diff --git a/.env.example b/.env.example index 964a449ae7..daaebdbcb4 100644 --- a/.env.example +++ b/.env.example @@ -98,6 +98,9 @@ POSTGRES_SHM_SIZE=8gb # NODE_NAME=transcode-01 # NODE_URL=http://transcode-01.example.com:8082 +# Optional metadata artwork cache concurrency. Defaults to 4 workers. +# SILO_METADATA_IMAGE_CACHE_WORKERS=4 + # Optional server mode override for source runs or standalone worker nodes. # MODE=integrated #SILO_MIGRATE_TIMEOUT=60m diff --git a/internal/taskmanager/tasks/cache_metadata_images.go b/internal/taskmanager/tasks/cache_metadata_images.go index f7a81e03c9..8cc5facf4c 100644 --- a/internal/taskmanager/tasks/cache_metadata_images.go +++ b/internal/taskmanager/tasks/cache_metadata_images.go @@ -4,6 +4,8 @@ import ( "context" "fmt" "os" + "strconv" + "strings" "time" "github.com/google/uuid" @@ -13,15 +15,25 @@ import ( ) const ( - cacheMetadataImagesIntervalMs = int64(60 * 1000) - // Claim only work that can start immediately. Claiming a large queue page - // stamps one lease on every row up front; with two workers, the unstarted + cacheMetadataImagesWorkersEnv = "SILO_METADATA_IMAGE_CACHE_WORKERS" + + cacheMetadataImagesIntervalMs = int64(60 * 1000) + cacheMetadataImagesDefaultWorkers = 4 + // Keep claim batches small independently of configured concurrency. Claiming + // a large queue page stamps one lease on every row up front, so an unstarted // tail could expire and be reclaimed before this execution reaches it. cacheMetadataImagesClaimLimit = 2 - cacheMetadataImagesWorkers = 2 cacheMetadataImagesMaxRuntime = 10 * time.Minute ) +func cacheMetadataImagesWorkers() int { + workers, err := strconv.Atoi(strings.TrimSpace(os.Getenv(cacheMetadataImagesWorkersEnv))) + if err != nil || workers <= 0 { + return cacheMetadataImagesDefaultWorkers + } + return workers +} + type MetadataImageCacheRunner interface { DrainUntilIdle(ctx context.Context, workerID string, claimLimit int, concurrency int, maxRuntime time.Duration, reportProgress metadata.ImageCacheRunProgressReporter) (metadata.ImageCacheRunStats, error) } @@ -135,7 +147,7 @@ func executeMetadataImages(ctx context.Context, progress taskmanager.ProgressRep ctx, workerID, cacheMetadataImagesClaimLimit, - cacheMetadataImagesWorkers, + cacheMetadataImagesWorkers(), maxRuntime, func(update metadata.ImageCacheRunStats) { percent := cacheMetadataImagesPercent(update) diff --git a/internal/taskmanager/tasks/cache_metadata_images_test.go b/internal/taskmanager/tasks/cache_metadata_images_test.go index 0814eb10a3..b662c09a1a 100644 --- a/internal/taskmanager/tasks/cache_metadata_images_test.go +++ b/internal/taskmanager/tasks/cache_metadata_images_test.go @@ -93,6 +93,7 @@ func TestBackfillMetadataImagesTaskProperties(t *testing.T) { } func TestCacheMetadataImagesTaskReportsStats(t *testing.T) { + t.Setenv(cacheMetadataImagesWorkersEnv, "6") runner := &fakeMetadataImageCacheRunner{ updates: []metadata.ImageCacheRunStats{{ Batches: 2, @@ -116,10 +117,10 @@ func TestCacheMetadataImagesTaskReportsStats(t *testing.T) { t.Fatalf("Execute() error = %v", err) } if runner.claimLimit != 2 { - t.Fatalf("claimLimit = %d, want one immediately-startable job per worker", runner.claimLimit) + t.Fatalf("claimLimit = %d, want fixed claim size 2", runner.claimLimit) } - if runner.concurrency != 2 { - t.Fatalf("concurrency = %d, want 2", runner.concurrency) + if runner.concurrency != 6 { + t.Fatalf("concurrency = %d, want configured 6", runner.concurrency) } if runner.maxRuntime != 10*time.Minute { t.Fatalf("maxRuntime = %s, want 10m", runner.maxRuntime) @@ -145,6 +146,7 @@ func TestCacheMetadataImagesTaskReportsStats(t *testing.T) { } func TestBackfillMetadataImagesTaskReportsDiscovery(t *testing.T) { + t.Setenv(cacheMetadataImagesWorkersEnv, "1") runner := &fakeMetadataImageCacheRunner{stats: metadata.ImageCacheRunStats{ Batches: 2, EnqueuedExisting: 5, @@ -162,7 +164,10 @@ func TestBackfillMetadataImagesTaskReportsDiscovery(t *testing.T) { t.Fatalf("maxRuntime = %s, want no deadline for manual backfill", runner.maxRuntime) } if runner.claimLimit != 2 { - t.Fatalf("claimLimit = %d, want one immediately-startable job per worker", runner.claimLimit) + t.Fatalf("claimLimit = %d, want fixed claim size 2", runner.claimLimit) + } + if runner.concurrency != 1 { + t.Fatalf("concurrency = %d, want configured 1", runner.concurrency) } if len(runner.workerIDs) != 1 || !strings.Contains(runner.workerIDs[0], ":backfill:") { t.Fatalf("backfill worker IDs = %#v, want one execution-scoped backfill owner", runner.workerIDs) @@ -176,6 +181,22 @@ func TestBackfillMetadataImagesTaskReportsDiscovery(t *testing.T) { } } +func TestCacheMetadataImagesWorkersDefaultsForInvalidOverride(t *testing.T) { + t.Setenv(cacheMetadataImagesWorkersEnv, "") + if got := cacheMetadataImagesWorkers(); got != 4 { + t.Fatalf("cacheMetadataImagesWorkers() = %d, want default 4", got) + } + + for _, value := range []string{"", "0", "-1", "not-a-number"} { + t.Run(value, func(t *testing.T) { + t.Setenv(cacheMetadataImagesWorkersEnv, value) + if got := cacheMetadataImagesWorkers(); got != cacheMetadataImagesDefaultWorkers { + t.Fatalf("cacheMetadataImagesWorkers() = %d, want default %d", got, cacheMetadataImagesDefaultWorkers) + } + }) + } +} + func TestMetadataImageTasksUseDistinctExecutionLeaseOwners(t *testing.T) { runner := &fakeMetadataImageCacheRunner{} progress := &recordingProgress{}