Merge pull request #15 from Silo-Server/fix/tv-scan-settle-window-cancel

fix(libraryingest): TV full scans cancelled at settle-window boundary
2026-05-27 13:44:57 -04:00
parent d9e6059079 01b0926748
commit bfe73a4dc9
2 changed files with 166 additions and 6 deletions
@@ -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()
@@ -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")
+	}
+}