From 8e008da364649965a1d1d1514c213e02ac7bc527 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Tue, 3 Nov 2026 10:36:19 +0100 Subject: [PATCH] Prober: the sweep by a signature index, pending rows locked in one order, deadlocked flushes kept The discard sweep of a dropped signature read the whole pending_releases table per signature (a parallel seq scan over the poster test, four minutes on the HDD that starved the lanes' claims): a partial index on (group_id, tool, poster) WHERE status = 0 makes it a range, and the sweep queries a poster's rows by it and a tool's and group's by the ladder index. The batch store's pending upsert, the record, and the sweep deadlocked on pending_releases and pending_classes rows locked in different orders, and a deadlocked flush dropped its outcomes: lockOrder() sorts every multi-row statement's set the same way, and a flush that deadlocks anyway keeps its outcomes for the next. Co-Authored-By: Claude Fable 5.1 --- REBUILD.md | 13 ++-- app/services/batch_store.ts | 10 ++- app/services/pending_ladder.ts | 64 ++++++++++++++----- app/services/prober.ts | 27 +++++++- .../1770000007200_pending_signature_index.ts | 20 ++++++ 5 files changed, 110 insertions(+), 24 deletions(-) create mode 100644 database/migrations/1770000007200_pending_signature_index.ts diff --git a/REBUILD.md b/REBUILD.md index 66d5810..3327149 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -936,10 +936,15 @@ f. **Nothing enters the catalog without a usable name** (amended `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: 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 + (every five minutes, in its own loop) 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, each by an index (`pending_releases_signature_idx` for a + poster's rows: tested on the rows, a signature was a scan of the whole + table; LOG.md (133), (134)). The batch store, the record, the claim + and the sweep lock pending rows in one order (`lockOrder`), and a + flush that deadlocks anyway keeps its outcomes for the next (LOG.md + (134)). 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, a chunk of articles diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index 70d8b42..efce2ec 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -9,7 +9,7 @@ import { SIDECAR } from '#services/sidecar_folder' import { identityKey, releaseKey } from '#lib/release/keys' import { sourceCode } from '#lib/release/source' import { uploadTool } from '#lib/rules/index' -import { byEffort } from '#services/pending_ladder' +import { byEffort, lockOrder } from '#services/pending_ladder' import { TEXT_KINDS, TEXT_MAX, textsToStore } from '#lib/release/text_kind' /** @@ -232,7 +232,8 @@ class BatchStore { target_bytes = coalesce(pending_releases.target_bytes, excluded.target_bytes) WHERE NOT pending_releases.finished RETURNING release_key, tool, klass, (xmax = 0) AS inserted`, - [groupId, JSON.stringify(rows)] + // In the record's lock order: the upsert locks a finished row too, before its WHERE passes it by. + [groupId, JSON.stringify(lockOrder(rows, (r) => r.release_key))] ) written += done.length // Each class seen gets its row in the ladder's record, so the claim explores it (pending_classes). @@ -248,7 +249,10 @@ class BatchStore { `INSERT INTO pending_classes (tool, group_id, klass, waiting) SELECT v.tool, ?, v.klass, v.waiting FROM jsonb_to_recordset(?::jsonb) AS v(tool text, klass int, waiting int) ON CONFLICT (tool, group_id, klass) DO UPDATE SET waiting = pending_classes.waiting + excluded.waiting`, - [groupId, JSON.stringify([...classes.values()])] + [ + groupId, + JSON.stringify(lockOrder(classes.values(), (c) => `${c.tool}\t${groupId}\t${c.klass}`)), + ] ) } } diff --git a/app/services/pending_ladder.ts b/app/services/pending_ladder.ts index f01ad14..05c4cf8 100644 --- a/app/services/pending_ladder.ts +++ b/app/services/pending_ladder.ts @@ -72,6 +72,19 @@ interface ClassRow { const CLASSES_MS = 60_000 /** A class found short of a claim's share is left alone this long. */ const EMPTY_MS = 5 * 60_000 +/** + * Rows in one order for a statement that locks many of them (pending_releases by key, pending_classes and + * pending_signatures by theirs): the batch store, the record, the claim and the sweep all lock the same way, so two + * of them can't deadlock on each other (LOG.md (134)). + */ +export function lockOrder(rows: Iterable, key: (row: T) => string): T[] { + return [...rows].sort((a, b) => { + const ka = key(a) + const kb = key(b) + return ka < kb ? -1 : ka > kb ? 1 : 0 + }) +} + /** The prior rate of an unexplored class and its weight in probes: the ladder's overall rate on 2026-11-02 (LOG.md (130)). */ const PRIOR_RATE = 0.15 const PRIOR_WEIGHT = 10 @@ -282,7 +295,14 @@ class PendingLadder { probed_at = CASE WHEN v.inspection IS NULL THEN p.probed_at ELSE now() END FROM jsonb_to_recordset(?::jsonb) AS v(release_key text, status int, tries int, named text, inspection jsonb) WHERE p.release_key = decode(v.release_key, 'hex')`, - [JSON.stringify(rows.map(({ final: _final, ...row }) => row))] + [ + JSON.stringify( + lockOrder( + rows.map(({ final: _final, ...row }) => row), + (r) => r.release_key + ) + ), + ] ) // The signatures: the poster's own and the tool's and group's together. const counts = new Map< @@ -312,7 +332,7 @@ class PendingLadder { `INSERT INTO pending_signatures (tool, group_id, poster, probed, named) SELECT v.tool, v.group_id, v.poster, v.probed, v.named FROM jsonb_to_recordset(?::jsonb) AS v(tool text, group_id int, poster text, probed int, named int) ON CONFLICT (tool, group_id, poster) DO UPDATE SET probed = pending_signatures.probed + excluded.probed, named = pending_signatures.named + excluded.named, updated_at = now()`, - [JSON.stringify([...counts.values()])] + [JSON.stringify(lockOrder(counts.values(), (c) => `${c.tool}\t${c.group_id}\t${c.poster}`))] ) await trx.rawQuery( `UPDATE pending_signatures s SET dropped = (s.probed >= ?::int AND s.named::float <= s.probed * ?::float), @@ -323,7 +343,9 @@ class PendingLadder { pendingMinProbes, pendingMaxNamed, JSON.stringify( - [...counts.values()].map(({ tool, group_id, poster }) => ({ tool, group_id, poster })) + lockOrder(counts.values(), (c) => `${c.tool}\t${c.group_id}\t${c.poster}`).map( + ({ tool, group_id, poster }) => ({ tool, group_id, poster }) + ) ), ] ) @@ -360,7 +382,7 @@ class PendingLadder { SELECT v.tool, v.group_id, v.klass, v.probed, v.named, v.waiting FROM jsonb_to_recordset(?::jsonb) AS v(tool text, group_id int, klass int, probed int, named int, waiting int) ON CONFLICT (tool, group_id, klass) DO UPDATE SET probed = pending_classes.probed + excluded.probed, named = pending_classes.named + excluded.named, waiting = pending_classes.waiting + excluded.waiting, updated_at = now()`, - [JSON.stringify([...classes.values()])] + [JSON.stringify(lockOrder(classes.values(), (c) => `${c.tool}\t${c.group_id}\t${c.klass}`))] ) await trx.rawQuery( `UPDATE pending_classes c SET dropped = (c.probed >= ?::int AND c.named::float <= c.probed * ?::float), @@ -371,7 +393,9 @@ class PendingLadder { pendingMinProbes, pendingMaxNamed, JSON.stringify( - [...classes.values()].map(({ tool, group_id, klass }) => ({ tool, group_id, klass })) + lockOrder(classes.values(), (c) => `${c.tool}\t${c.group_id}\t${c.klass}`).map( + ({ tool, group_id, klass }) => ({ tool, group_id, klass }) + ) ), ] ) @@ -380,10 +404,10 @@ class PendingLadder { /** * Waiting releases of a dropped class or signature discarded without a probe, the explored sample excepted; how * 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. + * `perSweep` of them at most, each by its index (the claim's for a class, the signature index for a poster's + * rows, the ladder's for a tool's and group's) in rounds of `limit`. 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, perSweep = 20): Promise { const { samplePercent } = await RuleSets.current() @@ -414,11 +438,20 @@ class PendingLadder { 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] - )) + // The poster's rows by the signature index, a tool's and group's by the ladder's: never a test on the rows, + // which was a scan of the whole table per signature on the HDD (LOG.md (134)). + ;({ rows } = + s.poster === '' + ? await db.rawQuery( + `SELECT encode(release_key, 'hex') AS release_key FROM pending_releases + WHERE status = ? AND group_id = ? AND tool = ? LIMIT ?`, + [PENDING.waiting, s.group_id, s.tool, limit] + ) + : await db.rawQuery( + `SELECT encode(release_key, 'hex') AS release_key FROM pending_releases + WHERE status = ? AND group_id = ? AND tool = ? AND poster = ? LIMIT ?`, + [PENDING.waiting, s.group_id, s.tool, s.poster, limit] + )) discarded += await this.discardKeys(rows, samplePercent) } while (rows.length === limit && rows.some((r) => !explored(r.release_key, samplePercent))) await db.rawQuery( @@ -434,6 +467,7 @@ class PendingLadder { const keys = rows .map((r: any) => r.release_key as string) .filter((key: string) => !explored(key, samplePercent)) + .sort() 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) @@ -457,7 +491,7 @@ class PendingLadder { `UPDATE pending_classes c SET waiting = greatest(0, c.waiting - v.n) FROM jsonb_to_recordset(?::jsonb) AS v(tool text, group_id int, klass int, n int) WHERE c.tool = v.tool AND c.group_id = v.group_id AND c.klass = v.klass`, - [JSON.stringify([...counts.values()])] + [JSON.stringify(lockOrder(counts.values(), (c) => `${c.tool}\t${c.group_id}\t${c.klass}`))] ) } return gone.length diff --git a/app/services/prober.ts b/app/services/prober.ts index 7ceb250..30a9764 100644 --- a/app/services/prober.ts +++ b/app/services/prober.ts @@ -41,6 +41,8 @@ const TIERS: { kind: string; unreadable: boolean }[] = [ const FLUSH_MS = 2000 /** Between sweeps that discard the waiting releases of dropped signatures. */ const SWEEP_MS = 5 * 60_000 +/** A flush that deadlocks (against the batch store's pending upsert) keeps its outcomes for the next, this often. */ +const DEADLOCK_RETRIES = 5 /** * JSON for a jsonb parameter, without NUL characters: jsonb can't hold \u0000 (a name read from an article had one), @@ -82,6 +84,8 @@ export class Prober { private refilling: Promise | null = null private stopped = false private stats = Prober.freshStats() + /** Flushes deadlocked in a row, their outcomes kept. */ + private deadlocks = 0 /** 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>() private running = 0 @@ -346,15 +350,34 @@ export class Prober { } } - /** The outcomes so far: candidates' status, inspections, and the releases' passworded and inner name. */ + /** + * The outcomes so far: candidates' status, inspections, and the releases' passworded and inner name. A deadlock + * (the batch store's pending upsert locks the same rows; LOG.md (134)) keeps the outcomes for the next flush + * instead of losing the probes. + */ private async flush() { const taken = this.outcomes.splice(0) if (taken.length === 0) return + try { + await this.store(taken) + } catch (error: any) { + if (error?.code === '40P01' && this.deadlocks < DEADLOCK_RETRIES) { + this.deadlocks++ + this.outcomes.unshift(...taken) + logger.warn(`Probing: storing results deadlocked, kept for the next flush (${this.deadlocks}).`) + return + } + throw error + } + this.deadlocks = 0 for (const o of taken) if (o.status !== 'waiting') this.stats[o.status]++ + this.stats.pending += taken.filter((o) => o.candidate.pending && o.status !== 'waiting').length + } + + private async store(taken: Outcome[]) { // The ladder's outcomes apart from the catalog's: the next target, a name for the namer, or the end of the climb. const pending = taken.filter((o) => o.candidate.pending) const outcomes = taken.filter((o) => !o.candidate.pending) - this.stats.pending += pending.filter((o) => o.status !== 'waiting').length await db.transaction(async (trx) => { if (outcomes.length > 0) { await trx.rawQuery( diff --git a/database/migrations/1770000007200_pending_signature_index.ts b/database/migrations/1770000007200_pending_signature_index.ts new file mode 100644 index 0000000..59b38a2 --- /dev/null +++ b/database/migrations/1770000007200_pending_signature_index.ts @@ -0,0 +1,20 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * The discard sweep of a dropped signature (tool, group, poster) read the whole pending_releases table per + * signature: the planner took a parallel seq scan over the poster test, four minutes on the HDD that starved the + * lanes' claims (LOG.md (134)). The waiting rows by signature make it an index range. + */ +export default class extends BaseSchema { + static disableTransactions = true + + async up() { + this.schema.raw( + `CREATE INDEX CONCURRENTLY IF NOT EXISTS pending_releases_signature_idx ON pending_releases (group_id, tool, poster) WHERE status = 0` + ) + } + + async down() { + this.schema.raw(`DROP INDEX CONCURRENTLY IF EXISTS pending_releases_signature_idx`) + } +}