# Plan: Video Pipeline Optimization — Error-Proofing and Latency Reduction for `sink`'s Compositor Companion to `plevion-docs/SPEC_VIDEO_PIPELINE_OPTIMIZATION.md` (post Spec Revision, human-approved; published to Confluence, page id `2293761`, [Spec — Video Pipeline Optimization](https://plevion.atlassian.net/wiki/spaces/PD/pages/2293761/Spec+Video+Pipeline+Optimization)). Read the spec in full before this plan — several tasks below only make sense against its corrected reasoning (the Spec Revision pass overturned parts of the original draft; this plan is built against the *current* spec text, not any earlier summary of this work). ## Goal Ship the spec's seven areas as independently mergeable, incrementally verifiable `sink`-repo PRs (this repo's own convention: one PR per task, not one PR per phase — unlike `frontend-admin`'s Build convention), in an order that respects the spec's own dependency reasoning: cheap/low-risk fixes first, the one fix most likely to explain the actual incident early (so its outcome is real evidence, not just a nice-to-have), and the corrected recovery mechanism (Area 3) only once its own prerequisite (Area 5's queue-contention question) has an answer — building it on the *original*, Spec-Revision-rejected sequencing would ship a recovery mechanism that can starve under exactly the load it exists to survive. **Two items the spec explicitly deferred stay deferred here, not silently scheduled**: Option B (isolating the x264 encoder specifically) and dedicated root-cause reproduction/correlation work for Open Question 1. Both wait until the fixes below ship and there's real signal on whether they're still needed. ## Planning-time decisions vs. open checkpoints Unlike this org's usual Plan convention of resolving a spec's remaining open questions before Build starts, **two of the spec's open questions are deliberately left as explicit checkpoints below, not resolved here** — the spec's own instruction (and this plan's own brief) was not to silently decide them: 1. **Area 5 — dedicated Celery queue vs. raised capacity** (Open Question 2). The spec leans toward a dedicated queue (mirrors the existing `capturer.*` routing pattern, doesn't require load data to be safe) but explicitly left the final call open. See Checkpoint B. 2. **Area 3a-ii — heartbeat field and write cadence** (Open Question 5): reuse `Composition.updated_at`, or add a dedicated `pipeline_heartbeat_at` field; write every tick, every Nth tick, or time-based. See Checkpoint C. Everything else below is planned concretely — the spec's own recommendations were specific enough to plan against directly. ## Risks | Risk | Mitigation | |---|---| | Any change to `apps.compositor.services.dispatcher` regresses a running broadcast — this is live, user-visible infrastructure, not a batch job with a retry budget. | Every task touching `dispatcher.py` is marked **ask first** below (per `sink/CLAUDE.md` Boundaries, carried into the spec's own Boundaries section) — confirm with the user before writing code, not just before merging. Test against a real local composition (per `sink/CLAUDE.md`'s own Testing section) before opening any such PR, not just unit tests. | | Area 3's watchdog (Task 5) ships before Area 5's queue question (Task 4/Checkpoint B) is resolved, reintroducing the exact contradiction Spec Revision caught — the recovery mechanism starves under the load that needs it most. | Task 5 is **hard-blocked** on Checkpoint B's resolution, not just sequenced after it in the task list — stated explicitly in Task 5's own preconditions. | | Area 2's JWT-caching fix introduces a *new* staleness bug (a cached JWT outlives its actual validity, or a stale cache entry survives past a source's re-attach). | Task 2's acceptance criteria require confirming the JWT's real TTL against the cache's TTL before picking a cache duration — not assumed equal. | | The watchdog (Task 5) double-starts a pipeline that's actually still alive but just had a slow tick (false-positive stale heartbeat under legitimate load, e.g. Area 2's own auth-latency stall before it's fixed). | Task 5's acceptance criteria require a heartbeat staleness threshold with real margin above the worst observed tick duration, and an explicit "is another run_compositor already alive for this id" guard before dispatching a duplicate `composition.start`. | | Scope creep — turning a "don't overengineer" spec into an overengineered Build. | Tasks 3, 6, and the two deferred items are explicitly scoped as *measurement*/*small addition*, not redesigns — see each task's own Acceptance criteria for what's deliberately out of scope. | ## Task Breakdown ### Task 1 — Area 1: DB query fixes (`select_related` + lazy `active_scene_id` resolution) **Not ask-first** — mechanical, additive, no behavior change to the running pipeline itself (query shape only). Independently mergeable, no dependency on anything else in this plan — do this first or in parallel with Task 2. **Files:** - `src/apps/compositor/tasks.py` — `_attach_initial_sources`'s base queryset (currently `CompositionSource.objects.filter(composition_id=object_id, attached_at__isnull=False)`, `tasks.py:454` as of spec-writing time, verify current line): add `.select_related('parent_source__capture_session', 'capture_session')`. - `src/apps/compositor/tasks.py` — the three `active_scene_id` re-fetch sites inside `_handle_add_overlay`/`_handle_add_source`/`_handle_set_source_visibility`: replace each handler's own `Composition.objects.filter(id=object_id) .values_list('active_scene_id', flat=True).first()` call with a value resolved **once per non-empty `get_compositor_commands` drain** in `run_compositor`'s own loop, threaded into `_dispatch_command` and down into whichever handler needs it — not resolved unconditionally every loop iteration (that would add 2 queries/sec to a currently-zero-query idle tick, the exact pessimization Spec Revision caught in the original draft). **Acceptance:** - [ ] A composition with at least one INTERACTIVE copy (`parent_source` set) attaches with the query count for `_attach_initial_sources` reduced by 2 per copy (verify via `django.db.connection.queries` around a controlled `_attach_initial_sources` call in a test, not just code inspection). - [ ] An idle `run_compositor` tick (no commands pending) still issues zero DB queries — a regression test asserting this explicitly, so a future change can't silently reintroduce the per-tick query the original draft's fix would have added. - [ ] A drain containing multiple commands that each need `active_scene_id` (e.g. two `add_overlay`s back to back) issues exactly one `active_scene_id` query for the whole drain, not one per command. - [ ] Existing `test_dispatcher.py`/`test_tasks.py` suites pass unchanged in behavior (this is a query-shape change, not a logic change — no existing assertion on handler *behavior* should need to change, only new assertions on query *count*). **Verification:** `docker compose -f docker/compose.yaml run --rm app_test` (lint + format + full suite, per `sink/CLAUDE.md`'s own Testing convention) — this task is fully offline-verifiable against the existing test DB, no live cluster needed. --- ### Task 2 — Area 2: pooled `requests.Session` + short-lived JWT cache for `get_object_jwt` **Ask first** — this changes a call path `dispatcher.py`'s `add_source` and `tasks.py`'s `start_composition_task` both depend on for every source attach; while the change itself lives in `apps/common/services/auth.py`, its blast radius is directly the compositor hot path. Sequence this **early** (right after or alongside Task 1) — the spec explicitly flags this as worth doing regardless of which of Open Question 1's three hypotheses turns out to be correct, and its own outcome (does the `c154ade3`-style incident stop recurring?) is real evidence toward answering that question, not just a latency optimization. **Files:** - `src/apps/common/services/auth.py` — `get_object_jwt`/`get_code_token_for_object`/ `_exchange_code_for_jwt`: introduce a shared, reused `requests.Session` (module-level or passed through, whichever fits this module's existing style — check for an existing session-management convention elsewhere in `apps/common` before inventing one) instead of a bare `requests.post` per call, plus a short-lived in-process cache keyed by `(object_id, object_type, purpose)` or equivalent. - Confirm the real JWT TTL (check `auth`'s own token-minting code/config, or the JWT's own `exp` claim at runtime) before picking the cache's TTL — per the Risks table, this must not be assumed equal to some arbitrary cache duration. **Acceptance:** - [ ] Two `add_source` calls for two different sources within one pipeline's lifetime reuse one TCP connection to `auth` (verified via the session object's own connection pool, or a request-count/connection-count assertion in a test) instead of opening a fresh connection per call. - [ ] A JWT fetched once for a given object is reused for any second request within the cache's TTL window, and a *new* JWT is fetched once that window expires — both paths covered by a test. - [ ] The `None`-JWT-on-failure behavior (`add_source`'s `if jwt:` guard, `dispatcher.py`) is unchanged in shape but now has an explicit test asserting what happens when `get_object_jwt` legitimately returns `None` after the caching change (this failure path already existed and is implicated in the incident's Open Question 1 hypothesis (b) — don't let the caching change accidentally paper over or swallow this case differently than before). - [ ] `apps/common/services/auth.py`'s own test suite covers the new session-reuse and cache-hit/miss/expiry behavior explicitly, not just "still returns a JWT." **Verification:** `docker compose -f docker/compose.yaml run --rm app_test`, plus a manual live check per `sink/CLAUDE.md`'s Testing conventions: publish a composition with 2+ `CONTROLLER_STREAM` sources locally, confirm via logs/tracing that `auth` sees fewer requests for the same pipeline start than before this change. > **Checkpoint A** — after Task 2 merges and deploys, watch for whether the > `c154ade3`-style incident (all branches failing near-simultaneously shortly after > pipeline start) recurs. This is real evidence toward Open Question 1, not a > formality — if it stops recurring, that's a strong signal hypothesis (b) (auth > latency) was dominant, and Option B / further root-cause work stays deferred with > more confidence. If it still recurs, that's evidence for hypothesis (a) or (c), and > worth revisiting before investing further down this plan. --- ### Task 3 — Area 4: re-measure command-dispatch latency after Task 2 **Not ask-first** — this is a measurement task, not a code change. Depends on Task 2 being deployed (the spec's own reasoning: Area 4's "no change" verdict rests on Task 2 removing the actual dominant term, not the 500ms poll interval itself — re-measure before treating that verdict as final). **Files:** none (a temporary instrumentation/logging addition is acceptable for the measurement itself, but should not land as a permanent code change unless the measurement reveals a real problem worth its own follow-up task). **Acceptance:** - [ ] Real command-dispatch latency (time from `send_compositor_command` to the corresponding handler actually running) measured for `add_source`/`add_overlay` specifically, post-Task-2, against a live local composition. - [ ] A written note (in this plan file or a short follow-up doc) stating whether the measured latency still supports Area 4's "no change to the polling architecture" verdict, or whether it reopens the question. **Verification:** Manual, against a live local cluster — not a unit test. This task's only deliverable is the measurement and its written conclusion. --- ### Task 4 / Checkpoint B — Area 5: resolve the dedicated-queue-vs-capacity decision **This task IS the checkpoint** — per this plan's own brief, the spec left this open deliberately, and it must be resolved by a human decision, not defaulted to the spec's own lean. **What needs deciding**: `CELERY_TASK_ROUTES` (`settings/common.py`) currently routes only `capturer.*` to its own queue; `composition.start`/`run_compositor`, `retry_add_source`, `apply_scene_auto_return`, and `recording.record` all share `sink-worker`'s single 4-slot (`--concurrency=4`) default queue. Area 3's watchdog (Task 5) would dispatch its recovery `composition.start` onto this same queue — meaning under saturation, the recovery mechanism can't run at exactly the moment it's needed. Two paths, as the spec itself framed them: - **(a) Dedicated queue** for `composition.start`/`run_compositor` (and arguably `retry_add_source`/`apply_scene_auto_return`, which have the identical starvation problem independent of the watchdog) — mirrors the existing `capturer.*` pattern exactly, doesn't require production load data to be safe, but needs a new worker queue/routing config and possibly a capacity decision for *that* queue too. - **(b) Raised capacity** on the existing queue (`--concurrency`/`replicas`) — simpler routing-wise, but needs real production load data to size correctly (unavailable today per the spec's own Assumptions), and doesn't fully close the starvation risk on its own (a large enough burst still saturates a single queue, just at a higher threshold). - **(c) Both** — a dedicated queue sized conservatively now, revisited once real load data exists. **Acceptance (of this checkpoint, not a code task):** - [ ] A human decision recorded here (or in a plan update) choosing (a), (b), or (c), with the reasoning — this becomes Task 5's actual precondition. - [ ] If (a) or (c): a follow-up task added to this plan for the routing/queue change itself (new `CELERY_TASK_ROUTES` entry, new worker Deployment or concurrency split in `plevion-k8s/services/sink/base/workers/`, mirroring `sink-capturer- worker.yml`'s existing shape) before Task 5 starts. **Verification:** N/A — this is a decision point, not a code change. --- ### Task 5 — Area 3a-i/3a-ii: liveness heartbeat + recovery watchdog **Ask first**, and **hard-blocked on Checkpoint B's resolution** — do not start this task until Task 4 has a recorded decision and (if applicable) its own follow-up queue/routing task has landed. Building this against the unresolved queue-contention risk reintroduces the exact defect Spec Revision found in the original draft. Also **blocked on Checkpoint C** (heartbeat field/cadence decision) — see below. **Files:** - `src/apps/compositor/models.py` — if Checkpoint C picks a dedicated field: add `pipeline_heartbeat_at` (nullable timestamp) to `Composition`, plus its migration (per `sink/CLAUDE.md`'s squash-to-`0001_initial` convention — this app is still pre-launch, so extend the existing initial migration rather than adding a numbered one, unless that convention has changed since the spec was written — verify). - `src/apps/compositor/tasks.py` — `run_compositor`'s main loop: write the chosen heartbeat signal at the chosen cadence (per Checkpoint C). - New: a periodic sweep (Celery beat task — note: **no Celery beat infrastructure exists in `sink` today**, confirmed absent from `docker/compose.yaml` and `plevion-k8s/services/sink/base/workers/`; this task includes standing it up, not just adding a task to an existing scheduler) that finds `Composition.objects.filter(phase__in=LIVE_PHASES, ...)` with a stale heartbeat and dispatches `composition.start` directly — bypassing `publish_composition`'s phase-transition validation entirely (this is recovery, not a phase change; calling `publish_composition` on an already-LIVE composition raises `COMP_SOURCE_008`, per the spec's own finding). - The dispatch must guard against double-starting: confirm no other `run_compositor` is already alive for that composition id before firing (e.g. a short-lived lock, mirroring `_acquire_retry_lock`'s existing Redis-`SETNX` pattern in `dispatcher.py`/`tasks.py`, adapted to whole-composition scope). - `plevion-k8s/services/sink/base/workers/` — a new beat Deployment (or an addition to an existing worker's own command, if a lighter-weight approach is preferred — decide during Build, not presupposed here) plus whatever queue/routing Checkpoint B settled on. **Acceptance:** - [ ] A composition whose `run_compositor` process is hard-killed (simulate via `kubectl delete pod` mid-composition, or a test-level SIGKILL equivalent) has its heartbeat go stale, gets picked up by the sweep, and resumes without manual intervention — this is the core scenario the whole task exists for, test it for real, not just unit-test the sweep's query logic in isolation. - [ ] A composition that's actually still alive (heartbeat fresh) is never double-started by a sweep tick that happens to run concurrently. - [ ] A deliberately `stop_composition`-ed composition (`phase = ENDED`) is never picked up by the sweep — the query must scope to `LIVE_PHASES` correctly. - [ ] The heartbeat write itself doesn't introduce a meaningful new per-tick cost — measure query/write cost added to `run_compositor`'s loop, confirm it's negligible relative to Task 1's "idle tick = zero queries" baseline (a *bounded*, deliberate addition, not a regression of that same finding). - [ ] A composition failing during Checkpoint A's rolling-deploy scenario (Task 3a-iv in the spec — a SIGKILL from a routine `sink-worker` redeploy) recovers via this same mechanism, confirmed against a real `bootstrap.sh` redeploy while a test composition is live. **Verification:** Both automated (new tests in `tests/compositor/`) and live, against a real local cluster — per the Risks table, this is exactly the kind of change that needs real-composition testing before merge, not unit tests alone. > **Checkpoint C** — before starting Task 5's implementation: decide the heartbeat > field (reuse `Composition.updated_at`, or add `pipeline_heartbeat_at`) and write > cadence (every tick vs. every Nth tick vs. time-based, e.g. "at most once per 5s > regardless of tick rate"). The spec left this open deliberately (Open Question 5) — > resolve it here, informed by Task 5's own "negligible per-tick cost" acceptance > criterion above. --- ### Task 6 — Area 3a-iv: graceful shutdown handling for `run_compositor` **Ask first** — touches `tasks.py`'s `run_compositor` main loop and `plevion-k8s/services/sink/base/workers/sink-worker.yml`'s pod spec. Sequenced **after** Task 5 ships, not before — this is a complementary reduction in how often a hard-kill happens at all (fewer SIGKILLs to recover from), not a substitute for the watchdog (which must work regardless of *why* a pipeline died). Building this first would leave the actual unrecoverable-state gap (3a-i) open the whole time. **Files:** - `src/apps/compositor/tasks.py` — `run_compositor`: SIGTERM handling that lets an in-progress composition wind down cleanly (or at least reach `_clear_compositor_ task_id` reliably) within a bounded grace period, rather than relying on Celery's default warm-shutdown behavior (which waits indefinitely for a task that never returns on its own). - `plevion-k8s/services/sink/base/workers/sink-worker.yml` — add `terminationGracePeriodSeconds` sized to match, and consider a `preStop` hook if that's the cleaner mechanism for signaling "stop accepting new pipeline work, finish winding down existing ones" during a rolling deploy. **Acceptance:** - [ ] A `sink-worker` rolling deploy with a live composition no longer relies solely on Task 5's watchdog to recover — the composition either winds down cleanly and restarts fast, or the watchdog still catches it, but the SIGKILL-and-recover path is no longer the *only* path. - [ ] No regression to `bootstrap.sh`'s own redeploy flow or its existing postflight health checks. **Verification:** Live, against a real `bootstrap.sh --with-frontend` redeploy with a test composition running throughout. --- ### Task 7 — Area 3b: retry-exhaustion WS notification **Ask first** (touches `tasks.py`'s `retry_add_source`) but small and fully independent of Tasks 4-6 — safe to schedule in parallel with them once Task 1/2 are done, no ordering dependency. **Files:** - `src/apps/compositor/tasks.py` — `retry_add_source`: on the final failed attempt (`attempt == _SOURCE_RETRY_MAX_ATTEMPTS`), broadcast a new, distinct WS event (e.g. `source_retry_exhausted`) via the existing `broadcast_group_event` mechanism `_notify_source_auto_removed` already uses. - `frontend-admin` — a small follow-up to distinguish this new event from the existing `source_removed` one in the composition-dashboard's WS handler (mirrors the existing `composition-dashboard/index.tsx` pattern for `source_removed`) — flagged here as a cross-repo follow-up, not detailed further in this sink-focused plan; open as its own small frontend-admin task once the sink side lands. **Acceptance:** - [ ] A source that exhausts all 30 retry attempts fires exactly one `source_retry_exhausted` event, distinguishable from `source_removed` by its own `event` key value. - [ ] A source that succeeds on any attempt before exhaustion never fires this event. **Verification:** `docker compose -f docker/compose.yaml run --rm app_test`. --- ### Task 8 — Area 7b: `sink-worker`/`sink-capturer-worker` resource limit review **Not ask-first** (plevion-k8s manifest change, not `dispatcher.py`) — independent of every other task, safe to schedule anytime, including in parallel with Task 1. **Files:** - `plevion-k8s/services/sink/base/workers/sink-worker.yml` — CPU limit currently 6 cores against a 4-slot `--concurrency`; review whether the limit should rise to actually cover 4 concurrent compositions at their observed ~2-core-each cost (~8 cores), or whether Task 4/Checkpoint B's capacity decision changes this calculation first (if concurrency itself changes, redo this sizing against the new number, not the old one). - `plevion-k8s/services/sink/base/workers/sink-capturer-worker.yml` — currently observed pinned at ~3.9 of its 4-core limit under a single interactive overlay; review whether this needs headroom independent of the compositor-side changes above. **Acceptance:** - [ ] A documented sizing decision (in this plan or a short follow-up note) for both workers' CPU limits, grounded in real observed per-composition/per-overlay cost, not a guess. - [ ] `microk8s kubectl kustomize environments/local/ --enable-helm` still builds cleanly with the new limits. **Verification:** `yamllint .` + `kubectl kustomize` build (plevion-k8s's own CI gate), plus a live resource-usage check post-deploy. --- ## Explicitly not scheduled (deferred per the spec, revisit after the above ships) - **Option B — isolate the x264 encoder specifically.** Revisit only if encoder failures keep recurring after Tasks 2/5/6 ship and Checkpoint A's evidence doesn't point clearly at hypothesis (b) or (c) instead. - **Dedicated root-cause reproduction/correlation for Open Question 1** (deliberately forcing concurrent-composition CPU load, or historically correlating `auth` latency/errors against the `14:44:47` incident timestamp) — worth doing if Checkpoint A's passive evidence (does the incident recur after Task 2?) isn't conclusive on its own. ## Critical files (read before Build starts on any task) - `sink/src/apps/compositor/services/dispatcher.py`, `tasks.py`, `services/lifecycle.py` - `sink/src/apps/common/services/auth.py` - `sink/src/settings/common.py` (Celery routing/broker config) - `sink/docker/compose.yaml` (worker concurrency) - `plevion-k8s/services/sink/base/workers/sink-worker.yml`, `sink-capturer-worker.yml` - `sink/CLAUDE.md` (Boundaries — `dispatcher.py` ask-first convention, migration- squash convention, Testing conventions) ## Verification (whole-plan) No single end-to-end test covers this plan — verification is per-task (each task's own Acceptance/Verification above). The one whole-plan check worth doing once Tasks 1-7 have shipped: confirm a hard-killed `run_compositor` process recovers automatically (Task 5's core scenario) **and** that recovery doesn't get starved under a deliberately saturated queue (Checkpoint B's resolution actually holding) — this is the specific failure mode Spec Revision caught in the original draft, and the one thing worth a real combined check rather than trusting each task's isolated tests. ## Next step (once this plan is approved) Per `plevion-docs/WORKFLOW.md`'s publishing convention: this plan, once approved, needs to go to Confluence as a child page of the Spec page (id `2293761`), titled `Plan — Video Pipeline Optimization`, plus one Jira Story (titled after the feature) with one Subtask per task above (8 subtasks, Tasks 1-8; the two deferred items are not tasks and don't get subtasks). Not done yet — flagging as the next step once this plan itself is approved, per this plan's own brief not to publish prematurely.