From 7958f0bbf0851113d9eac073bab23645b81c72ff Mon Sep 17 00:00:00 2001 From: Quick <31828688+Quick104@users.noreply.github.com> Date: Wed, 10 Jun 2026 17:18:18 -0400 Subject: [PATCH] feat(nodepool): node groups, per-node caps, and local transcode fallback control (#126) * feat(nodepool): node groups, per-node caps, and local transcode fallback control Group co-located transcode and proxy nodes so transcoded streams are served by a proxy on the same host/LAN instead of bouncing across the internal network (fixes #93): - New nodepool.Planner is the single selection entry point: it picks the transcode node and its group's proxy together (round-robin within the group), replacing the independent ProxyPool.Pick/TranscodePool.Acquire calls scattered across the native and jellycompat handlers, and absorbs the duplicated soft-affinity pick logic. - A group is only eligible while all of its enabled members are healthy; ungrouped nodes keep the historical behavior. - New per-node max_jobs cap (transcodes for transcode nodes, streams for proxies; NULL = unlimited), enforced via health-reported job counts plus short-lived reservations that expire once fresher health data arrives. Proxy health now reports real stream counts, including HLS sessions via idle-expiry tracking. - New playback.local_transcode_fallback setting (default on) lets admins refuse API-server transcoding when no eligible node exists. - Health checks now publish updated node copies under the pool lock instead of mutating shared structs in place, fixing a data race. - Admin UI: group + cap fields on the node form, group/cap columns, and the new fallback toggle in playback settings. Co-Authored-By: Claude Fable 5 * feat(nodepool): proxy bandwidth measurement and egress caps Proxy nodes now measure their stream egress (rolling 60s average over everything under /stream) and report it via the health endpoint. A new per-proxy max_bandwidth_kbps cap lets the planner route new streams away from saturated proxies: - Admission combines the measured egress with the estimated bitrate of the new stream (transcode target bitrate, or source bitrate for direct play/remux) so a stream is only admitted where it fits. - Recently admitted streams are bridged as bandwidth reservations for the meter window, since the rolling average only converges on a new stream's rate gradually. - A group whose proxies lack bandwidth headroom is treated as full: its transcode nodes are skipped, same as the job cap. - Admin UI: per-proxy "Max Egress Bandwidth (Mbps)" field and a live egress column; manual health checks return the measured rate. Active streams are never interrupted - the cap only gates new admissions. Co-Authored-By: Claude Fable 5 * perf(playback): trim node-mode time-to-stream-start Distributed playback paid several avoidable costs before the first frame that integrated mode doesn't have. This trims the safe ones: - Web player preconnects to the stream origin (the proxy node) as soon as /playback/start returns, overlapping DNS/TCP/TLS handshakes with the transcode dispatch instead of paying them at the first manifest fetch. - The transcode node no longer blocks its 202 on monitoring work: the Redis session-track write moves off the request path, and a replaced session's segment directory is renamed aside and deleted in the background instead of synchronously (RemoveAll of a long session can take seconds on slow disks during quality switches). - The proxy's node-facing HTTP client gets a tuned transport: a larger idle-connection pool (Go's default of 2 per host causes connection churn and TLS re-handshakes when many viewers stream through one proxy->node pair) and a response-header timeout so a hung transcode node can no longer hang client requests indefinitely. - jellycompat's remote transcode dispatch gains the same 10s timeout the native path has had; an unreachable node previously hung the compat manifest request until the OS gave up. Co-Authored-By: Claude Fable 5 --------- Co-authored-by: Claude Fable 5 --- cmd/silo/main.go | 4 +- internal/api/handlers/nodes.go | 8 +- internal/api/handlers/playback.go | 116 ++--- internal/api/handlers/playback_test.go | 2 +- internal/api/router.go | 10 +- internal/jellycompat/handlers_playback.go | 39 +- internal/jellycompat/router.go | 3 +- internal/jellycompat/server.go | 3 +- internal/jellycompat/streams.go | 25 +- internal/nodepool/health.go | 45 +- internal/nodepool/planner.go | 328 ++++++++++++ internal/nodepool/planner_test.go | 471 ++++++++++++++++++ internal/nodepool/proxy_pool.go | 9 + internal/nodepool/repository.go | 107 +++- internal/nodepool/transcode_pool.go | 29 +- internal/nodesessions/tracker.go | 68 ++- internal/proxy/egress.go | 86 ++++ internal/proxy/egress_test.go | 51 ++ internal/proxy/server.go | 77 ++- internal/transcodenode/server.go | 19 +- .../20260610130619_node_groups_and_caps.sql | 35 ++ web/src/api/types.ts | 12 + web/src/pages/AdminNodes.tsx | 105 +++- .../pages/admin-settings/PlaybackSettings.tsx | 8 + web/src/player/components/VideoPlayer.tsx | 8 + web/src/player/stream-url.ts | 29 ++ 26 files changed, 1505 insertions(+), 192 deletions(-) create mode 100644 internal/nodepool/planner.go create mode 100644 internal/nodepool/planner_test.go create mode 100644 internal/proxy/egress.go create mode 100644 internal/proxy/egress_test.go create mode 100644 migrations/sql/20260610130619_node_groups_and_caps.sql diff --git a/cmd/silo/main.go b/cmd/silo/main.go index 172c7fa2..67bb01c8 100644 --- a/cmd/silo/main.go +++ b/cmd/silo/main.go @@ -670,6 +670,7 @@ func main() { deps.ProxyPool = proxyPool deps.TranscodePool = transcodePool + deps.NodePlanner = nodepool.NewPlanner(proxyPool, transcodePool) healthChecker := nodepool.NewHealthChecker(proxyPool, transcodePool, nodeRepo) healthChecker.Start(appCtx) @@ -1897,8 +1898,7 @@ func main() { DB: deps.DB, SecretCipher: dataCipher, ClientIPResolver: ipResolver, - ProxyPool: deps.ProxyPool, - TranscodePool: deps.TranscodePool, + NodePlanner: deps.NodePlanner, JWTSecret: cfg.Auth.JWTSecret, RecWorker: recWorker, } diff --git a/internal/api/handlers/nodes.go b/internal/api/handlers/nodes.go index b96e851e..76ec84da 100644 --- a/internal/api/handlers/nodes.go +++ b/internal/api/handlers/nodes.go @@ -26,7 +26,7 @@ type NodeRepository interface { Create(ctx context.Context, input nodepool.CreateNodeInput) (*nodepool.Node, error) Update(ctx context.Context, id int, input nodepool.UpdateNodeInput) (*nodepool.Node, error) Delete(ctx context.Context, id int) error - UpdateHealth(ctx context.Context, id int, healthy bool, activeJobs int) error + UpdateHealth(ctx context.Context, id int, healthy bool, activeJobs, egressKbps int) error } // NodeListEnabled queries enabled nodes by type for pool reload. @@ -70,6 +70,7 @@ type ForceReloadResult struct { type checkNodeResult struct { Healthy bool `json:"healthy"` ActiveJobs int `json:"active_jobs"` + EgressKbps int `json:"egress_kbps"` } // HandleListNodes handles GET /admin/nodes. @@ -181,15 +182,16 @@ func (h *NodeHandler) HandleCheckNode(w http.ResponseWriter, r *http.Request) { return } - healthy, activeJobs := nodepool.CheckNode(r.Context(), node) + healthy, activeJobs, egressKbps := nodepool.CheckNode(r.Context(), node) - if err := h.repo.UpdateHealth(r.Context(), id, healthy, activeJobs); err != nil { + if err := h.repo.UpdateHealth(r.Context(), id, healthy, activeJobs, egressKbps); err != nil { slog.Error("persisting health check result", "node_id", id, "error", err) } writeJSON(w, http.StatusOK, checkNodeResult{ Healthy: healthy, ActiveJobs: activeJobs, + EgressKbps: egressKbps, }) } diff --git a/internal/api/handlers/playback.go b/internal/api/handlers/playback.go index 7149bb3a..f624101d 100644 --- a/internal/api/handlers/playback.go +++ b/internal/api/handlers/playback.go @@ -59,17 +59,6 @@ type sessionStarterWithFilesContext interface { StartSessionWithFilesContext(ctx context.Context, userID int, profileID string, effectiveFileID int, requestedFileID int, method playback.PlayMethod, transcodeAudio bool) (*playback.Session, error) } -// ProxyPicker selects a proxy node for stream routing. -type ProxyPicker interface { - Pick() *nodepool.Node -} - -// TranscodeAcquirer selects a transcode node for transcoding jobs. -type TranscodeAcquirer interface { - Acquire() *nodepool.Node - FindByURL(url string) *nodepool.Node -} - type PlaybackItemAccessChecker interface { EnsureAccessible(ctx context.Context, contentID string, filter catalog.AccessFilter) error } @@ -119,8 +108,7 @@ type PlaybackHandler struct { SessionSyncer PlaybackSessionSyncer // optional; enables immediate session sync to shared admin view EventsHub *evt.Hub MissingMarker MissingFileMarker - ProxyPool ProxyPicker // optional; enables proxy-based stream URLs - TranscodePool TranscodeAcquirer // optional; enables transcode node selection + NodePlanner nodepool.SessionPlanner // optional; enables proxy/transcode node selection JWTSecret string // needed for signing stream tokens ItemAccess PlaybackItemAccessChecker // optional; enables file authorization checks EpisodeLookup PlaybackEpisodeLookup // optional; resolves episode files to their series @@ -445,6 +433,13 @@ func playbackStreamURL(s *playback.Session) string { return fmt.Sprintf("/stream/%s", s.ID) } +func fileBitrateKbps(file *models.MediaFile) int { + if file == nil || file.Bitrate <= 0 { + return 0 + } + return file.Bitrate +} + func buildPlaybackInfo(session *playback.Session, file *models.MediaFile) *playbackInfoResult { if session == nil { return nil @@ -1372,10 +1367,14 @@ func (h *PlaybackHandler) HandleStartPlayback(w http.ResponseWriter, r *http.Req } resp.SubtitleURLs = buildSubtitleURLs(session.ID, effectiveFile, downloadedSubs) - // If proxy nodes are available, generate proxy-based stream URLs. - if h.ProxyPool != nil && h.JWTSecret != "" { - proxyNode := h.ProxyPool.Pick() - if proxyNode != nil { + // If stream nodes are available, generate proxy-based stream URLs. + // Remux and transcode both use HLS via a transcode node, so the planner + // picks the transcode node and its group's proxy together. + if h.NodePlanner != nil && h.JWTSecret != "" { + needsTranscode := session.PlayMethod == playback.PlayTranscode || session.PlayMethod == playback.PlayRemux + plan := h.NodePlanner.PlanSession(session.ID, "", needsTranscode, fileBitrateKbps(effectiveFile)) + proxyNode := plan.ProxyNode + if proxyNode != nil && (!needsTranscode || plan.TranscodeNode != nil) { tokenClaims := streamtoken.Claims{ SessionID: session.ID, PlayMethod: string(session.PlayMethod), @@ -1389,12 +1388,9 @@ func (h *PlaybackHandler) HandleStartPlayback(w http.ResponseWriter, r *http.Req tokenClaims.TranscodeAudio = session.TranscodeAudio tokenClaims.AudioTrackIndex = session.AudioTrackIndex - // Remux and transcode both use HLS via a transcode node. - if (session.PlayMethod == playback.PlayTranscode || session.PlayMethod == playback.PlayRemux) && h.TranscodePool != nil { - if tcNode := h.TranscodePool.Acquire(); tcNode != nil { - tokenClaims.TranscodeNode = tcNode.URL - _ = h.sessionMgr.SetTranscodeNodeURL(session.ID, tcNode.URL) - } + if plan.TranscodeNode != nil { + tokenClaims.TranscodeNode = plan.TranscodeNode.URL + _ = h.sessionMgr.SetTranscodeNodeURL(session.ID, plan.TranscodeNode.URL) } token, signErr := streamtoken.Sign(tokenClaims, h.JWTSecret, 24*time.Hour) @@ -1771,8 +1767,14 @@ func (h *PlaybackHandler) HandleChangeAudioTrack(w http.ResponseWriter, r *http. PlaybackInfo: buildPlaybackInfo(&updatedSession, file), } - if h.ProxyPool != nil && h.JWTSecret != "" { - if proxyNode := h.ProxyPool.Pick(); proxyNode != nil { + if h.NodePlanner != nil && h.JWTSecret != "" { + needsTranscode := updatedSession.PlayMethod == playback.PlayTranscode + estKbps := updatedSession.TargetBitrateKbps + if estKbps <= 0 { + estKbps = fileBitrateKbps(file) + } + plan := h.NodePlanner.PlanSession(sessionID, session.TranscodeNodeURL, needsTranscode, estKbps) + if proxyNode := plan.ProxyNode; proxyNode != nil && (!needsTranscode || plan.TranscodeNode != nil) { tokenClaims := streamtoken.Claims{ SessionID: sessionID, PlayMethod: string(updatedSession.PlayMethod), @@ -1780,11 +1782,9 @@ func (h *PlaybackHandler) HandleChangeAudioTrack(w http.ResponseWriter, r *http. TranscodeAudio: updatedSession.TranscodeAudio, AudioTrackIndex: req.AudioTrackIndex, } - if updatedSession.PlayMethod == playback.PlayTranscode && h.TranscodePool != nil { - if tcNode := h.TranscodePool.Acquire(); tcNode != nil { - tokenClaims.TranscodeNode = tcNode.URL - _ = h.sessionMgr.SetTranscodeNodeURL(sessionID, tcNode.URL) - } + if plan.TranscodeNode != nil { + tokenClaims.TranscodeNode = plan.TranscodeNode.URL + _ = h.sessionMgr.SetTranscodeNodeURL(sessionID, plan.TranscodeNode.URL) } if token, signErr := streamtoken.Sign(tokenClaims, h.JWTSecret, 24*time.Hour); signErr == nil { switch updatedSession.PlayMethod { @@ -2037,7 +2037,15 @@ func (h *PlaybackHandler) HandleStartTranscode(w http.ResponseWriter, r *http.Re } // Determine whether to run locally or forward to a remote transcode node. - tcNode := h.PickTranscodeNode(session.TranscodeNodeURL) + var plan nodepool.Plan + if h.NodePlanner != nil { + estKbps := req.TargetBitrateKbps + if estKbps <= 0 { + estKbps = fileBitrateKbps(file) + } + plan = h.NodePlanner.PlanSession(req.SessionID, session.TranscodeNodeURL, true, estKbps) + } + tcNode := plan.TranscodeNode if tcNode != nil { // Remote transcode: forward to the assigned node. @@ -2102,7 +2110,7 @@ func (h *PlaybackHandler) HandleStartTranscode(w http.ResponseWriter, r *http.Re effectiveHWAccel = strings.TrimSpace(nodeReq.HWAccel) } - manifestURL := h.buildProxyManifestURL(req.SessionID, session, tcNode.URL) + manifestURL := h.buildProxyManifestURL(req.SessionID, session, tcNode.URL, plan.ProxyNode) h.finalizeTranscodeStart(r, transcodeStartState{ req: req, file: file, @@ -2115,6 +2123,13 @@ func (h *PlaybackHandler) HandleStartTranscode(w http.ResponseWriter, r *http.Re } // Local transcode (integrated mode — no transcode nodes available). + // In distributed mode admins can disable this fallback so the API server + // never transcodes when no eligible node exists. + if h.NodePlanner != nil && !nodepool.LocalTranscodeFallbackAllowed(r.Context(), h.SettingsRepo) { + writeError(w, http.StatusServiceUnavailable, "no_transcode_node", + "No transcode node is available and local transcode fallback is disabled") + return + } if err := os.MkdirAll(h.TranscodeDir, 0o755); err != nil { writeError(w, http.StatusInternalServerError, "internal_error", "Failed to prepare transcode directory") return @@ -2456,40 +2471,11 @@ func (h *PlaybackHandler) closeTranscodeSession(sessionID, transcodeNodeURL stri } } -// PickTranscodeNode selects a transcode node with soft affinity. -// If currentURL is set and that node is healthy, reuse it unless another node -// has significantly fewer active jobs (2+ difference). -func (h *PlaybackHandler) PickTranscodeNode(currentURL string) *nodepool.Node { - if h.TranscodePool == nil { - return nil - } - best := h.TranscodePool.Acquire() - if best == nil { - return nil - } - if currentURL == "" || best.URL == currentURL { - return best - } - // Reuse current node unless the best alternative has significantly fewer jobs. - current := h.TranscodePool.FindByURL(currentURL) - if current != nil && current.Healthy && current.Enabled && - best.ActiveJobs+2 <= current.ActiveJobs { - return best // switch to less-loaded node - } - if current != nil && current.Healthy && current.Enabled { - return current // soft affinity — stay on current - } - return best // current node is gone or unhealthy -} - -// buildProxyManifestURL signs a stream token and builds the proxy-based manifest URL. -func (h *PlaybackHandler) buildProxyManifestURL(sessionID string, session *playback.Session, transcodeNodeURL string) string { - if h.ProxyPool == nil { - return fmt.Sprintf("/playback/transcode/%s/master.m3u8", sessionID) - } - proxyNode := h.ProxyPool.Pick() +// buildProxyManifestURL signs a stream token and builds the proxy-based +// manifest URL. proxyNode is the planner's pick for this session; when nil +// the URL falls back to the API-local path. +func (h *PlaybackHandler) buildProxyManifestURL(sessionID string, session *playback.Session, transcodeNodeURL string, proxyNode *nodepool.Node) string { if proxyNode == nil { - // No proxy — fall back to API-local path. return fmt.Sprintf("/playback/transcode/%s/master.m3u8", sessionID) } diff --git a/internal/api/handlers/playback_test.go b/internal/api/handlers/playback_test.go index f711869d..52d96dcc 100644 --- a/internal/api/handlers/playback_test.go +++ b/internal/api/handlers/playback_test.go @@ -1117,7 +1117,7 @@ func TestHandleStartTranscode_PreservesRecomputedBaseMethodAfterFallback(t *test Healthy: true, ActiveJobs: 0, }}) - handler.TranscodePool = pool + handler.NodePlanner = nodepool.NewPlanner(nodepool.NewProxyPool(), pool) transcodeReq := httptest.NewRequest( "POST", diff --git a/internal/api/router.go b/internal/api/router.go index a520e195..01cb24cb 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -98,6 +98,7 @@ type Dependencies struct { NodeRepo *nodepool.Repository // stream node repository (may be nil) ProxyPool *nodepool.ProxyPool // proxy node pool (may be nil) TranscodePool *nodepool.TranscodePool // transcode node pool (may be nil) + NodePlanner *nodepool.Planner // group/cap-aware node selection (may be nil) SessionSyncer handlers.PlaybackSessionSyncer // optional; immediate playback session sync trigger EventBus cache.EventBus AdminStatsProvider handlers.AdminStatsSource @@ -654,12 +655,9 @@ func NewRouter(deps Dependencies) chi.Router { } } - // Wire optional proxy/transcode pools and JWT secret for node-aware stream URLs. - if deps.ProxyPool != nil { - playbackHandler.ProxyPool = deps.ProxyPool - } - if deps.TranscodePool != nil { - playbackHandler.TranscodePool = deps.TranscodePool + // Wire the optional node planner and JWT secret for node-aware stream URLs. + if deps.NodePlanner != nil { + playbackHandler.NodePlanner = deps.NodePlanner } if deps.Config != nil && deps.Config.Auth.JWTSecret != "" { playbackHandler.JWTSecret = deps.Config.Auth.JWTSecret diff --git a/internal/jellycompat/handlers_playback.go b/internal/jellycompat/handlers_playback.go index aa90ebe2..a1a8dc65 100644 --- a/internal/jellycompat/handlers_playback.go +++ b/internal/jellycompat/handlers_playback.go @@ -119,8 +119,7 @@ type PlaybackHandler struct { sessionMgr SessionManagerInterface fileResolver FilePathResolver storeProvider userstore.UserStoreProvider - ProxyPool *nodepool.ProxyPool - TranscodePool *nodepool.TranscodePool + NodePlanner nodepool.SessionPlanner JWTSecret string profileStaler profileStaler profileRefreshRequester profileRefreshRequester @@ -215,28 +214,8 @@ func NewPlaybackHandler( return h } -func (h *PlaybackHandler) pickTranscodeNode(currentURL string) *nodepool.Node { - if h.TranscodePool == nil { - return nil - } - best := h.TranscodePool.Acquire() - if best == nil { - return nil - } - if currentURL == "" || best.URL == currentURL { - return best - } - current := h.TranscodePool.FindByURL(currentURL) - if current != nil && current.Healthy && current.Enabled && - best.ActiveJobs+2 <= current.ActiveJobs { - return best - } - if current != nil && current.Healthy && current.Enabled { - return current - } - return best -} - +// buildProxyRedirectURL signs a stream token and builds the redirect URL for +// the given proxy node (the planner's pick for this session). func (h *PlaybackHandler) buildProxyRedirectURL( playSessionID string, upstreamSessionID string, @@ -245,12 +224,9 @@ func (h *PlaybackHandler) buildProxyRedirectURL( source PlaybackMediaSource, transcodeNodeURL string, seekSeconds float64, + proxyNode *nodepool.Node, ) (string, error) { - if h.ProxyPool == nil || h.JWTSecret == "" { - return "", fmt.Errorf("proxy transport unavailable") - } - proxyNode := h.ProxyPool.Pick() - if proxyNode == nil { + if proxyNode == nil || h.JWTSecret == "" { return "", fmt.Errorf("proxy transport unavailable") } @@ -352,6 +328,11 @@ func (h *PlaybackHandler) startRemoteTranscode( if err != nil { return fmt.Errorf("marshal transcode request: %w", err) } + // Bound the dispatch like the native path does (playback.go) — without + // this, an unreachable transcode node hangs the compat manifest request + // until the OS gives up on the connection. + ctx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, transcodeNodeURL+"/transcode/start", strings.NewReader(string(body))) if err != nil { return fmt.Errorf("build transcode request: %w", err) diff --git a/internal/jellycompat/router.go b/internal/jellycompat/router.go index 8eab5064..1326ef63 100644 --- a/internal/jellycompat/router.go +++ b/internal/jellycompat/router.go @@ -96,8 +96,7 @@ func NewRouter(deps Dependencies) chi.Router { if deps.DB != nil { playbackHandler.profileStaler = recommendations.NewRepo(deps.DB) } - playbackHandler.ProxyPool = deps.ProxyPool - playbackHandler.TranscodePool = deps.TranscodePool + playbackHandler.NodePlanner = deps.NodePlanner playbackHandler.JWTSecret = deps.JWTSecret playbackHandler.profileRefreshRequester = deps.RecWorker playbackHandler.SettingsRepo = deps.SettingsRepo diff --git a/internal/jellycompat/server.go b/internal/jellycompat/server.go index ef6cf956..ea54cbd3 100644 --- a/internal/jellycompat/server.go +++ b/internal/jellycompat/server.go @@ -70,8 +70,7 @@ type Dependencies struct { FileResolver FilePathResolver UserStoreProvider userstore.UserStoreProvider AccessFilterFn AccessFilterResolver - ProxyPool *nodepool.ProxyPool - TranscodePool *nodepool.TranscodePool + NodePlanner nodepool.SessionPlanner JWTSecret string Recommender recommendations.Recommender RecWorker *recommendations.Worker diff --git a/internal/jellycompat/streams.go b/internal/jellycompat/streams.go index 9df8347e..f7e6677d 100644 --- a/internal/jellycompat/streams.go +++ b/internal/jellycompat/streams.go @@ -22,6 +22,7 @@ import ( "github.com/go-chi/chi/v5" "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/subtitles" ) @@ -96,9 +97,12 @@ func (h *PlaybackHandler) HandleVideoStream(w http.ResponseWriter, r *http.Reque if d := float64(source.Version.Duration); d > 0 && seekSeconds > d { seekSeconds = d } - if redirectURL, redirectErr := h.buildProxyRedirectURL(playSession.ID, playSession.UpstreamSessionID, method, file, *source, "", seekSeconds); redirectErr == nil { - http.Redirect(w, r, redirectURL, http.StatusTemporaryRedirect) - return + if h.NodePlanner != nil && h.JWTSecret != "" { + plan := h.NodePlanner.PlanSession(playSession.UpstreamSessionID, "", false, source.Version.Bitrate) + if redirectURL, redirectErr := h.buildProxyRedirectURL(playSession.ID, playSession.UpstreamSessionID, method, file, *source, "", seekSeconds, plan.ProxyNode); redirectErr == nil { + http.Redirect(w, r, redirectURL, http.StatusTemporaryRedirect) + return + } } switch method { @@ -191,7 +195,7 @@ func (h *PlaybackHandler) HandleMasterManifest(w http.ResponseWriter, r *http.Re } var err error - if h.ProxyPool != nil && h.JWTSecret != "" && h.TranscodePool != nil { + if h.NodePlanner != nil && h.JWTSecret != "" { playSession, err = h.ensureUpstreamPlayback(r.Context(), session, playSession.ID, *source, "transcode") if err != nil { writeCompatUpstreamError(w, err) @@ -199,7 +203,8 @@ func (h *PlaybackHandler) HandleMasterManifest(w http.ResponseWriter, r *http.Re } upstreamSession, upstreamErr := h.sessionMgr.GetSession(playSession.UpstreamSessionID) if upstreamErr == nil { - if tcNode := h.pickTranscodeNode(upstreamSession.TranscodeNodeURL); tcNode != nil { + plan := h.NodePlanner.PlanSession(playSession.UpstreamSessionID, upstreamSession.TranscodeNodeURL, true, source.Version.Bitrate) + if tcNode := plan.TranscodeNode; tcNode != nil { if h.fileResolver == nil { writeError(w, http.StatusInternalServerError, "ServerError", "File resolver not available") return @@ -221,7 +226,7 @@ func (h *PlaybackHandler) HandleMasterManifest(w http.ResponseWriter, r *http.Re writeError(w, http.StatusBadGateway, "TranscodeStartFailed", "Transcode node rejected the request") return } - redirectURL, redirectErr := h.buildProxyRedirectURL(playSession.ID, playSession.UpstreamSessionID, string(playback.PlayTranscode), file, *source, tcNode.URL, 0) + redirectURL, redirectErr := h.buildProxyRedirectURL(playSession.ID, playSession.UpstreamSessionID, string(playback.PlayTranscode), file, *source, tcNode.URL, 0, plan.ProxyNode) if redirectErr != nil { writeError(w, http.StatusInternalServerError, "ServerError", "Failed to sign proxy stream URL") return @@ -232,6 +237,14 @@ func (h *PlaybackHandler) HandleMasterManifest(w http.ResponseWriter, r *http.Re } } + // In distributed mode admins can disable the local fallback so the API + // server never transcodes when no eligible node exists. + if h.NodePlanner != nil && !nodepool.LocalTranscodeFallbackAllowed(r.Context(), h.SettingsRepo) { + writeError(w, http.StatusServiceUnavailable, "NoTranscodeNode", + "No transcode node is available and local transcode fallback is disabled") + return + } + // Ensure the transcode process is running. _, err = h.ensureTranscodeManifest(r.Context(), session, playSession.ID, *source) if err != nil { diff --git a/internal/nodepool/health.go b/internal/nodepool/health.go index 333ea1fc..e1f9d027 100644 --- a/internal/nodepool/health.go +++ b/internal/nodepool/health.go @@ -13,36 +13,37 @@ import ( type healthResponse struct { Status string `json:"status"` ActiveJobs int `json:"active_jobs"` + EgressKbps int `json:"egress_kbps"` } -// CheckNode pings a node's /health endpoint and returns its health status -// and active job count. -func CheckNode(ctx context.Context, n *Node) (healthy bool, activeJobs int) { +// CheckNode pings a node's /health endpoint and returns its health status, +// active job count, and reported egress bandwidth. +func CheckNode(ctx context.Context, n *Node) (healthy bool, activeJobs, egressKbps int) { client := &http.Client{Timeout: 5 * time.Second} ctx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() req, err := http.NewRequestWithContext(ctx, http.MethodGet, n.URL+"/api/v1/health", nil) if err != nil { - return false, 0 + return false, 0, 0 } resp, err := client.Do(req) if err != nil { - return false, 0 + return false, 0, 0 } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - return false, 0 + return false, 0, 0 } var hr healthResponse if err := json.NewDecoder(resp.Body).Decode(&hr); err != nil { - return false, 0 + return false, 0, 0 } - return true, hr.ActiveJobs + return true, hr.ActiveJobs, hr.EgressKbps } // HealthChecker runs periodic health checks on all nodes in both pools, @@ -82,33 +83,33 @@ func (hc *HealthChecker) Start(ctx context.Context) { } func (hc *HealthChecker) checkAll(ctx context.Context) { - var allNodes []*Node - allNodes = append(allNodes, hc.proxyPool.Nodes()...) - allNodes = append(allNodes, hc.transcodePool.Nodes()...) - var wg sync.WaitGroup - for _, n := range allNodes { + check := func(n *Node, applyHealth func(int, bool, int, int, time.Time)) { wg.Go(func() { - healthy, activeJobs := CheckNode(ctx, n) + healthy, activeJobs, egressKbps := CheckNode(ctx, n) - wasHealthy := n.Healthy - n.Healthy = healthy - n.ActiveJobs = activeJobs - now := time.Now() - n.LastHealthCheck = &now + // Publish the result through the pool lock so readers never see + // a Node struct mutated in place (the pool swaps in a copy). + applyHealth(n.ID, healthy, activeJobs, egressKbps, time.Now()) - if wasHealthy && !healthy { + if n.Healthy && !healthy { slog.Warn("stream node unhealthy", "id", n.ID, "name", n.Name, "url", n.URL) - } else if !wasHealthy && healthy { + } else if !n.Healthy && healthy { slog.Info("stream node recovered", "id", n.ID, "name", n.Name, "url", n.URL) } if hc.repo != nil { - if err := hc.repo.UpdateHealth(ctx, n.ID, healthy, activeJobs); err != nil { + if err := hc.repo.UpdateHealth(ctx, n.ID, healthy, activeJobs, egressKbps); err != nil { slog.Error("failed to persist node health", "id", n.ID, "error", err) } } }) } + for _, n := range hc.proxyPool.Nodes() { + check(n, hc.proxyPool.ApplyHealth) + } + for _, n := range hc.transcodePool.Nodes() { + check(n, hc.transcodePool.ApplyHealth) + } wg.Wait() } diff --git a/internal/nodepool/planner.go b/internal/nodepool/planner.go new file mode 100644 index 00000000..9f058e56 --- /dev/null +++ b/internal/nodepool/planner.go @@ -0,0 +1,328 @@ +package nodepool + +import ( + "context" + "sync" + "time" +) + +// Plan is the result of a node selection for one playback session. +// Either field may be nil when no suitable node exists. +type Plan struct { + TranscodeNode *Node + ProxyNode *Node +} + +// SessionPlanner selects transcode and proxy nodes for playback sessions. +// Implemented by *Planner; defined as an interface so handlers can be tested +// without a real pool. +type SessionPlanner interface { + PlanSession(sessionID, currentTranscodeURL string, needsTranscode bool, estBitrateKbps int) Plan +} + +// reservation bridges the gap between assigning a session to a node and the +// node's health reports reflecting that session. +// +// The job count stops counting toward a node's effective load as soon as the +// node delivers a health report newer than the reservation (the node's own +// count then includes the session), or after maxReservationAge as a safety +// net. The bandwidth estimate instead counts for a fixed bandwidthBridgeAge +// regardless of health freshness: a proxy's measured egress is a rolling +// average that only converges on the new stream's rate gradually, so an +// early health report would otherwise drop the estimate before the meter +// reflects it. +type reservation struct { + transcodeURL string + proxyURL string + kbps int // estimated stream bitrate, counted against the proxy + createdAt time.Time +} + +const ( + maxReservationAge = 90 * time.Second + bandwidthBridgeAge = 60 * time.Second // matches the proxy egress meter window +) + +// Planner makes group- and capacity-aware node selections on top of the +// existing pools. +// +// Grouping: nodes sharing a group label are co-located (same host/LAN). A +// group is eligible only while every enabled member is healthy. A transcode +// node from group G is always paired with a proxy from G so transcoded bytes +// never cross the LAN twice (round-robin when G has several proxies). +// Ungrouped nodes keep the historical behavior: least-connections transcode +// selection and global round-robin proxy selection. +// +// Capacity: a node with MaxJobs set is skipped once its effective load +// (health-reported active jobs plus unexpired reservations) reaches the cap. +// A proxy with MaxBandwidthKbps set is skipped once its measured egress plus +// the estimated bitrate of recently admitted streams would exceed the cap. +type Planner struct { + proxies *ProxyPool + transcodes *TranscodePool + + mu sync.Mutex + rr map[string]int // per-group round-robin cursor; "" = global + reserved map[string]*reservation // keyed by playback session ID + now func() time.Time // overridable for tests +} + +// NewPlanner creates a planner over the given pools. +func NewPlanner(proxies *ProxyPool, transcodes *TranscodePool) *Planner { + return &Planner{ + proxies: proxies, + transcodes: transcodes, + rr: make(map[string]int), + reserved: make(map[string]*reservation), + now: time.Now, + } +} + +// PlanSession picks the nodes serving one playback session. +// +// When needsTranscode is true it selects a transcode node (soft affinity to +// currentTranscodeURL, matching the historical quality-switch behavior) and a +// proxy from the same group. When false (direct play / proxy-side remux) it +// selects only a proxy. estBitrateKbps is the expected stream bitrate (target +// bitrate for transcodes, source bitrate otherwise; 0 = unknown), used for +// bandwidth-cap admission. Re-planning the same session replaces its previous +// reservation, so quality switches don't double-count. +func (p *Planner) PlanSession(sessionID, currentTranscodeURL string, needsTranscode bool, estBitrateKbps int) Plan { + if p == nil { + return Plan{} + } + p.mu.Lock() + defer p.mu.Unlock() + + now := p.now() + p.pruneReservations(now) + // Drop this session's own reservation before computing loads so a + // re-plan doesn't count the session against its current node. + delete(p.reserved, sessionID) + + if estBitrateKbps < 0 { + estBitrateKbps = 0 + } + proxies := p.proxies.Nodes() + transcodes := p.transcodes.Nodes() + groupHealthy := groupHealth(proxies, transcodes) + + var plan Plan + if needsTranscode { + 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 { + plan.ProxyNode = p.pickProxy(proxies, groupHealthy, nil, estBitrateKbps, now) + } + + if plan.TranscodeNode != nil || plan.ProxyNode != nil { + res := &reservation{createdAt: now} + if plan.TranscodeNode != nil { + res.transcodeURL = plan.TranscodeNode.URL + } + if plan.ProxyNode != nil { + res.proxyURL = plan.ProxyNode.URL + res.kbps = estBitrateKbps + } + p.reserved[sessionID] = res + } + return plan +} + +// groupHealth reports, for every group label present in either pool, whether +// all of its enabled members are healthy. Pools only hold enabled nodes, so +// disabled nodes never count against a group. +func groupHealth(proxies, transcodes []*Node) map[string]bool { + health := make(map[string]bool) + for _, nodes := range [][]*Node{proxies, transcodes} { + for _, n := range nodes { + if n.Group == nil { + continue + } + healthy, seen := health[*n.Group] + if !seen { + healthy = true + } + health[*n.Group] = healthy && n.Healthy + } + } + return health +} + +// pickTranscode returns the eligible transcode node with the fewest effective +// jobs, keeping the session on currentURL unless a candidate has at least two +// fewer jobs (the historical soft-affinity rule). +func (p *Planner) pickTranscode(transcodes, proxies []*Node, groupHealthy map[string]bool, currentURL string, estKbps int, now time.Time) *Node { + var best, current *Node + for _, n := range transcodes { + if !p.transcodeEligible(n, proxies, groupHealthy, estKbps, now) { + continue + } + if n.URL == currentURL { + current = n + } + if best == nil || p.effectiveJobs(n, now) < p.effectiveJobs(best, now) { + best = n + } + } + if current == nil || best == nil || current == best { + return best + } + if p.effectiveJobs(best, now)+2 <= p.effectiveJobs(current, now) { + return best + } + return current +} + +// transcodeEligible reports whether a transcode node may take a new session: +// it must be healthy and under cap, and a grouped node additionally requires +// its whole group healthy and — when the group contains proxies — at least +// one of them with job and bandwidth headroom (a group's capacity is bounded +// by its proxies). +func (p *Planner) transcodeEligible(n *Node, proxies []*Node, groupHealthy map[string]bool, estKbps int, now time.Time) bool { + if !n.Healthy || !n.Enabled || !p.underCap(n, now) { + return false + } + if n.Group == nil { + return true + } + if !groupHealthy[*n.Group] { + return false + } + groupHasProxy := false + for _, proxy := range proxies { + if proxy.Group == nil || *proxy.Group != *n.Group { + continue + } + groupHasProxy = true + if proxy.Healthy && proxy.Enabled && p.underCap(proxy, now) && p.underBandwidthCap(proxy, estKbps, now) { + return true + } + } + // A group without proxies pins nothing; its transcode nodes fall back + // to global proxy selection. + return !groupHasProxy +} + +// pickProxy selects a proxy round-robin. When group is set and contains +// proxies, only that group's proxies are considered (keeping transcoded +// traffic on the group's LAN); otherwise any healthy proxy qualifies. +func (p *Planner) pickProxy(proxies []*Node, groupHealthy map[string]bool, group *string, estKbps int, now time.Time) *Node { + var candidates []*Node + rrKey := "" + if group != nil { + for _, n := range proxies { + if n.Group != nil && *n.Group == *group && n.Healthy && n.Enabled && + groupHealthy[*group] && p.underCap(n, now) && p.underBandwidthCap(n, estKbps, now) { + candidates = append(candidates, n) + } + } + rrKey = *group + } + if len(candidates) == 0 { + if group != nil { + groupHasProxy := false + for _, n := range proxies { + if n.Group != nil && *n.Group == *group { + groupHasProxy = true + break + } + } + // Strict pinning: a group that has proxies but none usable + // never spills onto other LANs. (Unreachable from PlanSession + // for transcode plans — transcodeEligible already requires a + // usable group proxy — but enforced here for safety.) + if groupHasProxy { + return nil + } + } + rrKey = "" + for _, n := range proxies { + if n.Healthy && n.Enabled && p.underCap(n, now) && p.underBandwidthCap(n, estKbps, now) { + candidates = append(candidates, n) + } + } + } + if len(candidates) == 0 { + return nil + } + idx := p.rr[rrKey] % len(candidates) + p.rr[rrKey]++ + return candidates[idx] +} + +// underCap reports whether a node can take one more job. +func (p *Planner) underCap(n *Node, now time.Time) bool { + return n.MaxJobs == nil || p.effectiveJobs(n, now) < *n.MaxJobs +} + +// underBandwidthCap reports whether a proxy has bandwidth headroom for a +// stream of the given estimated bitrate. With an unknown bitrate (0) the +// node only needs to be below its cap. +func (p *Planner) underBandwidthCap(n *Node, estKbps int, now time.Time) bool { + if n.MaxBandwidthKbps == nil { + return true + } + egress := p.effectiveEgressKbps(n, now) + if estKbps <= 0 { + return egress < *n.MaxBandwidthKbps + } + return egress+estKbps <= *n.MaxBandwidthKbps +} + +// effectiveEgressKbps is the node's health-reported egress plus the estimated +// bitrate of streams admitted within the bandwidth bridge window, which the +// rolling egress average doesn't fully reflect yet. +func (p *Planner) effectiveEgressKbps(n *Node, now time.Time) int { + egress := n.EgressKbps + for _, res := range p.reserved { + if res.proxyURL != n.URL || res.kbps <= 0 { + continue + } + if now.Sub(res.createdAt) >= bandwidthBridgeAge { + continue + } + egress += res.kbps + } + return egress +} + +// effectiveJobs is the node's health-reported job count plus reservations the +// health checker hasn't had a chance to observe yet. +func (p *Planner) effectiveJobs(n *Node, now time.Time) int { + jobs := n.ActiveJobs + for _, res := range p.reserved { + if res.transcodeURL != n.URL && res.proxyURL != n.URL { + continue + } + if n.LastHealthCheck != nil && n.LastHealthCheck.After(res.createdAt) { + continue // a newer health report already reflects this session + } + jobs++ + } + return jobs +} + +func (p *Planner) pruneReservations(now time.Time) { + for id, res := range p.reserved { + if now.Sub(res.createdAt) > maxReservationAge { + delete(p.reserved, id) + } + } +} + +// LocalTranscodeFallbackAllowed reports whether the API server may transcode +// locally when no eligible transcode node exists, based on the +// playback.local_transcode_fallback setting. Defaults to allowed so +// deployments without the setting keep the historical behavior. +func LocalTranscodeFallbackAllowed(ctx context.Context, settings interface { + Get(ctx context.Context, key string) (string, error) +}) bool { + if settings == nil { + return true + } + v, _ := settings.Get(ctx, "playback.local_transcode_fallback") + return v != "false" +} diff --git a/internal/nodepool/planner_test.go b/internal/nodepool/planner_test.go new file mode 100644 index 00000000..f2f29b79 --- /dev/null +++ b/internal/nodepool/planner_test.go @@ -0,0 +1,471 @@ +package nodepool + +import ( + "testing" + "time" +) + +func strPtr(s string) *string { return &s } +func intPtr(i int) *int { return &i } + +type plannerFixture struct { + planner *Planner + proxies *ProxyPool + transcodes *TranscodePool + now time.Time +} + +func newFixture(proxies, transcodes []*Node) *plannerFixture { + pp := NewProxyPool() + pp.SetNodes(proxies) + tp := NewTranscodePool() + tp.SetNodes(transcodes) + f := &plannerFixture{ + planner: NewPlanner(pp, tp), + proxies: pp, + transcodes: tp, + now: time.Date(2026, 6, 10, 12, 0, 0, 0, time.UTC), + } + f.planner.now = func() time.Time { return f.now } + return f +} + +func proxyNode(id int, url string, group *string) *Node { + return &Node{ID: id, Name: url, Type: NodeTypeProxy, URL: url, Enabled: true, Healthy: true, Group: group} +} + +func transcodeNode(id int, url string, group *string, activeJobs int) *Node { + return &Node{ID: id, Name: url, Type: NodeTypeTranscode, URL: url, Enabled: true, Healthy: true, Group: group, ActiveJobs: activeJobs} +} + +func TestPlanTranscodePairsProxyFromSameGroup(t *testing.T) { + f := newFixture( + []*Node{ + proxyNode(1, "http://proxy-a", strPtr("rack-a")), + proxyNode(2, "http://proxy-b", strPtr("rack-b")), + }, + []*Node{ + transcodeNode(3, "http://tc-a", strPtr("rack-a"), 5), + transcodeNode(4, "http://tc-b", strPtr("rack-b"), 0), + }, + ) + + plan := f.planner.PlanSession("s1", "", true, 0) + if plan.TranscodeNode == nil || plan.TranscodeNode.URL != "http://tc-b" { + t.Fatalf("expected least-loaded tc-b, got %+v", plan.TranscodeNode) + } + if plan.ProxyNode == nil || plan.ProxyNode.URL != "http://proxy-b" { + t.Fatalf("expected same-group proxy-b, got %+v", plan.ProxyNode) + } +} + +func TestDegradedGroupExcludesItsTranscodeNodes(t *testing.T) { + unhealthyProxy := proxyNode(1, "http://proxy-a", strPtr("rack-a")) + unhealthyProxy.Healthy = false + f := newFixture( + []*Node{ + unhealthyProxy, + proxyNode(2, "http://proxy-b", strPtr("rack-b")), + }, + []*Node{ + transcodeNode(3, "http://tc-a", strPtr("rack-a"), 0), // idle but group degraded + transcodeNode(4, "http://tc-b", strPtr("rack-b"), 9), + }, + ) + + plan := f.planner.PlanSession("s1", "", true, 0) + if plan.TranscodeNode == nil || plan.TranscodeNode.URL != "http://tc-b" { + t.Fatalf("expected tc-b (rack-a degraded), got %+v", plan.TranscodeNode) + } + if plan.ProxyNode == nil || plan.ProxyNode.URL != "http://proxy-b" { + t.Fatalf("expected proxy-b, got %+v", plan.ProxyNode) + } +} + +func TestUnhealthyTranscodeMemberDegradesGroup(t *testing.T) { + deadTC := transcodeNode(5, "http://tc-a2", strPtr("rack-a"), 0) + deadTC.Healthy = false + f := newFixture( + []*Node{proxyNode(1, "http://proxy-a", strPtr("rack-a"))}, + []*Node{ + transcodeNode(3, "http://tc-a1", strPtr("rack-a"), 0), + deadTC, + }, + ) + + // All enabled members of a group must be healthy for the group to be + // eligible — even the healthy sibling is excluded. + plan := f.planner.PlanSession("s1", "", true, 0) + if plan.TranscodeNode != nil { + t.Fatalf("expected no transcode node, got %+v", plan.TranscodeNode) + } +} + +func TestUngroupedNodesKeepLegacyBehavior(t *testing.T) { + f := newFixture( + []*Node{ + proxyNode(1, "http://proxy-1", nil), + proxyNode(2, "http://proxy-2", nil), + }, + []*Node{ + transcodeNode(3, "http://tc-1", nil, 2), + transcodeNode(4, "http://tc-2", nil, 1), + }, + ) + + plan := f.planner.PlanSession("s1", "", true, 0) + if plan.TranscodeNode == nil || plan.TranscodeNode.URL != "http://tc-2" { + t.Fatalf("expected least-connections tc-2, got %+v", plan.TranscodeNode) + } + if plan.ProxyNode == nil { + t.Fatal("expected a proxy node") + } + + // Round-robin across both proxies for subsequent sessions. + first := plan.ProxyNode.URL + second := f.planner.PlanSession("s2", "", true, 0).ProxyNode.URL + if first == second { + t.Fatalf("expected round-robin to alternate proxies, got %s twice", first) + } +} + +func TestGroupWithoutProxiesFallsBackToGlobalProxy(t *testing.T) { + f := newFixture( + []*Node{proxyNode(1, "http://proxy-1", nil)}, + []*Node{transcodeNode(2, "http://tc-a", strPtr("rack-a"), 0)}, + ) + + plan := f.planner.PlanSession("s1", "", true, 0) + if plan.TranscodeNode == nil || plan.TranscodeNode.URL != "http://tc-a" { + t.Fatalf("expected tc-a, got %+v", plan.TranscodeNode) + } + if plan.ProxyNode == nil || plan.ProxyNode.URL != "http://proxy-1" { + t.Fatalf("expected global proxy fallback, got %+v", plan.ProxyNode) + } +} + +func TestSoftAffinityKeepsCurrentNode(t *testing.T) { + f := newFixture( + []*Node{proxyNode(1, "http://proxy-1", nil)}, + []*Node{ + transcodeNode(2, "http://tc-1", nil, 2), + transcodeNode(3, "http://tc-2", nil, 1), + }, + ) + + // Difference of 1 job: stay on current. + plan := f.planner.PlanSession("s1", "http://tc-1", true, 0) + if plan.TranscodeNode == nil || plan.TranscodeNode.URL != "http://tc-1" { + t.Fatalf("expected soft affinity to keep tc-1, got %+v", plan.TranscodeNode) + } + + // Difference of 2+: switch to the less-loaded node. + f.transcodes.Nodes()[0].ActiveJobs = 4 + plan = f.planner.PlanSession("s1", "http://tc-1", true, 0) + if plan.TranscodeNode == nil || plan.TranscodeNode.URL != "http://tc-2" { + t.Fatalf("expected switch to tc-2, got %+v", plan.TranscodeNode) + } +} + +func TestTranscodeCapSkipsFullNode(t *testing.T) { + capped := transcodeNode(2, "http://tc-1", nil, 3) + capped.MaxJobs = intPtr(3) + f := newFixture( + []*Node{proxyNode(1, "http://proxy-1", nil)}, + []*Node{ + capped, + transcodeNode(3, "http://tc-2", nil, 5), + }, + ) + + plan := f.planner.PlanSession("s1", "", true, 0) + if plan.TranscodeNode == nil || plan.TranscodeNode.URL != "http://tc-2" { + t.Fatalf("expected at-cap tc-1 to be skipped, got %+v", plan.TranscodeNode) + } + + // All nodes at cap: no transcode node. + f.transcodes.Nodes()[1].MaxJobs = intPtr(5) + plan = f.planner.PlanSession("s2", "", true, 0) + if plan.TranscodeNode != nil { + t.Fatalf("expected no eligible node, got %+v", plan.TranscodeNode) + } +} + +func TestProxyCapSkipsFullProxy(t *testing.T) { + capped := proxyNode(1, "http://proxy-1", nil) + capped.MaxJobs = intPtr(2) + capped.ActiveJobs = 2 + f := newFixture( + []*Node{capped, proxyNode(2, "http://proxy-2", nil)}, + []*Node{}, + ) + + for i := 0; i < 3; i++ { + plan := f.planner.PlanSession("s", "", false, 0) + if plan.ProxyNode == nil || plan.ProxyNode.URL != "http://proxy-2" { + t.Fatalf("expected proxy-2 (proxy-1 at cap), got %+v", plan.ProxyNode) + } + } +} + +func TestGroupAtProxyCapacityExcludesGroupTranscode(t *testing.T) { + groupProxy := proxyNode(1, "http://proxy-a", strPtr("rack-a")) + groupProxy.MaxJobs = intPtr(1) + groupProxy.ActiveJobs = 1 + f := newFixture( + []*Node{groupProxy, proxyNode(2, "http://proxy-1", nil)}, + []*Node{ + transcodeNode(3, "http://tc-a", strPtr("rack-a"), 0), + transcodeNode(4, "http://tc-1", nil, 7), + }, + ) + + // rack-a's only proxy is full, so its transcode node must not be used — + // streams pinned to rack-a would have nowhere to go. + plan := f.planner.PlanSession("s1", "", true, 0) + if plan.TranscodeNode == nil || plan.TranscodeNode.URL != "http://tc-1" { + t.Fatalf("expected ungrouped tc-1, got %+v", plan.TranscodeNode) + } + if plan.ProxyNode == nil || plan.ProxyNode.URL != "http://proxy-1" { + t.Fatalf("expected ungrouped proxy-1, got %+v", plan.ProxyNode) + } +} + +func TestGroupProxyReservationsGateGroupCapacity(t *testing.T) { + groupProxy := proxyNode(1, "http://proxy-a", strPtr("rack-a")) + groupProxy.MaxJobs = intPtr(1) + f := newFixture( + []*Node{groupProxy}, + []*Node{transcodeNode(2, "http://tc-a", strPtr("rack-a"), 0)}, + ) + + // The first session reserves the group's only proxy slot. + plan := f.planner.PlanSession("s1", "", true, 0) + if plan.TranscodeNode == nil || plan.ProxyNode == nil { + t.Fatalf("first session should get both nodes, got %+v", plan) + } + + // With the group's proxy fully reserved, its transcode node is + // ineligible too — streams pinned to the group would have nowhere to go. + plan = f.planner.PlanSession("s2", "", true, 0) + if plan.TranscodeNode != nil || plan.ProxyNode != nil { + t.Fatalf("second session should be rejected, got %+v", plan) + } +} + +func TestReservationsCountTowardCaps(t *testing.T) { + capped := transcodeNode(2, "http://tc-1", nil, 0) + capped.MaxJobs = intPtr(2) + lastCheck := time.Date(2026, 6, 10, 11, 59, 0, 0, time.UTC) + capped.LastHealthCheck = &lastCheck + f := newFixture( + []*Node{proxyNode(1, "http://proxy-1", nil)}, + []*Node{capped}, + ) + + // Two sessions fill the cap via reservations before any health refresh. + if f.planner.PlanSession("s1", "", true, 0).TranscodeNode == nil { + t.Fatal("first session should be admitted") + } + if f.planner.PlanSession("s2", "", true, 0).TranscodeNode == nil { + t.Fatal("second session should be admitted") + } + if got := f.planner.PlanSession("s3", "", true, 0).TranscodeNode; got != nil { + t.Fatalf("third session should be rejected, got %+v", got) + } + + // Re-planning an admitted session must not double-count it. + if f.planner.PlanSession("s2", "http://tc-1", true, 0).TranscodeNode == nil { + t.Fatal("re-plan of s2 should be admitted") + } + + // A health report newer than the reservations becomes authoritative: + // the node now says 1 job, so one slot is free again. + newer := f.now.Add(10 * time.Second) + capped.LastHealthCheck = &newer + capped.ActiveJobs = 1 + f.now = f.now.Add(20 * time.Second) + if f.planner.PlanSession("s4", "", true, 0).TranscodeNode == nil { + t.Fatal("session should be admitted after fresh health report") + } +} + +func TestReservationsExpire(t *testing.T) { + capped := transcodeNode(2, "http://tc-1", nil, 0) + capped.MaxJobs = intPtr(1) + f := newFixture( + []*Node{proxyNode(1, "http://proxy-1", nil)}, + []*Node{capped}, + ) + + if f.planner.PlanSession("s1", "", true, 0).TranscodeNode == nil { + t.Fatal("first session should be admitted") + } + if got := f.planner.PlanSession("s2", "", true, 0).TranscodeNode; got != nil { + t.Fatalf("second session should be rejected, got %+v", got) + } + + // Without health reports (LastHealthCheck nil) reservations still expire + // after maxReservationAge so a stalled health checker can't wedge admission. + f.now = f.now.Add(maxReservationAge + time.Second) + if f.planner.PlanSession("s3", "", true, 0).TranscodeNode == nil { + t.Fatal("session should be admitted after reservation expiry") + } +} + +func TestDirectPlayIgnoresGroups(t *testing.T) { + f := newFixture( + []*Node{proxyNode(1, "http://proxy-a", strPtr("rack-a"))}, + []*Node{}, + ) + + plan := f.planner.PlanSession("s1", "", false, 0) + if plan.ProxyNode == nil || plan.ProxyNode.URL != "http://proxy-a" { + t.Fatalf("expected grouped proxy to serve direct play, got %+v", plan.ProxyNode) + } + if plan.TranscodeNode != nil { + t.Fatalf("direct play must not pick a transcode node, got %+v", plan.TranscodeNode) + } +} + +func TestGroupRoundRobinAcrossGroupProxies(t *testing.T) { + f := newFixture( + []*Node{ + proxyNode(1, "http://proxy-a1", strPtr("rack-a")), + proxyNode(2, "http://proxy-a2", strPtr("rack-a")), + }, + []*Node{transcodeNode(3, "http://tc-a", strPtr("rack-a"), 0)}, + ) + + seen := map[string]bool{} + for i, id := range []string{"s1", "s2"} { + plan := f.planner.PlanSession(id, "", true, 0) + if plan.ProxyNode == nil { + t.Fatalf("plan %d: expected a proxy", i) + } + seen[plan.ProxyNode.URL] = true + } + if len(seen) != 2 { + t.Fatalf("expected round-robin across both group proxies, saw %v", seen) + } +} + +func TestNilPlannerReturnsEmptyPlan(t *testing.T) { + var p *Planner + plan := p.PlanSession("s1", "", true, 0) + if plan.TranscodeNode != nil || plan.ProxyNode != nil { + t.Fatalf("expected empty plan from nil planner, got %+v", plan) + } +} + +func TestBandwidthCapSkipsSaturatedProxy(t *testing.T) { + saturated := proxyNode(1, "http://proxy-1", nil) + saturated.MaxBandwidthKbps = intPtr(100_000) // 100 Mbps + saturated.EgressKbps = 97_000 + f := newFixture( + []*Node{saturated, proxyNode(2, "http://proxy-2", nil)}, + []*Node{}, + ) + + // A 6 Mbps stream doesn't fit in proxy-1's 3 Mbps of headroom. + for i := 0; i < 3; i++ { + plan := f.planner.PlanSession("s", "", false, 6_000) + if plan.ProxyNode == nil || plan.ProxyNode.URL != "http://proxy-2" { + t.Fatalf("expected proxy-2 (proxy-1 saturated), got %+v", plan.ProxyNode) + } + } + + // A 2 Mbps stream still fits. + plan := f.planner.PlanSession("s2", "", false, 2_000) + if plan.ProxyNode == nil { + t.Fatal("expected a proxy for a stream that fits") + } +} + +func TestBandwidthReservationsCountDuringBridge(t *testing.T) { + capped := proxyNode(1, "http://proxy-1", nil) + capped.MaxBandwidthKbps = intPtr(10_000) + f := newFixture([]*Node{capped}, []*Node{}) + + // Two 4 Mbps admissions fit; the third would exceed the 10 Mbps cap + // because the first two are still bridged as reservations. + if f.planner.PlanSession("s1", "", false, 4_000).ProxyNode == nil { + t.Fatal("first stream should be admitted") + } + if f.planner.PlanSession("s2", "", false, 4_000).ProxyNode == nil { + t.Fatal("second stream should be admitted") + } + if got := f.planner.PlanSession("s3", "", false, 4_000).ProxyNode; got != nil { + t.Fatalf("third stream should be rejected, got %+v", got) + } + + // Unlike job reservations, bandwidth bridges ignore health freshness — + // a report right after admission would not reflect the streams yet. + newer := f.now.Add(5 * time.Second) + f.proxies.ApplyHealth(1, true, 0, 0, newer) + f.now = f.now.Add(10 * time.Second) + if got := f.planner.PlanSession("s4", "", false, 4_000).ProxyNode; got != nil { + t.Fatalf("stream should still be rejected during bridge window, got %+v", got) + } + + // After the bridge window the measured egress is authoritative. The + // meter now reports 8 Mbps, so one more 4 Mbps stream still won't fit, + // but a 2 Mbps one will. + f.now = f.now.Add(bandwidthBridgeAge) + f.proxies.ApplyHealth(1, true, 0, 8_000, f.now) + if got := f.planner.PlanSession("s5", "", false, 4_000).ProxyNode; got != nil { + t.Fatalf("4 Mbps stream should not fit at 8/10 Mbps, got %+v", got) + } + if f.planner.PlanSession("s6", "", false, 2_000).ProxyNode == nil { + t.Fatal("2 Mbps stream should fit at 8/10 Mbps") + } +} + +func TestGroupBandwidthGatesGroupTranscode(t *testing.T) { + groupProxy := proxyNode(1, "http://proxy-a", strPtr("rack-a")) + groupProxy.MaxBandwidthKbps = intPtr(10_000) + groupProxy.EgressKbps = 9_000 + f := newFixture( + []*Node{groupProxy, proxyNode(2, "http://proxy-1", nil)}, + []*Node{ + transcodeNode(3, "http://tc-a", strPtr("rack-a"), 0), + transcodeNode(4, "http://tc-1", nil, 7), + }, + ) + + // rack-a's proxy has no bandwidth headroom for a 4 Mbps stream, so the + // group's idle transcode node must be skipped. + plan := f.planner.PlanSession("s1", "", true, 4_000) + if plan.TranscodeNode == nil || plan.TranscodeNode.URL != "http://tc-1" { + t.Fatalf("expected ungrouped tc-1, got %+v", plan.TranscodeNode) + } + if plan.ProxyNode == nil || plan.ProxyNode.URL != "http://proxy-1" { + t.Fatalf("expected ungrouped proxy-1, got %+v", plan.ProxyNode) + } + + // A 500 kbps stream fits and stays pinned to the group. + plan = f.planner.PlanSession("s2", "", true, 500) + if plan.TranscodeNode == nil || plan.TranscodeNode.URL != "http://tc-a" { + t.Fatalf("expected rack-a tc-a for small stream, got %+v", plan.TranscodeNode) + } + if plan.ProxyNode == nil || plan.ProxyNode.URL != "http://proxy-a" { + t.Fatalf("expected rack-a proxy, got %+v", plan.ProxyNode) + } +} + +func TestUnknownBitrateAdmittedBelowCap(t *testing.T) { + p := proxyNode(1, "http://proxy-1", nil) + p.MaxBandwidthKbps = intPtr(10_000) + p.EgressKbps = 9_999 + f := newFixture([]*Node{p}, []*Node{}) + + // Unknown bitrate (0): admitted while measured egress is below the cap. + if f.planner.PlanSession("s1", "", false, 0).ProxyNode == nil { + t.Fatal("unknown-bitrate stream should be admitted below cap") + } + + f.proxies.ApplyHealth(1, true, 0, 10_000, f.now) + if got := f.planner.PlanSession("s2", "", false, 0).ProxyNode; got != nil { + t.Fatalf("unknown-bitrate stream should be rejected at cap, got %+v", got) + } +} diff --git a/internal/nodepool/proxy_pool.go b/internal/nodepool/proxy_pool.go index fedac1c4..95c0480a 100644 --- a/internal/nodepool/proxy_pool.go +++ b/internal/nodepool/proxy_pool.go @@ -3,6 +3,7 @@ package nodepool import ( "sync" "sync/atomic" + "time" ) // ProxyPool manages proxy nodes with round-robin selection. @@ -53,3 +54,11 @@ func (p *ProxyPool) Nodes() []*Node { copy(cp, p.nodes) return cp } + +// ApplyHealth records a health check result by swapping the node for an +// updated copy, keeping published *Node values immutable. +func (p *ProxyPool) ApplyHealth(id int, healthy bool, activeJobs, egressKbps int, checkedAt time.Time) { + p.mu.Lock() + defer p.mu.Unlock() + applyNodeHealth(p.nodes, id, healthy, activeJobs, egressKbps, checkedAt) +} diff --git a/internal/nodepool/repository.go b/internal/nodepool/repository.go index a0d405b6..21a48e9a 100644 --- a/internal/nodepool/repository.go +++ b/internal/nodepool/repository.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "strings" "time" "github.com/jackc/pgx/v5" @@ -19,22 +20,29 @@ const ( // Node represents a stream node in the database. type Node struct { - ID int `json:"id"` - Name string `json:"name"` - Type string `json:"type"` - URL string `json:"url"` - Enabled bool `json:"enabled"` - Healthy bool `json:"healthy"` - ActiveJobs int `json:"active_jobs"` - LastHealthCheck *time.Time `json:"last_health_check"` - CreatedAt time.Time `json:"created_at"` + ID int `json:"id"` + Name string `json:"name"` + Type string `json:"type"` + URL string `json:"url"` + Enabled bool `json:"enabled"` + Healthy bool `json:"healthy"` + ActiveJobs int `json:"active_jobs"` + Group *string `json:"group"` // co-location group; nil = ungrouped + MaxJobs *int `json:"max_jobs"` // concurrent job cap; nil = unlimited + MaxBandwidthKbps *int `json:"max_bandwidth_kbps"` // egress cap in kilobits/s; nil = unlimited + EgressKbps int `json:"egress_kbps"` // health-reported rolling egress average + LastHealthCheck *time.Time `json:"last_health_check"` + CreatedAt time.Time `json:"created_at"` } // CreateNodeInput holds the fields for creating a new node. type CreateNodeInput struct { - Name string `json:"name"` - Type string `json:"type"` - URL string `json:"url"` + Name string `json:"name"` + Type string `json:"type"` + URL string `json:"url"` + Group string `json:"group"` // empty = ungrouped + MaxJobs *int `json:"max_jobs"` // nil or <= 0 = unlimited + MaxBandwidthKbps *int `json:"max_bandwidth_kbps"` // nil or <= 0 = unlimited } // Validate checks required fields and allowed values. @@ -52,10 +60,33 @@ func (i CreateNodeInput) Validate() error { } // UpdateNodeInput holds the fields for updating a node. +// The optional fields distinguish "leave unchanged" (nil) from "clear": +// an empty-string Group clears the group, and a non-positive MaxJobs or +// MaxBandwidthKbps clears that cap. type UpdateNodeInput struct { - Name *string `json:"name,omitempty"` - URL *string `json:"url,omitempty"` - Enabled *bool `json:"enabled,omitempty"` + Name *string `json:"name,omitempty"` + URL *string `json:"url,omitempty"` + Enabled *bool `json:"enabled,omitempty"` + Group *string `json:"group,omitempty"` + MaxJobs *int `json:"max_jobs,omitempty"` + MaxBandwidthKbps *int `json:"max_bandwidth_kbps,omitempty"` +} + +// normalizeGroup trims a group label and converts empty to NULL. +func normalizeGroup(group string) *string { + g := strings.TrimSpace(group) + if g == "" { + return nil + } + return &g +} + +// normalizeCap converts non-positive capacity values to NULL (unlimited). +func normalizeCap(v *int) *int { + if v == nil || *v <= 0 { + return nil + } + return v } // Repository provides CRUD operations for stream nodes. @@ -68,13 +99,15 @@ func NewRepository(pool *pgxpool.Pool) *Repository { return &Repository{pool: pool} } -const nodeColumns = `id, name, type, url, enabled, healthy, active_jobs, last_health_check, created_at` +const nodeColumns = `id, name, type, url, enabled, healthy, active_jobs, node_group, max_jobs, max_bandwidth_kbps, egress_kbps, last_health_check, created_at` func scanNode(row pgx.Row) (*Node, error) { var n Node err := row.Scan( &n.ID, &n.Name, &n.Type, &n.URL, &n.Enabled, &n.Healthy, &n.ActiveJobs, + &n.Group, &n.MaxJobs, + &n.MaxBandwidthKbps, &n.EgressKbps, &n.LastHealthCheck, &n.CreatedAt, ) if err != nil { @@ -90,6 +123,8 @@ func scanNodes(rows pgx.Rows) ([]*Node, error) { if err := rows.Scan( &n.ID, &n.Name, &n.Type, &n.URL, &n.Enabled, &n.Healthy, &n.ActiveJobs, + &n.Group, &n.MaxJobs, + &n.MaxBandwidthKbps, &n.EgressKbps, &n.LastHealthCheck, &n.CreatedAt, ); err != nil { return nil, err @@ -142,22 +177,43 @@ func (r *Repository) Create(ctx context.Context, input CreateNodeInput) (*Node, return nil, err } row := r.pool.QueryRow(ctx, - `INSERT INTO stream_nodes (name, type, url) VALUES ($1, $2, $3) + `INSERT INTO stream_nodes (name, type, url, node_group, max_jobs, max_bandwidth_kbps) + VALUES ($1, $2, $3, $4, $5, $6) RETURNING `+nodeColumns, - input.Name, input.Type, input.URL) + input.Name, input.Type, input.URL, normalizeGroup(input.Group), + normalizeCap(input.MaxJobs), normalizeCap(input.MaxBandwidthKbps)) return scanNode(row) } -// Update modifies a node's mutable fields. +// Update modifies a node's mutable fields. The optional fields use sentinel +// values to clear: an empty-string group and non-positive caps set the +// column to NULL (see UpdateNodeInput). func (r *Repository) Update(ctx context.Context, id int, input UpdateNodeInput) (*Node, error) { + var group *string + if input.Group != nil { + group = normalizeGroup(*input.Group) + } + var maxJobs, maxBandwidth *int + if input.MaxJobs != nil { + maxJobs = normalizeCap(input.MaxJobs) + } + if input.MaxBandwidthKbps != nil { + maxBandwidth = normalizeCap(input.MaxBandwidthKbps) + } row := r.pool.QueryRow(ctx, `UPDATE stream_nodes SET name = COALESCE($2, name), url = COALESCE($3, url), - enabled = COALESCE($4, enabled) + enabled = COALESCE($4, enabled), + node_group = CASE WHEN $5::boolean THEN $6::text ELSE node_group END, + max_jobs = CASE WHEN $7::boolean THEN $8::integer ELSE max_jobs END, + max_bandwidth_kbps = CASE WHEN $9::boolean THEN $10::integer ELSE max_bandwidth_kbps END WHERE id = $1 RETURNING `+nodeColumns, - id, input.Name, input.URL, input.Enabled) + id, input.Name, input.URL, input.Enabled, + input.Group != nil, group, + input.MaxJobs != nil, maxJobs, + input.MaxBandwidthKbps != nil, maxBandwidth) n, err := scanNode(row) if errors.Is(err, pgx.ErrNoRows) { return nil, ErrNodeNotFound @@ -180,12 +236,13 @@ func (r *Repository) Delete(ctx context.Context, id int) error { return nil } -// UpdateHealth updates a node's health status and active job count. -func (r *Repository) UpdateHealth(ctx context.Context, id int, healthy bool, activeJobs int) error { +// UpdateHealth updates a node's health status, active job count, and +// reported egress bandwidth. +func (r *Repository) UpdateHealth(ctx context.Context, id int, healthy bool, activeJobs, egressKbps int) error { tag, err := r.pool.Exec(ctx, - `UPDATE stream_nodes SET healthy = $2, active_jobs = $3, last_health_check = NOW() + `UPDATE stream_nodes SET healthy = $2, active_jobs = $3, egress_kbps = $4, last_health_check = NOW() WHERE id = $1`, - id, healthy, activeJobs) + id, healthy, activeJobs, egressKbps) if err != nil { return fmt.Errorf("update node health: %w", err) } diff --git a/internal/nodepool/transcode_pool.go b/internal/nodepool/transcode_pool.go index 3489c055..35088a70 100644 --- a/internal/nodepool/transcode_pool.go +++ b/internal/nodepool/transcode_pool.go @@ -1,6 +1,9 @@ package nodepool -import "sync" +import ( + "sync" + "time" +) // TranscodePool manages transcode nodes with least-connections selection. // Thread-safe for concurrent use. @@ -60,3 +63,27 @@ func (p *TranscodePool) Nodes() []*Node { copy(cp, p.nodes) return cp } + +// ApplyHealth records a health check result by swapping the node for an +// updated copy, keeping published *Node values immutable. +func (p *TranscodePool) ApplyHealth(id int, healthy bool, activeJobs, egressKbps int, checkedAt time.Time) { + p.mu.Lock() + defer p.mu.Unlock() + applyNodeHealth(p.nodes, id, healthy, activeJobs, egressKbps, checkedAt) +} + +// applyNodeHealth replaces the slice entry for id with an updated copy. +func applyNodeHealth(nodes []*Node, id int, healthy bool, activeJobs, egressKbps int, checkedAt time.Time) { + for i, n := range nodes { + if n.ID != id { + continue + } + clone := *n + clone.Healthy = healthy + clone.ActiveJobs = activeJobs + clone.EgressKbps = egressKbps + clone.LastHealthCheck = &checkedAt + nodes[i] = &clone + return + } +} diff --git a/internal/nodesessions/tracker.go b/internal/nodesessions/tracker.go index f9521d4a..4ded00eb 100644 --- a/internal/nodesessions/tracker.go +++ b/internal/nodesessions/tracker.go @@ -43,7 +43,8 @@ type Tracker struct { nodeHash string // first 8 chars of SHA-256 of nodeURL mu sync.Mutex - sessions map[string]struct{} // set of active session IDs + sessions map[string]struct{} // set of active session IDs + touched map[string]time.Time // ephemeral sessions by last-activity time } // NewTracker creates a session tracker for the given node. @@ -57,6 +58,7 @@ func NewTracker(rdb *redis.Client, nodeURL, nodeName, nodeType string) *Tracker nodeType: nodeType, nodeHash: hex.EncodeToString(h[:4]), // 8 hex chars sessions: make(map[string]struct{}), + touched: make(map[string]time.Time), } } @@ -80,11 +82,22 @@ func (tr *Tracker) NodeName() string { return tr.nodeName } -// ActiveCount returns the number of active sessions tracked by this node. +// ActiveCount returns the number of active sessions tracked by this node, +// including ephemeral sessions touched within the session TTL. func (tr *Tracker) ActiveCount() int { tr.mu.Lock() defer tr.mu.Unlock() - return len(tr.sessions) + now := time.Now() + count := len(tr.sessions) + for id, last := range tr.touched { + if _, dup := tr.sessions[id]; dup { + continue + } + if now.Sub(last) <= sessionTTL { + count++ + } + } + return count } // Track registers an active session in Redis with a TTL. @@ -108,6 +121,32 @@ func (tr *Tracker) Track(ctx context.Context, info SessionInfo) { tr.mu.Unlock() } +// Touch registers or refreshes an ephemeral session that has no explicit end, +// such as HLS manifest/segment fetches flowing through a proxy. The session is +// written to Redis on first touch and drops out of the active count after +// sessionTTL without further touches (pruned by the refresh loop). +func (tr *Tracker) Touch(ctx context.Context, info SessionInfo) { + if tr.rdb == nil { + return + } + tr.mu.Lock() + _, known := tr.touched[info.SessionID] + tr.touched[info.SessionID] = time.Now() + tr.mu.Unlock() + if known { + return + } + + data, err := json.Marshal(info) + if err != nil { + slog.Debug("session touch marshal failed", "error", err) + return + } + if err := tr.rdb.Set(ctx, tr.redisKey(info.SessionID), data, sessionTTL).Err(); err != nil { + slog.Debug("session touch set failed", "error", err, "session", info.SessionID) + } +} + // Remove deletes a session from Redis and the in-memory set. func (tr *Tracker) Remove(ctx context.Context, sessionID string) { if tr.rdb == nil { @@ -115,6 +154,7 @@ func (tr *Tracker) Remove(ctx context.Context, sessionID string) { } tr.mu.Lock() delete(tr.sessions, sessionID) + delete(tr.touched, sessionID) tr.mu.Unlock() if err := tr.rdb.Del(ctx, tr.redisKey(sessionID)).Err(); err != nil { @@ -128,11 +168,17 @@ func (tr *Tracker) Cleanup(ctx context.Context) { return } tr.mu.Lock() - ids := make([]string, 0, len(tr.sessions)) + ids := make([]string, 0, len(tr.sessions)+len(tr.touched)) for id := range tr.sessions { ids = append(ids, id) } + for id := range tr.touched { + if _, dup := tr.sessions[id]; !dup { + ids = append(ids, id) + } + } tr.sessions = make(map[string]struct{}) + tr.touched = make(map[string]time.Time) tr.mu.Unlock() if len(ids) == 0 { @@ -169,11 +215,23 @@ func (tr *Tracker) StartRefresh(ctx context.Context) { } func (tr *Tracker) refreshAll(ctx context.Context) { + now := time.Now() tr.mu.Lock() - ids := make([]string, 0, len(tr.sessions)) + ids := make([]string, 0, len(tr.sessions)+len(tr.touched)) for id := range tr.sessions { ids = append(ids, id) } + for id, last := range tr.touched { + if now.Sub(last) > sessionTTL { + // Idle ephemeral session: stop refreshing and let the Redis + // key expire on its own. + delete(tr.touched, id) + continue + } + if _, dup := tr.sessions[id]; !dup { + ids = append(ids, id) + } + } tr.mu.Unlock() if len(ids) == 0 { diff --git a/internal/proxy/egress.go b/internal/proxy/egress.go new file mode 100644 index 00000000..e1a9c2fa --- /dev/null +++ b/internal/proxy/egress.go @@ -0,0 +1,86 @@ +package proxy + +import ( + "net/http" + "sync" + "time" +) + +// meterWindowSeconds is the averaging window for the egress rate. HLS clients +// fetch segments in bursts (especially when buffering ahead), so a window of +// this size smooths the spikes into something close to the steady-state +// stream rate. The planner's bandwidth reservation bridge matches this value. +const meterWindowSeconds = 60 + +// egressMeter measures outbound stream bytes as a rolling per-second ring, +// reporting the average rate over the window. Safe for concurrent use. +type egressMeter struct { + mu sync.Mutex + // One extra bucket so the current (partial) second never collides with + // the oldest second still inside the window. + buckets [meterWindowSeconds + 1]int64 + stamps [meterWindowSeconds + 1]int64 // unix second each bucket holds + now func() time.Time +} + +func newEgressMeter() *egressMeter { + return &egressMeter{now: time.Now} +} + +// Add records n egressed bytes against the current second. +func (m *egressMeter) Add(n int64) { + if n <= 0 { + return + } + sec := m.now().Unix() + i := int(sec % int64(len(m.buckets))) + m.mu.Lock() + if m.stamps[i] != sec { + m.stamps[i] = sec + m.buckets[i] = 0 + } + m.buckets[i] += n + m.mu.Unlock() +} + +// RateKbps returns the average egress over the window in kilobits/s. +func (m *egressMeter) RateKbps() int { + sec := m.now().Unix() + var total int64 + m.mu.Lock() + for i := range m.buckets { + if sec-m.stamps[i] < meterWindowSeconds { + total += m.buckets[i] + } + } + m.mu.Unlock() + return int(total * 8 / 1000 / meterWindowSeconds) +} + +// meteredResponseWriter counts every byte written to the client. +// Embedding the interface intentionally hides optimizations like +// io.ReaderFrom so all writes flow through Write. +type meteredResponseWriter struct { + http.ResponseWriter + meter *egressMeter +} + +func (w *meteredResponseWriter) Write(b []byte) (int, error) { + n, err := w.ResponseWriter.Write(b) + w.meter.Add(int64(n)) + return n, err +} + +func (w *meteredResponseWriter) Flush() { + if f, ok := w.ResponseWriter.(http.Flusher); ok { + f.Flush() + } +} + +// meterEgress wraps stream handlers so their responses count toward the +// node's measured egress bandwidth. +func (s *Server) meterEgress(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + next.ServeHTTP(&meteredResponseWriter{ResponseWriter: w, meter: s.egress}, r) + }) +} diff --git a/internal/proxy/egress_test.go b/internal/proxy/egress_test.go new file mode 100644 index 00000000..344ece3c --- /dev/null +++ b/internal/proxy/egress_test.go @@ -0,0 +1,51 @@ +package proxy + +import ( + "testing" + "time" +) + +func TestEgressMeterAveragesOverWindow(t *testing.T) { + now := time.Date(2026, 6, 10, 12, 0, 0, 0, time.UTC) + m := newEgressMeter() + m.now = func() time.Time { return now } + + if got := m.RateKbps(); got != 0 { + t.Fatalf("empty meter rate = %d, want 0", got) + } + + // 1 MB/s for 10 seconds = 80 Mbit over a 60s window = ~1333 kbps average. + for i := 0; i < 10; i++ { + m.Add(1_000_000) + now = now.Add(time.Second) + } + got := m.RateKbps() + want := 10 * 1_000_000 * 8 / 1000 / meterWindowSeconds + if got != want { + t.Fatalf("rate = %d kbps, want %d", got, want) + } + + // Once the writes age out of the window the rate returns to zero. + now = now.Add(meterWindowSeconds * time.Second) + if got := m.RateKbps(); got != 0 { + t.Fatalf("rate after window = %d, want 0", got) + } +} + +func TestEgressMeterRingReusesSlots(t *testing.T) { + now := time.Date(2026, 6, 10, 12, 0, 0, 0, time.UTC) + m := newEgressMeter() + m.now = func() time.Time { return now } + + // Write into the same ring slot two window-laps apart; the stale value + // must be replaced, not accumulated. + m.Add(5_000_000) + now = now.Add(time.Duration(len(m.buckets)) * time.Second) + m.Add(1_000_000) + + got := m.RateKbps() + want := 1_000_000 * 8 / 1000 / meterWindowSeconds + if got != want { + t.Fatalf("rate = %d kbps, want %d (stale bucket leaked)", got, want) + } +} diff --git a/internal/proxy/server.go b/internal/proxy/server.go index f148cc00..fd8d20c1 100644 --- a/internal/proxy/server.go +++ b/internal/proxy/server.go @@ -24,17 +24,35 @@ type Server struct { watcher *nodeconfig.Watcher tracker *nodesessions.Tracker httpClient *http.Client + egress *egressMeter } // NewServer creates a new proxy server backed by a config watcher and session tracker. func NewServer(watcher *nodeconfig.Watcher, tracker *nodesessions.Tracker) *Server { return &Server{ - watcher: watcher, - tracker: tracker, - httpClient: &http.Client{}, // no timeout for long streams + watcher: watcher, + tracker: tracker, + // No overall timeout — stream bodies are long-lived. Hung nodes are + // bounded by the transport's response-header timeout instead. + httpClient: &http.Client{Transport: newStreamTransport()}, + egress: newEgressMeter(), } } +// newStreamTransport tunes the proxy→transcode-node connection pool. Many +// concurrent viewers fan their segment fetches through one proxy→node pair, +// and Go's default of 2 idle connections per host causes constant connection +// churn (and TLS re-handshakes) under load. The response-header timeout +// bounds requests to a hung node; the longest legitimate server-side wait is +// the 30s manifest-readiness poll on the transcode node. +func newStreamTransport() *http.Transport { + t := http.DefaultTransport.(*http.Transport).Clone() + t.MaxIdleConns = 128 + t.MaxIdleConnsPerHost = 32 + t.ResponseHeaderTimeout = 60 * time.Second + return t +} + // Handler returns the chi.Router with all proxy routes mounted. func (s *Server) Handler() http.Handler { r := chi.NewRouter() @@ -47,15 +65,20 @@ func (s *Server) Handler() http.Handler { MaxAge: 86400, })) r.Get("/api/v1/health", s.handleHealth) - r.Head("/stream/direct/{token}", s.handleDirectPlay) - r.Get("/stream/direct/{token}", s.handleDirectPlay) - r.Head("/stream/remux/{token}", s.handleRemux) - r.Get("/stream/remux/{token}", s.handleRemux) - r.Head("/stream/transcode/{token}/master.m3u8", s.handleTranscodeManifest) - r.Get("/stream/transcode/{token}/master.m3u8", s.handleTranscodeManifest) - r.Get("/stream/transcode/{token}/segment/{name}", s.handleTranscodeSegment) - r.Get("/stream/subtitles/{token}/{track}/fonts", s.handleSubtitleFonts) - r.Get("/stream/subtitles/{token}/{track}", s.handleSubtitle) + r.Group(func(r chi.Router) { + // Everything under /stream counts toward the node's measured + // egress bandwidth. + r.Use(s.meterEgress) + r.Head("/stream/direct/{token}", s.handleDirectPlay) + r.Get("/stream/direct/{token}", s.handleDirectPlay) + r.Head("/stream/remux/{token}", s.handleRemux) + r.Get("/stream/remux/{token}", s.handleRemux) + r.Head("/stream/transcode/{token}/master.m3u8", s.handleTranscodeManifest) + r.Get("/stream/transcode/{token}/master.m3u8", s.handleTranscodeManifest) + r.Get("/stream/transcode/{token}/segment/{name}", s.handleTranscodeSegment) + r.Get("/stream/subtitles/{token}/{track}/fonts", s.handleSubtitleFonts) + r.Get("/stream/subtitles/{token}/{track}", s.handleSubtitle) + }) // Admin routes — bearer-auth protected. r.Group(func(r chi.Router) { @@ -67,12 +90,22 @@ func (s *Server) Handler() http.Handler { } type healthResponse struct { - Status string `json:"status"` + Status string `json:"status"` + ActiveJobs int `json:"active_jobs"` + EgressKbps int `json:"egress_kbps"` } func (s *Server) handleHealth(w http.ResponseWriter, _ *http.Request) { + activeJobs := 0 + if s.tracker != nil { + activeJobs = s.tracker.ActiveCount() + } w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(healthResponse{Status: "ok"}) + json.NewEncoder(w).Encode(healthResponse{ + Status: "ok", + ActiveJobs: activeJobs, + EgressKbps: s.egress.RateKbps(), + }) } // requireBearer checks Authorization: Bearer {secret} for admin endpoints. @@ -149,6 +182,7 @@ func (s *Server) handleTranscodeManifest(w http.ResponseWriter, r *http.Request) if claims == nil { return } + s.touchTranscodeSession(r, claims) s.proxyToTranscodeNode(w, r, claims, "/transcode/"+claims.SessionID+"/master.m3u8") } @@ -157,10 +191,25 @@ func (s *Server) handleTranscodeSegment(w http.ResponseWriter, r *http.Request) if claims == nil { return } + s.touchTranscodeSession(r, claims) name := chi.URLParam(r, "name") s.proxyToTranscodeNode(w, r, claims, "/transcode/"+claims.SessionID+"/segment/"+name) } +// touchTranscodeSession keeps HLS sessions visible in the active stream count. +// Unlike direct play and remux, transcode playback reaches the proxy as many +// short manifest/segment requests, so the session is tracked by recent +// activity instead of request lifetime. +func (s *Server) touchTranscodeSession(r *http.Request, claims *streamtoken.Claims) { + s.tracker.Touch(r.Context(), nodesessions.SessionInfo{ + SessionID: claims.SessionID, + NodeURL: s.tracker.NodeURL(), + NodeName: s.tracker.NodeName(), + Type: "transcode", + StartedAt: time.Now().UTC().Format(time.RFC3339), + }) +} + func (s *Server) handleSubtitle(w http.ResponseWriter, r *http.Request) { claims := s.verifyToken(w, r) if claims == nil { diff --git a/internal/transcodenode/server.go b/internal/transcodenode/server.go index e0275446..b3aa1122 100644 --- a/internal/transcodenode/server.go +++ b/internal/transcodenode/server.go @@ -8,6 +8,7 @@ import ( "net/http" "os" "path/filepath" + "strconv" "strings" "sync" "sync/atomic" @@ -225,7 +226,16 @@ func (s *Server) handleStart(w http.ResponseWriter, r *http.Request) { s.mu.Unlock() s.activeJobs.Add(-1) _ = old.Close() - os.RemoveAll(outputDir) + // Move the old segment directory aside and delete it in the + // background: removing a long session's segments can take seconds + // on slow disks, and the playback start that triggered this switch + // is blocked waiting for our 202. + staleDir := outputDir + ".stale-" + strconv.FormatInt(time.Now().UnixNano(), 10) + if err := os.Rename(outputDir, staleDir); err == nil { + go func() { _ = os.RemoveAll(staleDir) }() + } else { + os.RemoveAll(outputDir) + } } else { s.mu.Unlock() } @@ -242,9 +252,12 @@ func (s *Server) handleStart(w http.ResponseWriter, r *http.Request) { s.mu.Unlock() s.activeJobs.Add(1) - // Track session in Redis + // Track session in Redis off the request path — the API server (and + // behind it the playback client) is blocked on this 202, and the + // tracking write is monitoring-only. effectiveHWAccel := session.Opts().HWAccel - s.tracker.Track(r.Context(), nodesessions.SessionInfo{ + trackCtx := context.WithoutCancel(r.Context()) + go s.tracker.Track(trackCtx, nodesessions.SessionInfo{ SessionID: req.SessionID, NodeURL: s.tracker.NodeURL(), NodeName: s.tracker.NodeName(), diff --git a/migrations/sql/20260610130619_node_groups_and_caps.sql b/migrations/sql/20260610130619_node_groups_and_caps.sql new file mode 100644 index 00000000..c4f39962 --- /dev/null +++ b/migrations/sql/20260610130619_node_groups_and_caps.sql @@ -0,0 +1,35 @@ +-- +goose Up +-- +goose StatementBegin +ALTER TABLE stream_nodes + ADD COLUMN node_group text, + ADD COLUMN max_jobs integer, + ADD COLUMN max_bandwidth_kbps integer, + ADD COLUMN egress_kbps integer NOT NULL DEFAULT 0; + +COMMENT ON COLUMN stream_nodes.node_group IS + 'Optional co-location group. Nodes sharing a group are assumed to be on the ' + 'same host/LAN. A group is only eligible for selection while every enabled ' + 'member is healthy; transcoded streams are served by a proxy from the same ' + 'group as the chosen transcode node.'; + +COMMENT ON COLUMN stream_nodes.max_jobs IS + 'Maximum concurrent jobs for this node (transcodes for transcode nodes, ' + 'streams for proxy nodes). NULL = unlimited.'; + +COMMENT ON COLUMN stream_nodes.max_bandwidth_kbps IS + 'Maximum egress bandwidth for proxy nodes in kilobits/s. New streams are ' + 'not admitted once the measured egress would exceed this. NULL = unlimited.'; + +COMMENT ON COLUMN stream_nodes.egress_kbps IS + 'Last health-reported egress bandwidth (rolling average, kilobits/s). ' + 'Currently only proxy nodes report a non-zero value.'; +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +ALTER TABLE stream_nodes + DROP COLUMN IF EXISTS node_group, + DROP COLUMN IF EXISTS max_jobs, + DROP COLUMN IF EXISTS max_bandwidth_kbps, + DROP COLUMN IF EXISTS egress_kbps; +-- +goose StatementEnd diff --git a/web/src/api/types.ts b/web/src/api/types.ts index beea5d27..06b585fc 100644 --- a/web/src/api/types.ts +++ b/web/src/api/types.ts @@ -2938,6 +2938,10 @@ export interface StreamNode { enabled: boolean; healthy: boolean; active_jobs: number; + group: string | null; + max_jobs: number | null; + max_bandwidth_kbps: number | null; + egress_kbps: number; last_health_check: string | null; created_at: string; } @@ -2946,17 +2950,25 @@ export interface CreateNodeRequest { name: string; type: string; url: string; + group?: string; + max_jobs?: number; + max_bandwidth_kbps?: number; } export interface UpdateNodeRequest { name?: string; url?: string; enabled?: boolean; + // Empty string clears the group; 0 clears the caps (unlimited). + group?: string; + max_jobs?: number; + max_bandwidth_kbps?: number; } export interface CheckNodeResponse { healthy: boolean; active_jobs: number; + egress_kbps: number; } // User-facing library (simplified, no admin fields) diff --git a/web/src/pages/AdminNodes.tsx b/web/src/pages/AdminNodes.tsx index 98760ddd..2300445a 100644 --- a/web/src/pages/AdminNodes.tsx +++ b/web/src/pages/AdminNodes.tsx @@ -28,6 +28,10 @@ import { ConfirmDialog } from "@/components/ConfirmDialog"; type NodeType = "proxy" | "transcode"; +function formatMbps(kbps: number): string { + return (Math.round(kbps / 100) / 10).toString(); +} + interface NodeSectionProps { type: NodeType; nodes: StreamNode[]; @@ -54,7 +58,7 @@ function NodeSection({ checkingHealthId, }: NodeSectionProps) { const label = type === "proxy" ? "Proxy" : "Transcode"; - const colCount = showJobs ? 7 : 6; + const colCount = (showJobs ? 8 : 7) + (type === "proxy" ? 1 : 0); return (
@@ -76,9 +80,11 @@ function NodeSection({ Name URL + Group Status Health - {showJobs && Jobs} + {showJobs && {type === "proxy" ? "Streams" : "Jobs"}} + {type === "proxy" && Egress} Last Check Actions @@ -106,6 +112,13 @@ function NodeSection({ {node.name} {node.url} + + {node.group ? ( + {node.group} + ) : ( + — + )} + onToggle(node)} /> @@ -125,7 +138,26 @@ function NodeSection({ - {showJobs && {node.active_jobs}} + {showJobs && ( + + {node.active_jobs} + {node.max_jobs != null && ( + / {node.max_jobs} + )} + + )} + {type === "proxy" && ( + + {formatMbps(node.egress_kbps)} + {node.max_bandwidth_kbps != null && ( + + {" "} + / {formatMbps(node.max_bandwidth_kbps)} + + )}{" "} + Mbps + + )} {node.last_health_check ? new Date(node.last_health_check).toLocaleString() @@ -188,6 +220,11 @@ function NodeForm({ }) { const [name, setName] = useState(node?.name ?? ""); const [url, setUrl] = useState(node?.url ?? ""); + const [group, setGroup] = useState(node?.group ?? ""); + const [maxJobs, setMaxJobs] = useState(node?.max_jobs?.toString() ?? ""); + const [maxBandwidthMbps, setMaxBandwidthMbps] = useState( + node?.max_bandwidth_kbps ? (node.max_bandwidth_kbps / 1000).toString() : "", + ); const createMutation = useCreateNode(); const updateMutation = useUpdateNode(); const isPending = createMutation.isPending || updateMutation.isPending; @@ -197,10 +234,23 @@ function NodeForm({ async function handleSubmit(e: FormEvent) { e.preventDefault(); + // The backend treats an empty group as "ungrouped" and caps <= 0 as + // "unlimited", so cleared inputs reset those fields. + const parsedMaxJobs = parseInt(maxJobs, 10); + const parsedMaxBandwidthMbps = parseFloat(maxBandwidthMbps); + const fields = { + name, + url, + group: group.trim(), + max_jobs: Number.isNaN(parsedMaxJobs) ? 0 : parsedMaxJobs, + max_bandwidth_kbps: Number.isNaN(parsedMaxBandwidthMbps) + ? 0 + : Math.round(parsedMaxBandwidthMbps * 1000), + }; if (node) { - updateMutation.mutate({ id: node.id, body: { name, url } }, { onSuccess: onClose }); + updateMutation.mutate({ id: node.id, body: fields }, { onSuccess: onClose }); } else { - const body: CreateNodeRequest = { name, type: nodeType, url }; + const body: CreateNodeRequest = { type: nodeType, ...fields }; createMutation.mutate(body, { onSuccess: onClose }); } } @@ -244,6 +294,49 @@ function NodeForm({ )}
+
+ + setGroup(e.target.value)} placeholder="e.g. rack-1" /> +

+ Optional. Nodes in the same group are treated as co-located: transcoded streams are served + by a proxy from the transcode node's group, keeping traffic on the same LAN. A group is + only used while all of its nodes are healthy. +

+
+ +
+ + setMaxJobs(e.target.value)} + placeholder="Unlimited" + /> +

+ Optional concurrency cap for this node. Leave empty (or 0) for unlimited. +

+
+ + {nodeType === "proxy" && ( +
+ + setMaxBandwidthMbps(e.target.value)} + placeholder="Unlimited" + /> +

+ Optional. New streams are routed elsewhere once this node's measured egress (plus the + expected bitrate of the new stream) would exceed the cap. Active streams are never + interrupted. Leave empty (or 0) for unlimited. +

+
+ )} + @@ -329,7 +422,7 @@ export default function AdminNodes() { handleAdd("proxy")} onEdit={handleEdit} onDelete={handleDelete} diff --git a/web/src/pages/admin-settings/PlaybackSettings.tsx b/web/src/pages/admin-settings/PlaybackSettings.tsx index 196f668d..7a0c98b6 100644 --- a/web/src/pages/admin-settings/PlaybackSettings.tsx +++ b/web/src/pages/admin-settings/PlaybackSettings.tsx @@ -10,6 +10,7 @@ const KEYS = [ "playback.transcode_dir", "playback.hw_accel", "playback.transcode_enabled", + "playback.local_transcode_fallback", "playback.allow_hevc_encoding", "allow_4k_transcode", "enable_transcode_throttle", @@ -87,6 +88,13 @@ export default function PlaybackSettings() { value={form.getValue("playback.transcode_enabled")} onChange={(v) => form.setValue("playback.transcode_enabled", v)} /> + form.setValue("playback.local_transcode_fallback", v)} + /> { + if (streamUrl) preconnectToStreamOrigin(streamUrl); + }, [streamUrl]); + useEffect(() => { setPendingSeekTime(null); }, [effectiveStreamUrl]); diff --git a/web/src/player/stream-url.ts b/web/src/player/stream-url.ts index 92a73f4b..1a7c505e 100644 --- a/web/src/player/stream-url.ts +++ b/web/src/player/stream-url.ts @@ -1,5 +1,34 @@ import type { PlayMethod } from "./types"; +const preconnectedOrigins = new Set(); + +/** + * Warm the connection (DNS + TCP + TLS) to a stream origin as soon as it is + * known. In distributed deployments the stream URL points at a proxy node the + * browser has never contacted, and without this the first manifest request + * pays all the handshakes after the transcode has already started. + */ +export function preconnectToStreamOrigin(streamUrl: string): void { + if (!streamUrl.startsWith("http://") && !streamUrl.startsWith("https://")) return; + let origin: string; + try { + origin = new URL(streamUrl).origin; + } catch { + return; + } + if (typeof document === "undefined" || origin === window.location.origin) return; + if (preconnectedOrigins.has(origin)) return; + preconnectedOrigins.add(origin); + + const link = document.createElement("link"); + link.rel = "preconnect"; + link.href = origin; + // hls.js fetches are anonymous-mode CORS requests; the warmed connection + // is only reused when the preconnect uses the same credentials mode. + link.crossOrigin = "anonymous"; + document.head.appendChild(link); +} + export function buildPlayerStreamUrl( apiBaseUrl: string, streamPath: string,