feat(metadata): configure image cache workers

.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

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)

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{}