NNTP shares: the account's connections shared by priority

The coordinator (38) and the prober (8) had a fixed split; with history
paused the coordinator's stood idle while the prober's were its limit.
nntp_shares: each process syncs every 15 s (priority, minimum, what it could
use, what it has open) and gets what it may open: minimums first, then by
priority what each wants, growing only into connections the others closed
or gave up. The pool's size and the prober's lanes follow the grant; the
coordinator doubles its want while work waits for a connection, else asks
for its peak in use and a margin; the prober wants its maximum unless the
queue is dry. Claims scale with the lanes.

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

app/services/prober.ts

@@ -6,6 +6,7 @@ import coordinatorConfig from '#config/coordinator'
 import { XoverClient } from '#lib/nntp/index'
 import { YencError } from '#lib/yenc/index'
 import { inspect, innerName, PROBE_VERSION, type Inspection } from '#lib/probe/index'
+import { NntpShare } from '#services/nntp_shares'
 
 interface Candidate {
   id: number
@@ -46,6 +47,12 @@ export class Prober {
   private refilling: Promise<void> | null = null
   private stopped = false
   private stats = Prober.freshStats()
+  /** Lanes running (each one connection, once it has work); at most the share's grant. Counted apart from the set: a lane leaving takes itself off at once. */
+  private lanes = new Set<Promise<void>>()
+  private running = 0
+  /** The last lane to look found nothing to claim: the prober wants its minimum only. */
+  private dry = false
+  private share = new NntpShare('prober', coordinatorConfig.nntpShares.proberPriority, coordinatorConfig.probe.minConnections, coordinatorConfig.probe.connections)
 
   /** Per minute: what was read, and where the lanes' time went (ms summed over lanes; flushes ms on their own). */
   private static freshStats() {
@@ -56,8 +63,7 @@ export class Prober {
     // A previous run's claims: one prober at a time.
     const { rowCount } = await db.rawQuery(`UPDATE probe_candidates SET status = 'waiting' WHERE status = 'taken'`)
     if (rowCount) logger.info(`Probing: ${rowCount} candidates from a previous run back in the queue.`)
-    const connections = coordinatorConfig.probe.connections
-    logger.info(`Probing with ${connections} connections.`)
+    logger.info(`Probing with ${this.share.min}–${this.share.max} connections (its share of the account's).`)
     const flusher = (async () => {
       while (!this.stopped || this.outcomes.length > 0) {
         await sleep(FLUSH_MS)
@@ -67,7 +73,26 @@ export class Prober {
         this.report()
       }
     })()
-    await Promise.all(Array.from({ length: connections }, () => this.lane()))
+    // Lanes as the share allows: the coordinator's history comes first (app/services/nntp_shares.ts).
+    while (!this.stopped) {
+      try {
+        const was = this.share.granted
+        const granted = await this.share.sync(this.dry ? 0 : this.share.max, this.running)
+        if (granted !== was) logger.info(`Probing: ${granted} connections (was ${was}).`)
+        while (this.running < granted) {
+          this.running++
+          const lane: Promise<void> = this.lane()
+            .catch((error) => logger.error(`Probing: a lane failed: ${error.message}`))
+            .finally(() => this.lanes.delete(lane))
+          this.lanes.add(lane)
+        }
+      } catch (error: any) {
+        logger.error(`Probing: NNTP share failed: ${error.message}`)
+      }
+      for (let waited = 0; waited < coordinatorConfig.nntpShares.everyMs && !this.stopped; waited += 1000) await sleep(1000)
+    }
+    await Promise.all(this.lanes)
+    await this.share.leave().catch(() => {})
     // What was claimed and not probed goes back.
     for (const candidate of this.queue.splice(0)) this.outcomes.push({ candidate, status: 'waiting' })
     this.stopped = true
@@ -78,68 +103,82 @@ export class Prober {
     this.stopped = true
   }
 
-  /** One connection's loop: PROBE_PIPELINE articles at a time (pipelined), reconnecting after a failure. */
+  /**
+   * One connection's loop: PROBE_PIPELINE articles at a time (pipelined), reconnecting after a failure; done when
+   * there are more lanes than the share (its connection closed, the coordinator takes it on its next sync).
+   */
   private async lane() {
     let client: XoverClient | null = null
-    while (!this.stopped) {
-      const batch: Candidate[] = []
-      const waited = Date.now()
-      while (batch.length < coordinatorConfig.probe.pipeline) {
-        const candidate = await this.next()
-        if (!candidate) break
-        batch.push(candidate)
-      }
-      this.stats.waitMs += Date.now() - waited
-      if (batch.length === 0) {
-        this.stats.idleMs += 30_000
-        await sleep(30_000)
-        continue
-      }
-      try {
-        client ??= await XoverClient.connect({
-          host: env.get('NNTP_HOST')!,
-          port: env.get('NNTP_PORT', 563),
-          user: env.get('NNTP_USER'),
-          password: env.get('NNTP_PASSWORD'),
-          secure: env.get('NNTP_SECURE', true),
-          compress: false,
-          timeoutMs: coordinatorConfig.nntpTimeoutMs,
-        })
-        const fetched = Date.now()
-        const bodies = await client.bodies(batch.map((c) => c.messageId))
-        this.stats.fetchMs += Date.now() - fetched
-        const inspected = Date.now()
-        for (const [i, candidate] of batch.entries()) {
-          const body = bodies[i]
-          this.stats.articles++
-          if (body === null) {
-            // Taken down or expired: final.
-            this.outcomes.push({ candidate, status: 'missing' })
-            continue
+    try {
+      while (!this.stopped) {
+        if (this.running > this.share.granted) break
+        const batch: Candidate[] = []
+        const waited = Date.now()
+        while (batch.length < coordinatorConfig.probe.pipeline) {
+          const candidate = await this.next()
+          if (!candidate) break
+          batch.push(candidate)
+        }
+        this.stats.waitMs += Date.now() - waited
+        this.dry = batch.length === 0
+        if (batch.length === 0) {
+          this.stats.idleMs += 30_000
+          await sleep(30_000)
+          continue
+        }
+        try {
+          client ??= await XoverClient.connect({
+            host: env.get('NNTP_HOST')!,
+            port: env.get('NNTP_PORT', 563),
+            user: env.get('NNTP_USER'),
+            password: env.get('NNTP_PASSWORD'),
+            secure: env.get('NNTP_SECURE', true),
+            compress: false,
+            timeoutMs: coordinatorConfig.nntpTimeoutMs,
+          })
+          const fetched = Date.now()
+          const bodies = await client.bodies(batch.map((c) => c.messageId))
+          this.stats.fetchMs += Date.now() - fetched
+          const inspected = Date.now()
+          for (const [i, candidate] of batch.entries()) {
+            const body = bodies[i]
+            this.stats.articles++
+            if (body === null) {
+              // Taken down or expired: final.
+              this.outcomes.push({ candidate, status: 'missing' })
+              continue
+            }
+            this.stats.bytes += body.length
+            try {
+              this.outcomes.push({ candidate, status: 'done', inspection: inspect(body) })
+            } catch (error: any) {
+              // Not yEnc, a broken one, or data a parser chokes on: a final answer, not a connection problem.
+              if (!(error instanceof YencError)) logger.warn(`Probing ${candidate.messageId}: ${error.message}`)
+              this.outcomes.push({ candidate, status: 'error' })
+            }
           }
-          this.stats.bytes += body.length
-          try {
-            this.outcomes.push({ candidate, status: 'done', inspection: inspect(body) })
-          } catch (error: any) {
-            // Not yEnc, a broken one, or data a parser chokes on: a final answer, not a connection problem.
-            if (!(error instanceof YencError)) logger.warn(`Probing ${candidate.messageId}: ${error.message}`)
-            this.outcomes.push({ candidate, status: 'error' })
-          }
+          this.stats.inspectMs += Date.now() - inspected
+        } catch (error: any) {
+          logger.warn(`Probing ${batch[0].messageId} and ${batch.length - 1} more: ${error.message}`)
+          for (const candidate of batch) this.outcomes.push({ candidate, status: 'waiting' })
+          client?.close()
+          client = null
+          await sleep(5000)
         }
-        this.stats.inspectMs += Date.now() - inspected
-      } catch (error: any) {
-        logger.warn(`Probing ${batch[0].messageId} and ${batch.length - 1} more: ${error.message}`)
-        for (const candidate of batch) this.outcomes.push({ candidate, status: 'waiting' })
-        client?.close()
-        client = null
-        await sleep(5000)
       }
+    } finally {
+      client?.close()
+      this.running--
     }
-    client?.close()
+  }
+
+  /** Candidates claimed at a time: a few rounds of every lane's pipeline (a claim takes seconds; 400 lasted ~2 s at 37 lanes). */
+  private get batch() {
+    return Math.max(coordinatorConfig.probe.batch, this.running * coordinatorConfig.probe.pipeline * 4)
   }
 
   private async next(): Promise<Candidate | undefined> {
-    if (this.queue.length < coordinatorConfig.probe.batch / 2) {
+    if (this.queue.length < this.batch / 2) {
       this.refilling ??= this.claim().finally(() => (this.refilling = null))
       if (this.queue.length === 0) await this.refilling
     }
@@ -163,7 +202,7 @@ export class Prober {
 
   private async claimTiers() {
     for (const { kind, unreadable } of TIERS) {
-      const want = coordinatorConfig.probe.batch - this.queue.length
+      const want = this.batch - this.queue.length
       if (want <= 0) return
       // The starting points chosen here: in the query Postgres drew random() once for all slices.
       // Over the claimable ones of this tier only: the rest of the queue isn't settled yet.