fix: Race conditions, memory usage, and Redis cleanup

[filthyrake]

Summary

Batch fix for 4 open reliability and performance issues:

• [Reliability] Stale Stream Detection Race Condition #551 (Critical): Stale stream detection race condition — replaced check-then-act with atomic UPDATE ... WHERE that includes status + timestamp conditions, so a segment arriving between SELECT and UPDATE causes the UPDATE to match 0 rows instead of prematurely terminating the stream
• [Reliability] Analytics Recomputation Race Condition #550 (Critical): Analytics recomputation race condition — wrapped the check-if-exists + insert/update in a database.transaction() to prevent concurrent requests from both attempting inserts and hitting duplicate key errors
• [Performance] Playlist Generation Fetches ALL Segments Into Memory #559 (Medium): Playlist generation fetches all segments into memory — pushed DVR cutoff filter into the SQL WHERE clause instead of fetching all segments and filtering in Python (saves ~2.6MB per playlist update for a 24h stream)
• [Reliability] SSE Connection Cleanup May Leak Redis Resources #562 (High): SSE connection cleanup leaks Redis resources — added force-reset of the underlying Redis pub/sub connection when subscriber.close() times out, preventing connection pool exhaustion

Note: #548 (WebSocket broadcast sequential sending) was already fixed in 7a0ce90 — broadcast_to_stream already uses asyncio.gather with bounded concurrency.

Files Changed

File	Issue	Change
api/live_tasks.py	#551	Atomic UPDATE with WHERE on status + timestamp
api/studio_analytics.py	#550	Transaction-wrapped upsert
api/live_playlist.py	#559	Database-level DVR cutoff filtering
api/studio_sse.py	#562	Force pub/sub reset on close timeout

Test plan

• All 4 modified files compile cleanly
• Existing analytics test suite passes (47/47 — 6 pre-existing integration failures unrelated to changes)
• Verify stale stream detection doesn't prematurely terminate streams under load
• Verify analytics recomputation handles concurrent requests without duplicate key errors
• Verify playlist generation memory usage is reduced for long streams
• Verify SSE connections don't leak Redis connections over time

Closes #551, closes #550, closes #559, closes #562

🤖 Generated with Claude Code

[Copilot]

Pull request overview

This PR targets reliability/performance issues in the live streaming subsystem by making state transitions and analytics writes race-safe, reducing playlist memory usage, and hardening SSE Redis pub/sub cleanup.

Changes:

• Make stale stream transitions atomic via conditional UPDATE ... WHERE checks (eliminates check-then-act races).
• Replace analytics “check then insert/update” with a single database-level upsert (ON CONFLICT DO UPDATE).
• Push DVR cutoff filtering into the segment query to avoid loading/filtering large segment lists in Python.
• Add Subscriber.force_close() and use it in SSE cleanup to prevent Redis pub/sub resource leaks on close timeouts.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
api/live_tasks.py	Uses conditional updates to prevent stale-detection races when segments arrive concurrently.
api/studio_analytics.py	Switches analytics summary persistence to an atomic upsert.
api/live_playlist.py	Applies DVR window cutoff at the database level to reduce memory usage.
api/studio_sse.py	Uses force_close() for subscriber cleanup to mitigate Redis pool exhaustion.
api/pubsub.py	Introduces Subscriber.force_close() to enforce cleanup even when graceful close hangs.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

force_close() clears _subscribed_channels / _subscribed_patterns and then calls pubsub.close() without attempting unsubscribe/punsubscribe like close() does. This makes cleanup semantics inconsistent and can leave server-side subscriptions around until the connection is torn down (and also makes it harder to reason about behavior when pubsub.close() succeeds). Consider capturing the channel/pattern lists before clearing state and attempting a best-effort unsubscribe/punsubscribe (with a short timeout) before the timed close, similar to close().

[Copilot]

There are existing unit tests covering Subscriber.close() cleanup behavior, but the newly added force_close() (including the timeout path that disconnects the underlying connection) isn't covered. Adding tests for both the normal path (calls pubsub.close()) and timeout path (forces disconnect) would help prevent Redis connection leaks from regressing.

[Copilot]

This upsert is built with PostgreSQL-specific pg_insert(...).on_conflict_do_update(...). The codebase supports SQLite via DATABASE_URL=sqlite:///..., so this will raise at runtime under SQLite. Consider a dialect-agnostic upsert strategy (e.g., transaction + insert with IntegrityError fallback to update), or branching on the active dialect to use the appropriate ON CONFLICT implementation.