feat(cache): make contiguous evictor unlink concurrency and sweep bound configurable

docs/envs.md

@@ -237,6 +237,8 @@ This document describes the environment variables that can be used to configure
 | ENABLE_CONTIGUOUS_DATA_CACHE_INDEX               | Boolean              | false                                          | Enables the SQLite cache cleanup index: each cache write records {hash, size, cached_at, tier} and a disk-pressure evictor reclaims by querying the (SSD-backed) index instead of walking the HDD-backed cache tree. Uses the LOW/HIGH_WATERMARK_PERCENT + MIN_FREE_BYTES thresholds above |
 | CONTIGUOUS_DATA_CACHE_INDEX_EVICTION_INTERVAL_MS | Number               | 60000                                          | How often (ms) the index-driven evictor checks disk pressure |
 | CONTIGUOUS_DATA_CACHE_INDEX_EVICTION_BATCH_SIZE  | Number               | 1000                                           | Max index rows evicted (and blobs unlinked) per batch within an eviction sweep |
+| CONTIGUOUS_DATA_CACHE_INDEX_UNLINK_CONCURRENCY   | Number               | max(1, floor(UV_THREADPOOL_SIZE/8))            | Concurrent blob unlinks per eviction batch. Each unlink holds a libuv thread, so this is derived from the pool rather than fixed: a large value takes the whole pool and queues every other file operation behind it, on a disk that is already saturated whenever the evictor runs |
+| CONTIGUOUS_DATA_CACHE_INDEX_MAX_BATCHES_PER_SWEEP | Number              | 50                                             | Batches per sweep. `EVICTION_BATCH_SIZE * this` bounds the unlinks a single sweep can issue, so a large backlog is reclaimed over several sweeps instead of one pass holding the disk |
 | CONTIGUOUS_DATA_CACHE_INDEX_UPDATE_ON_READ       | Boolean              | true                                           | Refresh a cache entry's last_access (and promote its tier on a preferred-ArNS read) on cache hits => LRU eviction. Set false for FIFO by cache-write time (e.g. operators without an edge cache who want to avoid the per-hit index writes) |
 | ENABLE_CONTIGUOUS_DATA_CACHE_INDEX_BACKFILL      | Boolean              | false                                          | One-time backfill: on startup, walk the existing on-disk cache once and seed index rows for untracked blobs (insert-if-absent, so live entries are untouched). Needed to adopt a pre-existing cache that predates the index; enable once, then disable. Runs in the background without blocking startup. Resumable across restarts (checkpoints per top-level shard); delete data/contiguous/.cache-index-backfill-checkpoint to force a fresh full pass |
 | CONTIGUOUS_DATA_CACHE_INDEX_BACKFILL_BATCH_SIZE  | Number               | 2000                                           | Rows buffered per backfill insert transaction |

src/config.ts

@@ -2877,6 +2877,26 @@ export const CONTIGUOUS_DATA_CACHE_INDEX_EVICTION_INTERVAL_MS =
 export const CONTIGUOUS_DATA_CACHE_INDEX_EVICTION_BATCH_SIZE =
   +env.varOrDefault('CONTIGUOUS_DATA_CACHE_INDEX_EVICTION_BATCH_SIZE', '1000');
 
+// Concurrent blob unlinks per eviction batch. DERIVED from UV_THREADPOOL_SIZE
+// for the same reason as CHUNK_DATA_CACHE_INDEX_UNLINK_CONCURRENCY: every
+// unlink occupies a libuv thread, so a hard-coded 50 takes the entire pool on a
+// stock node (UV_THREADPOOL_SIZE defaults to 4) and queues every other file
+// operation behind it -- on a device that is, by definition, already saturated
+// when the evictor is running.
+export const CONTIGUOUS_DATA_CACHE_INDEX_UNLINK_CONCURRENCY =
+  env.positiveIntOrDefault(
+    'CONTIGUOUS_DATA_CACHE_INDEX_UNLINK_CONCURRENCY',
+    Math.max(1, Math.floor(UV_THREADPOOL_SIZE / 8)),
+  );
+
+// Batches per sweep. batchSize * this is the upper bound on unlinks issued by a
+// single sweep, so it bounds how long one sweep can hold the disk and the pool.
+export const CONTIGUOUS_DATA_CACHE_INDEX_MAX_BATCHES_PER_SWEEP =
+  env.positiveIntOrDefault(
+    'CONTIGUOUS_DATA_CACHE_INDEX_MAX_BATCHES_PER_SWEEP',
+    50,
+  );
+
 // Whether to refresh a cache entry's last_access (and promote its tier on a
 // preferred-ArNS read) on cache HITS. On => LRU eviction; off => FIFO by
 // cache-write time. Operators without an edge cache see every read at the core,

src/workers/contiguous-data-cache-evictor.test.ts

@@ -100,6 +100,74 @@ describe('ContiguousDataCacheEvictor', () => {
     assert.equal(h.remaining.size, 20);
   });
 
+  // Every unlink occupies a libuv thread for its duration, so an unbounded (or
+  // hard-coded 50) fan-out takes the whole pool on a stock node and queues every
+  // other file operation behind it — on a disk that is already saturated, which
+  // is why the evictor is running at all.
+  it('never exceeds the configured unlink concurrency', async () => {
+    const h = makeHarness({
+      initialUsedPercent: 95,
+      entryCount: 200,
+      freePerEvict: 1,
+    });
+    let inFlight = 0;
+    let peak = 0;
+    (h.dataStore as any).delete = async (hash: string) => {
+      inFlight++;
+      peak = Math.max(peak, inFlight);
+      await new Promise((r) => setImmediate(r));
+      inFlight--;
+      h.unlinked.push(hash);
+    };
+
+    await makeEvictor(h, { batchSize: 50, unlinkConcurrency: 4 }).sweep();
+
+    assert.ok(
+      h.unlinked.length > 4,
+      `expected real work, got ${h.unlinked.length}`,
+    );
+    assert.ok(peak <= 4, `peak concurrent unlinks was ${peak}, limit was 4`);
+  });
+
+  // Clamping these would fail silently in the worst way: Math.max(1, NaN) is
+  // NaN, so the sweep loop runs zero batches and the cache never drains, while
+  // a fractional unlinkConcurrency makes p-limit throw mid-sweep — after the
+  // index rows are deleted but before the blobs are unlinked.
+  it('rejects invalid explicit limits at construction', async () => {
+    const h = makeHarness({
+      initialUsedPercent: 90,
+      entryCount: 10,
+      freePerEvict: 5,
+    });
+    for (const bad of [Number.NaN, Infinity, 0, -1, 2.5]) {
+      assert.throws(
+        () => makeEvictor(h, { unlinkConcurrency: bad }),
+        /unlinkConcurrency must be a positive integer/,
+        `unlinkConcurrency=${bad} should be rejected`,
+      );
+      assert.throws(
+        () => makeEvictor(h, { maxBatchesPerSweep: bad }),
+        /maxBatchesPerSweep must be a positive integer/,
+        `maxBatchesPerSweep=${bad} should be rejected`,
+      );
+    }
+  });
+
+  it('bounds unlinks per sweep by batchSize * maxBatchesPerSweep', async () => {
+    const h = makeHarness({
+      initialUsedPercent: 99,
+      entryCount: 500,
+      freePerEvict: 0, // never recovers: only the sweep bound can stop it
+    });
+
+    await makeEvictor(h, { batchSize: 5, maxBatchesPerSweep: 3 }).sweep();
+
+    // The sweep must stop at the bound rather than draining the index, so the
+    // next sweep resumes instead of one pass holding the disk indefinitely.
+    assert.equal(h.unlinked.length, 15);
+    assert.equal(h.remaining.size, 485);
+  });
+
   it('evicts oldest-first until usage recovers below the low watermark', async () => {
     const h = makeHarness({
       initialUsedPercent: 90, // over high(80)

src/workers/contiguous-data-cache-evictor.ts

@@ -12,14 +12,20 @@ import * as config from '../config.js';
 import * as metrics from '../metrics.js';
 import { ContiguousDataCacheIndex, ContiguousDataStore } from '../types.js';
 
-// Bound work per sweep so a large backlog is reclaimed over several sweeps
-// rather than one unbounded pass; the next sweep resumes.
-const MAX_BATCHES_PER_SWEEP = 50;
-
-// Concurrent blob unlinks per batch. The index row deletes are already batched
-// into one transaction; the unlinks are the remaining (HDD-bound) cost, so run
-// them in parallel to saturate the disk instead of one seek at a time.
-const UNLINK_CONCURRENCY = 50;
+// Explicit limits are validated rather than clamped: Math.max(1, NaN) is NaN,
+// which makes the sweep loop run zero batches and evict nothing; Infinity
+// removes the bound entirely; and a fractional unlinkConcurrency makes p-limit
+// throw mid-sweep, after the index rows are deleted but before the blobs are
+// unlinked, leaving orphans for the reconciler. Failing at construction is the
+// only outcome that cannot silently misbehave in production.
+function positiveIntLimit(name: string, value: number): number {
+  if (!Number.isInteger(value) || value < 1) {
+    throw new Error(
+      `${name} must be a positive integer, received ${String(value)}`,
+    );
+  }
+  return value;
+}
 
 /**
  * Disk-pressure evictor for the contiguous data cache, driven by the SQLite
@@ -44,6 +50,8 @@ export class ContiguousDataCacheEvictor {
   private minFreeBytes: number;
   private intervalMs: number;
   private batchSize: number;
+  private unlinkConcurrency: number;
+  private maxBatchesPerSweep: number;
 
   private timer: NodeJS.Timeout | undefined;
   private sweeping = false;
@@ -58,6 +66,8 @@ export class ContiguousDataCacheEvictor {
     minFreeBytes = config.CONTIGUOUS_DATA_CACHE_MIN_FREE_BYTES,
     intervalMs = config.CONTIGUOUS_DATA_CACHE_INDEX_EVICTION_INTERVAL_MS,
     batchSize = config.CONTIGUOUS_DATA_CACHE_INDEX_EVICTION_BATCH_SIZE,
+    unlinkConcurrency = config.CONTIGUOUS_DATA_CACHE_INDEX_UNLINK_CONCURRENCY,
+    maxBatchesPerSweep = config.CONTIGUOUS_DATA_CACHE_INDEX_MAX_BATCHES_PER_SWEEP,
   }: {
     log: Logger;
     dataStore: ContiguousDataStore;
@@ -68,6 +78,8 @@ export class ContiguousDataCacheEvictor {
     minFreeBytes?: number;
     intervalMs?: number;
     batchSize?: number;
+    unlinkConcurrency?: number;
+    maxBatchesPerSweep?: number;
   }) {
     this.log = log.child({ class: this.constructor.name });
     this.dataStore = dataStore;
@@ -78,6 +90,14 @@ export class ContiguousDataCacheEvictor {
     this.minFreeBytes = minFreeBytes;
     this.intervalMs = intervalMs;
     this.batchSize = Math.max(1, batchSize);
+    this.unlinkConcurrency = positiveIntLimit(
+      'unlinkConcurrency',
+      unlinkConcurrency,
+    );
+    this.maxBatchesPerSweep = positiveIntLimit(
+      'maxBatchesPerSweep',
+      maxBatchesPerSweep,
+    );
   }
 
   start(): void {
@@ -169,7 +189,7 @@ export class ContiguousDataCacheEvictor {
 
       let evicted = 0;
       let bytesFreed = 0;
-      for (let batch = 0; batch < MAX_BATCHES_PER_SWEEP; batch++) {
+      for (let batch = 0; batch < this.maxBatchesPerSweep; batch++) {
         const candidates =
           await this.cacheIndex.selectContiguousDataCacheEvictionCandidates(
             this.batchSize,
@@ -201,7 +221,7 @@ export class ContiguousDataCacheEvictor {
           metrics.cacheIndexEvictedTotal.inc({ reason: 'disk_pressure' });
           metrics.cacheIndexEvictedBytesTotal.inc(size);
         }
-        const unlinkLimit = pLimit(UNLINK_CONCURRENCY);
+        const unlinkLimit = pLimit(this.unlinkConcurrency);
         await Promise.all(
           deletedHashes.map((hash) =>
             unlinkLimit(() =>