diff --git a/internal/api/handlers/playback.go b/internal/api/handlers/playback.go index 24f995310a..3bdea5150a 100644 --- a/internal/api/handlers/playback.go +++ b/internal/api/handlers/playback.go @@ -218,22 +218,42 @@ func isClientCancellation(ctx context.Context, err error) bool { // logVirtualStreamFailure records the sanitized cause behind a virtual-stream // 502. A provider URL can be embedded in a wrapped *url.Error, so the cause is // passed through logredact; only the provider-neutral virtual URI, the file, -// the session, and the owner installation are logged directly. +// the session, the pinned and delivered candidate identities, and the owner +// installation are logged directly. +// +// deliveredPath is the virtual:// path the transport actually served (the +// ?result= identity the bytes came from), or "" when the failure happened +// before any candidate was served — a resolve failure, an unparseable relay +// target, a lapsed-registration retry that never re-resolved. It must be +// passed explicitly because the catalog row (file.FilePath) still names the +// pinned candidate after a heal or rotation rebinds the served path: reading +// the identity off the row would blame the pin for a sibling's failure. +// pinned_candidate_id is the ?result= on the row; delivered_candidate_id is +// the ?result= on deliveredPath, empty when nothing was served. The two agree +// on the common path and disagree exactly when attribution matters. // // A client cancellation is not a transport failure: the upstream fetch runs on // the request context, so the viewer navigating away (or hls.js giving up) // already canceled it. That case is a debug line with an explicit reason so it // cannot be mistaken for a provider outage; genuine timeouts and provider // errors stay at WARN. -func logVirtualStreamFailure(ctx context.Context, sessionID string, file *models.MediaFile, err error) { +func logVirtualStreamFailure(ctx context.Context, sessionID string, file *models.MediaFile, err error, deliveredPath ...string) { if err == nil || file == nil { return } + pinnedID := virtualCandidateID(file) + deliveredID := "" + if len(deliveredPath) > 0 { + deliveredID = virtualResultCandidateID(deliveredPath[0]) + } if isClientCancellation(ctx, err) { slog.DebugContext(ctx, "virtual stream transport canceled by client", "component", "api", "session", sessionID, + "playback_session_id", sessionID, "file_id", file.ID, + "pinned_candidate_id", pinnedID, + "delivered_candidate_id", deliveredID, "reason", "client_canceled", ) return @@ -241,7 +261,10 @@ func logVirtualStreamFailure(ctx context.Context, sessionID string, file *models slog.WarnContext(ctx, "virtual stream transport failed", "component", "api", "session", sessionID, + "playback_session_id", sessionID, "file_id", file.ID, + "pinned_candidate_id", pinnedID, + "delivered_candidate_id", deliveredID, "owner_installation_id", file.VirtualOwnerInstallationID, "virtual_uri", file.FilePath, "error", logredact.SanitizeURLError(err), diff --git a/internal/api/handlers/playback_virtual.go b/internal/api/handlers/playback_virtual.go index 4c124fe051..baabf12820 100644 --- a/internal/api/handlers/playback_virtual.go +++ b/internal/api/handlers/playback_virtual.go @@ -29,6 +29,7 @@ import ( "github.com/Silo-Server/silo-server/internal/scanner" "github.com/Silo-Server/silo-server/internal/virtuallibrary" "github.com/Silo-Server/silo-server/internal/virtuallibrary/resolver" + chimw "github.com/go-chi/chi/v5/middleware" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" "golang.org/x/text/language" @@ -1497,11 +1498,21 @@ func (t *virtualResolveTrace) finishBudget() { // runs first in log), so the reported elapsed and the exceeded verdict can // never disagree near the boundary. func (t *virtualResolveTrace) fields() []any { + return t.fieldsWithContext(context.Background()) +} + +// fieldsWithContext is fields plus the edge request id, read from the start +// request's context. It joins this line to the start that owns it (protocol +// v3 start timing) and to the provider listing it may have triggered +// (provider candidates fetched). Empty stays empty rather than a fabricated +// join key. Split out so tests can call fields() without a request context. +func (t *virtualResolveTrace) fieldsWithContext(ctx context.Context) []any { elapsedMS := t.budgetElapsed.Milliseconds() if !t.budgetRan { elapsedMS = time.Since(t.started).Milliseconds() } attrs := []any{ + requestIDLogKeyV3, chimw.GetReqID(ctx), "elapsed_ms", elapsedMS, "total_ms", t.totalMS(), //nolint:goconst // log attribute key/value, kept inline for readability. "candidates", t.candidates, //nolint:goconst // log attribute key/value, kept inline for readability. @@ -1537,7 +1548,7 @@ func (t *virtualResolveTrace) log(ctx context.Context, file *models.MediaFile) { } t.finishBudget() attrs := []any{logComponentKey, "api", "content_id", file.ContentID} //nolint:goconst // log attribute key/value, kept inline for readability. - attrs = append(attrs, t.fields()...) + attrs = append(attrs, t.fieldsWithContext(ctx)...) slog.InfoContext(ctx, "virtual resolve timing", attrs...) } diff --git a/internal/api/handlers/playback_virtual_trace_test.go b/internal/api/handlers/playback_virtual_trace_test.go index e0e29944cc..9bee39f3ae 100644 --- a/internal/api/handlers/playback_virtual_trace_test.go +++ b/internal/api/handlers/playback_virtual_trace_test.go @@ -12,6 +12,7 @@ import ( "time" "github.com/Silo-Server/silo-server/internal/models" + chimw "github.com/go-chi/chi/v5/middleware" ) // virtualTraceFieldsToMap converts the stable []any key/value shape returned by @@ -131,11 +132,18 @@ func TestResolveVirtualTimingReportsRanStages(t *testing.T) { entry := captureVirtualResolveTiming(t, func() { req := httptest.NewRequest(http.MethodPost, "/api/v1/playback/start", nil) + // The trace must carry the start's own edge id so it joins to its + // start timing and its provider fetch without timestamp forensics. + req = req.WithContext(context.WithValue(req.Context(), chimw.RequestIDKey, "trace-start-req-1")) if _, err := h.resolveVirtualPlaybackSource(req, file, "profile-1", true, nil, "", "", 0, false); err != nil { t.Fatalf("resolveVirtualPlaybackSource error: %v", err) } }) + if entry["request_id"] != "trace-start-req-1" { + t.Fatalf("request_id = %#v, want the start request's edge id", entry["request_id"]) + } + for _, stage := range []string{"list", "resolve"} { if entry[stage+"_ran"] != true { t.Fatalf("%s_ran = %#v, want true", stage, entry[stage+"_ran"]) diff --git a/internal/api/handlers/stream.go b/internal/api/handlers/stream.go index 6b3f78de5d..8bd6465d31 100644 --- a/internal/api/handlers/stream.go +++ b/internal/api/handlers/stream.go @@ -1005,6 +1005,109 @@ func resolvedVirtualCandidatePath(resolved ResolvedVirtualMedia) string { return uri } +// Disposition vocabulary for the direct-play proxy verdict. Each names one +// first-attempt outcome; the verdict log records exactly one of them. Named +// constants (not inline literals) so the closed set reads in one place and +// stays out of the repeated-literal lint. +const ( + virtualProxyDispositionClientCanceled = "client_canceled" + virtualProxyDispositionRelayNotFound = "relay_not_found" + virtualProxyDispositionRelayTemporary = "relay_temporary" + virtualProxyDispositionCommitted = "committed" + virtualProxyDispositionFailedOpen = "failed_open" +) + +// Attribute keys for the proxy verdict line. The verdict is the single site +// that emits this shape, so the keys live with it rather than as package +// shared constants; logVirtualStreamFailure in playback.go keeps its own +// inline keys (two occurrences stay below the repeated-literal lint). +const ( + virtualVerdictLogKeySession = "session" + virtualVerdictLogKeyPlaybackSession = "playback_session_id" +) + +// virtualProxyDisposition names how a failed direct-play proxy attempt ended +// before any heal or rotation runs: the first-attempt verdict the verdict log +// records. Client cancellation is the viewer's own doing; relay_not_found is +// the lapsed registration the same-release retry covers; relay_temporary is +// the upstream 5xx the rotation treats as availability, not a dead release; +// committed means the response had committed headers when the error arrived +// (headers written, not necessarily media bytes — mid-stream copy failures +// panic with ErrAbortHandler instead of reaching this verdict); failed_open +// is everything else. The disposition is a log value only and never steers +// recovery: the branches below keep their own conditions. +// +// The returned values are the verdict's contract vocabulary, pinned by +// TestVirtualProxyDisposition. +func virtualProxyDisposition(ctx context.Context, proxyErr error, statusCode int, relayNotFound, relayTemporary bool) string { + if isClientCancellation(ctx, proxyErr) { + return virtualProxyDispositionClientCanceled + } + if relayNotFound { + return virtualProxyDispositionRelayNotFound + } + if relayTemporary { + return virtualProxyDispositionRelayTemporary + } + if statusCode != 0 { + return virtualProxyDispositionCommitted + } + return virtualProxyDispositionFailedOpen +} + +// logVirtualProxyVerdict records the first-attempt serve verdict of a failed +// direct-play proxy: the session, the pinned candidate, the candidate the +// attempt served, and the disposition above. The delivered candidate id is the +// ?result= identity on the path the transport actually served; at verdict time +// no heal or rotation has run yet, so it agrees with the pin on the common +// path and is empty when the attempt failed before serving one. Provider URLs +// never appear here: only provider-neutral virtual URIs. Client cancellation +// logs at debug; a genuine failure keeps its WARN so it stays visible in the +// operational log. The verdict is first-attempt classification, not the final +// recovery outcome: a later heal or rotation can still serve the request. +// +// The attribute keys are the verdict's join contract (session, +// playback_session_id, candidate ids, disposition), pinned by +// TestLogVirtualProxyVerdictCarriesJoinKeys. Provider URLs never appear +// here: only provider-neutral virtual URIs. Client cancellation logs at +// debug; a genuine failure keeps its WARN so it stays visible in the +// operational log. The verdict is first-attempt classification, not the final +// recovery outcome: a later heal or rotation can still serve the request. +func logVirtualProxyVerdict(ctx context.Context, sessionID string, file *models.MediaFile, deliveredPath, disposition string, proxyErr error) { + pinnedID := virtualCandidateID(file) + deliveredID := virtualResultCandidateID(deliveredPath) + attrs := []any{ + logComponentKey, "api", //nolint:goconst // log attribute key/value, kept inline for readability. + virtualVerdictLogKeySession, sessionID, + virtualVerdictLogKeyPlaybackSession, sessionID, + "pinned_candidate_id", pinnedID, + "delivered_candidate_id", deliveredID, + "disposition", disposition, + } + if file != nil { + attrs = append(attrs, "file_id", file.ID) + } + if proxyErr != nil { + attrs = append(attrs, "error", logredact.SanitizeURLError(proxyErr)) + } + if disposition == virtualProxyDispositionClientCanceled { + slog.DebugContext(ctx, "virtual stream proxy verdict", attrs...) + return + } + slog.WarnContext(ctx, "virtual stream proxy verdict", attrs...) +} + +// virtualCandidateID is the ?result= identity bound to a catalog row's +// virtual URI, or "" when the row carries no explicit pick. The log field +// must never be a fabricated join key, so absence stays empty. A nil file +// has no identity either. +func virtualCandidateID(file *models.MediaFile) string { + if file == nil { + return "" + } + return virtualResultCandidateID(file.FilePath) +} + func hasVirtualMediaResolver(h *StreamHandler) bool { return h != nil && (h.VirtualMediaResolver != nil || h.VirtualMediaDetailedResolver != nil || h.VirtualMediaRefreshResolver != nil) } @@ -1417,6 +1520,17 @@ func (h *StreamHandler) HandleStream(w http.ResponseWriter, r *http.Request) { } proxy.ServeHTTP(streamWriter, r) if lastProxyErr != nil { + // Log the first-attempt serve verdict while the recovery state + // is still observable: which candidate the attempt served and + // how that attempt ended. This is first-attempt + // classification, not the final recovery outcome — a later + // heal or rotation can still serve the request. A later 503 + // names only the status; this line names the release and the + // path taken, so the next virtual_stream_unavailable is a + // join, not forensics. Client cancellation stays debug (see + // logVirtualStreamFailure); a genuine failure keeps its WARN. + disposition := virtualProxyDisposition(r.Context(), lastProxyErr, streamWriter.StatusCode(), relayNotFound, relayTemporary) + logVirtualProxyVerdict(r.Context(), sessionID, file, deliveredPath, disposition, lastProxyErr) // A client that navigated away is not a candidate failure: // never stamp the pinned release known-bad or spend a failover // resolve on it. The upstream request already aborted with the @@ -1609,7 +1723,7 @@ func (h *StreamHandler) HandleStream(w http.ResponseWriter, r *http.Request) { if lastProxyErr != nil { h.handleTransportStartFailure(r.Context(), session, file, lastProxyErr) if streamWriter.StatusCode() == 0 { - logVirtualStreamFailure(r.Context(), sessionID, file, lastProxyErr) + logVirtualStreamFailure(r.Context(), sessionID, file, lastProxyErr, deliveredPath) writeErrorCause(streamWriter, http.StatusBadGateway, "virtual_stream_unavailable", "Failed to stream virtual media source", lastProxyErr) } } @@ -2774,6 +2888,7 @@ func (h *StreamHandler) handleTransportStartFailure(ctx context.Context, session slog.WarnContext(ctx, "stream transport startup failed", "component", "api", "session", session.ID, "file_id", session.MediaFileID, + "pinned_candidate_id", virtualCandidateID(file), "error", err, "playback_session_id", session.ID, ) diff --git a/internal/api/handlers/transport_cancellation_log_test.go b/internal/api/handlers/transport_cancellation_log_test.go index 8af86f01ee..ccf35665ba 100644 --- a/internal/api/handlers/transport_cancellation_log_test.go +++ b/internal/api/handlers/transport_cancellation_log_test.go @@ -1,10 +1,13 @@ package handlers import ( + "bytes" "context" + "encoding/json" "errors" "fmt" "log/slog" + "strings" "sync" "testing" @@ -93,6 +96,160 @@ func TestLogVirtualStreamFailureDowngradesClientCancellation(t *testing.T) { } } +// The failure line must join to the serve attempt: it carries the playback +// session id and both the pinned and the delivered candidate ids, so a +// virtual_stream_unavailable names the release it actually served instead of +// only its session. A rotation that rebinds the served path must attribute +// the sibling, not the pin; a resolve failure that never served one reports +// both identities honestly (pin set, delivered empty). +func TestLogVirtualStreamFailureCarriesSessionAndCandidate(t *testing.T) { + capture := func(t *testing.T, file *models.MediaFile, err error, delivered ...string) map[string]any { + t.Helper() + var buf bytes.Buffer + previous := slog.Default() + slog.SetDefault(slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}))) + t.Cleanup(func() { slog.SetDefault(previous) }) + + logVirtualStreamFailure(context.Background(), "session-9", file, err, delivered...) + + for _, line := range strings.Split(strings.TrimSpace(buf.String()), "\n") { + var candidate map[string]any + if err := json.Unmarshal([]byte(line), &candidate); err != nil { + continue + } + if candidate["msg"] == "virtual stream transport failed" { + return candidate + } + } + t.Fatalf("no virtual stream transport failed line captured; log was:\n%s", buf.String()) + return nil + } + + // Common path: served the pin, both identities agree. + file := &models.MediaFile{ID: 7, FilePath: "virtual://movie/tt14538850?result=cand-9"} + entry := capture(t, file, errors.New("relay returned HTTP 502"), "virtual://movie/tt14538850?result=cand-9") + if entry["playback_session_id"] != "session-9" { + t.Fatalf("playback_session_id = %#v, want session-9", entry["playback_session_id"]) + } + if entry["pinned_candidate_id"] != "cand-9" { + t.Fatalf("pinned_candidate_id = %#v, want cand-9", entry["pinned_candidate_id"]) + } + if entry["delivered_candidate_id"] != "cand-9" { + t.Fatalf("delivered_candidate_id = %#v, want cand-9", entry["delivered_candidate_id"]) + } + + // A → B rotation that then fails attributes the sibling, not the pin. + entry = capture(t, file, errors.New("relay returned HTTP 502"), "virtual://movie/tt14538850?result=cand-4") + if entry["pinned_candidate_id"] != "cand-9" { + t.Fatalf("pinned_candidate_id = %#v, want cand-9", entry["pinned_candidate_id"]) + } + if entry["delivered_candidate_id"] != "cand-4" { + t.Fatalf("delivered_candidate_id = %#v, want cand-4 (the served sibling, not the pin)", entry["delivered_candidate_id"]) + } + + // A resolve failure serves nothing: delivered stays empty rather than + // echoing the pin. + entry = capture(t, file, errors.New("no streams available from provider")) + if entry["pinned_candidate_id"] != "cand-9" { + t.Fatalf("pinned_candidate_id = %#v, want cand-9", entry["pinned_candidate_id"]) + } + if delivered, present := entry["delivered_candidate_id"]; !present || delivered != "" { + t.Fatalf("delivered_candidate_id = %#v, want empty (nothing served)", entry["delivered_candidate_id"]) + } +} + +// The proxy verdict names how a failed direct-play proxy ended before any +// heal or rotation runs, so the verdict log records the release and the path +// taken while that state is still observable. Each disposition maps to one +// first-attempt outcome; client cancellation stays debug and never warns. +func TestVirtualProxyDisposition(t *testing.T) { + canceledCtx, cancel := context.WithCancel(context.Background()) + cancel() + cases := []struct { + name string + ctx context.Context + err error + status int + relayNotFound, relayTemporary bool + want string + }{ + {"client canceled wins", canceledCtx, errors.New("boom"), 0, false, false, "client_canceled"}, + {"relay not found", context.Background(), errors.New("relay token not found"), 0, true, false, "relay_not_found"}, + {"relay temporary", context.Background(), errors.New("relay returned HTTP 502"), 0, false, true, "relay_temporary"}, + {"committed headers", context.Background(), errors.New("boom"), 206, false, false, "committed"}, + {"failed open", context.Background(), errors.New("boom"), 0, false, false, "failed_open"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := virtualProxyDisposition(tc.ctx, tc.err, tc.status, tc.relayNotFound, tc.relayTemporary); got != tc.want { + t.Fatalf("virtualProxyDisposition = %q, want %q", got, tc.want) + } + }) + } +} + +// The verdict line joins to the serve attempt: session, pinned and delivered +// candidate ids, and the disposition above. The client only ever sees the +// generic problem envelope; this line is the server-side join key. +func TestLogVirtualProxyVerdictCarriesJoinKeys(t *testing.T) { + var buf bytes.Buffer + previous := slog.Default() + slog.SetDefault(slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}))) + t.Cleanup(func() { slog.SetDefault(previous) }) + + file := &models.MediaFile{ID: 7, FilePath: "virtual://movie/tt14538850?result=cand-9"} + logVirtualProxyVerdict(context.Background(), "session-9", file, "virtual://movie/tt14538850?result=cand-9", "failed_open", errors.New("relay returned HTTP 502")) + + var entry map[string]any + for _, line := range strings.Split(strings.TrimSpace(buf.String()), "\n") { + var candidate map[string]any + if err := json.Unmarshal([]byte(line), &candidate); err != nil { + continue + } + if candidate["msg"] == "virtual stream proxy verdict" { + entry = candidate + break + } + } + if entry == nil { + t.Fatalf("no virtual stream proxy verdict line captured; log was:\n%s", buf.String()) + } + for key, want := range map[string]any{ + "playback_session_id": "session-9", + "pinned_candidate_id": "cand-9", + "delivered_candidate_id": "cand-9", + "disposition": "failed_open", + } { + if entry[key] != want { + t.Fatalf("%s = %#v, want %#v", key, entry[key], want) + } + } + + // Client cancellation downgrades to debug so it cannot be mistaken for a + // provider outage. + buf.Reset() + canceledCtx, cancel := context.WithCancel(context.Background()) + cancel() + logVirtualProxyVerdict(canceledCtx, "session-9", file, "virtual://movie/tt14538850?result=cand-9", "client_canceled", context.Canceled) + found := false + for _, line := range strings.Split(strings.TrimSpace(buf.String()), "\n") { + var candidate map[string]any + if err := json.Unmarshal([]byte(line), &candidate); err != nil { + continue + } + if candidate["msg"] != "virtual stream proxy verdict" { + continue + } + found = true + if candidate["level"] != "DEBUG" { + t.Fatalf("canceled verdict level = %#v, want DEBUG", candidate["level"]) + } + } + if !found { + t.Fatalf("no canceled verdict line captured; log was:\n%s", buf.String()) + } +} + func TestHandleTransportStartFailureDowngradesClientCancellation(t *testing.T) { capture := &levelRecordingHandler{} previous := slog.Default()