perf(metadata): scale image-cache concurrency with host CPUs

internal/metadata/image_cache_job_repo.go

@@ -33,7 +33,10 @@ const (
 	ImageCacheStatusSucceeded = "succeeded"
 	ImageCacheStatusFailed    = "failed"
 
-	imageCacheLeaseDuration = 15 * time.Minute
+	// ImageCacheLeaseDuration is stamped on every claimed row up front.
+	// Exported so worker/claim-page sizing can assert a claimed page always
+	// drains inside it (see internal/taskmanager/tasks).
+	ImageCacheLeaseDuration = 15 * time.Minute
 	imageCacheMaxAttempts   = 8
 	imageCacheDeferredRetry = 7 * 24 * time.Hour
 
@@ -427,7 +430,7 @@ func (r *ImageCacheJobRepository) recoverExpiredRunning(ctx context.Context, tar
 		  AND locked_at < NOW() - $1::interval
 	`
 	args := []any{
-		intervalLiteral(imageCacheLeaseDuration),
+		intervalLiteral(ImageCacheLeaseDuration),
 		imageCacheMaxAttempts,
 		intervalLiteral(imageCacheFailedCooldown),
 	}

internal/metadata/image_cache_processor.go

@@ -34,11 +34,21 @@ const (
 	imageCacheDiscoveryBatchSize = 1000
 	// Waiting for a background worker to release a job polls with backoff and
 	// gives up after immediateImageCacheIdleTimeout without progress. The
-	// worker's own lease runs for imageCacheLeaseDuration, and its pod can die
+	// worker's own lease runs for ImageCacheLeaseDuration, and its pod can die
 	// while holding it, so an interactive refresh must not wait that long.
 	immediateImageCacheMinPoll     = 100 * time.Millisecond
 	immediateImageCacheMaxPoll     = 2 * time.Second
 	immediateImageCacheIdleTimeout = 30 * time.Second
+
+	// ImageCacheJobTimeout bounds one job end to end: source check, download
+	// (which also has its own 30-second cap in imagecache), variant encode, and
+	// uploads. Only the download had a deadline before; a hung upload or encode
+	// could hold a job past its claim lease, letting another worker reclaim and
+	// duplicate it. Two minutes is generous for the slowest realistic job, and
+	// worker/claim-page sizing (internal/taskmanager/tasks) relies on it to
+	// prove a claimed page always drains inside ImageCacheLeaseDuration. A job
+	// that hits it is marked failed and retried on the normal backoff.
+	ImageCacheJobTimeout = 2 * time.Minute
 )
 
 // ErrTargetArtworkPending reports that some of a refreshed target's artwork was
@@ -434,7 +444,9 @@ loop:
 		go func(job *models.MetadataImageCacheJob) {
 			defer wg.Done()
 			defer func() { <-sem }()
-			result := p.processOne(ctx, job)
+			jobCtx, cancelJob := context.WithTimeout(ctx, ImageCacheJobTimeout)
+			result := p.processOne(jobCtx, job)
+			cancelJob()
 			mu.Lock()
 			switch result.outcome {
 			case "succeeded":

internal/metadata/image_cache_processor_test.go

@@ -457,7 +457,7 @@ func TestImageCacheProcessorRetriesTargetAfterConcurrentWorkerFinishes(t *testin
 
 func TestImageCacheProcessorStopsWaitingOnStuckBackgroundWorker(t *testing.T) {
 	// A worker that claimed the job and died holds its lease for
-	// imageCacheLeaseDuration. The interactive refresh must not block that
+	// ImageCacheLeaseDuration. The interactive refresh must not block that
 	// long: it gives up and reports the artwork as still pending.
 	jobs := &targetImageCacheJobs{alwaysRunning: true}
 	processor := NewImageCacheProcessorWithTargets(

internal/nodemetrics/meminfo.go

@@ -138,6 +138,35 @@ func parseMeminfoLine(line string) (string, int64, bool) {
 	return strings.TrimSpace(key), value, true
 }
 
+// EffectiveMemoryLimitBytes returns the tightest cgroup memory limit in force
+// on this process, or 0 when none is set.
+//
+// The root-level limit files alone are only right inside a cgroup-namespaced
+// container. A systemd unit with MemoryMax=, or a leaf inheriting a tighter
+// limit from a slice or pod ancestor, publishes its limit at this process's
+// own cgroup or somewhere above it while the root file reads "max" — so this
+// walks the process's own cgroup, every ancestor, and the mount root, and
+// takes the tightest concrete limit found, mirroring what the sampler reports
+// (the kernel OOM-kills against the tightest level).
+func EffectiveMemoryLimitBytes() int64 {
+	return effectiveMemoryLimitBytes("/proc", cgroupMemoryUsagePaths)
+}
+
+func effectiveMemoryLimitBytes(procDir string, layouts []cgroupUsagePath) int64 {
+	relative := cgroupRelativePaths(procDir)
+	tightest := int64(0)
+	for _, level := range withCgroupSelfUsagePaths(relative, layouts) {
+		limit, err := ReadCgroupMemoryLimit(level.limit)
+		if err != nil || limit <= 0 {
+			continue
+		}
+		if tightest == 0 || limit < tightest {
+			tightest = limit
+		}
+	}
+	return tightest
+}
+
 // ReadCgroupMemoryLimit reads a cgroup memory-limit file and returns the limit
 // in bytes, or an error when the cgroup imposes none.
 //

internal/nodemetrics/meminfo_test.go

@@ -0,0 +1,61 @@
+package nodemetrics
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+// A systemd unit with MemoryMax= publishes its limit on the slice above the
+// leaf while both the leaf and the mount root read "max". The effective limit
+// must come from whichever level actually binds, not from the root files
+// alone.
+func TestEffectiveMemoryLimitBytesFindsTheBindingAncestor(t *testing.T) {
+	root := t.TempDir()
+	previousRoot := cgroupMountRoot
+	cgroupMountRoot = root
+	t.Cleanup(func() { cgroupMountRoot = previousRoot })
+
+	leaf := filepath.Join(root, "system.slice", "silo.service")
+	if err := os.MkdirAll(leaf, 0o755); err != nil {
+		t.Fatalf("create %s: %v", leaf, err)
+	}
+	write := func(path, body string) {
+		if err := os.WriteFile(path, []byte(body), 0o644); err != nil {
+			t.Fatalf("write %s: %v", path, err)
+		}
+	}
+	procDir := t.TempDir()
+	if err := os.MkdirAll(filepath.Join(procDir, "self"), 0o755); err != nil {
+		t.Fatalf("create proc self dir: %v", err)
+	}
+	write(filepath.Join(procDir, "self", "cgroup"), "0::/system.slice/silo.service\n")
+
+	layouts := []cgroupUsagePath{{
+		limit:        filepath.Join(root, "memory.max"),
+		usage:        filepath.Join(root, "memory.current"),
+		stat:         filepath.Join(root, "memory.stat"),
+		inactiveFile: cgroupInactiveFileKeyV2,
+	}}
+	const gib = int64(1) << 30
+
+	write(filepath.Join(root, "memory.max"), "max\n")
+	write(filepath.Join(root, "system.slice", "memory.max"), "2147483648\n")
+	write(filepath.Join(leaf, "memory.max"), "max\n")
+	if got := effectiveMemoryLimitBytes(procDir, layouts); got != 2*gib {
+		t.Fatalf("effective limit = %d, want the slice's 2GiB", got)
+	}
+
+	// A leaf tighter than its ancestor binds instead.
+	write(filepath.Join(leaf, "memory.max"), "1073741824\n")
+	if got := effectiveMemoryLimitBytes(procDir, layouts); got != gib {
+		t.Fatalf("effective limit = %d, want the leaf's 1GiB", got)
+	}
+
+	// No concrete limit anywhere reads as no limit, never as a sentinel.
+	write(filepath.Join(leaf, "memory.max"), "max\n")
+	write(filepath.Join(root, "system.slice", "memory.max"), "max\n")
+	if got := effectiveMemoryLimitBytes(procDir, layouts); got != 0 {
+		t.Fatalf("effective limit = %d, want 0 for no limit", got)
+	}
+}

internal/taskmanager/tasks/cache_metadata_images.go

@@ -4,25 +4,112 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"math"
 	"os"
+	"runtime"
+	"runtime/debug"
 	"time"
 
 	"github.com/google/uuid"
 
 	"github.com/Silo-Server/silo-server/internal/metadata"
+	"github.com/Silo-Server/silo-server/internal/nodemetrics"
 	"github.com/Silo-Server/silo-server/internal/taskmanager"
 )
 
 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
-	// tail could expire and be reclaimed before this execution reaches it.
-	cacheMetadataImagesClaimLimit = 2
-	cacheMetadataImagesWorkers    = 2
 	cacheMetadataImagesMaxRuntime = 10 * time.Minute
 )
 
+// imageCacheWorkerMemoryBudget is the memory reserved per concurrent image
+// job when sizing the pool against a detected memory bound. A job can hold
+// the compressed download (capped at 25 MiB in imagecache), the libvips
+// working set for the variant ladder, and — today — a full Go-heap decode of
+// the original for thumbhash, which for a large provider poster reaches the
+// low hundreds of MiB. 512 MiB per worker keeps even a burst of worst-case
+// originals from consuming the whole budget, since the baseline server also
+// lives inside it.
+const imageCacheWorkerMemoryBudget = 512 << 20
+
+// imageCacheWorkerCount sizes the image-cache worker pool for the host. Each
+// job downloads an original (30s timeout in imagecache) and runs a libvips
+// WEBP encode ladder, so the work is a CPU/network mix: 4× the scheduler's
+// CPU count keeps cores busy while other workers wait on downloads, and the
+// cap keeps a many-core server from monopolizing provider connections. The
+// previous fixed pool of 2 measured roughly 60 images/minute on a ~600k-item
+// library; 48 workers measured roughly 2,900/minute over a 60-minute window
+// (RXWatcher/silo-server@3b377f5c2).
+//
+// memoryBytes, when positive, is the tightest detected memory bound for this
+// process (GOMEMLIMIT, cgroup limit, or system memory) and caps the pool at
+// one worker per imageCacheWorkerMemoryBudget so a many-core container with a
+// small memory limit cannot be OOM-killed by concurrent decodes.
+//
+// The floor of 2 is the pool size this task shipped with, and it deliberately
+// overrides the per-worker budget below 2×imageCacheWorkerMemoryBudget: a
+// sub-1GiB deployment ran 2 workers before this sizing existed, so the memory
+// cap never reduces such a host below its long-standing baseline. The budget
+// is a sizing heuristic for how far to scale up, not a reservation.
+func imageCacheWorkerCount(numCPU int, memoryBytes int64) int {
+	workers := min(48, 4*max(numCPU, 1))
+	if memoryBytes > 0 {
+		workers = min(workers, max(int(memoryBytes/imageCacheWorkerMemoryBudget), 2))
+	}
+	return workers
+}
+
+// detectImageCacheMemoryBytes returns the tightest memory bound the process
+// can see, or 0 when none is detectable (macOS dev boxes, bare Linux without
+// cgroups). Every source is consulted and the smallest wins, because none
+// implies the others: GOMEMLIMIT can be set looser than a cgroup limit, and
+// the effective cgroup limit — own cgroup, ancestors, and root, so a systemd
+// MemoryMax= or an inherited pod/slice limit binds, not just a namespaced
+// container's root files — can sit above or below host memory.
+func detectImageCacheMemoryBytes() int64 {
+	goLimit := int64(0)
+	if limit := debug.SetMemoryLimit(-1); limit < math.MaxInt64 {
+		goLimit = limit
+	}
+	hostTotal, _ := nodemetrics.ReadMeminfoTotalBytes("/proc/meminfo")
+	return tightestMemoryLimit(goLimit, nodemetrics.EffectiveMemoryLimitBytes(), hostTotal)
+}
+
+// tightestMemoryLimit returns the smallest positive limit, or 0 when none is.
+func tightestMemoryLimit(limits ...int64) int64 {
+	tightest := int64(0)
+	for _, limit := range limits {
+		if limit > 0 && (tightest == 0 || limit < tightest) {
+			tightest = limit
+		}
+	}
+	return tightest
+}
+
+var cacheMetadataImagesWorkers = imageCacheWorkerCount(runtime.GOMAXPROCS(0), detectImageCacheMemoryBytes())
+
+// cacheMetadataImagesClaimPerWorker sizes the queue page stamped with one
+// lease up front. processClaimedJobs dispatches the page through a semaphore,
+// so a page larger than the worker count keeps the pool saturated instead of
+// waiting on every straggler in a worker-sized batch before the next page can
+// be claimed.
+//
+// The page must drain inside metadata.ImageCacheLeaseDuration (15 minutes) or
+// another worker reclaims the unstarted tail and duplicates it. Every job runs
+// under metadata.ImageCacheJobTimeout (2 minutes), but that context cannot
+// preempt the synchronous decode/encode segment (imageutil.Thumbhash,
+// GenerateVariants take no context) — it only stops the job at the next
+// context-aware step. That segment operates on inputs capped at 25 MiB
+// (imagecache's download limit), so its overshoot is bounded by CPU speed,
+// not by the network; 4 jobs per worker budgets the worst chain at
+// 4 × (timeout + overshoot), which stays inside the lease with a generous
+// overshoot allowance of nearly two minutes per job.
+// TestImageCacheWorkerCount asserts the timeout part of this arithmetic
+// against the exported constants.
+const cacheMetadataImagesClaimPerWorker = 4
+
+var cacheMetadataImagesClaimLimit = cacheMetadataImagesClaimPerWorker * cacheMetadataImagesWorkers
+
 type MetadataImageCacheRunner interface {
 	DrainUntilIdle(ctx context.Context, workerID string, claimLimit int, concurrency int, maxRuntime time.Duration, reportProgress metadata.ImageCacheRunProgressReporter) (metadata.ImageCacheRunStats, error)
 }

internal/taskmanager/tasks/cache_metadata_images_test.go

@@ -115,11 +115,11 @@ func TestCacheMetadataImagesTaskReportsStats(t *testing.T) {
 	if err := task.Execute(context.Background(), progress); err != nil {
 		t.Fatalf("Execute() error = %v", err)
 	}
-	if runner.claimLimit != 2 {
-		t.Fatalf("claimLimit = %d, want one immediately-startable job per worker", runner.claimLimit)
+	if runner.claimLimit != cacheMetadataImagesClaimLimit {
+		t.Fatalf("claimLimit = %d, want the shared page size %d", runner.claimLimit, cacheMetadataImagesClaimLimit)
 	}
-	if runner.concurrency != 2 {
-		t.Fatalf("concurrency = %d, want 2", runner.concurrency)
+	if runner.concurrency != cacheMetadataImagesWorkers {
+		t.Fatalf("concurrency = %d, want the shared worker count %d", runner.concurrency, cacheMetadataImagesWorkers)
 	}
 	if runner.maxRuntime != 10*time.Minute {
 		t.Fatalf("maxRuntime = %s, want 10m", runner.maxRuntime)
@@ -161,8 +161,8 @@ func TestBackfillMetadataImagesTaskReportsDiscovery(t *testing.T) {
 	if runner.maxRuntime != 0 {
 		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)
+	if runner.claimLimit != cacheMetadataImagesClaimLimit {
+		t.Fatalf("claimLimit = %d, want the shared page size %d", runner.claimLimit, cacheMetadataImagesClaimLimit)
 	}
 	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)
@@ -329,3 +329,69 @@ func TestBackfillMetadataImagesTaskProgressDoesNotFallWhenDiscoveryWidensTheRun(
 		t.Fatalf("recovered report = %g, want 75", progress.percents[3])
 	}
 }
+
+func TestImageCacheWorkerCount(t *testing.T) {
+	const gib = int64(1) << 30
+	cases := []struct {
+		numCPU int
+		memory int64
+		want   int
+	}{
+		{numCPU: 0, memory: 0, want: 4}, // defensive floor; GOMAXPROCS never reports < 1
+		{numCPU: 1, memory: 0, want: 4}, // small household box: modest but no longer crippled
+		{numCPU: 4, memory: 0, want: 16},
+		{numCPU: 12, memory: 0, want: 48},
+		{numCPU: 16, memory: 0, want: 48}, // cap: more cores must not monopolize providers
+		{numCPU: 64, memory: 0, want: 48},
+		// A many-core container with a small memory limit is bounded by
+		// memory, never below the original pool of 2 — a sub-1GiB deployment
+		// already ran 2 workers before this change, so 2 is the floor, not 1.
+		{numCPU: 16, memory: 256 << 20, want: 2},
+		{numCPU: 16, memory: 512 << 20, want: 2},
+		{numCPU: 16, memory: 1 * gib, want: 2},
+		{numCPU: 16, memory: 2 * gib, want: 4},
+		{numCPU: 16, memory: 8 * gib, want: 16},
+		{numCPU: 16, memory: 64 * gib, want: 48},
+		// Plenty of memory but few cores: CPU stays the binding cap.
+		{numCPU: 2, memory: 64 * gib, want: 8},
+	}
+	for _, tc := range cases {
+		if got := imageCacheWorkerCount(tc.numCPU, tc.memory); got != tc.want {
+			t.Errorf("imageCacheWorkerCount(%d, %d) = %d, want %d", tc.numCPU, tc.memory, got, tc.want)
+		}
+	}
+	// A claimed page is stamped with one lease up front, so it must fully
+	// drain before the lease expires or another worker reclaims the tail.
+	// The job timeout cannot preempt the synchronous decode/encode segment,
+	// so the timeout-based drain must leave real headroom under the lease
+	// for that bounded overshoot — at least one extra timeout per job.
+	drain := time.Duration(cacheMetadataImagesClaimPerWorker) * metadata.ImageCacheJobTimeout
+	overshootBudget := time.Duration(cacheMetadataImagesClaimPerWorker) * metadata.ImageCacheJobTimeout / 2
+	if drain+overshootBudget > metadata.ImageCacheLeaseDuration {
+		t.Errorf("page drain %s plus overshoot budget %s must stay within the %s claim lease", drain, overshootBudget, metadata.ImageCacheLeaseDuration)
+	}
+	if cacheMetadataImagesClaimLimit != cacheMetadataImagesClaimPerWorker*cacheMetadataImagesWorkers {
+		t.Errorf("claim limit = %d, want %d per worker (%d)", cacheMetadataImagesClaimLimit, cacheMetadataImagesClaimPerWorker, cacheMetadataImagesClaimPerWorker*cacheMetadataImagesWorkers)
+	}
+}
+
+// Conflicting memory bounds resolve to the smallest positive one: GOMEMLIMIT
+// can be set looser than the cgroup limit and the cgroup limit can sit above
+// host memory, so no single source can be preferred outright.
+func TestTightestMemoryLimit(t *testing.T) {
+	const gib = int64(1) << 30
+	cases := []struct {
+		limits []int64
+		want   int64
+	}{
+		{limits: []int64{4 * gib, 2 * gib, 8 * gib}, want: 2 * gib}, // smallest wins
+		{limits: []int64{0, 2 * gib, 0}, want: 2 * gib},             // zeros mean "no bound", not zero bytes
+		{limits: []int64{0, 0, 0}, want: 0},                         // nothing detectable
+		{limits: []int64{6 * gib}, want: 6 * gib},
+	}
+	for _, tc := range cases {
+		if got := tightestMemoryLimit(tc.limits...); got != tc.want {
+			t.Errorf("tightestMemoryLimit(%v) = %d, want %d", tc.limits, got, tc.want)
+		}
+	}
+}