fix(ebooks): stop stalled enrichment drains

internal/ebooks/enrichment.go

@@ -383,7 +383,11 @@ func (e *Enricher) runQueueBatch(
 					}
 					recordTransitionError(item.ContentID, transitionErr)
 					if transitionErr == nil {
-						atomic.AddInt64(&failed, 1)
+						if errorClass == EnrichmentErrorRateLimited {
+							atomic.AddInt64(&deferred, 1)
+						} else {
+							atomic.AddInt64(&failed, 1)
+						}
 					}
 					continue
 				}

internal/ebooks/enrichment_test.go

@@ -13,6 +13,10 @@ import (
 
 	"github.com/Silo-Server/silo-server/internal/metadata"
 	"github.com/Silo-Server/silo-server/internal/models"
+	"google.golang.org/genproto/googleapis/rpc/errdetails"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/durationpb"
 )
 
 func TestEbookContentType(t *testing.T) {
@@ -260,6 +264,44 @@ func TestEnricherRunTransitionsClaimedJobsByOutcome(t *testing.T) {
 	}
 }
 
+func TestEnricherRunCountsRateLimitedFailureAsDeferred(t *testing.T) {
+	const retryAfter = 45 * time.Minute
+	limited, err := status.New(codes.ResourceExhausted, "provider quota exhausted").WithDetails(
+		&errdetails.RetryInfo{RetryDelay: durationpb.New(retryAfter)},
+	)
+	if err != nil {
+		t.Fatalf("attach retry info: %v", err)
+	}
+	queue := &fakeEnrichmentQueue{
+		jobs: []EnrichmentJob{{ContentID: "rate-limited", Token: "token", Attempts: 1}},
+	}
+	e := &Enricher{
+		queue:     queue,
+		batchSize: 1,
+		workers:   1,
+		loadClaimedItemsFn: func(context.Context, []EnrichmentJob) ([]enrichmentItemRow, error) {
+			return []enrichmentItemRow{{ContentID: "rate-limited"}}, nil
+		},
+		enrichClaimedItemFn: func(context.Context, enrichmentItemRow) (EnrichmentOutcome, error) {
+			return "", limited.Err()
+		},
+	}
+
+	result, runErr := e.Run(context.Background(), EnrichmentScopeIncremental)
+	if runErr != nil {
+		t.Fatalf("Run() error = %v", runErr)
+	}
+	if result.Failed != 0 || result.Deferred != 1 {
+		t.Fatalf("Run() result = %+v, want one deferred and no failures", result)
+	}
+	if got := queue.failed["rate-limited"]; got != EnrichmentErrorRateLimited {
+		t.Fatalf("failure class = %q, want rate_limited", got)
+	}
+	if got := queue.retryAfter["rate-limited"]; got != retryAfter {
+		t.Fatalf("retry after = %s, want %s", got, retryAfter)
+	}
+}
+
 func TestEnricherRunReleasesEveryLeaseOnCancellation(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	queue := &fakeEnrichmentQueue{
@@ -1089,6 +1131,7 @@ type fakeEnrichmentQueue struct {
 	remaining                 int
 	completed                 map[string]EnrichmentOutcome
 	failed                    map[string]EnrichmentErrorClass
+	retryAfter                map[string]time.Duration
 	released                  []string
 	discarded                 []string
 	transitionedJobs          map[string]EnrichmentJob
@@ -1130,13 +1173,17 @@ func (f *fakeEnrichmentQueue) Complete(_ context.Context, job EnrichmentJob, out
 	return f.completeErr
 }
 
-func (f *fakeEnrichmentQueue) Fail(_ context.Context, job EnrichmentJob, errorClass EnrichmentErrorClass, _ string, _ time.Duration) error {
+func (f *fakeEnrichmentQueue) Fail(_ context.Context, job EnrichmentJob, errorClass EnrichmentErrorClass, _ string, retryAfter time.Duration) error {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	if f.failed == nil {
 		f.failed = make(map[string]EnrichmentErrorClass)
 	}
+	if f.retryAfter == nil {
+		f.retryAfter = make(map[string]time.Duration)
+	}
 	f.failed[job.ContentID] = errorClass
+	f.retryAfter[job.ContentID] = retryAfter
 	f.recordTransition(job)
 	return f.failErr
 }

internal/taskmanager/tasks/sync_ebook_metadata.go

@@ -97,6 +97,10 @@ func (t *ebookMetadataTask) Execute(ctx context.Context, progress taskmanager.Pr
 		if err := ctx.Err(); err != nil {
 			return err
 		}
+		if ebookEnrichmentBatchMadeNoProgress(batch) {
+			reportEbookEnrichmentCircuitBreak(progress, total)
+			return nil
+		}
 		if batch.Remaining == 0 {
 			reportEbookEnrichmentProgress(progress, total, true)
 			return nil
@@ -112,6 +116,13 @@ func (t *ebookMetadataTask) Execute(ctx context.Context, progress taskmanager.Pr
 	}
 }
 
+func ebookEnrichmentBatchMadeNoProgress(batch ebooks.EnrichmentRunResult) bool {
+	return batch.Claimed > 0 &&
+		batch.Enriched == 0 &&
+		batch.NoMatch == 0 &&
+		batch.Failed+batch.Deferred >= batch.Claimed
+}
+
 func addEbookEnrichmentResult(total *ebooks.EnrichmentRunResult, batch ebooks.EnrichmentRunResult) {
 	total.Claimed += batch.Claimed
 	total.Enriched += batch.Enriched
@@ -144,6 +155,25 @@ func reportEbookEnrichmentProgress(
 	))
 }
 
+func reportEbookEnrichmentCircuitBreak(
+	progress taskmanager.ProgressReporter,
+	result ebooks.EnrichmentRunResult,
+) {
+	data, _ := json.Marshal(result)
+	progress.SetResultData(data)
+	percent := ebookEnrichmentPercent(result)
+	if percent >= 100 {
+		percent = 99
+	}
+	progress.Report(percent, fmt.Sprintf(
+		"Paused after a full batch made no progress; remaining work will retry later. Claimed %d, failed %d, deferred %d, remaining %d",
+		result.Claimed,
+		result.Failed,
+		result.Deferred,
+		result.Remaining,
+	))
+}
+
 func ebookEnrichmentPercent(result ebooks.EnrichmentRunResult) float64 {
 	if result.Remaining == 0 {
 		return 100

internal/taskmanager/tasks/sync_ebook_metadata_test.go

@@ -112,6 +112,81 @@ func TestEbookMetadataTaskDrainsBatchesAndReportsHonestProgress(t *testing.T) {
 	}
 }
 
+func TestEbookMetadataTaskStopsAfterOneAllFailedBatch(t *testing.T) {
+	enricher := &fakeEbookMetadataEnricher{results: []ebooks.EnrichmentRunResult{
+		{Claimed: 4, Failed: 4, Remaining: 100},
+		{Claimed: 4, Enriched: 4, Remaining: 96},
+	}}
+	progress := &ebookMetadataProgressReporter{}
+
+	if err := NewBackfillEbookMetadataTask(enricher).Execute(context.Background(), progress); err != nil {
+		t.Fatalf("Execute() error = %v", err)
+	}
+	assertNoProgressCircuitBreak(t, enricher, progress, ebooks.EnrichmentRunResult{
+		Claimed: 4, Failed: 4, Remaining: 100,
+	})
+}
+
+func TestEbookMetadataTaskStopsAfterOneAllDeferredBatch(t *testing.T) {
+	enricher := &fakeEbookMetadataEnricher{results: []ebooks.EnrichmentRunResult{
+		{Claimed: 4, Deferred: 4, Remaining: 0},
+		{Claimed: 4, Enriched: 4, Remaining: 0},
+	}}
+	progress := &ebookMetadataProgressReporter{}
+
+	if err := NewBackfillEbookMetadataTask(enricher).Execute(context.Background(), progress); err != nil {
+		t.Fatalf("Execute() error = %v", err)
+	}
+	assertNoProgressCircuitBreak(t, enricher, progress, ebooks.EnrichmentRunResult{
+		Claimed: 4, Deferred: 4, Remaining: 0,
+	})
+}
+
+func TestEbookMetadataTaskContinuesAfterMixedBatchWithProgress(t *testing.T) {
+	enricher := &fakeEbookMetadataEnricher{results: []ebooks.EnrichmentRunResult{
+		{Claimed: 4, Enriched: 1, Failed: 2, Deferred: 1, Remaining: 2},
+		{Claimed: 2, NoMatch: 2, Remaining: 0},
+	}}
+	progress := &ebookMetadataProgressReporter{}
+
+	if err := NewBackfillEbookMetadataTask(enricher).Execute(context.Background(), progress); err != nil {
+		t.Fatalf("Execute() error = %v", err)
+	}
+	if len(enricher.scopes) != 2 {
+		t.Fatalf("Run calls = %d, want 2 when a mixed batch made progress", len(enricher.scopes))
+	}
+}
+
+func assertNoProgressCircuitBreak(
+	t *testing.T,
+	enricher *fakeEbookMetadataEnricher,
+	progress *ebookMetadataProgressReporter,
+	want ebooks.EnrichmentRunResult,
+) {
+	t.Helper()
+	if len(enricher.scopes) != 1 {
+		t.Fatalf("Run calls = %d, want exactly 1", len(enricher.scopes))
+	}
+	if len(progress.results) == 0 {
+		t.Fatal("no result JSON reported")
+	}
+	var result ebooks.EnrichmentRunResult
+	if err := json.Unmarshal(progress.results[len(progress.results)-1], &result); err != nil {
+		t.Fatalf("result JSON error: %v", err)
+	}
+	if result != want {
+		t.Fatalf("result JSON = %+v, want %+v", result, want)
+	}
+	if got := progress.percents[len(progress.percents)-1]; got >= 100 {
+		t.Fatalf("circuit-break progress = %.1f, must not report completion", got)
+	}
+	message := progress.messages[len(progress.messages)-1]
+	if !strings.Contains(strings.ToLower(message), "no progress") ||
+		!strings.Contains(strings.ToLower(message), "retry later") {
+		t.Fatalf("circuit-break progress message = %q", message)
+	}
+}
+
 func TestEbookMetadataBackfillUsesLegacyScope(t *testing.T) {
 	enricher := &fakeEbookMetadataEnricher{results: []ebooks.EnrichmentRunResult{{Remaining: 0}}}
 	task := NewBackfillEbookMetadataTask(enricher)