Rebuild: spread prefetch connections over ranges and groups

At most PREFETCH_PER_RANGE (2) connections on a range and
PREFETCH_PER_GROUP (4) on a group; SPOOL_AHEAD_PER_LEASE 2 -> 4 so there
are targets enough. The report says how many ranges and groups are in
flight.

Homelab, 40 connections, alternated 12 min each: unlimited (~10 ranges
in flight) 281k and 167k articles/s; spread (~22 ranges) 319k and 296k,
chunks 11.5 s against 13-24 s.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/services/prefetcher.ts

@@ -27,8 +27,9 @@ const SIZE_MS = 30_000
 /**
  * Keeps the news server busy ahead of the workers (REBUILD.md, prefetch).
  * Every connection takes the next missing chunk: the leased ranges' first
- * (a worker waits for them; all connections work on the same range, in
- * order), then the next planned ranges, about SPOOL_AHEAD_PER_LEASE per
+ * (a worker waits for them), then the next planned ranges, with at most
+ * PREFETCH_PER_RANGE connections on a range and PREFETCH_PER_GROUP on a
+ * group (Eweka answers faster spread out), about SPOOL_AHEAD_PER_LEASE per
  * active lease (at least SPOOL_MIN_AHEAD), within SPOOL_MAX_GB. One
  * connection fills ~26 k overview lines/s, so a range on one connection
  * kept its worker waiting (LOG.md 2026-10-09). A chunk that takes long is
@@ -44,6 +45,9 @@ export class Prefetcher {
 
   private targets: Target[] = []
   private claimed = new Set<string>()
+  /** Chunks being fetched per range and per group (spool.perRange, spool.perGroup). */
+  private perRange = new Map<number, number>()
+  private perGroup = new Map<string, number>()
   private running = 0
   private refreshedAt = 0
   private stopped = false
@@ -80,6 +84,8 @@ export class Prefetcher {
       idle: 0,
       inFlight: 0,
       inFlightMin: Infinity,
+      ranges: 0,
+      groups: 0,
       since: Date.now(),
     }
   }
@@ -154,6 +160,7 @@ export class Prefetcher {
         }
       } finally {
         this.claimed.delete(`${target.id}:${span.from}`)
+        this.count(target, -1)
       }
     }
   }
@@ -224,6 +231,8 @@ export class Prefetcher {
     s.idle += Math.max(0, this.loops - this.fetchingNow - this.writingNow)
     s.inFlight += NntpPool.busyCount
     s.inFlightMin = Math.min(s.inFlightMin, NntpPool.busyCount)
+    s.ranges += this.perRange.size
+    s.groups += this.perGroup.size
   }
 
   /** A line a minute: what the connections fetched, and how long a chunk took. */
@@ -244,7 +253,8 @@ export class Prefetcher {
       const avg = (n: number) => (samples > 0 ? n / samples : 0)
       const per = (ms: number) => (requests > 0 ? (ms / requests / 1000).toFixed(1) : '-')
       logger.info(
-        `Connections: ${avg(st.inFlight).toFixed(1)} of ${NntpPool.size} with a command in flight (min ${st.inFlightMin}); ` +
+        `Connections: ${avg(st.inFlight).toFixed(1)} of ${NntpPool.size} with a command in flight (min ${st.inFlightMin}), ` +
+          `on ${avg(st.ranges).toFixed(1)} ranges of ${avg(st.groups).toFixed(1)} groups; ` +
           `loops ${avg(st.fetching).toFixed(1)} fetching, ${avg(st.writing).toFixed(1)} writing, ${avg(st.idle).toFixed(1)} idle; ` +
           `a request: GROUP ${per(st.groupMs)} s, XOVER reply ${per(st.replyMs)} s, transfer ${per(st.bodyMs)} s; ` +
           `${st.silent} of ${requests} transfers silent >5 s; writing ${(st.writeMs / Math.max(1, chunks)).toFixed(0)} ms a chunk.`
@@ -254,16 +264,33 @@ export class Prefetcher {
   }
 
   private peek(): boolean {
-    return this.targets.some((t) => t.missing.some((s) => !this.claimed.has(`${t.id}:${s.from}`)))
+    return this.targets.some((t) => this.open(t) && t.missing.some((s) => !this.claimed.has(`${t.id}:${s.from}`)))
   }
 
-  /** The first unclaimed missing chunk, in target order. */
+  /** Whether the target is under its range's and its group's limit of connections. */
+  private open(target: Target) {
+    const { perRange, perGroup } = coordinatorConfig.spool
+    return (perRange <= 0 || (this.perRange.get(target.id) ?? 0) < perRange) && (perGroup <= 0 || (this.perGroup.get(target.group) ?? 0) < perGroup)
+  }
+
+  private count(target: Target, by: number) {
+    const range = (this.perRange.get(target.id) ?? 0) + by
+    const group = (this.perGroup.get(target.group) ?? 0) + by
+    if (range > 0) this.perRange.set(target.id, range)
+    else this.perRange.delete(target.id)
+    if (group > 0) this.perGroup.set(target.group, group)
+    else this.perGroup.delete(target.group)
+  }
+
+  /** The first unclaimed missing chunk, in target order, of a target under its limits. */
   private take(): { target: Target; span: ChunkSpan } | null {
     for (const target of this.targets) {
+      if (!this.open(target)) continue
       const index = target.missing.findIndex((s) => !this.claimed.has(`${target.id}:${s.from}`))
       if (index === -1) continue
       const [span] = target.missing.splice(index, 1)
       this.claimed.add(`${target.id}:${span.from}`)
+      this.count(target, 1)
       return { target, span }
     }
     return null

config/coordinator.ts

@@ -58,9 +58,20 @@ export default {
   spool: {
     dir: env.get('SPOOL_DIR', app.tmpPath('spool')),
     /** Ranges kept fetched ahead, per active lease. */
-    aheadPerLease: env.get('SPOOL_AHEAD_PER_LEASE', 2),
+    aheadPerLease: env.get('SPOOL_AHEAD_PER_LEASE', 4),
     /** At least this many ranges ahead, so the first leases start at once. */
     minAhead: env.get('SPOOL_MIN_AHEAD', 2),
     maxBytes: env.get('SPOOL_MAX_GB', 10) * 1e9,
+    /**
+     * At most this many connections on one range, and on one group, at a
+     * time (0: no limit): spreads the requests over ranges and groups instead
+     * of all on the next chunks of the first ranges. Homelab, 40 connections,
+     * 12 min each, alternated: unlimited (on ~10 ranges) 281k and 167k
+     * articles/s, 2 per range and 4 per group (on ~22) 319k and 296k, with
+     * shorter chunk times (11.5 vs 13–24 s; LOG.md 2026-10-09 (18)). Needs
+     * targets for it: aheadPerLease 4.
+     */
+    perRange: env.get('PREFETCH_PER_RANGE', 2),
+    perGroup: env.get('PREFETCH_PER_GROUP', 4),
   },
 }

start/env.ts

@@ -62,4 +62,6 @@ export default await Env.create(new URL('../', import.meta.url), {
   SPOOL_AHEAD_PER_LEASE: Env.schema.number.optional(),
   SPOOL_MIN_AHEAD: Env.schema.number.optional(),
   SPOOL_MAX_GB: Env.schema.number.optional(),
+  PREFETCH_PER_RANGE: Env.schema.number.optional(),
+  PREFETCH_PER_GROUP: Env.schema.number.optional(),
 })