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