diff --git a/internal/libraryingest/executor.go b/internal/libraryingest/executor.go index ab5b3415..688e326c 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 } @@ -160,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 { @@ -175,16 +181,23 @@ 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 { + if scanCtx.Err() != nil { + return + } select { case drainerErrCh <- fmt.Errorf("concurrent match scope %q: %w", scopePath, err): default: @@ -193,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: } @@ -219,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 { @@ -238,12 +254,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..073f2199 --- /dev/null +++ b/internal/libraryingest/executor_test.go @@ -0,0 +1,140 @@ +package libraryingest + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/Silo-Server/silo-server/internal/models" + "github.com/Silo-Server/silo-server/internal/scanner" +) + +// 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 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) + 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 *settleControlledMatcher) ProcessAllByFolderAndPathPrefix(context.Context, int, string, time.Time) (int, error) { + if !m.batchCompleted.Load() { + m.processAllBeforeBatch.Store(true) + } + return 0, nil +} + +func (m *settleControlledMatcher) 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 +} + +// 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: settleWindow, + } + folder := &models.MediaFolder{ID: 5, Type: "series", Paths: []string{"/tv"}} + + type ingestResult struct { + result *Result + err error + } + 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") + } +}