diff --git a/internal/ebooks/enrichment.go b/internal/ebooks/enrichment.go index 900aefa4..97e75b0e 100644 --- a/internal/ebooks/enrichment.go +++ b/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 } diff --git a/internal/ebooks/enrichment_test.go b/internal/ebooks/enrichment_test.go index 43b619e2..4f5102c8 100644 --- a/internal/ebooks/enrichment_test.go +++ b/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 } diff --git a/internal/taskmanager/tasks/sync_ebook_metadata.go b/internal/taskmanager/tasks/sync_ebook_metadata.go index 082c2b47..d2daa65f 100644 --- a/internal/taskmanager/tasks/sync_ebook_metadata.go +++ b/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 diff --git a/internal/taskmanager/tasks/sync_ebook_metadata_test.go b/internal/taskmanager/tasks/sync_ebook_metadata_test.go index 75ea8806..003951bd 100644 --- a/internal/taskmanager/tasks/sync_ebook_metadata_test.go +++ b/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)