From 0ccb850ea4dc722082f5c082871a49390d01b356 Mon Sep 17 00:00:00 2001
From: Daan Meijer <daan@studioseptember.nl>
Date: Wed, 28 Oct 2026 14:52:48 +0100
Subject: [PATCH] Prober: claim ahead of the lanes, not when they run dry

The next claim started at half a batch of four pipeline rounds a lane,
3-6 s of work, while a claim takes 1.5-4 s and more; at 20-24 lanes the
lanes waited for candidates up to 19% of their time. queueLevels: the
claim starts below eight rounds of every lane's pipeline (~13 s) and
fills the queue to twice that. Taken candidates go back to waiting at
start-up as before, so a bigger queue costs nothing on a restart.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 app/services/prober.ts                      | 25 ++++++++++++++++++---
 tests/unit/coordinator/prober_queue.spec.ts | 16 +++++++++++++
 2 files changed, 38 insertions(+), 3 deletions(-)
 create mode 100644 tests/unit/coordinator/prober_queue.spec.ts

diff --git a/app/services/prober.ts b/app/services/prober.ts
index 5a33d93..de8a577 100644
--- a/app/services/prober.ts
+++ b/app/services/prober.ts
@@ -49,6 +49,20 @@ function jsonb(value: unknown): string {
   return JSON.stringify(value, (_key, v) => (typeof v === 'string' ? v.replaceAll('\0', '') : v))
 }
 
+/** Rounds of every lane's pipeline still queued when the next claim starts. */
+const LOW_ROUNDS = 8
+
+/**
+ * The candidate queue's levels: the next claim starts below `low`, LOW_ROUNDS rounds of every lane's pipeline (at
+ * ~0.2 s an article and a pipeline of 8, ~13 s of work), and fills it to `target`, twice that. A claim takes 1.5–4 s
+ * and more (random slices of an 11 M-row table); refilling at half of four rounds left 3–6 s, and at 20–24 lanes the
+ * lanes waited for candidates up to 19% of their time (LOG.md 2026-10-28 (120)). At least `minBatch` either way.
+ */
+export function queueLevels(lanes: number, pipeline: number, minBatch: number): { low: number; target: number } {
+  const low = Math.max(minBatch, lanes * pipeline * LOW_ROUNDS)
+  return { low, target: 2 * low }
+}
+
 /** Slices a claim takes from random points of the queue. */
 const SLICES = 20
 const REPORT_MS = 60_000
@@ -209,13 +223,18 @@ export class Prober {
     }
   }
 
-  /** Candidates claimed at a time: a few rounds of every lane's pipeline (a claim takes seconds; 400 lasted ~2 s at 37 lanes). */
+  /** The queue's low-water mark and the level a claim fills it to (queueLevels). */
+  private get levels() {
+    return queueLevels(this.running, coordinatorConfig.probe.pipeline, coordinatorConfig.probe.batch)
+  }
+
+  /** Claims to the queue's target; the claim after starts when it's below its low-water mark (queueLevels). */
   private get batch() {
-    return Math.max(coordinatorConfig.probe.batch, this.running * coordinatorConfig.probe.pipeline * 4)
+    return this.levels.target
   }
 
   private async next(): Promise<Candidate | undefined> {
-    if (this.queue.length < this.batch / 2) {
+    if (this.queue.length < this.levels.low) {
       this.refilling ??= this.claim().finally(() => (this.refilling = null))
       if (this.queue.length === 0) await this.refilling
     }
diff --git a/tests/unit/coordinator/prober_queue.spec.ts b/tests/unit/coordinator/prober_queue.spec.ts
new file mode 100644
index 0000000..2ed6896
--- /dev/null
+++ b/tests/unit/coordinator/prober_queue.spec.ts
@@ -0,0 +1,16 @@
+import { test } from '@japa/runner'
+import { queueLevels } from '#services/prober'
+
+test.group('Prober / queue levels', () => {
+  test('the next claim starts with eight pipeline rounds of every lane still queued, and fills to twice that', ({
+    assert,
+  }) => {
+    assert.deepEqual(queueLevels(22, 8, 400), { low: 1408, target: 2816 })
+    assert.deepEqual(queueLevels(7, 8, 400), { low: 448, target: 896 })
+  })
+
+  test('few lanes: at least the configured batch', ({ assert }) => {
+    assert.deepEqual(queueLevels(2, 8, 400), { low: 400, target: 800 })
+    assert.deepEqual(queueLevels(0, 8, 400), { low: 400, target: 800 })
+  })
+})