diff --git a/internal/api/handlers/nodes.go b/internal/api/handlers/nodes.go index 79690c90..c306cc8d 100644 --- a/internal/api/handlers/nodes.go +++ b/internal/api/handlers/nodes.go @@ -15,10 +15,45 @@ import ( "github.com/Silo-Server/silo-server/internal/cache" "github.com/Silo-Server/silo-server/internal/nodepool" + "github.com/Silo-Server/silo-server/internal/nodesessions" + "github.com/Silo-Server/silo-server/internal/playback" + "github.com/Silo-Server/silo-server/internal/streammonitor" "github.com/go-chi/chi/v5" "github.com/redis/go-redis/v9" ) +// LiveLocalSessions maps the in-process playback sessions into monitoring +// records (the same SessionInfo shape edge nodes write to Redis), so integrated +// single-node streams — which never touch Redis — are still visible to the +// monitor and the admin active-streams view. Shared by the enforcer's FuncSource +// and the admin session list so there is exactly one Session→record mapping. +func LiveLocalSessions(sm *playback.SessionManager, nodeName string) []nodesessions.SessionInfo { + if sm == nil { + return nil + } + live := sm.AllSessions() + out := make([]nodesessions.SessionInfo, 0, len(live)) + for _, s := range live { + out = append(out, nodesessions.SessionInfo{ + SessionID: s.ID, + NodeName: nodeName, + AuthUserID: s.UserID, + ProfileID: s.ProfileID, + Type: string(s.PlayMethod), + Route: s.Origin(), + MediaFileID: s.MediaFileID, + ClientIP: s.ClientIP, + ClientName: s.ClientName, + Position: s.Position, + Resolution: s.TargetResolution, + HWAccel: s.TranscodeHWAccel, + StartedAt: s.StartedAt.UTC().Format(time.RFC3339), + LastServedAt: s.LastActivityAt.UTC().Format(time.RFC3339), + }) + } + return out +} + // NodeRepository defines the operations the NodeHandler needs on the node store. type NodeRepository interface { List(ctx context.Context) ([]*nodepool.Node, error) @@ -43,6 +78,18 @@ type NodeHandler struct { eventBus cache.EventBus redisClient *redis.Client // for reading session keys jwtSecret string // for bearer auth when calling force-reload on nodes + + // sessionMgr + localNodeName let the session list include integrated + // single-node streams (which never write Redis). Optional; nil in edge modes. + sessionMgr *playback.SessionManager + localNodeName string +} + +// SetLocalSessionSource wires the in-process session manager so HandleListSessions +// can union integrated streams with the Redis-backed edge records. +func (h *NodeHandler) SetLocalSessionSource(sm *playback.SessionManager, nodeName string) { + h.sessionMgr = sm + h.localNodeName = nodeName } // NewNodeHandler creates a new NodeHandler. @@ -301,51 +348,69 @@ func (h *NodeHandler) HandleForceReloadNode(w http.ResponseWriter, r *http.Reque } // HandleListSessions handles GET /admin/nodes/sessions — lists active playback -// sessions from Redis, optionally filtered by node_id query parameter. +// sessions, unioning the Redis-backed edge records with the in-process +// integrated sessions (which never write Redis) so a single-node deployment is +// not blind. The union is deduped by session id (the same stream is tracked by +// the central manager AND by the edge serving it — one row per stream, like the +// enforcer's monitoring picture, so an operator never sees a single stream +// counted twice against a cap). Optionally filtered by node_id: a filter +// targets a specific edge node, so integrated sessions are only included in the +// unfiltered listing. func (h *NodeHandler) HandleListSessions(w http.ResponseWriter, r *http.Request) { - if h.redisClient == nil { - writeError(w, http.StatusServiceUnavailable, "redis_unavailable", "Redis not configured") - return - } - ctx := r.Context() - pattern := "silo:sessions:*" + nodeFilter := r.URL.Query().Get("node_id") - if nodeIDStr := r.URL.Query().Get("node_id"); nodeIDStr != "" { - nodeID, err := strconv.Atoi(nodeIDStr) - if err == nil { - node, err := h.repo.GetByID(ctx, nodeID) - if err == nil { - hashBytes := sha256.Sum256([]byte(node.URL)) - nodeHash := hex.EncodeToString(hashBytes[:4]) - pattern = "silo:sessions:" + nodeHash + ":*" + var infos []nodesessions.SessionInfo + + if h.redisClient != nil { + pattern := nodesessions.KeyPrefix + "*" + if nodeFilter != "" { + if nodeID, err := strconv.Atoi(nodeFilter); err == nil { + if node, err := h.repo.GetByID(ctx, nodeID); err == nil { + hashBytes := sha256.Sum256([]byte(node.URL)) + nodeHash := hex.EncodeToString(hashBytes[:4]) + pattern = nodesessions.KeyPrefix + nodeHash + ":*" + } } } - } - var sessions []json.RawMessage - var cursor uint64 - for { - keys, next, err := h.redisClient.Scan(ctx, cursor, pattern, 100).Result() - if err != nil { - writeError(w, http.StatusInternalServerError, "redis_error", err.Error()) - return - } - for _, key := range keys { - val, err := h.redisClient.Get(ctx, key).Result() + var cursor uint64 + for { + keys, next, err := h.redisClient.Scan(ctx, cursor, pattern, 100).Result() if err != nil { - continue + writeError(w, http.StatusInternalServerError, "redis_error", err.Error()) + return + } + for _, key := range keys { + val, err := h.redisClient.Get(ctx, key).Result() + if err != nil { + continue + } + var info nodesessions.SessionInfo + if err := json.Unmarshal([]byte(val), &info); err != nil { + slog.Debug("skip malformed session record", "key", key, "error", err) + continue + } + infos = append(infos, info) + } + cursor = next + if cursor == 0 { + break } - sessions = append(sessions, json.RawMessage(val)) - } - cursor = next - if cursor == 0 { - break } } - if sessions == nil { - sessions = []json.RawMessage{} + // Integrated single-node streams live only in the in-process session manager. + // Include them in the unfiltered listing (a node_id filter targets an edge). + if h.sessionMgr != nil && nodeFilter == "" { + infos = append(infos, LiveLocalSessions(h.sessionMgr, h.localNodeName)...) + } + + sessions := []json.RawMessage{} + for _, info := range streammonitor.DedupeSessionInfos(infos) { + if data, err := json.Marshal(info); err == nil { + sessions = append(sessions, json.RawMessage(data)) + } } type sessionsResponse struct { diff --git a/internal/api/handlers/playback.go b/internal/api/handlers/playback.go index 9ee4350b..75dc2503 100644 --- a/internal/api/handlers/playback.go +++ b/internal/api/handlers/playback.go @@ -612,7 +612,9 @@ func (h *PlaybackHandler) signSessionToken(card playback.RecipeCard) string { if h.JWTSecret == "" { return "" } - token, err := streamtoken.Sign(card.ToClaims(), h.JWTSecret, playback.MaxTokenTTL) + claims := card.ToClaims() + claims.Origin = playback.OriginNative // native-only mint site + token, err := streamtoken.Sign(claims, h.JWTSecret, playback.MaxTokenTTL) if err != nil { slog.Warn("sign stream token failed", "error", err, "session", card.SessionID, "playback_session_id", card.SessionID) return "" @@ -1952,6 +1954,8 @@ func (h *PlaybackHandler) handleStartPlaybackLegacy(w http.ResponseWriter, r *ht UserID: session.UserID, ProfileID: session.ProfileID, MediaFileID: session.MediaFileID, + Origin: playback.OriginNative, + ClientName: session.ClientName, } // Resolve media path if possible. @@ -2603,6 +2607,11 @@ func (h *PlaybackHandler) HandleChangeAudioTrack(w http.ResponseWriter, r *http. SubtitleCodec: subtitleCodec, TotalDuration: float64(file.Duration), RequireReady: atomicLegacyReplacement, + AuthUserID: updatedSession.UserID, + ProfileID: updatedSession.ProfileID, + MediaFileID: updatedSession.MediaFileID, + Route: updatedSession.Origin(), + ClientName: updatedSession.ClientName, } if strings.TrimSpace(nodeReq.HWAccel) == "" { nodeReq.HWAccel = h.playbackConfig().HWAccel @@ -2728,6 +2737,8 @@ func (h *PlaybackHandler) HandleChangeAudioTrack(w http.ResponseWriter, r *http. // mode; a re-minted token must not silently shed either. TranscodeTransportID: updatedSession.TranscodeTransportID, RemuxDVMode: string(updatedSession.RemuxDVMode), + Origin: updatedSession.Origin(), + ClientName: updatedSession.ClientName, } if plan.TranscodeNode != nil { tokenClaims.TranscodeNode = plan.TranscodeNode.URL @@ -3273,6 +3284,11 @@ func (h *PlaybackHandler) HandleStartTranscode(w http.ResponseWriter, r *http.Re SubtitleCodec: subtitleCodec, TotalDuration: float64(file.Duration), RequireReady: atomicLegacyReplacement, + AuthUserID: session.UserID, + ProfileID: session.ProfileID, + MediaFileID: session.MediaFileID, + Route: session.Origin(), + ClientName: session.ClientName, } nodeResp, status, err := h.startRemotePlaybackTransport(context.Background(), tcNode.URL, nodeReq) @@ -3755,6 +3771,19 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht w.Header().Set("Cache-Control", "no-store, max-age=0") w.Header().Set("Pragma", "no-cache") + // Hold a transport marker for the whole segment pour so a slow client draining + // a single segment past the liveness grace cannot be reaped mid-serve and go + // invisible to the monitor — matching the direct-play/remux handlers. This is + // server-observed liveness: it keeps the session visible independent of any + // client progress report (the hidden-stream defense). + if err := h.sessionMgr.BeginTransport(sessionID); err == nil { + defer func() { _ = h.sessionMgr.EndTransport(sessionID) }() + } else { + // The marker is what keeps a slow single-segment drain visible; the + // segment is still served on failure, so log rather than fail — but log so + // the hidden-stream window is observable. + slog.Warn("begin transport marker failed", "session", sessionID, "segment", segmentPath, "error", err) + } http.ServeFile(w, r, segmentPath) } diff --git a/internal/nodesessions/tracker.go b/internal/nodesessions/tracker.go index cac36b12..a757e97c 100644 --- a/internal/nodesessions/tracker.go +++ b/internal/nodesessions/tracker.go @@ -13,7 +13,12 @@ import ( ) const ( - keyPrefix = "silo:sessions:" + // KeyPrefix is the Redis key namespace for node session records + // (silo:sessions:{nodeHash}:{sessionID}). Exported so readers of the + // monitoring picture (streammonitor, the admin session list) SCAN the same + // namespace this tracker writes instead of duplicating the literal. + KeyPrefix = "silo:sessions:" + sessionTTL = 60 * time.Second refreshInt = 30 * time.Second ) @@ -33,6 +38,14 @@ type SessionInfo struct { HWAccel string `json:"hw_accel,omitempty"` StartedAt string `json:"started_at"` + // LastServedAt / BytesServed are the authoritative, server-observed liveness + // signals: they are refreshed only when the node actually serves bytes for this + // session (a segment written, a direct-play/remux pour advancing), never from a + // client progress report. A "quiet" stream that keeps pulling stays fresh here; + // one that stops pulling ages out. This is the anti-abuse monitoring surface. + LastServedAt string `json:"last_served_at,omitempty"` + BytesServed int64 `json:"bytes_served,omitempty"` + // AuthUserID / ProfileID / MediaFileID are the numeric ownership keys the // node copies from the verified stream token. They enrich the live admin // "active streams" view (served by SCANning these records) so it can answer @@ -41,6 +54,17 @@ type SessionInfo struct { AuthUserID int `json:"auth_user_id,omitempty"` ProfileID string `json:"profile_id,omitempty"` MediaFileID int `json:"media_file_id,omitempty"` + + // Route is the origin protocol ("native" | "jellycompat"). Type is the play + // method (direct_play/remux/transcode); Route is orthogonal — a jellycompat + // stream and a native one share a Type but differ in Route. Client* are + // best-effort viewer identity for oversight and abuse heuristics (e.g. one + // re-streaming client fanning out to many viewers). Position is the last + // known playback position in seconds (secondary timing). + Route string `json:"route,omitempty"` + ClientIP string `json:"client_ip,omitempty"` + ClientName string `json:"client_name,omitempty"` + Position float64 `json:"position,omitempty"` } // Tracker manages session lifecycle in Redis for a single node. @@ -52,8 +76,10 @@ 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 - touched map[string]time.Time // ephemeral sessions by last-activity time + sessions map[string]struct{} // set of active (explicitly tracked) session IDs + touched map[string]time.Time // ephemeral sessions by last-activity time + records map[string]SessionInfo // last-written record per session, for enriched refresh + bytes map[string]int64 // cumulative bytes served per session (monitoring only) } // NewTracker creates a session tracker for the given node. @@ -68,12 +94,14 @@ func NewTracker(rdb *redis.Client, nodeURL, nodeName, nodeType string) *Tracker nodeHash: hex.EncodeToString(h[:4]), // 8 hex chars sessions: make(map[string]struct{}), touched: make(map[string]time.Time), + records: make(map[string]SessionInfo), + bytes: make(map[string]int64), } } // redisKey returns the full Redis key for a session. func (tr *Tracker) redisKey(sessionID string) string { - return keyPrefix + tr.nodeHash + ":" + sessionID + return KeyPrefix + tr.nodeHash + ":" + sessionID } // NodeHash returns the node's hash prefix used in Redis keys. @@ -109,12 +137,64 @@ func (tr *Tracker) ActiveCount() int { return count } +// Snapshot returns a copy of the currently-known session records on this node +// (including ephemeral ones still within their TTL), stamped with the latest +// LastServedAt/BytesServed. Used for the node /status view; central aggregates +// the equivalent picture by reading Redis. +func (tr *Tracker) Snapshot() []SessionInfo { + tr.mu.Lock() + defer tr.mu.Unlock() + now := time.Now() + out := make([]SessionInfo, 0, len(tr.records)) + for id, rec := range tr.records { + // Session-backed entries are always live until Remove; only ephemeral + // (non-session) entries age out by idle timeout. + if _, isSession := tr.sessions[id]; !isSession { + if last, ok := tr.touched[id]; ok && now.Sub(last) > sessionTTL { + continue + } + } + out = append(out, tr.enrichLocked(id, rec)) + } + return out +} + +// enrichLocked stamps a record with the latest observed liveness/bytes. Caller +// holds tr.mu. +func (tr *Tracker) enrichLocked(id string, rec SessionInfo) SessionInfo { + if last, ok := tr.touched[id]; ok { + rec.LastServedAt = last.UTC().Format(time.RFC3339) + } + if b, ok := tr.bytes[id]; ok { + rec.BytesServed = b + } + return rec +} + // Track registers an active session in Redis with a TTL. func (tr *Tracker) Track(ctx context.Context, info SessionInfo) { if tr.rdb == nil { return } - data, err := json.Marshal(info) + now := time.Now() + if info.LastServedAt == "" { + info.LastServedAt = now.UTC().Format(time.RFC3339) + } + + // Explicitly-tracked sessions (direct play / remux) live in tr.sessions for + // the whole connection. touched holds their last-served time so LastServedAt + // reflects real activity; refreshAll only idle-prunes ephemeral (non-session) + // touched entries, so a long quiet pour is never expired while its connection + // is open. + tr.mu.Lock() + tr.preserveStartedAtLocked(&info) + tr.sessions[info.SessionID] = struct{}{} + tr.records[info.SessionID] = info + tr.touched[info.SessionID] = now + enriched := tr.enrichLocked(info.SessionID, info) + tr.mu.Unlock() + + data, err := json.Marshal(enriched) if err != nil { slog.DebugContext(ctx, "session track marshal failed", "component", "nodesessions", "error", err) return @@ -124,29 +204,31 @@ func (tr *Tracker) Track(ctx context.Context, info SessionInfo) { slog.DebugContext(ctx, "session track set failed", "component", "nodesessions", "error", err, "session", info.SessionID) return } - - tr.mu.Lock() - tr.sessions[info.SessionID] = struct{}{} - 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). +// sessionTTL without further touches (pruned by the refresh loop). Subsequent +// touches update LastServedAt in memory; the body is re-flushed on the refresh +// tick so the monitoring record reflects real serve activity. func (tr *Tracker) Touch(ctx context.Context, info SessionInfo) { if tr.rdb == nil { return } + now := time.Now() tr.mu.Lock() + tr.preserveStartedAtLocked(&info) _, known := tr.touched[info.SessionID] - tr.touched[info.SessionID] = time.Now() + tr.touched[info.SessionID] = now + tr.records[info.SessionID] = info + enriched := tr.enrichLocked(info.SessionID, info) tr.mu.Unlock() if known { return } - data, err := json.Marshal(info) + data, err := json.Marshal(enriched) if err != nil { slog.DebugContext(ctx, "session touch marshal failed", "component", "nodesessions", "error", err) return @@ -156,6 +238,57 @@ func (tr *Tracker) Touch(ctx context.Context, info SessionInfo) { } } +// preserveStartedAtLocked keeps the first-seen StartedAt when a record is +// re-written for a session we already track. Track re-fires on every range +// reconnect and Touch stamps a fresh sessionInfo per segment fetch; without +// this, StartedAt degrades to "time of last request", corrupting the admin +// view and the enforcer's StartedAt tie-break in victim selection. Caller +// holds tr.mu. +func (tr *Tracker) preserveStartedAtLocked(info *SessionInfo) { + if prev, ok := tr.records[info.SessionID]; ok && prev.StartedAt != "" { + info.StartedAt = prev.StartedAt + } +} + +// AddBytes records that n bytes were served for a session and marks it as active +// now. Cheap and lock-guarded; the accumulated value is flushed to Redis on the +// refresh tick. Bytes attribution is best-effort monitoring and never gates the +// hot path. Bytes for a session with no live record are dropped: a late tally +// arriving after the record was idle-pruned (or removed) must not recreate a +// bytes entry nothing will ever clean up — that was a slow permanent leak on +// busy edges. +func (tr *Tracker) AddBytes(sessionID string, n int64) { + if tr.rdb == nil || n <= 0 { + return + } + tr.mu.Lock() + if _, known := tr.records[sessionID]; known { + tr.bytes[sessionID] += n + // Mark real serve activity so LastServedAt advances for every session + // type, including long direct-play/remux pours. + tr.touched[sessionID] = time.Now() + } + tr.mu.Unlock() +} + +// MarkServed records real serve activity for a session without byte +// attribution, advancing LastServedAt (flushed on the refresh tick). Used by +// the transcode node's own serve paths, where wrapping ServeFile for byte +// counting would cost the sendfile fast path for a signal the fronting proxy +// already measures — without it the node record's LastServedAt stays frozen at +// start time, which starves the enforcer's freshness ordering and the admin +// view of the node-side truth. +func (tr *Tracker) MarkServed(sessionID string) { + if tr == nil || tr.rdb == nil { + return + } + tr.mu.Lock() + if _, known := tr.records[sessionID]; known { + tr.touched[sessionID] = time.Now() + } + tr.mu.Unlock() +} + // Remove deletes a session from Redis and the in-memory set. func (tr *Tracker) Remove(ctx context.Context, sessionID string) { if tr.rdb == nil { @@ -164,6 +297,8 @@ func (tr *Tracker) Remove(ctx context.Context, sessionID string) { tr.mu.Lock() delete(tr.sessions, sessionID) delete(tr.touched, sessionID) + delete(tr.records, sessionID) + delete(tr.bytes, sessionID) tr.mu.Unlock() if err := tr.rdb.Del(ctx, tr.redisKey(sessionID)).Err(); err != nil { @@ -188,6 +323,8 @@ func (tr *Tracker) Cleanup(ctx context.Context) { } tr.sessions = make(map[string]struct{}) tr.touched = make(map[string]time.Time) + tr.records = make(map[string]SessionInfo) + tr.bytes = make(map[string]int64) tr.mu.Unlock() if len(ids) == 0 { @@ -225,31 +362,53 @@ func (tr *Tracker) StartRefresh(ctx context.Context) { func (tr *Tracker) refreshAll(ctx context.Context) { now := time.Now() + type flush struct { + id string + data []byte + } tr.mu.Lock() - ids := make([]string, 0, len(tr.sessions)+len(tr.touched)) + ids := make([]string, 0, len(tr.sessions)) 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. + _, isSession := tr.sessions[id] + if !isSession && now.Sub(last) > sessionTTL { + // Idle ephemeral session: stop refreshing and let the Redis key + // expire on its own. Session-backed entries (direct/remux) are pruned + // on Remove, never by idle timeout, so a quiet-but-open pour stays live. delete(tr.touched, id) + delete(tr.records, id) + delete(tr.bytes, id) continue } - if _, dup := tr.sessions[id]; !dup { + if !isSession { ids = append(ids, id) } } + flushes := make([]flush, 0, len(ids)) + for _, id := range ids { + rec, ok := tr.records[id] + if !ok { + continue + } + data, err := json.Marshal(tr.enrichLocked(id, rec)) + if err != nil { + continue + } + flushes = append(flushes, flush{id: id, data: data}) + } tr.mu.Unlock() - if len(ids) == 0 { + if len(flushes) == 0 { return } + // Re-SET (not just EXPIRE) so LastServedAt/BytesServed stay current in the + // monitoring record. 30s cadence keeps this cheap. pipe := tr.rdb.Pipeline() - for _, id := range ids { - pipe.Expire(ctx, tr.redisKey(id), sessionTTL) + for _, f := range flushes { + pipe.Set(ctx, tr.redisKey(f.id), f.data, sessionTTL) } if _, err := pipe.Exec(ctx); err != nil { slog.DebugContext(ctx, "session refresh pipeline failed", "component", "nodesessions", "error", err) diff --git a/internal/playback/session.go b/internal/playback/session.go index 9c66b5ed..f1c40397 100644 --- a/internal/playback/session.go +++ b/internal/playback/session.go @@ -135,6 +135,14 @@ var ErrSessionReplacementSuperseded = errors.New("session replacement was supers type clientInfoContextKey struct{} +// Origin identifies the protocol a playback session was started through, for +// first-class monitoring attribution (a jellycompat session shares this manager +// with native ones and is otherwise indistinguishable in the live snapshot). +const ( + OriginNative = "native" + OriginJellyfin = "jellycompat" +) + // ClientInfo carries best-effort client metadata from request handling into // the playback session manager. type ClientInfo struct { @@ -294,6 +302,17 @@ func normalizeClientMetadataValue(value string, maxLen int) string { return value } +// Origin reports the protocol this session was started through, as the label +// monitoring and stream tokens attribute by. It derives from the stored +// IsJellyfinCompat identity so the two can never disagree; native is the +// historical default for anything not minted through the compat layer. +func (s *Session) Origin() string { + if s == nil || !s.IsJellyfinCompat { + return OriginNative + } + return OriginJellyfin +} + // StartSession creates a new playback session using the same file as both the // requested and effective source. func (m *SessionManager) StartSession(userID int, profileID string, fileID int, method PlayMethod, transcodeAudio bool) (*Session, error) { diff --git a/internal/playback/session_test.go b/internal/playback/session_test.go index 0812abf2..98e3edbf 100644 --- a/internal/playback/session_test.go +++ b/internal/playback/session_test.go @@ -864,6 +864,35 @@ func TestStartSessionWithFiles(t *testing.T) { } } +func TestSessionOriginFromContext(t *testing.T) { + mgr := playback.NewSessionManager(0, 0) + + // A jellycompat-originated session records its origin for monitor attribution. + ctx := playback.WithClientInfo(context.Background(), playback.ClientInfo{ + Name: "Infuse", + IsCompat: true, + }) + s, err := mgr.StartSessionWithContext(ctx, 1, "p1", 100, playback.PlayTranscode, false) + if err != nil { + t.Fatalf("StartSessionWithContext: %v", err) + } + if s.Origin() != playback.OriginJellyfin { + t.Errorf("Origin = %q, want %q", s.Origin(), playback.OriginJellyfin) + } + if s.ClientName != "Infuse" { + t.Errorf("ClientName = %q, want Infuse", s.ClientName) + } + + // An unset origin defaults to native (historical behavior), never empty. + s2, err := mgr.StartSession(2, "p2", 101, playback.PlayDirect, false) + if err != nil { + t.Fatalf("StartSession: %v", err) + } + if s2.Origin() != playback.OriginNative { + t.Errorf("default Origin = %q, want %q", s2.Origin(), playback.OriginNative) + } +} + func TestUpdateStreamState(t *testing.T) { mgr := playback.NewSessionManager(0, 0) session, err := mgr.StartSession(1, "profile-1", 100, playback.PlayRemux, false) diff --git a/internal/streammonitor/monitor.go b/internal/streammonitor/monitor.go new file mode 100644 index 00000000..2686977d --- /dev/null +++ b/internal/streammonitor/monitor.go @@ -0,0 +1,386 @@ +// Package streammonitor produces a normalized "live streams" snapshot grouped by +// user from the authoritative server-side monitoring records written by +// internal/nodesessions. It backs an async enforcement loop and admin views. +// +// EXISTENCE is server-observed on every path and never gated by a client report, +// so a "hidden stream" (a disguised client that pulls bytes but withholds +// progress) is always counted: +// - Edge (multi-node): a record exists in Redis for the whole connection +// (direct/remux Track..Remove) or while segments are pulled (transcode Touch), +// and BytesServed/LastServedAt advance only on real bytes. +// - Integrated: the FuncSource reflects SessionManager.AllSessions(), and a +// session is unreapable while it holds an in-flight transport marker +// (BeginTransport/EndTransport around every byte pour) — no client progress +// required to stay visible. +// +// TIMING is a secondary signal. On the edge, LastServedAt is purely byte-observed. +// In integrated mode LastServedAt is mapped from SessionManager.LastActivityAt, +// which client progress reports also advance; that is acceptable because it is +// used only to order over-cap victims (selectVictims), never to decide whether a +// stream exists or is counted. See internal/nodesessions for record production. +package streammonitor + +import ( + "context" + "encoding/json" + "log/slog" + "time" + + "github.com/redis/go-redis/v9" + + "github.com/Silo-Server/silo-server/internal/nodesessions" +) + +// sessionKeyPrefix is the Redis key prefix under which nodesessions.Tracker +// stores its SessionInfo records — the tracker's own exported constant, so the +// write and read sides can never drift. +const sessionKeyPrefix = nodesessions.KeyPrefix + +// scanCount is the COUNT hint passed to SCAN. It bounds the amount of work per +// round trip while keeping the number of round trips reasonable. +const scanCount = 256 + +// LiveStream is a normalized view of a single active streaming session. +type LiveStream struct { + SessionID string + UserID int // from SessionInfo.AuthUserID + ProfileID string + NodeName string + NodeURL string + Type string // play method: direct_play | remux | transcode + Route string // origin protocol: native | jellycompat + MediaFileID int + ClientIP string + ClientName string + Position float64 // last known playback position (seconds); secondary timing + HWAccel string + LastServedAt time.Time // parsed from SessionInfo.LastServedAt (zero if absent) + BytesServed int64 + StartedAt time.Time // parsed from SessionInfo.StartedAt (zero if unparseable) +} + +// Snapshot is a point-in-time picture of the live streams. +type Snapshot struct { + Streams []LiveStream +} + +// CountByUser returns the number of live streams owned by userID. +func (s Snapshot) CountByUser(userID int) int { + n := 0 + for _, st := range s.Streams { + if st.UserID == userID { + n++ + } + } + return n +} + +// StreamsForUser returns the live streams owned by userID. +func (s Snapshot) StreamsForUser(userID int) []LiveStream { + out := make([]LiveStream, 0) + for _, st := range s.Streams { + if st.UserID == userID { + out = append(out, st) + } + } + return out +} + +// ByUser groups the live streams by owning user id. +func (s Snapshot) ByUser() map[int][]LiveStream { + out := make(map[int][]LiveStream) + for _, st := range s.Streams { + out[st.UserID] = append(out[st.UserID], st) + } + return out +} + +// Source yields the current live picture. +type Source interface { + Snapshot(ctx context.Context) (Snapshot, error) +} + +// MultiSource unions several sources into one snapshot, de-duplicating sessions +// that appear in more than one backend (e.g. a session held both in the central +// session manager and mirrored by an edge's Redis record) by keeping the most- +// recently-served copy. A source that errors is skipped (logged), so one +// unavailable backend never blinds the enforcer to the others. This is what lets +// the enforcer see BOTH locally-served (integrated) and edge-served (multi-node) +// streams regardless of whether Redis is configured. +type MultiSource struct { + sources []Source +} + +// NewMultiSource builds a union source over the given sources (nil entries are +// ignored). +func NewMultiSource(sources ...Source) *MultiSource { + return &MultiSource{sources: sources} +} + +// Snapshot merges every sub-source's snapshot, de-duplicated by session id. +func (m *MultiSource) Snapshot(ctx context.Context) (Snapshot, error) { + var all []LiveStream + for _, src := range m.sources { + if src == nil { + continue + } + snap, err := src.Snapshot(ctx) + if err != nil { + slog.Warn("streammonitor: source snapshot failed; skipping", "error", err) + continue + } + all = append(all, snap.Streams...) + } + return Snapshot{Streams: mergeStreams(all)}, nil +} + +// toLiveStream converts a nodesessions.SessionInfo into a LiveStream, parsing +// the RFC3339 timestamps and tolerating empty/unparseable values (which map to +// the zero time). +func toLiveStream(info nodesessions.SessionInfo) LiveStream { + return LiveStream{ + SessionID: info.SessionID, + UserID: info.AuthUserID, + ProfileID: info.ProfileID, + NodeName: info.NodeName, + NodeURL: info.NodeURL, + Type: info.Type, + Route: info.Route, + MediaFileID: info.MediaFileID, + ClientIP: info.ClientIP, + ClientName: info.ClientName, + Position: info.Position, + HWAccel: info.HWAccel, + LastServedAt: parseTime(info.LastServedAt), + BytesServed: info.BytesServed, + StartedAt: parseTime(info.StartedAt), + } +} + +// parseTime parses an RFC3339 timestamp, returning the zero time for empty or +// unparseable input. +func parseTime(s string) time.Time { + if s == "" { + return time.Time{} + } + t, err := time.Parse(time.RFC3339, s) + if err != nil { + slog.Debug("streammonitor: unparseable timestamp", "value", s, "error", err) + return time.Time{} + } + return t +} + +// mergeStreams collapses records that share a SessionID (the same session can be +// tracked by more than one node — e.g. a proxy record and a transcode-node +// record), keeping the one with the most recent LastServedAt. Records with a +// distinct SessionID are all retained. The relative order of the kept records is +// not guaranteed. +// +// Ownership and attribution are carried forward independently of the freshness +// pick: the transcode node's own start record has no resolved owner (UserID 0) +// and thinner attribution, while the proxy record fronting it does. If the +// ownerless node record happened to be the freshest copy, taking it wholesale +// would bucket the session under user 0 (which the enforcer skips — silently +// exempting it from the concurrency cap) and drop route/client detail from the +// monitor view. So a merged record adopts a resolved owner from either candidate +// and backfills any empty attribution/display field from the other copy. +func mergeStreams(streams []LiveStream) []LiveStream { + bySession := make(map[string]LiveStream, len(streams)) + for _, st := range streams { + existing, ok := bySession[st.SessionID] + if !ok { + bySession[st.SessionID] = st + continue + } + winner := existing + other := st + if st.LastServedAt.After(existing.LastServedAt) { + winner, other = st, existing + } + // If the freshest copy is ownerless, adopt a resolved owner from either + // candidate so the session is still attributed (and enforced) correctly. + if winner.UserID <= 0 { + for _, cand := range []LiveStream{existing, st} { + if cand.UserID > 0 { + winner.UserID = cand.UserID + winner.ProfileID = cand.ProfileID + winner.MediaFileID = cand.MediaFileID + break + } + } + } + // Backfill display/attribution fields the freshest copy lacks so the + // merged record is as complete as possible for the monitor view. + if winner.Route == "" { + winner.Route = other.Route + } + if winner.ClientIP == "" { + winner.ClientIP = other.ClientIP + } + if winner.ClientName == "" { + winner.ClientName = other.ClientName + } + if winner.HWAccel == "" { + winner.HWAccel = other.HWAccel + } + if winner.Position == 0 { + winner.Position = other.Position + } + bySession[st.SessionID] = winner + } + out := make([]LiveStream, 0, len(bySession)) + for _, st := range bySession { + out = append(out, st) + } + return out +} + +// DedupeSessionInfos collapses raw monitoring records that share a SessionID, +// applying the same rules as mergeStreams — keep the most-recently-served copy, +// carry a resolved owner forward, backfill missing attribution — but preserving +// the SessionInfo shape for surfaces whose wire format IS the raw record (the +// admin session list, which unions Redis edge records with the in-process +// integrated sessions and would otherwise show the same stream twice). Kept as +// a sibling of mergeStreams rather than a shared generic because mergeStreams +// operates on the parsed LiveStream form; keep the two rule sets in sync. +func DedupeSessionInfos(infos []nodesessions.SessionInfo) []nodesessions.SessionInfo { + bySession := make(map[string]nodesessions.SessionInfo, len(infos)) + order := make([]string, 0, len(infos)) + for _, in := range infos { + existing, ok := bySession[in.SessionID] + if !ok { + bySession[in.SessionID] = in + order = append(order, in.SessionID) + continue + } + winner, other := existing, in + if parseTime(in.LastServedAt).After(parseTime(existing.LastServedAt)) { + winner, other = in, existing + } + if winner.AuthUserID <= 0 && other.AuthUserID > 0 { + winner.AuthUserID = other.AuthUserID + winner.ProfileID = other.ProfileID + winner.MediaFileID = other.MediaFileID + } + if winner.Route == "" { + winner.Route = other.Route + } + if winner.ClientIP == "" { + winner.ClientIP = other.ClientIP + } + if winner.ClientName == "" { + winner.ClientName = other.ClientName + } + if winner.HWAccel == "" { + winner.HWAccel = other.HWAccel + } + if winner.Position == 0 { + winner.Position = other.Position + } + bySession[in.SessionID] = winner + } + out := make([]nodesessions.SessionInfo, 0, len(bySession)) + for _, id := range order { + out = append(out, bySession[id]) + } + return out +} + +// RedisSource reads silo:sessions:* records, producing the multi-node +// authoritative picture. +type RedisSource struct { + rdb *redis.Client +} + +// NewRedisSource creates a RedisSource backed by rdb. +func NewRedisSource(rdb *redis.Client) *RedisSource { + return &RedisSource{rdb: rdb} +} + +// Snapshot SCANs every silo:sessions:* record, decodes it, and returns the +// deduped live picture. The same session appearing on multiple nodes is +// collapsed to the record with the most recent LastServedAt. +func (r *RedisSource) Snapshot(ctx context.Context) (Snapshot, error) { + if r.rdb == nil { + return Snapshot{Streams: []LiveStream{}}, nil + } + + // Fully iterate the cursor to collect matching keys. SCAN (not KEYS) keeps + // this non-blocking against a large keyspace. + var keys []string + var cursor uint64 + for { + batch, next, err := r.rdb.Scan(ctx, cursor, sessionKeyPrefix+"*", scanCount).Result() + if err != nil { + return Snapshot{}, err + } + keys = append(keys, batch...) + cursor = next + if cursor == 0 { + break + } + } + + if len(keys) == 0 { + return Snapshot{Streams: []LiveStream{}}, nil + } + + vals, err := r.rdb.MGet(ctx, keys...).Result() + if err != nil { + return Snapshot{}, err + } + + streams := make([]LiveStream, 0, len(vals)) + for i, v := range vals { + if v == nil { + // Key expired between SCAN and MGET; skip. + continue + } + var raw string + switch val := v.(type) { + case string: + raw = val + case []byte: + raw = string(val) + default: + slog.Debug("streammonitor: unexpected redis value type", "key", keys[i]) + continue + } + var info nodesessions.SessionInfo + if err := json.Unmarshal([]byte(raw), &info); err != nil { + slog.Debug("streammonitor: unmarshal session record failed", "key", keys[i], "error", err) + continue + } + streams = append(streams, toLiveStream(info)) + } + + return Snapshot{Streams: mergeStreams(streams)}, nil +} + +// FuncSource adapts an in-process provider (integrated single-node) that returns +// the local tracker records. +type FuncSource struct { + fn func(ctx context.Context) ([]nodesessions.SessionInfo, error) +} + +// NewFuncSource creates a FuncSource backed by fn. +func NewFuncSource(fn func(ctx context.Context) ([]nodesessions.SessionInfo, error)) *FuncSource { + return &FuncSource{fn: fn} +} + +// Snapshot invokes the wrapped provider and returns the deduped live picture. +func (f *FuncSource) Snapshot(ctx context.Context) (Snapshot, error) { + if f.fn == nil { + return Snapshot{Streams: []LiveStream{}}, nil + } + infos, err := f.fn(ctx) + if err != nil { + return Snapshot{}, err + } + streams := make([]LiveStream, 0, len(infos)) + for _, info := range infos { + streams = append(streams, toLiveStream(info)) + } + return Snapshot{Streams: mergeStreams(streams)}, nil +} diff --git a/internal/streammonitor/monitor_test.go b/internal/streammonitor/monitor_test.go new file mode 100644 index 00000000..a8c2748a --- /dev/null +++ b/internal/streammonitor/monitor_test.go @@ -0,0 +1,276 @@ +package streammonitor + +import ( + "context" + "testing" + "time" + + "github.com/Silo-Server/silo-server/internal/nodesessions" +) + +func fakeFn(infos []nodesessions.SessionInfo) func(ctx context.Context) ([]nodesessions.SessionInfo, error) { + return func(ctx context.Context) ([]nodesessions.SessionInfo, error) { + return infos, nil + } +} + +func TestFuncSourceGrouping(t *testing.T) { + infos := []nodesessions.SessionInfo{ + {SessionID: "s1", AuthUserID: 1, Type: "direct_play", StartedAt: "2026-07-04T10:00:00Z", LastServedAt: "2026-07-04T10:05:00Z"}, + {SessionID: "s2", AuthUserID: 1, Type: "transcode", StartedAt: "2026-07-04T10:01:00Z", LastServedAt: "2026-07-04T10:06:00Z"}, + {SessionID: "s3", AuthUserID: 2, Type: "remux", StartedAt: "2026-07-04T10:02:00Z", LastServedAt: "2026-07-04T10:07:00Z"}, + {SessionID: "s4", AuthUserID: 0, Type: "direct_play"}, // no user, no timestamps + } + + snap, err := NewFuncSource(fakeFn(infos)).Snapshot(context.Background()) + if err != nil { + t.Fatalf("Snapshot: %v", err) + } + + if got := len(snap.Streams); got != 4 { + t.Fatalf("Streams len = %d, want 4", got) + } + + if got := snap.CountByUser(1); got != 2 { + t.Errorf("CountByUser(1) = %d, want 2", got) + } + if got := snap.CountByUser(2); got != 1 { + t.Errorf("CountByUser(2) = %d, want 1", got) + } + if got := snap.CountByUser(0); got != 1 { + t.Errorf("CountByUser(0) = %d, want 1", got) + } + if got := snap.CountByUser(99); got != 0 { + t.Errorf("CountByUser(99) = %d, want 0", got) + } + + u1 := snap.StreamsForUser(1) + if len(u1) != 2 { + t.Fatalf("StreamsForUser(1) len = %d, want 2", len(u1)) + } + for _, st := range u1 { + if st.UserID != 1 { + t.Errorf("StreamsForUser(1) returned stream with UserID %d", st.UserID) + } + } + + byUser := snap.ByUser() + if len(byUser) != 3 { + t.Fatalf("ByUser len = %d, want 3 (users 0,1,2)", len(byUser)) + } + if len(byUser[1]) != 2 || len(byUser[2]) != 1 || len(byUser[0]) != 1 { + t.Errorf("ByUser grouping wrong: %#v", map[int]int{0: len(byUser[0]), 1: len(byUser[1]), 2: len(byUser[2])}) + } +} + +func TestTimestampParsing(t *testing.T) { + infos := []nodesessions.SessionInfo{ + {SessionID: "s1", AuthUserID: 1, StartedAt: "2026-07-04T10:00:00Z", LastServedAt: "2026-07-04T10:05:00Z"}, + {SessionID: "s2", AuthUserID: 1, StartedAt: "not-a-time", LastServedAt: ""}, + } + snap, err := NewFuncSource(fakeFn(infos)).Snapshot(context.Background()) + if err != nil { + t.Fatalf("Snapshot: %v", err) + } + bySession := map[string]LiveStream{} + for _, st := range snap.Streams { + bySession[st.SessionID] = st + } + + want := time.Date(2026, 7, 4, 10, 5, 0, 0, time.UTC) + if !bySession["s1"].LastServedAt.Equal(want) { + t.Errorf("s1 LastServedAt = %v, want %v", bySession["s1"].LastServedAt, want) + } + if !bySession["s2"].StartedAt.IsZero() { + t.Errorf("s2 StartedAt = %v, want zero (unparseable)", bySession["s2"].StartedAt) + } + if !bySession["s2"].LastServedAt.IsZero() { + t.Errorf("s2 LastServedAt = %v, want zero (empty)", bySession["s2"].LastServedAt) + } +} + +func TestDedupeKeepsNewest(t *testing.T) { + // Same SessionID observed on two nodes (proxy vs transcode node). Keep the + // record with the most recent LastServedAt. + infos := []nodesessions.SessionInfo{ + {SessionID: "dup", AuthUserID: 5, NodeName: "proxy", LastServedAt: "2026-07-04T10:00:00Z", BytesServed: 100}, + {SessionID: "dup", AuthUserID: 5, NodeName: "transcode", LastServedAt: "2026-07-04T10:09:00Z", BytesServed: 900}, + {SessionID: "other", AuthUserID: 5, NodeName: "proxy", LastServedAt: "2026-07-04T10:03:00Z"}, + } + snap, err := NewFuncSource(fakeFn(infos)).Snapshot(context.Background()) + if err != nil { + t.Fatalf("Snapshot: %v", err) + } + + if got := len(snap.Streams); got != 2 { + t.Fatalf("Streams len = %d, want 2 (deduped)", got) + } + if got := snap.CountByUser(5); got != 2 { + t.Errorf("CountByUser(5) = %d, want 2", got) + } + + var dup LiveStream + found := false + for _, st := range snap.Streams { + if st.SessionID == "dup" { + dup = st + found = true + } + } + if !found { + t.Fatal("deduped session 'dup' not present") + } + if dup.NodeName != "transcode" { + t.Errorf("dedupe kept NodeName %q, want transcode (newest LastServedAt)", dup.NodeName) + } + if dup.BytesServed != 900 { + t.Errorf("dedupe kept BytesServed %d, want 900", dup.BytesServed) + } +} + +func TestMergeStreamsDirect(t *testing.T) { + // Direct unit test of the merge helper, mirroring RedisSource dedupe. + in := []LiveStream{ + {SessionID: "a", NodeName: "n1", LastServedAt: time.Unix(100, 0)}, + {SessionID: "a", NodeName: "n2", LastServedAt: time.Unix(200, 0)}, + {SessionID: "b", NodeName: "n1", LastServedAt: time.Unix(150, 0)}, + } + out := mergeStreams(in) + if len(out) != 2 { + t.Fatalf("mergeStreams len = %d, want 2", len(out)) + } + for _, st := range out { + if st.SessionID == "a" && st.NodeName != "n2" { + t.Errorf("merge kept %q for session a, want n2 (newest)", st.NodeName) + } + } +} + +func TestMergeStreamsCarriesOwnershipForward(t *testing.T) { + // The transcode node's own start record has no resolved owner (UserID 0) and + // can be the freshest copy of a session. Taking it wholesale would bucket the + // session under user 0, which the enforcer skips — exempting it from the cap. + // The merge must recover the owner from the proxy's (staler) owned record. + in := []LiveStream{ + {SessionID: "s", NodeName: "proxy", UserID: 42, ProfileID: "p1", MediaFileID: 7, LastServedAt: time.Unix(100, 0)}, + {SessionID: "s", NodeName: "transcode", UserID: 0, LastServedAt: time.Unix(200, 0)}, + } + out := mergeStreams(in) + if len(out) != 1 { + t.Fatalf("mergeStreams len = %d, want 1", len(out)) + } + got := out[0] + if got.NodeName != "transcode" { + t.Errorf("kept NodeName %q, want transcode (freshest)", got.NodeName) + } + if got.UserID != 42 { + t.Errorf("UserID = %d, want 42 (recovered from the proxy record)", got.UserID) + } + if got.ProfileID != "p1" || got.MediaFileID != 7 { + t.Errorf("ownership not fully recovered: ProfileID=%q MediaFileID=%d", got.ProfileID, got.MediaFileID) + } +} + +func TestToLiveStreamCarriesRouteAndClient(t *testing.T) { + // The normalized view must surface route (native vs jellycompat) and client + // identity so the monitor is first-class, not just session id + method. + info := nodesessions.SessionInfo{ + SessionID: "s1", + AuthUserID: 9, + Type: "transcode", + Route: "jellycompat", + ClientIP: "203.0.113.7", + ClientName: "Infuse", + Position: 61.5, + HWAccel: "vaapi", + } + got := toLiveStream(info) + if got.Route != "jellycompat" || got.ClientIP != "203.0.113.7" || got.ClientName != "Infuse" { + t.Fatalf("route/client not carried: %+v", got) + } + if got.Position != 61.5 || got.HWAccel != "vaapi" { + t.Fatalf("position/hwaccel not carried: %+v", got) + } +} + +func TestMergeStreamsBackfillsAttribution(t *testing.T) { + // A fresher-but-thinner record (e.g. an ownerless transcode-node start record) + // must not drop route/client that a staler record for the same session carries. + in := []LiveStream{ + {SessionID: "s", NodeName: "proxy", UserID: 5, Route: "native", ClientName: "SiloTV", ClientIP: "10.0.0.9", LastServedAt: time.Unix(100, 0)}, + {SessionID: "s", NodeName: "transcode", UserID: 0, LastServedAt: time.Unix(200, 0)}, + } + out := mergeStreams(in) + if len(out) != 1 { + t.Fatalf("merge len = %d, want 1", len(out)) + } + got := out[0] + if got.NodeName != "transcode" { + t.Errorf("kept NodeName %q, want transcode (freshest)", got.NodeName) + } + if got.UserID != 5 || got.Route != "native" || got.ClientName != "SiloTV" || got.ClientIP != "10.0.0.9" { + t.Errorf("attribution not backfilled from staler record: %+v", got) + } +} + +func TestFuncSourceNilFn(t *testing.T) { + snap, err := NewFuncSource(nil).Snapshot(context.Background()) + if err != nil { + t.Fatalf("Snapshot: %v", err) + } + if len(snap.Streams) != 0 { + t.Errorf("nil fn Streams len = %d, want 0", len(snap.Streams)) + } +} + +func TestRedisSourceNilClient(t *testing.T) { + snap, err := NewRedisSource(nil).Snapshot(context.Background()) + if err != nil { + t.Fatalf("Snapshot: %v", err) + } + if len(snap.Streams) != 0 { + t.Errorf("nil client Streams len = %d, want 0", len(snap.Streams)) + } +} + +// TestDedupeSessionInfos mirrors the mergeStreams rules on the raw SessionInfo +// shape used by the admin session list: one row per session id, freshest copy +// wins, a resolved owner and missing attribution are carried across the merge +// so the operator view matches the enforcer's picture (no double-counted +// streams, no user-0 rows when any copy knows the owner). +func TestDedupeSessionInfos(t *testing.T) { + newer := time.Now().UTC().Format(time.RFC3339) + older := time.Now().Add(-time.Minute).UTC().Format(time.RFC3339) + infos := []nodesessions.SessionInfo{ + {SessionID: "s1", NodeName: "central", AuthUserID: 7, ProfileID: "p1", MediaFileID: 42, + Route: "native", ClientName: "SiloTV", Position: 130, LastServedAt: older}, + // The same stream, seen from the edge serving it: freshest but ownerless. + {SessionID: "s1", NodeName: "edge-1", LastServedAt: newer, BytesServed: 9000}, + {SessionID: "s2", NodeName: "edge-1", AuthUserID: 8, LastServedAt: newer}, + } + + out := DedupeSessionInfos(infos) + if len(out) != 2 { + t.Fatalf("dedupe len = %d, want 2", len(out)) + } + byID := make(map[string]nodesessions.SessionInfo, len(out)) + for _, info := range out { + byID[info.SessionID] = info + } + s1, ok := byID["s1"] + if !ok { + t.Fatalf("s1 missing from dedupe output") + } + if s1.NodeName != "edge-1" || s1.BytesServed != 9000 { + t.Errorf("s1 freshest copy not kept: %+v", s1) + } + if s1.AuthUserID != 7 || s1.ProfileID != "p1" || s1.MediaFileID != 42 { + t.Errorf("s1 owner not carried across merge: %+v", s1) + } + if s1.Route != "native" || s1.ClientName != "SiloTV" || s1.Position != 130 { + t.Errorf("s1 attribution not backfilled: %+v", s1) + } + if s2 := byID["s2"]; s2.AuthUserID != 8 { + t.Errorf("s2 mangled by dedupe: %+v", s2) + } +} diff --git a/internal/streamtoken/token.go b/internal/streamtoken/token.go index 4df90711..09debaa8 100644 --- a/internal/streamtoken/token.go +++ b/internal/streamtoken/token.go @@ -43,6 +43,13 @@ type Claims struct { ProfileID string `json:"pid,omitempty"` MediaFileID int `json:"mfid,omitempty"` + // Monitoring attribution (not byte-affecting, not a trust assertion): the + // origin protocol ("native" | "jellycompat") and the reported client/app + // name, carried so an edge node — which never sees the originating API path — + // can stamp them onto its live-session record for the first-class monitor view. + Origin string `json:"org,omitempty"` + ClientName string `json:"cn,omitempty"` + // Reconstruction recipe — the byte-affecting encode parameters, mirroring the // former playback.RecipeCard. Zero for direct/remux tokens, which reconstruct // from identity alone plus the client-supplied position. diff --git a/internal/transcodenode/server.go b/internal/transcodenode/server.go index 0ac57de6..d990713a 100644 --- a/internal/transcodenode/server.go +++ b/internal/transcodenode/server.go @@ -48,6 +48,15 @@ type TranscodeStartRequest struct { SubtitleCodec string `json:"subtitle_codec,omitempty"` TotalDuration float64 `json:"total_duration"` RequireReady bool `json:"require_ready,omitempty"` + + // Monitoring attribution copied from the originating session/token so the + // node's live-session record is owner-attributed and route-tagged even when + // no proxy record fronts it (closes the ownerless-record window). + AuthUserID int `json:"auth_user_id,omitempty"` + ProfileID string `json:"profile_id,omitempty"` + MediaFileID int `json:"media_file_id,omitempty"` + Route string `json:"route,omitempty"` + ClientName string `json:"client_name,omitempty"` } // TranscodeStartResponse is the JSON response for POST /transcode/start. @@ -567,15 +576,20 @@ func (s *Server) handleStart(w http.ResponseWriter, r *http.Request) { effectiveHWAccel := session.Opts().HWAccel trackCtx := context.WithoutCancel(r.Context()) go s.tracker.Track(trackCtx, nodesessions.SessionInfo{ - SessionID: req.SessionID, - NodeURL: s.tracker.NodeURL(), - NodeName: s.tracker.NodeName(), - Type: "transcode", - CodecVideo: req.TargetCodecVideo, - CodecAudio: req.TargetCodecAudio, - Resolution: req.TargetResolution, - HWAccel: effectiveHWAccel, - StartedAt: time.Now().UTC().Format(time.RFC3339), + SessionID: req.SessionID, + NodeURL: s.tracker.NodeURL(), + NodeName: s.tracker.NodeName(), + Type: "transcode", + CodecVideo: req.TargetCodecVideo, + CodecAudio: req.TargetCodecAudio, + Resolution: req.TargetResolution, + HWAccel: effectiveHWAccel, + StartedAt: time.Now().UTC().Format(time.RFC3339), + AuthUserID: req.AuthUserID, + ProfileID: req.ProfileID, + MediaFileID: req.MediaFileID, + Route: req.Route, + ClientName: req.ClientName, }) w.WriteHeader(http.StatusAccepted) @@ -651,7 +665,7 @@ func (s *Server) reconstructFromToken(r *http.Request, sessionID string, request } resolved = *fetched } - return s.spawnReconstruct(r, sessionID, requestedSegment, resolved), nil + return s.spawnReconstruct(r, sessionID, requestedSegment, resolved, claims.Origin, claims.ClientName), nil }) if session, _ := v.(*playback.TranscodeSession); session != nil { return session @@ -663,7 +677,7 @@ func (s *Server) reconstructFromToken(r *http.Request, sessionID string, request // registers it in the live map. It is only ever called inside the per-session // single-flight in reconstructFromToken, so it is the sole writer racing to // register sessionID. Returns nil if the spawn fails or the slot wait is canceled. -func (s *Server) spawnReconstruct(r *http.Request, sessionID string, requestedSegment int, card playback.RecipeCard) *playback.TranscodeSession { +func (s *Server) spawnReconstruct(r *http.Request, sessionID string, requestedSegment int, card playback.RecipeCard, route, clientName string) *playback.TranscodeSession { // Pace the cold-start burst so a node restart that loses many sessions does not // launch every ffmpeg at once. A client that disconnects while waiting releases // its slot rather than queueing dead work. @@ -749,6 +763,8 @@ func (s *Server) spawnReconstruct(r *http.Request, sessionID string, requestedSe AuthUserID: card.UserID, ProfileID: card.ProfileID, MediaFileID: card.MediaFileID, + Route: route, + ClientName: clientName, }) slog.InfoContext(r.Context(), "transcode node session reconstructed from token", "component", "transcodenode", @@ -854,6 +870,10 @@ func (s *Server) handleManifest(w http.ResponseWriter, r *http.Request) { w.Header().Set("Cache-Control", "no-store, max-age=0") w.Header().Set("Pragma", "no-cache") w.Write(manifest) + // Serve-activity mark so the node's own record reflects real serving instead + // of a LastServedAt frozen at start time (bytes are measured by the fronting + // proxy; wrapping ServeFile here would cost sendfile for a duplicate count). + s.tracker.MarkServed(sessionID) } func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) { @@ -983,6 +1003,8 @@ func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) { w.Header().Set("Cache-Control", "no-store, max-age=0") w.Header().Set("Pragma", "no-cache") http.ServeFile(w, r, segPath) + // Serve-activity mark: see handleManifest. + s.tracker.MarkServed(sessionID) } func (s *Server) handleForceReload(w http.ResponseWriter, r *http.Request) {