From 5d3b57832201f3fee713537cc7d3e3c016d45d72 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Mon, 2 Nov 2026 21:12:51 +0100 Subject: [PATCH] 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 --- REBUILD.md | 12 ++- app/services/pending_ladder.ts | 73 ++++++++++++++----- app/services/prober.ts | 17 +++-- commands/pending_classes.ts | 2 +- .../migrations/1770000007100_pending_swept.ts | 26 +++++++ 5 files changed, 102 insertions(+), 28 deletions(-) create mode 100644 database/migrations/1770000007100_pending_swept.ts diff --git a/REBUILD.md b/REBUILD.md index 02e2fff..66d5810 100644 --- a/REBUILD.md +++ b/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 diff --git a/app/services/pending_ladder.ts b/app/services/pending_ladder.ts index 2ef22d2..f01ad14 100644 --- a/app/services/pending_ladder.ts +++ b/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 { + async discard(limit = 10_000, perSweep = 20): Promise { 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 { + 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) diff --git a/app/services/prober.ts b/app/services/prober.ts index d718bc6..7ceb250 100644 --- a/app/services/prober.ts +++ b/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 diff --git a/commands/pending_classes.ts b/commands/pending_classes.ts index 1cd766f..1d91805 100644 --- a/commands/pending_classes.ts +++ b/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). diff --git a/database/migrations/1770000007100_pending_swept.ts b/database/migrations/1770000007100_pending_swept.ts new file mode 100644 index 0000000..3bbb0b0 --- /dev/null +++ b/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') + }) + } +}