diff --git a/internal/taskmanager/tasks/sync_ebook_metadata.go b/internal/taskmanager/tasks/sync_ebook_metadata.go index d2daa65f..97d8ee53 100644 --- a/internal/taskmanager/tasks/sync_ebook_metadata.go +++ b/internal/taskmanager/tasks/sync_ebook_metadata.go @@ -4,13 +4,20 @@ import ( "context" "encoding/json" "fmt" + "os" + "strconv" + "strings" "time" "github.com/Silo-Server/silo-server/internal/ebooks" "github.com/Silo-Server/silo-server/internal/taskmanager" ) -const ebookMetadataExecutionBudget = 4 * time.Minute +const ( + ebookMetadataExecutionBudget = 4 * time.Minute + ebookBackfillMaxClaimsEnv = "SILO_EBOOK_BACKFILL_MAX_CLAIMS" + ebookBackfillBatchDelayEnv = "SILO_EBOOK_BACKFILL_BATCH_DELAY" +) type ebookMetadataEnricher interface { Run(ctx context.Context, scope ebooks.EnrichmentScope) (ebooks.EnrichmentRunResult, error) @@ -25,6 +32,9 @@ type ebookMetadataTask struct { triggers []taskmanager.TriggerConfig budget time.Duration now func() time.Time + maxClaims int + batchDelay time.Duration + sleep func(context.Context, time.Duration) error errorPrefix string } @@ -61,6 +71,9 @@ func NewBackfillEbookMetadataTask(enricher ebookMetadataEnricher) *BackfillEbook description: "Manually enriches the legacy ebook backlog without competing with scheduled metadata sync", budget: ebookMetadataExecutionBudget, now: time.Now, + maxClaims: parseEbookBackfillMaxClaims(os.Getenv(ebookBackfillMaxClaimsEnv)), + batchDelay: parseEbookBackfillBatchDelay(os.Getenv(ebookBackfillBatchDelayEnv)), + sleep: sleepEbookBackfill, errorPrefix: "ebook metadata backfill", }} } @@ -108,11 +121,66 @@ func (t *ebookMetadataTask) Execute(ctx context.Context, progress taskmanager.Pr if batch.Claimed == 0 { return nil } + if t.maxClaims > 0 && total.Claimed >= t.maxClaims { + deferEbookEnrichmentRemaining(&total) + reportEbookEnrichmentPause(progress, total, fmt.Sprintf( + "Paused after reaching manual backfill claim cap of %d; remaining work will retry later.", + t.maxClaims, + )) + return nil + } if t.now().Sub(started) >= t.budget { - total.Deferred += total.Remaining + deferEbookEnrichmentRemaining(&total) reportEbookEnrichmentProgress(progress, total, false) return nil } + if t.batchDelay > 0 && ebookEnrichmentBatchMadeProgress(batch) { + if t.batchDelay >= t.budget-t.now().Sub(started) { + deferEbookEnrichmentRemaining(&total) + reportEbookEnrichmentPause(progress, total, + "Paused before the next batch because its delay would exceed the ebook metadata execution budget; remaining work will retry later.") + return nil + } + if err := t.sleep(ctx, t.batchDelay); err != nil { + return err + } + if err := ctx.Err(); err != nil { + return err + } + if t.now().Sub(started) >= t.budget { + deferEbookEnrichmentRemaining(&total) + reportEbookEnrichmentPause(progress, total, + "Paused after reaching the ebook metadata execution budget; remaining work will retry later.") + return nil + } + } + } +} + +func parseEbookBackfillMaxClaims(raw string) int { + value, err := strconv.Atoi(strings.TrimSpace(raw)) + if err != nil || value <= 0 { + return 0 + } + return value +} + +func parseEbookBackfillBatchDelay(raw string) time.Duration { + value, err := time.ParseDuration(strings.TrimSpace(raw)) + if err != nil || value <= 0 { + return 0 + } + return value +} + +func sleepEbookBackfill(ctx context.Context, delay time.Duration) error { + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil } } @@ -123,6 +191,14 @@ func ebookEnrichmentBatchMadeNoProgress(batch ebooks.EnrichmentRunResult) bool { batch.Failed+batch.Deferred >= batch.Claimed } +func ebookEnrichmentBatchMadeProgress(batch ebooks.EnrichmentRunResult) bool { + return batch.Enriched > 0 || batch.NoMatch > 0 +} + +func deferEbookEnrichmentRemaining(result *ebooks.EnrichmentRunResult) { + result.Deferred += result.Remaining +} + func addEbookEnrichmentResult(total *ebooks.EnrichmentRunResult, batch ebooks.EnrichmentRunResult) { total.Claimed += batch.Claimed total.Enriched += batch.Enriched @@ -174,6 +250,29 @@ func reportEbookEnrichmentCircuitBreak( )) } +func reportEbookEnrichmentPause( + progress taskmanager.ProgressReporter, + result ebooks.EnrichmentRunResult, + message string, +) { + data, _ := json.Marshal(result) + progress.SetResultData(data) + percent := ebookEnrichmentPercent(result) + if percent >= 100 { + percent = 99 + } + progress.Report(percent, fmt.Sprintf( + "%s Claimed %d, enriched %d, no match %d, failed %d, deferred %d, remaining %d", + message, + result.Claimed, + result.Enriched, + result.NoMatch, + 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 003951bd..2839c01b 100644 --- a/internal/taskmanager/tasks/sync_ebook_metadata_test.go +++ b/internal/taskmanager/tasks/sync_ebook_metadata_test.go @@ -113,13 +113,20 @@ func TestEbookMetadataTaskDrainsBatchesAndReportsHonestProgress(t *testing.T) { } func TestEbookMetadataTaskStopsAfterOneAllFailedBatch(t *testing.T) { + t.Setenv("SILO_EBOOK_BACKFILL_MAX_CLAIMS", "1") + t.Setenv("SILO_EBOOK_BACKFILL_BATCH_DELAY", "1h") enricher := &fakeEbookMetadataEnricher{results: []ebooks.EnrichmentRunResult{ {Claimed: 4, Failed: 4, Remaining: 100}, {Claimed: 4, Enriched: 4, Remaining: 96}, }} progress := &ebookMetadataProgressReporter{} + task := NewBackfillEbookMetadataTask(enricher) + task.sleep = func(context.Context, time.Duration) error { + t.Fatal("no-progress batch must circuit-break before pacing") + return nil + } - if err := NewBackfillEbookMetadataTask(enricher).Execute(context.Background(), progress); err != nil { + if err := task.Execute(context.Background(), progress); err != nil { t.Fatalf("Execute() error = %v", err) } assertNoProgressCircuitBreak(t, enricher, progress, ebooks.EnrichmentRunResult{ @@ -157,6 +164,152 @@ func TestEbookMetadataTaskContinuesAfterMixedBatchWithProgress(t *testing.T) { } } +func TestEbookMetadataBackfillStopsAtClaimCap(t *testing.T) { + t.Setenv("SILO_EBOOK_BACKFILL_MAX_CLAIMS", "4") + t.Setenv("SILO_EBOOK_BACKFILL_BATCH_DELAY", "0") + enricher := &fakeEbookMetadataEnricher{results: []ebooks.EnrichmentRunResult{ + {Claimed: 2, Enriched: 2, Remaining: 10}, + {Claimed: 2, Enriched: 2, Remaining: 8}, + {Claimed: 2, Enriched: 2, Remaining: 6}, + }} + 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 at claim cap", len(enricher.scopes)) + } + var result ebooks.EnrichmentRunResult + if err := json.Unmarshal(progress.results[len(progress.results)-1], &result); err != nil { + t.Fatalf("result JSON error: %v", err) + } + want := ebooks.EnrichmentRunResult{Claimed: 4, Enriched: 4, Deferred: 8, Remaining: 8} + if result != want { + t.Fatalf("result JSON = %+v, want %+v", result, want) + } + if got := progress.percents[len(progress.percents)-1]; got >= 100 { + t.Fatalf("claim-capped progress = %.1f, must not report completion", got) + } + message := strings.ToLower(progress.messages[len(progress.messages)-1]) + if !strings.Contains(message, "claim cap") || !strings.Contains(message, "retry later") { + t.Fatalf("claim-cap progress message = %q", message) + } +} + +func TestEbookMetadataBackfillDelaysOnlyBetweenProductiveBatches(t *testing.T) { + enricher := &fakeEbookMetadataEnricher{results: []ebooks.EnrichmentRunResult{ + {Claimed: 1, Enriched: 1, Remaining: 2}, + {Claimed: 1, NoMatch: 1, Remaining: 1}, + {Claimed: 1, Enriched: 1, Remaining: 0}, + }} + task := NewBackfillEbookMetadataTask(enricher) + task.batchDelay = time.Second + var sleeps []time.Duration + task.sleep = func(_ context.Context, delay time.Duration) error { + sleeps = append(sleeps, delay) + return nil + } + + if err := task.Execute(context.Background(), &ebookMetadataProgressReporter{}); err != nil { + t.Fatalf("Execute() error = %v", err) + } + if len(sleeps) != 2 || sleeps[0] != time.Second || sleeps[1] != time.Second { + t.Fatalf("sleeps = %v, want two 1s inter-batch delays", sleeps) + } +} + +func TestEbookMetadataBackfillCancelsDuringBatchDelay(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + enricher := &fakeEbookMetadataEnricher{results: []ebooks.EnrichmentRunResult{ + {Claimed: 1, Enriched: 1, Remaining: 2}, + {Claimed: 1, Enriched: 1, Remaining: 1}, + }} + task := NewBackfillEbookMetadataTask(enricher) + task.batchDelay = time.Second + task.sleep = func(ctx context.Context, _ time.Duration) error { + cancel() + <-ctx.Done() + return ctx.Err() + } + + err := task.Execute(ctx, &ebookMetadataProgressReporter{}) + if !errors.Is(err, context.Canceled) { + t.Fatalf("Execute() error = %v, want context.Canceled", err) + } + if len(enricher.scopes) != 1 { + t.Fatalf("Run calls = %d, want 1 before cancellation", len(enricher.scopes)) + } +} + +func TestEbookMetadataBackfillDoesNotDelayPastExecutionBudget(t *testing.T) { + now := time.Date(2026, 7, 19, 12, 0, 0, 0, time.UTC) + enricher := &fakeEbookMetadataEnricher{results: []ebooks.EnrichmentRunResult{ + {Claimed: 1, Enriched: 1, Remaining: 2}, + {Claimed: 1, Enriched: 1, Remaining: 1}, + }} + task := NewBackfillEbookMetadataTask(enricher) + task.now = func() time.Time { return now } + task.budget = 1500 * time.Millisecond + task.batchDelay = time.Second + var sleeps int + task.sleep = func(_ context.Context, delay time.Duration) error { + sleeps++ + now = now.Add(delay) + return nil + } + progress := &ebookMetadataProgressReporter{} + + if err := task.Execute(context.Background(), progress); err != nil { + t.Fatalf("Execute() error = %v", err) + } + if sleeps != 1 { + t.Fatalf("sleep calls = %d, want 1 before remaining budget became too short", sleeps) + } + if len(enricher.scopes) != 2 { + t.Fatalf("Run calls = %d, want 2", len(enricher.scopes)) + } + message := strings.ToLower(progress.messages[len(progress.messages)-1]) + if !strings.Contains(message, "execution budget") { + t.Fatalf("budget progress message = %q", message) + } +} + +func TestEbookMetadataBackfillInvalidEnvironmentFallsBackToDisabledControls(t *testing.T) { + for _, tc := range []struct { + name string + max string + delay string + wantMax int + wantWait time.Duration + }{ + {name: "empty", max: "", delay: ""}, + {name: "malformed", max: "many", delay: "later"}, + {name: "negative", max: "-5", delay: "-1s"}, + {name: "zero", max: "0", delay: "0"}, + {name: "trimmed valid", max: " 20 ", delay: " 1s ", wantMax: 20, wantWait: time.Second}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Setenv("SILO_EBOOK_BACKFILL_MAX_CLAIMS", tc.max) + t.Setenv("SILO_EBOOK_BACKFILL_BATCH_DELAY", tc.delay) + task := NewBackfillEbookMetadataTask(&fakeEbookMetadataEnricher{}) + if task.maxClaims != tc.wantMax || task.batchDelay != tc.wantWait { + t.Fatalf("controls = (%d, %s), want (%d, %s)", + task.maxClaims, task.batchDelay, tc.wantMax, tc.wantWait) + } + }) + } +} + +func TestEbookMetadataSyncIgnoresBackfillCanaryEnvironment(t *testing.T) { + t.Setenv("SILO_EBOOK_BACKFILL_MAX_CLAIMS", "1") + t.Setenv("SILO_EBOOK_BACKFILL_BATCH_DELAY", "1h") + task := NewSyncEbookMetadataTask(&fakeEbookMetadataEnricher{}) + if task.maxClaims != 0 || task.batchDelay != 0 { + t.Fatalf("sync controls = (%d, %s), want disabled", task.maxClaims, task.batchDelay) + } +} + func assertNoProgressCircuitBreak( t *testing.T, enricher *fakeEbookMetadataEnricher,