haeaeae/silo-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ebooks): add backfill canary controls

Browse Source
This commit is contained in:
rxwatcher
2026-07-22 13:43:57 +02:00
parent 194e5ff527
commit 3a5d318714
2 changed files with 255 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/taskmanager/tasks/sync_ebook_metadata.go
+101 -2
View File
@@ -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
internal/taskmanager/tasks/sync_ebook_metadata_test.go
+154 -1
View File
@@ -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,