From 37c8807c7a9b69b1b0b33dd1faa134d5a046e5ab Mon Sep 17 00:00:00 2001 From: randrini Date: Tue, 6 Oct 2026 06:12:28 +0200 Subject: [PATCH 1/4] WIP: #227 guarded stale-pin recovery (partial, aborted lane) --- internal/api/handlers/playback_v3.go | 12 +- .../handlers/playback_virtual_stale_pin.go | 246 ++++++++++++++ .../playback_virtual_stale_pin_test.go | 318 ++++++++++++++++++ 3 files changed, 575 insertions(+), 1 deletion(-) create mode 100644 internal/api/handlers/playback_virtual_stale_pin.go create mode 100644 internal/api/handlers/playback_virtual_stale_pin_test.go diff --git a/internal/api/handlers/playback_v3.go b/internal/api/handlers/playback_v3.go index 39a8f098e8..8c5ea4a7b6 100644 --- a/internal/api/handlers/playback_v3.go +++ b/internal/api/handlers/playback_v3.go @@ -2631,11 +2631,21 @@ func (h *PlaybackHandler) resolveVirtualStartWithVersionFallback( // elapses rather than waiting out the full cold-start budget per version. walkCtx, cancel := context.WithTimeout(r.Context(), virtualStartVersionFallbackDecisionBudget) defer cancel() + walkReq := r.WithContext(walkCtx) + + // Guarded stale-pin recovery for identity-less rows: when the pin id is + // absent from a non-empty live listing and the row carries no durable + // provider identity to rematch on, re-pin in memory to the best + // fingerprint-matched live candidate and retry once. Bounded to one attempt; + // a miss falls through to the alternate walk below with the terminal intact. + if recovered, recoveredOK := h.recoverStaleIdentityLessPinV3(walkReq, file, profileID, req, bandwidthCapKbps); recoveredOK { + return recovered, nil + } + alternates, alternateErr := h.findAlternateFiles(walkCtx, file, alternateOrderingForClient(req.Capabilities)) if alternateErr != nil || len(alternates) == 0 { return resolved, resolveErr } - walkReq := r.WithContext(walkCtx) // candidateRows keeps one entry per walkable alternate, in fallback order. // Skipping an AltMount-failed version here (rather than inside the resolve diff --git a/internal/api/handlers/playback_virtual_stale_pin.go b/internal/api/handlers/playback_virtual_stale_pin.go new file mode 100644 index 0000000000..231995a983 --- /dev/null +++ b/internal/api/handlers/playback_virtual_stale_pin.go @@ -0,0 +1,246 @@ +package handlers + +import ( + "context" + "log/slog" + "net/http" + "os" + "strings" + "time" + + apimw "github.com/Silo-Server/silo-server/internal/api/middleware" + "github.com/Silo-Server/silo-server/internal/models" + "github.com/Silo-Server/silo-server/internal/playback" +) + +// Guarded stale-pin recovery for identity-less rows. +// +// Background: a catalog row pinned to a concrete `?result=` id can go stale +// within minutes when the provider renumbers its per-listing result ids. When +// the row carries durable provider identity the resolver re-identifies the same +// release under its new id (the same-release rematch in +// internal/virtuallibrary/resolve.go). An identity-less row has nothing to rematch +// on, so a non-empty listing whose only change is the result id used to terminal +// the cold start even though the release was still listed. +// +// This recovery fires only in the cold-start version-fallback walk (never on the +// session-bound serve/replan path, which must keep refusing) when ALL of: +// - the row is identity-less (no video hash / GUID / release name); +// - the live provider listing is non-empty (an EMPTY listing is a transient +// blackout, not a renumber, and keeps its existing terminal); +// - the bound pin id is absent from that listing (otherwise the normal resolve +// owns the candidate); +// - a live candidate for the SAME content (same provider-neutral path, so never +// across a different content_id) matches the row's stored evidence +// fingerprint: size within virtualStalePinSizeTolerance, an equal video codec, +// and a probed duration within virtualStalePinDurationTolerance when both are +// known. +// +// When it fires it re-pins IN MEMORY ONLY (the durable catalog row's ?result= is +// never rewritten) and resolves the matched candidate once. If no candidate +// matches, the caller's terminal is preserved. +// +// Mismatch-rate concern: a fingerprint is evidence, not proof. A renumbered +// release whose size drifted past the tolerance, or two releases of one title +// with coincidentally equal size+codec+duration, could be confused. Adopting the +// wrong bytes is worse than a terminal, so the recovery is deliberately one-shot +// and kill-switchable. Every recovery logs at Info with the matched ids; if the +// observed mismatch rate exceeds ~1%, disable it with the kill switch below. +// +// Kill switch: set SILO_DISABLE_VIRTUAL_STALE_PIN_RECOVERY=1 at process start to +// disable the recovery entirely (fast, default-on). Tests override the var +// directly. +var virtualStalePinRecoveryEnabled = os.Getenv("SILO_DISABLE_VIRTUAL_STALE_PIN_RECOVERY") == "" + +const ( + // virtualStalePinSizeTolerance is the maximum relative drift allowed between + // the row's stored file size and the live candidate's declared size. Provider + // display sizes are rounded, so an exact match is too strict; 5% absorbs the + // rounding without admitting a different release. + virtualStalePinSizeTolerance = 0.05 + // virtualStalePinDurationTolerance is the maximum relative drift allowed + // between the row's stored duration and the candidate's probed duration. It is + // tighter than the runtime tolerance used for probe plausibility because this + // gate authorizes adopting different bytes. + virtualStalePinDurationTolerance = 0.02 + // virtualStalePinListBudget bounds the one provider listing the recovery pays. + // The walk's own decision budget still bounds the whole start. + virtualStalePinListBudget = 5 * time.Second +) + +// recoverStaleIdentityLessPinV3 attempts the guarded in-memory re-pin described +// above. It returns ok=true only when a fingerprint-matched live candidate was +// resolved; any other outcome leaves the caller's terminal in place. It performs +// exactly one candidate resolve (no loops) and never mutates the catalog pin. +func (h *PlaybackHandler) recoverStaleIdentityLessPinV3( + r *http.Request, + file *models.MediaFile, + profileID string, + req playback.StartRequestV3, + bandwidthCapKbps int, +) (resolvedVirtualPlaybackSource, bool) { + if !virtualStalePinRecoveryEnabled || h == nil || file == nil || r == nil { + return resolvedVirtualPlaybackSource{}, false + } + if h.VirtualPlaybackStreamLister == nil || h.VirtualMediaDetailedResolver == nil { + return resolvedVirtualPlaybackSource{}, false + } + pinnedID := virtualResultCandidateID(file.FilePath) + if pinnedID == "" { + // A neutral row has no pin to go stale. + return resolvedVirtualPlaybackSource{}, false + } + // Identity-less only. A row with durable identity is already re-identified + // by the resolver's same-release rematch; running here too would second-guess + // a stronger signal with a weaker one. + if _, hasIdentity := persistedVirtualIdentity(file); hasIdentity { + return resolvedVirtualPlaybackSource{}, false + } + neutralKey := virtualPlaybackNeutralKey(file.FilePath) + if neutralKey == "" || neutralKey == file.FilePath { + return resolvedVirtualPlaybackSource{}, false + } + ctx := r.Context() + userID := apimw.GetUserID(ctx) + listCtx, cancel := context.WithTimeout(ctx, virtualStalePinListBudget) + streams, err := h.VirtualPlaybackStreamLister.ListVirtualPlaybackStreams( + listCtx, neutralKey, userID, profileID, file.VirtualOwnerInstallationID, + ) + cancel() + if err != nil || len(streams) == 0 { + // An empty (or failed) listing is a transient outage, not a renumber. + // Preserve the caller's terminal. + return resolvedVirtualPlaybackSource{}, false + } + // The pin must be absent from the live listing. If it is still listed the + // ordinary resolve owns it and a recovery would only mask a different fault. + for _, stream := range streams { + if stream.ID == pinnedID || stream.URI == file.FilePath { + return resolvedVirtualPlaybackSource{}, false + } + } + // Same-content candidates only: filterVirtualPlaybackStreams drops any + // candidate whose scheme/host/path differs from the row's, so an adoption + // can never cross a content_id (or a provider). + candidates := filterVirtualPlaybackStreams(file, streams) + if len(candidates) == 0 { + return resolvedVirtualPlaybackSource{}, false + } + match, found := virtualStalePinBestStreamMatch(file, candidates) + if !found { + slog.DebugContext(ctx, "virtual stale-pin recovery found no fingerprint match", + "component", "api", "file_id", file.ID, "old_candidate_id", pinnedID, + "candidate_count", len(candidates)) + return resolvedVirtualPlaybackSource{}, false + } + // Re-pin in memory and resolve the matched candidate exactly once. The row's + // durable ?result= is never rewritten; only process state moves. + matchedID := virtualResultCandidateID(match.FilePath) + recovered, resolveErr := h.resolveVirtualPlaybackSource( + r, &match, profileID, false, nil, "", req.QualityPreference, bandwidthCapKbps, false, + virtualResolveOptionsV3{sessionBound: false, bypassProviderFloor: true}, + ) + if resolveErr != nil || recovered.File == nil { + return resolvedVirtualPlaybackSource{}, false + } + // Refuse a substitution: the recovery authorized serving the exact matched + // candidate, so a resolve that returned a different result id (a sibling the + // resolver preferred) must not be adopted under the re-pin. This keeps the + // recovery's blast radius to the candidate the fingerprint actually matched. + if got := virtualResultCandidateID(recovered.URI); got != "" && matchedID != "" && got != matchedID { + slog.DebugContext(ctx, "virtual stale-pin recovery refused a substituted candidate", + "component", "api", "file_id", file.ID, "old_candidate_id", pinnedID, + "matched_candidate_id", matchedID, "resolved_candidate_id", got) + return resolvedVirtualPlaybackSource{}, false + } + if !virtualStalePinDurationMatch(file.Duration, recovered.File.Duration) { + slog.DebugContext(ctx, "virtual stale-pin recovery rejected a duration mismatch", + "component", "api", "file_id", file.ID, "old_candidate_id", pinnedID, + "new_candidate_id", virtualResultCandidateID(recovered.URI), + "row_duration", file.Duration, "candidate_duration", recovered.File.Duration) + return resolvedVirtualPlaybackSource{}, false + } + slog.InfoContext(ctx, "virtual stale-pin recovery re-pinned an identity-less candidate in memory", + "component", "api", "file_id", file.ID, "old_candidate_id", pinnedID, + "new_candidate_id", virtualResultCandidateID(recovered.URI), + "virtual_uri", recovered.URI) + return recovered, true +} + +// virtualStalePinBestStreamMatch returns the best live candidate whose declared +// evidence matches the row's stored fingerprint, re-pinned as a copy of the row. +// "Best" is the first match in provider rank order: the listing is already +// ranked, so the first fingerprint match is the strongest one, and a single +// deterministic choice keeps the recovery one-shot. It reports ok=false when no +// candidate matches. +func virtualStalePinBestStreamMatch(row *models.MediaFile, candidates []VirtualPlaybackStream) (models.MediaFile, bool) { + for _, candidate := range candidates { + if !virtualStalePinStreamFingerprintMatch(row, candidate) { + continue + } + match := *row + match.FilePath = candidate.URI + // Keep the row's own identity for the in-memory re-pin: the live candidate + // has no catalog row yet, and adopting its row id would fabricate one. Only + // the served path changes. + match.VirtualOwnerInstallationID = effectiveVirtualOwner(candidate.OwnerInstallationID, row.VirtualOwnerInstallationID) + return match, true + } + return models.MediaFile{}, false +} + +// virtualStalePinStreamFingerprintMatch is the cheap half of the fingerprint, +// evaluated on the live listing record before any resolve: the size must be +// known on both sides and agree within tolerance, and the video codec must be +// known on both sides and match. An audio codec mismatch is also a veto when both +// sides carry one; an unknown audio codec is neutral, since some providers omit +// it. +func virtualStalePinStreamFingerprintMatch(row *models.MediaFile, candidate VirtualPlaybackStream) bool { + if row == nil { + return false + } + if row.FileSize <= 0 || candidate.FileSize <= 0 || + !virtualStalePinWithinTolerance(row.FileSize, candidate.FileSize, virtualStalePinSizeTolerance) { + return false + } + rowVideo := strings.TrimSpace(row.CodecVideo) + candidateVideo := strings.TrimSpace(candidate.CodecVideo) + if rowVideo == "" || candidateVideo == "" || !strings.EqualFold(rowVideo, candidateVideo) { + return false + } + rowAudio := strings.TrimSpace(row.CodecAudio) + candidateAudio := strings.TrimSpace(candidate.CodecAudio) + if rowAudio != "" && candidateAudio != "" && !strings.EqualFold(rowAudio, candidateAudio) { + return false + } + return true +} + +// virtualStalePinDurationMatch applies the duration half of the fingerprint to a +// probed candidate. It is neutral when either duration is unknown (a probe that +// reported no runtime), so the size and codec tiers decide; when both are known +// the drift must stay inside the tolerance. +func virtualStalePinDurationMatch(rowDuration, candidateDuration int) bool { + if rowDuration <= 0 || candidateDuration <= 0 { + return true + } + return virtualStalePinWithinTolerance(int64(rowDuration), int64(candidateDuration), virtualStalePinDurationTolerance) +} + +// virtualStalePinWithinTolerance reports whether two positive quantities agree +// within a relative tolerance, using the larger as the denominator so the test +// is symmetric. +func virtualStalePinWithinTolerance(a, b int64, tolerance float64) bool { + if a <= 0 || b <= 0 { + return false + } + diff := a - b + if diff < 0 { + diff = -diff + } + largest := a + if b > largest { + largest = b + } + return float64(diff)/float64(largest) <= tolerance +} diff --git a/internal/api/handlers/playback_virtual_stale_pin_test.go b/internal/api/handlers/playback_virtual_stale_pin_test.go new file mode 100644 index 0000000000..8b97b5bddb --- /dev/null +++ b/internal/api/handlers/playback_virtual_stale_pin_test.go @@ -0,0 +1,318 @@ +package handlers + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + + "github.com/Silo-Server/silo-server/internal/models" + "github.com/Silo-Server/silo-server/internal/playback" +) + +// stalePinRow builds an identity-less virtual row. FileSize/CodecVideo/Duration +// are the fingerprint tiers; probe evidence is intentionally absent so the P0 +// fast path cannot serve the row. +func stalePinRow(id int, contentID, neutralURI, resultID string) *models.MediaFile { + return &models.MediaFile{ + ID: id, + ContentID: contentID, + FilePath: neutralURI + "?result=" + resultID, + Container: "mkv", + CodecVideo: "h264", + CodecAudio: "aac", + Resolution: "1080p", + VirtualOwnerInstallationID: 5, + FileSize: 8_500_000_000, + Duration: 7200, + } +} + +// stalePinLiveStream is a live listing record for the same content under a new +// result id. FileSize and codec are overridable so a test can force a mismatch. +func stalePinLiveStream(neutralURI, resultID string, fileSize int64, videoCodec string) VirtualPlaybackStream { + return VirtualPlaybackStream{ + ID: resultID, URI: neutralURI + "?result=" + resultID, + Resolution: "1080p", CodecVideo: videoCodec, CodecAudio: "aac", + Container: "mkv", FileSize: fileSize, + } +} + +// stalePinRecoveryHandler wires the recovery's two collaborators: the lister +// answers a fixed live set, and the resolver records every call and succeeds for +// any non-pinned candidate (or fails for all, when failAll is set). +func stalePinRecoveryHandler(live []VirtualPlaybackStream, pinnedURI string, failAll bool) (*PlaybackHandler, *atomic.Int32, *atomic.Int32) { + var resolveCalls, listCalls atomic.Int32 + h := &PlaybackHandler{ + VirtualPlaybackResolver: VirtualPlaybackResolverFunc(func(context.Context, string, int, string, int) (string, error) { + return "", errors.New("simple resolver must not be used when the detailed resolver is set") + }), + VirtualPlaybackStreamLister: VirtualPlaybackStreamListerFunc(func(context.Context, string, int, string, int) ([]VirtualPlaybackStream, error) { + listCalls.Add(1) + return live, nil + }), + VirtualMediaDetailedResolver: VirtualMediaDetailedResolverFunc(func(_ context.Context, uri string, _ int, _ int, _ string, _ bool, _ []string, _ string) (ResolvedVirtualMedia, error) { + resolveCalls.Add(1) + if failAll || uri == pinnedURI { + return ResolvedVirtualMedia{}, absentSessionPinError("STALE") + } + return ResolvedVirtualMedia{URL: "http://127.0.0.1:8080/stream?uri=" + uri, URI: uri, CandidateID: "NEW"}, nil + }), + } + return h, &resolveCalls, &listCalls +} + +func stalePinRecoveryRequest() *http.Request { + return httptest.NewRequest(http.MethodPost, "/api/v2/playback/start", nil) +} + +func stalePinRequest() playback.StartRequestV3 { + return playback.StartRequestV3{QualityPreference: "auto"} +} + +// (a) A dead pin absent from a NON-EMPTY listing, on an identity-less row, with a +// live fingerprint match recovers: the recovery resolves the matched candidate +// exactly once and returns it, in memory only. +func TestStalePinRecoveryRecoversFingerprintMatch(t *testing.T) { + const neutral = "virtual://movie/tt-stale-recover" + pinnedURI := neutral + "?result=STALE" + row := stalePinRow(1, "movie-stale-recover", neutral, "STALE") + live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} + h, resolveCalls, _ := stalePinRecoveryHandler(live, pinnedURI, false) + + recovered, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1", stalePinRequest(), 0) + if !ok { + t.Fatal("a fingerprint-matched live candidate must recover the pin") + } + if recovered.File == nil || recovered.URI != neutral+"?result=NEW" { + t.Fatalf("recovered = %#v, want the fingerprint-matched live candidate", recovered) + } + if got := virtualResultCandidateID(recovered.URI); got != "NEW" { + t.Fatalf("recovered candidate = %q, want NEW", got) + } + if got := resolveCalls.Load(); got != 1 { + t.Fatalf("resolver calls = %d, want exactly 1 (one recovery retry, no loop)", got) + } + // The row's durable pin is never rewritten: only the returned copy changed. + if got := virtualResultCandidateID(row.FilePath); got != "STALE" { + t.Fatalf("catalog row pin = %q, want the durable STALE unchanged", got) + } +} + +// (b) An EMPTY listing is a transient blackout, not a renumber: no recovery. +func TestStalePinRecoveryEmptyListingPreservesTerminal(t *testing.T) { + const neutral = "virtual://movie/tt-stale-empty" + row := stalePinRow(2, "movie-stale-empty", neutral, "STALE") + h, resolveCalls, _ := stalePinRecoveryHandler(nil, neutral+"?result=STALE", false) + + _, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1", stalePinRequest(), 0) + if ok { + t.Fatal("an empty listing must not recover") + } + if got := resolveCalls.Load(); got != 0 { + t.Fatalf("resolver calls = %d, want 0 for an empty listing", got) + } +} + +// (c) A non-empty listing with NO fingerprint match (size outside tolerance) +// does not recover. +func TestStalePinRecoveryNoFingerprintMatchPreservesTerminal(t *testing.T) { + const neutral = "virtual://movie/tt-stale-nomatch" + row := stalePinRow(3, "movie-stale-nomatch", neutral, "STALE") + // 30% larger: outside the 5% size tolerance. + live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "OTHER", row.FileSize*13/10, "h264")} + h, resolveCalls, _ := stalePinRecoveryHandler(live, neutral+"?result=STALE", false) + + _, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1", stalePinRequest(), 0) + if ok { + t.Fatal("a fingerprint miss must not recover") + } + if got := resolveCalls.Load(); got != 0 { + t.Fatalf("resolver calls = %d, want 0 without a fingerprint match", got) + } +} + +// (c2) A non-empty listing with the right size but a mismatched video codec does +// not recover. +func TestStalePinRecoveryCodecMismatchPreservesTerminal(t *testing.T) { + const neutral = "virtual://movie/tt-stale-codec" + row := stalePinRow(4, "movie-stale-codec", neutral, "STALE") + live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "HEVC", row.FileSize, "hevc")} + h, resolveCalls, _ := stalePinRecoveryHandler(live, neutral+"?result=STALE", false) + + _, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1", stalePinRequest(), 0) + if ok { + t.Fatal("a codec mismatch must not recover") + } + if got := resolveCalls.Load(); got != 0 { + t.Fatalf("resolver calls = %d, want 0", got) + } +} + +// (d) The recovery never adopts across a different content_id: a live candidate +// under another neutral path is dropped by the same-content filter before +// matching, so no resolve is attempted for it. +func TestStalePinRecoveryNoCrossContentAdoption(t *testing.T) { + const ( + neutral = "virtual://movie/tt-stale-cross" + foreign = "virtual://movie/tt-some-other-content" + ) + row := stalePinRow(5, "movie-stale-cross", neutral, "STALE") + // Same size and codec, but a different content path. + live := []VirtualPlaybackStream{stalePinLiveStream(foreign, "FOREIGN", row.FileSize, "h264")} + h, resolveCalls, _ := stalePinRecoveryHandler(live, neutral+"?result=STALE", false) + + _, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1", stalePinRequest(), 0) + if ok { + t.Fatal("a foreign-content candidate must not recover the pin") + } + if got := resolveCalls.Load(); got != 0 { + t.Fatalf("resolver calls = %d, want 0 (the foreign candidate is never resolved)", got) + } +} + +// (e) The recovery is a single retry, not a loop: even when the matched +// candidate also fails, exactly one recovery resolve is attempted. +func TestStalePinRecoverySingleRetryBound(t *testing.T) { + const neutral = "virtual://movie/tt-stale-retry" + row := stalePinRow(6, "movie-stale-retry", neutral, "STALE") + live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} + h, resolveCalls, _ := stalePinRecoveryHandler(live, neutral+"?result=STALE", true) + + _, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1", stalePinRequest(), 0) + if ok { + t.Fatal("a persistently failing recovery resolve must not report success") + } + if got := resolveCalls.Load(); got != 1 { + t.Fatalf("resolver calls = %d, want exactly 1 (bounded single retry)", got) + } +} + +// (f) A row WITH durable identity is left to the resolver's same-release rematch: +// the recovery does not fire and no listing or resolve is attempted. +func TestStalePinRecoverySkippedForIdentityCarryingRow(t *testing.T) { + const neutral = "virtual://movie/tt-stale-identity" + row := stalePinRow(7, "movie-stale-identity", neutral, "STALE") + row.ProviderVideoHash = "hash-a" // durable identity present + h, resolveCalls, listCalls := stalePinRecoveryHandler( + []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")}, + neutral+"?result=STALE", false) + + _, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1", stalePinRequest(), 0) + if ok { + t.Fatal("an identity-carrying row must not be recovered by the identity-less path") + } + if got := resolveCalls.Load(); got != 0 { + t.Fatalf("resolver calls = %d, want 0 (identity-carrying rows use the resolver rematch)", got) + } + if got := listCalls.Load(); got != 0 { + t.Fatalf("lister calls = %d, want 0 (the recovery must not even list)", got) + } +} + +// (g) The kill switch disables the recovery entirely. +func TestStalePinRecoveryKillSwitchDisables(t *testing.T) { + prev := virtualStalePinRecoveryEnabled + virtualStalePinRecoveryEnabled = false + t.Cleanup(func() { virtualStalePinRecoveryEnabled = prev }) + + const neutral = "virtual://movie/tt-stale-kill" + row := stalePinRow(8, "movie-stale-kill", neutral, "STALE") + h, resolveCalls, listCalls := stalePinRecoveryHandler( + []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")}, + neutral+"?result=STALE", false) + + _, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1", stalePinRequest(), 0) + if ok { + t.Fatal("the kill switch must disable recovery") + } + if got := resolveCalls.Load(); got != 0 { + t.Fatalf("resolver calls = %d, want 0 with the kill switch off", got) + } + if got := listCalls.Load(); got != 0 { + t.Fatalf("lister calls = %d, want 0 with the kill switch off", got) + } +} + +// TestStalePinRecoveryIntegratedInVersionFallbackWalk proves the walk actually +// invokes the recovery: an identity-less pin absent from the live listing +// recovers to the fingerprint-matched candidate instead of terminaling. The +// control run below (recovery disabled) proves the recovery is load-bearing. +func TestStalePinRecoveryIntegratedInVersionFallbackWalk(t *testing.T) { + const neutral = "virtual://movie/tt-stale-walk" + pinnedURI := neutral + "?result=STALE" + row := stalePinRow(9, "movie-stale-walk", neutral, "STALE") + live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} + h, resolveCalls, _ := stalePinRecoveryHandler(live, pinnedURI, false) + h.FileVersionFetcher = testPlaybackFileVersionFetcher{byContent: map[string][]*models.MediaFile{row.ContentID: {row}}} + + req := stalePinRecoveryRequest() + resolved, err := h.resolveVirtualStartWithVersionFallback(req, row, "profile-1", stalePinRequest(), false, 0) + if err != nil { + t.Fatalf("walk recovery: %v", err) + } + if resolved.File == nil || virtualResultCandidateID(resolved.URI) != "NEW" { + t.Fatalf("resolved = %#v, want the fingerprint-matched candidate NEW", resolved) + } + if got := resolveCalls.Load(); got == 0 { + t.Fatal("the walk never consulted the resolver") + } + + // Control: with the recovery disabled the same scenario terminals, so the + // recovery is what recovers it. + prev := virtualStalePinRecoveryEnabled + virtualStalePinRecoveryEnabled = false + t.Cleanup(func() { virtualStalePinRecoveryEnabled = prev }) + control := stalePinRow(10, "movie-stale-walk", neutral, "STALE") + controlHandler, _, _ := stalePinRecoveryHandler(live, pinnedURI, false) + controlHandler.FileVersionFetcher = testPlaybackFileVersionFetcher{byContent: map[string][]*models.MediaFile{control.ContentID: {control}}} + if _, controlErr := controlHandler.resolveVirtualStartWithVersionFallback(stalePinRecoveryRequest(), control, "profile-1", stalePinRequest(), false, 0); controlErr == nil { + t.Fatal("with the recovery disabled the dead pin must terminal; the recovery is masking a walk that already substitutes") + } +} + +// The fingerprint predicate is the boundary: inside tolerance accepts, outside +// refuses, and only the duration tier is neutral for an unknown side. +func TestStalePinFingerprintTolerance(t *testing.T) { + row := &models.MediaFile{CodecVideo: "h264", CodecAudio: "aac", FileSize: 1_000_000, Duration: 7200} + cases := []struct { + name string + candidate VirtualPlaybackStream + want bool + }{ + {"exact", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "aac", FileSize: 1_000_000}, true}, + {"size +4%", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "aac", FileSize: 1_040_000}, true}, + {"size -5%", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "aac", FileSize: 950_000}, true}, + {"size +6% refused", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "aac", FileSize: 1_060_000}, false}, + {"unknown candidate size refused", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "aac", FileSize: 0}, false}, + {"codec mismatch refused", VirtualPlaybackStream{CodecVideo: "hevc", CodecAudio: "aac", FileSize: 1_000_000}, false}, + {"unknown candidate codec refused", VirtualPlaybackStream{CodecAudio: "aac", FileSize: 1_000_000}, false}, + {"audio codec mismatch refused", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "eac3", FileSize: 1_000_000}, false}, + {"unknown candidate audio neutral", VirtualPlaybackStream{CodecVideo: "h264", FileSize: 1_000_000}, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := virtualStalePinStreamFingerprintMatch(row, tc.candidate); got != tc.want { + t.Fatalf("fingerprint = %v, want %v", got, tc.want) + } + }) + } + + durations := []struct { + row, candidate int + want bool + }{ + {7200, 7200, true}, + {7200, 7300, true}, // +1.4% + {7200, 7400, false}, // +2.8% + {0, 7300, true}, // unknown row duration is neutral + {7200, 0, true}, // unknown candidate duration is neutral + } + for _, tc := range durations { + if got := virtualStalePinDurationMatch(tc.row, tc.candidate); got != tc.want { + t.Fatalf("duration(%d,%d) = %v, want %v", tc.row, tc.candidate, got, tc.want) + } + } +} From 05fe5ed8c635442dd82a9ad4b59b30cfb9025d06 Mon Sep 17 00:00:00 2001 From: randrini Date: Tue, 6 Oct 2026 12:40:13 +0200 Subject: [PATCH 2/4] fix(playback): complete guarded stale-pin recovery for identity-less rows Finish the #227 partial: the version-fallback walk now declares its cold-start (unbound) intent, the recovery refuses a session-bound resolve explicitly, and a re-pinned row no longer carries the previous pin's stored URL, headers, durable identity, or probe evidence onto the matched candidate (clearing lifecycle verdict/delivery untouched). Rewrite the integrated walk test to be load-bearing: the primary resolve genuinely terminals before the recovery is consulted, and the recovery-disabled control proves the recovery is what recovers it. Add coverage for the session-bound refusal and the re-pin field hygiene. --- internal/api/handlers/playback_v3.go | 7 +- .../handlers/playback_virtual_stale_pin.go | 61 ++++++++ .../playback_virtual_stale_pin_test.go | 144 ++++++++++++++++-- 3 files changed, 196 insertions(+), 16 deletions(-) diff --git a/internal/api/handlers/playback_v3.go b/internal/api/handlers/playback_v3.go index 8c5ea4a7b6..a600c4aecd 100644 --- a/internal/api/handlers/playback_v3.go +++ b/internal/api/handlers/playback_v3.go @@ -2631,7 +2631,12 @@ func (h *PlaybackHandler) resolveVirtualStartWithVersionFallback( // elapses rather than waiting out the full cold-start budget per version. walkCtx, cancel := context.WithTimeout(r.Context(), virtualStartVersionFallbackDecisionBudget) defer cancel() - walkReq := r.WithContext(walkCtx) + // Declare the fresh-start intent explicitly: the version-fallback walk is a + // cold start, never a session-bound resolve. The stale-pin recovery checks + // this declaration before it will re-pin, so a future session-bound caller + // that reuses the walk with an undeclared context keeps the conservative + // session-bound default and refuses to touch a live session's binding. + walkReq := r.WithContext(withVirtualSessionBindingV3(walkCtx, false)) // Guarded stale-pin recovery for identity-less rows: when the pin id is // absent from a non-empty live listing and the row carries no durable diff --git a/internal/api/handlers/playback_virtual_stale_pin.go b/internal/api/handlers/playback_virtual_stale_pin.go index 231995a983..f7b09dbbc3 100644 --- a/internal/api/handlers/playback_virtual_stale_pin.go +++ b/internal/api/handlers/playback_virtual_stale_pin.go @@ -85,6 +85,18 @@ func (h *PlaybackHandler) recoverStaleIdentityLessPinV3( if h.VirtualPlaybackStreamLister == nil || h.VirtualMediaDetailedResolver == nil { return resolvedVirtualPlaybackSource{}, false } + // Cold start only, never a session-bound resolve. This function is called + // from the fresh-start version-fallback walk, which is unbound by + // construction, but the session binding is the one contract a stale-pin + // re-pin must never touch: a session already serving a release keeps + // refusing an absent pin so the serve layer's rotation policy owns it. The + // guard makes that invariant explicit and testable rather than implicit in + // the caller's wiring. + if VirtualSessionBinding(r.Context()) { + slog.DebugContext(r.Context(), "virtual stale-pin recovery skipped: session-bound resolve", + "component", "api", "file_id", file.ID, "candidate_uri", file.FilePath) + return resolvedVirtualPlaybackSource{}, false + } pinnedID := virtualResultCandidateID(file.FilePath) if pinnedID == "" { // A neutral row has no pin to go stale. @@ -173,6 +185,17 @@ func (h *PlaybackHandler) recoverStaleIdentityLessPinV3( // ranked, so the first fingerprint match is the strongest one, and a single // deterministic choice keeps the recovery one-shot. It reports ok=false when no // candidate matches. +// +// The returned copy keeps the row's catalog id, its declared fingerprint fields +// (size, codec, resolution, duration) — those are what the match was made +// against — and the row's own lifecycle fields (verdict, delivery stamp). Every +// field that describes the OLD pinned bytes is cleared: the stored provider URL +// and its headers, the durable identity tiers, and the probe evidence. Without +// this the re-pinned candidate would inherit the old pin's URL (serving the old +// bytes under the new id) and its probe stamp (reporting the old inventory as +// verified for the new candidate). The cleared state makes the caller resolve +// and probe the matched candidate afresh, which is the only thing that may +// authorize serving its bytes or inventory. func virtualStalePinBestStreamMatch(row *models.MediaFile, candidates []VirtualPlaybackStream) (models.MediaFile, bool) { for _, candidate := range candidates { if !virtualStalePinStreamFingerprintMatch(row, candidate) { @@ -184,11 +207,49 @@ func virtualStalePinBestStreamMatch(row *models.MediaFile, candidates []VirtualP // has no catalog row yet, and adopting its row id would fabricate one. Only // the served path changes. match.VirtualOwnerInstallationID = effectiveVirtualOwner(candidate.OwnerInstallationID, row.VirtualOwnerInstallationID) + clearVirtualStalePinOldCandidateState(&match) return match, true } return models.MediaFile{}, false } +// clearVirtualStalePinOldCandidateState drops the fields a re-pinned row must +// not carry from its previous pin. It is deliberately narrower than +// clearVirtualCandidateDeclaredMetadata: the row's declared fingerprint fields +// (size, codec, resolution, duration) are kept because they are the evidence the +// match was made against and drive candidate ranking, while the previous pin's +// transport and probe evidence are cleared so the matched candidate is resolved +// and probed on its own merits. +// +// The verdict (failed_at) and delivery (last_delivered_at) lifecycle fields are +// deliberately left in place: they describe the row's own lifecycle, not the +// bytes a candidate names, and the recovery must not alter verdict semantics. +// Clearing the stored URL is what matters most: without it the durable-resume +// fast path would serve the previous pin's URL under the new candidate's id. +func clearVirtualStalePinOldCandidateState(file *models.MediaFile) { + if file == nil { + return + } + // The stored URL and its headers name the previous pin's bytes. + file.ResolvedURL = "" + file.ResolvedURLExpiresAt = nil + file.ProviderRequestHeaders = nil + // Durable identity tiers (empty on an identity-less row, cleared for safety + // so a partially-populated legacy row cannot leak them onto the new pin). + file.ProviderVideoHash = "" + file.ProviderGUID = "" + file.ProviderReleaseName = "" + file.ProviderReleaseSize = 0 + // Probe evidence describes the previous pin's bytes; serving it as verified + // for the new candidate would be the premature-verified failure. + file.ProbeUpdatedAt = nil + file.ProbeSource = "" + file.VideoTracks = nil + file.AudioTracks = nil + file.SubtitleTracks = nil + file.ExternalSubtitles = nil +} + // virtualStalePinStreamFingerprintMatch is the cheap half of the fingerprint, // evaluated on the live listing record before any resolve: the size must be // known on both sides and agree within tolerance, and the video codec must be diff --git a/internal/api/handlers/playback_virtual_stale_pin_test.go b/internal/api/handlers/playback_virtual_stale_pin_test.go index 8b97b5bddb..d508a302ae 100644 --- a/internal/api/handlers/playback_virtual_stale_pin_test.go +++ b/internal/api/handlers/playback_virtual_stale_pin_test.go @@ -7,6 +7,7 @@ import ( "net/http/httptest" "sync/atomic" "testing" + "time" "github.com/Silo-Server/silo-server/internal/models" "github.com/Silo-Server/silo-server/internal/playback" @@ -65,6 +66,14 @@ func stalePinRecoveryHandler(live []VirtualPlaybackStream, pinnedURI string, fai } func stalePinRecoveryRequest() *http.Request { + // Declare the cold-start intent the version-fallback walk threads: the + // recovery refuses a session-bound resolve, and the conservative default is + // bound, so a unit test must opt in exactly as the walk does. + req := httptest.NewRequest(http.MethodPost, "/api/v2/playback/start", nil) + return req.WithContext(withVirtualSessionBindingV3(req.Context(), false)) +} + +func stalePinSessionBoundRequest() *http.Request { return httptest.NewRequest(http.MethodPost, "/api/v2/playback/start", nil) } @@ -237,42 +246,147 @@ func TestStalePinRecoveryKillSwitchDisables(t *testing.T) { } // TestStalePinRecoveryIntegratedInVersionFallbackWalk proves the walk actually -// invokes the recovery: an identity-less pin absent from the live listing -// recovers to the fingerprint-matched candidate instead of terminaling. The -// control run below (recovery disabled) proves the recovery is load-bearing. +// invokes the recovery and that it is load-bearing: the primary resolve fails +// because the provider answers empty for the pin's own listing AND for the +// stale-source fallback's re-list, but the recovery's neutral re-list answers a +// renumbered candidate whose fingerprint matches, so the walk recovers instead +// of terminaling. The control run below (recovery disabled) reaches the same +// dead pin and must terminal, proving the recovery — not an ordinary +// substitution — is what recovered it. +// +// The lister answers empty for the first two calls (the primary listing and the +// fallback's re-list) and the renumbered sibling thereafter, so the primary +// genuinely terminals before the recovery is consulted. The resolver answers +// the pin with an absent-pin refusal (a genuine renumber) and the sibling with a +// successful resolve. func TestStalePinRecoveryIntegratedInVersionFallbackWalk(t *testing.T) { const neutral = "virtual://movie/tt-stale-walk" - pinnedURI := neutral + "?result=STALE" row := stalePinRow(9, "movie-stale-walk", neutral, "STALE") live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} - h, resolveCalls, _ := stalePinRecoveryHandler(live, pinnedURI, false) - h.FileVersionFetcher = testPlaybackFileVersionFetcher{byContent: map[string][]*models.MediaFile{row.ContentID: {row}}} + + newHandler := func() *PlaybackHandler { + var listCalls atomic.Int32 + return &PlaybackHandler{ + FileVersionFetcher: testPlaybackFileVersionFetcher{byContent: map[string][]*models.MediaFile{row.ContentID: {row}}}, + VirtualPlaybackResolver: VirtualPlaybackResolverFunc(func(context.Context, string, int, string, int) (string, error) { + return "", errors.New("simple resolver must not be used when the detailed resolver is set") + }), + VirtualPlaybackStreamLister: VirtualPlaybackStreamListerFunc(func(context.Context, string, int, string, int) ([]VirtualPlaybackStream, error) { + if listCalls.Add(1) <= 2 { + // The primary listing and the fallback's re-list: the + // provider is empty, so the primary resolve terminal. + return nil, nil + } + // The recovery's re-list: the renumbered sibling. + return live, nil + }), + VirtualMediaDetailedResolver: VirtualMediaDetailedResolverFunc(func(_ context.Context, uri string, _ int, _ int, _ string, _ bool, _ []string, _ string) (ResolvedVirtualMedia, error) { + if virtualResultCandidateID(uri) == "STALE" { + return ResolvedVirtualMedia{}, absentSessionPinError("STALE") + } + return ResolvedVirtualMedia{URL: "http://127.0.0.1:8080/stream?uri=" + uri, URI: uri, CandidateID: virtualResultCandidateID(uri)}, nil + }), + } + } req := stalePinRecoveryRequest() - resolved, err := h.resolveVirtualStartWithVersionFallback(req, row, "profile-1", stalePinRequest(), false, 0) + resolved, err := newHandler().resolveVirtualStartWithVersionFallback(req, row, "profile-1", stalePinRequest(), false, 0) if err != nil { t.Fatalf("walk recovery: %v", err) } if resolved.File == nil || virtualResultCandidateID(resolved.URI) != "NEW" { t.Fatalf("resolved = %#v, want the fingerprint-matched candidate NEW", resolved) } - if got := resolveCalls.Load(); got == 0 { - t.Fatal("the walk never consulted the resolver") - } - // Control: with the recovery disabled the same scenario terminals, so the + // Control: with the recovery disabled the same dead pin terminals, so the // recovery is what recovers it. prev := virtualStalePinRecoveryEnabled virtualStalePinRecoveryEnabled = false t.Cleanup(func() { virtualStalePinRecoveryEnabled = prev }) - control := stalePinRow(10, "movie-stale-walk", neutral, "STALE") - controlHandler, _, _ := stalePinRecoveryHandler(live, pinnedURI, false) - controlHandler.FileVersionFetcher = testPlaybackFileVersionFetcher{byContent: map[string][]*models.MediaFile{control.ContentID: {control}}} - if _, controlErr := controlHandler.resolveVirtualStartWithVersionFallback(stalePinRecoveryRequest(), control, "profile-1", stalePinRequest(), false, 0); controlErr == nil { + if _, controlErr := newHandler().resolveVirtualStartWithVersionFallback(req, row, "profile-1", stalePinRequest(), false, 0); controlErr == nil { t.Fatal("with the recovery disabled the dead pin must terminal; the recovery is masking a walk that already substitutes") } } +// TestStalePinRecoveryRefusesSessionBound proves the cold-start-only contract: +// a session-bound resolve must keep refusing an absent pin, never re-pin it. +// The default (no declaration) is the conservative session-bound intent. +func TestStalePinRecoveryRefusesSessionBound(t *testing.T) { + const neutral = "virtual://movie/tt-stale-bound" + row := stalePinRow(11, "movie-stale-bound", neutral, "STALE") + live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} + h, resolveCalls, listCalls := stalePinRecoveryHandler(live, neutral+"?result=STALE", false) + + _, ok := h.recoverStaleIdentityLessPinV3(stalePinSessionBoundRequest(), row, "profile-1", stalePinRequest(), 0) + if ok { + t.Fatal("a session-bound resolve must not be re-pinned") + } + if got := resolveCalls.Load(); got != 0 { + t.Fatalf("resolver calls = %d, want 0 for a session-bound resolve", got) + } + if got := listCalls.Load(); got != 0 { + t.Fatalf("lister calls = %d, want 0 (the recovery must not even list)", got) + } +} + +// TestStalePinRecoveryClearsOldCandidateState pins the in-memory re-pin's field +// hygiene: the matched candidate must not inherit the previous pin's stored +// URL, headers, durable identity, or probe evidence. Carrying the stored URL +// would serve the old bytes under the new id; carrying the probe stamp would +// report the old inventory as verified for the new candidate. The row's own +// lifecycle fields (verdict and delivery stamp) are deliberately preserved, so +// the recovery does not alter verdict semantics. +func TestStalePinRecoveryClearsOldCandidateState(t *testing.T) { + const neutral = "virtual://movie/tt-stale-clear" + expiry := time.Now().Add(2 * time.Hour) + delivered := time.Now().Add(-time.Minute) + failedAt := time.Now().Add(-time.Minute) + probedAt := time.Now().Add(-time.Minute) + row := stalePinRow(13, "movie-stale-clear", neutral, "STALE") + row.ResolvedURL = "https://old.example/stale.mkv" + row.ResolvedURLExpiresAt = &expiry + row.ProviderRequestHeaders = map[string]string{"Referer": "https://old.example/"} + row.ProviderVideoHash = "old-hash" + row.ProviderReleaseName = "Old.Release" + row.ProviderReleaseSize = row.FileSize + row.LastDeliveredAt = &delivered + row.FailedAt = &failedAt + row.ProbeUpdatedAt = &probedAt + row.ProbeSource = virtualCollectionProbeSource + row.VideoTracks = []models.VideoTrack{{Codec: "h264", Width: 1920, Height: 1080}} + row.AudioTracks = []models.AudioTrack{{Codec: "aac", Channels: 2}} + row.SubtitleTracks = []models.SubtitleTrack{{Codec: "subrip", Language: "eng"}} + row.ExternalSubtitles = []models.ExternalSubtitle{{Path: "/tmp/old.srt"}} + + live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} + + match, found := virtualStalePinBestStreamMatch(row, live) + if !found { + t.Fatal("the fingerprint match was not found") + } + if match.FilePath != neutral+"?result=NEW" { + t.Fatalf("matched path = %q, want the live candidate", match.FilePath) + } + if match.ResolvedURL != "" || match.ResolvedURLExpiresAt != nil || match.ProviderRequestHeaders != nil { + t.Fatalf("re-pin carried the old transport state: url=%q expires=%v headers=%v", match.ResolvedURL, match.ResolvedURLExpiresAt, match.ProviderRequestHeaders) + } + if match.ProviderVideoHash != "" || match.ProviderReleaseName != "" || match.ProviderReleaseSize != 0 { + t.Fatalf("re-pin carried the old durable identity: hash=%q name=%q size=%d", match.ProviderVideoHash, match.ProviderReleaseName, match.ProviderReleaseSize) + } + if match.LastDeliveredAt == nil || match.FailedAt == nil { + t.Fatalf("re-pin cleared the row's own lifecycle state: delivered=%v failed=%v (verdict/delivery semantics must be untouched)", match.LastDeliveredAt, match.FailedAt) + } + if match.ProbeUpdatedAt != nil || match.ProbeSource != "" || len(match.VideoTracks) != 0 || len(match.AudioTracks) != 0 || len(match.SubtitleTracks) != 0 || len(match.ExternalSubtitles) != 0 { + t.Fatalf("re-pin carried the old probe evidence: stamp=%v source=%q video=%d audio=%d subs=%d ext=%d", + match.ProbeUpdatedAt, match.ProbeSource, len(match.VideoTracks), len(match.AudioTracks), len(match.SubtitleTracks), len(match.ExternalSubtitles)) + } + // The declared fingerprint fields are what the match was made against and + // must survive so candidate ranking still sees a 1080p h264 release. + if match.FileSize != row.FileSize || match.CodecVideo != "h264" || match.Resolution != "1080p" { + t.Fatalf("re-pin dropped the matched fingerprint fields: size=%d codec=%q res=%q", match.FileSize, match.CodecVideo, match.Resolution) + } +} + // The fingerprint predicate is the boundary: inside tolerance accepts, outside // refuses, and only the duration tier is neutral for an unknown side. func TestStalePinFingerprintTolerance(t *testing.T) { From 9b5a95bbcdff8629d167bffd21edbf8043046f6e Mon Sep 17 00:00:00 2001 From: randrini Date: Tue, 6 Oct 2026 17:39:25 +0200 Subject: [PATCH 3/4] fix(playback): bound and validate the stale-pin recovery Address the PR #234 review remainders: 1. Enforce an exact-candidate single attempt. The recovery no longer calls the general resolver (which iterates siblings and invokes the stale-source fallback, paying extra listings and breaking the promised retry bound). It resolves the re-pinned candidate once through resolveStalePinCandidateOnceV3, which has no sibling iteration, no fallback invocation, and rejects a substituted result id. 2. Validate before side effects. Identity and duration are checked on the probed result before anything is persisted or pinned; a rejected recovery leaves no new sticky selection and publishes no evidence. 3. Use the measured candidate duration. The probe copies the row with its duration zeroed and the fingerprint's duration tier is compared against the probe's own measured runtime; a no-prober or failed probe is a no-match rather than a comparison against the copied stale value. 4. Reserve time for alternatives. The recovery runs inside its own virtualStalePinRecoveryBudget (well under the walk's decision budget) so a slow recovery yields to the healthy alternates the walk still has to try. 5. Preserve session-bound intent. The walk threads the caller's binding intent instead of forcing it false, and the fresh-start path declares unbound at the call site; a session-bound caller reusing the walk keeps refusing. Tests cover the exact-candidate single attempt, substitution rejection, no-prober and probe-failure no-matches, measured-duration accept/reject with side-effect checks, a slow recovery yielding to a healthy alternate, and session-bound refusal through the walk. AI-assisted: implemented by AI coding agents on behalf of the repository owner. --- internal/api/handlers/playback_v3.go | 27 +- .../handlers/playback_virtual_stale_pin.go | 225 +++++-- .../playback_virtual_stale_pin_test.go | 592 +++++++++++++----- 3 files changed, 620 insertions(+), 224 deletions(-) diff --git a/internal/api/handlers/playback_v3.go b/internal/api/handlers/playback_v3.go index a600c4aecd..04db99fd87 100644 --- a/internal/api/handlers/playback_v3.go +++ b/internal/api/handlers/playback_v3.go @@ -1865,7 +1865,12 @@ func (h *PlaybackHandler) startPlaybackApplicationV3(r *http.Request, body []byt // earlier auto pick; only an explicit pick or a forced relink re-tries // the known-bad candidate. allowFailedCandidate := req.FileSelection == playback.FileSelectionExplicitV3 || req.ForceRelink - resolved, resolveErr := h.resolveVirtualStartWithVersionFallback(r, requestedFile, profileID, req, allowFailedCandidate, intOrZeroHandlerV3(req.BandwidthCapKbps)) + // The start path is a fresh selection: declare the unbound intent on the + // request handed to the walk, which threads it (rather than forcing it) + // so the stale-pin recovery may re-pin here while a future session-bound + // caller reusing the walk keeps its own binding intent and refuses. + walkSourceReq := r.WithContext(withVirtualSessionBindingV3(r.Context(), false)) + resolved, resolveErr := h.resolveVirtualStartWithVersionFallback(walkSourceReq, requestedFile, profileID, req, allowFailedCandidate, intOrZeroHandlerV3(req.BandwidthCapKbps)) if resolveErr != nil { // A confirmed-dead pinned release is indicted here so a retry does // not re-resolve it; an empty provider listing is not a verdict and @@ -2631,19 +2636,21 @@ func (h *PlaybackHandler) resolveVirtualStartWithVersionFallback( // elapses rather than waiting out the full cold-start budget per version. walkCtx, cancel := context.WithTimeout(r.Context(), virtualStartVersionFallbackDecisionBudget) defer cancel() - // Declare the fresh-start intent explicitly: the version-fallback walk is a - // cold start, never a session-bound resolve. The stale-pin recovery checks - // this declaration before it will re-pin, so a future session-bound caller - // that reuses the walk with an undeclared context keeps the conservative - // session-bound default and refuses to touch a live session's binding. - walkReq := r.WithContext(withVirtualSessionBindingV3(walkCtx, false)) + // Thread the caller's session-binding intent rather than forcing it false. + // The fresh-start start path declares unbound on the request context before + // calling this walk, so the stale-pin recovery may re-pin; a future + // session-bound caller that reuses the walk keeps its declaration (or the + // conservative session-bound default) and the recovery refuses to touch a + // live session's binding. + walkReq := r.WithContext(withVirtualSessionBindingV3(walkCtx, VirtualSessionBinding(r.Context()))) // Guarded stale-pin recovery for identity-less rows: when the pin id is // absent from a non-empty live listing and the row carries no durable // provider identity to rematch on, re-pin in memory to the best - // fingerprint-matched live candidate and retry once. Bounded to one attempt; - // a miss falls through to the alternate walk below with the terminal intact. - if recovered, recoveredOK := h.recoverStaleIdentityLessPinV3(walkReq, file, profileID, req, bandwidthCapKbps); recoveredOK { + // fingerprint-matched live candidate and retry once. Bounded to one attempt + // inside its own sub-budget; a miss falls through to the alternate walk + // below with the terminal intact. + if recovered, recoveredOK := h.recoverStaleIdentityLessPinV3(walkReq, file, profileID); recoveredOK { return recovered, nil } diff --git a/internal/api/handlers/playback_virtual_stale_pin.go b/internal/api/handlers/playback_virtual_stale_pin.go index f7b09dbbc3..37b931d3c5 100644 --- a/internal/api/handlers/playback_virtual_stale_pin.go +++ b/internal/api/handlers/playback_virtual_stale_pin.go @@ -2,6 +2,8 @@ package handlers import ( "context" + "errors" + "fmt" "log/slog" "net/http" "os" @@ -10,7 +12,7 @@ import ( apimw "github.com/Silo-Server/silo-server/internal/api/middleware" "github.com/Silo-Server/silo-server/internal/models" - "github.com/Silo-Server/silo-server/internal/playback" + "github.com/Silo-Server/silo-server/internal/virtuallibrary" ) // Guarded stale-pin recovery for identity-less rows. @@ -63,21 +65,54 @@ const ( // tighter than the runtime tolerance used for probe plausibility because this // gate authorizes adopting different bytes. virtualStalePinDurationTolerance = 0.02 - // virtualStalePinListBudget bounds the one provider listing the recovery pays. - // The walk's own decision budget still bounds the whole start. - virtualStalePinListBudget = 5 * time.Second ) +// virtualStalePinRecoveryBudget bounds the entire stale-pin recovery: its one +// provider listing plus the single exact-candidate resolve and probe. It is a +// small fraction of the version-fallback walk's decision budget so a slow +// recovery yields the remaining time to the healthy alternate rows the walk +// still has to try; a recovery that cannot finish inside this window is +// abandoned and the walk proceeds to the alternates. Without this sub-budget a +// synchronous recovery could consume the whole parent budget on one listing and +// starve every alternate. It is a var so tests can pin it. +var virtualStalePinRecoveryBudget = 500 * time.Millisecond + +// stalePinCandidateResolutionV3 is the outcome of the recovery's single +// exact-candidate resolve+probe: the probed file (its Duration is the value the +// probe measured, never a copied or stale value), the canonical candidate URI, +// and the transport facts the caller needs to pin and persist it. +type stalePinCandidateResolutionV3 struct { + file *models.MediaFile + uri string + url string + headers map[string]string + providerVideoHash string + providerGUID string + providerReleaseName string + providerReleaseSize int64 +} + // recoverStaleIdentityLessPinV3 attempts the guarded in-memory re-pin described // above. It returns ok=true only when a fingerprint-matched live candidate was -// resolved; any other outcome leaves the caller's terminal in place. It performs -// exactly one candidate resolve (no loops) and never mutates the catalog pin. +// resolved; any other outcome leaves the caller's terminal in place. +// +// The recovery obeys three safety contracts: +// - a single attempt: it resolves the exact re-pinned candidate through a +// bespoke one-candidate resolve (no sibling iteration, no stale-source +// fallback), never the general resolver whose candidate loop and fallback +// would break the promised retry bound; +// - validation before side effects: identity and duration are checked before +// anything is persisted or pinned, so a rejected recovery publishes no +// evidence and leaves no new sticky selection; +// - measured duration: the duration comparison uses the probe result's own +// measured runtime, and a no-prober or failed probe is a no-match rather +// than a comparison against the copied row's stale value. +// +// It never mutates the durable catalog pin. func (h *PlaybackHandler) recoverStaleIdentityLessPinV3( r *http.Request, file *models.MediaFile, profileID string, - req playback.StartRequestV3, - bandwidthCapKbps int, ) (resolvedVirtualPlaybackSource, bool) { if !virtualStalePinRecoveryEnabled || h == nil || file == nil || r == nil { return resolvedVirtualPlaybackSource{}, false @@ -85,10 +120,8 @@ func (h *PlaybackHandler) recoverStaleIdentityLessPinV3( if h.VirtualPlaybackStreamLister == nil || h.VirtualMediaDetailedResolver == nil { return resolvedVirtualPlaybackSource{}, false } - // Cold start only, never a session-bound resolve. This function is called - // from the fresh-start version-fallback walk, which is unbound by - // construction, but the session binding is the one contract a stale-pin - // re-pin must never touch: a session already serving a release keeps + // Cold start only, never a session-bound resolve. The binding intent is the + // caller's, preserved by the walk: a session already serving a release keeps // refusing an absent pin so the serve layer's rotation policy owns it. The // guard makes that invariant explicit and testable rather than implicit in // the caller's wiring. @@ -114,11 +147,13 @@ func (h *PlaybackHandler) recoverStaleIdentityLessPinV3( } ctx := r.Context() userID := apimw.GetUserID(ctx) - listCtx, cancel := context.WithTimeout(ctx, virtualStalePinListBudget) + // Bound the whole recovery well under the walk's decision budget so the + // alternates the walk still has to try keep their time. + recoveryCtx, cancel := context.WithTimeout(ctx, virtualStalePinRecoveryBudget) + defer cancel() streams, err := h.VirtualPlaybackStreamLister.ListVirtualPlaybackStreams( - listCtx, neutralKey, userID, profileID, file.VirtualOwnerInstallationID, + recoveryCtx, neutralKey, userID, profileID, file.VirtualOwnerInstallationID, ) - cancel() if err != nil || len(streams) == 0 { // An empty (or failed) listing is a transient outage, not a renumber. // Preserve the caller's terminal. @@ -138,53 +173,151 @@ func (h *PlaybackHandler) recoverStaleIdentityLessPinV3( if len(candidates) == 0 { return resolvedVirtualPlaybackSource{}, false } - match, found := virtualStalePinBestStreamMatch(file, candidates) + match, matchedStream, found := virtualStalePinBestStreamMatch(file, candidates) if !found { slog.DebugContext(ctx, "virtual stale-pin recovery found no fingerprint match", "component", "api", "file_id", file.ID, "old_candidate_id", pinnedID, "candidate_count", len(candidates)) return resolvedVirtualPlaybackSource{}, false } - // Re-pin in memory and resolve the matched candidate exactly once. The row's - // durable ?result= is never rewritten; only process state moves. + // Resolve and probe the EXACT re-pinned candidate in a single attempt. + ownerID := effectiveVirtualOwner(match.VirtualOwnerInstallationID, file.VirtualOwnerInstallationID) matchedID := virtualResultCandidateID(match.FilePath) - recovered, resolveErr := h.resolveVirtualPlaybackSource( - r, &match, profileID, false, nil, "", req.QualityPreference, bandwidthCapKbps, false, - virtualResolveOptionsV3{sessionBound: false, bypassProviderFloor: true}, - ) - if resolveErr != nil || recovered.File == nil { - return resolvedVirtualPlaybackSource{}, false - } - // Refuse a substitution: the recovery authorized serving the exact matched - // candidate, so a resolve that returned a different result id (a sibling the - // resolver preferred) must not be adopted under the re-pin. This keeps the - // recovery's blast radius to the candidate the fingerprint actually matched. - if got := virtualResultCandidateID(recovered.URI); got != "" && matchedID != "" && got != matchedID { - slog.DebugContext(ctx, "virtual stale-pin recovery refused a substituted candidate", + resolution, resolveErr := h.resolveStalePinCandidateOnceV3(recoveryCtx, &match, matchedID, matchedStream, ownerID, userID, profileID) + if resolveErr != nil { + slog.DebugContext(ctx, "virtual stale-pin recovery did not resolve the matched candidate", "component", "api", "file_id", file.ID, "old_candidate_id", pinnedID, - "matched_candidate_id", matchedID, "resolved_candidate_id", got) + "matched_candidate_id", matchedID, "error", resolveErr) return resolvedVirtualPlaybackSource{}, false } - if !virtualStalePinDurationMatch(file.Duration, recovered.File.Duration) { + if !virtualStalePinDurationMatch(file.Duration, resolution.file.Duration) { slog.DebugContext(ctx, "virtual stale-pin recovery rejected a duration mismatch", "component", "api", "file_id", file.ID, "old_candidate_id", pinnedID, - "new_candidate_id", virtualResultCandidateID(recovered.URI), - "row_duration", file.Duration, "candidate_duration", recovered.File.Duration) + "new_candidate_id", virtualResultCandidateID(resolution.uri), + "row_duration", file.Duration, "candidate_duration", resolution.file.Duration) return resolvedVirtualPlaybackSource{}, false } + // Validation passed; only now may the recovery touch shared state. Pin the + // matched candidate for later starts and persist its probed evidence. + fingerprint := "" + if caps, ok := h.requestDeviceCapabilities(r); ok { + fingerprint = caps.Fingerprint() + } + stickyKey := bestResultCacheKey(file.ContentID, neutralKey, file.VirtualOwnerInstallationID, fingerprint) + h.pinVirtualStickyAt(stickyKey, resolution.uri, nextVirtualCacheGeneration()) + h.persistVirtualProbeEvidence(ctx, file, resolution.uri, resolution.file, true, true) slog.InfoContext(ctx, "virtual stale-pin recovery re-pinned an identity-less candidate in memory", "component", "api", "file_id", file.ID, "old_candidate_id", pinnedID, - "new_candidate_id", virtualResultCandidateID(recovered.URI), - "virtual_uri", recovered.URI) - return recovered, true + "new_candidate_id", virtualResultCandidateID(resolution.uri), + "virtual_uri", resolution.uri) + return resolvedVirtualPlaybackSource{ + URL: resolution.url, + URI: resolution.uri, + OwnerID: ownerID, + File: resolution.file, + ProbeSucceeded: true, + Provenance: ProbeProvenanceVerified, + ResolvedURL: resolution.url, + ProviderVideoHash: resolution.providerVideoHash, + ProviderGUID: resolution.providerGUID, + ProviderReleaseName: resolution.providerReleaseName, + ProviderReleaseSize: resolution.providerReleaseSize, + RequestHeaders: cloneHeaderMap(resolution.headers), + }, true +} + +// resolveStalePinCandidateOnceV3 resolves and probes exactly one candidate for +// the stale-pin recovery. It performs one provider resolve and at most one +// probe, and it has no persistence or sticky-pin side effect: the general +// resolver would iterate siblings and invoke the stale-source fallback before +// returning, and would persist evidence and pin the candidate before the +// recovery's own identity and duration checks ran. The caller validates the +// returned identity and measured duration and only then persists and pins. +func (h *PlaybackHandler) resolveStalePinCandidateOnceV3( + ctx context.Context, + match *models.MediaFile, + matchedID string, + matchedStream VirtualPlaybackStream, + ownerID int, + userID int, + profileID string, +) (stalePinCandidateResolutionV3, error) { + // The recovery is a declared re-list: it must see the provider past the + // fresh-serve floor and the short provider-failure fail-fast, exactly as the + // general resolver's bypassProviderFloor did, without which a recent + // background failure would hide the renumbered candidate. + ctx = virtuallibrary.WithProviderOutageRelist(ctx) + res, err := h.VirtualMediaDetailedResolver.ResolveVirtualMediaDetailed( + ctx, match.FilePath, ownerID, userID, profileID, true, nil, "", + ) + if err != nil { + return stalePinCandidateResolutionV3{}, err + } + if strings.TrimSpace(res.URL) == "" { + return stalePinCandidateResolutionV3{}, errors.New("stale-pin candidate resolved without a stream URL") + } + resolvedURI := strings.TrimSpace(res.URI) + if resolvedURI == "" { + resolvedURI = match.FilePath + } + // The resolver must have returned the exact matched candidate. A sibling it + // preferred (a substitution) is not the bytes the fingerprint authorized. + if got := virtualResultCandidateID(resolvedURI); got != "" && matchedID != "" && got != matchedID { + return stalePinCandidateResolutionV3{}, fmt.Errorf("resolver substituted candidate %q for matched %q", got, matchedID) + } + probeTransient := cloneVirtualProbeTransient(*match) + probeTransient.FilePath = resolvedURI + probeTransient.VirtualOwnerInstallationID = ownerID + // Zero the copied row's duration so the fingerprint's duration tier is + // measured by the probe, never inherited from the previous pin. A prober + // that reports no runtime then leaves Duration at zero and the caller treats + // the recovery as a no-match rather than comparing a stale value. + probeTransient.Duration = 0 + probed, probeErr := h.probeVirtualSource(ctx, res.URL, &probeTransient, cloneHeaderMap(res.RequestHeaders)) + if probeErr != nil || probed == nil { + // No probe result means no measured duration: the recovery must not fall + // back to the copied row's stale runtime. + return stalePinCandidateResolutionV3{}, fmt.Errorf("stale-pin candidate probe failed: %w", probeErr) + } + if probed.Duration <= 0 { + // A probe that measured no runtime cannot confirm the fingerprint's + // duration tier, so the recovery is a no-match. + return stalePinCandidateResolutionV3{}, errors.New("stale-pin candidate probe measured no duration") + } + // A candidate the serve layer already indicted must never be re-served by the + // recovery. Check the identity the resolver actually returned, exactly as the + // general resolver's verdict gate does. + if err := h.virtualCandidateVerdictError(ctx, resolvedURI, match, ownerID, false); err != nil { + return stalePinCandidateResolutionV3{}, err + } + probed.FilePath = resolvedURI + if match.ID > 0 { + probed.ID = match.ID + probed.MediaFolderID = match.MediaFolderID + } + // Fill declared gaps the same way the general resolver does, so the served + // file carries the candidate's language/codec hints where the probe left + // them empty. This never overwrites probed evidence. + mergeVirtualCandidateTracks(probed, matchedStream) + return stalePinCandidateResolutionV3{ + file: probed, + uri: resolvedURI, + url: res.URL, + headers: res.RequestHeaders, + providerVideoHash: res.ProviderVideoHash, + providerGUID: res.ProviderGUID, + providerReleaseName: res.ProviderReleaseName, + providerReleaseSize: res.ProviderReleaseSize, + }, nil } // virtualStalePinBestStreamMatch returns the best live candidate whose declared -// evidence matches the row's stored fingerprint, re-pinned as a copy of the row. -// "Best" is the first match in provider rank order: the listing is already -// ranked, so the first fingerprint match is the strongest one, and a single -// deterministic choice keeps the recovery one-shot. It reports ok=false when no -// candidate matches. +// evidence matches the row's stored fingerprint, re-pinned as a copy of the row +// alongside the matched stream itself (whose declared tracks the caller may merge +// into the probe result). "Best" is the first match in provider rank order: the +// listing is already ranked, so the first fingerprint match is the strongest one, +// and a single deterministic choice keeps the recovery one-shot. It reports +// ok=false when no candidate matches. // // The returned copy keeps the row's catalog id, its declared fingerprint fields // (size, codec, resolution, duration) — those are what the match was made @@ -196,7 +329,7 @@ func (h *PlaybackHandler) recoverStaleIdentityLessPinV3( // verified for the new candidate). The cleared state makes the caller resolve // and probe the matched candidate afresh, which is the only thing that may // authorize serving its bytes or inventory. -func virtualStalePinBestStreamMatch(row *models.MediaFile, candidates []VirtualPlaybackStream) (models.MediaFile, bool) { +func virtualStalePinBestStreamMatch(row *models.MediaFile, candidates []VirtualPlaybackStream) (models.MediaFile, VirtualPlaybackStream, bool) { for _, candidate := range candidates { if !virtualStalePinStreamFingerprintMatch(row, candidate) { continue @@ -208,9 +341,9 @@ func virtualStalePinBestStreamMatch(row *models.MediaFile, candidates []VirtualP // the served path changes. match.VirtualOwnerInstallationID = effectiveVirtualOwner(candidate.OwnerInstallationID, row.VirtualOwnerInstallationID) clearVirtualStalePinOldCandidateState(&match) - return match, true + return match, candidate, true } - return models.MediaFile{}, false + return models.MediaFile{}, VirtualPlaybackStream{}, false } // clearVirtualStalePinOldCandidateState drops the fields a re-pinned row must diff --git a/internal/api/handlers/playback_virtual_stale_pin_test.go b/internal/api/handlers/playback_virtual_stale_pin_test.go index d508a302ae..d861d2b3ae 100644 --- a/internal/api/handlers/playback_virtual_stale_pin_test.go +++ b/internal/api/handlers/playback_virtual_stale_pin_test.go @@ -41,34 +41,87 @@ func stalePinLiveStream(neutralURI, resultID string, fileSize int64, videoCodec } } -// stalePinRecoveryHandler wires the recovery's two collaborators: the lister -// answers a fixed live set, and the resolver records every call and succeeds for -// any non-pinned candidate (or fails for all, when failAll is set). -func stalePinRecoveryHandler(live []VirtualPlaybackStream, pinnedURI string, failAll bool) (*PlaybackHandler, *atomic.Int32, *atomic.Int32) { - var resolveCalls, listCalls atomic.Int32 +// stalePinProbedTrack is the real ffprobe evidence the recovery's probe must +// return: a complete track plus a measured duration that can be tuned to match +// or miss the row's stored runtime. +func stalePinProbedTrack(duration int) *models.MediaFile { + return &models.MediaFile{ + Duration: duration, + Container: "mkv", CodecVideo: "h264", CodecAudio: "aac", Resolution: "1080p", Bitrate: 8000, + VideoTracks: []models.VideoTrack{{ + Codec: "h264", Width: 1920, Height: 1080, FrameRate: "24000/1001", BitDepth: 8, Bitrate: 8000, + }}, + AudioTracks: []models.AudioTrack{{Codec: "aac", Channels: 2}}, + } +} + +// stalePinRecoveryHandler wires the recovery's collaborators: the lister answers +// a fixed live set, the detailed resolver records every call and resolves the +// matched candidate (or substitutes/fails per the test), and the prober records +// every probe and returns a measured duration (or fails). resolveCalls counts +// handler-level resolves (must stay 1); listCalls counts listings; saveCalls +// counts evidence writes. +type stalePinRecoveryHandlerOpts struct { + live []VirtualPlaybackStream + resolveErr error + substituteURI string + probeDuration int + probeErr error + noProber bool +} + +type stalePinRecoveryFixture struct { + handler *PlaybackHandler + resolveCalls *atomic.Int32 + listCalls *atomic.Int32 + saveCalls *atomic.Int32 +} + +func newStalePinRecoveryFixture(pinnedURI string, opts stalePinRecoveryHandlerOpts) *stalePinRecoveryFixture { + var resolveCalls, listCalls, saveCalls atomic.Int32 h := &PlaybackHandler{ VirtualPlaybackResolver: VirtualPlaybackResolverFunc(func(context.Context, string, int, string, int) (string, error) { return "", errors.New("simple resolver must not be used when the detailed resolver is set") }), VirtualPlaybackStreamLister: VirtualPlaybackStreamListerFunc(func(context.Context, string, int, string, int) ([]VirtualPlaybackStream, error) { listCalls.Add(1) - return live, nil + return opts.live, nil }), VirtualMediaDetailedResolver: VirtualMediaDetailedResolverFunc(func(_ context.Context, uri string, _ int, _ int, _ string, _ bool, _ []string, _ string) (ResolvedVirtualMedia, error) { resolveCalls.Add(1) - if failAll || uri == pinnedURI { - return ResolvedVirtualMedia{}, absentSessionPinError("STALE") + if opts.resolveErr != nil { + return ResolvedVirtualMedia{}, opts.resolveErr + } + resolvedURI := uri + if opts.substituteURI != "" { + resolvedURI = opts.substituteURI } - return ResolvedVirtualMedia{URL: "http://127.0.0.1:8080/stream?uri=" + uri, URI: uri, CandidateID: "NEW"}, nil + return ResolvedVirtualMedia{URL: "http://127.0.0.1:8080/stream?uri=" + resolvedURI, URI: resolvedURI, CandidateID: virtualResultCandidateID(resolvedURI)}, nil }), + VirtualFileSaver: func(context.Context, models.VirtualFilePersistArgs) (int64, error) { + saveCalls.Add(1) + return 1, nil + }, + } + fixture := &stalePinRecoveryFixture{handler: h, resolveCalls: &resolveCalls, listCalls: &listCalls, saveCalls: &saveCalls} + if !opts.noProber { + h.VirtualPlaybackSourceProber = func(_ context.Context, _ string, f *models.MediaFile) (*models.MediaFile, error) { + if opts.probeErr != nil { + return nil, opts.probeErr + } + probed := stalePinProbedTrack(opts.probeDuration) + probed.ID = f.ID + probed.FilePath = f.FilePath + return probed, nil + } } - return h, &resolveCalls, &listCalls + return fixture } func stalePinRecoveryRequest() *http.Request { - // Declare the cold-start intent the version-fallback walk threads: the - // recovery refuses a session-bound resolve, and the conservative default is - // bound, so a unit test must opt in exactly as the walk does. + // Declare the cold-start intent the start path threads: the recovery refuses + // a session-bound resolve, and the conservative default is bound, so a unit + // test must opt in exactly as the start path does. req := httptest.NewRequest(http.MethodPost, "/api/v2/playback/start", nil) return req.WithContext(withVirtualSessionBindingV3(req.Context(), false)) } @@ -77,21 +130,21 @@ func stalePinSessionBoundRequest() *http.Request { return httptest.NewRequest(http.MethodPost, "/api/v2/playback/start", nil) } -func stalePinRequest() playback.StartRequestV3 { - return playback.StartRequestV3{QualityPreference: "auto"} +func (f *stalePinRecoveryFixture) stickyKey(row *models.MediaFile) string { + return bestResultCacheKey(row.ContentID, virtualPlaybackNeutralKey(row.FilePath), row.VirtualOwnerInstallationID) } // (a) A dead pin absent from a NON-EMPTY listing, on an identity-less row, with a -// live fingerprint match recovers: the recovery resolves the matched candidate -// exactly once and returns it, in memory only. +// live fingerprint match and a probe that measures a matching duration recovers: +// one handler-level resolve, one probe, in memory only. func TestStalePinRecoveryRecoversFingerprintMatch(t *testing.T) { const neutral = "virtual://movie/tt-stale-recover" pinnedURI := neutral + "?result=STALE" row := stalePinRow(1, "movie-stale-recover", neutral, "STALE") live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} - h, resolveCalls, _ := stalePinRecoveryHandler(live, pinnedURI, false) + fx := newStalePinRecoveryFixture(pinnedURI, stalePinRecoveryHandlerOpts{live: live, probeDuration: row.Duration}) - recovered, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1", stalePinRequest(), 0) + recovered, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") if !ok { t.Fatal("a fingerprint-matched live candidate must recover the pin") } @@ -101,8 +154,18 @@ func TestStalePinRecoveryRecoversFingerprintMatch(t *testing.T) { if got := virtualResultCandidateID(recovered.URI); got != "NEW" { t.Fatalf("recovered candidate = %q, want NEW", got) } - if got := resolveCalls.Load(); got != 1 { - t.Fatalf("resolver calls = %d, want exactly 1 (one recovery retry, no loop)", got) + if got := fx.resolveCalls.Load(); got != 1 { + t.Fatalf("handler-level resolves = %d, want exactly 1 (single attempt, no sibling iteration)", got) + } + // One listing only: the general resolver's stale-source fallback would have + // paid a second listing, proving the recovery did not invoke it. + if got := fx.listCalls.Load(); got != 1 { + t.Fatalf("provider listings = %d, want exactly 1 (no fallback invocation)", got) + } + // The sticky pin is set synchronously only after validation passes, so its + // presence is the deterministic proof the recovery authorized the candidate. + if got := fx.handler.peekVirtualSticky(fx.stickyKey(row)); got != neutral+"?result=NEW" { + t.Fatalf("sticky pin = %q, want the matched candidate after a successful recovery", got) } // The row's durable pin is never rewritten: only the returned copy changed. if got := virtualResultCandidateID(row.FilePath); got != "STALE" { @@ -110,18 +173,48 @@ func TestStalePinRecoveryRecoversFingerprintMatch(t *testing.T) { } } +// (a2) A single attempt is enforced even when the matched candidate's resolve +// yields a substitution: the recovery rejects the different result id and never +// iterates to another sibling or falls back. +func TestStalePinRecoverySingleAttemptRejectsSubstitution(t *testing.T) { + const neutral = "virtual://movie/tt-stale-subst" + pinnedURI := neutral + "?result=STALE" + row := stalePinRow(2, "movie-stale-subst", neutral, "STALE") + live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} + fx := newStalePinRecoveryFixture(pinnedURI, stalePinRecoveryHandlerOpts{ + live: live, + probeDuration: row.Duration, + // The detailed resolver substituted a different sibling. + substituteURI: neutral + "?result=OTHER", + }) + + _, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") + if ok { + t.Fatal("a substituted candidate must not recover the pin") + } + if got := fx.resolveCalls.Load(); got != 1 { + t.Fatalf("handler-level resolves = %d, want exactly 1 (no iteration after a substitution)", got) + } + if got := fx.listCalls.Load(); got != 1 { + t.Fatalf("provider listings = %d, want exactly 1 (no stale-source fallback)", got) + } + if got := fx.saveCalls.Load(); got != 0 { + t.Fatalf("evidence writes = %d, want 0 when the recovery is rejected", got) + } +} + // (b) An EMPTY listing is a transient blackout, not a renumber: no recovery. func TestStalePinRecoveryEmptyListingPreservesTerminal(t *testing.T) { const neutral = "virtual://movie/tt-stale-empty" - row := stalePinRow(2, "movie-stale-empty", neutral, "STALE") - h, resolveCalls, _ := stalePinRecoveryHandler(nil, neutral+"?result=STALE", false) + row := stalePinRow(3, "movie-stale-empty", neutral, "STALE") + fx := newStalePinRecoveryFixture(neutral+"?result=STALE", stalePinRecoveryHandlerOpts{probeDuration: row.Duration}) - _, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1", stalePinRequest(), 0) + _, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") if ok { t.Fatal("an empty listing must not recover") } - if got := resolveCalls.Load(); got != 0 { - t.Fatalf("resolver calls = %d, want 0 for an empty listing", got) + if got := fx.resolveCalls.Load(); got != 0 { + t.Fatalf("resolves = %d, want 0 for an empty listing", got) } } @@ -129,17 +222,17 @@ func TestStalePinRecoveryEmptyListingPreservesTerminal(t *testing.T) { // does not recover. func TestStalePinRecoveryNoFingerprintMatchPreservesTerminal(t *testing.T) { const neutral = "virtual://movie/tt-stale-nomatch" - row := stalePinRow(3, "movie-stale-nomatch", neutral, "STALE") + row := stalePinRow(4, "movie-stale-nomatch", neutral, "STALE") // 30% larger: outside the 5% size tolerance. live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "OTHER", row.FileSize*13/10, "h264")} - h, resolveCalls, _ := stalePinRecoveryHandler(live, neutral+"?result=STALE", false) + fx := newStalePinRecoveryFixture(neutral+"?result=STALE", stalePinRecoveryHandlerOpts{live: live, probeDuration: row.Duration}) - _, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1", stalePinRequest(), 0) + _, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") if ok { t.Fatal("a fingerprint miss must not recover") } - if got := resolveCalls.Load(); got != 0 { - t.Fatalf("resolver calls = %d, want 0 without a fingerprint match", got) + if got := fx.resolveCalls.Load(); got != 0 { + t.Fatalf("resolves = %d, want 0 without a fingerprint match", got) } } @@ -147,16 +240,16 @@ func TestStalePinRecoveryNoFingerprintMatchPreservesTerminal(t *testing.T) { // not recover. func TestStalePinRecoveryCodecMismatchPreservesTerminal(t *testing.T) { const neutral = "virtual://movie/tt-stale-codec" - row := stalePinRow(4, "movie-stale-codec", neutral, "STALE") + row := stalePinRow(5, "movie-stale-codec", neutral, "STALE") live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "HEVC", row.FileSize, "hevc")} - h, resolveCalls, _ := stalePinRecoveryHandler(live, neutral+"?result=STALE", false) + fx := newStalePinRecoveryFixture(neutral+"?result=STALE", stalePinRecoveryHandlerOpts{live: live, probeDuration: row.Duration}) - _, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1", stalePinRequest(), 0) + _, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") if ok { t.Fatal("a codec mismatch must not recover") } - if got := resolveCalls.Load(); got != 0 { - t.Fatalf("resolver calls = %d, want 0", got) + if got := fx.resolveCalls.Load(); got != 0 { + t.Fatalf("resolves = %d, want 0", got) } } @@ -168,34 +261,44 @@ func TestStalePinRecoveryNoCrossContentAdoption(t *testing.T) { neutral = "virtual://movie/tt-stale-cross" foreign = "virtual://movie/tt-some-other-content" ) - row := stalePinRow(5, "movie-stale-cross", neutral, "STALE") + row := stalePinRow(6, "movie-stale-cross", neutral, "STALE") // Same size and codec, but a different content path. live := []VirtualPlaybackStream{stalePinLiveStream(foreign, "FOREIGN", row.FileSize, "h264")} - h, resolveCalls, _ := stalePinRecoveryHandler(live, neutral+"?result=STALE", false) + fx := newStalePinRecoveryFixture(neutral+"?result=STALE", stalePinRecoveryHandlerOpts{live: live, probeDuration: row.Duration}) - _, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1", stalePinRequest(), 0) + _, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") if ok { t.Fatal("a foreign-content candidate must not recover the pin") } - if got := resolveCalls.Load(); got != 0 { - t.Fatalf("resolver calls = %d, want 0 (the foreign candidate is never resolved)", got) + if got := fx.resolveCalls.Load(); got != 0 { + t.Fatalf("resolves = %d, want 0 (the foreign candidate is never resolved)", got) } } -// (e) The recovery is a single retry, not a loop: even when the matched -// candidate also fails, exactly one recovery resolve is attempted. -func TestStalePinRecoverySingleRetryBound(t *testing.T) { +// (e) A fingerprint match whose resolve fails is a single attempt, not a loop: +// exactly one resolve is tried, and no evidence is persisted. +func TestStalePinRecoveryResolveFailureIsSingleAttempt(t *testing.T) { const neutral = "virtual://movie/tt-stale-retry" - row := stalePinRow(6, "movie-stale-retry", neutral, "STALE") + row := stalePinRow(7, "movie-stale-retry", neutral, "STALE") live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} - h, resolveCalls, _ := stalePinRecoveryHandler(live, neutral+"?result=STALE", true) + fx := newStalePinRecoveryFixture(neutral+"?result=STALE", stalePinRecoveryHandlerOpts{ + live: live, + resolveErr: errors.New("resolve failed"), + probeDuration: row.Duration, + }) - _, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1", stalePinRequest(), 0) + _, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") if ok { - t.Fatal("a persistently failing recovery resolve must not report success") + t.Fatal("a failing recovery resolve must not report success") } - if got := resolveCalls.Load(); got != 1 { - t.Fatalf("resolver calls = %d, want exactly 1 (bounded single retry)", got) + if got := fx.resolveCalls.Load(); got != 1 { + t.Fatalf("resolves = %d, want exactly 1 (bounded single attempt)", got) + } + if got := fx.listCalls.Load(); got != 1 { + t.Fatalf("listings = %d, want exactly 1 (no fallback after a failed resolve)", got) + } + if got := fx.saveCalls.Load(); got != 0 { + t.Fatalf("evidence writes = %d, want 0 on a failed resolve", got) } } @@ -203,21 +306,22 @@ func TestStalePinRecoverySingleRetryBound(t *testing.T) { // the recovery does not fire and no listing or resolve is attempted. func TestStalePinRecoverySkippedForIdentityCarryingRow(t *testing.T) { const neutral = "virtual://movie/tt-stale-identity" - row := stalePinRow(7, "movie-stale-identity", neutral, "STALE") + row := stalePinRow(8, "movie-stale-identity", neutral, "STALE") row.ProviderVideoHash = "hash-a" // durable identity present - h, resolveCalls, listCalls := stalePinRecoveryHandler( - []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")}, - neutral+"?result=STALE", false) + fx := newStalePinRecoveryFixture(neutral+"?result=STALE", stalePinRecoveryHandlerOpts{ + live: []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")}, + probeDuration: row.Duration, + }) - _, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1", stalePinRequest(), 0) + _, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") if ok { t.Fatal("an identity-carrying row must not be recovered by the identity-less path") } - if got := resolveCalls.Load(); got != 0 { - t.Fatalf("resolver calls = %d, want 0 (identity-carrying rows use the resolver rematch)", got) + if got := fx.resolveCalls.Load(); got != 0 { + t.Fatalf("resolves = %d, want 0 (identity-carrying rows use the resolver rematch)", got) } - if got := listCalls.Load(); got != 0 { - t.Fatalf("lister calls = %d, want 0 (the recovery must not even list)", got) + if got := fx.listCalls.Load(); got != 0 { + t.Fatalf("listings = %d, want 0 (the recovery must not even list)", got) } } @@ -228,121 +332,293 @@ func TestStalePinRecoveryKillSwitchDisables(t *testing.T) { t.Cleanup(func() { virtualStalePinRecoveryEnabled = prev }) const neutral = "virtual://movie/tt-stale-kill" - row := stalePinRow(8, "movie-stale-kill", neutral, "STALE") - h, resolveCalls, listCalls := stalePinRecoveryHandler( - []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")}, - neutral+"?result=STALE", false) + row := stalePinRow(9, "movie-stale-kill", neutral, "STALE") + fx := newStalePinRecoveryFixture(neutral+"?result=STALE", stalePinRecoveryHandlerOpts{ + live: []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")}, + probeDuration: row.Duration, + }) - _, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1", stalePinRequest(), 0) + _, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") if ok { t.Fatal("the kill switch must disable recovery") } - if got := resolveCalls.Load(); got != 0 { - t.Fatalf("resolver calls = %d, want 0 with the kill switch off", got) + if got := fx.resolveCalls.Load(); got != 0 { + t.Fatalf("resolves = %d, want 0 with the kill switch off", got) + } + if got := fx.listCalls.Load(); got != 0 { + t.Fatalf("listings = %d, want 0 with the kill switch off", got) + } +} + +// (3a) A NO-PROBER handler is a no-match: with no probe there is no measured +// duration, so the recovery must not compare against the copied row's stale +// runtime, and it must not persist or pin. +func TestStalePinRecoveryNoProberIsNoMatch(t *testing.T) { + const neutral = "virtual://movie/tt-stale-noprober" + row := stalePinRow(10, "movie-stale-noprober", neutral, "STALE") + live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} + fx := newStalePinRecoveryFixture(neutral+"?result=STALE", stalePinRecoveryHandlerOpts{live: live, noProber: true}) + + _, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") + if ok { + t.Fatal("a no-prober handler must not recover: there is no measured duration") + } + if got := fx.saveCalls.Load(); got != 0 { + t.Fatalf("evidence writes = %d, want 0 without a probe", got) + } + if got := fx.handler.peekVirtualSticky(fx.stickyKey(row)); got != "" { + t.Fatalf("sticky pin = %q, want none without a probe", got) + } +} + +// (3b) A failing probe is a no-match: the stale copied duration must not be used +// to authorize the recovery. +func TestStalePinRecoveryProbeFailureIsNoMatch(t *testing.T) { + const neutral = "virtual://movie/tt-stale-probefail" + row := stalePinRow(11, "movie-stale-probefail", neutral, "STALE") + live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} + fx := newStalePinRecoveryFixture(neutral+"?result=STALE", stalePinRecoveryHandlerOpts{ + live: live, + probeErr: errors.New("probe refused the stream"), + }) + + _, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") + if ok { + t.Fatal("a failed probe must not recover") + } + if got := fx.saveCalls.Load(); got != 0 { + t.Fatalf("evidence writes = %d, want 0 on a failed probe", got) + } + if got := fx.handler.peekVirtualSticky(fx.stickyKey(row)); got != "" { + t.Fatalf("sticky pin = %q, want none on a failed probe", got) + } +} + +// (3c) The measured duration is used: a probe that measures a runtime outside +// the tolerance rejects the recovery, and the rejection happens BEFORE any +// evidence persist or sticky-pin update. +func TestStalePinRecoveryDurationMismatchLeavesNoSideEffects(t *testing.T) { + const neutral = "virtual://movie/tt-stale-duration" + row := stalePinRow(12, "movie-stale-duration", neutral, "STALE") + live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} + // 300s against a 7200s row: far outside the 2% duration tolerance. + fx := newStalePinRecoveryFixture(neutral+"?result=STALE", stalePinRecoveryHandlerOpts{live: live, probeDuration: 300}) + + _, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") + if ok { + t.Fatal("a measured duration mismatch must not recover") + } + if got := fx.saveCalls.Load(); got != 0 { + t.Fatalf("evidence writes = %d, want 0: validation must precede persistence", got) } - if got := listCalls.Load(); got != 0 { - t.Fatalf("lister calls = %d, want 0 with the kill switch off", got) + if got := fx.handler.peekVirtualSticky(fx.stickyKey(row)); got != "" { + t.Fatalf("sticky pin = %q, want none: validation must precede the pin update", got) } } -// TestStalePinRecoveryIntegratedInVersionFallbackWalk proves the walk actually -// invokes the recovery and that it is load-bearing: the primary resolve fails -// because the provider answers empty for the pin's own listing AND for the -// stale-source fallback's re-list, but the recovery's neutral re-list answers a -// renumbered candidate whose fingerprint matches, so the walk recovers instead -// of terminaling. The control run below (recovery disabled) reaches the same -// dead pin and must terminal, proving the recovery — not an ordinary -// substitution — is what recovered it. -// -// The lister answers empty for the first two calls (the primary listing and the -// fallback's re-list) and the renumbered sibling thereafter, so the primary -// genuinely terminals before the recovery is consulted. The resolver answers -// the pin with an absent-pin refusal (a genuine renumber) and the sibling with a -// successful resolve. -func TestStalePinRecoveryIntegratedInVersionFallbackWalk(t *testing.T) { - const neutral = "virtual://movie/tt-stale-walk" - row := stalePinRow(9, "movie-stale-walk", neutral, "STALE") +// (3d) The measured duration is used: a probe that measures a matching runtime +// (even when the copied row carried a different one) recovers, proving the +// comparison is against the probe result rather than the copied value. +func TestStalePinRecoveryUsesMeasuredDuration(t *testing.T) { + const neutral = "virtual://movie/tt-stale-measured" + row := stalePinRow(13, "movie-stale-measured", neutral, "STALE") + // The copied row carries a stale runtime; the probe measures the true one. + row.Duration = 10 live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} + fx := newStalePinRecoveryFixture(neutral+"?result=STALE", stalePinRecoveryHandlerOpts{live: live, probeDuration: row.Duration}) + + recovered, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") + if !ok { + t.Fatal("a probe measuring a matching duration must recover") + } + if recovered.File == nil || recovered.File.Duration != row.Duration { + t.Fatalf("recovered file duration = %v, want the measured %d", recovered.File, row.Duration) + } + if got := fx.handler.peekVirtualSticky(fx.stickyKey(row)); got != neutral+"?result=NEW" { + t.Fatalf("sticky pin = %q, want the matched candidate after a successful recovery", got) + } +} +// (4) A slow recovery yields to healthy alternates: the recovery's sub-budget +// expires and the version-fallback walk still resolves a healthy alternate +// instead of the recovery consuming the whole parent budget. +func TestStalePinRecoverySlowYieldsToHealthyAlternate(t *testing.T) { + prevBudget := virtualStalePinRecoveryBudget + virtualStalePinRecoveryBudget = 100 * time.Millisecond + t.Cleanup(func() { virtualStalePinRecoveryBudget = prevBudget }) + + const neutral = "virtual://movie/tt-stale-slow" + row := stalePinRow(14, "movie-stale-slow", neutral, "STALE") + alt := &models.MediaFile{ID: 15, ContentID: row.ContentID, FilePath: neutral + "?result=ALT", VirtualOwnerInstallationID: 5} + + // The primary resolve (its listing and the stale-source fallback's re-list) + // answers empty so the primary genuinely terminals; the recovery's own + // listing (call 3) blocks until its sub-budget expires, simulating a hung + // provider; later alternate listings answer empty so the alternate resolves + // through the detailed resolver. + var listCalls atomic.Int32 + h := &PlaybackHandler{ + FileVersionFetcher: testPlaybackFileVersionFetcher{byContent: map[string][]*models.MediaFile{row.ContentID: {row, alt}}}, + VirtualPlaybackResolver: VirtualPlaybackResolverFunc(func(context.Context, string, int, string, int) (string, error) { + return "", errors.New("simple resolver must not be used when the detailed resolver is set") + }), + VirtualPlaybackStreamLister: VirtualPlaybackStreamListerFunc(func(ctx context.Context, _ string, _ int, _ string, _ int) ([]VirtualPlaybackStream, error) { + switch listCalls.Add(1) { + case 3: + // The recovery's listing: hang until the recovery sub-budget + // expires. + <-ctx.Done() + return nil, ctx.Err() + default: + return nil, nil + } + }), + VirtualMediaDetailedResolver: VirtualMediaDetailedResolverFunc(func(_ context.Context, uri string, _ int, _ int, _ string, _ bool, _ []string, _ string) (ResolvedVirtualMedia, error) { + if virtualResultCandidateID(uri) == "ALT" { + return ResolvedVirtualMedia{URL: "http://127.0.0.1:8080/stream?uri=" + uri, URI: uri, CandidateID: "ALT"}, nil + } + return ResolvedVirtualMedia{}, absentSessionPinError(virtualResultCandidateID(uri)) + }), + } + req := stalePinRecoveryRequest() + start := time.Now() + resolved, err := h.resolveVirtualStartWithVersionFallback(req, row, "profile-1", playback.StartRequestV3{QualityPreference: "auto"}, false, 0) + elapsed := time.Since(start) + if err != nil { + t.Fatalf("walk after a slow recovery: %v", err) + } + if resolved.File == nil || virtualResultCandidateID(resolved.URI) != "ALT" { + t.Fatalf("resolved = %#v, want the healthy alternate ALT after the recovery yielded", resolved) + } + // The recovery must not have consumed the whole parent decision budget: the + // alternate resolved well inside it. + if elapsed >= virtualStartVersionFallbackDecisionBudget { + t.Fatalf("walk took %s, want < the %s parent budget", elapsed, virtualStartVersionFallbackDecisionBudget) + } +} + +// (5) Session-bound refusal through the WALK: a walk that carries the +// session-bound intent must not re-pin, while the same walk carrying the +// fresh-start intent does. This proves the recovery preserves the caller's +// binding intent rather than forcing it false. +func TestStalePinRecoveryBindingIntentPreservedThroughWalk(t *testing.T) { + const neutral = "virtual://movie/tt-stale-walk-bind" + row := stalePinRow(16, "movie-stale-walk-bind", neutral, "STALE") + live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} + + // newHandler wires a primary outage followed by a non-empty recovery listing + // so the walk reaches the recovery; the resolver answers the pin with an + // absent-pin refusal and the sibling with a successful resolve. newHandler := func() *PlaybackHandler { - var listCalls atomic.Int32 - return &PlaybackHandler{ + var listCalls, resolveCalls atomic.Int32 + h := &PlaybackHandler{ FileVersionFetcher: testPlaybackFileVersionFetcher{byContent: map[string][]*models.MediaFile{row.ContentID: {row}}}, VirtualPlaybackResolver: VirtualPlaybackResolverFunc(func(context.Context, string, int, string, int) (string, error) { return "", errors.New("simple resolver must not be used when the detailed resolver is set") }), - VirtualPlaybackStreamLister: VirtualPlaybackStreamListerFunc(func(context.Context, string, int, string, int) ([]VirtualPlaybackStream, error) { - if listCalls.Add(1) <= 2 { - // The primary listing and the fallback's re-list: the - // provider is empty, so the primary resolve terminal. + VirtualPlaybackStreamLister: VirtualPlaybackStreamListerFunc(func(ctx context.Context, _ string, _ int, _ string, _ int) ([]VirtualPlaybackStream, error) { + if listCalls.Add(1) == 1 { return nil, nil } - // The recovery's re-list: the renumbered sibling. return live, nil }), VirtualMediaDetailedResolver: VirtualMediaDetailedResolverFunc(func(_ context.Context, uri string, _ int, _ int, _ string, _ bool, _ []string, _ string) (ResolvedVirtualMedia, error) { + resolveCalls.Add(1) if virtualResultCandidateID(uri) == "STALE" { return ResolvedVirtualMedia{}, absentSessionPinError("STALE") } return ResolvedVirtualMedia{URL: "http://127.0.0.1:8080/stream?uri=" + uri, URI: uri, CandidateID: virtualResultCandidateID(uri)}, nil }), + VirtualPlaybackSourceProber: func(_ context.Context, _ string, f *models.MediaFile) (*models.MediaFile, error) { + probed := stalePinProbedTrack(row.Duration) + probed.ID = f.ID + probed.FilePath = f.FilePath + return probed, nil + }, } + return h } - req := stalePinRecoveryRequest() - resolved, err := newHandler().resolveVirtualStartWithVersionFallback(req, row, "profile-1", stalePinRequest(), false, 0) + // Fresh start: the request declares unbound, the walk threads it, recovery runs. + freshReq := stalePinRecoveryRequest() + fresh, err := newHandler().resolveVirtualStartWithVersionFallback(freshReq, row, "profile-1", playback.StartRequestV3{QualityPreference: "auto"}, false, 0) if err != nil { - t.Fatalf("walk recovery: %v", err) + t.Fatalf("fresh-start walk: %v", err) } - if resolved.File == nil || virtualResultCandidateID(resolved.URI) != "NEW" { - t.Fatalf("resolved = %#v, want the fingerprint-matched candidate NEW", resolved) + if fresh.File == nil || virtualResultCandidateID(fresh.URI) != "NEW" { + t.Fatalf("fresh-start resolved = %#v, want the recovered NEW", fresh) } - // Control: with the recovery disabled the same dead pin terminals, so the - // recovery is what recovers it. - prev := virtualStalePinRecoveryEnabled - virtualStalePinRecoveryEnabled = false - t.Cleanup(func() { virtualStalePinRecoveryEnabled = prev }) - if _, controlErr := newHandler().resolveVirtualStartWithVersionFallback(req, row, "profile-1", stalePinRequest(), false, 0); controlErr == nil { - t.Fatal("with the recovery disabled the dead pin must terminal; the recovery is masking a walk that already substitutes") + // Session-bound: the request declares bound, the walk preserves it, and the + // recovery must refuse to re-pin. + boundReq := stalePinSessionBoundRequest() + bound, err := newHandler().resolveVirtualStartWithVersionFallback(boundReq, row, "profile-1", playback.StartRequestV3{QualityPreference: "auto"}, false, 0) + if err == nil { + t.Fatalf("session-bound walk = %#v, want the primary terminal preserved", bound) + } + if virtualResultCandidateID(bound.URI) == "NEW" { + t.Fatal("a session-bound walk must not re-pin the absent candidate") } } -// TestStalePinRecoveryRefusesSessionBound proves the cold-start-only contract: -// a session-bound resolve must keep refusing an absent pin, never re-pin it. -// The default (no declaration) is the conservative session-bound intent. -func TestStalePinRecoveryRefusesSessionBound(t *testing.T) { - const neutral = "virtual://movie/tt-stale-bound" - row := stalePinRow(11, "movie-stale-bound", neutral, "STALE") - live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} - h, resolveCalls, listCalls := stalePinRecoveryHandler(live, neutral+"?result=STALE", false) - - _, ok := h.recoverStaleIdentityLessPinV3(stalePinSessionBoundRequest(), row, "profile-1", stalePinRequest(), 0) - if ok { - t.Fatal("a session-bound resolve must not be re-pinned") +// The fingerprint predicate is the boundary: inside tolerance accepts, outside +// refuses, and only the duration tier is neutral for an unknown side. +func TestStalePinFingerprintTolerance(t *testing.T) { + row := &models.MediaFile{CodecVideo: "h264", CodecAudio: "aac", FileSize: 1_000_000, Duration: 7200} + cases := []struct { + name string + candidate VirtualPlaybackStream + want bool + }{ + {"exact", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "aac", FileSize: 1_000_000}, true}, + {"size +4%", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "aac", FileSize: 1_040_000}, true}, + {"size -5%", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "aac", FileSize: 950_000}, true}, + {"size +6% refused", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "aac", FileSize: 1_060_000}, false}, + {"unknown candidate size refused", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "aac", FileSize: 0}, false}, + {"codec mismatch refused", VirtualPlaybackStream{CodecVideo: "hevc", CodecAudio: "aac", FileSize: 1_000_000}, false}, + {"unknown candidate codec refused", VirtualPlaybackStream{CodecAudio: "aac", FileSize: 1_000_000}, false}, + {"audio codec mismatch refused", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "eac3", FileSize: 1_000_000}, false}, + {"unknown candidate audio neutral", VirtualPlaybackStream{CodecVideo: "h264", FileSize: 1_000_000}, true}, } - if got := resolveCalls.Load(); got != 0 { - t.Fatalf("resolver calls = %d, want 0 for a session-bound resolve", got) + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := virtualStalePinStreamFingerprintMatch(row, tc.candidate); got != tc.want { + t.Fatalf("fingerprint = %v, want %v", got, tc.want) + } + }) } - if got := listCalls.Load(); got != 0 { - t.Fatalf("lister calls = %d, want 0 (the recovery must not even list)", got) + + durations := []struct { + row, candidate int + want bool + }{ + {7200, 7200, true}, + {7200, 7300, true}, // +1.4% + {7200, 7400, false}, // +2.8% + {0, 7300, true}, // unknown row duration is neutral + {7200, 0, true}, // unknown candidate duration is neutral + } + for _, tc := range durations { + if got := virtualStalePinDurationMatch(tc.row, tc.candidate); got != tc.want { + t.Fatalf("duration(%d,%d) = %v, want %v", tc.row, tc.candidate, got, tc.want) + } } } // TestStalePinRecoveryClearsOldCandidateState pins the in-memory re-pin's field -// hygiene: the matched candidate must not inherit the previous pin's stored -// URL, headers, durable identity, or probe evidence. Carrying the stored URL -// would serve the old bytes under the new id; carrying the probe stamp would -// report the old inventory as verified for the new candidate. The row's own -// lifecycle fields (verdict and delivery stamp) are deliberately preserved, so -// the recovery does not alter verdict semantics. +// hygiene via the WALK (the recovery itself): the matched candidate must not +// inherit the previous pin's stored URL, headers, durable identity, or probe +// evidence. Carrying the stored URL would serve the old bytes under the new id; +// carrying the probe stamp would report the old inventory as verified for the +// new candidate. The row's own lifecycle fields (verdict and delivery stamp) are +// deliberately preserved, so the recovery does not alter verdict semantics. func TestStalePinRecoveryClearsOldCandidateState(t *testing.T) { const neutral = "virtual://movie/tt-stale-clear" expiry := time.Now().Add(2 * time.Hour) delivered := time.Now().Add(-time.Minute) failedAt := time.Now().Add(-time.Minute) probedAt := time.Now().Add(-time.Minute) - row := stalePinRow(13, "movie-stale-clear", neutral, "STALE") + row := stalePinRow(17, "movie-stale-clear", neutral, "STALE") row.ResolvedURL = "https://old.example/stale.mkv" row.ResolvedURLExpiresAt = &expiry row.ProviderRequestHeaders = map[string]string{"Referer": "https://old.example/"} @@ -359,14 +635,16 @@ func TestStalePinRecoveryClearsOldCandidateState(t *testing.T) { row.ExternalSubtitles = []models.ExternalSubtitle{{Path: "/tmp/old.srt"}} live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} - - match, found := virtualStalePinBestStreamMatch(row, live) + match, matchedStream, found := virtualStalePinBestStreamMatch(row, live) if !found { t.Fatal("the fingerprint match was not found") } if match.FilePath != neutral+"?result=NEW" { t.Fatalf("matched path = %q, want the live candidate", match.FilePath) } + if matchedStream.URI != neutral+"?result=NEW" { + t.Fatalf("matched stream URI = %q, want the live candidate", matchedStream.URI) + } if match.ResolvedURL != "" || match.ResolvedURLExpiresAt != nil || match.ProviderRequestHeaders != nil { t.Fatalf("re-pin carried the old transport state: url=%q expires=%v headers=%v", match.ResolvedURL, match.ResolvedURLExpiresAt, match.ProviderRequestHeaders) } @@ -387,46 +665,24 @@ func TestStalePinRecoveryClearsOldCandidateState(t *testing.T) { } } -// The fingerprint predicate is the boundary: inside tolerance accepts, outside -// refuses, and only the duration tier is neutral for an unknown side. -func TestStalePinFingerprintTolerance(t *testing.T) { - row := &models.MediaFile{CodecVideo: "h264", CodecAudio: "aac", FileSize: 1_000_000, Duration: 7200} - cases := []struct { - name string - candidate VirtualPlaybackStream - want bool - }{ - {"exact", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "aac", FileSize: 1_000_000}, true}, - {"size +4%", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "aac", FileSize: 1_040_000}, true}, - {"size -5%", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "aac", FileSize: 950_000}, true}, - {"size +6% refused", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "aac", FileSize: 1_060_000}, false}, - {"unknown candidate size refused", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "aac", FileSize: 0}, false}, - {"codec mismatch refused", VirtualPlaybackStream{CodecVideo: "hevc", CodecAudio: "aac", FileSize: 1_000_000}, false}, - {"unknown candidate codec refused", VirtualPlaybackStream{CodecAudio: "aac", FileSize: 1_000_000}, false}, - {"audio codec mismatch refused", VirtualPlaybackStream{CodecVideo: "h264", CodecAudio: "eac3", FileSize: 1_000_000}, false}, - {"unknown candidate audio neutral", VirtualPlaybackStream{CodecVideo: "h264", FileSize: 1_000_000}, true}, - } - for _, tc := range cases { - t.Run(tc.name, func(t *testing.T) { - if got := virtualStalePinStreamFingerprintMatch(row, tc.candidate); got != tc.want { - t.Fatalf("fingerprint = %v, want %v", got, tc.want) - } - }) - } +// TestStalePinRecoveryRefusesSessionBound is a focused unit test of the helper: +// a session-bound resolve must keep refusing an absent pin, never re-pin it. +func TestStalePinRecoveryRefusesSessionBound(t *testing.T) { + const neutral = "virtual://movie/tt-stale-bound" + row := stalePinRow(18, "movie-stale-bound", neutral, "STALE") + fx := newStalePinRecoveryFixture(neutral+"?result=STALE", stalePinRecoveryHandlerOpts{ + live: []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")}, + probeDuration: row.Duration, + }) - durations := []struct { - row, candidate int - want bool - }{ - {7200, 7200, true}, - {7200, 7300, true}, // +1.4% - {7200, 7400, false}, // +2.8% - {0, 7300, true}, // unknown row duration is neutral - {7200, 0, true}, // unknown candidate duration is neutral + _, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinSessionBoundRequest(), row, "profile-1") + if ok { + t.Fatal("a session-bound resolve must not be re-pinned") } - for _, tc := range durations { - if got := virtualStalePinDurationMatch(tc.row, tc.candidate); got != tc.want { - t.Fatalf("duration(%d,%d) = %v, want %v", tc.row, tc.candidate, got, tc.want) - } + if got := fx.resolveCalls.Load(); got != 0 { + t.Fatalf("resolves = %d, want 0 for a session-bound resolve", got) + } + if got := fx.listCalls.Load(); got != 0 { + t.Fatalf("listings = %d, want 0 (the recovery must not even list)", got) } } From 130c36f3adbefc2c15ad08dbd5b8247b2ad3927c Mon Sep 17 00:00:00 2001 From: randrini Date: Tue, 6 Oct 2026 21:00:32 +0200 Subject: [PATCH 4/4] fix(playback): keep stale-pin recovery transient and unambiguous Recovery for an identity-less stale pin re-pinned in memory but then handed the adoption-capable evidence writer the original row with the replacement URI, which took the path-adoption branch and rewrote the durable ?result= pin it promised never to touch. Persist replacement evidence only against a row that already verifiably owns the matched candidate's concrete path, and then as metadata only; with no existing owner the recovery leaves no durable trace. Allocate the write generation before the listing instead of at publication, so an earlier-started, later-finishing recovery cannot overwrite a newer selection that finished first. Refuse when more than one live candidate identity carries the row's fingerprint rather than rank-picking the first: same-title releases share codec, size and runtime, so size+codec cannot separate language or edition variants. Veto known resolution and audio-language conflicts so those variants are excluded before the uniqueness count. Require a concrete matched candidate from the resolver: reject a provider-neutral URI, a different release under the same neutral key, a candidate id inconsistent with the returned URI, and an owner outside the matched release scope. Add DB-backed tests that read the whole media_files row back and prove the durable pin is byte-identical, plus unit tests for ambiguous listings in both orders, interleaved generation, and the substitution cases. --- .../handlers/playback_virtual_stale_pin.go | 231 +++++++++-- .../playback_virtual_stale_pin_db_test.go | 189 +++++++++ .../playback_virtual_stale_pin_test.go | 364 +++++++++++++++++- 3 files changed, 740 insertions(+), 44 deletions(-) create mode 100644 internal/api/handlers/playback_virtual_stale_pin_db_test.go diff --git a/internal/api/handlers/playback_virtual_stale_pin.go b/internal/api/handlers/playback_virtual_stale_pin.go index 37b931d3c5..dbaf9779cf 100644 --- a/internal/api/handlers/playback_virtual_stale_pin.go +++ b/internal/api/handlers/playback_virtual_stale_pin.go @@ -13,6 +13,7 @@ import ( apimw "github.com/Silo-Server/silo-server/internal/api/middleware" "github.com/Silo-Server/silo-server/internal/models" "github.com/Silo-Server/silo-server/internal/virtuallibrary" + "github.com/Silo-Server/silo-server/internal/virtuallibrary/stream" ) // Guarded stale-pin recovery for identity-less rows. @@ -32,26 +33,32 @@ import ( // blackout, not a renumber, and keeps its existing terminal); // - the bound pin id is absent from that listing (otherwise the normal resolve // owns the candidate); -// - a live candidate for the SAME content (same provider-neutral path, so never -// across a different content_id) matches the row's stored evidence -// fingerprint: size within virtualStalePinSizeTolerance, an equal video codec, -// and a probed duration within virtualStalePinDurationTolerance when both are +// - EXACTLY ONE live candidate for the SAME content (same provider-neutral +// path, so never across a different content_id) matches the row's stored +// evidence fingerprint: size within virtualStalePinSizeTolerance, an equal +// video codec, no conflicting known resolution or audio-language signal, and +// a probed duration within virtualStalePinDurationTolerance when both are // known. // -// When it fires it re-pins IN MEMORY ONLY (the durable catalog row's ?result= is -// never rewritten) and resolves the matched candidate once. If no candidate -// matches, the caller's terminal is preserved. +// When it fires it re-pins IN MEMORY ONLY: the durable catalog row's ?result= is +// never rewritten. Replacement evidence is persisted only against a row that +// already verifiably owns the matched candidate's concrete path (an alternate +// owner), and then as metadata only; when no such row exists the recovery leaves +// no durable trace. It resolves the matched candidate once. If no candidate +// matches, or more than one does, the caller's terminal is preserved. // -// Mismatch-rate concern: a fingerprint is evidence, not proof. A renumbered -// release whose size drifted past the tolerance, or two releases of one title -// with coincidentally equal size+codec+duration, could be confused. Adopting the -// wrong bytes is worse than a terminal, so the recovery is deliberately one-shot -// and kill-switchable. Every recovery logs at Info with the matched ids; if the +// Mismatch-rate concern: a fingerprint is evidence, not proof. A single matching +// candidate is a heuristic, not a proof that it is the same release: same-title +// releases share codec, size and runtime, so size+codec cannot separate language +// or edition variants. When more than one candidate identity matches, the +// recovery refuses rather than rank-picking, because adopting the wrong bytes is +// worse than a terminal. The recovery is deliberately one-shot and +// kill-switchable. Every recovery logs at Info with the matched ids; if the // observed mismatch rate exceeds ~1%, disable it with the kill switch below. // -// Kill switch: set SILO_DISABLE_VIRTUAL_STALE_PIN_RECOVERY=1 at process start to -// disable the recovery entirely (fast, default-on). Tests override the var -// directly. +// Kill switch: set SILO_DISABLE_VIRTUAL_STALE_PIN_RECOVERY to any non-empty +// value (conventionally 1) at process start to disable the recovery entirely +// (fast, default-on). Tests override the var directly. var virtualStalePinRecoveryEnabled = os.Getenv("SILO_DISABLE_VIRTUAL_STALE_PIN_RECOVERY") == "" const ( @@ -147,6 +154,14 @@ func (h *PlaybackHandler) recoverStaleIdentityLessPinV3( } ctx := r.Context() userID := apimw.GetUserID(ctx) + // Capture ONE write generation before the recovery does any work, exactly as + // the general resolver does. Allocating it at publication time would let an + // earlier-started, later-finishing recovery (or a selection racing it) carry + // a newer generation than a newer selection that actually finished first, and + // so overwrite that newer selection's pin and evidence. Allocating here makes + // the generation order the start order, and every write below carries this + // token. + generation := nextVirtualCacheGeneration() // Bound the whole recovery well under the walk's decision budget so the // alternates the walk still has to try keep their time. recoveryCtx, cancel := context.WithTimeout(ctx, virtualStalePinRecoveryBudget) @@ -173,14 +188,24 @@ func (h *PlaybackHandler) recoverStaleIdentityLessPinV3( if len(candidates) == 0 { return resolvedVirtualPlaybackSource{}, false } - match, matchedStream, found := virtualStalePinBestStreamMatch(file, candidates) - if !found { + match, matchedStream, candidatesMatched := virtualStalePinMatchStream(file, candidates) + if candidatesMatched == 0 { slog.DebugContext(ctx, "virtual stale-pin recovery found no fingerprint match", "component", "api", "file_id", file.ID, "old_candidate_id", pinnedID, "candidate_count", len(candidates)) return resolvedVirtualPlaybackSource{}, false } - // Resolve and probe the EXACT re-pinned candidate in a single attempt. + if candidatesMatched != 1 { + // More than one live candidate identity carries the row's fingerprint. + // The fingerprint cannot separate same-title releases (shared codec, + // size and runtime), so every one of them remains a candidate for the + // pin, and rank-picking would authorize a coin flip. Refuse. + slog.InfoContext(ctx, "virtual stale-pin recovery refused: ambiguous candidate identities share the row's fingerprint", + "component", "api", "file_id", file.ID, "old_candidate_id", pinnedID, + "candidate_count", len(candidates), "matching_identities", candidatesMatched) + return resolvedVirtualPlaybackSource{}, false + } + // Resolve and probe the single matched candidate in one attempt. ownerID := effectiveVirtualOwner(match.VirtualOwnerInstallationID, file.VirtualOwnerInstallationID) matchedID := virtualResultCandidateID(match.FilePath) resolution, resolveErr := h.resolveStalePinCandidateOnceV3(recoveryCtx, &match, matchedID, matchedStream, ownerID, userID, profileID) @@ -198,14 +223,30 @@ func (h *PlaybackHandler) recoverStaleIdentityLessPinV3( return resolvedVirtualPlaybackSource{}, false } // Validation passed; only now may the recovery touch shared state. Pin the - // matched candidate for later starts and persist its probed evidence. + // matched candidate for later starts with the generation captured before the + // listing. fingerprint := "" if caps, ok := h.requestDeviceCapabilities(r); ok { fingerprint = caps.Fingerprint() } stickyKey := bestResultCacheKey(file.ContentID, neutralKey, file.VirtualOwnerInstallationID, fingerprint) - h.pinVirtualStickyAt(stickyKey, resolution.uri, nextVirtualCacheGeneration()) - h.persistVirtualProbeEvidence(ctx, file, resolution.uri, resolution.file, true, true) + h.pinVirtualStickyAt(stickyKey, resolution.uri, generation) + // Persist the replacement evidence ONLY against a row that already + // verifiably owns the matched candidate's concrete path. The requested row + // does not: that is precisely why this recovery runs, and handing the + // adoption-capable writer the original identity-less row with the + // replacement URI would take its path-adoption branch and rewrite the + // durable pin. The recovery is transient: it must never rewrite the + // requested row's ?result=, so when no alternate owner exists it persists + // nothing rather than adopt the path. + if ownerRow, ownerErr := h.virtualPathOwnerRow(recoveryCtx, file, resolution.uri, ownerID, file.MediaFolderID); ownerErr == nil && + ownerRow != nil && ownerRow.ID != file.ID { + // ownerRow already owns this exact path, so a metadata-only write + // (no AdoptPath) stamps the verified inventory in place without moving + // any file_path. Pass the owner row so the shared builder cannot fall + // back to adopting onto a different row. + h.persistVirtualProbeEvidence(ctx, ownerRow, resolution.uri, resolution.file, true, true) + } slog.InfoContext(ctx, "virtual stale-pin recovery re-pinned an identity-less candidate in memory", "component", "api", "file_id", file.ID, "old_candidate_id", pinnedID, "new_candidate_id", virtualResultCandidateID(resolution.uri), @@ -258,12 +299,30 @@ func (h *PlaybackHandler) resolveStalePinCandidateOnceV3( } resolvedURI := strings.TrimSpace(res.URI) if resolvedURI == "" { - resolvedURI = match.FilePath + return stalePinCandidateResolutionV3{}, errors.New("stale-pin candidate resolved without a concrete URI") + } + // The resolver must have returned the exact matched candidate. A returned + // URI that carries no concrete pick is not the matched identity, and a + // sibling it preferred (a different pick under the same neutral key, or a + // candidate under a different release/content) is not the bytes the + // fingerprint authorized. The match is made on the candidate's concrete + // ?result= identity, so a concrete, matching identity is required on every + // returned surface; there is no neutral or same-id-on-other-content escape. + returnedID := virtualResultCandidateID(resolvedURI) + if returnedID == "" { + return stalePinCandidateResolutionV3{}, fmt.Errorf("resolver returned a provider-neutral URI %q for matched candidate %q", resolvedURI, matchedID) } - // The resolver must have returned the exact matched candidate. A sibling it - // preferred (a substitution) is not the bytes the fingerprint authorized. - if got := virtualResultCandidateID(resolvedURI); got != "" && matchedID != "" && got != matchedID { - return stalePinCandidateResolutionV3{}, fmt.Errorf("resolver substituted candidate %q for matched %q", got, matchedID) + if trimmedCandidateID := strings.TrimSpace(res.CandidateID); trimmedCandidateID != "" && trimmedCandidateID != returnedID { + return stalePinCandidateResolutionV3{}, fmt.Errorf("resolver candidate id %q is inconsistent with returned URI %q", res.CandidateID, resolvedURI) + } + if returnedID != matchedID { + return stalePinCandidateResolutionV3{}, fmt.Errorf("resolver substituted candidate %q for matched %q", returnedID, matchedID) + } + if virtualPlaybackNeutralKey(resolvedURI) != virtualPlaybackNeutralKey(match.FilePath) { + return stalePinCandidateResolutionV3{}, fmt.Errorf("resolver returned candidate %q under a different release than matched %q", resolvedURI, match.FilePath) + } + if ownerID > 0 && res.OwnerID > 0 && res.OwnerID != ownerID { + return stalePinCandidateResolutionV3{}, fmt.Errorf("resolver returned an owner installation %d outside the matched release scope %d", res.OwnerID, ownerID) } probeTransient := cloneVirtualProbeTransient(*match) probeTransient.FilePath = resolvedURI @@ -311,13 +370,17 @@ func (h *PlaybackHandler) resolveStalePinCandidateOnceV3( }, nil } -// virtualStalePinBestStreamMatch returns the best live candidate whose declared -// evidence matches the row's stored fingerprint, re-pinned as a copy of the row -// alongside the matched stream itself (whose declared tracks the caller may merge -// into the probe result). "Best" is the first match in provider rank order: the -// listing is already ranked, so the first fingerprint match is the strongest one, -// and a single deterministic choice keeps the recovery one-shot. It reports -// ok=false when no candidate matches. +// virtualStalePinMatchStream scans the live candidates and returns the single +// fingerprint match as the row re-pinned as a copy, alongside the matched stream +// itself (whose declared tracks the caller may merge into the probe result). +// +// A fingerprint is evidence, not proof: same-title releases share video codec, +// declared size and runtime, so more than one distinct candidate identity can +// match. In that case there is no way to tell which one the row pinned, and +// rank-picking would authorize a coin flip. matched is the number of matching +// candidate identities: 0 means no candidate matches, 1 means match and stream +// are the unique match, and >1 means the caller must refuse and ignore match and +// stream. // // The returned copy keeps the row's catalog id, its declared fingerprint fields // (size, codec, resolution, duration) — those are what the match was made @@ -329,21 +392,32 @@ func (h *PlaybackHandler) resolveStalePinCandidateOnceV3( // verified for the new candidate). The cleared state makes the caller resolve // and probe the matched candidate afresh, which is the only thing that may // authorize serving its bytes or inventory. -func virtualStalePinBestStreamMatch(row *models.MediaFile, candidates []VirtualPlaybackStream) (models.MediaFile, VirtualPlaybackStream, bool) { +func virtualStalePinMatchStream(row *models.MediaFile, candidates []VirtualPlaybackStream) (models.MediaFile, VirtualPlaybackStream, int) { + matched := 0 + var ( + match models.MediaFile + stream VirtualPlaybackStream + ) for _, candidate := range candidates { if !virtualStalePinStreamFingerprintMatch(row, candidate) { continue } - match := *row + matched++ + if matched > 1 { + // The decision is already refusal; stop scanning the rest of a + // listing that could be arbitrarily long. + return models.MediaFile{}, VirtualPlaybackStream{}, matched + } + match = *row match.FilePath = candidate.URI // Keep the row's own identity for the in-memory re-pin: the live candidate // has no catalog row yet, and adopting its row id would fabricate one. Only // the served path changes. match.VirtualOwnerInstallationID = effectiveVirtualOwner(candidate.OwnerInstallationID, row.VirtualOwnerInstallationID) clearVirtualStalePinOldCandidateState(&match) - return match, candidate, true + stream = candidate } - return models.MediaFile{}, VirtualPlaybackStream{}, false + return match, stream, matched } // clearVirtualStalePinOldCandidateState drops the fields a re-pinned row must @@ -388,7 +462,10 @@ func clearVirtualStalePinOldCandidateState(file *models.MediaFile) { // known on both sides and agree within tolerance, and the video codec must be // known on both sides and match. An audio codec mismatch is also a veto when both // sides carry one; an unknown audio codec is neutral, since some providers omit -// it. +// it. A known resolution or track-language conflict is a veto too: those are the +// few declared signals that can separate a language/edition variant from a +// same-title release whose codec, size and runtime coincide. Every unknown side +// is neutral. func virtualStalePinStreamFingerprintMatch(row *models.MediaFile, candidate VirtualPlaybackStream) bool { if row == nil { return false @@ -407,9 +484,87 @@ func virtualStalePinStreamFingerprintMatch(row *models.MediaFile, candidate Virt if rowAudio != "" && candidateAudio != "" && !strings.EqualFold(rowAudio, candidateAudio) { return false } + rowRes := strings.TrimSpace(row.Resolution) + candidateRes := strings.TrimSpace(candidate.Resolution) + if rowRes != "" && candidateRes != "" && + !strings.EqualFold(stream.NormalizeResolution(rowRes), stream.NormalizeResolution(candidateRes)) { + return false + } + if virtualStalePinKnownAudioLanguageConflict(row, candidate) { + return false + } + return true +} + +// virtualStalePinKnownAudioLanguageConflict reports whether both the row and the +// candidate carry a known audio-language set and the candidate shares no language +// with the row. An empty or unparseable-only side is neutral: the provider or the +// probe may simply not have recorded a language, and treating absence as a +// conflict would veto every unknown candidate. Both sides are canonicalized to +// ISO base subtags so "ENG" and "en" agree. +func virtualStalePinKnownAudioLanguageConflict(row *models.MediaFile, candidate VirtualPlaybackStream) bool { + rowLangs := virtualStalePinRowAudioLanguages(row) + if len(rowLangs) == 0 { + return false + } + candidateLangs := virtualStalePinCanonicalLanguages(candidate.AudioLanguages) + if len(candidateLangs) == 0 { + return false + } + for lang := range candidateLangs { + if rowLangs[lang] { + return false + } + } return true } +// virtualStalePinRowAudioLanguages collects the row's known audio languages from +// its probed tracks, canonicalized to ISO base subtags. A row whose tracks carry +// no parseable language is unknown and yields an empty set, which the caller +// treats as neutral. +func virtualStalePinRowAudioLanguages(row *models.MediaFile) map[string]bool { + if row == nil { + return nil + } + out := make(map[string]bool) + for _, track := range row.AudioTracks { + values := append([]string{track.Language}, track.Languages...) + for _, raw := range values { + if lang := virtualStalePinCanonicalLanguage(raw); lang != "" { + out[lang] = true + } + } + } + if len(out) == 0 { + return nil + } + return out +} + +// virtualStalePinCanonicalLanguages canonicalizes a provider-declared language +// list, dropping release markers (MULTI/DUAL) that are not real languages. An +// all-marker list yields an empty set, which is neutral rather than a conflict. +func virtualStalePinCanonicalLanguages(values []string) map[string]bool { + out := make(map[string]bool, len(values)) + for _, raw := range values { + if lang := virtualStalePinCanonicalLanguage(raw); lang != "" { + out[lang] = true + } + } + return out +} + +// virtualStalePinCanonicalLanguage canonicalizes one language token to its ISO +// base subtag, or "" when it is not a real language tag. +func virtualStalePinCanonicalLanguage(raw string) string { + trimmed := strings.TrimSpace(raw) + if trimmed == "" || !isRealVirtualLanguageTag(trimmed) { + return "" + } + return virtualLanguageBaseSubtag(trimmed) +} + // virtualStalePinDurationMatch applies the duration half of the fingerprint to a // probed candidate. It is neutral when either duration is unknown (a probe that // reported no runtime), so the size and codec tiers decide; when both are known diff --git a/internal/api/handlers/playback_virtual_stale_pin_db_test.go b/internal/api/handlers/playback_virtual_stale_pin_db_test.go new file mode 100644 index 0000000000..0771ed728b --- /dev/null +++ b/internal/api/handlers/playback_virtual_stale_pin_db_test.go @@ -0,0 +1,189 @@ +package handlers + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/Silo-Server/silo-server/internal/models" + "github.com/Silo-Server/silo-server/internal/scanner" +) + +// errStalePinSimpleResolverUnexpected guards the DB tests against accidentally +// wiring the simple resolver: the recovery always uses the detailed one. +var errStalePinSimpleResolverUnexpected = errors.New("simple resolver must not be used when the detailed resolver is set") + +// TestStalePinRecoveryLeavesDurablePinUnchangedInDatabase is the DB-backed proof +// the reviewer asked for: the recovery runs end to end against the real +// VirtualFileMetadataUpdate, and the requested row's durable pin comes out +// byte-identical. A struct-only assertion cannot show this — the dangerous path +// is the adoption-capable writer landing an UPDATE on the row — so the test reads +// the whole row back from Postgres and compares it. It also counts saver writes: +// with no existing owner the recovery must admit none. +func TestStalePinRecoveryLeavesDurablePinUnchangedInDatabase(t *testing.T) { + pool := virtualMetadataUpdateTestPool(t) + ctx := context.Background() + const folderID, ownerID = 994461, 7161 + seedVirtualMetadataUpdateFolder(t, pool, folderID, ownerID) + + const neutral = "virtual://movie/tt-stale-db-transient" + pinnedPath := neutral + "?result=STALE" + var rowID int + var updatedAt time.Time + var probeUpdatedAt *time.Time + if err := pool.QueryRow(ctx, ` + INSERT INTO media_files(content_id, media_folder_id, file_path, container, virtual_owner_installation_id, probe_source, + codec_video, codec_audio, resolution, file_size, duration) + VALUES('movie-stale-db-transient', $1, $2, 'virtual', $3, 'virtual', + 'h264', 'aac', '1080p', 8500000000, 7200) + RETURNING id, updated_at, probe_updated_at`, folderID, pinnedPath, ownerID, + ).Scan(&rowID, &updatedAt, &probeUpdatedAt); err != nil { + t.Fatalf("insert identity-less pinned row: %v", err) + } + before := readVirtualEvidenceRowJSON(t, pool, rowID) + + repo := scanner.NewFileRepository(pool) + row, err := repo.GetByID(ctx, rowID) + if err != nil || row == nil { + t.Fatalf("load pinned row: %v", err) + } + + var saveCalls atomic.Int32 + h := &PlaybackHandler{ + fileResolver: repo, + VirtualPlaybackResolver: VirtualPlaybackResolverFunc(func(context.Context, string, int, string, int) (string, error) { + return "", errStalePinSimpleResolverUnexpected + }), + VirtualPlaybackStreamLister: VirtualPlaybackStreamListerFunc(func(context.Context, string, int, string, int) ([]VirtualPlaybackStream, error) { + return []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")}, nil + }), + VirtualMediaDetailedResolver: VirtualMediaDetailedResolverFunc(func(_ context.Context, uri string, _ int, _ int, _ string, _ bool, _ []string, _ string) (ResolvedVirtualMedia, error) { + return ResolvedVirtualMedia{URL: "http://127.0.0.1:8080/stream?uri=" + uri, URI: uri, CandidateID: virtualResultCandidateID(uri)}, nil + }), + VirtualPlaybackSourceProber: func(_ context.Context, _ string, f *models.MediaFile) (*models.MediaFile, error) { + probed := stalePinProbedTrack(row.Duration) + probed.ID = f.ID + probed.FilePath = f.FilePath + return probed, nil + }, + VirtualFileSaver: func(writeCtx context.Context, args models.VirtualFilePersistArgs) (int64, error) { + saveCalls.Add(1) + return ExecVirtualFileMetadataUpdate(writeCtx, pool, args) + }, + } + + if _, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1"); !ok { + t.Fatal("the unique fingerprint match should recover the pin") + } + h.StopVirtualEvidence() + + if got := saveCalls.Load(); got != 0 { + t.Fatalf("evidence writes = %d, want 0: the recovery must not persist against the requested row", got) + } + if after := readVirtualEvidenceRowJSON(t, pool, rowID); after != before { + t.Fatalf("identity-less pinned row changed during a transient recovery:\nbefore=%s\nafter =%s", before, after) + } + if got := virtualResultCandidateID(row.FilePath); got != "STALE" { + t.Fatalf("in-memory row pin = %q, want the durable STALE unchanged", got) + } +} + +// TestStalePinRecoveryStampsExistingOwnerWithoutMovingPath is the allowed half +// against a real database: when a distinct row already owns the matched +// candidate's concrete path, the recovery persists the probed evidence against +// that row as a metadata-only write. The owner keeps its path; the requested +// row stays byte-identical. +func TestStalePinRecoveryStampsExistingOwnerWithoutMovingPath(t *testing.T) { + pool := virtualMetadataUpdateTestPool(t) + ctx := context.Background() + const folderID, ownerID = 994462, 7162 + seedVirtualMetadataUpdateFolder(t, pool, folderID, ownerID) + + const ( + neutral = "virtual://movie/tt-stale-db-owner" + pinnedPath = neutral + "?result=STALE" + ownerPath = neutral + "?result=NEW" + ) + var pinnedID int + var updatedAt time.Time + var probeUpdatedAt *time.Time + if err := pool.QueryRow(ctx, ` + INSERT INTO media_files(content_id, media_folder_id, file_path, container, virtual_owner_installation_id, probe_source, + codec_video, codec_audio, resolution, file_size, duration) + VALUES('movie-stale-db-owner', $1, $2, 'virtual', $3, 'virtual', + 'h264', 'aac', '1080p', 8500000000, 7200) + RETURNING id, updated_at, probe_updated_at`, folderID, pinnedPath, ownerID, + ).Scan(&pinnedID, &updatedAt, &probeUpdatedAt); err != nil { + t.Fatalf("insert pinned row: %v", err) + } + pinnedBefore := readVirtualEvidenceRowJSON(t, pool, pinnedID) + + var ownerID2 int + if err := pool.QueryRow(ctx, ` + INSERT INTO media_files(content_id, media_folder_id, file_path, container, virtual_owner_installation_id, probe_source, + codec_video, codec_audio, resolution, file_size, duration) + VALUES('movie-stale-db-owner', $1, $2, 'virtual', $3, 'virtual', + 'h264', 'aac', '1080p', 8500000000, 7200) + RETURNING id`, folderID, ownerPath, ownerID, + ).Scan(&ownerID2); err != nil { + t.Fatalf("insert owner row: %v", err) + } + + repo := scanner.NewFileRepository(pool) + row, err := repo.GetByID(ctx, pinnedID) + if err != nil || row == nil { + t.Fatalf("load pinned row: %v", err) + } + + h := &PlaybackHandler{ + fileResolver: repo, + VirtualPlaybackResolver: VirtualPlaybackResolverFunc(func(context.Context, string, int, string, int) (string, error) { + return "", errStalePinSimpleResolverUnexpected + }), + VirtualPlaybackStreamLister: VirtualPlaybackStreamListerFunc(func(context.Context, string, int, string, int) ([]VirtualPlaybackStream, error) { + return []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")}, nil + }), + VirtualMediaDetailedResolver: VirtualMediaDetailedResolverFunc(func(_ context.Context, uri string, _ int, _ int, _ string, _ bool, _ []string, _ string) (ResolvedVirtualMedia, error) { + return ResolvedVirtualMedia{URL: "http://127.0.0.1:8080/stream?uri=" + uri, URI: uri, CandidateID: virtualResultCandidateID(uri)}, nil + }), + VirtualPlaybackSourceProber: func(_ context.Context, _ string, f *models.MediaFile) (*models.MediaFile, error) { + probed := stalePinProbedTrack(row.Duration) + probed.ID = f.ID + probed.FilePath = f.FilePath + return probed, nil + }, + VirtualFileSaver: func(writeCtx context.Context, args models.VirtualFilePersistArgs) (int64, error) { + return ExecVirtualFileMetadataUpdate(writeCtx, pool, args) + }, + } + + if _, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1"); !ok { + t.Fatal("the unique fingerprint match should recover the pin") + } + h.StopVirtualEvidence() + + // The requested row must not have moved or gained evidence. + if after := readVirtualEvidenceRowJSON(t, pool, pinnedID); after != pinnedBefore { + t.Fatalf("requested row changed during recovery:\nbefore=%s\nafter =%s", pinnedBefore, after) + } + + // The existing owner keeps its path and gains the probed inventory/stamp. + var ownerPathAfter, ownerResolution string + var ownerStampedAt *time.Time + if err := pool.QueryRow(ctx, + `SELECT file_path, resolution, probe_updated_at FROM media_files WHERE id = $1`, ownerID2). + Scan(&ownerPathAfter, &ownerResolution, &ownerStampedAt); err != nil { + t.Fatalf("read owner row: %v", err) + } + if ownerPathAfter != ownerPath { + t.Fatalf("owner file_path = %q, want it unchanged at %q", ownerPathAfter, ownerPath) + } + if ownerStampedAt == nil { + t.Fatal("owner row was not stamped with the probed evidence") + } + if len(ownerResolution) == 0 { + t.Fatal("owner row lost its resolution evidence") + } +} diff --git a/internal/api/handlers/playback_virtual_stale_pin_test.go b/internal/api/handlers/playback_virtual_stale_pin_test.go index d861d2b3ae..04d4bdc1fe 100644 --- a/internal/api/handlers/playback_virtual_stale_pin_test.go +++ b/internal/api/handlers/playback_virtual_stale_pin_test.go @@ -5,6 +5,7 @@ import ( "errors" "net/http" "net/http/httptest" + "sync" "sync/atomic" "testing" "time" @@ -68,6 +69,19 @@ type stalePinRecoveryHandlerOpts struct { probeDuration int probeErr error noProber bool + // resolverURI, when set, is returned verbatim as the resolved URI (and the + // CandidateID is derived from it) instead of echoing the requested URI. It + // lets a test return a provider-neutral URI or a cross-content URI. + resolverURI string + // resolverCandidateID, when set, overrides the CandidateID the resolver + // returns, so a test can return a CandidateID inconsistent with the URI. + resolverCandidateID string + // resolverOwnerID, when set, is returned as the resolver's OwnerID. + resolverOwnerID int + // pathOwnerRow, when set, is returned by the handler's exact-path lookup, so + // the recovery can persist replacement evidence against an existing owner. + pathOwnerRow *models.MediaFile + pathOwnerResolves bool } type stalePinRecoveryFixture struct { @@ -75,10 +89,22 @@ type stalePinRecoveryFixture struct { resolveCalls *atomic.Int32 listCalls *atomic.Int32 saveCalls *atomic.Int32 + // saved records every write admitted by the evidence saver; savedArgsMu + // guards it for the interleaving test, which drives two recoveries from + // separate goroutines. + savedMu sync.Mutex + savedArgs []models.VirtualFilePersistArgs +} + +func (f *stalePinRecoveryFixture) saved() []models.VirtualFilePersistArgs { + f.savedMu.Lock() + defer f.savedMu.Unlock() + return append([]models.VirtualFilePersistArgs(nil), f.savedArgs...) } func newStalePinRecoveryFixture(pinnedURI string, opts stalePinRecoveryHandlerOpts) *stalePinRecoveryFixture { var resolveCalls, listCalls, saveCalls atomic.Int32 + fixture := &stalePinRecoveryFixture{} h := &PlaybackHandler{ VirtualPlaybackResolver: VirtualPlaybackResolverFunc(func(context.Context, string, int, string, int) (string, error) { return "", errors.New("simple resolver must not be used when the detailed resolver is set") @@ -96,14 +122,43 @@ func newStalePinRecoveryFixture(pinnedURI string, opts stalePinRecoveryHandlerOp if opts.substituteURI != "" { resolvedURI = opts.substituteURI } - return ResolvedVirtualMedia{URL: "http://127.0.0.1:8080/stream?uri=" + resolvedURI, URI: resolvedURI, CandidateID: virtualResultCandidateID(resolvedURI)}, nil + if opts.resolverURI != "" { + resolvedURI = opts.resolverURI + } + candidateID := virtualResultCandidateID(resolvedURI) + if opts.resolverCandidateID != "" { + candidateID = opts.resolverCandidateID + } + resolved := ResolvedVirtualMedia{URL: "http://127.0.0.1:8080/stream?uri=" + resolvedURI, URI: resolvedURI, CandidateID: candidateID} + if opts.resolverOwnerID != 0 { + resolved.OwnerID = opts.resolverOwnerID + } + return resolved, nil }), - VirtualFileSaver: func(context.Context, models.VirtualFilePersistArgs) (int64, error) { + VirtualFileSaver: func(_ context.Context, args models.VirtualFilePersistArgs) (int64, error) { saveCalls.Add(1) + fixture.savedMu.Lock() + fixture.savedArgs = append(fixture.savedArgs, args) + fixture.savedMu.Unlock() return 1, nil }, } - fixture := &stalePinRecoveryFixture{handler: h, resolveCalls: &resolveCalls, listCalls: &listCalls, saveCalls: &saveCalls} + if opts.pathOwnerRow != nil || opts.pathOwnerResolves { + owner := opts.pathOwnerRow + h.fileResolver = stalePinPathResolver{ + getByID: func(int) (*models.MediaFile, error) { return owner, nil }, + getByPath: func(string) (*models.MediaFile, error) { + if opts.pathOwnerResolves { + return owner, nil + } + return nil, ErrVirtualCandidateNotFound + }, + } + } + fixture.handler = h + fixture.resolveCalls = &resolveCalls + fixture.listCalls = &listCalls + fixture.saveCalls = &saveCalls if !opts.noProber { h.VirtualPlaybackSourceProber = func(_ context.Context, _ string, f *models.MediaFile) (*models.MediaFile, error) { if opts.probeErr != nil { @@ -118,6 +173,22 @@ func newStalePinRecoveryFixture(pinnedURI string, opts stalePinRecoveryHandlerOp return fixture } +// stalePinPathResolver is the minimal FilePathResolver the recovery needs: the +// exact-path lookup behind the owner check. Only GetByID and GetByPath are +// reached by the recovery; both are supplied per test. +type stalePinPathResolver struct { + getByID func(int) (*models.MediaFile, error) + getByPath func(string) (*models.MediaFile, error) +} + +func (r stalePinPathResolver) GetByID(_ context.Context, id int) (*models.MediaFile, error) { + return r.getByID(id) +} + +func (r stalePinPathResolver) GetByPath(_ context.Context, path string) (*models.MediaFile, error) { + return r.getByPath(path) +} + func stalePinRecoveryRequest() *http.Request { // Declare the cold-start intent the start path threads: the recovery refuses // a session-bound resolve, and the conservative default is bound, so a unit @@ -635,9 +706,9 @@ func TestStalePinRecoveryClearsOldCandidateState(t *testing.T) { row.ExternalSubtitles = []models.ExternalSubtitle{{Path: "/tmp/old.srt"}} live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} - match, matchedStream, found := virtualStalePinBestStreamMatch(row, live) - if !found { - t.Fatal("the fingerprint match was not found") + match, matchedStream, matched := virtualStalePinMatchStream(row, live) + if matched != 1 { + t.Fatalf("the fingerprint match was not found uniquely (matched=%d)", matched) } if match.FilePath != neutral+"?result=NEW" { t.Fatalf("matched path = %q, want the live candidate", match.FilePath) @@ -686,3 +757,284 @@ func TestStalePinRecoveryRefusesSessionBound(t *testing.T) { t.Fatalf("listings = %d, want 0 (the recovery must not even list)", got) } } + +// TestStalePinRecoveryRefusesAmbiguousFingerprint is the fix-boundary test for +// ambiguous fingerprints: two same-title releases share codec, size and +// runtime, so both fingerprint-match the row. The recovery must REFUSE rather +// than rank-pick the first, whatever the listing order. Both orders are checked +// so "first match wins" cannot pass by accident. +func TestStalePinRecoveryRefusesAmbiguousFingerprint(t *testing.T) { + const neutral = "virtual://movie/tt-stale-ambiguous" + row := stalePinRow(19, "movie-stale-ambiguous", neutral, "STALE") + + first := stalePinLiveStream(neutral, "AAA", row.FileSize, "h264") + second := stalePinLiveStream(neutral, "BBB", row.FileSize, "h264") + if first.URI == second.URI { + t.Fatal("fixture candidates must be distinct identities") + } + + for _, tc := range []struct { + name string + live []VirtualPlaybackStream + }{ + {"listed AAA then BBB", []VirtualPlaybackStream{first, second}}, + {"listed BBB then AAA", []VirtualPlaybackStream{second, first}}, + } { + t.Run(tc.name, func(t *testing.T) { + fx := newStalePinRecoveryFixture(neutral+"?result=STALE", stalePinRecoveryHandlerOpts{ + live: tc.live, + probeDuration: row.Duration, + }) + _, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") + if ok { + t.Fatal("an ambiguous fingerprint must not recover: rank-picking could adopt a different release") + } + if got := fx.resolveCalls.Load(); got != 0 { + t.Fatalf("resolves = %d, want 0 on an ambiguous match", got) + } + if got := fx.saveCalls.Load(); got != 0 { + t.Fatalf("evidence writes = %d, want 0 on an ambiguous match", got) + } + if got := fx.handler.peekVirtualSticky(fx.stickyKey(row)); got != "" { + t.Fatalf("sticky pin = %q, want none on an ambiguous match", got) + } + if got := virtualResultCandidateID(row.FilePath); got != "STALE" { + t.Fatalf("catalog row pin = %q, want the durable STALE unchanged", got) + } + }) + } +} + +// TestStalePinRecoveryVetoesResolutionConflict proves a known resolution +// conflict vetoes a candidate that the size+codec tiers alone would have let +// through, so an edition at a different resolution cannot be adopted and does +// not make the remaining match ambiguous. +func TestStalePinRecoveryVetoesResolutionConflict(t *testing.T) { + const neutral = "virtual://movie/tt-stale-resolution" + row := stalePinRow(20, "movie-stale-resolution", neutral, "STALE") + row.Resolution = "1080p" + + hd := stalePinLiveStream(neutral, "HD", row.FileSize, "h264") + hd.Resolution = "1080p" + sd := stalePinLiveStream(neutral, "SD", row.FileSize, "h264") + sd.Resolution = "720p" + + for _, tc := range []struct { + name string + live []VirtualPlaybackStream + }{ + {"HD then SD", []VirtualPlaybackStream{hd, sd}}, + {"SD then HD", []VirtualPlaybackStream{sd, hd}}, + } { + t.Run(tc.name, func(t *testing.T) { + match, stream, matched := virtualStalePinMatchStream(row, tc.live) + if matched != 1 { + t.Fatalf("matched=%d, want the single resolution-compatible candidate", matched) + } + if stream.URI != hd.URI || match.FilePath != hd.URI { + t.Fatalf("matched %q, want the 1080p candidate %q", stream.URI, hd.URI) + } + }) + } +} + +// TestStalePinRecoveryVetoesAudioLanguageConflict proves a known +// audio-language conflict vetoes a same-size same-codec candidate that is a +// different language variant of the title. +func TestStalePinRecoveryVetoesAudioLanguageConflict(t *testing.T) { + const neutral = "virtual://movie/tt-stale-language" + row := stalePinRow(21, "movie-stale-language", neutral, "STALE") + row.AudioTracks = []models.AudioTrack{{Codec: "aac", Channels: 2, Language: "eng"}} + + french := stalePinLiveStream(neutral, "FRA", row.FileSize, "h264") + french.AudioLanguages = []string{"fra"} + english := stalePinLiveStream(neutral, "ENG", row.FileSize, "h264") + english.AudioLanguages = []string{"ENG"} + + for _, tc := range []struct { + name string + live []VirtualPlaybackStream + want string + }{ + {"french then english", []VirtualPlaybackStream{french, english}, english.URI}, + {"english then french", []VirtualPlaybackStream{english, french}, english.URI}, + } { + t.Run(tc.name, func(t *testing.T) { + match, stream, matched := virtualStalePinMatchStream(row, tc.live) + if matched != 1 { + t.Fatalf("matched=%d, want only the language-compatible candidate", matched) + } + if stream.URI != tc.want || match.FilePath != tc.want { + t.Fatalf("matched %q, want %q", stream.URI, tc.want) + } + }) + } + + // A listing whose only candidate is a conflicting language is ambiguous in + // the sense that matters: it is not the same release, so no match at all. + _, _, matched := virtualStalePinMatchStream(row, []VirtualPlaybackStream{french}) + if matched != 0 { + t.Fatalf("a lone conflicting-language candidate matched=%d, want no match", matched) + } +} + +// TestStalePinRecoveryNewerSelectionFinishesFirst is the generation-fence proof +// for the early allocation. An earlier-started recovery blocks in its listing; +// a newer selection pins and caches first; the recovery then finishes. Because +// the recovery captured its generation before listing, its publication carries +// an older generation and must NOT overwrite the newer selection. Allocating the +// generation at publication instead would let the recovery's later finish win. +func TestStalePinRecoveryNewerSelectionFinishesFirst(t *testing.T) { + const neutral = "virtual://movie/tt-stale-interleave" + row := stalePinRow(22, "movie-stale-interleave", neutral, "STALE") + newerURI := neutral + "?result=NEWER" + + started := make(chan struct{}) + releaseList := make(chan struct{}) + var listOnce sync.Once + h := &PlaybackHandler{ + VirtualPlaybackResolver: VirtualPlaybackResolverFunc(func(context.Context, string, int, string, int) (string, error) { + return "", errors.New("simple resolver must not be used") + }), + VirtualPlaybackStreamLister: VirtualPlaybackStreamListerFunc(func(context.Context, string, int, string, int) ([]VirtualPlaybackStream, error) { + listOnce.Do(func() { close(started) }) + <-releaseList + return []VirtualPlaybackStream{stalePinLiveStream(neutral, "RECOVERED", row.FileSize, "h264")}, nil + }), + VirtualMediaDetailedResolver: VirtualMediaDetailedResolverFunc(func(_ context.Context, uri string, _ int, _ int, _ string, _ bool, _ []string, _ string) (ResolvedVirtualMedia, error) { + return ResolvedVirtualMedia{URL: "http://127.0.0.1:8080/stream?uri=" + uri, URI: uri, CandidateID: virtualResultCandidateID(uri)}, nil + }), + VirtualPlaybackSourceProber: func(_ context.Context, _ string, f *models.MediaFile) (*models.MediaFile, error) { + probed := stalePinProbedTrack(row.Duration) + probed.ID = f.ID + probed.FilePath = f.FilePath + return probed, nil + }, + } + + stickyKey := bestResultCacheKey(row.ContentID, neutral, row.VirtualOwnerInstallationID) + done := make(chan bool, 1) + go func() { + _, ok := h.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") + done <- ok + }() + + // Wait until the recovery has captured its generation and entered the + // listing, then let a newer selection finish first. + <-started + h.pinVirtualStickyAt(stickyKey, newerURI, nextVirtualCacheGeneration()) + close(releaseList) + + if ok := <-done; !ok { + t.Fatal("the recovery should still resolve its unique candidate") + } + if got := h.peekVirtualSticky(stickyKey); got != newerURI { + t.Fatalf("sticky pin = %q, want the newer selection %q: the late-finishing recovery must not overwrite it", got, newerURI) + } +} + +// TestStalePinRecoveryRejectsNeutralAndCrossContentSubstitutions proves the +// exact-candidate check requires a concrete matched candidate: a neutral URI +// (no result= pick) and a same-id candidate under a different content both +// reject before any publication. +func TestStalePinRecoveryRejectsNeutralAndCrossContentSubstitutions(t *testing.T) { + const ( + neutral = "virtual://movie/tt-stale-exact" + foreign = "virtual://movie/tt-stale-foreign" + ) + row := stalePinRow(23, "movie-stale-exact", neutral, "STALE") + + cases := []struct { + name string + resolverURI string + candidateID string + }{ + {"neutral URI", neutral, ""}, + {"same result id under a different content", foreign + "?result=NEW", ""}, + {"candidate id inconsistent with the URI", neutral + "?result=NEW", "DIFFERENT"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + fx := newStalePinRecoveryFixture(neutral+"?result=STALE", stalePinRecoveryHandlerOpts{ + live: []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")}, + probeDuration: row.Duration, + resolverURI: tc.resolverURI, + resolverCandidateID: tc.candidateID, + }) + _, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") + if ok { + t.Fatal("a substituted candidate identity must not recover the pin") + } + if got := fx.resolveCalls.Load(); got != 1 { + t.Fatalf("resolves = %d, want exactly 1", got) + } + if got := fx.saveCalls.Load(); got != 0 { + t.Fatalf("evidence writes = %d, want 0 on a rejected substitution", got) + } + if got := fx.handler.peekVirtualSticky(fx.stickyKey(row)); got != "" { + t.Fatalf("sticky pin = %q, want none on a rejected substitution", got) + } + }) + } +} + +// TestStalePinRecoveryPersistsOnlyAgainstExistingOwner proves the recovery is +// transient: with no row already owning the matched concrete path, it persists +// no evidence at all (so it can never adopt the path onto the requested row); +// with an existing alternate owner it writes a metadata-only update against that +// owner, never the requested row. +func TestStalePinRecoveryPersistsOnlyAgainstExistingOwner(t *testing.T) { + const neutral = "virtual://movie/tt-stale-owner" + row := stalePinRow(24, "movie-stale-owner", neutral, "STALE") + live := []VirtualPlaybackStream{stalePinLiveStream(neutral, "NEW", row.FileSize, "h264")} + + t.Run("no existing owner persists nothing", func(t *testing.T) { + fx := newStalePinRecoveryFixture(neutral+"?result=STALE", stalePinRecoveryHandlerOpts{ + live: live, + probeDuration: row.Duration, + }) + _, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") + if !ok { + t.Fatal("the unique candidate should recover") + } + if got := fx.saveCalls.Load(); got != 0 { + t.Fatalf("evidence writes = %d, want 0: the recovery must not adopt the candidate onto the requested row", got) + } + if got := virtualResultCandidateID(row.FilePath); got != "STALE" { + t.Fatalf("catalog row pin = %q, want the durable STALE unchanged", got) + } + }) + + t.Run("existing alternate owner gets a metadata-only write", func(t *testing.T) { + ownerRow := &models.MediaFile{ + ID: 99, + ContentID: row.ContentID, + FilePath: neutral + "?result=NEW", + VirtualOwnerInstallationID: row.VirtualOwnerInstallationID, + MediaFolderID: row.MediaFolderID, + ProbeSource: "virtual", + } + fx := newStalePinRecoveryFixture(neutral+"?result=STALE", stalePinRecoveryHandlerOpts{ + live: live, + probeDuration: row.Duration, + pathOwnerRow: ownerRow, + pathOwnerResolves: true, + }) + _, ok := fx.handler.recoverStaleIdentityLessPinV3(stalePinRecoveryRequest(), row, "profile-1") + if !ok { + t.Fatal("the unique candidate should recover") + } + fx.handler.StopVirtualEvidence() + + saved := fx.saved() + if len(saved) != 1 { + t.Fatalf("evidence writes = %d, want exactly 1 against the existing owner", len(saved)) + } + if saved[0].FileID != ownerRow.ID { + t.Fatalf("evidence write targeted file %d, want the owner %d", saved[0].FileID, ownerRow.ID) + } + if saved[0].AdoptPath != "" || saved[0].RequireAdopt { + t.Fatalf("owner write nominated an adoption target: %+v", saved[0]) + } + }) +}