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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-06-10 17:18:18 -04:00
parent 52dd5dff7f
commit 7958f0bbf0
26 changed files with 1505 additions and 192 deletions
@@ -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,
 		}
@@ -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,
 	})
 }

@@ -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)
 	}

@@ -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",
@@ -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
@@ -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)
@@ -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
@@ -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
@@ -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 {
@@ -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()
 }
@@ -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"
+}
@@ -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)
+	}
+}
@@ -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)
+}
@@ -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)
 	}
@@ -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
+	}
+}
@@ -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 {
@@ -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)
+	})
+}
@@ -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)
+	}
+}
@@ -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 {
@@ -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(),
@@ -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
@@ -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)
@@ -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 (
    <div className="space-y-3">
@@ -76,9 +80,11 @@ function NodeSection({
            <TableRow>
              <TableHead>Name</TableHead>
              <TableHead>URL</TableHead>
+              <TableHead>Group</TableHead>
              <TableHead>Status</TableHead>
              <TableHead>Health</TableHead>
-              {showJobs && <TableHead>Jobs</TableHead>}
+              {showJobs && <TableHead>{type === "proxy" ? "Streams" : "Jobs"}</TableHead>}
+              {type === "proxy" && <TableHead>Egress</TableHead>}
              <TableHead>Last Check</TableHead>
              <TableHead className="w-32">Actions</TableHead>
            </TableRow>
@@ -106,6 +112,13 @@ function NodeSection({
                  <TableRow key={node.id}>
                    <TableCell className="font-medium">{node.name}</TableCell>
                    <TableCell className="font-mono text-sm">{node.url}</TableCell>
+                    <TableCell>
+                      {node.group ? (
+                        <Badge variant="outline">{node.group}</Badge>
+                      ) : (
+                        <span className="text-muted-foreground">—</span>
+                      )}
+                    </TableCell>
                    <TableCell>
                      <Switch checked={node.enabled} onCheckedChange={() => onToggle(node)} />
                    </TableCell>
@@ -125,7 +138,26 @@ function NodeSection({
                        </span>
                      </span>
                    </TableCell>
-                    {showJobs && <TableCell>{node.active_jobs}</TableCell>}
+                    {showJobs && (
+                      <TableCell>
+                        {node.active_jobs}
+                        {node.max_jobs != null && (
+                          <span className="text-muted-foreground"> / {node.max_jobs}</span>
+                        )}
+                      </TableCell>
+                    )}
+                    {type === "proxy" && (
+                      <TableCell className="text-sm whitespace-nowrap">
+                        {formatMbps(node.egress_kbps)}
+                        {node.max_bandwidth_kbps != null && (
+                          <span className="text-muted-foreground">
+                            {" "}
+                            / {formatMbps(node.max_bandwidth_kbps)}
+                          </span>
+                        )}{" "}
+                        Mbps
+                      </TableCell>
+                    )}
                    <TableCell className="text-muted-foreground text-xs">
                      {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({
        )}
      </div>

+      <div className="space-y-2">
+        <Label>Group</Label>
+        <Input value={group} onChange={(e) => setGroup(e.target.value)} placeholder="e.g. rack-1" />
+        <p className="text-muted-foreground text-sm">
+          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.
+        </p>
+      </div>
+
+      <div className="space-y-2">
+        <Label>{nodeType === "proxy" ? "Max Streams" : "Max Transcodes"}</Label>
+        <Input
+          type="number"
+          min={0}
+          value={maxJobs}
+          onChange={(e) => setMaxJobs(e.target.value)}
+          placeholder="Unlimited"
+        />
+        <p className="text-muted-foreground text-sm">
+          Optional concurrency cap for this node. Leave empty (or 0) for unlimited.
+        </p>
+      </div>
+
+      {nodeType === "proxy" && (
+        <div className="space-y-2">
+          <Label>Max Egress Bandwidth (Mbps)</Label>
+          <Input
+            type="number"
+            min={0}
+            step="any"
+            value={maxBandwidthMbps}
+            onChange={(e) => setMaxBandwidthMbps(e.target.value)}
+            placeholder="Unlimited"
+          />
+          <p className="text-muted-foreground text-sm">
+            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.
+          </p>
+        </div>
+      )}
+
      <Button type="submit" className="w-full" disabled={isPending}>
        {isPending ? "Saving..." : "Save"}
      </Button>
@@ -329,7 +422,7 @@ export default function AdminNodes() {
      <NodeSection
        type="proxy"
        nodes={proxyNodes}
-        showJobs={false}
+        showJobs={true}
        onAdd={() => handleAdd("proxy")}
        onEdit={handleEdit}
        onDelete={handleDelete}
@@ -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)}
          />
+          <SettingField
+            label="Local Transcode Fallback"
+            type="toggle"
+            hint="When no eligible transcode node is available, transcode on this server instead. Disable to keep all transcoding on dedicated nodes — playback that requires transcoding fails while no node is eligible."
+            value={form.getValue("playback.local_transcode_fallback") || "true"}
+            onChange={(v) => form.setValue("playback.local_transcode_fallback", v)}
+          />
          <SettingField
            label="Allow HEVC Encoding"
            type="toggle"
@@ -24,6 +24,7 @@ import type { WatchTogetherRoomConnectionResult } from "../hooks/useWatchTogethe
 import { getPersistedVolume, persistVolume } from "./VolumeControl";
 import { usePlayerConfig } from "../context/PlayerConfigContext";
 import { deriveDisplayedPlaybackState } from "../playback-info";
+import { preconnectToStreamOrigin } from "../stream-url";
 import { WatchTogetherPanel } from "./WatchTogetherPanel";
 import type {
  PlaybackRealtimeCommandEnvelope,
@@ -466,6 +467,13 @@ export function VideoPlayer({
    compatibilityFallbackKeyRef.current = null;
  }, [sessionId]);

+  // Warm the connection to the stream origin (a proxy node in distributed
+  // deployments) while the transcode start request is still in flight, so
+  // the first manifest fetch doesn't pay DNS/TCP/TLS handshakes.
+  useEffect(() => {
+    if (streamUrl) preconnectToStreamOrigin(streamUrl);
+  }, [streamUrl]);
+
  useEffect(() => {
    setPendingSeekTime(null);
  }, [effectiveStreamUrl]);
@@ -1,5 +1,34 @@
 import type { PlayMethod } from "./types";

+const preconnectedOrigins = new Set<string>();
+
+/**
+ * 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,