From a3397be489352e65cdb7404c214a147a08f0ca75 Mon Sep 17 00:00:00 2001 From: CoffeeKnyte <67730400+CoffeeKnyte@users.noreply.github.com> Date: Wed, 27 May 2026 15:46:24 +0000 Subject: [PATCH 1/2] fix(libraryingest): treat drainer shutdown cancel as clean stop TV/series full scans (libraries with new or updated items) were recorded as "cancelled" with an empty error message and never completed matching. When the file-walk finishes, the ingest executor waits out a settle window and then calls stopDrainers() to shut down the concurrent match goroutines. That cancels the drainer context while a ProcessBatchByFolderAndPathPrefix call may still be in flight. The drainer treated the resulting context.Canceled as a fatal error: it pushed the error to drainerErrCh and called cancel() on the whole scan context, so scanqueue.process() mapped it to cancelRun(). Large/slow libraries (many series, slow provider lookups) keep a batch in flight continuously, so stopDrainers() almost always landed mid-call and the scan was cancelled; small/fast libraries were usually idle at that instant and completed normally. Treat a cancelled drainer context as a deliberate shutdown: return cleanly without escalating. Genuine external cancellation still reaches the run via the main goroutine's scanCtx checks, so real cancels are not swallowed. Adds a regression test (settle window made injectable) that fails against the old handler with 'concurrent match scope ...: context canceled' and passes with the fix. --- internal/libraryingest/executor.go | 23 ++++++- internal/libraryingest/executor_test.go | 82 +++++++++++++++++++++++++ 2 files changed, 103 insertions(+), 2 deletions(-) create mode 100644 internal/libraryingest/executor_test.go diff --git a/internal/libraryingest/executor.go b/internal/libraryingest/executor.go index ab5b3415..4cc5a0cc 100644 --- a/internal/libraryingest/executor.go +++ b/internal/libraryingest/executor.go @@ -90,6 +90,11 @@ type Executor struct { realtime *notifications.Hub now func() time.Time + // tvDrainSettleWindow overrides scopedTVDrainSettleWindow when > 0. Kept + // injectable so tests can exercise the settle-window shutdown path without + // waiting the full production interval. + tvDrainSettleWindow time.Duration + mu sync.Mutex running []runningClaim } @@ -185,6 +190,16 @@ func (e *Executor) ingest(ctx context.Context, folder *models.MediaFolder, mode concurrentMatched.Add(int64(processed)) } if err != nil { + // A cancelled drainer context means this is a deliberate + // shutdown (the settle window ended and stopDrainers ran, or + // the whole scan is being torn down), not a real failure. The + // in-flight ProcessBatch call returns context.Canceled here, so + // exit cleanly without escalating. Genuine external + // cancellation still reaches the run via the scanCtx checks in + // the main goroutine, so this does not swallow real cancels. + if drainerCtx.Err() != nil { + return + } select { case drainerErrCh <- fmt.Errorf("concurrent match scope %q: %w", scopePath, err): default: @@ -238,12 +253,16 @@ func (e *Executor) ingest(ctx context.Context, folder *models.MediaFolder, mode } matchScopes = scanMatchScopes if shouldWaitForTVQueueSettle(folder, scanResult) { + settleWindow := e.tvDrainSettleWindow + if settleWindow <= 0 { + settleWindow = scopedTVDrainSettleWindow + } slog.Info("library ingest: waiting for tv queue settle window", "folder_id", folder.ID, "mode", mode, - "wait", scopedTVDrainSettleWindow, + "wait", settleWindow, ) - timer := time.NewTimer(scopedTVDrainSettleWindow) + timer := time.NewTimer(settleWindow) select { case <-scanCtx.Done(): timer.Stop() diff --git a/internal/libraryingest/executor_test.go b/internal/libraryingest/executor_test.go new file mode 100644 index 00000000..17bcaf02 --- /dev/null +++ b/internal/libraryingest/executor_test.go @@ -0,0 +1,82 @@ +package libraryingest + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "github.com/Silo-Server/silo-server/internal/models" + "github.com/Silo-Server/silo-server/internal/scanner" +) + +// settleBlockingMatcher simulates a TV match drainer whose provider lookup is +// still in flight when the settle window expires: the batch call blocks until +// its context is cancelled, then returns context.Canceled — exactly what +// stopDrainers triggers when it cancels the drainer context. +type settleBlockingMatcher struct { + batchCalls atomic.Int64 +} + +func (m *settleBlockingMatcher) ProcessBatchByFolderAndPathPrefix(ctx context.Context, _ int, _ string, _ time.Time) (int, error) { + m.batchCalls.Add(1) + <-ctx.Done() + return 0, ctx.Err() +} + +func (m *settleBlockingMatcher) ProcessAllByFolderAndPathPrefix(context.Context, int, string, time.Time) (int, error) { + return 0, nil +} + +func (m *settleBlockingMatcher) RetryUnmatchedItemsByFolderAndPathPrefix(context.Context, int, string) (int, int, error) { + return 0, 0, nil +} + +// settleStubScanner returns a scan result that triggers the TV settle window +// (a series library with new items) and accepts the post-match finalize call. +type settleStubScanner struct { + result *scanner.ScanResult +} + +func (s *settleStubScanner) ScanFolder(context.Context, *models.MediaFolder) (*scanner.ScanResult, error) { + return s.result, nil +} + +func (s *settleStubScanner) ScanSubtree(context.Context, *models.MediaFolder, string) (*scanner.ScanResult, error) { + return s.result, nil +} + +func (s *settleStubScanner) ScanFile(context.Context, string, *models.MediaFolder) error { + return nil +} + +func (s *settleStubScanner) FinalizeVariantsByPathPrefix(context.Context, *models.MediaFolder, string) error { + return nil +} + +// TestIngestFolderCompletesWhenDrainerCanceledMidBatch is a regression test for +// the settle-window cancellation bug: a TV library full scan was recorded as +// "cancelled" because a drainer batch in flight when stopDrainers() ran returned +// context.Canceled, which the drainer escalated as a fatal scan error. The +// ingest must instead complete normally. +func TestIngestFolderCompletesWhenDrainerCanceledMidBatch(t *testing.T) { + matcher := &settleBlockingMatcher{} + exec := &Executor{ + scanner: &settleStubScanner{result: &scanner.ScanResult{New: 1}}, + matcher: matcher, + now: time.Now, + tvDrainSettleWindow: 50 * time.Millisecond, + } + folder := &models.MediaFolder{ID: 5, Type: "series", Paths: []string{"/tv"}} + + result, err := exec.IngestFolder(context.Background(), folder) + if err != nil { + t.Fatalf("expected ingest to complete, got error: %v", err) + } + if result == nil || result.Skipped { + t.Fatalf("expected a non-skipped result, got %+v", result) + } + if matcher.batchCalls.Load() == 0 { + t.Fatal("drainer never ran a batch; test did not exercise the settle-window shutdown path") + } +} From 01b0926748b1fd877b2b306ff6720b7a0c715d25 Mon Sep 17 00:00:00 2001 From: Quick <31828688+Quick104@users.noreply.github.com> Date: Wed, 27 May 2026 13:37:33 -0400 Subject: [PATCH 2/2] Fix settle-window drainer cancellation in library ingest --- internal/libraryingest/executor.go | 25 +++--- internal/libraryingest/executor_test.go | 106 ++++++++++++++++++------ 2 files changed, 95 insertions(+), 36 deletions(-) diff --git a/internal/libraryingest/executor.go b/internal/libraryingest/executor.go index 4cc5a0cc..688e326c 100644 --- a/internal/libraryingest/executor.go +++ b/internal/libraryingest/executor.go @@ -165,7 +165,8 @@ func (e *Executor) ingest(ctx context.Context, folder *models.MediaFolder, mode CurrentScope: claim.path, }) runStartedAt := time.Now().UTC() - drainerCtx, stopDrainers := context.WithCancel(scanCtx) + drainerStopCtx, stopDrainers := context.WithCancel(context.Background()) + defer stopDrainers() drainerErrCh := make(chan error, 1) var drainerWG sync.WaitGroup if len(matchScopes) > 0 { @@ -180,24 +181,21 @@ func (e *Executor) ingest(ctx context.Context, folder *models.MediaFolder, mode ticker := time.NewTicker(scopedDrainInterval) defer ticker.Stop() for { - if drainerCtx.Err() != nil { + select { + case <-scanCtx.Done(): return + case <-drainerStopCtx.Done(): + return + default: } started := time.Now() - processed, err := e.matcher.ProcessBatchByFolderAndPathPrefix(drainerCtx, folder.ID, scopePath, runStartedAt) + processed, err := e.matcher.ProcessBatchByFolderAndPathPrefix(scanCtx, folder.ID, scopePath, runStartedAt) concurrentMatchDuration.Add(time.Since(started).Nanoseconds()) if processed > 0 { concurrentMatched.Add(int64(processed)) } if err != nil { - // A cancelled drainer context means this is a deliberate - // shutdown (the settle window ended and stopDrainers ran, or - // the whole scan is being torn down), not a real failure. The - // in-flight ProcessBatch call returns context.Canceled here, so - // exit cleanly without escalating. Genuine external - // cancellation still reaches the run via the scanCtx checks in - // the main goroutine, so this does not swallow real cancels. - if drainerCtx.Err() != nil { + if scanCtx.Err() != nil { return } select { @@ -208,7 +206,9 @@ func (e *Executor) ingest(ctx context.Context, folder *models.MediaFolder, mode return } select { - case <-drainerCtx.Done(): + case <-scanCtx.Done(): + return + case <-drainerStopCtx.Done(): return case <-ticker.C: } @@ -234,6 +234,7 @@ func (e *Executor) ingest(ctx context.Context, folder *models.MediaFolder, mode scanStarted := time.Now() scanMatchScopes, scanResult, err := e.scan(scanProgressCtx, folder, mode, claim.path) if err != nil { + cancel() stopDrainers() drainerWG.Wait() if len(matchScopes) > 0 { diff --git a/internal/libraryingest/executor_test.go b/internal/libraryingest/executor_test.go index 17bcaf02..073f2199 100644 --- a/internal/libraryingest/executor_test.go +++ b/internal/libraryingest/executor_test.go @@ -2,6 +2,7 @@ package libraryingest import ( "context" + "sync" "sync/atomic" "testing" "time" @@ -10,25 +11,50 @@ import ( "github.com/Silo-Server/silo-server/internal/scanner" ) -// settleBlockingMatcher simulates a TV match drainer whose provider lookup is -// still in flight when the settle window expires: the batch call blocks until -// its context is cancelled, then returns context.Canceled — exactly what -// stopDrainers triggers when it cancels the drainer context. -type settleBlockingMatcher struct { - batchCalls atomic.Int64 +// settleControlledMatcher simulates a TV match drainer whose provider lookup is +// still in flight when the settle window expires. The batch only returns once +// the test releases it, so the test can catch stopDrainers cancelling the active +// batch context instead of only stopping the drainer loop. +type settleControlledMatcher struct { + batchCalls atomic.Int64 + batchCompleted atomic.Bool + batchCtxCanceled atomic.Bool + processAllBeforeBatch atomic.Bool + batchStarted chan struct{} + releaseBatch chan struct{} + closeBatchStartedOnce sync.Once } -func (m *settleBlockingMatcher) ProcessBatchByFolderAndPathPrefix(ctx context.Context, _ int, _ string, _ time.Time) (int, error) { +func newSettleControlledMatcher() *settleControlledMatcher { + return &settleControlledMatcher{ + batchStarted: make(chan struct{}), + releaseBatch: make(chan struct{}), + } +} + +func (m *settleControlledMatcher) ProcessBatchByFolderAndPathPrefix(ctx context.Context, _ int, _ string, _ time.Time) (int, error) { m.batchCalls.Add(1) - <-ctx.Done() - return 0, ctx.Err() + m.closeBatchStartedOnce.Do(func() { + close(m.batchStarted) + }) + select { + case <-ctx.Done(): + m.batchCtxCanceled.Store(true) + return 0, ctx.Err() + case <-m.releaseBatch: + m.batchCompleted.Store(true) + return 1, nil + } } -func (m *settleBlockingMatcher) ProcessAllByFolderAndPathPrefix(context.Context, int, string, time.Time) (int, error) { +func (m *settleControlledMatcher) ProcessAllByFolderAndPathPrefix(context.Context, int, string, time.Time) (int, error) { + if !m.batchCompleted.Load() { + m.processAllBeforeBatch.Store(true) + } return 0, nil } -func (m *settleBlockingMatcher) RetryUnmatchedItemsByFolderAndPathPrefix(context.Context, int, string) (int, int, error) { +func (m *settleControlledMatcher) RetryUnmatchedItemsByFolderAndPathPrefix(context.Context, int, string) (int, int, error) { return 0, 0, nil } @@ -54,29 +80,61 @@ func (s *settleStubScanner) FinalizeVariantsByPathPrefix(context.Context, *model return nil } -// TestIngestFolderCompletesWhenDrainerCanceledMidBatch is a regression test for -// the settle-window cancellation bug: a TV library full scan was recorded as -// "cancelled" because a drainer batch in flight when stopDrainers() ran returned -// context.Canceled, which the drainer escalated as a fatal scan error. The -// ingest must instead complete normally. -func TestIngestFolderCompletesWhenDrainerCanceledMidBatch(t *testing.T) { - matcher := &settleBlockingMatcher{} +// TestIngestFolderLetsActiveDrainerBatchFinishAfterSettleWindow is a regression +// test for the settle-window cancellation bug: a TV library full scan was +// recorded as "cancelled" because stopDrainers cancelled a drainer batch that +// was already in flight. The active batch must be allowed to finish; otherwise +// rows it already claimed can be excluded from the final scoped matcher by the +// runStartedAt attempt window. +func TestIngestFolderLetsActiveDrainerBatchFinishAfterSettleWindow(t *testing.T) { + const settleWindow = 25 * time.Millisecond + + matcher := newSettleControlledMatcher() exec := &Executor{ scanner: &settleStubScanner{result: &scanner.ScanResult{New: 1}}, matcher: matcher, now: time.Now, - tvDrainSettleWindow: 50 * time.Millisecond, + tvDrainSettleWindow: settleWindow, } folder := &models.MediaFolder{ID: 5, Type: "series", Paths: []string{"/tv"}} - result, err := exec.IngestFolder(context.Background(), folder) - if err != nil { - t.Fatalf("expected ingest to complete, got error: %v", err) + type ingestResult struct { + result *Result + err error } - if result == nil || result.Skipped { - t.Fatalf("expected a non-skipped result, got %+v", result) + done := make(chan ingestResult, 1) + go func() { + result, err := exec.IngestFolder(context.Background(), folder) + done <- ingestResult{result: result, err: err} + }() + + select { + case <-matcher.batchStarted: + case <-time.After(time.Second): + t.Fatal("drainer never started a batch") + } + time.Sleep(2 * settleWindow) + close(matcher.releaseBatch) + + var got ingestResult + select { + case got = <-done: + case <-time.After(time.Second): + t.Fatal("ingest did not complete after releasing the active drainer batch") + } + if got.err != nil { + t.Fatalf("expected ingest to complete, got error: %v", got.err) + } + if got.result == nil || got.result.Skipped { + t.Fatalf("expected a non-skipped result, got %+v", got.result) } if matcher.batchCalls.Load() == 0 { t.Fatal("drainer never ran a batch; test did not exercise the settle-window shutdown path") } + if matcher.batchCtxCanceled.Load() { + t.Fatal("settle-window shutdown cancelled the active drainer batch context") + } + if matcher.processAllBeforeBatch.Load() { + t.Fatal("final scoped matcher ran before the active drainer batch completed") + } }