diff --git a/cmd/silo/main.go b/cmd/silo/main.go index 181c358a..89cc18c4 100644 --- a/cmd/silo/main.go +++ b/cmd/silo/main.go @@ -2183,6 +2183,7 @@ func main() { JWTSecret: cfg.Auth.JWTSecret, RecWorker: recWorker, FrontendFS: deps.FrontendFS, + SessionSyncer: deps.SessionSyncer, } // Wire direct dependencies when DB is available. diff --git a/internal/jellycompat/content_direct_test.go b/internal/jellycompat/content_direct_test.go index 9438b8eb..37982416 100644 --- a/internal/jellycompat/content_direct_test.go +++ b/internal/jellycompat/content_direct_test.go @@ -232,6 +232,9 @@ func (s *progressCountingStore) ListProgress(context.Context, string, string, in func (s *progressCountingStore) ListProgressFiltered(context.Context, string, string, []string, *int, int, int) ([]userstore.WatchProgress, error) { panic("unused") } +func (s *progressCountingStore) ListProgressSince(context.Context, string, string) ([]userstore.WatchProgress, string, error) { + panic("unused") +} func (s *progressCountingStore) AddHistory(context.Context, userstore.WatchHistoryEntry) error { panic("unused") } diff --git a/internal/jellycompat/handlers_playback.go b/internal/jellycompat/handlers_playback.go index 68112f54..71397bd7 100644 --- a/internal/jellycompat/handlers_playback.go +++ b/internal/jellycompat/handlers_playback.go @@ -110,6 +110,14 @@ type SettingsReader interface { Get(ctx context.Context, key string) (string, error) } +// PlaybackSessionSyncer flushes the in-memory native-session snapshot into the +// shared admin live-session table (playback_sessions_sync). Without it, compat +// session starts and stops only become visible on the periodic reconciler +// tick, leaving ghost rows in the activity dashboard for several seconds. +type PlaybackSessionSyncer interface { + SyncNow(ctx context.Context) error +} + // PlaybackHandler serves Jellyfin playback negotiation endpoints. type PlaybackHandler struct { cfg *config.Config @@ -129,10 +137,11 @@ type PlaybackHandler struct { TranscodeDir string transcodeMu sync.RWMutex transcodes map[string]*playback.TranscodeSession - SubtitleRepo subtitles.Repository // optional; enables downloaded subtitles - S3Client subtitles.S3Client // optional; for serving S3 subtitles - S3Bucket string // bucket for subtitle storage - SettingsRepo SettingsReader // optional; reads watched threshold setting + SubtitleRepo subtitles.Repository // optional; enables downloaded subtitles + S3Client subtitles.S3Client // optional; for serving S3 subtitles + S3Bucket string // bucket for subtitle storage + SettingsRepo SettingsReader // optional; reads watched threshold setting + SessionSyncer PlaybackSessionSyncer // optional; enables immediate session sync to shared admin view } // playbackThresholds reads the playback.watched_threshold and @@ -799,14 +808,14 @@ var losslessPassthroughCodecs = map[string]bool{ // compatFallbackCodecs are broadly supported audio codecs suitable for // software decoding. Lower index = higher preference. var compatFallbackCodecRank = map[string]int{ - "eac3": 1, - "ac3": 2, - "dts": 3, - "aac": 4, - "flac": 5, - "opus": 6, - "vorbis": 7, - "mp3": 8, + "eac3": 1, + "ac3": 2, + "dts": 3, + "aac": 4, + "flac": 5, + "opus": 6, + "vorbis": 7, + "mp3": 8, "pcm_s16le": 9, "pcm_s24le": 10, } diff --git a/internal/jellycompat/router.go b/internal/jellycompat/router.go index 7db59f94..f3875a49 100644 --- a/internal/jellycompat/router.go +++ b/internal/jellycompat/router.go @@ -109,6 +109,7 @@ func NewRouter(deps Dependencies) chi.Router { playbackHandler.JWTSecret = deps.JWTSecret playbackHandler.profileRefreshRequester = deps.RecWorker playbackHandler.SettingsRepo = deps.SettingsRepo + playbackHandler.SessionSyncer = deps.SessionSyncer if subtitleRepo != nil { playbackHandler.SubtitleRepo = subtitleRepo playbackHandler.S3Client = deps.S3Client diff --git a/internal/jellycompat/server.go b/internal/jellycompat/server.go index 1c88464d..67020bce 100644 --- a/internal/jellycompat/server.go +++ b/internal/jellycompat/server.go @@ -79,7 +79,12 @@ type Dependencies struct { PresignTTL time.Duration // Playback - SessionMgr SessionManagerInterface + SessionMgr SessionManagerInterface + // SessionSyncer flushes native session-manager state into the shared + // admin live-session table right after compat playback starts/stops, so + // the activity dashboard doesn't wait for the periodic reconciler tick. + // Optional. + SessionSyncer PlaybackSessionSyncer FileResolver FilePathResolver UserStoreProvider userstore.UserStoreProvider AccessFilterFn AccessFilterResolver diff --git a/internal/jellycompat/streams.go b/internal/jellycompat/streams.go index d4c83147..ed92c3e4 100644 --- a/internal/jellycompat/streams.go +++ b/internal/jellycompat/streams.go @@ -727,7 +727,7 @@ func (h *PlaybackHandler) HandleDeleteActiveEncodings(w http.ResponseWriter, r * return } - h.teardownPlaySession(playSession) + h.teardownPlaySession(r.Context(), playSession) w.WriteHeader(http.StatusNoContent) } @@ -736,7 +736,7 @@ func (h *PlaybackHandler) HandleDeleteActiveEncodings(w http.ResponseWriter, r * // play session from the store. Every step is idempotent, so it is safe to call // from both the explicit ActiveEncodings teardown and a Stopped playback report // (which may race). It is a no-op-safe teardown for an already-gone session. -func (h *PlaybackHandler) teardownPlaySession(playSession *PlaybackSession) { +func (h *PlaybackHandler) teardownPlaySession(ctx context.Context, playSession *PlaybackSession) { transcodeNodeURL := "" if h.sessionMgr != nil { if upstreamSession, err := h.sessionMgr.GetSession(playSession.UpstreamSessionID); err == nil { @@ -748,6 +748,28 @@ func (h *PlaybackHandler) teardownPlaySession(playSession *PlaybackSession) { _ = h.sessionMgr.StopSession(playSession.UpstreamSessionID) } h.playbackStore.Delete(playSession.ID) + // Clients often drop the connection right after reporting a stop, so detach + // the sync from request cancellation to keep the admin view accurate. + h.syncSessionsNow(context.WithoutCancel(ctx), "compat_stop") +} + +// compatSessionSyncTimeout bounds the immediate session sync issued from +// request paths, so a stalled database degrades to the periodic reconciler +// tick instead of pinning request goroutines. +const compatSessionSyncTimeout = 5 * time.Second + +// syncSessionsNow flushes the native-session snapshot to the shared admin +// live-session table so compat start/stop events are visible immediately +// instead of on the next reconciler tick. +func (h *PlaybackHandler) syncSessionsNow(ctx context.Context, reason string) { + if h == nil || h.SessionSyncer == nil { + return + } + ctx, cancel := context.WithTimeout(ctx, compatSessionSyncTimeout) + defer cancel() + if err := h.SessionSyncer.SyncNow(ctx); err != nil { + slog.Error("jellycompat: failed to sync sessions", "reason", reason, "error", err) + } } func (h *PlaybackHandler) handlePlaybackReport(w http.ResponseWriter, r *http.Request, stop bool) { @@ -833,7 +855,7 @@ func (h *PlaybackHandler) handlePlaybackReport(w http.ResponseWriter, r *http.Re } } if stop { - h.teardownPlaySession(playSession) + h.teardownPlaySession(r.Context(), playSession) } w.WriteHeader(http.StatusNoContent) @@ -904,6 +926,7 @@ func (h *PlaybackHandler) ensureUpstreamPlayback(ctx context.Context, compatSess if !ok { return nil, ErrSessionNotFound } + h.syncSessionsNow(ctx, "compat_start") return updated, nil } diff --git a/internal/jellycompat/streams_test.go b/internal/jellycompat/streams_test.go index 38c7e6e9..c7eb31b9 100644 --- a/internal/jellycompat/streams_test.go +++ b/internal/jellycompat/streams_test.go @@ -272,3 +272,132 @@ func TestHandleDeleteActiveEncodings_NotYetStartedNotTornDown(t *testing.T) { t.Fatalf("expected no StopSession calls; got %v", mgr.stopCalls) } } + +// recordingSessionSyncer counts SyncNow calls and records the context state at +// call time, standing in for the reconciler's immediate-sync trigger. +type recordingSessionSyncer struct { + calls int + lastCtxErr error + lastHadDeadline bool +} + +func (s *recordingSessionSyncer) SyncNow(ctx context.Context) error { + s.calls++ + s.lastCtxErr = ctx.Err() + _, s.lastHadDeadline = ctx.Deadline() + return nil +} + +// TestHandleSessionPlayingStopped_TearsDownAndSyncsImmediately verifies the +// Stopped report path removes the compat session AND flushes the live-session +// snapshot right away, so the activity dashboard doesn't show a ghost stream +// until the next reconciler tick (issue #205). +func TestHandleSessionPlayingStopped_TearsDownAndSyncsImmediately(t *testing.T) { + mgr := &testCompatSessionManager{sessions: map[string]*playback.Session{"upstream-1": {ID: "upstream-1"}}} + h, store := newActiveEncodingsHandler(mgr) + syncer := &recordingSessionSyncer{} + h.SessionSyncer = syncer + store.Put(PlaybackSession{ID: "ps-1", UpstreamSessionID: "upstream-1", CompatToken: "tok"}) + + body := strings.NewReader(`{"PlaySessionId":"ps-1"}`) + // Cancel the request context up front to simulate the client dropping the + // connection right after firing the stop report — the sync must still run. + ctx, cancel := context.WithCancel(context.Background()) + cancel() + req := withCompatSession(httptest.NewRequest("POST", "/Sessions/Playing/Stopped", body).WithContext(ctx), "tok") + rec := httptest.NewRecorder() + h.HandleSessionPlayingStopped(rec, req) + + if rec.Code != 204 { + t.Fatalf("status = %d, body = %s; want 204", rec.Code, rec.Body.String()) + } + if _, ok := store.Get("ps-1"); ok { + t.Fatal("play session should be deleted") + } + if len(mgr.stopCalls) != 1 || mgr.stopCalls[0] != "upstream-1" { + t.Fatalf("expected StopSession(upstream-1); got %v", mgr.stopCalls) + } + if syncer.calls != 1 { + t.Fatalf("SyncNow calls = %d; want 1", syncer.calls) + } + if syncer.lastCtxErr != nil { + t.Fatalf("sync context canceled with request: %v", syncer.lastCtxErr) + } + if !syncer.lastHadDeadline { + t.Fatal("sync context must carry a deadline so a stalled DB cannot pin the request goroutine") + } +} + +// TestHandleSessionPlayingStopped_UnknownSessionDoesNotSync verifies a stop +// report that tears nothing down doesn't trigger a sync round trip. +func TestHandleSessionPlayingStopped_UnknownSessionDoesNotSync(t *testing.T) { + mgr := &testCompatSessionManager{} + h, _ := newActiveEncodingsHandler(mgr) + syncer := &recordingSessionSyncer{} + h.SessionSyncer = syncer + + body := strings.NewReader(`{"PlaySessionId":"ps-missing"}`) + req := withCompatSession(httptest.NewRequest("POST", "/Sessions/Playing/Stopped", body), "tok") + rec := httptest.NewRecorder() + h.HandleSessionPlayingStopped(rec, req) + + if rec.Code != 204 { + t.Fatalf("status = %d, body = %s; want 204", rec.Code, rec.Body.String()) + } + if syncer.calls != 0 { + t.Fatalf("SyncNow calls = %d; want 0", syncer.calls) + } +} + +// TestHandleDeleteActiveEncodings_SyncsSessionsImmediately verifies the +// explicit encoder-teardown path also flushes the live-session snapshot. +func TestHandleDeleteActiveEncodings_SyncsSessionsImmediately(t *testing.T) { + mgr := &testCompatSessionManager{sessions: map[string]*playback.Session{"upstream-1": {ID: "upstream-1"}}} + h, store := newActiveEncodingsHandler(mgr) + syncer := &recordingSessionSyncer{} + h.SessionSyncer = syncer + store.Put(PlaybackSession{ID: "ps-1", UpstreamSessionID: "upstream-1", CompatToken: "tok"}) + + req := withCompatSession(httptest.NewRequest("DELETE", "/Videos/ActiveEncodings?PlaySessionId=ps-1", nil), "tok") + rec := httptest.NewRecorder() + h.HandleDeleteActiveEncodings(rec, req) + + if rec.Code != 204 { + t.Fatalf("status = %d, body = %s; want 204", rec.Code, rec.Body.String()) + } + if syncer.calls != 1 { + t.Fatalf("SyncNow calls = %d; want 1", syncer.calls) + } +} + +// TestEnsureUpstreamPlayback_SyncsOnNewSession verifies a fresh upstream +// session start flushes the live-session snapshot so the new stream appears in +// the activity dashboard immediately. +func TestEnsureUpstreamPlayback_SyncsOnNewSession(t *testing.T) { + mgr := &testCompatSessionManager{} + h, store := newActiveEncodingsHandler(mgr) + syncer := &recordingSessionSyncer{} + h.SessionSyncer = syncer + store.Put(PlaybackSession{ID: "ps-1", CompatToken: "tok"}) + + compatSession := &Session{Token: "tok", StreamAppUserID: 7, ProfileID: "prof-1"} + source := PlaybackMediaSource{ID: "src-1", FileID: 42} + playSession, err := h.ensureUpstreamPlayback(context.Background(), compatSession, "ps-1", source, "direct") + if err != nil { + t.Fatalf("ensureUpstreamPlayback: %v", err) + } + if playSession.UpstreamSessionID == "" { + t.Fatal("expected upstream session to be started") + } + if syncer.calls != 1 { + t.Fatalf("SyncNow calls = %d; want 1", syncer.calls) + } + + // Re-entering with the same method reuses the session and must not sync again. + if _, err := h.ensureUpstreamPlayback(context.Background(), compatSession, "ps-1", source, "direct"); err != nil { + t.Fatalf("ensureUpstreamPlayback reuse: %v", err) + } + if syncer.calls != 1 { + t.Fatalf("SyncNow calls after reuse = %d; want 1", syncer.calls) + } +} diff --git a/internal/worker/reconciler.go b/internal/worker/reconciler.go index 490dbb8c..07ae4978 100644 --- a/internal/worker/reconciler.go +++ b/internal/worker/reconciler.go @@ -6,6 +6,7 @@ import ( "log" "sort" "strings" + "sync" "time" "github.com/jackc/pgx/v5" @@ -74,6 +75,14 @@ type Reconciler struct { EventBus cache.EventBus EventsHub *evt.Hub PreSync PreSyncHook + // syncMu guards syncRunning/syncPending. Session syncs are coalesced onto a + // single owner so concurrent callers (the periodic tick plus request-path + // start/stop triggers) can never commit an older session snapshot after a + // newer one — which would resurrect stopped sessions or drop freshly + // started ones — and so request goroutines never queue behind a slow sync. + syncMu sync.Mutex + syncRunning bool + syncPending bool } // NewReconciler creates a new Reconciler with sensible defaults. The default @@ -434,14 +443,56 @@ func (r *Reconciler) tick() { } } -// SyncNow runs one immediate session reconciliation using the current local -// session snapshot. When nodeName is configured, an empty snapshot still -// clears any rows previously reported by that node. +// SyncNow reconciles the current local session snapshot into the shared +// table. When nodeName is configured, an empty snapshot still clears any rows +// previously reported by that node. +// +// Syncs are coalesced: only one reconciliation runs at a time, and a call that +// arrives while one is in flight returns immediately after asking the running +// owner for one follow-up pass with a fresh snapshot. The follow-up capture +// happens after the caller's state change, so its effect is never lost, and +// snapshots always commit in capture order. func (r *Reconciler) SyncNow(ctx context.Context) error { if r.sessionProvider == nil { return nil } + r.syncMu.Lock() + if r.syncRunning { + // The in-flight sync may have captured a snapshot that predates this + // caller's state change; have the owner run one more pass. + r.syncPending = true + r.syncMu.Unlock() + return nil + } + r.syncRunning = true + // The fresh capture below supersedes any pass queued before ownership. + r.syncPending = false + r.syncMu.Unlock() + + err := r.syncOnce(ctx) + for { + r.syncMu.Lock() + // Leave a queued pass for the next caller (the periodic tick at the + // latest) rather than burning it on an already-expired context. + if !r.syncPending || ctx.Err() != nil { + r.syncRunning = false + r.syncMu.Unlock() + return err + } + r.syncPending = false + r.syncMu.Unlock() + if passErr := r.syncOnce(ctx); err == nil { + err = passErr + } + } +} + +// syncOnce captures one session snapshot and reconciles it. +func (r *Reconciler) syncOnce(ctx context.Context) error { + if err := ctx.Err(); err != nil { + return err + } sessions := r.sessionProvider() if len(sessions) == 0 { if r.nodeName == "" { diff --git a/internal/worker/reconciler_test.go b/internal/worker/reconciler_test.go new file mode 100644 index 00000000..b467ca30 --- /dev/null +++ b/internal/worker/reconciler_test.go @@ -0,0 +1,91 @@ +package worker + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" +) + +// TestSyncNowSerializesSnapshotCapture guards the SyncNow ordering contract: +// snapshot capture and reconciliation run under one lock, so a request-path +// sync (playback start/stop) can never interleave with the periodic tick and +// commit an older session snapshot after a newer one. +func TestSyncNowSerializesSnapshotCapture(t *testing.T) { + var inflight atomic.Int32 + var overlapped atomic.Bool + provider := func() []SessionSync { + if inflight.Add(1) > 1 { + overlapped.Store(true) + } + time.Sleep(2 * time.Millisecond) + inflight.Add(-1) + return nil + } + + // No pool is needed: an empty snapshot with no node name returns before + // any database work, keeping the test focused on the locking contract. + r := NewReconciler(nil, "", provider) + + var wg sync.WaitGroup + for range 8 { + wg.Add(1) + go func() { + defer wg.Done() + if err := r.SyncNow(context.Background()); err != nil { + t.Errorf("SyncNow: %v", err) + } + }() + } + wg.Wait() + + if overlapped.Load() { + t.Fatal("concurrent SyncNow calls captured snapshots concurrently; capture and reconcile must be serialized") + } +} + +// TestSyncNowCoalescesPendingPass guards the follow-up contract: a SyncNow +// call that arrives while a sync is in flight returns immediately, and the +// running owner re-captures a fresh snapshot afterwards — so a stop that lands +// mid-sync is still reflected without waiting for the periodic tick. +func TestSyncNowCoalescesPendingPass(t *testing.T) { + captures := make(chan struct{}, 16) + release := make(chan struct{}) + first := true + provider := func() []SessionSync { + captures <- struct{}{} + if first { + first = false + <-release // hold the first sync mid-flight + } + return nil + } + r := NewReconciler(nil, "", provider) + + ownerDone := make(chan error, 1) + go func() { ownerDone <- r.SyncNow(context.Background()) }() + <-captures // owner is now blocked inside its snapshot capture + + // A second sync while the first is in flight must not block. + done := make(chan struct{}) + go func() { + _ = r.SyncNow(context.Background()) + close(done) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("SyncNow blocked behind an in-flight sync; it must coalesce and return") + } + + close(release) + if err := <-ownerDone; err != nil { + t.Fatalf("owner SyncNow: %v", err) + } + select { + case <-captures: // the owner's follow-up pass with a fresh snapshot + default: + t.Fatal("no follow-up snapshot capture ran; the coalesced sync was lost") + } +}