2026-07-10 05:21:26 -07:00
|
|
|
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())
|
|
|
|
|
}
|
2026-08-10 18:14:49 -04:00
|
|
|
|
|
|
|
|
// 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)
|
|
|
|
|
}
|
2026-07-10 05:21:26 -07:00
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// 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
|
|
|
|
|
}
|