From 821de668d0c648a7a0e868f77cffedd54ab9fb4c Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Fri, 9 Oct 2026 02:49:45 +0200 Subject: [PATCH] Rebuild: the merger stops when a pass over the file keys decides nothing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Waiting files counted as work, so the merge loop re-read them for its whole 30 s budget every minute: Postgres used ~a core even without workers. Now ~4% at rest, and storing a boneless range costs Postgres ~0.7–1.9 s of CPU (it was 33 s with the loop running). Chunk payloads are stored EXTERNAL (zstd already; no pglz attempt). Co-Authored-By: Claude Opus 5.5 --- app/services/fragment_merger.ts | 7 +++++-- commands/coordinator_run.ts | 10 ++++++++-- .../migrations/1770000000500_create_release_chunks.ts | 2 ++ 3 files changed, 15 insertions(+), 4 deletions(-) diff --git a/app/services/fragment_merger.ts b/app/services/fragment_merger.ts index 16a9d1b..6ffc18d 100644 --- a/app/services/fragment_merger.ts +++ b/app/services/fragment_merger.ts @@ -12,6 +12,8 @@ export interface MergeResult { completed: number incomplete: number waiting: number + /** This round reached the end of the file keys (the next one starts over). */ + wrapped: boolean } /** File keys per round. */ @@ -29,14 +31,15 @@ class FragmentMerger { private cursor = '' async run(): Promise { - const result: MergeResult = { files: 0, completed: 0, incomplete: 0, waiting: 0 } + const result: MergeResult = { files: 0, completed: 0, incomplete: 0, waiting: 0, wrapped: false } const { rows: keys } = await db.rawQuery( `SELECT DISTINCT f.file_key FROM fragments f JOIN ranges r ON r.id = f.range_id WHERE r.status = 'done' AND f.file_key > ? ORDER BY f.file_key LIMIT ?`, [this.cursor, BATCH] ) // At the end of the keys: the next round starts over. - this.cursor = keys.length < BATCH ? '' : keys.at(-1).file_key + result.wrapped = keys.length < BATCH + this.cursor = result.wrapped ? '' : keys.at(-1).file_key if (keys.length === 0) return result const rules = new RuleSet(await RuleSets.current()) const rows = await db diff --git a/commands/coordinator_run.ts b/commands/coordinator_run.ts index 21b46ab..097e5a0 100644 --- a/commands/coordinator_run.ts +++ b/commands/coordinator_run.ts @@ -48,14 +48,20 @@ export default class CoordinatorRun extends BaseCommand { } if (Date.now() - mergedAt >= MERGE_EVERY_MS) { mergedAt = Date.now() - // Rounds until nothing more can be decided, or MERGE_BUDGET_MS. + // Rounds until a whole pass over the file keys decides nothing more, or MERGE_BUDGET_MS. + // (Going on while only waiting files were left kept Postgres busy all the time.) const totals = { completed: 0, incomplete: 0, waiting: 0 } + let decidedThisPass = 0 while (!stopping && Date.now() - mergedAt < MERGE_BUDGET_MS) { const merged = await FragmentMerger.run() totals.completed += merged.completed totals.incomplete += merged.incomplete totals.waiting = merged.waiting - if (merged.files === 0) break + decidedThisPass += merged.completed + merged.incomplete + if (merged.wrapped) { + if (decidedThisPass === 0) break + decidedThisPass = 0 + } } if (totals.completed || totals.incomplete) this.logger.info(`Fragments: ${totals.completed} files completed, ${totals.incomplete} incomplete, ${totals.waiting} waiting.`) } diff --git a/database/migrations/1770000000500_create_release_chunks.ts b/database/migrations/1770000000500_create_release_chunks.ts index afef892..1bfbcb5 100644 --- a/database/migrations/1770000000500_create_release_chunks.ts +++ b/database/migrations/1770000000500_create_release_chunks.ts @@ -24,6 +24,8 @@ export default class extends BaseSchema { created_at timestamptz NOT NULL DEFAULT now(), PRIMARY KEY (id, created_at) ) PARTITION BY RANGE (created_at)${where}`) + // The payload is zstd already: no pglz attempt on it. + this.schema.raw(`ALTER TABLE release_chunks ALTER COLUMN payload SET STORAGE EXTERNAL`) this.schema.raw(`CREATE INDEX release_chunks_release_idx ON release_chunks (release_id)`) this.schema.raw(`CREATE INDEX release_chunks_range_idx ON release_chunks (range_id)`) }