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 } type retryRecordingMatcher struct { processAllCalls atomic.Int64 retryCalls atomic.Int64 } func (m *retryRecordingMatcher) ProcessBatchByFolderAndPathPrefix(context.Context, int, string, time.Time) (int, error) { return 0, nil } func (m *retryRecordingMatcher) ProcessAllByFolderAndPathPrefix(context.Context, int, string, time.Time) (int, error) { m.processAllCalls.Add(1) return 0, nil } func (m *retryRecordingMatcher) RetryUnmatchedItemsByFolderAndPathPrefix(context.Context, int, string) (int, int, error) { m.retryCalls.Add(1) return 0, 0, nil } type finalizeRecordingScanner struct { finalizeCalls atomic.Int64 } func (s *finalizeRecordingScanner) ScanFolder(context.Context, *models.MediaFolder) (*scanner.ScanResult, error) { return &scanner.ScanResult{}, nil } func (s *finalizeRecordingScanner) ScanSubtree(context.Context, *models.MediaFolder, string) (*scanner.ScanResult, error) { return &scanner.ScanResult{}, nil } func (s *finalizeRecordingScanner) ScanFile(context.Context, string, *models.MediaFolder) error { return nil } func (s *finalizeRecordingScanner) FinalizeVariantsByPathPrefix(context.Context, *models.MediaFolder, string) error { s.finalizeCalls.Add(1) return nil } func TestIngestFolderSkipsGenericMatchingForDedicatedEnrichment(t *testing.T) { tests := []struct { name string folderType string expectedProcessAll int64 expectedRetry int64 }{ {name: "ebooks", folderType: "ebooks", expectedProcessAll: 0, expectedRetry: 0}, {name: "movies", folderType: "movies", expectedProcessAll: 1, expectedRetry: 1}, {name: "series", folderType: "series", expectedProcessAll: 1, expectedRetry: 1}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { matcher := &retryRecordingMatcher{} scanner := &finalizeRecordingScanner{} exec := &Executor{ scanner: scanner, matcher: matcher, now: time.Now, } folder := &models.MediaFolder{ID: 5, Type: tt.folderType, Paths: []string{"/library"}} if _, err := exec.IngestFolder(context.Background(), folder); err != nil { t.Fatalf("ingest folder: %v", err) } if got := matcher.processAllCalls.Load(); got != tt.expectedProcessAll { t.Fatalf("ProcessAllByFolderAndPathPrefix calls = %d, want %d", got, tt.expectedProcessAll) } if got := matcher.retryCalls.Load(); got != tt.expectedRetry { t.Fatalf("RetryUnmatchedItemsByFolderAndPathPrefix calls = %d, want %d", got, tt.expectedRetry) } if got := scanner.finalizeCalls.Load(); got != 1 { t.Fatalf("FinalizeVariantsByPathPrefix calls = %d, want 1", got) } }) } } // 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") } }