Files

708 lines
21 KiB
Go
Raw Permalink Normal View History

package playback
import (
"context"
"errors"
"fmt"
"io"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"sync"
"testing"
"time"
)
// newTestCache builds a cache rooted under a temp transcode dir and returns
// it with the path of a fake source media file.
func newTestCache(t *testing.T) (*SubtitleCache, string) {
t.Helper()
base := t.TempDir()
source := filepath.Join(base, "movie.mkv")
if err := os.WriteFile(source, []byte("fake mkv contents"), 0o644); err != nil {
t.Fatal(err)
}
return NewSubtitleCache(func() string { return base }), source
}
// fillEntry populates the cache for source+track with the given payload via
// 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)
if fill == nil {
t.Fatalf("BeginFill returned nil for track %d", track)
}
if _, err := fill.Tee(io.Discard).Write([]byte(payload)); err != nil {
t.Fatalf("tee write: %v", err)
}
if err := fill.Commit(); err != nil {
t.Fatalf("commit: %v", err)
}
}
// supExtractOpts builds the base extract options a handler would pass to
// ServeSUPExtract for a PGS track.
func supExtractOpts(source string, track int) StreamExtractOpts {
return StreamExtractOpts{
InputPath: source,
TrackIndex: track,
SourceCodec: "hdmv_pgs_subtitle",
}
}
// windowedSupOpts builds options for a windowed (?windowed=1) PGS request.
func windowedSupOpts(source string, track int, seek, duration float64) StreamExtractOpts {
opts := supExtractOpts(source, track)
opts.AllowWindow = true
opts.SeekSeconds = seek
opts.DurationSeconds = duration
return opts
}
// waitForCacheEntry polls until the cache holds a committed entry for
// source+track — used to observe asynchronous background warms.
func waitForCacheEntry(t *testing.T, c *SubtitleCache, source string, track int) {
t.Helper()
deadline := time.Now().Add(5 * time.Second)
for time.Now().Before(deadline) {
if f, _, ok := c.Lookup(source, track); ok {
_ = f.Close()
return
}
time.Sleep(5 * time.Millisecond)
}
t.Fatalf("cache entry for track %d never appeared", track)
}
func readAllAndClose(t *testing.T, f *os.File) string {
t.Helper()
defer f.Close()
data, err := io.ReadAll(f)
if err != nil {
t.Fatal(err)
}
return string(data)
}
func TestSubtitleCacheMissThenHit(t *testing.T) {
c, source := newTestCache(t)
if _, _, ok := c.Lookup(source, 0); ok {
t.Fatal("expected miss on empty cache")
}
fillEntry(t, c, source, 0, "PGS DATA TRACK 0")
f, modTime, ok := c.Lookup(source, 0)
if !ok {
t.Fatal("expected hit after commit")
}
if got := readAllAndClose(t, f); got != "PGS DATA TRACK 0" {
t.Fatalf("cached content = %q", got)
}
src, err := os.Stat(source)
if err != nil {
t.Fatal(err)
}
if !modTime.Equal(src.ModTime()) {
t.Fatalf("hit modTime = %v, want source mtime %v", modTime, src.ModTime())
}
// A different track ordinal is a distinct entry.
if _, _, ok := c.Lookup(source, 1); ok {
t.Fatal("expected miss for uncached track ordinal")
}
}
func TestSubtitleCacheInvalidatedBySourceMtime(t *testing.T) {
c, source := newTestCache(t)
fillEntry(t, c, source, 0, "old extract")
newTime := time.Now().Add(2 * time.Hour)
if err := os.Chtimes(source, newTime, newTime); err != nil {
t.Fatal(err)
}
if _, _, ok := c.Lookup(source, 0); ok {
t.Fatal("expected miss after source mtime changed")
}
// Re-filling under the new source identity overwrites, and the stale
// sibling entry is cleaned up.
fillEntry(t, c, source, 0, "new extract")
f, _, ok := c.Lookup(source, 0)
if !ok {
t.Fatal("expected hit after refill")
}
if got := readAllAndClose(t, f); got != "new extract" {
t.Fatalf("cached content = %q", got)
}
if n := countCacheEntries(t, c); n != 1 {
t.Fatalf("stale sibling not removed: %d entries", n)
}
}
func TestSubtitleCacheInvalidatedBySourceSize(t *testing.T) {
c, source := newTestCache(t)
fillEntry(t, c, source, 0, "old extract")
src, err := os.Stat(source)
if err != nil {
t.Fatal(err)
}
if err := os.WriteFile(source, []byte("different length contents!"), 0o644); err != nil {
t.Fatal(err)
}
// Restore the original mtime so only size differs.
if err := os.Chtimes(source, src.ModTime(), src.ModTime()); err != nil {
t.Fatal(err)
}
if _, _, ok := c.Lookup(source, 0); ok {
t.Fatal("expected miss after source size changed")
}
}
func TestSubtitleCacheDiscardLeavesNothing(t *testing.T) {
c, source := newTestCache(t)
fill := c.BeginFill(source, 0)
if fill == nil {
t.Fatal("BeginFill returned nil")
}
if _, err := fill.Tee(io.Discard).Write([]byte("partial byt")); err != nil {
t.Fatal(err)
}
fill.Discard()
if _, _, ok := c.Lookup(source, 0); ok {
t.Fatal("discarded fill must not be served")
}
if n := countCacheFiles(t, c); n != 0 {
t.Fatalf("discard left %d files (temp not removed?)", n)
}
}
func TestSubtitleCacheCommitRefusesChangedSource(t *testing.T) {
c, source := newTestCache(t)
fill := c.BeginFill(source, 0)
if fill == nil {
t.Fatal("BeginFill returned nil")
}
if _, err := fill.Tee(io.Discard).Write([]byte("extract from old source")); err != nil {
t.Fatal(err)
}
// Source replaced mid-extract.
newTime := time.Now().Add(time.Hour)
if err := os.Chtimes(source, newTime, newTime); err != nil {
t.Fatal(err)
}
if err := fill.Commit(); err == nil {
t.Fatal("Commit must refuse when source changed mid-fill")
}
if n := countCacheFiles(t, c); n != 0 {
t.Fatalf("refused commit left %d files", n)
}
}
func TestSubtitleCacheTeeWriteFailureKeepsServingClient(t *testing.T) {
c, source := newTestCache(t)
fill := c.BeginFill(source, 0)
if fill == nil {
t.Fatal("BeginFill returned nil")
}
// Force temp-file writes to fail (simulates disk full).
_ = fill.tmp.Close()
var client strings.Builder
n, err := fill.Tee(&client).Write([]byte("bytes for the viewer"))
if err != nil || n != len("bytes for the viewer") {
t.Fatalf("client write must succeed despite cache failure: n=%d err=%v", n, err)
}
if client.String() != "bytes for the viewer" {
t.Fatalf("client got %q", client.String())
}
if err := fill.Commit(); err == nil {
t.Fatal("Commit must fail after tee write error")
}
if _, _, ok := c.Lookup(source, 0); ok {
t.Fatal("failed fill must not be served")
}
}
func TestSubtitleCacheEvictionUnderCap(t *testing.T) {
c, source := newTestCache(t)
c.maxBytes = 25 // each payload below is 10 bytes
base := time.Now().Add(-time.Hour)
for track := 0; track < 3; track++ {
fillEntry(t, c, source, track, fmt.Sprintf("0123456%03d", track))
// Pin distinct LRU mtimes: track 0 oldest.
path := entryPath(t, c, source, track)
mt := base.Add(time.Duration(track) * time.Minute)
if err := os.Chtimes(path, mt, mt); err != nil {
t.Fatal(err)
}
}
// 4th commit (10 bytes) pushes the total to 40 > 25; eviction must
// remove the two oldest entries (tracks 0 and 1) to get back to 20.
fillEntry(t, c, source, 3, "0123456003")
for track, want := range map[int]bool{0: false, 1: false, 2: true, 3: true} {
_, _, ok := c.Lookup(source, track)
if ok != want {
t.Errorf("track %d cached = %v, want %v", track, ok, want)
}
}
}
func TestSubtitleCacheCoalescing(t *testing.T) {
c, source := newTestCache(t)
first := c.BeginFill(source, 0)
if first == nil {
t.Fatal("first BeginFill returned nil")
}
if second := c.BeginFill(source, 0); second != nil {
second.Discard()
t.Fatal("second BeginFill for in-flight track must return nil")
}
// A different track is independent.
other := c.BeginFill(source, 1)
if other == nil {
t.Fatal("BeginFill for a different track must not be blocked")
}
other.Discard()
if _, err := first.Tee(io.Discard).Write([]byte("data")); err != nil {
t.Fatal(err)
}
if err := first.Commit(); err != nil {
t.Fatal(err)
}
// Slot released after commit.
if again := c.BeginFill(source, 0); again == nil {
t.Fatal("BeginFill must work again after Commit")
} else {
again.Discard()
}
}
func TestSubtitleCacheCoalescingConcurrent(t *testing.T) {
c, source := newTestCache(t)
const workers = 16
var (
wg sync.WaitGroup
mu sync.Mutex
fills []*SubtitleCacheFill
)
start := make(chan struct{})
for i := 0; i < workers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
<-start
if f := c.BeginFill(source, 0); f != nil {
mu.Lock()
fills = append(fills, f)
mu.Unlock()
}
}()
}
close(start)
wg.Wait()
if len(fills) != 1 {
t.Fatalf("exactly one concurrent BeginFill must win, got %d", len(fills))
}
fills[0].Discard()
}
func TestServeSUPExtractCacheFlow(t *testing.T) {
c, source := newTestCache(t)
extractCalls := 0
extract := func(_ context.Context, opts StreamExtractOpts) error {
extractCalls++
_, err := opts.Writer.Write([]byte("SUP PAYLOAD"))
return err
}
// 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 {
t.Fatal(err)
}
if extractCalls != 1 {
t.Fatalf("extract calls = %d", extractCalls)
}
if rec.Body.String() != "SUP PAYLOAD" {
t.Fatalf("miss body = %q", rec.Body.String())
}
if cc := rec.Header().Get("Cache-Control"); cc != "no-store" {
t.Fatalf("miss Cache-Control = %q", cc)
}
// Second request: hit → served from cache, no extract, revalidatable.
rec = httptest.NewRecorder()
if err := c.ServeSUPExtract(rec, req, supExtractOpts(source, 0), extract); err != nil {
t.Fatal(err)
}
if extractCalls != 1 {
t.Fatal("cache hit must not invoke extract")
}
if rec.Body.String() != "SUP PAYLOAD" {
t.Fatalf("hit body = %q", rec.Body.String())
}
if cc := rec.Header().Get("Cache-Control"); cc != "private, no-cache" {
t.Fatalf("hit Cache-Control = %q", cc)
}
if rec.Header().Get("Last-Modified") == "" {
t.Fatal("hit must carry Last-Modified")
}
if cl := rec.Header().Get("Content-Length"); cl != "11" {
t.Fatalf("hit Content-Length = %q", cl)
}
// Range request against the cached entry.
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 {
t.Fatal(err)
}
if rec.Code != http.StatusPartialContent || rec.Body.String() != "PAYLOAD" {
t.Fatalf("range: code=%d body=%q", rec.Code, rec.Body.String())
}
// HEAD uses the same cached binary representation and metadata without
// invoking a fresh extract or returning a body.
rec = httptest.NewRecorder()
headReq := httptest.NewRequest(http.MethodHead, "/sub.sup", nil)
if err := c.ServeSUPExtract(rec, headReq, supExtractOpts(source, 0), extract); err != nil {
t.Fatal(err)
}
if rec.Code != http.StatusOK || rec.Body.Len() != 0 {
t.Fatalf("head: code=%d body=%q", rec.Code, rec.Body.String())
}
if extractCalls != 1 {
t.Fatal("cached HEAD must not invoke extract")
}
if got := rec.Header().Get("Content-Type"); got != "application/octet-stream" {
t.Fatalf("head Content-Type = %q", got)
}
if got := rec.Header().Get("Content-Length"); got != "11" {
t.Fatalf("head Content-Length = %q", got)
}
if got := rec.Header().Get("Accept-Ranges"); got != "bytes" {
t.Fatalf("head Accept-Ranges = %q", got)
}
}
// 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) {
c, source := newTestCache(t)
fillEntry(t, c, source, 0, "FULL TRACK")
// Age the entry so the LRU recency bump is observable.
entry := entryPath(t, c, source, 0)
old := time.Now().Add(-time.Hour)
if err := os.Chtimes(entry, old, old); err != nil {
t.Fatal(err)
}
var got StreamExtractOpts
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 {
extractCalls++
got = opts
_, err := opts.Writer.Write([]byte("WINDOW SLICE"))
return err
})
if err != nil {
t.Fatal(err)
}
if extractCalls != 1 {
t.Fatalf("extract calls = %d", extractCalls)
}
if rec.Body.String() != "WINDOW SLICE" {
t.Fatalf("windowed body = %q", rec.Body.String())
}
if cc := rec.Header().Get("Cache-Control"); cc != "no-store" {
t.Fatalf("windowed Cache-Control = %q", cc)
}
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.SeekSeconds != 1200 || got.DurationSeconds != 3600 || !got.AllowWindow {
t.Fatalf("window parameters not preserved: %+v", got)
}
// The full-track entry must be untouched, with recency bumped.
f, _, ok := c.Lookup(source, 0)
if !ok {
t.Fatal("full-track entry lost")
}
if content := readAllAndClose(t, f); content != "FULL TRACK" {
t.Fatalf("full-track entry corrupted: %q", content)
}
info, err := os.Stat(entry)
if err != nil {
t.Fatal(err)
}
if !info.ModTime().After(old.Add(time.Minute)) {
t.Fatalf("windowed serve must bump LRU recency: mtime = %v", info.ModTime())
}
}
// 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) {
c, source := newTestCache(t)
var (
mu sync.Mutex
warmOpts []StreamExtractOpts
windowOpts []StreamExtractOpts
warmRelease = make(chan struct{})
)
extract := func(_ context.Context, opts StreamExtractOpts) error {
if opts.AllowWindow {
mu.Lock()
windowOpts = append(windowOpts, opts)
mu.Unlock()
_, err := opts.Writer.Write([]byte("WINDOW SLICE"))
return err
}
mu.Lock()
warmOpts = append(warmOpts, opts)
mu.Unlock()
<-warmRelease
_, err := opts.Writer.Write([]byte("FULL TRACK"))
return err
}
// 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).
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 {
t.Fatal(err)
}
if rec.Body.String() != "WINDOW SLICE" {
t.Fatalf("windowed body = %q", rec.Body.String())
}
}
close(warmRelease)
waitForCacheEntry(t, c, source, 0)
mu.Lock()
if len(warmOpts) != 1 {
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 {
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 {
t.Fatalf("pre-warm windowed extract must read the original file: %+v", wo)
}
}
mu.Unlock()
// 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 {
t.Fatal(err)
}
mu.Lock()
last := windowOpts[len(windowOpts)-1]
mu.Unlock()
if last.InputPath != entryPath(t, c, source, 0) || !last.InputIsExtractedSup {
t.Fatalf("post-warm windowed extract must read the cached track: %+v", last)
}
}
// Warms beyond the server-wide slot budget are dropped, not queued, and a
// dropped warm must not leave an in-flight reservation behind.
func TestWarmInBackgroundSemaphoreDrop(t *testing.T) {
c, source := newTestCache(t)
release := make(chan struct{})
extract := func(_ context.Context, opts StreamExtractOpts) error {
<-release
_, err := opts.Writer.Write([]byte("FULL TRACK"))
return err
}
// Occupy every warm slot (slots are acquired synchronously).
for track := 0; track < subtitleCacheWarmSlots; track++ {
c.WarmInBackground(supExtractOpts(source, track), extract)
}
// One more: dropped without reserving the track's in-flight slot.
overflow := subtitleCacheWarmSlots
c.WarmInBackground(supExtractOpts(source, overflow), extract)
if fill := c.BeginFill(source, overflow); fill == nil {
t.Fatal("dropped warm must not hold the in-flight slot")
} else {
fill.Discard()
}
close(release)
for track := 0; track < subtitleCacheWarmSlots; track++ {
waitForCacheEntry(t, c, source, track)
}
if _, _, ok := c.Lookup(source, overflow); ok {
t.Fatal("dropped warm must not populate the cache")
}
// With slots free again, the overflow track's warm goes through.
c.WarmInBackground(supExtractOpts(source, overflow), extract)
waitForCacheEntry(t, c, source, overflow)
}
// A warm that races an already-in-flight client fill must skip (BeginFill
// coalescing) and release its warm slot for other tracks.
func TestWarmInBackgroundSkipsInFlightFill(t *testing.T) {
c, source := newTestCache(t)
clientFill := c.BeginFill(source, 0)
if clientFill == nil {
t.Fatal("BeginFill returned nil")
}
warmed := make(chan struct{}, 1)
c.WarmInBackground(supExtractOpts(source, 0), func(_ context.Context, opts StreamExtractOpts) error {
warmed <- struct{}{}
_, err := opts.Writer.Write([]byte("WARM"))
return err
})
// The skipped warm must have released its slot synchronously: all
// subtitleCacheWarmSlots slots are still available.
for track := 1; track <= subtitleCacheWarmSlots; track++ {
c.WarmInBackground(supExtractOpts(source, track), func(_ context.Context, opts StreamExtractOpts) error {
_, err := opts.Writer.Write([]byte("FULL TRACK"))
return err
})
}
for track := 1; track <= subtitleCacheWarmSlots; track++ {
waitForCacheEntry(t, c, source, track)
}
select {
case <-warmed:
t.Fatal("warm for an in-flight track must not run")
default:
}
clientFill.Discard()
}
func TestServeSUPExtractDiscardsOnExtractError(t *testing.T) {
c, source := newTestCache(t)
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/sub.sup", nil)
wantErr := errors.New("ffmpeg exploded")
err := c.ServeSUPExtract(rec, req, supExtractOpts(source, 0), 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 {
t.Fatal("partial extract must not be cached")
}
if n := countCacheFiles(t, c); n != 0 {
t.Fatalf("failed extract left %d files", n)
}
}
func TestServeSUPExtractNilCacheStreams(t *testing.T) {
var c *SubtitleCache
extract := func(_ context.Context, opts StreamExtractOpts) error {
_, err := opts.Writer.Write([]byte("UNCACHED"))
return err
}
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/sub.sup", nil)
if err := c.ServeSUPExtract(rec, req, supExtractOpts("/nonexistent.mkv", 0), extract); err != nil {
t.Fatal(err)
}
if rec.Body.String() != "UNCACHED" {
t.Fatalf("body = %q", rec.Body.String())
}
// 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 {
t.Fatal(err)
}
if rec.Body.String() != "UNCACHED" {
t.Fatalf("windowed body = %q", rec.Body.String())
}
}
// entryPath computes the committed entry path for source+track.
func entryPath(t *testing.T, c *SubtitleCache, source string, track int) string {
t.Helper()
src, err := os.Stat(source)
if err != nil {
t.Fatal(err)
}
return filepath.Join(c.dir(), subtitleCacheKey(source, track, src.ModTime(), src.Size()))
}
// countCacheEntries counts committed .sup entries in the cache dir.
func countCacheEntries(t *testing.T, c *SubtitleCache) int {
t.Helper()
return countMatching(t, c, func(name string) bool {
return strings.HasSuffix(name, ".sup") && !strings.Contains(name, ".part-")
})
}
// countCacheFiles counts every file in the cache dir, temp files included.
func countCacheFiles(t *testing.T, c *SubtitleCache) int {
t.Helper()
return countMatching(t, c, func(string) bool { return true })
}
func countMatching(t *testing.T, c *SubtitleCache, match func(string) bool) int {
t.Helper()
entries, err := os.ReadDir(c.dir())
if os.IsNotExist(err) {
return 0
}
if err != nil {
t.Fatal(err)
}
n := 0
for _, e := range entries {
if match(e.Name()) {
n++
}
}
return n
}