diff --git a/internal/metadata/image_cache_job_repo.go b/internal/metadata/image_cache_job_repo.go index 706ad8a1f4..ddeab1d1ad 100644 --- a/internal/metadata/image_cache_job_repo.go +++ b/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), } diff --git a/internal/metadata/image_cache_processor.go b/internal/metadata/image_cache_processor.go index ce3b4b2bb9..b6369149cf 100644 --- a/internal/metadata/image_cache_processor.go +++ b/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": diff --git a/internal/metadata/image_cache_processor_test.go b/internal/metadata/image_cache_processor_test.go index aa77734b62..473b6dfd9a 100644 --- a/internal/metadata/image_cache_processor_test.go +++ b/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( diff --git a/internal/nodemetrics/meminfo.go b/internal/nodemetrics/meminfo.go index 581bef4a10..24b3c90377 100644 --- a/internal/nodemetrics/meminfo.go +++ b/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. // diff --git a/internal/nodemetrics/meminfo_test.go b/internal/nodemetrics/meminfo_test.go new file mode 100644 index 0000000000..23b9c3513f --- /dev/null +++ b/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) + } +} diff --git a/internal/taskmanager/tasks/cache_metadata_images.go b/internal/taskmanager/tasks/cache_metadata_images.go index 58d37d98e9..5bd72eab82 100644 --- a/internal/taskmanager/tasks/cache_metadata_images.go +++ b/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) } diff --git a/internal/taskmanager/tasks/cache_metadata_images_test.go b/internal/taskmanager/tasks/cache_metadata_images_test.go index 0814eb10a3..d1166b32ff 100644 --- a/internal/taskmanager/tasks/cache_metadata_images_test.go +++ b/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) + } + } +}