diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..dba9cd4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,20 @@ +__pycache__/ +*.py[cod] +.venv/ +venv/ +.pytest_cache/ +.coverage +htmlcov/ + +.env +.env.* +!.env.example + +*.sqlite3 +*.sqlite3-* +controller-data/ +data/ +playlists.yaml.pre-v2.*.bak + +cookies*.txt +secrets/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..19151c5 --- /dev/null +++ b/README.md @@ -0,0 +1,975 @@ +# Video Player Controller v2 + +Video Player Controller downloads media into ten stable slots, publishes the +existing Home Assistant and Companion state topics, and coordinates playback +between the authoritative Video Player Node and the Audio Router. + +Controller v2 is deliberately conservative: publishing a command is not proof +that playback changed. Only fresh Player Node status or telemetry can establish +playing, paused, or stopped state. A command accepted by the MQTT broker remains +`pending` until that evidence arrives. + +The Controller does not run or supervise `mpv`. The Player Node owns the MPV +lifecycle and sends audio only to `pipewire/ytplayer_bus`; the Controller owns +the request/confirmation/restoration workflow for Audio Router mode. + +## Architecture and safety model + +```text +Browser / HA / Companion + | + | download requests and Controller commands + v +Flask -> bounded event queue -> single Controller coordinator + | | | + | | +-> SQLite state/audit + | +------------> bounded yt-dlp workers + +---------------------> MQTT + | + +-----------------------------+------------------+ + | | + v v + Video Player Node Audio Router + (authoritative state) (physical audio mode) + | + +-> mpv -> pipewire/ytplayer_bus +``` + +MQTT callbacks, web requests, timers, and worker notifications converge on a +single coordinator thread for orchestration and publication. Download workers +perform their external work independently and commit only generation-fenced +SQLite transitions. SQLite transactions are the final persistence boundary. +This provides the following invariants: + +- Slots `1` through `10` are permanent identities. A hole stays a hole; deleting + slot 3 never renames slot 4 or its directory. +- A download slot is reserved in an immediate SQLite transaction before a + worker is submitted. A partial unique index prevents two live jobs in one + slot. +- Every reuse increments the slot generation. A late worker or deletion token + cannot commit against a newer occupant. +- Active, pending, uncertain, or operator-review playback guards block slot + replacement and deletion. +- A reserved or running download blocks deletion and expiry. +- Filesystem deletion is limited to the canonical `slotN` directory. The + database first atomically fences the generation as `deleting`; filesystem + work then runs asynchronously. A directory is atomically moved under + `.controller-trash` before recursive removal, and interrupted deletion + recovery is deferred until Player safety is established. +- MQTT can be unavailable at startup. The web/coordinator process remains up, + reports degraded health, and retries with bounded exponential backoff. + +The canonical managed media layout is: + +```text +/mnt/Media/YTPlayer/ + slot1/playlist.m3u + slot2/playlist.m3u + ... + slot10/playlist.m3u +``` + +The base path is configurable, but both Controller and Player Node must see the +same absolute paths. + +## Player-authoritative operation model + +Controller commands are serialized. A second transition is rejected while a +play, stop, pause, or resume operation is waiting for evidence. + +For a play request, the Controller: + +1. Validates the stable slot or explicit local path and protects a managed + slot before any external command is sent. +2. Acquires the Audio Router mode using the bounded policy described below. +3. Publishes the Player command at QoS 1 with `retain=false` and waits up to + `mqtt.publish_ack_timeout_seconds` for the broker PUBACK. +4. Reports the request as pending. It does **not** publish `playing` merely + because the command was sent. +5. Confirms, rejects, or marks the operation uncertain from matching Player + status/telemetry. The normal Player-evidence deadline is 10 seconds. + +A PUBACK proves only broker receipt, not Player execution. If the PUBACK is +uncertain, the Controller replaces its MQTT transport so an unacknowledged QoS +1 command cannot be replayed later, and it preserves uncertain playback state +for reconciliation. + +Player evidence is interpreted using the current Player Node contract: + +- `playing` status or telemetry with a path establishes active playback. +- Telemetry's boolean `paused` field is authoritative for pause state. Player + `paused` and `toggled` status only show that control was issued; Controller + waits for telemetry before confirming the resulting pause value. If pause, + resume, or toggle confirmation times out, both phase and pause value project + as `unknown` until fresh evidence arrives. +- `busy` means the new play did not replace the Player's existing process. When + a slot/path is present, or an earlier identity is still known, that existing + playback remains authoritative. A real pathless `busy` with no prior identity + becomes unknown; Controller keeps the Router lease and global cleanup fence + until identity/terminal evidence is reasserted or an administrator reconciles + the uncertainty. +- Terminal `idle` requires one of `stop`, `finished`, `mpv_error`, `sigterm`, + `sigint`, or `shutdown`, or cleared telemetry. An ambiguous idle does not + erase active ownership. +- A generic Player `error` is not proof that MPV stopped. Known managed or + external playback becomes reviewable uncertainty; an identity-less error + keeps the global cleanup fence closed. Only error reasons `startup_error` and + `worker_start_error` prove that playback never became active and reduce to + idle even if cleared telemetry was lost. +- On reconnect, `idle` with `message:"player online"` and a path means the + Player is still active. The same announcement without a path means idle. +- The retained flat `youtube/player/state` topic is never playback or cleanup + evidence. Its value is historical across outages, and Player reconnect + deliberately emits flat `idle` even while MPV is active. Controller keeps the + subscription for Companion compatibility but does not use it to unlock + downloads, expiry, deletion recovery, or playback guards. + +Use `youtube/controller/cmd` for operational control. Publishing directly to +`youtube/player/cmd` bypasses Controller Audio Router acquisition and slot +guard creation. Controller will eventually reduce resulting Player evidence, +but there is an unavoidable cross-service race before that evidence arrives. + +## MQTT contracts + +Defaults remain compatible with the existing Player Node, Home Assistant, and +Companion integrations. Topic names can be configured except where another +component treats one as fixed; coordinate any change across all consumers. + +### Topic matrix + +| Direction | Default topic | Payload | QoS used by Controller | Retained | Contract status | +|---|---|---|---:|---:|---| +| Controller -> Player | `youtube/player/cmd` | JSON command | publish 1 | no | Preserved; required | +| Player -> Controller | `youtube/player/status` | JSON status | subscribe 1 | no | Player-authoritative | +| Player -> Controller | `youtube/player/state` | plain flat state | subscribe 0 | yes, by Player | Preserved Companion contract only; never cleanup evidence | +| Player -> Controller | `youtube/player/telemetry` | JSON telemetry | subscribe 0 | no | Player-authoritative | +| HA/Companion -> Controller | `youtube/controller/cmd` | JSON command | subscribe 1 | must be no | Preserved; retained commands are rejected | +| Controller -> consumers | `youtube/ingest/progress` | JSON progress | publish 0 | no | Preserved | +| Controller -> consumers | `youtube/media/playback//title` | plain string | publish 0 | yes | Preserved | +| Controller -> consumers | `youtube/media/playback//state` | plain string | publish 0 | yes | Preserved | +| Controller -> consumers | `youtube/media/playback//count` | decimal string | publish 0 | yes | Preserved | +| Controller -> consumers | `youtube/media/playback//items/` | plain string | publish 0 | yes | Preserved; empty payload clears stale items | +| Controller/operator -> Router | `audio/player/mode/set` | plain mode string | publish and subscribe 1 | Controller publishes no | Router contract | +| Router -> Controller | `audio/player/mode/state` | plain mode string | subscribe 1 | Router-owned | Router confirmation | +| Controller -> consumers | `youtube/controller/v2/events` | JSON event | publish 1 | no | Optional, additive v2 contract | +| Controller -> Home Assistant | `homeassistant/sensor/yt_slot__/config` | discovery JSON | publish 0 | yes | Preserved IDs | + +QoS in the incoming rows is the Controller's requested subscription QoS; the +publisher and broker determine the delivered QoS. Player commands and Router +mode commands are always emitted with QoS 1 and `retain=false`. + +The Controller uses Paho manual acknowledgement for incoming QoS 1 messages. +It acknowledges after successful coordinator handling and deduplicates repeated +packets by MQTT connection generation, topic, and packet ID. Retained Controller +commands are rejected so a reconnect cannot replay an old control operation. + +Detailed Player status and telemetry are valid evidence only when +`retain=false`. If either arrives retained, Controller treats it as historical, +does not reduce it, does not unlock cleanup, and does not clear guards. It logs +and audits `retained_player_evidence_rejected`, publishing that warning when the +optional v2 event topic is enabled; a retained QoS 1 message is deliberately +acknowledged after rejection so the same invalid evidence is not redelivered +indefinitely. + +### Controller command payloads + +Publish all examples at QoS 1 without retain: + +```bash +mosquitto_pub -h 10.40.0.2 -t youtube/controller/cmd -q 1 \ + -m '{"action":"play","slot":1,"label":"Opening"}' +``` + +Supported JSON objects are: + +```json +{"action":"play","slot":1,"label":"Opening"} +``` + +```json +{"action":"play","path":"/mnt/Media/YTPlayer/special/welcome.mp4","label":"Welcome"} +``` + +```json +{"action":"play","playlist":"/mnt/Media/YTPlayer/special/playlist.m3u","title":"Special"} +``` + +```json +{"action":"stop"} +``` + +```json +{"action":"pause"} +``` + +```json +{"action":"resume"} +``` + +```json +{"action":"toggle"} +``` + +```json +{"action":"delete_slot","slot":4,"url":"optional stale-command guard"} +``` + +`play` requires a ready managed slot or an existing local `path`/`playlist`. +An explicit path inside a managed slot derives that stable slot identity; an +explicit conflicting slot is rejected. `toggle` is translated to `pause` or +`resume` from the latest telemetry and is rejected when pause state is unknown. + +When the global cleanup gate lacks fresh Player evidence, manual deletion also +requires the exact boolean override: + +```json +{"action":"delete_slot","slot":4,"confirm_player_unknown":true} +``` + +That override only bypasses the global evidence gate. It does not bypass an +active download or a per-slot active/uncertain playback guard. It is separate +from administrative reconciliation, authorizes only that manual delete, and +does not open automatic expiry/recovery or consume a cleanup token. + +An accepted manual delete is atomically fenced in SQLite and returns +`{"status":"deleting","slot":4}`, not `deleted`. Filesystem removal happens +on a worker afterward. MQTT commands have no synchronous response topic, so an +MQTT consumer observes the retained slot state (and optional v2 warnings) rather +than depending on that in-process return value. + +### Preserved slot and Home Assistant state + +For every stable slot, Controller publishes retained `title`, `state`, and +`count` values plus retained item titles. Projected state values include +`empty`, `queued`, `downloading`, `ready`, `error`, `deleting`, `playing`, +`paused`, `unknown`, and `operator_review`. A pending play remains `ready` +until Player evidence says otherwise. + +The thirty discovery entities keep the established identifiers: + +- Discovery topic: `homeassistant/sensor/yt_slot__/config` +- Unique ID: `yt_slot__` +- Device identifier: `ytplayer-controller` +- Device name/manufacturer/model: `YouTube Controller` / `River of Glory` / + `YT Controller` + +Changing `homeassistant.discovery_prefix` changes only the discovery prefix. +Set `homeassistant.enabled: false` to suppress discovery; legacy slot state +publication continues. + +### Ingest progress + +`youtube/ingest/progress` is a non-retained JSON stream. Payloads always contain +`slot` and `state`, with applicable fields such as `title`, `count`, `message`, +`url`, `submitter`, and `note`. A `queued` event also contains `reused_slot`. +Common states are `queued`, `downloading`, `ready`, and `error`. + +### Optional v2 events + +`topics.controller_result` defaults to `youtube/controller/v2/events`. Set it +to YAML `null` to disable publication. This topic is deliberately additive: +legacy MQTT, Home Assistant, and Companion consumers never need it. + +Events are QoS 1, non-retained JSON: + +```json +{ + "version": 2, + "event_id": "6ed8f0e49a8b4d598e5285ca14ec89ef", + "type": "audio_router_warning", + "severity": "warning", + "message": "Audio Router did not confirm 'video' within 3 seconds", + "ts": 1786060800.0, + "slot": 1, + "operation_id": "4e32c8d628454426bff69c879e709f33", + "continued": true +} +``` + +`slot`, `operation_id`, and `continued` appear only when relevant. Each event +is also written to `audit_events` and logged, so the non-retained stream is not +the durable record. Event publication itself depends on MQTT availability. + +## Audio Router orchestration + +The normal playback target is the plain mode `video`: + +- Command: `audio/player/mode/set` +- Confirmed state: `audio/player/mode/state` + +Before play, Controller snapshots the latest confirmed Router state and creates +one persisted lease for the playback session. Router acquisition and restoration +both use the finite `router.confirmation_timeout_seconds`; neither can hang a +play operation indefinitely. + +### Unknown-state and failure policy + +The four Router settings are independent: + +| Setting | Default | Effect | +|---|---|---| +| `router.confirmation_timeout_seconds` | `3` | Maximum wait for acquisition or restoration confirmation | +| `router.when_state_unknown` | `attempt` | `attempt`, `continue`, or `reject` when no current state exists | +| `router.continue_on_confirmation_failure` | `true` | Continue playback after publish failure, disconnect, operator conflict, or confirmation timeout | +| `router.fallback_restore_mode` | `null` | Optional restoration target when the prior mode was unknown | + +`when_state_unknown` has these meanings: + +- `attempt`: request `video`, wait for the bounded confirmation, then apply + `continue_on_confirmation_failure` if it cannot be confirmed. +- `continue`: request `video` and start playback immediately without claiming + restoration ownership. This still records and publishes a warning. +- `reject`: reject this play request without sending a Player command. The + caller may retry after a current Router state becomes available. + +The safe service default is `attempt` plus +`continue_on_confirmation_failure: true`: it tries to set and confirm `video` +for three seconds, but unavailable Router state cannot hang or permanently +disable playback. The warning is persisted, logged, and published on the +optional v2 event topic when MQTT permits. + +`fallback_restore_mode` is used only when the previous mode was unknown and +acquisition subsequently becomes confirmed. A play that continues immediately +or after failed confirmation has no safe restoration ownership, so it does not +apply the fallback later. + +### Safe restoration and manual override + +At authoritative playback termination, Controller restores the prior mode only +when all of these remain true: + +- Controller actually changed the mode and received confirmation of `video`. +- The same lease is still held. +- The latest Router state is still the Controller-owned target. +- MQTT continuity was not lost. +- No live, non-self mode command and no conflicting state change indicated an + operator override. + +Controller subscribes to both Router mode set and state topics. Its exact +in-flight echo is recognized; every other live mode-set publication is treated +conservatively as operator intent, even if it repeats `video`. A retained set +message is ignored as stale intent. Controller does not publish Router state; +it accepts a received live or retained state message as the Router's current +mode. Once ownership is lost, Controller does not take it back merely because +state later returns to `video`. + +Restoration is requested once and confirmed within the same bounded timeout. +Failure produces a warning and no automatic retry. The Router's plain string +protocol has no publisher identity or compare-and-set token, so an exactly +simultaneous same-value command cannot be distinguished perfectly; conservative +ownership loss avoids overwriting a known manual choice. + +## Downloads, retention, and stable slots + +The web form reserves the first empty stable slot, or reuses the slot already +associated with the same URL when it is safe. Reservation and the `reserved` +download job commit before a worker starts. `downloads.max_workers` bounds +concurrent yt-dlp work; it does not change the ten-slot capacity. + +Each successful download builds `slotN/playlist.m3u` atomically and records item +metadata in the same slot generation. Downloads run yt-dlp with an explicit Node +JavaScript runtime and ffmpeg support. On restart, database jobs left in +`reserved` or `running` become `interrupted`, and their slots become `error` +instead of pretending success. Whenever a failed or restart-interrupted job +would otherwise leave an error slot without expiry, Controller assigns +`now + storage.retention_seconds`. This fallback does not replace an existing +expiry and prevents repeated failed downloads from permanently exhausting all +ten slots. + +Ready and error media can expire after `storage.retention_seconds`, checked at +`storage.expiry_interval_seconds`. Cleanup skips any slot with a live job or +playback guard. Cleanup and interrupted-deletion recovery are globally deferred +until safe detailed Player status/telemetry or an explicit audited +reconciliation makes cleanup safe. The retained flat Player state never unlocks +this fence. + +Manual and expiry deletion use the same asynchronous sequence: + +1. A transaction rechecks media state, URL guard, live jobs, and playback + guards, increments the generation, and commits state `deleting`. +2. A bounded worker rechecks playback guards before touching the filesystem, + atomically moves the canonical directory to generation-specific trash, and + checks again before recursive removal. If a late guard appears after the + move, the worker restores the directory and leaves deletion pending. +3. Only successful filesystem completion changes the database row to `empty`. + A failure remains durably `deleting`, emits `slot_deletion_pending`, and is + retried without blocking the coordinator. Error retries are accelerated to + at most 30 seconds; otherwise pending deletions are reconsidered during each + cleanup/expiry cycle and whenever safe Player evidence or reconciliation + releases the fence. + +Only one worker handles a given slot generation at a time. A late guard or a +stale generation therefore defers/fences completion instead of deleting a newer +occupant. A `deleting` state can be visible for more than one interval during a +filesystem failure or active playback review; that is deliberate, recoverable +state rather than a completed delete. + +Expiry never compacts slots. An expired slot becomes empty; every other slot +keeps its number and canonical directory. + +## Restart uncertainty and administrative reconciliation + +On startup, a last-known managed active slot or pending play is converted to a +`restart_uncertain` guard. An active path outside managed slots receives its +own persisted reconciliation ID. These records remain deletion-protected until +fresh Player evidence confirms active playback or a terminal state clears it. + +Separately, every Controller start and every MQTT connect/disconnect closes the +global cleanup fence until safe detailed Player evidence arrives. These are two +different reconciliation scopes: durable playback ownership reviews protect a +known slot/path, while a connection-scoped cleanup token handles the narrower +case where no active identity or guard is known but Player evidence is absent. + +### Durable playback ownership reviews + +`storage.uncertain_review_after_seconds` defaults to 900 seconds. Reaching that +deadline changes `awaiting_evidence` to `operator_review`; it **never** assumes +the Player stopped and never removes protection. Set the value to `0` for no +automatic review deadline. Review events are re-announced after each broker +connection. + +Find durable review tokens in `GET /healthz` under +`reconciliation_reviews`. When the optional event topic is enabled, the same +token appears as `operation_id` on `playback_reconciliation_required`. + +To clear a managed-slot review after independently verifying that playback is +not active: + +```json +{ + "action": "reconcile_playback", + "decision": "clear", + "confirm": true, + "slot": 1, + "review_id": "copy-the-current-review-id", + "actor": "operator-name", + "reason": "Player host was replaced and old playback cannot still exist" +} +``` + +For uncertain external playback, omit `slot`: + +```json +{ + "action": "reconcile_playback", + "decision": "clear", + "confirm": true, + "review_id": "copy-the-current-review-id", + "actor": "operator-name", + "reason": "Retired Player host verified offline" +} +``` + +The exact current review ID, `decision:"clear"`, and boolean `confirm:true` +are required. A stale ID is rejected, as is reconciliation while a Player +operation is pending. Clearing a review does not send stop to the Player; it +states that the operator has established safety and can permit later cleanup. +An incorrect clear can therefore allow deletion of media still in use. + +### Connection-scoped cleanup reconciliation + +Detailed evidence unlocks the global cleanup fence only when it proves either +terminal idle or an online active identity. Examples are cleared telemetry, +terminal/reconnect-idle status without a path, matching active telemetry/status, +or a busy/control/reconnect status backed by a known slot/path. Retained flat +state, pathless `busy`, identity-less generic `error`, malformed messages, and +mere MQTT connectivity do not qualify. + +When cleanup is fenced but there is no known active identity and no live slot +guard, `/healthz` exposes the current token: + +```json +{ + "cleanup_reconciliation": { + "review_id": "copy-the-current-connection-token", + "kind": "player_evidence", + "state": "awaiting_evidence", + "clearable": true + } +} +``` + +The value is `null` once detailed evidence/override has opened cleanup, and is +also withheld when a known identity or guard requires the durable playback +review workflow instead. `clearable:false` means a Router-gated play or Player +operation is still pending. + +After independently verifying that no Player can be using managed media, +publish this exact administrative command: + +```json +{ + "action": "reconcile_playback", + "scope": "cleanup", + "decision": "clear", + "confirm": true, + "review_id": "copy-the-current-connection-token", + "actor": "operator-name", + "reason": "Player appliance was independently verified offline" +} +``` + +The accepted in-process result is: + +```json +{ + "status": "cleared", + "review_id": "copy-the-current-connection-token", + "scope": "cleanup" +} +``` + +This token is intentionally not durable: it rotates on Controller startup and +every MQTT connection or continuity loss. Fetch `/healthz` again immediately +before publishing; an older token is rejected. The clear is connection-scoped, +recorded as `player_evidence_admin_override` in the audit database, and announced +as the optional `playback_reconciled` event. Later MQTT continuity loss closes +the fence and creates a new token. + +For a pathless `busy`, accepting this cleanup reconciliation also abandons the +held Router lease instead of guessing that playback ended and restoring an old +mode. The operator then owns any subsequent Router change. Fresh reasserted +playback followed by terminal evidence uses the normal safe restoration path. + +Cleanup scope accepts no `slot` and cannot clear or bypass a known active path, +slot guard, active download, or pending Player/Router transition. Use the +durable slot/external review payloads above for known playback uncertainty. This +override opens cleanup scheduling; each slot deletion still performs its own +transactional job/guard checks and late-guard checks. + +`actor` and `reason` are durable audit fields, but `actor` is a caller-supplied +label, not authentication or authorization. Enforce administrative access with +broker credentials and topic ACLs. + +## SQLite persistence + +Controller v2 uses one embedded SQLite database at +`/`, by default +`/controller-data/controller.sqlite3`. It introduces no external database +service. + +Schema version 2 contains: + +| Table | Purpose and important constraints | +|---|---| +| `schema_migrations` | Applied internal schema versions; v1 databases upgrade transactionally to v2 | +| `data_migrations` | One-time import provenance, source hash, backup path, and warnings | +| `slots` | Exactly one row for each stable slot 1-10, generation, media state, identity, timestamps, expiry, and error | +| `slot_items` | Ordered item title/path metadata keyed by slot and position | +| `download_jobs` | Reservation/worker lifecycle keyed by job ID and slot generation; one live job per slot | +| `slot_guards` | Active, pending, restart-uncertain, and legacy-uncertain playback protection; one live guard per slot | +| `player_observation` | Singleton last status/telemetry reduction, generation, active identity, pause, and reconciliation state/token | +| `audio_router_observation` | Singleton last mode, freshness, timestamp, and warning | +| `audio_leases` | Acquisition, ownership, restoration, deadline, and failure history for Router sessions | +| `audit_events` | Durable migration, safety, reconciliation, and warning history | + +Runtime connections enable foreign keys, a 5-second busy timeout, WAL journaling, +and `synchronous=FULL`. Writes use `BEGIN IMMEDIATE` transactions behind a +re-entrant process lock. Opening and migrating the store also takes the +cross-process `/controller-data/.controller-store.lock` file lock. Production +must run one application worker; these database protections are not permission +to start duplicate coordinators or MQTT clients. + +For a consistent manual backup, stop Controller cleanly before copying the +entire data directory. If an unclean live copy is unavoidable, preserve +`controller.sqlite3`, `controller.sqlite3-wal`, and `controller.sqlite3-shm` as +a set. + +## One-time `playlists.yaml` migration + +Migration runs automatically only when both conditions are true: + +1. The final SQLite database does not exist. +2. `/` exists, normally + `/controller-data/playlists.yaml`. + +The migration is non-destructive and idempotent: + +1. It takes the process lock and rechecks database existence. +2. It reads at most 8 MiB and rejects a source whose inode, size, or mtime + changes during the read. +3. It computes the full SHA-256 and creates a content-addressed, fsynced backup + named `playlists.yaml.pre-v2..bak`. An existing backup is reused + only if its hash matches. +4. It validates UTF-8, unique YAML mapping keys, slot keys in 1-10, and agreement + between each key and embedded `slot` identity. +5. It creates a separate staging SQLite file, initializes all ten slots, imports + the YAML in one transaction, records migration provenance, and checks schema, + foreign keys, integrity, and the exact 1-10 slot set. +6. It fsyncs and atomically replaces the final database path only after all + validation succeeds. + +The original YAML is never renamed, edited, or removed. The backup is created +before YAML parsing, so even malformed YAML can leave a verified recovery copy. +A parsing, import, or validation failure rolls back the transaction, removes the +staging database, and occurs before final database installation. The final step +is an atomic replace followed by a directory fsync; if that last fsync itself +fails, a validated final database may already be present and must be inspected +rather than overwritten or blindly reimported. Fix or restore the YAML and retry +only when the final database is absent, or return to the old deployment; do not +discard the original or hash-named backup. + +Once `controller.sqlite3` exists, startup never imports YAML again and later +YAML edits are ignored. That database-presence rule plus the `data_migrations` +record makes the migration idempotent. An existing schema-v1 Controller SQLite +database is upgraded transactionally rather than reimporting YAML. + +Legacy state mapping is explicit: + +| Legacy state | v2 media state | Extra treatment | +|---|---|---| +| `ready`, `idle`, `finished` | `ready` | Preserves slot number, metadata, and expiry | +| `playing`, `paused`, `busy` | `ready` | Creates a `legacy_uncertain` operator-review guard | +| `queued`, `downloading` | `error` | Records an interrupted download job and warning | +| `error` | `error` | Preserves or supplies the error message | +| any unknown value | `error` | Records a warning rather than guessing | + +Missing URLs and invalid counts/timestamps are recorded as errors or migration +warnings. Parseable legacy UTC timestamps are converted to Unix timestamps. +Any migrated error/interrupted slot with no usable legacy expiry receives the +same migration-time `now + storage.retention_seconds` fallback, recorded as a +migration warning, so it can later expire when cleanup is safe. An existing +legacy expiry is preserved. +Migration does not move, rename, compact, or delete media. Legacy YAML does not +contain complete per-item paths, so v2 does not invent them; existing playlist +files and retained item topics remain untouched until authoritative v2 metadata +is available. + +## Configuration + +`config.yaml` is loaded relative to the application, not the invoking shell's +working directory. The production image bind-mounts it read-only at +`/app/config.yaml`. Configuration is validated before service startup and has no +hot reload; restart the one-worker service after changes. + +### Environment overrides + +Only the following environment variables override YAML: + +| Environment variable | YAML setting | Application default | +|---|---|---| +| `MQTT_HOST` | `mqtt.host` | `10.40.0.2` | +| `MQTT_PORT` | `mqtt.port` | `1883` | +| `MQTT_USERNAME` | `mqtt.username` | unset | +| `MQTT_PASSWORD` | `mqtt.password` | unset | +| `MQTT_CLIENT_ID` | `mqtt.client_id` | `yt-controller` | +| `YTPLAYER_MEDIA_BASE` | `storage.media_base` | `/mnt/Media/YTPlayer` | +| `YTPLAYER_DATA_BASE` | `storage.data_base` | `/controller-data` | +| `YTPLAYER_RETENTION_SECONDS` | `storage.retention_seconds` | `172800` | +| `YTPLAYER_COOKIE_PATH` | `downloads.cookie_path` | `/run/secrets/yt-dlp-cookies.txt` | +| `PORT` | `web.port` | `7531` | + +MQTT username and password must either both be set or both be unset. Media, +data, and cookie paths must be absolute. `storage.slot_count` is validated as +exactly 10 and cannot be used to compact or resize the stable namespace. + +Compose-only deployment variables are documented in `.env.example`: + +| Variable | Purpose | +|---|---| +| `FLASK_SECRET_KEY` | Required long random Flask signing secret; no production fallback | +| `CONTROLLER_DATA_DIR` | Host directory mounted at `/controller-data` | +| `YTPLAYER_MEDIA_DIR` | Host media directory mounted at `/mnt/Media/YTPlayer` | +| `CONTROLLER_SECRETS_DIR` | Host secrets directory mounted read-only at `/run/secrets` | +| `CONTROLLER_WEB_PORT` | Host-side published HTTP port | +| `CONTROLLER_UID`, `CONTROLLER_GID` | Non-root image user/group matching writable host paths | + +### YAML settings + +The checked-in `config.yaml` is the canonical example. Important defaults are: + +- MQTT: 30-second keepalive, 1-to-30-second reconnect backoff, 512-entry event + queue, and 2-second QoS 1 PUBACK timeout. +- Storage: 48-hour retention, 60-second expiry scan, and 15-minute uncertainty + review deadline. +- Downloads: two workers, 2-hour subprocess timeout, maximum height 1080, and + optional cookies. +- Router: required mode `video`, 3-second confirmation timeout, unknown-state + policy `attempt`, continue after confirmation failure, and no fallback restore + mode. +- Home Assistant: discovery enabled under `homeassistant`. +- Web: bind `0.0.0.0:7531`. + +All remaining MQTT topic defaults are listed in the topic matrix. Numeric values +must be finite and within their validated ranges. + +### Cookies + +Place an optional Netscape-format cookie file at: + +```text +/yt-dlp-cookies.txt +``` + +Compose exposes it as `/run/secrets/yt-dlp-cookies.txt`. For each yt-dlp command, +Controller makes a mode-0600 temporary copy under `/tmp`, passes that copy to +yt-dlp, and removes it afterward. The source is never copied into the image. + +With `downloads.cookie_required: false` (the default), a missing cookie file is +logged and download continues without cookies. Set it to `true` when silent +cookie omission is unacceptable. Keep the host file out of Git, restrict its +permissions, and rotate it as a credential. + +## HTTP and health + +The HTTP surface is intentionally small: + +- `GET /` renders the playlist submission form. +- `POST /submit` accepts an HTTP(S) URL plus optional submitter and note. +- `GET /healthz` returns service and reconciliation state as JSON. + +Health includes `status`, `started`, `coordinator_alive`, `mqtt_connected`, +`router_state`, `player`, durable `reconciliation_reviews`, connection-scoped +`cleanup_reconciliation`, and `degraded_reasons`. Typical degraded reasons are +`mqtt_disconnected`, `awaiting_player_evidence`, and +`router_state_unavailable`. + +The endpoint returns HTTP 200 whenever the service is started and its coordinator +thread is alive, including degraded MQTT or Router conditions. It returns 503 +when the service is stopped, the coordinator is dead, or health evaluation +fails. This means a broker outage does not make Docker restart an otherwise +healthy Controller. A configuration, storage, or migration failure prevents +startup and therefore cannot produce a healthy endpoint. + +## Docker deployment + +The production image and Compose file intentionally differ from Controller v1: + +- Gunicorn runs one `gthread` worker with four HTTP threads and `preload=false`. + One process must own the singleton coordinator, MQTT client, executor, and + SQLite connection. Do not increase `workers` or enable development hot reload. +- The container runs as configurable non-root UID/GID with all Linux + capabilities dropped and `no-new-privileges` enabled. +- The root filesystem is read-only. Only the data/media bind mounts and bounded + `/tmp` tmpfs are writable. +- The persistent Controller data mount is `/controller-data`; media is mounted + at the same `/mnt/Media/YTPlayer` path used by Player Node. +- Cookies are supplied through the read-only `/run/secrets` bind mount instead + of the old `/cookies_ro.txt` path. +- The image includes ffmpeg and a Node 22 runtime required by the pinned yt-dlp + YouTube JavaScript support. +- The build currently uses `node:22.23.1-bookworm-slim` as a JavaScript-runtime + stage and `python:3.12-slim-bookworm` as the final base. Runtime Python + requirements are exact pins, including Flask 3.1.3, Gunicorn 26.0.0, + paho-mqtt 2.1.0, PyYAML 6.0.3, and yt-dlp 2026.7.4. +- The HTTP health check, 45-second stop grace, bounded JSON logs, init process, + and `unless-stopped` restart policy are explicit. +- `tabernet` remains an external Docker network and must already exist. + +Create a deployment environment file and secret: + +```bash +cp .env.example .env +python3 -c 'import secrets; print(secrets.token_urlsafe(48))' +``` + +Put the generated value in `FLASK_SECRET_KEY`, set every host path, and ensure +`CONTROLLER_UID`/`CONTROLLER_GID` can write the data and media directories. +Then validate and start: + +```bash +docker network inspect tabernet +docker compose config --quiet +docker compose build --pull +docker compose up -d +docker compose ps +docker compose logs --tail=200 controller +curl -fsS http://127.0.0.1:7531/healthz +``` + +## Upgrade from Controller v1 + +Use a maintenance window. Replace the example paths below with the paths from +the existing deployment; the historical Compose deployment commonly used +`/home/media/Scripts/Video-Player` as `/controller-data`. + +1. Record the exact old Git commit or image ID, Compose file, configuration, + environment, UID/GID, and MQTT settings. Keep the old image locally. +2. Stop new submissions and let downloads finish. Prefer to stop playback and + verify authoritative Player terminal/cleared telemetry before the cutover. + If Player must remain active, leave it running; v2 will start conservatively + and protect last-known activity until fresh evidence arrives. +3. Stop only the old Controller. Do not modify, upgrade, or replace Player Node + as part of this migration. +4. While Controller is stopped, take a matched, read-only rollback snapshot of + the old data directory and `/mnt/Media/YTPlayer`. Preserve permissions and + timestamps. Record a SHA-256 of `playlists.yaml`. +5. Create a dedicated v2 data directory. Copy, do not move, the legacy + `playlists.yaml` into its root before first v2 startup. The v2 directory must + not already contain `controller.sqlite3`, or automatic YAML import will not + run. +6. Create the secrets directory and, if needed, install the old cookies in it as + `yt-dlp-cookies.txt`. Do not bake cookies into the image. +7. Copy `.env.example` to `.env`; set a new long `FLASK_SECRET_KEY`, the dedicated + `CONTROLLER_DATA_DIR`, the shared `YTPLAYER_MEDIA_DIR`, secrets directory, + MQTT values, web port, and matching UID/GID. Review `config.yaml`, especially + Router policy and uncertainty deadline. +8. Validate and build the v2 deployment, then start it once. Watch logs for the + migration report or any startup failure. +9. Verify that the original YAML still exists, its full-hash `.pre-v2.*.bak` + exists, and `controller.sqlite3` exists. Confirm the source and backup hashes + match. +10. Check `/healthz`. Resolve or deliberately reconcile any review token; do not + equate a 200/degraded response with current Player or Router evidence. +11. Verify retained slot `title/state/count` values and all thirty HA discovery + IDs. Perform a controlled play/stop through `youtube/controller/cmd`, confirm + Player evidence drives state, and verify Router restoration/manual override + behavior. + +A concrete copy/start sequence, using the common legacy path, is: + +```bash +sudo install -d -o 1000 -g 1000 -m 0750 /srv/video-player-controller/data +sudo install -m 0640 -o 1000 -g 1000 \ + /home/media/Scripts/Video-Player/playlists.yaml \ + /srv/video-player-controller/data/playlists.yaml +sudo install -d -o root -g 1000 -m 0750 /srv/video-player-controller/secrets + +cp .env.example .env +# Edit .env and config.yaml before continuing. + +docker compose config --quiet +docker compose build --pull +docker compose up -d +docker compose logs --tail=200 controller +curl -fsS http://127.0.0.1:7531/healthz + +sha256sum \ + /srv/video-player-controller/data/playlists.yaml \ + /srv/video-player-controller/data/playlists.yaml.pre-v2.*.bak +ls -l /srv/video-player-controller/data/controller.sqlite3 +``` + +If first startup reports migration failure, stop the restart loop, retain the +original YAML and generated hash backup, inspect/fix a copy of the source, and +retry only while no final SQLite database exists. Returning to the recorded v1 +image with the untouched v1 data/media snapshot remains safe because migration +does not modify them. + +## Rollback + +Code rollback and state rollback are different operations. Controller v2 does +not write changes back to `playlists.yaml`; v1 does not understand +`controller.sqlite3`. After any v2 download, delete, expiry, or slot reuse, the +pre-migration YAML and current media tree may no longer agree. + +1. Stop Controller v2 cleanly. Leave Player Node unchanged. +2. Archive the complete v2 data directory, logs, `.env`, `config.yaml`, image ID, + and current media tree. Do not delete the SQLite database, original YAML, or + hash backup. +3. Restore the matched pre-upgrade v1 `playlists.yaml` **and media snapshot as a + pair** into the paths used by the old deployment. Do not point v1 at the live + v2 SQLite file or mix old YAML with media already mutated by v2. +4. Restore the recorded v1 image/commit, Compose file, cookie mount, environment, + and path layout. +5. Start only v1 and verify MQTT connectivity, retained stable-topic consumers, + submissions, and a controlled Player operation. +6. Keep the archived v2 database/media pair. A later roll-forward must restore + those together before starting v2 again. + +If v2 has made no media or inventory mutation, the untouched original YAML and +pre-upgrade media may still be a valid rollback pair. Verify that fact rather +than assuming it. Controller v1 can compact and rename slots during prune or +delete, so starting it against mismatched/current v2 media can immediately undo +the stable-slot guarantee. + +For a migration-only failure before final SQLite installation, simply keep v2 +stopped and restart the recorded v1 deployment against the untouched original +data and media; no v2 database state needs conversion. + +## Development and tests + +Application imports are side-effect free. Constructing the Flask app does not +open SQLite, start MQTT, create directories, or launch threads. `wsgi.py` owns +the production lifecycle; `python app.py` explicitly owns the development +lifecycle. + +Create an isolated environment and run the complete regression suite: + +```bash +python3 -m venv .venv +.venv/bin/python -m pip install --upgrade pip +.venv/bin/python -m pip install --requirement requirements.txt +PYTHONDONTWRITEBYTECODE=1 .venv/bin/python -m unittest discover -s tests -v +``` + +The suite covers configuration and import side effects, atomic reservation and +stale generations, asynchronous deletion/retry/late-guard safety, migration +recovery/idempotence, MQTT startup/reconnect/PUBACK/manual acknowledgement, +Player generation, pathless busy/error handling and pause-timeout reduction, +durable playback plus connection-scoped cleanup reconciliation, Audio Router +timeouts and ownership loss, service integration, HA compatibility, HTTP +health, and Docker/runtime assets. It uses fakes and temporary directories; a +real-broker/container smoke test is still required for a deployment. + +Useful static and container checks are: + +```bash +docker compose config --quiet +docker compose build +docker compose up -d +curl -fsS http://127.0.0.1:7531/healthz +docker compose logs --tail=200 controller +``` + +The authoritative Player Node has its own isolated regression suite. Validate +the sibling without installing dependencies globally or modifying its source: + +```bash +cd ../Video-Player-Node +python3 -m venv .venv +.venv/bin/python -m pip install --upgrade pip +.venv/bin/python -m pip install --requirement requirements.txt +PYTHONDONTWRITEBYTECODE=1 .venv/bin/python -m unittest discover -s tests -v +``` + +## Security and operational limitations + +- The web form is unauthenticated and listens on `0.0.0.0` by default. Restrict + the published port with host firewall/network ACLs or place it behind an + authenticated reverse proxy. URL validation checks only the `http`/`https` + shape; yt-dlp performs the outbound request. +- MQTT username/password support is available, but application-level TLS is not + configured here. Keep this connection on a trusted/protected network or add a + separately managed secure tunnel/proxy, and grant the Controller only the + required publish/subscribe topic ACLs. In particular, restrict + `youtube/controller/cmd` and Router mode commands. +- Administrative reconciliation's `actor` is an audit label, not an identity + proof. Authorization comes from who can publish to the command topic. +- Direct commands to `youtube/player/cmd` bypass Controller orchestration and + can race its Router and deletion guards. Operational publishers should use + `youtube/controller/cmd` exclusively. +- Player command PUBACK is transport acknowledgement, not an application-level + correlated acknowledgement. Controller infers completion from authoritative + status/telemetry and serializes transitions; the Player protocol still has no + request ID or generation echoed in evidence. +- Audio Router's plain set/state topics have no publisher identity or atomic + compare-and-set. Manual override detection is deliberately conservative but + cannot eliminate a perfectly simultaneous same-value race. +- Optional v2 events are non-retained and can be missed while a consumer or MQTT + is offline. SQLite audit rows and durable playback reconciliation records are + the historical sources; the cleanup-only `/healthz` token is deliberately + connection-scoped and must be refetched. Existing consumers must not depend + on the event stream. +- Production is intentionally one Gunicorn worker with no preload or hot reload. + More workers would duplicate the coordinator, MQTT client, and downloader. +- Shutdown actively terminates tracked yt-dlp process groups (TERM, then KILL + after a short bound) rather than waiting for the configured 2-hour command + timeout. The current job fails instead of resuming. A container kill before + that bookkeeping completes is recovered as `interrupted`/`error`; queued + deletion work remains durably `deleting` for the evidence-gated retry path. +- The container is read-only and non-root, but the data and media mounts must be + writable. Controller can intentionally delete canonical expired or + administratively deleted slot directories; maintain independent backups. +- Python dependencies and yt-dlp are exactly pinned, but Docker base images are + tag-pinned rather than digest-pinned and Debian apt packages are not version + pinned. Builds are therefore not bit-for-bit reproducible. Pin verified image + digests and use a controlled package mirror if that is required. +- YouTube changes regularly. Update the explicit `yt-dlp[default,pin]` version + in `requirements.txt`, rebuild the image, and run both regression and download + smoke tests on a deliberate maintenance cadence. Runtime self-update is not + enabled. The copied Node binary must match the target image architecture and + remain supported by yt-dlp. +- SQLite and its `fcntl` process lock are designed for a local persistent + filesystem. Do not place `/controller-data` on storage whose locking or atomic + rename semantics are unreliable. diff --git a/controller/downloads.py b/controller/downloads.py index ad0d8a0..e4bcae9 100644 --- a/controller/downloads.py +++ b/controller/downloads.py @@ -90,6 +90,7 @@ class DownloadManager: self._lock = threading.Lock() self._futures: set[Future[Any]] = set() self._processes: set[subprocess.Popen[str]] = set() + self._deletions_in_flight: set[tuple[int, int]] = set() self._stopping = False def enqueue( @@ -484,21 +485,43 @@ class DownloadManager: slot_directory = canonical_slot_directory(self.config.storage.media_base, token.slot) trash_root = self.config.storage.media_base / ".controller-trash" trash_path: Path | None = None + if self._slot_is_guarded(token.slot): + return False if os.path.lexists(slot_directory): if slot_directory.is_symlink(): slot_directory.unlink() else: trash_root.mkdir(parents=True, exist_ok=True) + self._fsync_directory(self.config.storage.media_base) trash_path = trash_root / ( f"slot{token.slot}.generation{token.generation}.{uuid.uuid4().hex}" ) os.replace(slot_directory, trash_path) self._fsync_directory(self.config.storage.media_base) + self._fsync_directory(trash_root) + if self._slot_is_guarded(token.slot): + if ( + trash_path is not None + and not os.path.lexists(slot_directory) + and os.path.lexists(trash_path) + ): + os.replace(trash_path, slot_directory) + self._fsync_directory(self.config.storage.media_base) + self._fsync_directory(trash_root) + return False if trash_path is not None: shutil.rmtree(trash_path) + self._fsync_directory(trash_root) return self.store.finish_delete(token) + def _slot_is_guarded(self, slot: int) -> bool: + return any( + guard.slot == slot for guard in self.store.list_live_guards() + ) + def expire_due_slots(self) -> list[int]: + """Synchronously expire slots for maintenance callers and tests.""" + deleted: list[int] = [] for token in self.store.begin_expired_deletions(): try: @@ -508,47 +531,164 @@ class DownloadManager: LOG.exception("could not finish expired deletion for slot %s", token.slot) return deleted + def schedule_delete_slot( + self, + slot: int, + *, + reason: str, + expected_url: str | None = None, + ) -> DeleteToken: + """Atomically fence a manual deletion, then finish it off-thread.""" + + token = self.store.begin_delete( + slot, + reason=reason, + expected_url=expected_url, + ) + self._schedule_deletion(token) + return token + + def schedule_expired_deletions(self) -> list[DeleteToken]: + """Fence all currently due slots and queue filesystem work.""" + + tokens = self.store.begin_expired_deletions() + for token in tokens: + self._schedule_deletion(token) + return tokens + def recover_pending_deletions(self) -> list[int]: recovered: list[int] = [] - trash_root = self.config.storage.media_base / ".controller-trash" for token in self.store.pending_deletions(): - if any( - guard.slot == token.slot - for guard in self.store.list_live_guards() - ): - LOG.warning( - "deferring interrupted deletion for protected slot %s", - token.slot, - ) - continue - if trash_root.is_dir(): - prefix = f"slot{token.slot}.generation{token.generation}." - for candidate in trash_root.iterdir(): - if candidate.name.startswith(prefix): - if candidate.is_symlink() or candidate.is_file(): - candidate.unlink() - elif candidate.is_dir(): - shutil.rmtree(candidate) - if self.finish_deletion(token): + if self._recover_deletion(token): recovered.append(token.slot) return recovered def schedule_pending_deletion_recovery(self) -> bool: - """Run recoverable trash cleanup without delaying service startup.""" + """Queue every safe pending deletion without blocking the coordinator.""" with self._lock: if self._stopping: return False + scheduled = False + for token in self.store.pending_deletions(): + scheduled = self._schedule_deletion(token) or scheduled + return scheduled + + def _schedule_deletion(self, token: DeleteToken) -> bool: + key = (token.slot, token.generation) + with self._lock: + if self._stopping or key in self._deletions_in_flight: + return False + self._deletions_in_flight.add(key) try: - future = self._executor.submit(self.recover_pending_deletions) - except BaseException: - LOG.exception("could not schedule interrupted deletion recovery") + future = self._executor.submit(self._deletion_worker, token) + except BaseException as exc: + with self._lock: + self._deletions_in_flight.discard(key) + self._emit( + DownloadEvent( + slot=token.slot, + state="delete_error", + message=f"could not start deletion worker: {exc}", + generation=token.generation, + ) + ) return False with self._lock: self._futures.add(future) future.add_done_callback(self._forget_future) return True + def _deletion_worker(self, token: DeleteToken) -> None: + key = (token.slot, token.generation) + try: + if self._recover_deletion(token): + self._emit( + DownloadEvent( + slot=token.slot, + state="deleted", + generation=token.generation, + ) + ) + elif self.store.get_slot(token.slot).media_state == "deleting": + self._emit( + DownloadEvent( + slot=token.slot, + state="delete_deferred", + message="slot deletion was deferred by playback protection", + generation=token.generation, + ) + ) + except BaseException as exc: + LOG.exception( + "could not finish deletion for slot %s generation %s", + token.slot, + token.generation, + ) + self._emit( + DownloadEvent( + slot=token.slot, + state="delete_error", + message=f"deletion remains pending: {exc}", + generation=token.generation, + ) + ) + finally: + with self._lock: + self._deletions_in_flight.discard(key) + + def _recover_deletion(self, token: DeleteToken) -> bool: + trash_root = self.config.storage.media_base / ".controller-trash" + slot_directory = canonical_slot_directory( + self.config.storage.media_base, + token.slot, + ) + if self._slot_is_guarded(token.slot): + LOG.warning( + "deferring interrupted deletion for protected slot %s", + token.slot, + ) + self._restore_guarded_trash(token, trash_root, slot_directory) + return False + if trash_root.is_dir(): + prefix = f"slot{token.slot}.generation{token.generation}." + for candidate in trash_root.iterdir(): + if not candidate.name.startswith(prefix): + continue + if self._slot_is_guarded(token.slot): + self._restore_guarded_trash( + token, + trash_root, + slot_directory, + ) + return False + if candidate.is_symlink() or candidate.is_file(): + candidate.unlink() + elif candidate.is_dir(): + shutil.rmtree(candidate) + self._fsync_directory(trash_root) + return self.finish_deletion(token) + + def _restore_guarded_trash( + self, + token: DeleteToken, + trash_root: Path, + slot_directory: Path, + ) -> None: + if not trash_root.is_dir() or os.path.lexists(slot_directory): + return + prefix = f"slot{token.slot}.generation{token.generation}." + candidates = sorted( + candidate + for candidate in trash_root.iterdir() + if candidate.name.startswith(prefix) + ) + if not candidates: + return + os.replace(candidates[0], slot_directory) + self._fsync_directory(self.config.storage.media_base) + self._fsync_directory(trash_root) + def shutdown(self, *, wait: bool = True) -> None: with self._lock: self._stopping = True diff --git a/controller/playback.py b/controller/playback.py index 3b6356a..190091a 100644 --- a/controller/playback.py +++ b/controller/playback.py @@ -18,6 +18,7 @@ from .store import ControllerStore TERMINAL_REASONS = frozenset( {"stop", "finished", "mpv_error", "sigterm", "sigint", "shutdown"} ) +TERMINAL_ERROR_REASONS = frozenset({"startup_error", "worker_start_error"}) CONTROL_ACTIONS = frozenset({"play", "stop", "pause", "resume", "toggle"}) @@ -485,13 +486,31 @@ class PlaybackReducer: return effects if state == "busy": - if slot is not None or path is not None: + previous = self.store.player_observation() + if ( + slot is not None + or path is not None + or previous.active_slot is not None + or previous.active_path is not None + ): self._active_evidence( slot=slot, path=path, paused=None, status=payload, ) + else: + self.store.set_player_observation( + availability="online", + phase="unknown", + active_slot=None, + active_path=None, + paused=None, + reconciliation_state="unknown", + reconciliation_due_at=None, + reconciliation_id=None, + status_payload=payload, + ) effects.append(PlaybackEffect("player_busy", slot=slot, message=payload.get("message"))) if self.pending is not None and self.pending.action == "play": effects.append(self._finish_operation("failed", "Player is busy")) @@ -499,40 +518,87 @@ class PlaybackReducer: if state == "error": previous = self.store.player_observation() + message = str(payload.get("message") or "Player error") + reason = payload.get("reason") + if reason in TERMINAL_ERROR_REASONS: + terminal_slot = slot if slot is not None else previous.active_slot + self._idle_evidence(status=payload) + effects.extend( + ( + PlaybackEffect( + "player_error", + slot=terminal_slot, + message=message, + ), + PlaybackEffect( + "player_idle", + slot=terminal_slot, + message=str(reason), + ), + ) + ) + if self.pending is not None: + effects.append(self._finish_operation("failed", message)) + return effects + active_slot = slot if slot is not None else previous.active_slot active_path = path if path is not None else previous.active_path + due = None + if self.config.storage.uncertain_review_after_seconds > 0: + due = ( + self._wall_clock() + + self.config.storage.uncertain_review_after_seconds + ) if active_slot is not None: - self.store.reconcile_active_guard( + guard = self.store.mark_slot_uncertain( active_slot, - kind="player_active", - details={"error": payload.get("message")}, + review_after=due, + details={"reason": "Player error", "message": message}, + ) + due = guard.review_after + self.store.set_player_observation( + availability="online", + phase="error", + active_slot=active_slot, + active_path=active_path, + paused=( + previous.paused + if self._same_playback_identity( + previous, + active_slot, + active_path, + ) + else None + ), + reconciliation_state=( + "operator_review" + if guard.state == "review_required" + else "awaiting_evidence" + ), + reconciliation_due_at=due, + reconciliation_id=guard.guard_id, + status_payload=payload, ) elif active_path is not None: - self.store.reconcile_external_active() - self.store.set_player_observation( - availability="online", - phase="error", - active_slot=active_slot, - active_path=active_path, - paused=( - previous.paused - if self._same_playback_identity( - previous, - active_slot, - active_path, - ) - else None - ), - reconciliation_state=( - "confirmed" - if active_slot is not None or active_path is not None - else "unknown" - ), - reconciliation_due_at=None, - reconciliation_id=None, - status_payload=payload, - ) - message = str(payload.get("message") or "Player error") + self.store.mark_external_uncertain( + active_path, + review_after=due, + availability="online", + details={"reason": "Player error", "message": message}, + status_payload=payload, + ) + else: + self.store.set_player_observation( + availability="online", + phase="unknown", + active_slot=None, + active_path=None, + paused=None, + reconciliation_state="unknown", + reconciliation_due_at=None, + reconciliation_id=None, + status_payload=payload, + ) effects.append(PlaybackEffect("player_error", slot=active_slot, message=message)) if self.pending is not None: effects.append(self._finish_operation("failed", message)) @@ -645,6 +711,19 @@ class PlaybackReducer: "reason": "stop timeout", }, ) + elif operation.action in {"pause", "resume", "toggle"}: + previous = self.store.player_observation() + if previous.active_slot is not None or previous.active_path is not None: + self.store.set_player_observation( + availability=previous.availability, + phase="unknown", + active_slot=previous.active_slot, + active_path=previous.active_path, + paused=None, + reconciliation_state=previous.reconciliation_state, + reconciliation_due_at=previous.reconciliation_due_at, + reconciliation_id=previous.reconciliation_id, + ) return self._finish_operation("timed_out", "no authoritative Player confirmation") def _mark_play_uncertain(self, operation: PlayerOperation, reason: str) -> None: @@ -753,8 +832,6 @@ class PlaybackReducer: def projected_slot_state(self, slot: int) -> str: record = self.store.get_slot(slot) - if record.media_state != "ready": - return record.media_state observation = self.store.player_observation() if observation.active_slot == slot: if observation.reconciliation_state == "operator_review": @@ -771,9 +848,9 @@ class PlaybackReducer: ) if guard is not None: if guard.kind == "pending_play": - return "ready" + return record.media_state return "operator_review" if guard.state == "review_required" else "unknown" - return "ready" + return record.media_state def mark_due_reconciliation(self) -> list[PlaybackEffect]: now = self._wall_clock() diff --git a/controller/service.py b/controller/service.py index d8bf238..0e2768a 100644 --- a/controller/service.py +++ b/controller/service.py @@ -99,7 +99,7 @@ class ControllerService: self._processed_packets: dict[tuple[int, str, int], float] = {} self._player_evidence_ready = False self._operator_cleanup_override = False - self._recovery_scheduled = False + self._cleanup_review_id: str | None = None @property def started(self) -> bool: @@ -149,9 +149,7 @@ class ControllerService: ) # Cleanup is deliberately deferred until fresh Player evidence # (or an explicit administrative reconciliation) makes it safe. - self._player_evidence_ready = False - self._operator_cleanup_override = False - self._recovery_scheduled = False + self._reset_cleanup_evidence() self._stop_event.clear() self._next_expiry = ( self._monotonic() @@ -271,8 +269,13 @@ class ControllerService: guards = self.store.list_live_guards() if started else [] player_operation_pending = bool( started - and self.playback is not None - and self.playback.pending is not None + and ( + self._pending_play is not None + or ( + self.playback is not None + and self.playback.pending is not None + ) + ) ) mqtt_connected = bool(started and self.mqtt.connected) degraded: list[str] = [] @@ -321,6 +324,22 @@ class ControllerService: "clearable": not player_operation_pending, } ) + cleanup_reconciliation = None + if ( + started + and not self._cleanup_allowed() + and self._cleanup_review_id is not None + and not guards + and player_observation is not None + and player_observation.active_slot is None + and player_observation.active_path is None + ): + cleanup_reconciliation = { + "review_id": self._cleanup_review_id, + "kind": "player_evidence", + "state": "awaiting_evidence", + "clearable": not player_operation_pending, + } return { "status": ( "stopped" @@ -346,6 +365,7 @@ class ControllerService: } ), "reconciliation_reviews": reviews, + "cleanup_reconciliation": cleanup_reconciliation, "degraded_reasons": degraded, } @@ -416,16 +436,14 @@ class ControllerService: handled_successfully = True return if event.kind == "connected": - self._player_evidence_ready = False - self._operator_cleanup_override = False + self._reset_cleanup_evidence() self._announce_discovery() self._sync_slot_states(force=True) self._announce_reconciliation_reviews() handled_successfully = True return if event.kind == "disconnected": - self._player_evidence_ready = False - self._operator_cleanup_override = False + self._reset_cleanup_evidence() assert self.playback is not None assert self.router is not None playback_effect = self.playback.mark_transport_uncertain( @@ -441,30 +459,41 @@ class ControllerService: handled_successfully = True return if event.topic == self.config.topics.player_status: - payload = self._decode_json(event.payload, "Player status") - if payload is not None: - assert self.playback is not None - before_generation = self.store.player_observation().generation - self._handle_playback_effects(self.playback.handle_status(payload)) - self._refresh_player_evidence(before_generation) + if event.retain: + self._publish_event( + "retained_player_evidence_rejected", + "warning", + "Retained Player status was ignored as historical evidence", + ) + else: + payload = self._decode_json(event.payload, "Player status") + if payload is not None: + assert self.playback is not None + before_generation = self.store.player_observation().generation + effects = self.playback.handle_status(payload) + self._handle_playback_effects(effects) + self._refresh_player_evidence(before_generation, effects) elif event.topic == self.config.topics.player_telemetry: - payload = self._decode_json(event.payload, "Player telemetry") - if payload is not None: - assert self.playback is not None - before_generation = self.store.player_observation().generation - self._handle_playback_effects(self.playback.handle_telemetry(payload)) - self._refresh_player_evidence(before_generation) + if event.retain: + self._publish_event( + "retained_player_evidence_rejected", + "warning", + "Retained Player telemetry was ignored as historical evidence", + ) + else: + payload = self._decode_json(event.payload, "Player telemetry") + if payload is not None: + assert self.playback is not None + before_generation = self.store.player_observation().generation + effects = self.playback.handle_telemetry(payload) + self._handle_playback_effects(effects) + self._refresh_player_evidence(before_generation, effects) elif event.topic == self.config.topics.player_state: - try: - flat_state = event.payload.decode("utf-8").strip() - except UnicodeDecodeError: - flat_state = "" - # Retained flat idle is sufficient only as a quiescence gate; - # it never mutates detailed playback state or clears guards. - self._player_evidence_ready = flat_state == "idle" - self._operator_cleanup_override = False - if self._player_evidence_ready: - self._schedule_deletion_recovery() + # The Player owns this retained Companion topic. Its value is + # historical across outages and reconnect deliberately emits + # flat ``idle`` even while MPV is active, so it is never fresh + # cleanup authority. + pass elif event.topic == self.config.topics.controller_command: if event.retain: self._publish_event( @@ -511,8 +540,9 @@ class ControllerService: else: decision = self.router.observe_mode(mode) self._handle_router_decision(decision) - # Flat state is only a cold-start cleanup gate. Detailed playback - # truth continues to come exclusively from status and telemetry. + # The flat state subscription is compatibility observation only. + # Playback and cleanup truth come exclusively from detailed status + # and telemetry or an explicit administrative reconciliation. handled_successfully = True finally: if handled_successfully: @@ -873,17 +903,20 @@ class ControllerService: expected_url = payload.get("url") if expected_url is not None and not isinstance(expected_url, str): raise ValueError("delete URL guard must be a string") - self.downloads.delete_slot( + self.downloads.schedule_delete_slot( slot, reason="manual", expected_url=expected_url, ) self._announce_slot(slot, force=True) - return {"status": "deleted", "slot": slot} + return {"status": "deleting", "slot": slot} def _request_reconciliation(self, payload: Mapping[str, Any]) -> dict[str, Any]: if payload.get("decision") != "clear" or payload.get("confirm") is not True: raise ValueError("reconciliation clear requires decision=clear and confirm=true") + scope = payload.get("scope") + if scope not in {None, "cleanup"}: + raise ValueError("reconciliation scope must be cleanup when provided") slot = ( self._parse_slot(payload.get("slot")) if payload.get("slot") is not None @@ -899,7 +932,36 @@ class ControllerService: raise ValueError( "playback reconciliation cannot be cleared while a Player operation is pending" ) - if slot is None: + if scope == "cleanup": + if slot is not None: + raise ValueError("cleanup reconciliation does not accept a slot") + observation = self.store.player_observation() + if ( + self._cleanup_allowed() + or self._cleanup_review_id is None + or review_id.strip() != self._cleanup_review_id + ): + raise StaleOperationError("cleanup reconciliation review_id is stale") + if ( + self.store.list_live_guards() + or observation.active_slot is not None + or observation.active_path is not None + ): + raise SlotProtectedError( + "cleanup reconciliation cannot bypass known playback protection" + ) + self.store.record_audit_event( + "player_evidence_admin_override", + actor=actor.strip(), + details={ + "review_id": review_id.strip(), + "reason": reason.strip(), + }, + ) + self._operator_cleanup_override = True + self._cleanup_review_id = None + cleared = True + elif slot is None: cleared = self.store.clear_external_reconciliation( review_id.strip(), actor=actor.strip(), @@ -933,7 +995,9 @@ class ControllerService: and observation.active_slot is None and observation.active_path is None ): - self._operator_cleanup_override = True + if scope != "cleanup": + self._operator_cleanup_override = True + self._cleanup_review_id = None self._schedule_deletion_recovery() assert self.router is not None self.router.abandon( @@ -954,6 +1018,8 @@ class ControllerService: "status": "cleared", "review_id": review_id.strip(), } + if scope == "cleanup": + result["scope"] = "cleanup" if slot is not None: result["slot"] = slot return result @@ -996,7 +1062,11 @@ class ControllerService: no_active_player = ( observation.active_slot is None and observation.active_path is None ) - if terminal or (failed_play and no_active_player): + unidentified_busy = ( + no_active_player + and any(effect.kind == "player_busy" for effect in effects) + ) + if terminal or (failed_play and no_active_player and not unidentified_busy): self._handle_router_decision(self.router.release(self._session_lease_id)) self._session_lease_id = None for effect in effects: @@ -1028,6 +1098,27 @@ class ControllerService: self._sync_slot_states() def _handle_download_event(self, event: DownloadEvent) -> None: + if event.state == "deleted": + self._announce_slot(event.slot, force=True) + return + if event.state == "delete_deferred": + self._announce_slot(event.slot, force=True) + return + if event.state == "delete_error": + message = event.message or "slot deletion remains pending" + self._publish_event( + "slot_deletion_pending", + "warning", + message, + slot=event.slot, + ) + self._announce_slot(event.slot, force=True) + retry_delay = max( + 1.0, + min(30.0, self.config.storage.expiry_interval_seconds), + ) + self._next_expiry = self._monotonic() + retry_delay + return payload: dict[str, Any] = {"slot": event.slot, "state": event.state} for key in ("title", "count", "message", "url", "submitter", "note"): value = getattr(event, key) @@ -1171,21 +1262,55 @@ class ControllerService: return self._player_evidence_ready or self._operator_cleanup_override def _schedule_deletion_recovery(self) -> None: - if self._recovery_scheduled or not self._cleanup_allowed(): + if not self._cleanup_allowed(): return assert self.downloads is not None - self._recovery_scheduled = ( - self.downloads.schedule_pending_deletion_recovery() - ) + self.downloads.schedule_pending_deletion_recovery() - def _refresh_player_evidence(self, before_generation: int) -> None: + def _refresh_player_evidence( + self, + before_generation: int, + effects: list[PlaybackEffect], + ) -> None: observation = self.store.player_observation() if observation.generation == before_generation: return - self._operator_cleanup_override = False - self._player_evidence_ready = observation.availability == "online" - if self._player_evidence_ready and observation.phase == "idle": + kinds = {effect.kind for effect in effects} + identity_known = ( + observation.active_slot is not None + or observation.active_path is not None + ) + safe_evidence = "player_idle" in kinds or ( + identity_known + and bool( + kinds + & { + "player_active", + "player_busy", + "player_reconnected_active", + "player_telemetry", + "control_issued", + } + ) + ) + if safe_evidence and observation.availability == "online": + self._player_evidence_ready = True + self._operator_cleanup_override = False + self._cleanup_review_id = None self._schedule_deletion_recovery() + return + self._mark_cleanup_unknown() + + def _reset_cleanup_evidence(self) -> None: + self._player_evidence_ready = False + self._operator_cleanup_override = False + self._cleanup_review_id = uuid.uuid4().hex + + def _mark_cleanup_unknown(self) -> None: + self._player_evidence_ready = False + self._operator_cleanup_override = False + if self._cleanup_review_id is None: + self._cleanup_review_id = uuid.uuid4().hex def _run_timers(self) -> None: assert self.playback is not None @@ -1203,6 +1328,7 @@ class ControllerService: self._handle_playback_effects(self.playback.mark_due_reconciliation()) self._next_reconciliation = now + 1.0 if now >= self._next_expiry and self._cleanup_allowed(): - for slot in self.downloads.expire_due_slots(): - self._announce_slot(slot, force=True) + self._schedule_deletion_recovery() + for token in self.downloads.schedule_expired_deletions(): + self._announce_slot(token.slot, force=True) self._next_expiry = now + self.config.storage.expiry_interval_seconds diff --git a/controller/store.py b/controller/store.py index e315b20..6d89bb1 100644 --- a/controller/store.py +++ b/controller/store.py @@ -704,7 +704,16 @@ class ControllerStore: title = str(legacy.get("title") or "") count = _coerce_count(legacy.get("item_count", 0), slot, warnings) created_at = _utc_timestamp(legacy.get("created_at"), f"slot {slot} created_at", warnings) - expires_at = _utc_timestamp(legacy.get("expires_at"), f"slot {slot} expires_at", warnings) + expires_at = _utc_timestamp( + legacy.get("expires_at"), + f"slot {slot} expires_at", + warnings, + ) + if media_state == "error" and expires_at is None: + expires_at = now + self.config.retention_seconds + warnings.append( + f"slot {slot}: missing error expiry defaulted to retention window" + ) generation = 1 connection.execute( """ @@ -816,10 +825,17 @@ class ControllerStore: connection.execute( """ UPDATE slots - SET media_state = 'error', last_error = ?, updated_at = ? + SET media_state = 'error', last_error = ?, + expires_at = COALESCE(expires_at, ?), updated_at = ? WHERE slot_number = ? AND generation = ? """, - (message, now, row["slot_number"], row["generation"]), + ( + message, + now + self.config.retention_seconds, + now, + row["slot_number"], + row["generation"], + ), ) def list_slots(self) -> list[SlotRecord]: @@ -1060,10 +1076,17 @@ class ControllerStore: connection.execute( """ UPDATE slots - SET media_state = 'error', last_error = ?, updated_at = ? + SET media_state = 'error', last_error = ?, + expires_at = COALESCE(expires_at, ?), updated_at = ? WHERE slot_number = ? AND generation = ? """, - (error, now, slot, generation), + ( + error, + now + self.config.retention_seconds, + now, + slot, + generation, + ), ) return True diff --git a/tests/test_downloads.py b/tests/test_downloads.py index feb7ecb..5c04eb9 100644 --- a/tests/test_downloads.py +++ b/tests/test_downloads.py @@ -24,6 +24,24 @@ class ImmediateExecutor: return future +class QueuedExecutor: + def __init__(self): + self.calls = [] + + def submit(self, function, *args, **kwargs): + future = Future() + self.calls.append((future, function, args, kwargs)) + return future + + def run_next(self): + future, function, args, kwargs = self.calls.pop(0) + try: + future.set_result(function(*args, **kwargs)) + except BaseException as exc: + future.set_exception(exc) + return future + + class FakeYtDlp: def __init__( self, @@ -267,6 +285,155 @@ class DownloadManagerTests(unittest.TestCase): self.assertEqual(manager.recover_pending_deletions(), [1]) self.assertFalse(slot.exists()) + def test_async_deletion_failure_remains_durable_and_is_retryable(self): + reservation = self.store.reserve_download( + "https://example.test/retry-delete", + None, + None, + job_id="retry-delete", + ) + self.store.fail_download( + "retry-delete", + 1, + reservation.generation, + "failed", + ) + slot = self.media / "slot1" + slot.mkdir(parents=True) + (slot / "retry.mp4").write_bytes(b"retry") + events = [] + manager = self.manager(FakeYtDlp(self.media), events) + original_recover = manager._recover_deletion + attempts = 0 + + def fail_once(token): + nonlocal attempts + attempts += 1 + if attempts == 1: + raise OSError("transient media failure") + return original_recover(token) + + manager._recover_deletion = fail_once + token = manager.schedule_delete_slot(1, reason="manual") + + self.assertEqual(token.slot, 1) + self.assertEqual(self.store.get_slot(1).media_state, "deleting") + self.assertEqual(events[-1].state, "delete_error") + self.assertEqual(len(self.store.pending_deletions()), 1) + + self.assertTrue(manager.schedule_pending_deletion_recovery()) + self.assertEqual(attempts, 2) + self.assertEqual(events[-1].state, "deleted") + self.assertEqual(self.store.get_slot(1).media_state, "empty") + self.assertFalse(slot.exists()) + + def test_async_pending_recovery_retries_after_guard_is_cleared(self): + reservation = self.store.reserve_download( + "https://example.test/deferred-async-delete", + None, + None, + job_id="deferred-async-delete", + ) + self.store.fail_download( + "deferred-async-delete", + 1, + reservation.generation, + "failed", + ) + self.store.begin_delete(1, reason="manual") + slot = self.media / "slot1" + slot.mkdir(parents=True) + (slot / "active.mp4").write_bytes(b"active") + self.store.protect_slot(1, "player_active", guard_id="active-player") + events = [] + manager = self.manager(FakeYtDlp(self.media), events) + + self.assertTrue(manager.schedule_pending_deletion_recovery()) + self.assertEqual(events[-1].state, "delete_deferred") + self.assertEqual(self.store.get_slot(1).media_state, "deleting") + self.assertTrue(slot.exists()) + + self.store.clear_playback_guards(reason="fresh terminal idle evidence") + self.assertTrue(manager.schedule_pending_deletion_recovery()) + self.assertEqual(events[-1].state, "deleted") + self.assertEqual(self.store.get_slot(1).media_state, "empty") + self.assertFalse(slot.exists()) + + def test_guarded_recovery_restores_crash_staged_trash_to_canonical_slot(self): + reservation = self.store.reserve_download( + "https://example.test/restore-trash", + None, + None, + job_id="restore-trash", + ) + self.store.fail_download( + "restore-trash", + 1, + reservation.generation, + "failed", + ) + token = self.store.begin_delete(1, reason="expired") + trash = ( + self.media + / ".controller-trash" + / f"slot1.generation{token.generation}.crash" + ) + trash.mkdir(parents=True) + (trash / "still-active.mp4").write_bytes(b"active") + self.store.protect_slot(1, "player_active", guard_id="fresh-active") + events = [] + manager = self.manager(FakeYtDlp(self.media), events) + + self.assertTrue(manager.schedule_pending_deletion_recovery()) + + canonical = self.media / "slot1" + self.assertEqual(events[-1].state, "delete_deferred") + self.assertTrue((canonical / "still-active.mp4").exists()) + self.assertFalse(trash.exists()) + self.assertEqual(self.store.get_slot(1).media_state, "deleting") + + def test_queued_deletion_rechecks_guard_before_filesystem_mutation(self): + reservation = self.store.reserve_download( + "https://example.test/queued-delete-guard", + None, + None, + job_id="queued-delete-guard", + ) + self.store.fail_download( + "queued-delete-guard", + 1, + reservation.generation, + "failed", + ) + self.store.begin_delete(1, reason="manual") + slot = self.media / "slot1" + slot.mkdir(parents=True) + (slot / "must-survive.mp4").write_bytes(b"active") + events = [] + executor = QueuedExecutor() + manager = DownloadManager( + self.config, + self.store, + runner=FakeYtDlp(self.media), + executor=executor, + on_event=events.append, + ) + + self.assertTrue(manager.schedule_pending_deletion_recovery()) + self.store.protect_slot(1, "player_active", guard_id="late-player") + executor.run_next() + + self.assertEqual(events[-1].state, "delete_deferred") + self.assertEqual(self.store.get_slot(1).media_state, "deleting") + self.assertTrue((slot / "must-survive.mp4").exists()) + + self.store.clear_playback_guards(reason="fresh terminal idle evidence") + self.assertTrue(manager.schedule_pending_deletion_recovery()) + executor.run_next() + self.assertEqual(events[-1].state, "deleted") + self.assertEqual(self.store.get_slot(1).media_state, "empty") + self.assertFalse(slot.exists()) + def test_shutdown_terminates_tracked_process_without_waiting_for_job_timeout(self): manager = DownloadManager(self.config, self.store) errors = [] diff --git a/tests/test_playback.py b/tests/test_playback.py index f9d66fd..c9d140a 100644 --- a/tests/test_playback.py +++ b/tests/test_playback.py @@ -156,8 +156,6 @@ class PlaybackReducerTests(unittest.TestCase): effects = self.reducer.handle_status( { "state": "busy", - "slot": 1, - "path": "/mnt/Media/YTPlayer/slot1/playlist.m3u", "message": "mpv already playing; stop first", } ) @@ -166,6 +164,25 @@ class PlaybackReducerTests(unittest.TestCase): self.assertEqual(self.reducer.projected_slot_state(1), "playing") self.assertEqual(self.reducer.projected_slot_state(2), "ready") + def test_real_pathless_busy_without_prior_identity_becomes_unknown(self): + self.reducer.begin_operation( + "play", + slot=2, + expected_path="/mnt/Media/YTPlayer/slot2/playlist.m3u", + ) + effects = self.reducer.handle_status( + { + "state": "busy", + "message": "mpv already playing; stop first", + } + ) + + self.assertIn("operation_failed", [effect.kind for effect in effects]) + observation = self.store.player_observation() + self.assertEqual(observation.phase, "unknown") + self.assertIsNone(observation.active_slot) + self.assertIsNone(observation.active_path) + def test_pause_status_does_not_override_actual_telemetry(self): path = "/mnt/Media/YTPlayer/slot1/playlist.m3u" self.reducer.handle_telemetry( @@ -206,9 +223,91 @@ class PlaybackReducerTests(unittest.TestCase): {"state": "error", "slot": 1, "path": path, "message": "cannot stop"} ) self.assertEqual(self.store.player_observation().phase, "error") - self.assertEqual(self.store.list_live_guards()[0].slot, 1) + guard = self.store.list_live_guards()[0] + self.assertEqual(guard.slot, 1) + self.assertEqual(guard.kind, "restart_uncertain") self.assertEqual(self.reducer.projected_slot_state(1), "unknown") + def test_pathless_error_converts_previous_active_slot_to_reviewable_unknown(self): + path = "/mnt/Media/YTPlayer/slot1/playlist.m3u" + self.reducer.handle_status({"state": "playing", "slot": 1, "path": path}) + self.reducer.handle_status( + {"state": "error", "message": "nothing playing to pause"} + ) + + observation = self.store.player_observation() + guard = self.store.list_live_guards()[0] + self.assertEqual(observation.active_slot, 1) + self.assertEqual(observation.reconciliation_id, guard.guard_id) + self.assertEqual(observation.reconciliation_state, "awaiting_evidence") + self.assertEqual(guard.kind, "restart_uncertain") + + def test_terminal_startup_error_clears_identity_even_if_clear_telemetry_was_lost(self): + path = "/mnt/Media/YTPlayer/slot1/playlist.m3u" + self.reducer.begin_operation("play", slot=1, expected_path=path) + effects = self.reducer.handle_status( + { + "state": "error", + "slot": 1, + "path": path, + "reason": "startup_error", + "message": "failed to start mpv", + } + ) + + self.assertIn("player_idle", [effect.kind for effect in effects]) + self.assertIn("operation_failed", [effect.kind for effect in effects]) + observation = self.store.player_observation() + self.assertEqual(observation.phase, "idle") + self.assertIsNone(observation.active_slot) + self.assertIsNone(observation.active_path) + self.assertEqual(self.store.list_live_guards(), []) + + def test_terminal_worker_error_after_cleared_telemetry_remains_idle(self): + path = "/mnt/Media/YTPlayer/slot1/playlist.m3u" + self.reducer.begin_operation("play", slot=1, expected_path=path) + self.reducer.handle_telemetry( + {"state": "telemetry", "slot": 1, "path": None, "paused": False} + ) + effects = self.reducer.handle_status( + { + "state": "error", + "slot": 1, + "path": path, + "reason": "worker_start_error", + "message": "failed to start playback worker", + } + ) + + self.assertIn("player_idle", [effect.kind for effect in effects]) + observation = self.store.player_observation() + self.assertEqual(observation.phase, "idle") + self.assertIsNone(observation.active_path) + self.assertEqual(self.store.list_live_guards(), []) + + def test_player_state_takes_precedence_over_deferred_storage_deletion(self): + reservation = self.store.reserve_download( + "https://example.test/late-player", + None, + None, + job_id="late-player", + ) + self.store.fail_download( + "late-player", + 1, + reservation.generation, + "failed", + ) + self.store.begin_delete(1, reason="expired") + path = "/mnt/Media/YTPlayer/slot1/playlist.m3u" + + self.reducer.handle_status( + {"state": "playing", "slot": 1, "path": path} + ) + + self.assertEqual(self.store.get_slot(1).media_state, "deleting") + self.assertEqual(self.reducer.projected_slot_state(1), "playing") + def test_play_timeout_becomes_unknown_then_operator_review_without_unprotecting(self): self.reducer.begin_operation( "play", @@ -225,6 +324,39 @@ class PlaybackReducerTests(unittest.TestCase): self.assertEqual(self.reducer.projected_slot_state(1), "operator_review") self.assertEqual(self.store.list_live_guards()[0].state, "review_required") + def test_pause_timeout_invalidates_stale_playing_projection(self): + path = "/mnt/Media/YTPlayer/slot1/playlist.m3u" + self.reducer.handle_telemetry( + {"state": "telemetry", "slot": 1, "path": path, "paused": False} + ) + self.reducer.begin_operation("pause", slot=1, expected_path=path) + self.monotonic.value += 6 + + effect = self.reducer.check_timeout() + + self.assertEqual(effect.kind, "operation_timed_out") + observation = self.store.player_observation() + self.assertEqual(observation.phase, "unknown") + self.assertIsNone(observation.paused) + self.assertEqual(self.reducer.projected_slot_state(1), "unknown") + self.assertEqual(self.store.list_live_guards()[0].kind, "player_active") + + def test_resume_timeout_invalidates_stale_paused_projection(self): + path = "/mnt/Media/YTPlayer/slot1/playlist.m3u" + self.reducer.handle_telemetry( + {"state": "telemetry", "slot": 1, "path": path, "paused": True} + ) + self.reducer.begin_operation("resume", slot=1, expected_path=path) + self.monotonic.value += 6 + + effect = self.reducer.check_timeout() + + self.assertEqual(effect.kind, "operation_timed_out") + observation = self.store.player_observation() + self.assertEqual(observation.phase, "unknown") + self.assertIsNone(observation.paused) + self.assertEqual(self.reducer.projected_slot_state(1), "unknown") + def test_malformed_and_unknown_statuses_are_isolated(self): for payload in (None, [], {}, {"state": "mystery"}): effects = self.reducer.handle_status(payload) diff --git a/tests/test_service.py b/tests/test_service.py index f44e1b4..462f7f3 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -14,7 +14,7 @@ from controller.config import ( StorageConfig, ) from controller.downloads import canonical_playlist_path -from controller.models import SlotProtectedError +from controller.models import SlotProtectedError, StaleOperationError from controller.mqtt_adapter import MqttEvent, PublishReceipt from controller.service import ControllerService, ControllerUnavailableError from controller.store import ControllerStore @@ -81,11 +81,13 @@ class FakeMqttAdapter: class FakeDownloadManager: - def __init__(self): + def __init__(self, store=None): + self.store = store self.recovered = False self.recovery_calls = 0 self.expiry_calls = 0 self.stopped = False + self.delete_requests = [] def recover_pending_deletions(self): self.recovered = True @@ -95,10 +97,20 @@ class FakeDownloadManager: self.recovery_calls += 1 return True - def expire_due_slots(self): + def schedule_expired_deletions(self): self.expiry_calls += 1 return [] + def schedule_delete_slot(self, slot, *, reason, expected_url=None): + self.delete_requests.append((slot, reason, expected_url)) + if self.store is None: + return None + return self.store.begin_delete( + slot, + reason=reason, + expected_url=expected_url, + ) + def shutdown(self, *, wait): self.stopped = wait @@ -142,7 +154,7 @@ class ControllerServiceTests(unittest.TestCase): self.events = queue.Queue() self.store = ControllerStore(config.storage, now=self.wall) self.mqtt = mqtt or FakeMqttAdapter() - self.downloads = FakeDownloadManager() + self.downloads = FakeDownloadManager(self.store) self.service = ControllerService( config, store=self.store, @@ -395,6 +407,61 @@ class ControllerServiceTests(unittest.TestCase): self.assertEqual((qos, retain), (1, False)) self.assertEqual(self.mqtt.acknowledged[0].message_id, 42) + def test_retained_detailed_player_evidence_cannot_clear_restart_protection(self): + self.start_service() + self.store.protect_slot( + 1, + "restart_uncertain", + guard_id="restart-review", + ) + + self.send_message( + self.config.topics.player_status, + {"state": "idle", "slot": 1, "reason": "stop"}, + qos=1, + retain=True, + message_id=41, + ) + self.send_message( + self.config.topics.player_telemetry, + { + "state": "telemetry", + "slot": 1, + "path": None, + "paused": False, + }, + retain=True, + ) + self.wait_for( + lambda: any(event.message_id == 41 for event in self.mqtt.acknowledged), + "retained QoS 1 status was not deliberately acknowledged", + ) + self.wait_for( + lambda: len( + [ + event + for event in self.result_events() + if event["type"] == "retained_player_evidence_rejected" + ] + ) + == 2, + "retained telemetry rejection was not reported", + ) + + guards = self.store.list_live_guards() + self.assertEqual(len(guards), 1) + self.assertEqual(guards[0].guard_id, "restart-review") + self.assertIn( + "awaiting_player_evidence", + self.service.health()["degraded_reasons"], + ) + retained_warnings = [ + event + for event in self.result_events() + if event["type"] == "retained_player_evidence_rejected" + ] + self.assertEqual(len(retained_warnings), 2) + def test_optional_result_topic_can_be_disabled_without_affecting_ack(self): config = replace( self.config, @@ -447,19 +514,106 @@ class ControllerServiceTests(unittest.TestCase): "idle", retain=True, ) + time.sleep(0.2) + self.assertFalse(self.downloads.recovered) + self.assertEqual(self.downloads.expiry_calls, 0) + self.assertIn( + "awaiting_player_evidence", + self.service.health()["degraded_reasons"], + ) + + self.send_message( + self.config.topics.player_status, + {"state": "idle", "message": "player online"}, + qos=1, + ) self.wait_for( lambda: self.downloads.recovered, - "retained Player idle did not release deferred recovery", + "detailed Player idle did not release deferred recovery", ) self.wait_for( lambda: self.downloads.expiry_calls > 0, - "Player idle did not release deferred expiry", + "detailed Player idle did not release deferred expiry", ) self.assertNotIn( "awaiting_player_evidence", self.service.health()["degraded_reasons"], ) + def test_administrator_can_explicitly_unlock_cleanup_without_player_evidence(self): + self.start_service() + review = self.service.health()["cleanup_reconciliation"] + self.assertIsNotNone(review) + + result = self.service.submit_control( + { + "action": "reconcile_playback", + "scope": "cleanup", + "decision": "clear", + "confirm": True, + "review_id": review["review_id"], + "actor": "operator", + "reason": "Player appliance was independently verified offline", + } + ) + + self.assertEqual( + result, + { + "status": "cleared", + "review_id": review["review_id"], + "scope": "cleanup", + }, + ) + self.wait_for( + lambda: self.downloads.recovered, + "administrative cleanup override did not schedule recovery", + ) + health = self.service.health() + self.assertIsNone(health["cleanup_reconciliation"]) + self.assertNotIn("awaiting_player_evidence", health["degraded_reasons"]) + + def test_cleanup_reconciliation_token_rotates_and_cannot_bypass_a_guard(self): + self.start_service() + first = self.service.health()["cleanup_reconciliation"]["review_id"] + self.events.put(MqttEvent(kind="connected", connection_generation=1)) + self.wait_for( + lambda: ( + self.service.health()["cleanup_reconciliation"] is not None + and self.service.health()["cleanup_reconciliation"]["review_id"] + != first + ), + "cleanup reconciliation token did not rotate on MQTT connect", + ) + current = self.service.health()["cleanup_reconciliation"]["review_id"] + + with self.assertRaisesRegex(StaleOperationError, "review_id is stale"): + self.service.submit_control( + { + "action": "reconcile_playback", + "scope": "cleanup", + "decision": "clear", + "confirm": True, + "review_id": first, + "actor": "operator", + "reason": "stale request", + } + ) + + self.store.protect_slot(1, "restart_uncertain", guard_id="late-guard") + with self.assertRaisesRegex(SlotProtectedError, "known playback protection"): + self.service.submit_control( + { + "action": "reconcile_playback", + "scope": "cleanup", + "decision": "clear", + "confirm": True, + "review_id": current, + "actor": "operator", + "reason": "must not bypass a guard", + } + ) + def test_manual_delete_requires_explicit_override_while_player_is_unknown(self): self.start_service() self.ready_slot(1) @@ -536,8 +690,6 @@ class ControllerServiceTests(unittest.TestCase): config.topics.player_status, { "state": "busy", - "slot": 2, - "path": str(second_playlist), "message": "mpv already playing; stop first", }, qos=1, @@ -546,12 +698,34 @@ class ControllerServiceTests(unittest.TestCase): lambda: self.service.playback.pending is None, "busy response was not reduced", ) - self.assertEqual(self.store.player_observation().active_slot, 2) + observation = self.store.player_observation() + self.assertIsNone(observation.active_slot) + self.assertIsNone(observation.active_path) + self.assertEqual(observation.phase, "unknown") + self.assertIn( + "awaiting_player_evidence", + self.service.health()["degraded_reasons"], + ) self.assertEqual(self.mqtt.router_modes, ["video"]) lease = self.store.live_audio_lease() self.assertIsNotNone(lease) self.assertEqual(lease.state, "held") + self.send_message( + config.topics.player_status, + { + "state": "playing", + "slot": 2, + "path": str(second_playlist), + }, + qos=1, + ) + self.wait_for( + lambda: self.store.player_observation().active_slot == 2, + "Player did not reassert its actual playback identity", + ) + self.assertEqual(self.store.live_audio_lease().state, "held") + def test_disconnected_synchronous_play_failure_is_not_reported_pending(self): self.start_service(mqtt=FakeMqttAdapter(connected=False)) self.ready_slot(1) @@ -735,6 +909,24 @@ class ControllerServiceTests(unittest.TestCase): self.assertEqual(observation.reconciliation_id, review["review_id"]) self.assertEqual(observation.active_path, str(playlist)) + def test_router_gated_play_marks_all_reconciliation_reviews_not_clearable(self): + self.start_service() + self.ready_slot(1) + self.store.set_router_observation(mode="service", freshness="current") + result = self.service.submit_control({"action": "play", "slot": 1}) + self.assertEqual(result["router"], "waiting") + # Model a durable/reconnect review becoming visible while Router + # acquisition is pending. + self.store.protect_slot( + 2, + "restart_uncertain", + guard_id="review-during-router-acquisition", + ) + + reviews = self.service.health()["reconciliation_reviews"] + self.assertEqual(len(reviews), 1) + self.assertIs(reviews[0]["clearable"], False) + def test_definite_external_publish_failure_clears_prepublication_guard(self): self.start_service(mqtt=FakeMqttAdapter(connected=False)) playlist = Path(self.tempdir.name) / "operator-playlist.m3u" diff --git a/tests/test_store.py b/tests/test_store.py index 1f91263..b491700 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -127,6 +127,51 @@ class FreshStoreTests(StoreTestCase): 7, ) + def test_failed_downloads_receive_expiry_and_cannot_exhaust_slots_forever(self): + with self.open_store() as store: + for slot in range(1, 11): + reservation = store.reserve_download( + f"https://example.test/failure-{slot}", + None, + None, + job_id=f"failure-{slot}", + ) + self.assertEqual(reservation.slot, slot) + self.assertTrue( + store.fail_download( + reservation.job_id, + slot, + reservation.generation, + "download failed", + ) + ) + self.assertEqual( + store.get_slot(slot).expires_at, + self.clock.value + self.config.retention_seconds, + ) + + self.assertEqual( + store.reserve_download( + "https://example.test/temporarily-full", + None, + None, + ).status, + "queue_full", + ) + + self.clock.value += self.config.retention_seconds + 1 + tokens = store.begin_expired_deletions() + self.assertEqual([token.slot for token in tokens], list(range(1, 11))) + for token in tokens: + self.assertTrue(store.finish_delete(token)) + + replacement = store.reserve_download( + "https://example.test/recovered-capacity", + None, + None, + ) + self.assertEqual(replacement.slot, 1) + def test_stale_worker_cannot_overwrite_reused_slot(self): with self.open_store() as store: first = store.reserve_download("https://example.test/one", None, None, job_id="old")