Rebuild: give up stalled requests, not slow ones

a.b.0day.stuffz sends steadily but slowly (~400 kB a minute, 5-7
minutes per 100k chunk): the fixed deadline (90 s, doubling up to 360 s)
gave its chunks up again and again, and two of its ranges' worker
streams timed out. A stall is a request that got less than
NNTP_STALL_KB_PER_MINUTE (100) in the last minute (the stuck replies
got ~25 kB a minute); then its connection is closed and the chunk
fetched again. Replaces NNTP_CHUNK_DEADLINE_SECONDS.

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

app/services/prefetcher.ts

@@ -16,6 +16,10 @@ interface Target {
   total: number
 }
 
+/** A request's bytes in are sampled this often; STALL_SAMPLES of them make the minute a stall is judged over. */
+const STALL_SAMPLE_MS = 15_000
+const STALL_SAMPLES = 60_000 / STALL_SAMPLE_MS
+
 /** Connection ages the report counts slow first requests by, in minutes. */
 const AGE_BUCKETS = [2, 10, Infinity]
 
@@ -59,8 +63,6 @@ export class Prefetcher {
   private fetchingNow = 0
   private writingNow = 0
   private sampler?: NodeJS.Timeout
-  /** Deadlines a chunk hit, by range and chunk start: each doubles its next deadline, so a slow chunk still comes. */
-  private lateChunks = new Map<string, number>()
   /** Requests in flight: what, since when, on which connection (the report names the oldest). */
   private inFlight = new Set<{ rangeId: number; from: number; to: number; since: number; client: XoverClient }>()
 
@@ -88,7 +90,7 @@ export class Prefetcher {
       idle: 0,
       inFlight: 0,
       inFlightMin: Infinity,
-      /** Requests given up at nntpChunkDeadlineMs. */
+      /** Requests given up as stalled (nntpStallBytesPerMinute). */
       deadlines: 0,
       ranges: 0,
       groups: 0,
@@ -191,15 +193,19 @@ export class Prefetcher {
         if (first) age = NntpPool.ageOf(c)
         const entry = { rangeId, from: span.from, to: span.to, since: Date.now(), client: c }
         this.inFlight.add(entry)
+        // Bytes in, sampled every STALL_SAMPLE_MS: less than nntpStallBytesPerMinute over the last minute is a stall.
         let late = false
-        const key = `${rangeId}:${span.from}`
-        const deadlineMs = coordinatorConfig.nntpChunkDeadlineMs * 2 ** Math.min(this.lateChunks.get(key) ?? 0, 4)
-        const deadline = setTimeout(() => {
-          late = true
-          this.stats.deadlines++
-          this.lateChunks.set(key, (this.lateChunks.get(key) ?? 0) + 1)
-          c.close()
-        }, deadlineMs)
+        const samples: number[] = []
+        const watch = setInterval(() => {
+          samples.push(c.buffered)
+          if (samples.length <= STALL_SAMPLES) return
+          const minuteAgo = samples.shift()!
+          if (c.buffered - minuteAgo < coordinatorConfig.nntpStallBytesPerMinute) {
+            late = true
+            this.stats.deadlines++
+            c.close()
+          }
+        }, STALL_SAMPLE_MS)
         let data: Buffer
         let groupMs = 0
         try {
@@ -209,12 +215,11 @@ export class Prefetcher {
           data = await c.xoverRaw(span.from, span.to)
         } catch (error) {
           if (!late) throw error
-          throw new Error(`Chunk ${span.from}–${span.to} of range ${rangeId} not in after ${deadlineMs / 1000} s; fetching it again`)
+          throw new Error(`Chunk ${span.from}–${span.to} of range ${rangeId} stalled (under ${coordinatorConfig.nntpStallBytesPerMinute / 1000} kB in a minute, ${Math.round(c.buffered / 1000)} kB in); fetching it again`)
         } finally {
-          clearTimeout(deadline)
+          clearInterval(watch)
           this.inFlight.delete(entry)
         }
-        this.lateChunks.delete(key)
         if (!cancelled) {
           const t = c.lastTiming
           this.stats.requests++
@@ -294,7 +299,7 @@ export class Prefetcher {
           `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; ` +
-          `${st.deadlines} given up after ${coordinatorConfig.nntpChunkDeadlineMs / 1000} s; ${this.oldest()}.`
+          `${st.deadlines} stalled (under ${coordinatorConfig.nntpStallBytesPerMinute / 1000} kB a minute); ${this.oldest()}.`
       )
     }
     this.stats = this.freshStats()