diff --git a/docs/superpowers/plans/2026-07-17-subtitle-extract-cache.md b/docs/superpowers/plans/2026-07-17-subtitle-extract-cache.md new file mode 100644 index 00000000..bfcefb22 --- /dev/null +++ b/docs/superpowers/plans/2026-07-17-subtitle-extract-cache.md @@ -0,0 +1,290 @@ +# 2026-07-17 — Embedded subtitle extraction: investigation and fix + +Commands assume the repository root is the cwd. + +This document records both the **final design** and the **investigation that +produced it**, including four superseded revisions. The dead ends are kept +deliberately: every one of them looked correct, three were disproved only by +measurement, and the last was a release-blocking bug caught in review. Anyone +tempted to "clean up" the subtitle windowing code should read Evolution first. + +## Summary + +Extracting an embedded subtitle track walks the interleaved container: subtitle +packets sit between video and audio across clusters, so harvesting a few KB of +text means demuxing that stretch of a multi-GB file off network storage (CephFS). + +The server believes it bounds this with a 600s window. **It does not** — the flag +is inert, so every request extracts from the seek point to end-of-file. We cannot +turn the window on, because two of three clients depend on receiving whole tracks. + +**Fix: cache the extract. Change nothing a client can observe.** + +## Evidence + +41.6h of production logs (2026-07-15 09:19 → 2026-07-17 02:56, 1.29M requests), +`/api/v1/stream/{session_id}/subtitles/{track}`: + +| client | requests | p50 | p95 | max | ≥10s | +|---|---|---|---|---|---| +| browser (web player) | 220 | 72ms | **120,021ms** | 121,470ms | **45** | +| native (Android/TV) | 42 | 145ms | 9,566ms | 41,607ms | 2 | + +Sidecar/external subtitles: **118 requests, all <1s** — standalone files, read +directly, no demux. That fast path already exists and is why the distribution +first looked bimodal. + +Both clients are slow; they differ in *how often they pay*. The web player +re-fetches a window every ~10 minutes of playback, so each slow fetch **stalls +subtitles mid-film** — that is the 120s cluster. Native fetches once per track and +holds all cues, so it rolls the dice once: up to ~41s before subtitles appear, then +fine. + +The 120s ceiling is `WriteTimeout: 120 * time.Second` (`cmd/silo/main.go:2389`) +cutting the connection mid-body. `WriteHeader(200)` was already called +(`stream.go:576`) and the logging middleware records that status +(`internal/api/middleware/request_logger.go:29`) — so **truncated subtitles are +logged `status=200`** and are invisible in error metrics. + +## Root cause: the window flag is inert + +`streamExtractArgs` (`subtitle_stream.go:161-168`) passes `-t` as an **input** +option, reasoning it "caps how much of the file we read". Measured against the +production binary (`ffmpeg 7.1.4-Jellyfin` at `/usr/lib/jellyfin-ffmpeg/ffmpeg` +inside the `silo` container — not on `$PATH`; there is no host ffmpeg): + +| command | wall | bytes | last cue | +|---|---|---|---| +| `-ss 4000 -t 30 -i F` | 5s | 38,562 | **02:10:04** | +| `-ss 4000 -t 300 -i F` | 6s | 38,562 | **02:10:04** | +| `-ss 4000 -i F` (no `-t`) | — | 38,562 | 02:10:04 | +| `-t 600 -i F` (seek 0) | **86s** | 83,092 | **02:10:04** | + +`-t 30` and `-t 300` are byte-identical to each other **and to passing no `-t` at +all**. Only `-ss` works. Cost is therefore *(duration − seek) × bitrate*, which the +logs confirm exactly: `seek=8s/130s/519s` → ≥60s; `seek=7797s/8392s/8987s` → <1s. +Bitrate correlates monotonically (39.8 Mbps → 120.6s; 5.0 Mbps → 8.0s; every 120s +outlier is a ≥28 Mbps remux). + +`-to` as an **output** option does bound it correctly — `-ss 4000 … -to 4600` → +1s, 9,170 bytes, last cue 01:16:32. **We are deliberately not using it.** + +## Why the window stays off + +Subagent scans of the sibling repos found the accidental whole-track behaviour has +become the de-facto contract: + +- **silo-apple — BREAKS.** `SidecarSubtitleFetcher.swift:48-95` issues a single + GET; `SubtitleSession.swift:244-287` caches the whole response in + `sidecarCache[urlIndex]` and never re-requests. No `position=`/`duration=`, no + window constant, no sliding window. Its comment states the "buffered path" is + deliberate. +- **silo-android — BREAKS.** `SubtitleTrackMerge.kt:56` builds the URL bare and + hands it to Media3 as a `MediaItem.SubtitleConfiguration` + (`SiloPlayerFactory.kt:413`) → `ProgressiveMediaSource` fetches once, parses every + cue. No timer, no position listener, no re-fetch. +- **web** windows correctly: sends `position=` + (`web/src/player/hooks/useSubtitleTracks.ts:79`), advances by 595 + (`WINDOW_DURATION 600 − WINDOW_OVERLAP 5`, `:9`/`:16`/`:409-410`) — the exact + stride observed in production. + +Enabling the window would silently kill subtitles ~10 minutes into every film on +iOS, tvOS, macOS and Android. Caching removes the need: keep whole-track delivery, +make it cheap. + +## Design + +`SubtitleCache` (`internal/playback/subtitle_cache.go`) already implements the +needed shape for PGS: a full-track miss streams to the client while teeing into a +temp file, atomically renamed on clean exit (`ServeSUPExtract:113`); a hit serves +from disk; a windowed request re-extracts from the small cached artifact +(`serveWindowedSUP:161`, via `InputIsExtractedSup`) or starts `WarmInBackground:188`. +Keyed by path + track ordinal + source mtime+size. Text was excluded only by the +assumption at `stream.go:557-562` that "VTT is already windowed and fast" — which +the inert `-t` makes false. + +Windowing a **cached** 83 KB VTT costs **52ms** vs 16,000ms against the original +27 GB remux (`-ss 4000 -i cached.vtt … -to 4600` → 9,216 bytes, cues 01:04:40 → +01:16:32). Note `-ss` on a webvtt input lands ~2 min *before* the target — a +coverage superset, not byte-equivalence. + +### The load-bearing rule: canonical artifact + effective-partial predicate + +A cache entry may only ever hold a **canonical extract**: source track → requested +output format, `seek=0`, `duration=0`, original-container input, original track +mapping. + +**`AllowWindow` must not be used to decide this.** It is set only in the PGS branch +(`stream.go:531`), so it is always false for text — while `streamExtractArgs:155` +applies `-ss` to any non-ASS/non-PGS source *regardless* of it. Routing on +`AllowWindow` therefore lets a seeked text extract be committed as canonical: + +1. subrip request at 900s → `AllowWindow=false` → full-track path. +2. ffmpeg applies `-ss 900`; inert `-t` does not save it → output is 900s→EOF. +3. ffmpeg exits cleanly → `Commit` publishes the partial as canonical. +4. Every later viewer from 0 gets a hit and **receives no cues before 900s**. + +118 of 143 production requests carry a non-zero seek, so this would poison +immediately and silently truncate subtitles for everyone until eviction. Classify +from the **effective argv** instead — ideally via the same helper that builds it: + +- effective seek/duration ⇒ partial ⇒ **never fills** the canonical entry; on a hit + re-extract from the cached artifact, on a miss stream the seeked extract uncached + and start a canonical background warm. +- seek=0 & duration=0 ⇒ canonical ⇒ may fill. + +(ASS→VTT happens to be safe because `IsASS` blocks `-ss` — that does **not** +generalize to subrip, which is 135 of 143 requests.) + +### Steps + +1. `subtitle_stream.go` — export the output-format resolver (`streamExtractOutput` + already is one; export it). Do **not** add a second mapping: the handler's + duplicate at `stream.go:493-499` is exactly what let two copies disagree. + Preserve current windowing behaviour **exactly** — this commit changes latency + only. +2. `stream.go` — delete the duplicated mapping; resolve format once. Keep + `seek`/`duration` selection keyed so the log line cannot report a window ffmpeg + will ignore. +3. `subtitle_cache.go` — generalize: + - **Key**: add the resolved output profile (out codec + muxer, not the caller's + `"vtt"` string) **and a schema version** — mtime+size cannot detect an in-place + ffmpeg upgrade or a future argv change. `.ass` and `.vtt` are both reachable + for one ASS source, so format must be in the key. + - **`removeStaleSiblings:468` groups by a prefix excluding format** — committing + `.vtt` would delete a valid `.ass` sibling. Cleanup (`:476`) and eviction + (`:517`) hardcode `.sup`, so text would never be reclaimed or counted. + - Generalize `InputIsExtractedSup` (used at `subtitle_stream.go:170-176` to force + `-f sup` + remap `0:0`) to carry the cached input's format. + - `serveWindowedSUP:171` hardcodes `application/octet-stream`; drive Content-Type + from the resolved format (`text/vtt`, `text/x-ssa`). + - Rename now that it carries text: `SUPExtractFunc` → `SubtitleExtractFunc`, + `ServeSUPExtract` → `ServeExtract`, `serveWindowedSUP` → `serveWindowed`. + - In-flight coalescing must include the output profile. + - Fix the doc comment (`:19-38`) and the `defaultSubtitleCacheMaxBytes` sizing + note, which both claim PGS-only. +4. `stream.go:563` — remove the `outFormat == "sup"` gate so text routes through the + cache. Correct the false comment at `:557-562`. +5. **Keep HTTP semantics unchanged**: copy the cached file with the current + Content-Type and `no-store`. Do **not** adopt `http.ServeContent` here — it adds + Range/`206`/`304`/`Content-Length`, and Media3 uses range-capable data sources, + so responses would differ depending on whether the cache happened to be warm. + Revisit separately with client testing. + +## Risk / follow-ups + +- **The inert `-t` stays.** Removing or "fixing" it changes what clients receive. + Its comment (`subtitle_stream.go:161-165`) confidently describes a bound that does + not exist and **must be corrected in place** — otherwise the next reader fixes it + and breaks both native clients. This is the most important comment in the change. +- **Cache-key completeness is correctness-critical.** A missing dimension serves + wrong bytes. mtime+size remains a residual risk (a replaced file with identical + size and preserved timestamp reuses stale content); `MediaFile.FileHash` exists if + we want content identity later. Documented, not solved. +- **First-touch is unimproved.** Native clients fetch once, so the first viewer of a + file+track still pays full price; they benefit only from another viewer's warm. + The web player's 2nd–14th windows are the win. +- **Cached-vs-cold output is a coverage superset, not byte-identical** (`-ss` on a + webvtt input lands early). Verification standard is semantic cue coverage. +- **120s WriteTimeout masking is not fixed** (`status=200` on a truncated body) — + largely inherent once headers are committed; + `LogSubtitleStreamError` (`subtitle_stream.go:273-277`) documents it as + deliberate. A cold first extract of a large remux can still exceed it. Follow-up: + the rolling deadline full downloads already use + (`internal/api/handlers/downloads.go:402`, `internal/httpstream/rolling_deadline.go`). +- **Not in scope:** the standalone proxy has separate uncached ASS/text extraction + paths (`internal/proxy/server.go:273`, `:286`); if remote/proxy subtitle URLs reach + those, they stay uncached. +- **Dropped: warm-on-playback-start.** `BeginFill` coalesces cache *writes* but does + not share extraction *output* — so a warm at session start plus the client's + immediate request (exactly what auto-enabled subtitles do) yields **two whole-file + demuxes and no latency benefit**. `WarmInBackground:188` also does `stat`/`mkdir`/ + temp-file work synchronously before detaching, which could block session start on + CephFS. Revisit only if commit 1's metrics show a real gap between playback start + and subtitle selection. +- **Deferred:** (a) real windowing, once `silo-android`/`silo-apple` send + `position=`/`duration=` and re-fetch on coverage exhaustion — then `-t`→`-to` + becomes safe and this cost drops again; (b) scan-time extraction of embedded tracks + into sidecars, putting every subtitle on the already-instant external path and + retiring on-demand extraction entirely. (b) is the real endgame. +- **Non-finite `?position=`** (`+Inf` parses and passes `>= 0`, reaching ffmpeg and + failing after the 200) is a genuine bug found en route — its own commit, not this. + +## Evolution (what we got wrong, and how it was caught) + +Kept because each error is one a careful reader would repeat. + +**R1 — "ASS→VTT skips windowing."** `streamExtractArgs:155` keys `windowable` on +`SourceCodec`, so an ASS source requested as `.vtt` never seeks → full-file demux. +Real, and the code reads exactly that way. Proposed a one-line predicate fix. +*Caught by Codex:* a **no-op**. The handler computes `seek`/`duration` from a +source-derived format *before* the `.vtt` override (`stream.go:493-520`), so ASS +yields `seek=0, duration=0` and no `-ss`/`-t` is emitted whatever the predicate +says. The proposed unit test would have passed while production stayed broken — +it hand-built opts the handler never produces. + +**R2 — fix both layers.** Correct, and Codex's implementation was clean. +*Caught by the logs:* the premise was wrong. The `subtitle stream requested` line +records `track_codec`/`seek`/`duration`; correlating it against request durations +showed **all 7 ASS requests logged `seek=0 dur=0` but never exceeded 10s**, while +**every request ≥60s was subrip with `seek=NNN dur=600`** — already windowed. The +same file (71458) ranged 41ms→120,574ms. The tidy "bimodality splits by codec" +story was a subagent's inference that I propagated without testing it against logs +I already had open. + +**R3 — `-t`→`-to` plus 600s→60s.** Found by running ffmpeg instead of reading it: +`-t` is inert. This is the true root cause, and it hits subrip (135 of 143 +requests), not ASS. *Caught by the client scans:* both native clients fetch once and +expect whole tracks, so enabling the window breaks them. Also caught: 600→60 would +have been worse — the web player hardcodes `WINDOW_DURATION = 600` and advances by +595 without checking what it received, so a 60s server window means a **535-second +subtitle blackout** between fetches. + +**R4 — cache text tracks.** *Caught by Codex:* routing on `AllowWindow` lets a +seeked extract be committed as canonical → cache poisoning (see Design). I had +generalized from ASS, the one case where `IsASS` blocks `-ss`. + +**Method errors worth naming:** +- I twice reported findings from a subagent's code reading without checking them + against production data I already had. +- I greped `position=`/`duration=`/`windowed=1` on `/api/v1` request logs and + concluded no client sends them. **The api request log has no `query=` field** — + the greps were vacuous. Ground truth is `seek_seconds` in the handler log (118 of + 143 non-zero). Do not grep query params off api log lines. +- The first "86s → 1s" measurement was cache-flattered; the honest cold number is + 16s. Always re-measure cold, in a fresh order. +- Two rounds of code review missed the inert `-t` because we both read *intent*: + the comment says it windows, the test asserts the flag is present, and it is + present. Only execution disproved it. + +## Verification + +- `go build ./...`, `go vet ./...`, `gofmt -l` clean. +- `go test -race ./internal/playback/ ./internal/api/handlers/`. Note + `TestHandleReplanPlaybackV3SeekFailureRecoveryNeverChangesMediaVersion` **already + fails on clean `origin/main`** — pre-existing, unrelated, do not chase it. +- Required tests: + - **a seeked `.vtt` miss must not commit a canonical entry**, and a subsequent + seek-0 request must still receive beginning-of-track cues (the poisoning + regression — the single most important test here); + - seek-0 fill, then a seeked request re-extracts from the cached artifact (assert + the rewritten input path/format, not the original); + - `.ass` and `.vtt` for the same file+track coexist and do not evict each other + via `removeStaleSiblings`; + - text entries participate in eviction accounting; + - a partial/failed extract never commits; + - Content-Type per format on hit and miss. +- **Behavioural equivalence is the bar**: for every (source codec, requested format, + seek), the bytes a client receives must be unchanged from `origin/main`. Only + latency may differ. +- Post-deploy: re-run the ≥1s aggregation over `docker logs silo`. Expect browser + p95 to fall away from 120s and the ≥10s count (45) to collapse toward the handful + of cold first-touches. Watch `subtitle cache warm` lines for failures and + dropped-slot rates. + +## AI-use disclosure + +Investigation, root-cause analysis, and this document were prepared with AI (Claude +Code) assistance. The plan was adversarially reviewed by Codex (`gpt-5.6-sol`), +which caught both the R1 no-op and the R4 cache-poisoning bug; client impact was +assessed by subagent scans of `silo-apple` and `silo-android`. diff --git a/internal/api/handlers/stream.go b/internal/api/handlers/stream.go index ef198fc3..57cdbe1c 100644 --- a/internal/api/handlers/stream.go +++ b/internal/api/handlers/stream.go @@ -490,13 +490,7 @@ func (h *StreamHandler) handleTransportStartFailure(ctx context.Context, session // pipeline, it works the same for direct play, remux, and transcode. func (h *StreamHandler) streamEmbeddedSubtitle(w http.ResponseWriter, r *http.Request, file *models.MediaFile, embeddedIndex int, session *playback.Session, requestedFormat ...string) { track := file.SubtitleTracks[embeddedIndex] - outFormat := "vtt" - switch { - case playback.IsASS(track.Codec): - outFormat = "ass" - case playback.IsPGS(track.Codec): - outFormat = "sup" - } + _, outFormat := playback.StreamExtractOutput(track.Codec, requestedFormat...) // ASS is fetched exactly once and consumed whole by its client-side // renderer (JASSUB), so it must never be windowed. PGS defaults to @@ -511,11 +505,11 @@ func (h *StreamHandler) streamEmbeddedSubtitle(w http.ResponseWriter, r *http.Re // misleading nonzero seek here. var seek, duration float64 var allowWindow bool - switch outFormat { - case "vtt": + switch { + case outFormat == "webvtt" && !playback.IsASS(track.Codec): seek = subtitleSeekPosition(r, session) duration = subtitleWindowDuration(r) - case "sup": + case outFormat == "sup": allowWindow, seek, duration = playback.PGSWindowRequest(r.URL.Query()) } slog.InfoContext(r.Context(), "subtitle stream requested", "component", "api", @@ -549,38 +543,16 @@ func (h *StreamHandler) streamEmbeddedSubtitle(w http.ResponseWriter, r *http.Re return } opts.TargetFormat = "vtt" - outFormat = "vtt" } w.Header().Set("Access-Control-Allow-Origin", "*") - // Full-track PGS extracts are expensive (whole-file demux) and byte- - // identical across requests, so they are served from / teed into the - // subtitle cache; windowed PGS requests extract their slice from the - // cached full track when present (warming it in the background when - // not). All other formats stream uncached: VTT is already windowed - // and fast, ASS is small. - if outFormat == "sup" { - err := h.SubtitleCache.ServeSUPExtract(w, r, opts, playback.StreamExtractSubtitle) - playback.LogSubtitleStreamError(r.Context(), err, file.ID, embeddedIndex) - return - } - - switch outFormat { - case "ass": - w.Header().Set("Content-Type", "text/x-ssa; charset=utf-8") - default: - w.Header().Set("Content-Type", "text/vtt; charset=utf-8") - } - w.Header().Set("Cache-Control", "no-store") - w.WriteHeader(http.StatusOK) - - opts.Writer = w - if err := playback.StreamExtractSubtitle(r.Context(), opts); err != nil { - // Headers already committed — best we can do is log and let - // the client see a truncated response. - playback.LogSubtitleStreamError(r.Context(), err, file.ID, embeddedIndex) - } + // Embedded extracts are served from or teed into the subtitle cache. + // Requests whose effective ffmpeg argv includes seek/duration are never + // canonical: they stream uncached and warm a seek-0/duration-0 artifact, + // or re-extract from that artifact when it is already present. + err := h.SubtitleCache.ServeExtract(w, r, opts, playback.StreamExtractSubtitle) + playback.LogSubtitleStreamError(r.Context(), err, file.ID, embeddedIndex) } // subtitleSeekPosition picks the best-known starting position for a @@ -599,12 +571,10 @@ func subtitleSeekPosition(r *http.Request, session *playback.Session) float64 { return 0 } -// subtitleWindowDuration picks the bounded extract length. The client -// overrides via ?duration=; absent that we use a 10-minute window, -// which is long enough that a single fetch covers many minutes of -// uninterrupted playback but short enough that the ffmpeg process -// finishes (and frees its input handle) well before the next window -// is requested. +// subtitleWindowDuration picks the compatibility duration argument. The +// client overrides via ?duration=; absent that we pass 10 minutes. ffmpeg +// currently ignores this input-side -t for subtitle extracts and runs to EOF; +// changing it would alter the whole-track response native clients depend on. func subtitleWindowDuration(r *http.Request) float64 { const defaultDuration = 600.0 const maxDuration = 3600.0 diff --git a/internal/playback/subtitle_cache.go b/internal/playback/subtitle_cache.go index 14f6504d..f1713527 100644 --- a/internal/playback/subtitle_cache.go +++ b/internal/playback/subtitle_cache.go @@ -16,26 +16,25 @@ import ( "time" ) -// SubtitleCache stores full-track PGS (.sup) subtitle extracts on disk so +// SubtitleCache stores canonical full-track subtitle extracts on disk so // repeat selections of the same embedded track don't re-run a whole-file -// ffmpeg demux (minutes for a large remux). Only complete, unwindowed .sup -// extracts are cached — VTT extracts are already windowed and fast, and ASS -// extracts are small; neither pays the full-demux cost PGS does. +// ffmpeg demux (minutes for a large remux). Only extracts whose effective +// ffmpeg argv has no seek or duration are cached; partial requests may use a +// canonical cached artifact as their input but are never published themselves. // -// Entries are keyed by the source file path, subtitle stream ordinal, and the -// source's mtime+size, all encoded in the cache filename. Invalidation is -// therefore implicit: when the source changes, the lookup key changes and the -// old entry becomes garbage that eviction reclaims. Entry recency for LRU is -// tracked by bumping the cache file's mtime on every hit (portable, unlike -// atime which is often disabled via noatime/relatime mounts). +// Entries are keyed by schema version, source file path, subtitle stream +// ordinal, resolved output codec+muxer, and source mtime+size, all encoded in +// the cache filename. Invalidation is therefore implicit when any dimension +// changes. Entry recency for LRU is tracked by bumping the cache file's mtime +// on every hit (portable, unlike atime on noatime/relatime mounts). // // Concurrency: the first requester of an uncached track streams the extract // progressively to its client while teeing bytes into a temp file that is // atomically renamed into the cache on clean ffmpeg exit (and discarded on // any error, so a partial entry is never served). Concurrent requesters for -// the same track while a fill is in flight simply run their own un-teed -// extract — no worse than today's behavior, and it avoids making a viewer's -// first-byte latency depend on another client's connection. +// the same track and output profile while a fill is in flight simply run their +// own un-teed extract — no worse than today's behavior, and it avoids making a +// viewer's first-byte latency depend on another client's connection. type SubtitleCache struct { // transcodeDir returns the current transcode directory; the cache lives // in a subtitle-cache subdirectory beneath it, created lazily. An empty @@ -56,8 +55,12 @@ type SubtitleCache struct { const ( subtitleCacheDirName = "subtitle-cache" - // defaultSubtitleCacheMaxBytes caps the cache at 2 GiB — PGS tracks run - // 15-80 MB, so this holds a few dozen tracks. + // subtitleCacheSchemaVersion invalidates artifacts when extraction argv or + // artifact semantics change in a way source mtime+size cannot detect. + subtitleCacheSchemaVersion = 1 + // defaultSubtitleCacheMaxBytes caps all cached subtitle formats at 2 GiB. + // Bitmap tracks dominate sizing at roughly 15-80 MB apiece; text tracks + // are much smaller. // TODO: expose as a config knob following the download.artifact_max_bytes // pattern (internal/config/config.go DownloadConfig.ArtifactMaxBytes). defaultSubtitleCacheMaxBytes = 2 << 30 @@ -74,12 +77,27 @@ const ( subtitleCacheWarmTimeout = 30 * time.Minute ) -// SUPExtractFunc runs one ffmpeg subtitle extract described by opts, writing +// ffmpeg muxer names for the subtitle formats an extract can produce (see +// StreamExtractOutput), plus the on-disk extension and the source codec the +// PGS-only entry points key on. Named because the extraction plan, the cache +// key, and the response Content-Type all have to agree on the same spellings. +const ( + subtitleFormatWebVTT = "webvtt" + subtitleFormatASS = "ass" + subtitleFormatSUP = "sup" + // subtitleExtVTT is the file extension for a webvtt artifact; the ffmpeg + // muxer name and the conventional extension differ only for this format. + subtitleExtVTT = "vtt" + subtitleCodecPGS = "hdmv_pgs_subtitle" + subtitleTypeOctetStream = "application/octet-stream" +) + +// SubtitleExtractFunc runs one ffmpeg subtitle extract described by opts, writing // output to opts.Writer. Production callers pass StreamExtractSubtitle; // tests substitute fakes. The cache invokes it with the caller's options -// rewritten as needed (tee writer for fills, cached-.sup input for windowed +// rewritten as needed (tee writer for fills, cached-artifact input for partial // serves, cleared window for background warms). -type SUPExtractFunc func(ctx context.Context, opts StreamExtractOpts) error +type SubtitleExtractFunc func(ctx context.Context, opts StreamExtractOpts) error // NewSubtitleCache builds a cache rooted under the transcode directory // returned by transcodeDir at call time (so runtime config changes are @@ -93,46 +111,48 @@ func NewSubtitleCache(transcodeDir func() string) *SubtitleCache { } } -// ServeSUPExtract serves the .sup extract for one source+track described by -// opts (opts.Writer is ignored; the cache supplies it). Full-track requests -// (no AllowWindow): a cache hit is served with http.ServeContent (Range -// support, Content-Length, Last-Modified from the source file's mtime, -// revalidatable instead of no-store); a miss invokes extract with a writer -// that streams to the client while teeing bytes into a temp file, atomically -// published as the cache entry on clean extract exit and discarded on any -// error (ffmpeg failure or client disconnect) — a partial entry is never -// served. Windowed requests (opts.AllowWindow): the output covers only a -// slice of the track, so it is never cached; but when the full-track entry -// already exists, the windowed extract runs against the small cached .sup -// instead of re-demuxing the original file, and when it doesn't, a detached -// background warm is kicked off so subsequent windows get that fast path. A -// nil receiver disables caching and just streams. +// ServeExtract serves one embedded subtitle extract (opts.Writer is ignored; +// the cache supplies it). Canonical requests may fill the cache on a miss. +// Requests whose effective ffmpeg argv applies seek or duration are partial: +// they never fill, re-extract from a cached canonical artifact on a hit, and +// trigger a canonical background warm on a miss. +// +// PGS cache hits retain the existing http.ServeContent behavior. Text hits +// deliberately use a plain copy with no-store, matching cold-path HTTP +// semantics without adding Range, validators, or Content-Length based on cache +// warmth. A nil receiver disables caching and just streams. // // The caller sets any extra response headers (e.g. CORS) before calling. // The returned error is the extract error; cache hits return nil. -func (c *SubtitleCache) ServeSUPExtract(w http.ResponseWriter, r *http.Request, opts StreamExtractOpts, extract SUPExtractFunc) error { - if opts.AllowWindow { - return c.serveWindowedSUP(w, r, opts, extract) +func (c *SubtitleCache) ServeExtract(w http.ResponseWriter, r *http.Request, opts StreamExtractOpts, extract SubtitleExtractFunc) error { + plan := streamExtractPlanFor(opts) + if plan.partial() { + return c.serveWindowed(w, r, opts, plan, extract) } - if cached, modTime, ok := c.Lookup(opts.InputPath, opts.TrackIndex); ok { + if cached, modTime, ok := c.lookup(opts); ok { defer func() { _ = cached.Close() }() slog.DebugContext(r.Context(), "subtitle stream served from cache", "input", opts.InputPath, "track", opts.TrackIndex) - w.Header().Set("Content-Type", "application/octet-stream") - w.Header().Set("Cache-Control", "private, no-cache") - http.ServeContent(w, r, "", modTime, cached) - return nil + w.Header().Set("Content-Type", subtitleContentType(plan.outFormat)) + if plan.outFormat == subtitleFormatSUP { + w.Header().Set("Cache-Control", "private, no-cache") + http.ServeContent(w, r, "", modTime, cached) + return nil + } + w.Header().Set("Cache-Control", "no-store") + w.WriteHeader(http.StatusOK) + return copyAndFlush(w, cached) } - w.Header().Set("Content-Type", "application/octet-stream") + w.Header().Set("Content-Type", subtitleContentType(plan.outFormat)) w.Header().Set("Cache-Control", "no-store") w.WriteHeader(http.StatusOK) // BeginFill returns nil when another fill for this track is already in // flight (or the cache dir is unusable); this request then streams its // own uncached extract. - fill := c.BeginFill(opts.InputPath, opts.TrackIndex) + fill := c.beginFill(opts) var writer io.Writer = w if fill != nil { writer = fill.Tee(w) @@ -151,24 +171,24 @@ func (c *SubtitleCache) ServeSUPExtract(w http.ResponseWriter, r *http.Request, return err } -// serveWindowedSUP streams a windowed slice of the track. The output is a -// position-dependent slice so it is never cached itself, but the cache still -// speeds it up: with a committed full-track entry the extract's input is -// rewritten to the cached .sup (15-80 MB, so the -ss scan is near-instant +// serveWindowed streams a partial request. Its effective argv is noncanonical, +// so its output is never cached itself, but the cache still speeds it up: with +// a committed full-track entry the extract's input is +// rewritten to the cached artifact (small enough that the -ss scan is fast // versus re-demuxing a multi-GB source); without one, a background warm is // started so later windows — the client re-fetches on every seek — hit the // fast path. -func (c *SubtitleCache) serveWindowedSUP(w http.ResponseWriter, r *http.Request, opts StreamExtractOpts, extract SUPExtractFunc) error { - if cachedPath, _, ok := c.cachedEntryPath(opts.InputPath, opts.TrackIndex); ok { +func (c *SubtitleCache) serveWindowed(w http.ResponseWriter, r *http.Request, opts StreamExtractOpts, plan streamExtractPlan, extract SubtitleExtractFunc) error { + if cachedPath, _, ok := c.cachedEntryPath(opts); ok { slog.DebugContext(r.Context(), "windowed subtitle extract using cached full track", "input", opts.InputPath, "track", opts.TrackIndex, "cache_entry", cachedPath) opts.InputPath = cachedPath - opts.InputIsExtractedSup = true + opts.ExtractedInputFormat = plan.outFormat } else { c.WarmInBackground(opts, extract) } - w.Header().Set("Content-Type", "application/octet-stream") + w.Header().Set("Content-Type", subtitleContentType(plan.outFormat)) w.Header().Set("Cache-Control", "no-store") w.WriteHeader(http.StatusOK) @@ -177,15 +197,15 @@ func (c *SubtitleCache) serveWindowedSUP(w http.ResponseWriter, r *http.Request, } // WarmInBackground starts a detached full-track extract that fills the cache -// entry for opts' source+track, so future windowed requests can extract from -// the small cached .sup instead of the original file. The warm runs on a -// background context with a generous timeout — it must survive the request -// that triggered it. BeginFill's in-flight coalescing guarantees at most one -// fill per track (a concurrent client-driven fill wins and the warm is +// entry for opts' source+track+output profile, so future partial requests can +// extract from the small cached artifact instead of the original file. The +// warm runs on a background context with a generous timeout — it must survive +// the request that triggered it. BeginFill's in-flight coalescing guarantees at +// most one fill per output profile (a concurrent client-driven fill wins and the warm is // skipped), and warmSem bounds warms server-wide: beyond the budget the warm // is dropped, not queued — the next windowed miss re-attempts it. A nil // receiver is a no-op. -func (c *SubtitleCache) WarmInBackground(opts StreamExtractOpts, extract SUPExtractFunc) { +func (c *SubtitleCache) WarmInBackground(opts StreamExtractOpts, extract SubtitleExtractFunc) { if c == nil || extract == nil { return } @@ -196,20 +216,19 @@ func (c *SubtitleCache) WarmInBackground(opts StreamExtractOpts, extract SUPExtr "input", opts.InputPath, "track", opts.TrackIndex) return } - fill := c.BeginFill(opts.InputPath, opts.TrackIndex) + // Full-track options: the warm ignores the triggering request's window + // and writes only to the cache temp file (no response writer). + opts.SeekSeconds = 0 + opts.DurationSeconds = 0 + opts.AllowWindow = false + opts.ExtractedInputFormat = "" + fill := c.beginFill(opts) if fill == nil { // Another fill (client-driven or a previous warm) is already in // flight, or the cache is unusable — either way, nothing to do. <-c.warmSem return } - - // Full-track options: the warm ignores the triggering request's window - // and writes only to the cache temp file (no response writer). - opts.SeekSeconds = 0 - opts.DurationSeconds = 0 - opts.AllowWindow = false - opts.InputIsExtractedSup = false opts.Writer = fill.Tee(io.Discard) go func() { @@ -250,27 +269,65 @@ func (c *SubtitleCache) dir() string { return filepath.Join(base, subtitleCacheDirName) } -// subtitleCacheKeyPrefix identifies a source file + track ordinal regardless -// of source version; the full key appends mtime+size so a changed source -// yields a different filename. -func subtitleCacheKeyPrefix(inputPath string, trackIndex int) string { +type subtitleCacheProfile struct { + codec string + format string +} + +func subtitleProfile(opts StreamExtractOpts) subtitleCacheProfile { + plan := streamExtractPlanFor(opts) + return subtitleCacheProfile{codec: plan.outCodec, format: plan.outFormat} +} + +// subtitleCacheKeyPrefix identifies a source file + track ordinal + resolved +// output profile regardless of source version. Including the profile keeps, +// for example, ASS and WebVTT artifacts for the same source track independent. +func subtitleCacheKeyPrefix(inputPath string, trackIndex int, profile subtitleCacheProfile) string { sum := sha256.Sum256([]byte(inputPath)) - return fmt.Sprintf("%x-s%d-", sum[:12], trackIndex) + return fmt.Sprintf("v%d-%x-s%d-%s-%s-", subtitleCacheSchemaVersion, sum[:12], trackIndex, profile.codec, profile.format) } -func subtitleCacheKey(inputPath string, trackIndex int, mtime time.Time, size int64) string { - return fmt.Sprintf("%s%d-%d.sup", subtitleCacheKeyPrefix(inputPath, trackIndex), mtime.UnixNano(), size) +func subtitleCacheKey(inputPath string, trackIndex int, profile subtitleCacheProfile, mtime time.Time, size int64) string { + return fmt.Sprintf("%s%d-%d.%s", subtitleCacheKeyPrefix(inputPath, trackIndex, profile), mtime.UnixNano(), size, subtitleCacheExtension(profile.format)) } -// Lookup opens the cached full-track .sup extract for the given source file -// and subtitle stream ordinal. The source is stat'ed on every lookup: an +func subtitleCacheExtension(format string) string { + if format == subtitleFormatWebVTT { + return subtitleExtVTT + } + return format +} + +func subtitleContentType(format string) string { + switch format { + case subtitleFormatASS: + return "text/x-ssa; charset=utf-8" + case subtitleFormatWebVTT: + return "text/vtt; charset=utf-8" + default: + return subtitleTypeOctetStream + } +} + +// Lookup opens the cached canonical PGS extract for the source and track. +// It is retained as a focused cache primitive; format-aware serving uses +// lookup with the caller's resolved extraction options. +// // mtime or size mismatch means the entry (if any) is stale and reads as a // miss. On a hit the returned modTime is the *source* file's mtime — stable // across hits, suitable for Last-Modified — while the cache file's own mtime // is bumped to record recency for LRU eviction. The caller owns closing the // returned file. func (c *SubtitleCache) Lookup(inputPath string, trackIndex int) (f *os.File, modTime time.Time, ok bool) { - path, modTime, ok := c.cachedEntryPath(inputPath, trackIndex) + return c.lookup(StreamExtractOpts{ + InputPath: inputPath, + TrackIndex: trackIndex, + SourceCodec: subtitleCodecPGS, + }) +} + +func (c *SubtitleCache) lookup(opts StreamExtractOpts) (f *os.File, modTime time.Time, ok bool) { + path, modTime, ok := c.cachedEntryPath(opts) if !ok { return nil, time.Time{}, false } @@ -287,16 +344,16 @@ func (c *SubtitleCache) Lookup(inputPath string, trackIndex int) (f *os.File, mo // miss) and bumps the entry's mtime to record recency for LRU eviction. // Callers that hand the path to an external reader (ffmpeg) rather than // opening it themselves use this instead of Lookup. -func (c *SubtitleCache) cachedEntryPath(inputPath string, trackIndex int) (path string, srcModTime time.Time, ok bool) { +func (c *SubtitleCache) cachedEntryPath(opts StreamExtractOpts) (path string, srcModTime time.Time, ok bool) { dir := c.dir() if dir == "" { return "", time.Time{}, false } - src, err := os.Stat(inputPath) + src, err := os.Stat(opts.InputPath) if err != nil { return "", time.Time{}, false } - path = filepath.Join(dir, subtitleCacheKey(inputPath, trackIndex, src.ModTime(), src.Size())) + path = filepath.Join(dir, subtitleCacheKey(opts.InputPath, opts.TrackIndex, subtitleProfile(opts), src.ModTime(), src.Size())) if _, err := os.Stat(path); err != nil { return "", time.Time{}, false } @@ -318,6 +375,7 @@ type SubtitleCacheFill struct { key string inputPath string trackIndex int + profile subtitleCacheProfile srcMtime time.Time srcSize int64 tmp *os.File @@ -326,17 +384,25 @@ type SubtitleCacheFill struct { failed bool } -// BeginFill reserves the in-flight slot for the given track and creates the +// BeginFill reserves the in-flight slot for the given PGS track and creates the // temp file the tee will write into. Returns nil — meaning "stream without // caching" — when caching is disabled, the source can't be stat'ed, the // cache directory can't be created, or another fill for the same track is // already in flight. func (c *SubtitleCache) BeginFill(inputPath string, trackIndex int) *SubtitleCacheFill { + return c.beginFill(StreamExtractOpts{ + InputPath: inputPath, + TrackIndex: trackIndex, + SourceCodec: subtitleCodecPGS, + }) +} + +func (c *SubtitleCache) beginFill(opts StreamExtractOpts) *SubtitleCacheFill { dir := c.dir() if dir == "" { return nil } - src, err := os.Stat(inputPath) + src, err := os.Stat(opts.InputPath) if err != nil { return nil } @@ -344,7 +410,8 @@ func (c *SubtitleCache) BeginFill(inputPath string, trackIndex int) *SubtitleCac slog.Warn("subtitle cache dir create failed", "dir", dir, "error", err) return nil } - key := subtitleCacheKey(inputPath, trackIndex, src.ModTime(), src.Size()) + profile := subtitleProfile(opts) + key := subtitleCacheKey(opts.InputPath, opts.TrackIndex, profile, src.ModTime(), src.Size()) c.mu.Lock() if _, busy := c.inflight[key]; busy { @@ -363,8 +430,9 @@ func (c *SubtitleCache) BeginFill(inputPath string, trackIndex int) *SubtitleCac return &SubtitleCacheFill{ c: c, key: key, - inputPath: inputPath, - trackIndex: trackIndex, + inputPath: opts.InputPath, + trackIndex: opts.TrackIndex, + profile: profile, srcMtime: src.ModTime(), srcSize: src.Size(), tmp: tmp, @@ -442,7 +510,7 @@ func (f *SubtitleCacheFill) Commit() error { return fmt.Errorf("publish subtitle cache entry: %w", err) } - f.c.removeStaleSiblings(dir, f.inputPath, f.trackIndex, f.key) + f.c.removeStaleSiblings(dir, f.inputPath, f.trackIndex, f.profile, f.key) f.c.evict(dir) return nil } @@ -462,18 +530,18 @@ func (f *SubtitleCacheFill) closeAndRemoveTmp() { } } -// removeStaleSiblings deletes committed entries for the same source+track -// with a different mtime/size suffix — the source was replaced, so those can -// never be served again. -func (c *SubtitleCache) removeStaleSiblings(dir, inputPath string, trackIndex int, keepKey string) { - prefix := subtitleCacheKeyPrefix(inputPath, trackIndex) +// removeStaleSiblings deletes committed entries for the same +// source+track+output profile with a different mtime/size suffix. Other output +// profiles remain valid siblings and must coexist. +func (c *SubtitleCache) removeStaleSiblings(dir, inputPath string, trackIndex int, profile subtitleCacheProfile, keepKey string) { + prefix := subtitleCacheKeyPrefix(inputPath, trackIndex, profile) entries, err := os.ReadDir(dir) if err != nil { return } for _, e := range entries { name := e.Name() - if name == keepKey || !strings.HasPrefix(name, prefix) || !strings.HasSuffix(name, ".sup") { + if name == keepKey || !strings.HasPrefix(name, prefix) || strings.Contains(name, ".part-") { continue } if err := os.Remove(filepath.Join(dir, name)); err != nil && !os.IsNotExist(err) { @@ -514,7 +582,7 @@ func (c *SubtitleCache) evict(dir string) { } continue } - if !strings.HasSuffix(e.Name(), ".sup") { + if !isCommittedSubtitleCacheEntry(e.Name()) { continue } ents = append(ents, cacheEnt{path: path, size: info.Size(), mtime: info.ModTime()}) @@ -538,3 +606,10 @@ func (c *SubtitleCache) evict(dir string) { total -= e.size } } + +func isCommittedSubtitleCacheEntry(name string) bool { + if strings.Contains(name, ".part-") { + return false + } + return strings.HasSuffix(name, ".sup") || strings.HasSuffix(name, ".ass") || strings.HasSuffix(name, ".vtt") +} diff --git a/internal/playback/subtitle_cache_test.go b/internal/playback/subtitle_cache_test.go index f2255878..bdd3631f 100644 --- a/internal/playback/subtitle_cache_test.go +++ b/internal/playback/subtitle_cache_test.go @@ -31,9 +31,14 @@ func newTestCache(t *testing.T) (*SubtitleCache, string) { // the real BeginFill → Tee → Commit path. func fillEntry(t *testing.T, c *SubtitleCache, source string, track int, payload string) { t.Helper() - fill := c.BeginFill(source, track) + fillOpts(t, c, supExtractOpts(source, track), payload) +} + +func fillOpts(t *testing.T, c *SubtitleCache, opts StreamExtractOpts, payload string) { + t.Helper() + fill := c.beginFill(opts) if fill == nil { - t.Fatalf("BeginFill returned nil for track %d", track) + t.Fatalf("beginFill returned nil for track %d", opts.TrackIndex) } if _, err := fill.Tee(io.Discard).Write([]byte(payload)); err != nil { t.Fatalf("tee write: %v", err) @@ -43,8 +48,17 @@ func fillEntry(t *testing.T, c *SubtitleCache, source string, track int, payload } } +func vttExtractOpts(source string, track int) StreamExtractOpts { + return StreamExtractOpts{ + InputPath: source, + TrackIndex: track, + SourceCodec: "subrip", + TargetFormat: "vtt", + } +} + // supExtractOpts builds the base extract options a handler would pass to -// ServeSUPExtract for a PGS track. +// ServeExtract for a PGS track. func supExtractOpts(source string, track int) StreamExtractOpts { return StreamExtractOpts{ InputPath: source, @@ -319,7 +333,7 @@ func TestSubtitleCacheCoalescingConcurrent(t *testing.T) { fills[0].Discard() } -func TestServeSUPExtractCacheFlow(t *testing.T) { +func TestServeExtractCacheFlow(t *testing.T) { c, source := newTestCache(t) extractCalls := 0 @@ -332,7 +346,7 @@ func TestServeSUPExtractCacheFlow(t *testing.T) { // First request: miss → streamed 200 with no-store, entry committed. rec := httptest.NewRecorder() req := httptest.NewRequest(http.MethodGet, "/sub.sup", nil) - if err := c.ServeSUPExtract(rec, req, supExtractOpts(source, 0), extract); err != nil { + if err := c.ServeExtract(rec, req, supExtractOpts(source, 0), extract); err != nil { t.Fatal(err) } if extractCalls != 1 { @@ -347,7 +361,7 @@ func TestServeSUPExtractCacheFlow(t *testing.T) { // Second request: hit → served from cache, no extract, revalidatable. rec = httptest.NewRecorder() - if err := c.ServeSUPExtract(rec, req, supExtractOpts(source, 0), extract); err != nil { + if err := c.ServeExtract(rec, req, supExtractOpts(source, 0), extract); err != nil { t.Fatal(err) } if extractCalls != 1 { @@ -370,7 +384,7 @@ func TestServeSUPExtractCacheFlow(t *testing.T) { rec = httptest.NewRecorder() rangeReq := httptest.NewRequest(http.MethodGet, "/sub.sup", nil) rangeReq.Header.Set("Range", "bytes=4-10") - if err := c.ServeSUPExtract(rec, rangeReq, supExtractOpts(source, 0), extract); err != nil { + if err := c.ServeExtract(rec, rangeReq, supExtractOpts(source, 0), extract); err != nil { t.Fatal(err) } if rec.Code != http.StatusPartialContent || rec.Body.String() != "PAYLOAD" { @@ -378,11 +392,185 @@ func TestServeSUPExtractCacheFlow(t *testing.T) { } } +// A seeked text request has effective -ss/-t argv even though AllowWindow is +// false. Its successful output is only seek->EOF and must never become the +// canonical cache artifact used by a later viewer starting at zero. +func TestServeExtractSeekedVTTMissNeverPoisonsCanonicalEntry(t *testing.T) { + c, source := newTestCache(t) + // Disable detached warms so the test observes only the request-driven + // cache behavior and can prove the partial response itself was not stored. + c.warmSem = make(chan struct{}) + + partial := vttExtractOpts(source, 0) + partial.SeekSeconds = 900 + partial.DurationSeconds = 600 + extract := func(_ context.Context, opts StreamExtractOpts) error { + payload := "BEGINNING CUE\nLATE CUE" + if streamExtractPlanFor(opts).partial() { + payload = "LATE CUE" + } + _, err := opts.Writer.Write([]byte(payload)) + return err + } + + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/sub.vtt?position=900", nil) + if err := c.ServeExtract(rec, req, partial, extract); err != nil { + t.Fatal(err) + } + if rec.Body.String() != "LATE CUE" { + t.Fatalf("seeked body = %q", rec.Body.String()) + } + if _, _, ok := c.lookup(vttExtractOpts(source, 0)); ok { + t.Fatal("seeked VTT extract must not commit a canonical entry") + } + + rec = httptest.NewRecorder() + if err := c.ServeExtract(rec, httptest.NewRequest(http.MethodGet, "/sub.vtt", nil), + vttExtractOpts(source, 0), extract); err != nil { + t.Fatal(err) + } + if rec.Body.String() != "BEGINNING CUE\nLATE CUE" { + t.Fatalf("seek-0 body lost beginning-of-track cues: %q", rec.Body.String()) + } +} + +func TestServeExtractSeekedVTTUsesCachedArtifactAndFormat(t *testing.T) { + c, source := newTestCache(t) + canonical := vttExtractOpts(source, 0) + fillOpts(t, c, canonical, "WEBVTT\n\nFULL TRACK") + entry, _, ok := c.cachedEntryPath(canonical) + if !ok { + t.Fatal("canonical VTT entry missing") + } + + partial := canonical + partial.SeekSeconds = 1200 + partial.DurationSeconds = 600 + var got StreamExtractOpts + rec := httptest.NewRecorder() + err := c.ServeExtract(rec, httptest.NewRequest(http.MethodGet, "/sub.vtt?position=1200", nil), partial, + func(_ context.Context, opts StreamExtractOpts) error { + got = opts + _, err := opts.Writer.Write([]byte("WINDOW")) + return err + }) + if err != nil { + t.Fatal(err) + } + if got.InputPath != entry { + t.Fatalf("partial input = %q, want cached artifact %q", got.InputPath, entry) + } + if got.ExtractedInputFormat != "webvtt" { + t.Fatalf("cached input format = %q, want webvtt", got.ExtractedInputFormat) + } + args := strings.Join(streamExtractArgs(got), " ") + if !strings.Contains(args, "-f webvtt -i "+entry) || !strings.Contains(args, "-map 0:s:0") { + t.Fatalf("cached VTT argv does not force format and sole stream: %s", args) + } +} + +func TestSubtitleCacheOutputProfilesCoexist(t *testing.T) { + c, source := newTestCache(t) + ass := StreamExtractOpts{InputPath: source, TrackIndex: 0, SourceCodec: "ass"} + vtt := ass + vtt.TargetFormat = "vtt" + + fillOpts(t, c, ass, "ASS TRACK") + fillOpts(t, c, vtt, "WEBVTT TRACK") + + assFile, _, assOK := c.lookup(ass) + if !assOK { + t.Fatal("ASS entry was removed when VTT sibling committed") + } + if got := readAllAndClose(t, assFile); got != "ASS TRACK" { + t.Fatalf("ASS entry = %q", got) + } + vttFile, _, vttOK := c.lookup(vtt) + if !vttOK { + t.Fatal("VTT entry missing") + } + if got := readAllAndClose(t, vttFile); got != "WEBVTT TRACK" { + t.Fatalf("VTT entry = %q", got) + } + + // In-flight coalescing is profile-specific too. + assFill := c.beginFill(ass) + vttFill := c.beginFill(vtt) + if assFill == nil || vttFill == nil { + t.Fatal("ASS and VTT fills must not coalesce with each other") + } + assFill.Discard() + vttFill.Discard() +} + +func TestSubtitleCacheEvictionCountsTextEntries(t *testing.T) { + c, source := newTestCache(t) + c.maxBytes = 5 + vtt := vttExtractOpts(source, 0) + fillOpts(t, c, vtt, "123456") + if _, _, ok := c.lookup(vtt); ok { + t.Fatal("VTT entry over the cache budget must be evicted") + } +} + +func TestServeExtractContentTypeOnMissAndHit(t *testing.T) { + tests := []struct { + name string + opts func(string) StreamExtractOpts + contentType string + path string + }{ + {"sup", func(source string) StreamExtractOpts { return supExtractOpts(source, 0) }, "application/octet-stream", "/sub.sup"}, + {"ass", func(source string) StreamExtractOpts { + return StreamExtractOpts{InputPath: source, SourceCodec: "ass"} + }, "text/x-ssa; charset=utf-8", "/sub.ass"}, + {"vtt", func(source string) StreamExtractOpts { return vttExtractOpts(source, 0) }, "text/vtt; charset=utf-8", "/sub.vtt"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + c, source := newTestCache(t) + opts := tc.opts(source) + extractCalls := 0 + extract := func(_ context.Context, opts StreamExtractOpts) error { + extractCalls++ + _, err := opts.Writer.Write([]byte("FULL TRACK")) + return err + } + + for request := 0; request < 2; request++ { + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, tc.path, nil) + if request == 1 && tc.name != "sup" { + req.Header.Set("Range", "bytes=5-9") + } + if err := c.ServeExtract(rec, req, opts, extract); err != nil { + t.Fatal(err) + } + if got := rec.Header().Get("Content-Type"); got != tc.contentType { + t.Fatalf("request %d Content-Type = %q, want %q", request, got, tc.contentType) + } + if tc.name != "sup" { + if rec.Code != http.StatusOK || rec.Body.String() != "FULL TRACK" { + t.Fatalf("text cache warmth changed response: code=%d body=%q", rec.Code, rec.Body.String()) + } + if rec.Header().Get("Cache-Control") != "no-store" || rec.Header().Get("Content-Length") != "" || rec.Header().Get("Last-Modified") != "" { + t.Fatalf("text response added cache-dependent HTTP headers: %v", rec.Header()) + } + } + } + if extractCalls != 1 { + t.Fatalf("extract calls = %d, want one miss then one hit", extractCalls) + } + }) + } +} + // A windowed request against a cached track must run its extract with the // cached .sup as input (small file → near-instant window) instead of // re-demuxing the original media, must never publish its sliced output as a // cache entry, and must bump the entry's LRU recency. -func TestServeSUPExtractWindowedUsesCachedTrack(t *testing.T) { +func TestServeExtractWindowedUsesCachedTrack(t *testing.T) { c, source := newTestCache(t) fillEntry(t, c, source, 0, "FULL TRACK") @@ -397,7 +585,7 @@ func TestServeSUPExtractWindowedUsesCachedTrack(t *testing.T) { extractCalls := 0 rec := httptest.NewRecorder() req := httptest.NewRequest(http.MethodGet, "/sub.sup?windowed=1&position=1200&duration=3600", nil) - err := c.ServeSUPExtract(rec, req, windowedSupOpts(source, 0, 1200, 3600), func(_ context.Context, opts StreamExtractOpts) error { + err := c.ServeExtract(rec, req, windowedSupOpts(source, 0, 1200, 3600), func(_ context.Context, opts StreamExtractOpts) error { extractCalls++ got = opts _, err := opts.Writer.Write([]byte("WINDOW SLICE")) @@ -418,8 +606,8 @@ func TestServeSUPExtractWindowedUsesCachedTrack(t *testing.T) { if got.InputPath != entry { t.Fatalf("windowed extract input = %q, want cached entry %q", got.InputPath, entry) } - if !got.InputIsExtractedSup { - t.Fatal("windowed extract from cache must set InputIsExtractedSup") + if got.ExtractedInputFormat != "sup" { + t.Fatalf("windowed extract cached input format = %q, want sup", got.ExtractedInputFormat) } if got.SeekSeconds != 1200 || got.DurationSeconds != 3600 || !got.AllowWindow { t.Fatalf("window parameters not preserved: %+v", got) @@ -445,7 +633,7 @@ func TestServeSUPExtractWindowedUsesCachedTrack(t *testing.T) { // A windowed miss must trigger exactly one detached background warm no // matter how many windowed requests arrive while it runs, and once the warm // commits, the next windowed request extracts from the cached track. -func TestServeSUPExtractWindowedMissWarmsOnce(t *testing.T) { +func TestServeExtractWindowedMissWarmsOnce(t *testing.T) { c, source := newTestCache(t) var ( @@ -473,11 +661,11 @@ func TestServeSUPExtractWindowedMissWarmsOnce(t *testing.T) { // N windowed misses: each still streams its own windowed slice from the // original file; only the first starts a warm (BeginFill coalescing keeps // the rest out — deterministic because the in-flight slot is reserved - // synchronously before ServeSUPExtract returns). + // synchronously before ServeExtract returns). for i := 0; i < 4; i++ { rec := httptest.NewRecorder() req := httptest.NewRequest(http.MethodGet, "/sub.sup?windowed=1&position=100&duration=3600", nil) - if err := c.ServeSUPExtract(rec, req, windowedSupOpts(source, 0, 100, 3600), extract); err != nil { + if err := c.ServeExtract(rec, req, windowedSupOpts(source, 0, 100, 3600), extract); err != nil { t.Fatal(err) } if rec.Body.String() != "WINDOW SLICE" { @@ -492,14 +680,14 @@ func TestServeSUPExtractWindowedMissWarmsOnce(t *testing.T) { t.Fatalf("warm extracts = %d, want exactly 1", len(warmOpts)) } warm := warmOpts[0] - if warm.InputPath != source || warm.SeekSeconds != 0 || warm.DurationSeconds != 0 || warm.AllowWindow || warm.InputIsExtractedSup { + if warm.InputPath != source || warm.SeekSeconds != 0 || warm.DurationSeconds != 0 || warm.AllowWindow || warm.ExtractedInputFormat != "" { t.Fatalf("warm must be a full-track extract of the original file: %+v", warm) } if len(windowOpts) != 4 { t.Fatalf("windowed extracts = %d, want 4", len(windowOpts)) } for _, wo := range windowOpts { - if wo.InputPath != source || wo.InputIsExtractedSup { + if wo.InputPath != source || wo.ExtractedInputFormat != "" { t.Fatalf("pre-warm windowed extract must read the original file: %+v", wo) } } @@ -508,13 +696,13 @@ func TestServeSUPExtractWindowedMissWarmsOnce(t *testing.T) { // Warm committed → the next windowed request reads the cached track. rec := httptest.NewRecorder() req := httptest.NewRequest(http.MethodGet, "/sub.sup?windowed=1&position=200&duration=3600", nil) - if err := c.ServeSUPExtract(rec, req, windowedSupOpts(source, 0, 200, 3600), extract); err != nil { + if err := c.ServeExtract(rec, req, windowedSupOpts(source, 0, 200, 3600), extract); err != nil { t.Fatal(err) } mu.Lock() last := windowOpts[len(windowOpts)-1] mu.Unlock() - if last.InputPath != entryPath(t, c, source, 0) || !last.InputIsExtractedSup { + if last.InputPath != entryPath(t, c, source, 0) || last.ExtractedInputFormat != "sup" { t.Fatalf("post-warm windowed extract must read the cached track: %+v", last) } } @@ -593,20 +781,21 @@ func TestWarmInBackgroundSkipsInFlightFill(t *testing.T) { clientFill.Discard() } -func TestServeSUPExtractDiscardsOnExtractError(t *testing.T) { +func TestServeExtractDiscardsOnExtractError(t *testing.T) { c, source := newTestCache(t) + opts := vttExtractOpts(source, 0) rec := httptest.NewRecorder() - req := httptest.NewRequest(http.MethodGet, "/sub.sup", nil) + req := httptest.NewRequest(http.MethodGet, "/sub.vtt", nil) wantErr := errors.New("ffmpeg exploded") - err := c.ServeSUPExtract(rec, req, supExtractOpts(source, 0), func(_ context.Context, opts StreamExtractOpts) error { + err := c.ServeExtract(rec, req, opts, func(_ context.Context, opts StreamExtractOpts) error { _, _ = opts.Writer.Write([]byte("PARTIAL")) return wantErr }) if !errors.Is(err, wantErr) { t.Fatalf("err = %v", err) } - if _, _, ok := c.Lookup(source, 0); ok { + if _, _, ok := c.lookup(opts); ok { t.Fatal("partial extract must not be cached") } if n := countCacheFiles(t, c); n != 0 { @@ -614,7 +803,7 @@ func TestServeSUPExtractDiscardsOnExtractError(t *testing.T) { } } -func TestServeSUPExtractNilCacheStreams(t *testing.T) { +func TestServeExtractNilCacheStreams(t *testing.T) { var c *SubtitleCache extract := func(_ context.Context, opts StreamExtractOpts) error { _, err := opts.Writer.Write([]byte("UNCACHED")) @@ -623,7 +812,7 @@ func TestServeSUPExtractNilCacheStreams(t *testing.T) { rec := httptest.NewRecorder() req := httptest.NewRequest(http.MethodGet, "/sub.sup", nil) - if err := c.ServeSUPExtract(rec, req, supExtractOpts("/nonexistent.mkv", 0), extract); err != nil { + if err := c.ServeExtract(rec, req, supExtractOpts("/nonexistent.mkv", 0), extract); err != nil { t.Fatal(err) } if rec.Body.String() != "UNCACHED" { @@ -633,7 +822,7 @@ func TestServeSUPExtractNilCacheStreams(t *testing.T) { // Windowed requests on a nil cache stream too (no lookup, no warm). rec = httptest.NewRecorder() req = httptest.NewRequest(http.MethodGet, "/sub.sup?windowed=1&position=10", nil) - if err := c.ServeSUPExtract(rec, req, windowedSupOpts("/nonexistent.mkv", 0, 10, 3600), extract); err != nil { + if err := c.ServeExtract(rec, req, windowedSupOpts("/nonexistent.mkv", 0, 10, 3600), extract); err != nil { t.Fatal(err) } if rec.Body.String() != "UNCACHED" { @@ -648,7 +837,8 @@ func entryPath(t *testing.T, c *SubtitleCache, source string, track int) string if err != nil { t.Fatal(err) } - return filepath.Join(c.dir(), subtitleCacheKey(source, track, src.ModTime(), src.Size())) + return filepath.Join(c.dir(), subtitleCacheKey(source, track, + subtitleProfile(supExtractOpts(source, track)), src.ModTime(), src.Size())) } // countCacheEntries counts committed .sup entries in the cache dir. diff --git a/internal/playback/subtitle_stream.go b/internal/playback/subtitle_stream.go index d56aa66f..5395909b 100644 --- a/internal/playback/subtitle_stream.go +++ b/internal/playback/subtitle_stream.go @@ -35,11 +35,10 @@ type StreamExtractOpts struct { // cues the client will never display. Ignored for ASS because ASS // output needs the script header that only appears at offset 0. SeekSeconds float64 - // DurationSeconds bounds the extract to a window of this length - // (passed as ffmpeg's `-t`). Zero means "until end of file". A - // bounded window lets the client consume one fetch to completion - // while keeping memory and in-flight state finite; the client - // requests subsequent windows as playback approaches the tail. + // DurationSeconds preserves the existing input-side ffmpeg `-t` argument. + // It does not actually bound subtitle extraction: ffmpeg ignores it here + // and runs from the effective seek point to EOF. Zero omits the argument. + // See streamExtractArgs for why changing this would break native clients. DurationSeconds float64 // AllowWindow lets SeekSeconds/DurationSeconds apply to PGS extracts. // By default PGS is never windowed because clients fetch the .sup @@ -49,18 +48,12 @@ type StreamExtractOpts struct { // [Script Info] header exists only at stream offset 0, so a seeked // extract would be structurally broken. AllowWindow bool - // InputIsExtractedSup marks InputPath as a cached full-track .sup - // elementary stream (a previous full extract, produced with -copyts so - // its timestamps are absolute source PTS) rather than the original - // media container. The input format is forced with `-f sup` — the - // headerless stream is probeable via its "PG" magic, but an explicit - // format is robust against probe-size edge cases — and the stream - // mapping is forced to `0:s:0`: a .sup holds exactly one stream, so - // TrackIndex (which names the ordinal in the *original* container) no - // longer applies. Seeking such an input with -copyts re-emits the same - // absolute timestamps, so windowed output is byte-compatible with a - // window cut from the original file. - InputIsExtractedSup bool + // ExtractedInputFormat marks InputPath as a cached full-track subtitle + // artifact rather than the original media container. The input demuxer is + // forced to this value and mapping is forced to `0:s:0`, because a cached + // artifact holds exactly one stream and TrackIndex names the ordinal in the + // original container. Seeking with -copyts preserves absolute source PTS. + ExtractedInputFormat string // FFmpegPath overrides the ffmpeg binary lookup. FFmpegPath string // Writer receives ffmpeg's stdout bytes as they arrive. When it @@ -136,7 +129,7 @@ func StreamExtractSubtitle(ctx context.Context, opts StreamExtractOpts) error { // streamExtractArgs builds the ffmpeg argument list for a streaming // subtitle extract. func streamExtractArgs(opts StreamExtractOpts) []string { - outCodec, outFormat := streamExtractOutput(opts.SourceCodec, opts.TargetFormat) + plan := streamExtractPlanFor(opts) args := []string{ "-hide_banner", "-nostats", "-loglevel", "error", @@ -150,34 +143,32 @@ func streamExtractArgs(opts StreamExtractOpts) []string { // from offset 0 — windowing would silently drop every cue outside // the window. Clients that manage their own sliding window opt in // via AllowWindow; -copyts below keeps the windowed output on - // absolute source timestamps so cues stay in sync. The same logic - // governs the -t duration cap below. - windowable := !IsASS(opts.SourceCodec) && (!IsPGS(opts.SourceCodec) || opts.AllowWindow) - seekApplied := opts.SeekSeconds > 0 && windowable - if seekApplied { + // absolute source timestamps so cues stay in sync. The same predicate + // controls whether the compatibility -t argument is emitted below. + if plan.seekApplied { args = append(args, "-ss", strconv.FormatFloat(opts.SeekSeconds, 'f', 3, 64)) } - // Duration limit must be an *input* option (placed before -i) so it - // caps how much of the file we read. Placed as an output option, -t - // combined with -copyts stops output when PTS reaches the given - // value — which with a non-zero seek is already in the past, so - // ffmpeg would emit only the WEBVTT header and zero cues. - if opts.DurationSeconds > 0 && windowable { + // This input-side -t is intentionally retained for byte compatibility, + // even though ffmpeg silently ignores it for these subtitle extracts and + // runs from the seek point to EOF. Moving it after -i or replacing it with + // -to would actually bound the output and break native clients that fetch + // once and depend on receiving the rest of the track. + if plan.durationApplied { args = append(args, "-t", strconv.FormatFloat(opts.DurationSeconds, 'f', 3, 64)) } - // A cached .sup input has no container magic worth probing and exactly - // one stream: force the demuxer and remap to the sole stream ordinal. + // A cached artifact has exactly one stream: force its demuxer and remap to + // the sole stream ordinal. trackIndex := opts.TrackIndex - if opts.InputIsExtractedSup { - args = append(args, "-f", "sup") + if opts.ExtractedInputFormat != "" { + args = append(args, "-f", opts.ExtractedInputFormat) trackIndex = 0 } args = append(args, "-i", opts.InputPath, "-map", fmt.Sprintf("0:s:%d", trackIndex), - "-c:s", outCodec, + "-c:s", plan.outCodec, ) // When we seek the input, preserve the absolute source timestamps @@ -185,16 +176,41 @@ func streamExtractArgs(opts StreamExtractOpts) []string { // which makes every cue play `opts.SeekSeconds` earlier than it // should — the symptom is subtitles that look "out of sync" with // the video the player is showing at the same media time. - if seekApplied { + if plan.seekApplied { args = append(args, "-copyts", "-avoid_negative_ts", "disabled") } return append(args, - "-f", outFormat, + "-f", plan.outFormat, "pipe:1", ) } +type streamExtractPlan struct { + outCodec string + outFormat string + seekApplied bool + durationApplied bool +} + +// streamExtractPlanFor is the single source of truth for both ffmpeg argv and +// cache canonicality. Any effective seek or duration makes an extract partial, +// regardless of the handler flag that caused the argument to be applied. +func streamExtractPlanFor(opts StreamExtractOpts) streamExtractPlan { + outCodec, outFormat := StreamExtractOutput(opts.SourceCodec, opts.TargetFormat) + windowable := !IsASS(opts.SourceCodec) && (!IsPGS(opts.SourceCodec) || opts.AllowWindow) + return streamExtractPlan{ + outCodec: outCodec, + outFormat: outFormat, + seekApplied: opts.SeekSeconds > 0 && windowable, + durationApplied: opts.DurationSeconds > 0 && windowable, + } +} + +func (p streamExtractPlan) partial() bool { + return p.seekApplied || p.durationApplied +} + // PGSWindowRequest reports whether a subtitle request explicitly opts in // to windowed PGS extraction (?windowed=1) and, if so, the seek position // and window duration to use. Only explicit query params count — there is @@ -247,12 +263,12 @@ func copyAndFlush(dst io.Writer, src io.Reader) error { } } -// streamExtractOutput picks the ffmpeg output codec and muxer format for +// StreamExtractOutput picks the ffmpeg output codec and muxer format for // a given source codec. ASS/SSA is copied so styling survives; PGS is // copied into a .sup elementary stream for client-side bitmap rendering // (libpgs); everything else is transmuxed to WebVTT for direct `` // consumption. -func streamExtractOutput(codec string, targetFormat ...string) (outCodec, outFormat string) { +func StreamExtractOutput(codec string, targetFormat ...string) (outCodec, outFormat string) { // A forced WebVTT target only applies to text sources: bitmap codecs // carry no text for ffmpeg's webvtt encoder, so honoring the override // would build a command that always fails mid-response. Fall through to diff --git a/internal/playback/subtitle_stream_test.go b/internal/playback/subtitle_stream_test.go index 3f7b2d0b..cdadfbcd 100644 --- a/internal/playback/subtitle_stream_test.go +++ b/internal/playback/subtitle_stream_test.go @@ -42,9 +42,9 @@ func TestStreamExtractOutput(t *testing.T) { {"mov_text", "webvtt", "webvtt"}, } for _, tc := range cases { - outCodec, outFormat := streamExtractOutput(tc.codec) + outCodec, outFormat := StreamExtractOutput(tc.codec) if outCodec != tc.wantCodec || outFormat != tc.wantFormat { - t.Errorf("streamExtractOutput(%q) = (%q, %q), want (%q, %q)", + t.Errorf("StreamExtractOutput(%q) = (%q, %q), want (%q, %q)", tc.codec, outCodec, outFormat, tc.wantCodec, tc.wantFormat) } } @@ -64,7 +64,7 @@ func TestStreamExtractArgs_TextCodecIsWindowed(t *testing.T) { t.Fatalf("text extract should seek the input: %s", joined) } if !strings.Contains(joined, "-t 600.000") { - t.Fatalf("text extract should cap the read duration: %s", joined) + t.Fatalf("text extract should preserve the duration argument: %s", joined) } if !strings.Contains(joined, "-copyts") { t.Fatalf("seeked extract must preserve source timestamps: %s", joined) @@ -99,7 +99,8 @@ func TestStreamExtractArgs_WholeTrackCodecsIgnoreWindow(t *testing.T) { } } -// A client that opts in via AllowWindow gets a seeked, duration-capped PGS +// A client that opts in via AllowWindow gets a seeked PGS extract with the +// existing duration argument and // extract with -copyts preserving absolute source timestamps — the -ss must // be an input option (before -i) so ffmpeg uses the container index. func TestStreamExtractArgs_WindowedPGS(t *testing.T) { @@ -122,7 +123,7 @@ func TestStreamExtractArgs_WindowedPGS(t *testing.T) { t.Fatalf("-ss must be an input option (before -i): %s", joined) } if !strings.Contains(joined, "-t 3600.000") { - t.Fatalf("windowed PGS extract should cap the read duration: %s", joined) + t.Fatalf("windowed PGS extract should preserve the duration argument: %s", joined) } if !strings.Contains(joined, "-copyts") { t.Fatalf("windowed PGS extract must preserve source timestamps: %s", joined) @@ -139,13 +140,13 @@ func TestStreamExtractArgs_WindowedPGS(t *testing.T) { // the cached stream's absolute timestamps survive into the output. func TestStreamExtractArgs_ExtractedSupInput(t *testing.T) { args := streamExtractArgs(StreamExtractOpts{ - InputPath: "/transcode/subtitle-cache/abc-s3-1-2.sup", - TrackIndex: 3, - SourceCodec: "hdmv_pgs_subtitle", - SeekSeconds: 1200, - DurationSeconds: 3600, - AllowWindow: true, - InputIsExtractedSup: true, + InputPath: "/transcode/subtitle-cache/abc-s3-1-2.sup", + TrackIndex: 3, + SourceCodec: "hdmv_pgs_subtitle", + SeekSeconds: 1200, + DurationSeconds: 3600, + AllowWindow: true, + ExtractedInputFormat: "sup", }) joined := strings.Join(args, " ") @@ -245,9 +246,9 @@ func TestStreamExtractOutput_TargetFormatVTTGatedToTextSources(t *testing.T) { {"hdmv_pgs_subtitle", "copy", "sup"}, } for _, tc := range cases { - outCodec, outFormat := streamExtractOutput(tc.codec, "vtt") + outCodec, outFormat := StreamExtractOutput(tc.codec, "vtt") if outCodec != tc.wantCodec || outFormat != tc.wantFormat { - t.Errorf("streamExtractOutput(%q, \"vtt\") = (%q, %q), want (%q, %q)", + t.Errorf("StreamExtractOutput(%q, \"vtt\") = (%q, %q), want (%q, %q)", tc.codec, outCodec, outFormat, tc.wantCodec, tc.wantFormat) } } diff --git a/internal/proxy/server.go b/internal/proxy/server.go index 9c5562dc..5c38aee4 100644 --- a/internal/proxy/server.go +++ b/internal/proxy/server.go @@ -252,7 +252,7 @@ func (s *Server) handleSubtitle(w http.ResponseWriter, r *http.Request) { // track when one exists (warming it in the background when not). if requestedFormat == "sup" { allowWindow, seek, duration := playback.PGSWindowRequest(r.URL.Query()) - err := s.subCache.ServeSUPExtract(w, r, playback.StreamExtractOpts{ + err := s.subCache.ServeExtract(w, r, playback.StreamExtractOpts{ InputPath: claims.MediaPath, TrackIndex: trackIndex, SourceCodec: "hdmv_pgs_subtitle", // .sup URLs are only generated for PGS tracks