fix(jellycompat): sync activity dashboard immediately on compat playback start/stop (#261)

* fix(jellycompat): sync activity dashboard immediately on compat playback start/stop

Compat (Infuse/Jellyfin) sessions were only reconciled into
playback_sessions_sync on the periodic 15s reconciler tick, so stopped
streams lingered in the admin activity dashboard and overlapped with
newly started ones as ghost sessions. Native playback handlers already
trigger an immediate SessionSyncer.SyncNow on start/stop; wire the same
syncer into the jellycompat playback handler and flush after
teardownPlaySession (Stopped report + ActiveEncodings teardown) and
after a new upstream session starts.

The stop-path sync detaches from request cancellation
(context.WithoutCancel) because clients often drop the connection right
after reporting a stop.

Also adds the ListProgressSince method to the jellycompat test fake so
the package's tests compile again after the downloads-v2 interface
change.

Part of #205

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(worker): serialize SyncNow and bound compat request-path session sync

Adversarial review follow-ups: SyncNow snapshots could commit out of
order when request-path syncs raced the periodic tick (an older
snapshot committing last would resurrect stopped sessions or drop
fresh ones), and the detached stop-path sync had no deadline, letting
a stalled DB pin request goroutines. Serialize snapshot capture +
reconcile under one lock and cap request-path syncs at 5s so failures
degrade to the periodic tick.

Part of #205

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(worker): coalesce immediate session syncs instead of queueing on a lock

Review iteration 2: a stop-triggered sync queued on a bare mutex could
expire its 5s deadline waiting behind a slow tick and fail before
removing the stopped row. Coalesce instead: one owner reconciles at a
time, callers that arrive mid-flight return immediately after flagging
a follow-up pass, and the owner re-captures a fresh (post-change)
snapshot afterwards. Request goroutines never block behind another
sync, snapshots still commit in capture order, and an expired-context
owner leaves the queued pass for the next tick instead of burning it.

Part of #205

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

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.

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

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,
 }

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

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

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
 }
 

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

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 == "" {

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")
+	}
+}