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.
This commit is contained in:
@@ -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()
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user