diff --git a/docs/design/schemas/playback-v3/v3/fixtures/valid/capability_response.json b/docs/design/schemas/playback-v3/v3/fixtures/valid/capability_response.json index c66a8100..ca2e7674 100644 --- a/docs/design/schemas/playback-v3/v3/fixtures/valid/capability_response.json +++ b/docs/design/schemas/playback-v3/v3/fixtures/valid/capability_response.json @@ -10,6 +10,7 @@ "playback_route_diagnostics", "device_quirks_v1", "seek_reanchor_v1", + "output_change_v1", "direct_stream_resume_v1", "plan_source_duration_v1" ], diff --git a/docs/design/schemas/playback-v3/v3/fixtures/valid/decision_response.json b/docs/design/schemas/playback-v3/v3/fixtures/valid/decision_response.json index d7345acf..2cca07df 100644 --- a/docs/design/schemas/playback-v3/v3/fixtures/valid/decision_response.json +++ b/docs/design/schemas/playback-v3/v3/fixtures/valid/decision_response.json @@ -7,6 +7,7 @@ "playback_route_diagnostics", "device_quirks_v1", "seek_reanchor_v1", + "output_change_v1", "direct_stream_resume_v1", "plan_source_duration_v1" ], diff --git a/internal/api/handlers/playback.go b/internal/api/handlers/playback.go index b5d16ff1..424fd87b 100644 --- a/internal/api/handlers/playback.go +++ b/internal/api/handlers/playback.go @@ -49,6 +49,7 @@ type SessionManagerInterface interface { TouchActivity(sessionID string) error BeginTransport(sessionID string) error EndTransport(sessionID string) error + SetRemoteTransport(sessionID string, remote bool) error SetEffectiveMediaFileID(sessionID string, fileID int) error SetTranscodeNodeURL(sessionID, url string) error SetTranscodeRoute(sessionID string, route playback.TranscodeRoute) error @@ -396,12 +397,20 @@ const streamTokenParam = "st" // reconstruction recipe. Returns "" when no signing secret is configured // (reconstruct effectively disabled, e.g. in tests). func (h *PlaybackHandler) signSessionToken(card playback.RecipeCard) string { + return h.signStreamClaims(card.ToClaims()) +} + +// signStreamClaims mints a stream token from claims that are already assembled. +// Callers serving a session from another node use it to add the claims a +// RecipeCard does not model (the file's Dolby Vision profile, audio-only flag), +// which a remote executor cannot look up for itself. +func (h *PlaybackHandler) signStreamClaims(claims streamtoken.Claims) string { if h.JWTSecret == "" { return "" } - token, err := streamtoken.Sign(card.ToClaims(), h.JWTSecret, playback.MaxTokenTTL) + token, err := streamtoken.Sign(claims, h.JWTSecret, playback.MaxTokenTTL) if err != nil { - slog.Warn("sign stream token failed", "error", err, "session", card.SessionID, "playback_session_id", card.SessionID) + slog.Warn("sign stream token failed", "error", err, "session", claims.SessionID, "playback_session_id", claims.SessionID) return "" } return token diff --git a/internal/api/handlers/playback_test.go b/internal/api/handlers/playback_test.go index bcf53dc0..0a431178 100644 --- a/internal/api/handlers/playback_test.go +++ b/internal/api/handlers/playback_test.go @@ -181,6 +181,8 @@ func (failingSessionManager) BeginTransport(string) error { return nil } func (failingSessionManager) EndTransport(string) error { return nil } +func (failingSessionManager) SetRemoteTransport(string, bool) error { return nil } + func (failingSessionManager) SetEffectiveMediaFileID(string, int) error { return nil } func (failingSessionManager) SetTranscodeNodeURL(string, string) error { return nil } diff --git a/internal/api/handlers/playback_v3.go b/internal/api/handlers/playback_v3.go index 301782ca..207ce32a 100644 --- a/internal/api/handlers/playback_v3.go +++ b/internal/api/handlers/playback_v3.go @@ -198,6 +198,12 @@ type transcodeNodeEnumeratorV3 interface { TranscodeNodeURLs() []string } +// proxyNodeEnumeratorV3 is the proxy-pool counterpart, letting identity +// planning narrow selection to proxies that can execute the plan's recipe. +type proxyNodeEnumeratorV3 interface { + ProxyNodeURLs() []string +} + // hlsPlanningRegistryV3 returns the registry HLS deliveries plan against: the // local probe plus every pooled transcode node's advertised transformations. // Only availability of locally-defined specs widens (name and recipe version @@ -672,7 +678,7 @@ func (h *PlaybackHandler) prepareTransportV3(r *http.Request, session *playback. return preparedTransportV3{}, timelineErr } if result.Plan.Delivery != playback.DeliveryTranscodeHLSV3 && result.Plan.Delivery != playback.DeliveryRemuxHLSV3 { - return h.prepareIdentityTransportV3(session, result, timeline), nil + return h.prepareIdentityTransportV3(r, session, file, result, timeline) } if h.NodePlanner != nil { plan := h.planNodeSessionV3(r.Context(), session, result) @@ -777,7 +783,7 @@ func planRequiresServerTransformationsV3(plan *playback.PlanV3) bool { return false } -func (h *PlaybackHandler) prepareIdentityTransportV3(session *playback.Session, result playback.PlannerResultV3, timeline preparedTimelineV3) preparedTransportV3 { +func (h *PlaybackHandler) prepareIdentityTransportV3(r *http.Request, session *playback.Session, file *models.MediaFile, result playback.PlannerResultV3, timeline preparedTimelineV3) (preparedTransportV3, *transportErrorV3) { routeSession := *session routeSession.PlayMethod = result.PlayMethod routeSession.BasePlayMethod = result.PlayMethod @@ -788,11 +794,31 @@ func (h *PlaybackHandler) prepareIdentityTransportV3(session *playback.Session, routeSession.TargetAudioChannels = result.TargetAudioChannels routeSession.TargetAudioBitrateKbps = result.TargetAudioBitrateKbps routeSession.RemuxDVMode = remuxDVModeForPlanV3(result.Plan) + + proxyNode, proxyErr := h.planIdentityProxyV3(r, session.ID, result) + if proxyErr != nil { + return preparedTransportV3{}, proxyErr + } + streamURL, servedByProxy := h.identityStreamURLV3(&routeSession, file, proxyNode) + releaseProxyReservation := func() { + if releaser, ok := h.NodePlanner.(sessionReservationReleaserV3); ok { + releaser.ReleaseSession(session.ID) + } + } + if proxyNode != nil && !servedByProxy { + // A planned proxy that could not be addressed (no signable token, no + // file record) falls back to the local path, so its reservation must be + // dropped now rather than pinning that node's budget until it ages out. + releaseProxyReservation() + if err := h.refuseLocalIdentityWorkV3(r, result); err != nil { + return preparedTransportV3{}, err + } + } + previousNodeURL := session.TranscodeNodeURL previousTransportID := remoteTransportID(session) unlock := h.tm.LockSessionLifecycle(session.ID) committed := false - streamURL := h.playbackStreamURL(&routeSession) if result.Plan != nil && result.Plan.Delivery == playback.DeliveryRemuxProgressiveV3 { if seek := timeline.seekSeconds; seek > 0 { streamURL = appendPlaybackQueryV3(streamURL, "seek", strconv.FormatFloat(seek, 'f', -1, 64)) @@ -809,6 +835,7 @@ func (h *PlaybackHandler) prepareIdentityTransportV3(session *playback.Session, if previousNodeURL != "" { h.tm.StopRemoteTranscode(previousTransportID, previousNodeURL) } + h.applyRemoteTransportMarkV3(r.Context(), session.ID, servedByProxy) unlock() }, rollback: func() { @@ -816,9 +843,158 @@ func (h *PlaybackHandler) prepareIdentityTransportV3(session *playback.Session, return } committed = true + // The session never reached the client, so a proxy admitted for it + // must not keep consuming that node's job/bandwidth budget until the + // reservation ages out. + if servedByProxy { + releaseProxyReservation() + } unlock() }, + }, nil +} + +// planIdentityProxyV3 selects the proxy node that will serve a direct-play or +// progressive-remux session. These deliveries need no transcode node — the +// bytes are either the source file or a single remux pipe — so the planner is +// asked for a proxy alone, exactly as the Jellyfin-compat transport does. +// +// A nil node (no planner, no signing secret, or no eligible proxy) means the +// API server serves the stream itself, which is both the single-node case and +// the correct degradation when every proxy is unhealthy or at capacity. The one +// exception is a remux that must run ffmpeg: that is transcode work, so it +// honors the same local-fallback gate as the HLS routes rather than quietly +// spawning an encoder on an API-only node. +func (h *PlaybackHandler) planIdentityProxyV3(r *http.Request, sessionID string, result playback.PlannerResultV3) (*nodepool.Node, *transportErrorV3) { + if h.NodePlanner == nil || h.JWTSecret == "" { + return nil, h.refuseLocalIdentityWorkV3(r, result) } + // Reserve against the session id the rest of the transport uses, so a + // re-plan replaces its own reservation instead of double-counting, and the + // rollback path can release it. + plan := h.planIdentityProxySessionV3(r.Context(), sessionID, result) + if plan.ProxyNode == nil { + return nil, h.refuseLocalIdentityWorkV3(r, result) + } + return plan.ProxyNode, nil +} + +// applyRemoteTransportMarkV3 records whether the committed route's media bytes +// leave this server or another node. +// +// Every committed transition calls it, including the ones that serve locally: +// a session replanned from a proxy onto the integrated transcoder would +// otherwise keep a stale mark, and the widened idle grace it grants would hold +// that session's stream and transcode slots for five minutes after the local +// stream disconnected without an explicit stop. +func (h *PlaybackHandler) applyRemoteTransportMarkV3(ctx context.Context, sessionID string, remote bool) { + if err := h.sessionMgr.SetRemoteTransport(sessionID, remote); err != nil && + !errors.Is(err, playback.ErrSessionNotFound) { + slog.WarnContext(ctx, "failed to record transport locality", + "component", "api", "playback_session_id", sessionID, "remote", remote, "error", err) + } +} + +// planIdentityProxySessionV3 asks the planner for a proxy, narrowing selection +// to proxies that can execute the plan's frozen recipe when one is required. +// +// The narrowing happens before selection rather than after, mirroring how HLS +// filters transcode nodes: rejecting a single round-robin pick would abandon +// the whole pool, so a capable sibling with free capacity would sit unused +// while playback fell back to the API — or was refused outright when local +// fallback is disabled. +// +// Direct play copies bytes and needs no recipe, so it skips the probe entirely +// and any healthy proxy serves it. +func (h *PlaybackHandler) planIdentityProxySessionV3(ctx context.Context, sessionID string, result playback.PlannerResultV3) nodepool.Plan { + estBitrate := identityStreamBitrateKbpsV3(result) + selector, selectable := h.NodePlanner.(capabilitySessionPlannerV3) + enumerator, enumerable := h.NodePlanner.(proxyNodeEnumeratorV3) + if !selectable || !enumerable || !planRequiresServerTransformationsV3(result.Plan) { + return h.NodePlanner.PlanSession(sessionID, "", false, estBitrate) + } + + capable := make(map[string]struct{}) + for nodeURL, advertised := range h.pooledNodeTransformationsV3(ctx, enumerator.ProxyNodeURLs()) { + if validateAdvertisedTransformationsV3(result.Plan, advertised) == nil { + capable[nodeURL] = struct{}{} + } + } + if len(capable) == 0 { + slog.WarnContext(ctx, "protocol v3 no proxy advertises the planned recipe", + "component", "api", "delivery", result.Plan.Delivery) + } + // The predicate runs under the planner lock: a set lookup only. + return selector.PlanSessionWith(sessionID, "", false, estBitrate, func(node *nodepool.Node) bool { + if node == nil { + return false + } + _, ok := capable[node.URL] + return ok + }) +} + +// refuseLocalIdentityWorkV3 enforces playback.local_transcode_fallback for the +// identity deliveries. Direct play moves bytes and always falls back locally; +// a progressive remux that must convert audio is ffmpeg work, so on an API-only +// node it is refused for the same reason the HLS routes refuse it, instead of +// silently spawning an encoder the operator disabled. +func (h *PlaybackHandler) refuseLocalIdentityWorkV3(r *http.Request, result playback.PlannerResultV3) *transportErrorV3 { + if result.Plan == nil || result.Plan.Delivery != playback.DeliveryRemuxProgressiveV3 || + !planRequiresServerTransformationsV3(result.Plan) || + nodepool.LocalTranscodeFallbackAllowed(r.Context(), h.SettingsRepo) { + return nil + } + return &transportErrorV3{reason: "capacity_unavailable", message: "No proxy node is available and local fallback is disabled.", retryable: true} +} + +// identityStreamBitrateKbpsV3 estimates the bitrate a proxy will egress for an +// identity delivery, so bandwidth-capped proxies admit it accurately. The plan's +// effective recipe is authoritative (a remux that downmixes audio egresses less +// than the source); the source descriptor is the fallback. +func identityStreamBitrateKbpsV3(result playback.PlannerResultV3) int { + if result.Plan == nil { + return 0 + } + if effective := result.Plan.EffectiveRecipe.BitrateKbps; effective != nil && *effective > 0 { + return *effective + } + if source := result.Plan.Source.BitrateKbps; source > 0 { + return source + } + return 0 +} + +// identityStreamURLV3 builds the stream URL for a direct-play or +// progressive-remux session: an absolute proxy URL when one was planned, +// otherwise the API-local path. +// +// The proxy serves from the token alone, so the token has to carry everything +// the API-local path would have read from the session and the file record: the +// media path it opens, and the Dolby Vision profile its remux needs to strip a +// dangling Profile 7 RPU. Omitting either would not fail loudly — the proxy +// would serve a subtly different stream than the plan promised. +// +// The bool reports whether the returned URL is actually a proxy URL, so the +// caller can release the planner reservation when it is not. +func (h *PlaybackHandler) identityStreamURLV3(s *playback.Session, file *models.MediaFile, proxyNode *nodepool.Node) (string, bool) { + if proxyNode == nil || file == nil { + return h.playbackStreamURL(s), false + } + card := identityRecipeCard(s) + card.InputPath = file.FilePath + claims := card.ToClaims() + claims.DVProfile = file.PrimaryDVProfile() + claims.AudioOnly = file.IsAudioOnly() + token := h.signStreamClaims(claims) + if token == "" { + return h.playbackStreamURL(s), false + } + base := strings.TrimRight(proxyNode.URL, "/") + if s.PlayMethod == playback.PlayRemux { + return base + "/stream/remux/" + token, true + } + return base + "/stream/direct/" + token, true } // sessionOwnsResumeTimelineV3 reports whether the session's own position is a @@ -1029,6 +1205,7 @@ func (h *PlaybackHandler) prepareLocalTransportV3(r *http.Request, session *play } committed = true previous := h.tm.SwapTranscodeSession(session.ID, ts) + h.applyRemoteTransportMarkV3(r.Context(), session.ID, false) unlock() if previous != nil && previous != ts { _ = previous.Close() @@ -1087,6 +1264,10 @@ func (h *PlaybackHandler) prepareRemoteTransportV3(r *http.Request, session *pla hw := firstNonEmptyHandlerV3(strings.TrimSpace(nodeResp.HWAccel), strings.TrimSpace(req.HWAccel)) card := playback.NewRecipeCard(session.UserID, session.ProfileID, file.ID, node.URL, playback.TranscodeOpts{InputPath: req.InputPath, SessionID: session.ID, TranscodeTransportID: transportID, SourceVideoCodec: req.SourceVideoCodec, SourceVideoProfile: req.SourceVideoProfile, SourceVideoBitDepth: req.SourceVideoBitDepth, SoftwareVideoDecode: req.SoftwareVideoDecode, VideoBitstreamFilter: req.VideoBitstreamFilter, SeekSeconds: req.SeekSeconds, StreamOriginSeconds: req.StreamOriginSeconds, CopySeekAnchorResolved: req.CopySeekAnchorResolved, StartSegmentNumber: req.StartSegmentNumber, TargetResolution: req.TargetResolution, TargetCodecVideo: req.TargetCodecVideo, TargetCodecAudio: req.TargetCodecAudio, TargetAudioChannels: req.TargetAudioChannels, TargetAudioBitrateKbps: req.TargetAudioBitrateKbps, TargetBitrateKbps: req.TargetBitrateKbps, SegmentDuration: req.SegmentDuration, HWAccel: hw, AudioTrackIndex: req.AudioTrackIndex, SubtitleTrackIndex: req.SubtitleTrackIndex, SubtitleBurnIn: req.SubtitleBurnIn, SubtitleCodec: req.SubtitleCodec, TotalDuration: req.TotalDuration}) url := h.buildProxyManifestURL(card, nodePlan.ProxyNode) + // buildProxyManifestURL only returns an absolute proxy URL when a proxy was + // planned and the token could be signed; otherwise the client fetches the + // manifest from this server and the local liveness path applies. + servedByProxy := nodePlan.ProxyNode != nil && strings.HasPrefix(url, "http") committed := false previousNodeURL := session.TranscodeNodeURL previousTransportID := remoteTransportID(session) @@ -1100,6 +1281,7 @@ func (h *PlaybackHandler) prepareRemoteTransportV3(r *http.Request, session *pla if previousNodeURL != "" { h.tm.StopRemoteTranscode(previousTransportID, previousNodeURL) } + h.applyRemoteTransportMarkV3(r.Context(), session.ID, servedByProxy) unlock() }, rollback: func() { if committed { @@ -2360,15 +2542,32 @@ func shouldTryAlternateFileV3(qualityPreference string) bool { } const ( - terminalNoAlternateVersionV3 = "no_alternate_version" - terminalHDRTranscodeUnsupportedV3 = "hdr_transcode_unsupported" + terminalNoAlternateVersionV3 = "no_alternate_version" + terminalHDRTranscodeUnsupportedV3 = "hdr_transcode_unsupported" + terminalSubtitleConversionUnsupportedV3 = "subtitle_conversion_unsupported" ) +// terminalAllowsAlternateFileV3 reports whether a refusal is the kind another +// version of the same item could satisfy. +// +// subtitle_conversion_unsupported belongs here because it is not only a +// subtitle-format refusal: when a burn-in requirement is the sole trigger of an +// adaptation the source cannot take, the planner reports the blocker in terms +// of the subtitle rather than the HDR pipeline or the 4K policy. Those cases +// used to surface as hdr_transcode_unsupported / no_alternate_version and were +// the exact reason this gate exists — a bitmap subtitle needing burn-in that an +// HDR source cannot support while an SDR alternate can. Leaving the new reason +// out silently retired that fallback and refused playback outright. func terminalAllowsAlternateFileV3(terminal *playback.TerminalV3) bool { if terminal == nil { return false } - return terminal.Reason == terminalNoAlternateVersionV3 || terminal.Reason == terminalHDRTranscodeUnsupportedV3 + switch terminal.Reason { + case terminalNoAlternateVersionV3, terminalHDRTranscodeUnsupportedV3, terminalSubtitleConversionUnsupportedV3: + return true + default: + return false + } } func replanAllowsAlternateFileV3(operation playback.ReplanOperationV3, qualityPreference string) bool { diff --git a/internal/api/handlers/playback_v3_test.go b/internal/api/handlers/playback_v3_test.go index f8c9c8c2..525c4416 100644 --- a/internal/api/handlers/playback_v3_test.go +++ b/internal/api/handlers/playback_v3_test.go @@ -23,6 +23,7 @@ import ( "github.com/Silo-Server/silo-server/internal/models" "github.com/Silo-Server/silo-server/internal/nodepool" "github.com/Silo-Server/silo-server/internal/playback" + "github.com/Silo-Server/silo-server/internal/streamtoken" "github.com/Silo-Server/silo-server/internal/subtitles" "github.com/Silo-Server/silo-server/internal/transcodenode" "github.com/Silo-Server/silo-server/internal/userstore" @@ -3634,3 +3635,518 @@ func postPlaybackReplanV3(t *testing.T, handler *PlaybackHandler, sessionID stri } return response } + +// recordingNodePlannerV3 is a planner double that records what the transport +// asked for and whether the reservation was released. +type recordingNodePlannerV3 struct { + plan nodepool.Plan + needsTranscode bool + estBitrateKbps int + plannedSessionID string + released []string +} + +func (p *recordingNodePlannerV3) PlanSession(sessionID, _ string, needsTranscode bool, estBitrateKbps int) nodepool.Plan { + p.plannedSessionID = sessionID + p.needsTranscode = needsTranscode + p.estBitrateKbps = estBitrateKbps + return p.plan +} + +func (p *recordingNodePlannerV3) ReleaseSession(sessionID string) { + p.released = append(p.released, sessionID) +} + +// PlanSessionWith mirrors the real planner: the eligibility predicate narrows +// the pool before selection, so a proxy that cannot execute the recipe is +// skipped rather than picked and then rejected. +func (p *recordingNodePlannerV3) PlanSessionWith(sessionID, currentTranscodeURL string, needsTranscode bool, estBitrateKbps int, eligible func(*nodepool.Node) bool) nodepool.Plan { + plan := p.PlanSession(sessionID, currentTranscodeURL, needsTranscode, estBitrateKbps) + if eligible == nil { + return plan + } + if !needsTranscode && plan.ProxyNode != nil && !eligible(plan.ProxyNode) { + return nodepool.Plan{} + } + if needsTranscode && plan.TranscodeNode != nil && !eligible(plan.TranscodeNode) { + return nodepool.Plan{} + } + return plan +} + +func (p *recordingNodePlannerV3) ProxyNodeURLs() []string { + if p.plan.ProxyNode == nil { + return nil + } + return []string{p.plan.ProxyNode.URL} +} + +func (p *recordingNodePlannerV3) TranscodeNodeURLs() []string { + if p.plan.TranscodeNode == nil { + return nil + } + return []string{p.plan.TranscodeNode.URL} +} + +func identityProxyPlanV3(delivery playback.DeliveryV3, transformations ...playback.TransformationV3) *playback.PlanV3 { + return &playback.PlanV3{ + PlanID: "plan:identity-proxy", + Delivery: delivery, + Transformations: transformations, + Source: playback.SourceDescriptorV3{BitrateKbps: 5_673}, + } +} + +func TestPrepareTransportV3RoutesDirectPlayThroughProxyNode(t *testing.T) { + handler := NewPlaybackHandler(playback.NewSessionManager(0, 0)) + handler.JWTSecret = "test-secret" + planner := &recordingNodePlannerV3{plan: nodepool.Plan{ProxyNode: &nodepool.Node{URL: "http://proxy-1"}}} + handler.NodePlanner = planner + + transport, transportErr := handler.prepareTransportV3( + httptest.NewRequest(http.MethodPost, "/", nil), + &playback.Session{ID: "session-direct-proxy", UserID: 7, ProfileID: "profile-1"}, + v3HandlerFixtureFile(t), + playback.PlannerResultV3{Plan: identityProxyPlanV3(playback.DeliveryOriginalHTTPV3), PlayMethod: playback.PlayDirect}, + ) + if transportErr != nil { + t.Fatalf("prepare identity transport: %v", transportErr) + } + defer transport.rollback() + + if !strings.HasPrefix(transport.url, "http://proxy-1/stream/direct/") { + t.Fatalf("direct-play stream url = %q, want a proxy /stream/direct URL", transport.url) + } + if planner.needsTranscode { + t.Fatal("direct play must not request a transcode node") + } + if planner.plannedSessionID != "session-direct-proxy" { + t.Fatalf("reservation session id = %q, want the playback session id", planner.plannedSessionID) + } + if planner.estBitrateKbps != 5_673 { + t.Fatalf("estimated bitrate = %d, want the source bitrate for cap admission", planner.estBitrateKbps) + } + + token := strings.TrimPrefix(transport.url, "http://proxy-1/stream/direct/") + claims, err := streamtoken.Verify(token, handler.JWTSecret) + if err != nil { + t.Fatalf("verify proxy stream token: %v", err) + } + if claims.MediaPath == "" { + t.Fatal("proxy token carries no media path; the proxy cannot open the source") + } + if claims.PlayMethod != string(playback.PlayDirect) { + t.Fatalf("token play method = %q, want direct", claims.PlayMethod) + } +} + +func TestPrepareTransportV3RoutesProgressiveRemuxThroughProxyNodeWithSeekAndDV(t *testing.T) { + handler := NewPlaybackHandler(playback.NewSessionManager(0, 0)) + handler.JWTSecret = "test-secret" + stubCopySeekAnchorV3(handler) + proxy := capableProxyStubV3(t) + handler.NodePlanner = &recordingNodePlannerV3{plan: nodepool.Plan{ProxyNode: &nodepool.Node{URL: proxy.URL + "/"}}} + + file := v3HandlerFixtureFile(t) + file.VideoTracks[0].DVProfile = 7 + + plan := identityProxyPlanV3(playback.DeliveryRemuxProgressiveV3, playback.TransformationV3{Name: playback.TransformationAudioToAACV3, Executor: playback.ExecutorServerV3, RecipeVersion: "1"}) + plan.Timeline = playback.TimelineV3{SourceStartSeconds: 39.5} + + transport, transportErr := handler.prepareTransportV3( + httptest.NewRequest(http.MethodPost, "/", nil), + &playback.Session{ID: "session-remux-proxy", UserID: 7, ProfileID: "profile-1"}, + file, + playback.PlannerResultV3{Plan: plan, PlayMethod: playback.PlayRemux, TranscodeAudio: true, TargetAudioCodec: "aac"}, + ) + if transportErr != nil { + t.Fatalf("prepare identity transport: %v", transportErr) + } + defer transport.rollback() + + remuxPrefix := proxy.URL + "/stream/remux/" + if !strings.HasPrefix(transport.url, remuxPrefix) { + t.Fatalf("remux stream url = %q, want a proxy /stream/remux URL", transport.url) + } + if !strings.Contains(transport.url, "seek=39.5") { + t.Fatalf("remux stream url = %q, want the resolved seek preserved", transport.url) + } + token := strings.TrimSuffix(strings.TrimPrefix(transport.url, remuxPrefix), "?seek=39.5") + claims, err := streamtoken.Verify(token, handler.JWTSecret) + if err != nil { + t.Fatalf("verify proxy stream token: %v", err) + } + if claims.DVProfile != 7 { + t.Fatalf("token DV profile = %d, want 7 so the proxy strips the dangling RPU", claims.DVProfile) + } + if !claims.TranscodeAudio { + t.Fatal("token must tell the proxy to convert audio") + } +} + +func TestPrepareTransportV3FallsBackLocallyWithoutEligibleProxy(t *testing.T) { + handler := NewPlaybackHandler(playback.NewSessionManager(0, 0)) + handler.JWTSecret = "test-secret" + handler.NodePlanner = &recordingNodePlannerV3{plan: nodepool.Plan{}} + + transport, transportErr := handler.prepareTransportV3( + httptest.NewRequest(http.MethodPost, "/", nil), + &playback.Session{ID: "session-direct-local", UserID: 7, ProfileID: "profile-1"}, + v3HandlerFixtureFile(t), + playback.PlannerResultV3{Plan: identityProxyPlanV3(playback.DeliveryOriginalHTTPV3), PlayMethod: playback.PlayDirect}, + ) + if transportErr != nil { + t.Fatalf("prepare identity transport: %v", transportErr) + } + defer transport.rollback() + + if !strings.HasPrefix(transport.url, "/stream/session-direct-local") { + t.Fatalf("stream url = %q, want the API-local path when no proxy is eligible", transport.url) + } +} + +func TestPrepareTransportV3RefusesLocalRemuxWhenFallbackDisabled(t *testing.T) { + handler := NewPlaybackHandler(playback.NewSessionManager(0, 0)) + handler.JWTSecret = "test-secret" + stubCopySeekAnchorV3(handler) + handler.NodePlanner = &recordingNodePlannerV3{plan: nodepool.Plan{}} + handler.SettingsRepo = &mutablePlaybackSettingsV3{values: map[string]string{"playback.local_transcode_fallback": "false"}} + + _, transportErr := handler.prepareTransportV3( + httptest.NewRequest(http.MethodPost, "/", nil), + &playback.Session{ID: "session-remux-refused", UserID: 7, ProfileID: "profile-1"}, + v3HandlerFixtureFile(t), + playback.PlannerResultV3{ + Plan: identityProxyPlanV3(playback.DeliveryRemuxProgressiveV3, playback.TransformationV3{Name: playback.TransformationAudioToAACV3, Executor: playback.ExecutorServerV3, RecipeVersion: "1"}), + PlayMethod: playback.PlayRemux, + TranscodeAudio: true, + }, + ) + if transportErr == nil || transportErr.reason != "capacity_unavailable" { + t.Fatalf("transport error = %#v, want capacity_unavailable when local remux work is disabled", transportErr) + } +} + +func TestPrepareTransportV3AllowsLocalDirectPlayWhenFallbackDisabled(t *testing.T) { + handler := NewPlaybackHandler(playback.NewSessionManager(0, 0)) + handler.JWTSecret = "test-secret" + handler.NodePlanner = &recordingNodePlannerV3{plan: nodepool.Plan{}} + handler.SettingsRepo = &mutablePlaybackSettingsV3{values: map[string]string{"playback.local_transcode_fallback": "false"}} + + // Direct play moves bytes rather than running ffmpeg, so the transcode + // fallback gate must not strand a single-node deployment. + transport, transportErr := handler.prepareTransportV3( + httptest.NewRequest(http.MethodPost, "/", nil), + &playback.Session{ID: "session-direct-allowed", UserID: 7, ProfileID: "profile-1"}, + v3HandlerFixtureFile(t), + playback.PlannerResultV3{Plan: identityProxyPlanV3(playback.DeliveryOriginalHTTPV3), PlayMethod: playback.PlayDirect}, + ) + if transportErr != nil { + t.Fatalf("direct play refused with local fallback disabled: %#v", transportErr) + } + transport.rollback() +} + +func TestPrepareTransportV3ReleasesProxyReservationOnRollback(t *testing.T) { + handler := NewPlaybackHandler(playback.NewSessionManager(0, 0)) + handler.JWTSecret = "test-secret" + planner := &recordingNodePlannerV3{plan: nodepool.Plan{ProxyNode: &nodepool.Node{URL: "http://proxy-1"}}} + handler.NodePlanner = planner + + transport, transportErr := handler.prepareTransportV3( + httptest.NewRequest(http.MethodPost, "/", nil), + &playback.Session{ID: "session-rollback", UserID: 7, ProfileID: "profile-1"}, + v3HandlerFixtureFile(t), + playback.PlannerResultV3{Plan: identityProxyPlanV3(playback.DeliveryOriginalHTTPV3), PlayMethod: playback.PlayDirect}, + ) + if transportErr != nil { + t.Fatalf("prepare identity transport: %v", transportErr) + } + transport.rollback() + + if len(planner.released) != 1 || planner.released[0] != "session-rollback" { + t.Fatalf("released reservations = %v, want the session released once", planner.released) + } +} + +// capableProxyStubV3 is a proxy that advertises the audio_to_aac and DV strip +// recipes, mirroring the /hw-capabilities endpoint a real proxy now serves. +func capableProxyStubV3(t *testing.T) *httptest.Server { + t.Helper() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/hw-capabilities" { + w.WriteHeader(http.StatusNotFound) + return + } + writeJSON(w, http.StatusOK, playback.HWAccelInfo{Transformations: []playback.TransformationV3{ + {Name: playback.TransformationAudioToAACV3, Executor: playback.ExecutorServerV3, RecipeVersion: "1"}, + {Name: playback.TransformationServerDV7HDR10V3, Executor: playback.ExecutorServerV3, RecipeVersion: "1"}, + }}) + })) + t.Cleanup(server.Close) + return server +} + +func TestPrepareTransportV3KeepsRemuxLocalWhenProxyLacksTheRecipe(t *testing.T) { + // A proxy on an older ffmpeg build advertises no aac encoder. Sending the + // remux there would 500 at stream time, so selection must reject it. + proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, playback.HWAccelInfo{Transformations: []playback.TransformationV3{ + {Name: playback.TransformationVideoToH264V3, Executor: playback.ExecutorServerV3, RecipeVersion: "2"}, + }}) + })) + defer proxy.Close() + + handler := NewPlaybackHandler(playback.NewSessionManager(0, 0)) + handler.JWTSecret = "test-secret" + stubCopySeekAnchorV3(handler) + planner := &recordingNodePlannerV3{plan: nodepool.Plan{ProxyNode: &nodepool.Node{URL: proxy.URL}}} + handler.NodePlanner = planner + + transport, transportErr := handler.prepareTransportV3( + httptest.NewRequest(http.MethodPost, "/", nil), + &playback.Session{ID: "session-incapable-proxy", UserID: 7, ProfileID: "profile-1"}, + v3HandlerFixtureFile(t), + playback.PlannerResultV3{ + Plan: identityProxyPlanV3(playback.DeliveryRemuxProgressiveV3, playback.TransformationV3{Name: playback.TransformationAudioToAACV3, Executor: playback.ExecutorServerV3, RecipeVersion: "1"}), + PlayMethod: playback.PlayRemux, + TranscodeAudio: true, + }, + ) + if transportErr != nil { + t.Fatalf("prepare identity transport: %v", transportErr) + } + defer transport.rollback() + + if strings.HasPrefix(transport.url, proxy.URL) { + t.Fatalf("stream url = %q, want local fallback when the proxy lacks the recipe", transport.url) + } + // Narrowing happens before selection, so no reservation is made against an + // incapable proxy in the first place and none needs releasing. + if len(planner.released) != 0 { + t.Fatalf("released = %v, want no reservation taken on an incapable proxy", planner.released) + } +} + +func TestPrepareTransportV3DirectPlaySkipsProxyCapabilityProbe(t *testing.T) { + // Direct play copies bytes and needs no recipe, so it must not pay a + // capability fetch nor be rejected by one. + probes := 0 + proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + probes++ + w.WriteHeader(http.StatusNotFound) + })) + defer proxy.Close() + + handler := NewPlaybackHandler(playback.NewSessionManager(0, 0)) + handler.JWTSecret = "test-secret" + handler.NodePlanner = &recordingNodePlannerV3{plan: nodepool.Plan{ProxyNode: &nodepool.Node{URL: proxy.URL}}} + + transport, transportErr := handler.prepareTransportV3( + httptest.NewRequest(http.MethodPost, "/", nil), + &playback.Session{ID: "session-direct-noprobe", UserID: 7, ProfileID: "profile-1"}, + v3HandlerFixtureFile(t), + playback.PlannerResultV3{Plan: identityProxyPlanV3(playback.DeliveryOriginalHTTPV3), PlayMethod: playback.PlayDirect}, + ) + if transportErr != nil { + t.Fatalf("prepare identity transport: %v", transportErr) + } + defer transport.rollback() + + if !strings.HasPrefix(transport.url, proxy.URL+"/stream/direct/") { + t.Fatalf("stream url = %q, want the proxy direct URL", transport.url) + } + if probes != 0 { + t.Fatalf("capability probes = %d, want none for direct play", probes) + } +} + +func TestIdentityStreamBitrateKbpsV3PrefersEffectiveRecipe(t *testing.T) { + // A downmixing remux egresses less than the source, so the effective + // recipe — not the source descriptor — is what a capped proxy admits on. + effective := 1_200 + plan := identityProxyPlanV3(playback.DeliveryRemuxProgressiveV3) + plan.EffectiveRecipe.BitrateKbps = &effective + + if got := identityStreamBitrateKbpsV3(playback.PlannerResultV3{Plan: plan}); got != effective { + t.Fatalf("bitrate estimate = %d, want the effective recipe value %d", got, effective) + } + + plan.EffectiveRecipe.BitrateKbps = nil + if got := identityStreamBitrateKbpsV3(playback.PlannerResultV3{Plan: plan}); got != 5_673 { + t.Fatalf("bitrate estimate = %d, want the source fallback 5673", got) + } +} + +func TestPrepareTransportV3MarksProxySessionsAsRemotelyTransported(t *testing.T) { + manager := playback.NewSessionManager(0, 0) + handler := NewPlaybackHandler(manager) + handler.JWTSecret = "test-secret" + handler.NodePlanner = &recordingNodePlannerV3{plan: nodepool.Plan{ProxyNode: &nodepool.Node{URL: "http://proxy-1"}}} + + session, err := manager.StartSession(7, "profile-1", 42, playback.PlayDirect, false) + if err != nil { + t.Fatalf("start session: %v", err) + } + + transport, transportErr := handler.prepareTransportV3( + httptest.NewRequest(http.MethodPost, "/", nil), + session, + v3HandlerFixtureFile(t), + playback.PlannerResultV3{Plan: identityProxyPlanV3(playback.DeliveryOriginalHTTPV3), PlayMethod: playback.PlayDirect}, + ) + if transportErr != nil { + t.Fatalf("prepare identity transport: %v", transportErr) + } + transport.commit() + + // The client talks to the proxy, so this server sees no transport request. + // Without the remote-transport mark the idle sweep would reap a healthy + // stream as soon as heartbeats lapsed. + if expired := manager.CleanInactive(time.Nanosecond, time.Nanosecond); len(expired) != 0 { + t.Fatalf("expired %d proxy-served sessions, want them preserved", len(expired)) + } + if _, err := manager.GetSession(session.ID); err != nil { + t.Fatalf("proxy-served session was reaped: %v", err) + } +} + +// poolNodePlannerV3 models a multi-proxy pool so capability narrowing can be +// observed end to end rather than only at the planner unit level. +type poolNodePlannerV3 struct { + proxies []*nodepool.Node + released []string +} + +func (p *poolNodePlannerV3) PlanSession(string, string, bool, int) nodepool.Plan { + if len(p.proxies) == 0 { + return nodepool.Plan{} + } + return nodepool.Plan{ProxyNode: p.proxies[0]} +} + +func (p *poolNodePlannerV3) PlanSessionWith(sessionID, currentTranscodeURL string, needsTranscode bool, estBitrateKbps int, eligible func(*nodepool.Node) bool) nodepool.Plan { + for _, node := range p.proxies { + if eligible == nil || eligible(node) { + return nodepool.Plan{ProxyNode: node} + } + } + return nodepool.Plan{} +} + +func (p *poolNodePlannerV3) ProxyNodeURLs() []string { + urls := make([]string, 0, len(p.proxies)) + for _, node := range p.proxies { + urls = append(urls, node.URL) + } + return urls +} + +func (p *poolNodePlannerV3) TranscodeNodeURLs() []string { return nil } + +func (p *poolNodePlannerV3) ReleaseSession(sessionID string) { + p.released = append(p.released, sessionID) +} + +func TestPrepareTransportV3PrefersACapableSiblingProxy(t *testing.T) { + // A rolling upgrade leaves one proxy on an ffmpeg without an aac encoder. + // The remux must land on its capable sibling rather than falling back to + // the API node, which would be refused outright with local fallback off. + stale := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, playback.HWAccelInfo{Transformations: []playback.TransformationV3{ + {Name: playback.TransformationVideoToH264V3, Executor: playback.ExecutorServerV3, RecipeVersion: "2"}, + }}) + })) + defer stale.Close() + upgraded := capableProxyStubV3(t) + + handler := NewPlaybackHandler(playback.NewSessionManager(0, 0)) + handler.JWTSecret = "test-secret" + stubCopySeekAnchorV3(handler) + handler.SettingsRepo = &mutablePlaybackSettingsV3{values: map[string]string{"playback.local_transcode_fallback": "false"}} + handler.NodePlanner = &poolNodePlannerV3{proxies: []*nodepool.Node{ + {ID: 1, URL: stale.URL}, + {ID: 2, URL: upgraded.URL}, + }} + + transport, transportErr := handler.prepareTransportV3( + httptest.NewRequest(http.MethodPost, "/", nil), + &playback.Session{ID: "session-sibling", UserID: 7, ProfileID: "profile-1"}, + v3HandlerFixtureFile(t), + playback.PlannerResultV3{ + Plan: identityProxyPlanV3(playback.DeliveryRemuxProgressiveV3, playback.TransformationV3{Name: playback.TransformationAudioToAACV3, Executor: playback.ExecutorServerV3, RecipeVersion: "1"}), + PlayMethod: playback.PlayRemux, + TranscodeAudio: true, + }, + ) + if transportErr != nil { + t.Fatalf("prepare identity transport: %#v", transportErr) + } + defer transport.rollback() + + if !strings.HasPrefix(transport.url, upgraded.URL+"/stream/remux/") { + t.Fatalf("stream url = %q, want the capable sibling proxy %q", transport.url, upgraded.URL) + } +} + +func TestPrepareTransportV3ClearsRemoteTransportMarkWhenServingLocally(t *testing.T) { + // A session replanned from a proxy onto this server must lose the widened + // idle grace, or its stream and transcode slots stay held for minutes after + // a local stream disconnects without an explicit stop. + manager := playback.NewSessionManager(0, 0) + handler := NewPlaybackHandler(manager) + handler.JWTSecret = "test-secret" + handler.NodePlanner = &recordingNodePlannerV3{plan: nodepool.Plan{ProxyNode: &nodepool.Node{URL: "http://proxy-1"}}} + + session, err := manager.StartSession(7, "profile-1", 42, playback.PlayDirect, false) + if err != nil { + t.Fatalf("start session: %v", err) + } + file := v3HandlerFixtureFile(t) + + proxied, transportErr := handler.prepareTransportV3( + httptest.NewRequest(http.MethodPost, "/", nil), session, file, + playback.PlannerResultV3{Plan: identityProxyPlanV3(playback.DeliveryOriginalHTTPV3), PlayMethod: playback.PlayDirect}, + ) + if transportErr != nil { + t.Fatalf("prepare proxy transport: %v", transportErr) + } + proxied.commit() + + // Re-plan with no eligible proxy: the same delivery now serves locally. + handler.NodePlanner = &recordingNodePlannerV3{plan: nodepool.Plan{}} + local, transportErr := handler.prepareTransportV3( + httptest.NewRequest(http.MethodPost, "/", nil), session, file, + playback.PlannerResultV3{Plan: identityProxyPlanV3(playback.DeliveryOriginalHTTPV3), PlayMethod: playback.PlayDirect}, + ) + if transportErr != nil { + t.Fatalf("prepare local transport: %v", transportErr) + } + local.commit() + + if expired := manager.CleanInactive(time.Nanosecond, time.Nanosecond); len(expired) != 1 { + t.Fatalf("expired %d sessions, want the locally-served session to lose its widened grace", len(expired)) + } +} + +// A burn-in requirement that only an SDR alternate can satisfy must still reach +// the alternate-version fallback. The planner reports that refusal in terms of +// the subtitle rather than the HDR pipeline, so a gate listing only the HDR and +// version reasons silently retires the fallback and refuses playback outright. +func TestTerminalAllowsAlternateFileV3CoversSubtitleForcedRefusals(t *testing.T) { + for _, reason := range []string{ + terminalNoAlternateVersionV3, + terminalHDRTranscodeUnsupportedV3, + terminalSubtitleConversionUnsupportedV3, + } { + if !terminalAllowsAlternateFileV3(&playback.TerminalV3{Reason: reason}) { + t.Fatalf("terminal %q must allow an alternate-version retry", reason) + } + } + if terminalAllowsAlternateFileV3(&playback.TerminalV3{Reason: "client_hls_unsupported"}) { + t.Fatal("a client-route refusal must not trigger an alternate-version retry") + } + if terminalAllowsAlternateFileV3(nil) { + t.Fatal("a nil terminal must not trigger an alternate-version retry") + } +} diff --git a/internal/nodepool/planner.go b/internal/nodepool/planner.go index 94d42609..112c7bfc 100644 --- a/internal/nodepool/planner.go +++ b/internal/nodepool/planner.go @@ -193,24 +193,27 @@ func (p *Planner) PlanSessionWith(sessionID, currentTranscodeURL string, needsTr } proxies := p.proxies.Nodes() transcodes := p.transcodes.Nodes() + // Group health is computed over the full pool before any narrowing: + // eligibility restricts what may be selected, not co-location semantics. groupHealthy := groupHealth(proxies, transcodes) - if eligible != nil { - filtered := make([]*Node, 0, len(transcodes)) - for _, node := range transcodes { - if eligible(node) { - filtered = append(filtered, node) - } - } - transcodes = filtered - } var plan Plan if needsTranscode { + if eligible != nil { + transcodes = filterNodes(transcodes, eligible) + } plan.TranscodeNode = p.pickTranscode(transcodes, proxies, groupHealthy, currentTranscodeURL, estBitrateKbps, now) if plan.TranscodeNode != nil { plan.ProxyNode = p.pickProxy(proxies, groupHealthy, plan.TranscodeNode.Group, estBitrateKbps, now) } } else { + // A proxy-only plan has no transcode node, so the predicate applies to + // the proxy: it is the node that will execute the recipe. Filtering + // before selection means a capability mismatch skips to a capable + // sibling instead of abandoning the pool after one round-robin pick. + if eligible != nil { + proxies = filterNodes(proxies, eligible) + } plan.ProxyNode = p.pickProxy(proxies, groupHealthy, nil, estBitrateKbps, now) } @@ -228,6 +231,36 @@ func (p *Planner) PlanSessionWith(sessionID, currentTranscodeURL string, needsTr return plan } +// filterNodes returns the nodes accepted by keep, preserving pool order so +// round-robin cursors stay meaningful across selections. +func filterNodes(nodes []*Node, keep func(*Node) bool) []*Node { + filtered := make([]*Node, 0, len(nodes)) + for _, node := range nodes { + if keep(node) { + filtered = append(filtered, node) + } + } + return filtered +} + +// ProxyNodeURLs lists the URLs of every enabled pooled proxy node, healthy or +// not, mirroring TranscodeNodeURLs. Capability planning wants the deployment's +// toolchain; an unreachable node excludes itself when its capability fetch +// fails. +func (p *Planner) ProxyNodeURLs() []string { + if p == nil || p.proxies == nil { + return nil + } + nodes := p.proxies.Nodes() + urls := make([]string, 0, len(nodes)) + for _, node := range nodes { + if node != nil && node.URL != "" { + urls = append(urls, node.URL) + } + } + return urls +} + // TranscodeNodeURLs lists the URLs of every enabled pooled transcode node, // healthy or not: capability planning wants the deployment's toolchain, and // an unreachable node excludes itself when its capability fetch fails. An diff --git a/internal/nodepool/planner_test.go b/internal/nodepool/planner_test.go index 78749e7c..47a06b8f 100644 --- a/internal/nodepool/planner_test.go +++ b/internal/nodepool/planner_test.go @@ -1,6 +1,7 @@ package nodepool import ( + "strconv" "testing" "time" ) @@ -615,3 +616,58 @@ func TestPlanDownloadFallsBackWhenOriginGroupHasNoProxy(t *testing.T) { t.Fatalf("plan = %+v, want proxy-a fallback", plan) } } + +// A proxy-only plan applies the eligibility predicate to the proxy: it is the +// node that executes the recipe. Without this, one incapable round-robin pick +// would abandon a pool that still holds a capable sibling. +func TestPlanSessionWithFiltersProxiesForProxyOnlyPlans(t *testing.T) { + proxies := NewProxyPool() + proxies.SetNodes([]*Node{ + {ID: 1, URL: "http://proxy-old", Enabled: true, Healthy: true}, + {ID: 2, URL: "http://proxy-new", Enabled: true, Healthy: true}, + }) + planner := NewPlanner(proxies, NewTranscodePool()) + + // Only the upgraded proxy can run the recipe; every selection must land + // there regardless of where the round-robin cursor happens to be. + for i := 0; i < 4; i++ { + plan := planner.PlanSessionWith("session-"+strconv.Itoa(i), "", false, 0, func(n *Node) bool { + return n.URL == "http://proxy-new" + }) + if plan.ProxyNode == nil || plan.ProxyNode.URL != "http://proxy-new" { + t.Fatalf("selection %d = %#v, want the capable proxy", i, plan.ProxyNode) + } + } +} + +func TestPlanSessionWithReturnsNoProxyWhenNoneAreCapable(t *testing.T) { + proxies := NewProxyPool() + proxies.SetNodes([]*Node{{ID: 1, URL: "http://proxy-old", Enabled: true, Healthy: true}}) + planner := NewPlanner(proxies, NewTranscodePool()) + + plan := planner.PlanSessionWith("session-none", "", false, 0, func(*Node) bool { return false }) + if plan.ProxyNode != nil { + t.Fatalf("proxy = %#v, want none when the pool cannot execute the recipe", plan.ProxyNode) + } + // An empty plan must not leave a reservation pinning capacity. + if _, reserved := planner.reserved["session-none"]; reserved { + t.Fatal("an unsatisfiable plan left a reservation behind") + } +} + +func TestProxyNodeURLsListsEnabledProxies(t *testing.T) { + proxies := NewProxyPool() + proxies.SetNodes([]*Node{ + {ID: 1, URL: "http://proxy-1", Enabled: true, Healthy: true}, + {ID: 2, URL: "http://proxy-2", Enabled: true, Healthy: false}, + }) + planner := NewPlanner(proxies, NewTranscodePool()) + + // Unhealthy nodes are still listed: capability planning wants the + // deployment's toolchain, and an unreachable node excludes itself when its + // capability fetch fails. + urls := planner.ProxyNodeURLs() + if len(urls) != 2 { + t.Fatalf("proxy urls = %v, want both pooled proxies", urls) + } +} diff --git a/internal/playback/session.go b/internal/playback/session.go index 61f409ea..0db781fa 100644 --- a/internal/playback/session.go +++ b/internal/playback/session.go @@ -62,6 +62,10 @@ type Session struct { activeTransportCount int replacementPlayMethod PlayMethod streamRevision uint64 + // remoteTransport marks a session whose media bytes are served by another + // node, so this server never sees the transport request that would + // otherwise keep it alive. See SetRemoteTransport. + remoteTransport bool } // SessionStreamState stores the mutable stream-specific details that can @@ -231,6 +235,12 @@ const ( // observed playback activity before it stops counting toward limits. DefaultActiveSessionGrace = 45 * time.Second + // remoteTransportIdleGrace is the floor on the idle windows for a session + // whose media is served by another node. See remoteTransportGrace: it must + // outlast a heartbeat gap on an otherwise healthy proxy stream, while still + // reaping a session whose client vanished without stopping. + remoteTransportIdleGrace = 5 * time.Minute + // DefaultPausedSessionGrace is the longer grace period for paused // sessions. It must comfortably cover an intentional pause (dinner // break, phone call): reaping a paused session kills its transcode @@ -1095,6 +1105,34 @@ func (m *SessionManager) BeginTransport(sessionID string) error { return nil } +// SetRemoteTransport records whether this session's media bytes are served by +// another node (a proxy), rather than by a transport request this server +// handles itself. +// +// A locally-served stream is protected from the idle reaper by +// BeginTransport/EndTransport around the serve call. A proxy-served stream +// never enters that path: the client talks to the proxy directly, so from this +// server's point of view a healthy multi-hour stream looks identical to an +// abandoned session, and a heartbeat gap longer than the active grace would +// reap it mid-playback — after which progress, stop, and replan all fail with +// session-not-found while bytes are still flowing. +// +// This marks the session as remotely transported for its whole lifetime rather +// than per-request, because there is no request to bracket. Progress +// heartbeats still drive UI liveness; this only prevents reaping. +func (m *SessionManager) SetRemoteTransport(sessionID string, remote bool) error { + m.mu.Lock() + defer m.mu.Unlock() + + s, ok := m.sessions[sessionID] + if !ok { + return ErrSessionNotFound + } + s.remoteTransport = remote + m.touchSessionLocked(s) + return nil +} + // EndTransport decrements the count of in-flight media transport requests for // the session and refreshes its activity timestamp. func (m *SessionManager) EndTransport(sessionID string) error { @@ -1275,6 +1313,7 @@ func (m *SessionManager) CleanInactive(activeIdle, pausedIdle time.Duration) []* if s.activeTransportCount > 0 { continue } + activeIdle, pausedIdle := remoteTransportGrace(s, activeIdle, pausedIdle) if m.sessionIsInactiveLocked(s, now, activeIdle, pausedIdle) { cp := *s expired = append(expired, &cp) @@ -1305,7 +1344,32 @@ func (m *SessionManager) countsTowardLimitsLocked(s *Session, now time.Time) boo if s.activeTransportCount > 0 { return true } - return !m.sessionIsInactiveLocked(s, now, m.activeGrace, m.pausedGrace) + activeGrace, pausedGrace := remoteTransportGrace(s, m.activeGrace, m.pausedGrace) + return !m.sessionIsInactiveLocked(s, now, activeGrace, pausedGrace) +} + +// remoteTransportGrace widens the idle windows for a session whose bytes are +// served by another node. +// +// A locally-served stream is held open by an in-flight transport request. A +// proxy-served one has no such request here, so it is protected only by the +// client's progress heartbeats — and a gap longer than the active grace would +// reap it while media is still flowing. The windows are widened rather than +// made infinite: there is no absolute session lifetime cap in this manager, so +// unconditional immunity would leak a session forever whenever a client +// disappears without stopping. A client that has gone quiet for this long has +// genuinely stopped watching. +func remoteTransportGrace(s *Session, activeIdle, pausedIdle time.Duration) (time.Duration, time.Duration) { + if s == nil || !s.remoteTransport { + return activeIdle, pausedIdle + } + if activeIdle > 0 && activeIdle < remoteTransportIdleGrace { + activeIdle = remoteTransportIdleGrace + } + if pausedIdle > 0 && pausedIdle < remoteTransportIdleGrace { + pausedIdle = remoteTransportIdleGrace + } + return activeIdle, pausedIdle } func (m *SessionManager) sessionIsInactiveLocked(s *Session, now time.Time, activeIdle, pausedIdle time.Duration) bool { diff --git a/internal/playback/session_remote_transport_test.go b/internal/playback/session_remote_transport_test.go new file mode 100644 index 00000000..4c5751ab --- /dev/null +++ b/internal/playback/session_remote_transport_test.go @@ -0,0 +1,97 @@ +package playback + +import ( + "testing" + "time" +) + +// A proxy-served session has no in-flight transport request on this server, so +// the idle sweep is all that stands between a healthy multi-hour stream and +// being reaped mid-playback. +func TestRemoteTransportGraceWidensIdleWindows(t *testing.T) { + local := &Session{} + remote := &Session{remoteTransport: true} + + if active, paused := remoteTransportGrace(local, 45*time.Second, 2*time.Minute); active != 45*time.Second || paused != 2*time.Minute { + t.Fatalf("local grace = (%v, %v), want the configured windows unchanged", active, paused) + } + + active, paused := remoteTransportGrace(remote, 45*time.Second, 2*time.Minute) + if active != remoteTransportIdleGrace || paused != remoteTransportIdleGrace { + t.Fatalf("remote grace = (%v, %v), want both widened to %v", active, paused, remoteTransportIdleGrace) + } +} + +// The widened window must not become immunity: this manager has no absolute +// session lifetime, so a client that disappears without stopping would leak the +// session forever. +func TestRemoteTransportGraceStillExpires(t *testing.T) { + manager := NewSessionManager(0, 0) + session, err := manager.StartSession(7, "profile-1", 42, PlayDirect, false) + if err != nil { + t.Fatalf("start session: %v", err) + } + if err := manager.SetRemoteTransport(session.ID, true); err != nil { + t.Fatalf("mark remote transport: %v", err) + } + + // Backdate the session past the widened window, as an abandoned client would. + manager.mu.Lock() + stale := time.Now().Add(-2 * remoteTransportIdleGrace) + manager.sessions[session.ID].LastActivityAt = stale + manager.sessions[session.ID].UpdatedAt = stale + manager.mu.Unlock() + + if expired := manager.CleanInactive(45*time.Second, 2*time.Minute); len(expired) != 1 { + t.Fatalf("expired %d abandoned proxy sessions, want 1", len(expired)) + } +} + +// A session inside the widened window survives a heartbeat gap that would have +// reaped it before, which is the reported failure this guards against. +func TestRemoteTransportSurvivesHeartbeatGap(t *testing.T) { + manager := NewSessionManager(0, 0) + session, err := manager.StartSession(7, "profile-1", 42, PlayDirect, false) + if err != nil { + t.Fatalf("start session: %v", err) + } + if err := manager.SetRemoteTransport(session.ID, true); err != nil { + t.Fatalf("mark remote transport: %v", err) + } + + manager.mu.Lock() + gap := time.Now().Add(-90 * time.Second) // twice the default active grace + manager.sessions[session.ID].LastActivityAt = gap + manager.sessions[session.ID].UpdatedAt = gap + manager.mu.Unlock() + + if expired := manager.CleanInactive(DefaultActiveSessionGrace, DefaultPausedSessionGrace); len(expired) != 0 { + t.Fatalf("reaped %d healthy proxy-served sessions during a heartbeat gap", len(expired)) + } +} + +// Clearing the mark restores normal reaping, so a re-plan that moves a session +// back onto this server does not leave it with a stale widened grace. +func TestRemoteTransportCanBeCleared(t *testing.T) { + manager := NewSessionManager(0, 0) + session, err := manager.StartSession(7, "profile-1", 42, PlayDirect, false) + if err != nil { + t.Fatalf("start session: %v", err) + } + if err := manager.SetRemoteTransport(session.ID, true); err != nil { + t.Fatalf("mark remote transport: %v", err) + } + if err := manager.SetRemoteTransport(session.ID, false); err != nil { + t.Fatalf("clear remote transport: %v", err) + } + + manager.mu.Lock() + gap := time.Now().Add(-90 * time.Second) + manager.sessions[session.ID].LastActivityAt = gap + manager.sessions[session.ID].UpdatedAt = gap + manager.mu.Unlock() + + if expired := manager.CleanInactive(DefaultActiveSessionGrace, DefaultPausedSessionGrace); len(expired) != 1 { + t.Fatalf("expired %d sessions after clearing the mark, want 1", len(expired)) + } +} diff --git a/internal/playback/testdata/protocol_v3/capability_response.json b/internal/playback/testdata/protocol_v3/capability_response.json index c66a8100..ca2e7674 100644 --- a/internal/playback/testdata/protocol_v3/capability_response.json +++ b/internal/playback/testdata/protocol_v3/capability_response.json @@ -10,6 +10,7 @@ "playback_route_diagnostics", "device_quirks_v1", "seek_reanchor_v1", + "output_change_v1", "direct_stream_resume_v1", "plan_source_duration_v1" ], diff --git a/internal/playback/testdata/protocol_v3/conformance_matrix.json b/internal/playback/testdata/protocol_v3/conformance_matrix.json index c803cebb..5a850578 100644 --- a/internal/playback/testdata/protocol_v3/conformance_matrix.json +++ b/internal/playback/testdata/protocol_v3/conformance_matrix.json @@ -5406,6 +5406,7 @@ "playback_route_diagnostics", "device_quirks_v1", "seek_reanchor_v1", + "output_change_v1", "direct_stream_resume_v1", "plan_source_duration_v1" ], diff --git a/internal/playback/testdata/protocol_v3/decision_response.json b/internal/playback/testdata/protocol_v3/decision_response.json index d7345acf..2cca07df 100644 --- a/internal/playback/testdata/protocol_v3/decision_response.json +++ b/internal/playback/testdata/protocol_v3/decision_response.json @@ -7,6 +7,7 @@ "playback_route_diagnostics", "device_quirks_v1", "seek_reanchor_v1", + "output_change_v1", "direct_stream_resume_v1", "plan_source_duration_v1" ], diff --git a/internal/proxy/capabilities_test.go b/internal/proxy/capabilities_test.go new file mode 100644 index 00000000..0bf263e2 --- /dev/null +++ b/internal/proxy/capabilities_test.go @@ -0,0 +1,50 @@ +package proxy + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/Silo-Server/silo-server/internal/playback" +) + +// The API validates a frozen remux recipe against the proxy that will execute +// it, so a proxy must answer the same capability probe a transcode node does. +// Without it the API cannot tell a capable proxy from one on an older ffmpeg, +// and the mismatch only surfaces as a failed stream. +func TestProxyAdvertisesTransformationCapabilities(t *testing.T) { + const secret = "capability-secret" + server := newDownloadProxyServer(t, secret) + + request := httptest.NewRequest(http.MethodGet, "/hw-capabilities", nil) + request.Header.Set("Authorization", "Bearer "+secret) + recorder := httptest.NewRecorder() + server.Handler().ServeHTTP(recorder, request) + + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", recorder.Code) + } + var info playback.HWAccelInfo + if err := json.Unmarshal(recorder.Body.Bytes(), &info); err != nil { + t.Fatalf("decode capabilities: %v", err) + } + // The probe reflects whatever ffmpeg this machine has; the contract under + // test is that the endpoint exists and answers in the shared shape. + for _, transformation := range info.Transformations { + if transformation.Name == "" || transformation.RecipeVersion == "" { + t.Fatalf("advertised transformation is missing identity: %#v", transformation) + } + } +} + +func TestProxyCapabilitiesRequireBearer(t *testing.T) { + server := newDownloadProxyServer(t, "capability-secret") + + recorder := httptest.NewRecorder() + server.Handler().ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/hw-capabilities", nil)) + + if recorder.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401 for an unauthenticated capability probe", recorder.Code) + } +} diff --git a/internal/proxy/egress.go b/internal/proxy/egress.go index e1a9c2fa..1ae0d9d1 100644 --- a/internal/proxy/egress.go +++ b/internal/proxy/egress.go @@ -77,6 +77,16 @@ func (w *meteredResponseWriter) Flush() { } } +// Unwrap lets http.ResponseController traverse to the underlying writer. +// +// Without it the metering wrapper is a dead end: RollingDeadlineWriter's +// SetWriteDeadline call fails, it degrades to a plain pass-through, and since +// the standalone proxy runs with WriteTimeout 0 there is no server-level guard +// behind it. A client that stops reading without closing its connection would +// then block a stream write forever, holding the tracked session, the open +// file, the goroutine and the connection. +func (w *meteredResponseWriter) Unwrap() http.ResponseWriter { return w.ResponseWriter } + // meterEgress wraps stream handlers so their responses count toward the // node's measured egress bandwidth. func (s *Server) meterEgress(next http.Handler) http.Handler { diff --git a/internal/proxy/server.go b/internal/proxy/server.go index 89c2fc40..32146eef 100644 --- a/internal/proxy/server.go +++ b/internal/proxy/server.go @@ -103,6 +103,17 @@ func (s *Server) Handler() http.Handler { "Accept", "Authorization", "Content-Type", "Range", "If-Match", "If-Modified-Since", "If-None-Match", "If-Range", "If-Unmodified-Since", }, + // direct_stream_resume_v1 has the client re-request a byte range with + // If-Range against the entity tag it stored. Cross-origin JavaScript + // can only read a response header that is explicitly exposed, so + // without these the client can send the conditional request headers + // above but never learn the values to put in them — the resume + // contract silently degrades to a full restart on a proxy that is on a + // different origin than the web app, which is the normal deployment. + ExposedHeaders: []string{ + "Accept-Ranges", "Content-Encoding", "Content-Length", "Content-Range", + "ETag", "Last-Modified", + }, MaxAge: 86400, })) r.Get("/api/v1/health", s.handleHealth) @@ -125,12 +136,34 @@ func (s *Server) Handler() http.Handler { // Admin routes — bearer-auth protected. r.Group(func(r chi.Router) { r.Use(s.requireBearer) + r.Get("/hw-capabilities", s.handleHWCapabilities) r.Post("/admin/force-reload", s.handleForceReload) r.Get("/status", s.handleStatus) }) return r } +// handleHWCapabilities advertises what this proxy's ffmpeg can actually do, in +// the same shape and at the same path as a transcode node. +// +// A proxy executes recipes too: /stream/remux runs ffmpeg to convert audio or +// strip a Dolby Vision RPU. Without this endpoint the API has no way to tell +// whether the proxy it just picked can run the transformations a plan froze, so +// a pool whose proxies carry a different ffmpeg build (a rolling upgrade, a +// custom image) would fail at stream time rather than at selection time. +func (s *Server) handleHWCapabilities(w http.ResponseWriter, r *http.Request) { + ffmpegPath := "" + if cfg := s.watcher.Config(); cfg != nil { + ffmpegPath = cfg.Playback.FFmpegPath + } + info := playback.DetectHWAccelWithFFmpeg(ffmpegPath) + info.Transformations = playback.ProbeTransformationRegistryV3(r.Context(), ffmpegPath).Advertised() + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(info); err != nil { + slog.WarnContext(r.Context(), "encode proxy capabilities", "component", "proxy", "error", err) + } +} + type healthResponse struct { Status string `json:"status"` ActiveJobs int `json:"active_jobs"` @@ -185,7 +218,12 @@ func (s *Server) handleDirectPlay(w http.ResponseWriter, r *http.Request) { s.tracker.Track(r.Context(), info) defer s.tracker.Remove(r.Context(), claims.SessionID) - http.ServeFile(w, r, claims.MediaPath) + // Serve through the same path the integrated server uses rather than a bare + // http.ServeFile: direct_stream_resume_v1 requires the strong ETag that + // ServeDirectPlay sets before ServeContent (ServeFile sets none, so + // If-Range never validates and a resumed range silently restarts at 200), + // and it carries the rolling write deadline and stream metrics with it. + _ = playback.ServeDirectPlay(w, r, claims.MediaPath) } func (s *Server) handleDownloadFile(w http.ResponseWriter, r *http.Request) {