The discard sweep beside the lanes, each dropped signature swept once

After aa40222 the prober still idled: a dropped signature's range of the
ladder index is minutes of HDD (309 dropped, 242 with a poster tested on
the rows) and the sweep read the same ranges again every five minutes
inside the claim. Now each dropped class and signature is swept once
(swept_at, migration 7100; null again when the record drops it anew),
20 of each per sweep in rounds of 10k rows, in a loop of its own in the
prober beside the lanes and the namer (LOG.md (133)).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

REBUILD.md

@@ -935,12 +935,16 @@ f. **Nothing enters the catalog without a usable name** (amended
    target or status 3 (exhausted); each finished climb counts on
    `pending_signatures` for (tool, group, poster) and (tool, group), and a
    signature with PENDING_MIN_PROBES (100) climbs and at most
-   PENDING_MAX_NAMED (0.5%) named is dropped: every five minutes the
-   waiting releases of dropped signatures are discarded (status 2), the
-   rules' `samplePercent` of them left to climb. The namer
+   PENDING_MAX_NAMED (0.5%) named is dropped: a sweep beside the lanes
+   (every five minutes, in its own loop: a signature's rows are minutes
+   of HDD) discards the waiting releases of each dropped signature and
+   class once (`swept_at`; the edge drops their new releases itself), the
+   rules' `samplePercent` of them left to climb (LOG.md (133)). The namer
    (`app/services/pending_namer.ts`, in the prober's process): a named
    release's article span read again from the corpus (the stored range's
-   chunks) or by XOVER on a connection of its own, through the TypeScript
+   chunks) or by XOVER on a connection of its own, a chunk of articles
+   at a time (one string of a wide span topped V8's limit; spans over 3M
+   articles are given up), through the TypeScript
    indexer with `partialFiles` (a file given up short of parts stays a
    partial file of its release, as the Rust window writes it; without it
    the range path's fragment rule left nothing to find: by 2026-10-29

app/services/pending_ladder.ts

@@ -315,7 +315,8 @@ class PendingLadder {
       [JSON.stringify([...counts.values()])]
     )
     await trx.rawQuery(
-      `UPDATE pending_signatures s SET dropped = (s.probed >= ?::int AND s.named::float <= s.probed * ?::float)
+      `UPDATE pending_signatures s SET dropped = (s.probed >= ?::int AND s.named::float <= s.probed * ?::float),
+         swept_at = CASE WHEN s.dropped THEN s.swept_at END
        FROM jsonb_to_recordset(?::jsonb) AS v(tool text, group_id int, poster text)
        WHERE s.tool = v.tool AND s.group_id = v.group_id AND s.poster = v.poster`,
       [
@@ -362,7 +363,8 @@ class PendingLadder {
       [JSON.stringify([...classes.values()])]
     )
     await trx.rawQuery(
-      `UPDATE pending_classes c SET dropped = (c.probed >= ?::int AND c.named::float <= c.probed * ?::float)
+      `UPDATE pending_classes c SET dropped = (c.probed >= ?::int AND c.named::float <= c.probed * ?::float),
+         swept_at = CASE WHEN c.dropped THEN c.swept_at END
        FROM jsonb_to_recordset(?::jsonb) AS v(tool text, group_id int, klass int)
        WHERE c.tool = v.tool AND c.group_id = v.group_id AND c.klass = v.klass`,
       [
@@ -377,26 +379,61 @@ class PendingLadder {
 
   /**
    * Waiting releases of a dropped class or signature discarded without a probe, the explored sample excepted; how
-   * many. Driven from the dropped classes and signatures into the claim's index and the ladder's: a scan of the
-   * waiting rows testing each (44M on 2026-11-02) ran 9+ minutes on the HDD every sweep (LOG.md (133)).
+   * many. Each dropped class or signature is swept once (swept_at; null again when the record drops it anew),
+   * `perSweep` of them at most, each by its index (the claim's for a class, the ladder's for a signature, whose
+   * poster is tested on the rows) in rounds of `limit`. A sweep of a signature's rows takes minutes on the HDD,
+   * so the prober runs this beside the lanes, not in the claim (LOG.md (133)); the edge drops new releases of
+   * dropped signatures and classes itself, so one sweep is all a drop needs.
    */
-  async discard(limit = 10_000): Promise<number> {
+  async discard(limit = 10_000, perSweep = 20): Promise<number> {
     const { samplePercent } = await RuleSets.current()
-    const { rows: byClass } = await db.rawQuery(
-      `SELECT encode(p.release_key, 'hex') AS release_key FROM pending_classes c
-       JOIN pending_releases p ON p.status = ? AND p.tool = c.tool AND p.group_id = c.group_id AND p.klass = c.klass
-       WHERE c.dropped LIMIT ?`,
-      [PENDING.waiting, limit]
+    let discarded = 0
+    const { rows: classes } = await db.rawQuery(
+      `SELECT tool, group_id, klass FROM pending_classes WHERE dropped AND swept_at IS NULL ORDER BY group_id, tool, klass LIMIT ?`,
+      [perSweep]
     )
-    const { rows: bySignature } = await db.rawQuery(
-      `SELECT encode(p.release_key, 'hex') AS release_key FROM pending_signatures s
-       JOIN pending_releases p ON p.status = ? AND p.group_id = s.group_id AND p.tool = s.tool AND (s.poster = '' OR s.poster = coalesce(p.poster, ''))
-       WHERE s.dropped LIMIT ?`,
-      [PENDING.waiting, limit]
+    for (const c of classes) {
+      let rows: any[]
+      do {
+        ;({ rows } = await db.rawQuery(
+          `SELECT encode(release_key, 'hex') AS release_key FROM pending_releases
+           WHERE status = ? AND tool = ? AND group_id = ? AND klass = ? LIMIT ?`,
+          [PENDING.waiting, c.tool, c.group_id, c.klass, limit]
+        ))
+        discarded += await this.discardKeys(rows, samplePercent)
+      } while (rows.length === limit && rows.some((r) => !explored(r.release_key, samplePercent)))
+      await db.rawQuery(
+        `UPDATE pending_classes SET swept_at = now() WHERE tool = ? AND group_id = ? AND klass = ?`,
+        [c.tool, c.group_id, c.klass]
+      )
+    }
+    const { rows: signatures } = await db.rawQuery(
+      `SELECT tool, group_id, poster FROM pending_signatures WHERE dropped AND swept_at IS NULL ORDER BY group_id, tool, poster LIMIT ?`,
+      [perSweep]
     )
-    const keys = [
-      ...new Set([...byClass, ...bySignature].map((r: any) => r.release_key as string)),
-    ].filter((key: string) => !explored(key, samplePercent))
+    for (const s of signatures) {
+      let rows: any[]
+      do {
+        ;({ rows } = await db.rawQuery(
+          `SELECT encode(release_key, 'hex') AS release_key FROM pending_releases
+           WHERE status = ? AND group_id = ? AND tool = ? AND (?::text = '' OR coalesce(poster, '') = ?::text) LIMIT ?`,
+          [PENDING.waiting, s.group_id, s.tool, s.poster, s.poster, limit]
+        ))
+        discarded += await this.discardKeys(rows, samplePercent)
+      } while (rows.length === limit && rows.some((r) => !explored(r.release_key, samplePercent)))
+      await db.rawQuery(
+        `UPDATE pending_signatures SET swept_at = now() WHERE tool = ? AND group_id = ? AND poster = ?`,
+        [s.tool, s.group_id, s.poster]
+      )
+    }
+    return discarded
+  }
+
+  /** The rows discarded (status 2), the explored sample excepted, their classes' waiting counts lowered; how many. */
+  private async discardKeys(rows: any[], samplePercent: number): Promise<number> {
+    const keys = rows
+      .map((r: any) => r.release_key as string)
+      .filter((key: string) => !explored(key, samplePercent))
     if (keys.length === 0) return 0
     const { rows: gone } = await db.rawQuery(
       `UPDATE pending_releases SET status = ? WHERE status = ? AND release_key = ANY(SELECT decode(k, 'hex') FROM unnest(?::text[]) AS k)

app/services/prober.ts

@@ -97,7 +97,6 @@ export class Prober {
   }
   /** The pending ladder's namer: named pending releases into the catalog, alongside the lanes. */
   private namer = new PendingNamer()
-  private sweptAt = 0
 
   async run() {
     // A previous run's claims: one prober at a time.
@@ -106,6 +105,17 @@ export class Prober {
     const released = await PendingLadder.release()
     if (released) logger.info(`Probing: ${released} pending releases from a previous run back on the ladder.`)
     const namer = this.namer.run(() => this.stopped).catch((error) => logger.error(`Naming failed: ${error.message}`))
+    // The ladder's discard sweep beside the lanes: a dropped signature's rows take minutes on the HDD (LOG.md (133)).
+    const sweeper = (async () => {
+      while (!this.stopped) {
+        try {
+          this.stats.discarded += await PendingLadder.discard()
+        } catch (error: any) {
+          logger.error(`Probing: the discard sweep failed: ${error.message}`)
+        }
+        for (let waited = 0; waited < SWEEP_MS && !this.stopped; waited += 1000) await sleep(1000)
+      }
+    })()
     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) {
@@ -136,6 +146,7 @@ export class Prober {
     }
     await Promise.all(this.lanes)
     await namer
+    await sweeper
     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' })
@@ -266,10 +277,6 @@ export class Prober {
 
   /** The pending ladder's share of a claim: at least pendingShare of the batch, and whatever the catalog's tiers left. */
   private async claimPending() {
-    if (Date.now() - this.sweptAt > SWEEP_MS) {
-      this.sweptAt = Date.now()
-      this.stats.discarded += await PendingLadder.discard()
-    }
     const want = Math.max(Math.floor(this.batch * coordinatorConfig.probe.pendingShare), this.batch - this.queue.length)
     const probes = await PendingLadder.claim(want)
     if (probes.length === 0) return

commands/pending_classes.ts

@@ -47,7 +47,7 @@ export default class PendingClasses extends BaseCommand {
     )
     const { pendingMinProbes, pendingMaxNamed } = coordinatorConfig.probe
     await db.rawQuery(
-      `UPDATE pending_classes SET dropped = true WHERE NOT dropped AND probed >= ?::int AND named::float <= probed * ?::float`,
+      `UPDATE pending_classes SET dropped = true, swept_at = NULL WHERE NOT dropped AND probed >= ?::int AND named::float <= probed * ?::float`,
       [pendingMinProbes, pendingMaxNamed]
     )
     // The waiting rows by windows of created_at (pending_releases_waiting_idx reaches them).

database/migrations/1770000007100_pending_swept.ts

@@ -0,0 +1,26 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * The ladder's discard sweep (PendingLadder.discard) reads a dropped signature's or class's waiting rows once:
+ * `swept_at` marks it done (null again when the record drops it anew). Before, every sweep read every dropped
+ * signature's rows over again, minutes of HDD each, inside the claim (LOG.md (133)).
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.alterTable('pending_signatures', (table) => {
+      table.timestamp('swept_at', { useTz: true }).nullable()
+    })
+    this.schema.alterTable('pending_classes', (table) => {
+      table.timestamp('swept_at', { useTz: true }).nullable()
+    })
+  }
+
+  async down() {
+    this.schema.alterTable('pending_signatures', (table) => {
+      table.dropColumn('swept_at')
+    })
+    this.schema.alterTable('pending_classes', (table) => {
+      table.dropColumn('swept_at')
+    })
+  }
+}