Rebuild: prefetch report samples what the connections do

Once a second: pool connections with a command in flight, loops
fetching/writing/idle; per request GROUP time, XOVER time to the status
line and the transfer, transfers with a silence over 5 s. The old
"connections busy" (fetch time of the chunks finished in the minute)
said 59-75% while all 40 connections were in flight all the time:
dropped.

Homelab: 39-40 of 40 in flight, no idle loops; the status line comes at
once, the whole 8-35 s is the transfer, with data flowing (few gaps
over 5 s): ~3 MB per chunk at 0.1-0.4 MB/s per connection.

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

app/services/nntp_pool.ts

@@ -32,6 +32,11 @@ class NntpPool {
     return this.acquire().then((client) => this.run(client, work))
   }
 
+  /** Connections with a command in flight (the prefetcher's, the planner's). */
+  get busyCount() {
+    return this.busy.size
+  }
+
   /** How long ago the connection was opened, in ms. */
   ageOf(client: XoverClient): number {
     return Date.now() - (this.openedAt.get(client) ?? Date.now())

app/services/prefetcher.ts

@@ -51,6 +51,10 @@ export class Prefetcher {
   private sizedAt = 0
   /** Fetched since the last report: chunks, articles, ms spent fetching. */
   private stats = this.freshStats()
+  /** Loops in a fetch or a spool write right now. */
+  private fetchingNow = 0
+  private writingNow = 0
+  private sampler?: NodeJS.Timeout
 
   private freshStats() {
     return {
@@ -61,6 +65,21 @@ export class Prefetcher {
       hedgeWon: 0,
       /** First requests per connection-age bucket, and how many of them were slow. */
       byAge: AGE_BUCKETS.map(() => ({ requests: 0, slow: 0 })),
+      /** Spool writes and the range's update, ms. */
+      writeMs: 0,
+      /** Per answered request, ms: GROUP, XOVER until its status line, the transfer; transfers with a silence over 5 s. */
+      requests: 0,
+      groupMs: 0,
+      replyMs: 0,
+      bodyMs: 0,
+      silent: 0,
+      /** Once a second: loops fetching, writing, not running; pool connections with a command in flight. */
+      samples: 0,
+      fetching: 0,
+      writing: 0,
+      idle: 0,
+      inFlight: 0,
+      inFlightMin: Infinity,
       since: Date.now(),
     }
   }
@@ -72,6 +91,7 @@ export class Prefetcher {
 
   /** One round: refreshes the targets now and then, and puts idle connections to work. */
   async tick() {
+    this.sampler ??= setInterval(() => this.sample(), 1000).unref()
     if (Date.now() - this.refreshedAt >= REFRESH_MS) {
       this.refreshedAt = Date.now()
       this.targets = await this.readTargets()
@@ -87,6 +107,7 @@ export class Prefetcher {
 
   stop() {
     this.stopped = true
+    clearInterval(this.sampler)
   }
 
   get active() {
@@ -101,23 +122,36 @@ export class Prefetcher {
       const { target, span } = next
       try {
         const fetchedAt = Date.now()
+        this.fetchingNow++
         const fetch = async (client: Parameters<Parameters<typeof NntpPool.use>[0]>[0]) => {
           await client.group(target.group)
           return client.xoverRaw(span.from, span.to)
         }
-        const data = target.plain.has(span.from) ? await NntpPool.usePlain(fetch) : await this.fetchHedged(target.group, span)
+        let data: Buffer
+        try {
+          data = target.plain.has(span.from) ? await NntpPool.usePlain(fetch) : await this.fetchHedged(target.group, span)
+        } finally {
+          this.fetchingNow--
+        }
         this.stats.chunks++
         this.stats.articles += span.to - span.from + 1
         this.stats.fetchMs += Date.now() - fetchedAt
-        const existed = await Spool.has(target.id, span)
-        await Spool.write(target.id, span, data)
-        if (existed) continue
-        this.spoolBytes += data.length
-        const { rows } = await db.rawQuery(
-          `UPDATE ranges SET spooled_chunks = spooled_chunks + 1 WHERE id = ? RETURNING spooled_chunks`,
-          [target.id]
-        )
-        if (Number(rows[0]?.spooled_chunks) >= target.total) await db.from('ranges').where('id', target.id).update({ spooled_at: new Date() })
+        const writtenAt = Date.now()
+        this.writingNow++
+        try {
+          const existed = await Spool.has(target.id, span)
+          await Spool.write(target.id, span, data)
+          if (existed) continue
+          this.spoolBytes += data.length
+          const { rows } = await db.rawQuery(
+            `UPDATE ranges SET spooled_chunks = spooled_chunks + 1 WHERE id = ? RETURNING spooled_chunks`,
+            [target.id]
+          )
+          if (Number(rows[0]?.spooled_chunks) >= target.total) await db.from('ranges').where('id', target.id).update({ spooled_at: new Date() })
+        } finally {
+          this.writingNow--
+          this.stats.writeMs += Date.now() - writtenAt
+        }
       } finally {
         this.claimed.delete(`${target.id}:${span.from}`)
       }
@@ -142,8 +176,19 @@ export class Prefetcher {
         if (cancelled) return Buffer.alloc(0)
         client = c
         if (first) age = NntpPool.ageOf(c)
+        const groupAt = performance.now()
         await c.group(group)
-        return c.xoverRaw(span.from, span.to)
+        const groupMs = performance.now() - groupAt
+        const data = await c.xoverRaw(span.from, span.to)
+        if (!cancelled) {
+          const t = c.lastTiming
+          this.stats.requests++
+          this.stats.groupMs += groupMs
+          this.stats.replyMs += t.replyMs
+          this.stats.bodyMs += t.bodyMs
+          if (t.longestGapMs > 5000) this.stats.silent++
+        }
+        return data
       })
       if (!done) return null
       done.catch(() => {})
@@ -171,20 +216,39 @@ export class Prefetcher {
     return result.value
   }
 
+  private sample() {
+    const s = this.stats
+    s.samples++
+    s.fetching += this.fetchingNow
+    s.writing += this.writingNow
+    s.idle += Math.max(0, this.loops - this.fetchingNow - this.writingNow)
+    s.inFlight += NntpPool.busyCount
+    s.inFlightMin = Math.min(s.inFlightMin, NntpPool.busyCount)
+  }
+
   /** A line a minute: what the connections fetched, and how long a chunk took. */
   private report() {
     const seconds = (Date.now() - this.stats.since) / 1000
     if (seconds < 60) return
-    const { chunks, articles, fetchMs, hedged, hedgeWon, byAge } = this.stats
+    const { chunks, articles, fetchMs, hedged, hedgeWon, byAge, samples, requests } = this.stats
     if (chunks > 0) {
       const ages = byAge
         .map((b, i) => `${i === 0 ? '<' : `${AGE_BUCKETS[i - 1]}–`}${AGE_BUCKETS[i] === Infinity ? '' : AGE_BUCKETS[i]} min ${b.slow}/${b.requests}`.replace('– min', '+ min'))
         .join(', ')
       logger.info(
         `Prefetched ${chunks} chunks in ${seconds.toFixed(0)} s: ${Math.round(articles / seconds).toLocaleString('en')} articles/s, ` +
-          `${(fetchMs / chunks / 1000).toFixed(1)} s a chunk, loops busy ${Math.round((fetchMs / (seconds * 1000 * this.loops)) * 100)}%; ` +
+          `${(fetchMs / chunks / 1000).toFixed(1)} s a chunk; ` +
           `hedged ${hedged} (second won ${hedgeWon}); slow by connection age: ${ages}.`
       )
+      const st = this.stats
+      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}); ` +
+          `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.`
+      )
     }
     this.stats = this.freshStats()
   }

lib/nntp/xover_client.ts

@@ -40,12 +40,26 @@ export class XoverClient {
   private wake?: () => void
   private failure?: Error
   compressed = false
+  /**
+   * The last xoverRaw's timing, ms: until the status line came (the server
+   * thinking), from there to the end (the transfer), and the longest silence
+   * during the transfer.
+   */
+  lastTiming = { replyMs: 0, bodyMs: 0, longestGapMs: 0 }
+  private measuring = false
+  private lastDataAt = 0
+  private longestGap = 0
 
   private constructor(
     private sock: net.Socket,
     private timeoutMs: number
   ) {
     sock.on('data', (chunk: Buffer) => {
+      if (this.measuring) {
+        const now = performance.now()
+        this.longestGap = Math.max(this.longestGap, now - this.lastDataAt)
+        this.lastDataAt = now
+      }
       this.chunks.push(chunk)
       this.length += chunk.length
       this.wake?.()
@@ -119,12 +133,22 @@ export class XoverClient {
 
   async xoverRaw(from: number, to: number): Promise<Buffer> {
     return this.timed(async () => {
+      const sentAt = performance.now()
       const reply = await this.command(`XOVER ${from}-${to}`)
+      const repliedAt = performance.now()
       if (reply.startsWith('420') || reply.startsWith('423')) return deflateSync(Buffer.alloc(0))
       await this.expect(reply, ['224'])
-      if (reply.includes('COMPRESS=GZIP')) return (await this.compressedRaw(false)).raw
-      // Plain replies (the relay's default): compressed off the event loop, at the fastest level.
-      return deflateAsync(Buffer.from(await this.plainBody()), { level: 1 })
+      this.measuring = true
+      this.lastDataAt = repliedAt
+      this.longestGap = 0
+      try {
+        if (reply.includes('COMPRESS=GZIP')) return (await this.compressedRaw(false)).raw
+        // Plain replies (the relay's default): compressed off the event loop, at the fastest level.
+        return deflateAsync(Buffer.from(await this.plainBody()), { level: 1 })
+      } finally {
+        this.measuring = false
+        this.lastTiming = { replyMs: repliedAt - sentAt, bodyMs: performance.now() - repliedAt, longestGapMs: this.longestGap }
+      }
     })
   }