From 80c920f67834b7f7755c98b847bbf4f6a1455841 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Fri, 9 Oct 2026 02:33:47 +0200 Subject: [PATCH] Rebuild: settling quiet releases Hourly in coordinator:run: a release unchanged for 6 hours is settled, complete when every file its [i/N] counter announced was found, unknown without a counter (phase 2 reads par2 sets for those). Partial index on unsettled releases. Co-Authored-By: Claude Opus 5.5 --- app/services/settler.ts | 28 +++++++++++++++++++ commands/coordinator_run.ts | 7 +++-- .../1770000001000_releases_unsettled_index.ts | 12 ++++++++ tests/functional/settler.spec.ts | 19 +++++++++++++ 4 files changed, 64 insertions(+), 2 deletions(-) create mode 100644 app/services/settler.ts create mode 100644 database/migrations/1770000001000_releases_unsettled_index.ts create mode 100644 tests/functional/settler.spec.ts diff --git a/app/services/settler.ts b/app/services/settler.ts new file mode 100644 index 0000000..97c5294 --- /dev/null +++ b/app/services/settler.ts @@ -0,0 +1,28 @@ +import db from '@adonisjs/lucid/services/db' + +/** Releases unchanged for this long are settled (ranges are read in order; the merger adds within minutes). */ +const QUIET_HOURS = 6 +const BATCH = 50_000 + +/** + * Settles releases (REBUILD.md, part 8): once nothing has been added for + * QUIET_HOURS, its completeness is decided: complete when every file the + * [i/N] counter announced was found, unknown without a counter (phase 2 + * reads par2 sets for those). Phase 2 picks its work from settled releases. + */ +class Settler { + async run(): Promise { + let settled = 0 + while (true) { + const { rowCount } = await db.rawQuery( + `UPDATE releases SET settled_at = now(), complete = CASE WHEN file_total IS NULL THEN NULL ELSE files >= file_total END + WHERE id IN (SELECT id FROM releases WHERE settled_at IS NULL AND last_seen_at < now() - make_interval(hours => ?) LIMIT ?)`, + [QUIET_HOURS, BATCH] + ) + settled += rowCount ?? 0 + if ((rowCount ?? 0) < BATCH) return settled + } + } +} + +export default new Settler() diff --git a/commands/coordinator_run.ts b/commands/coordinator_run.ts index d46a9f1..21b46ab 100644 --- a/commands/coordinator_run.ts +++ b/commands/coordinator_run.ts @@ -4,6 +4,7 @@ import { setTimeout as sleep } from 'node:timers/promises' import { Prefetcher } from '#services/prefetcher' import Planner from '#services/planner' import FragmentMerger from '#services/fragment_merger' +import Settler from '#services/settler' import NntpPool from '#services/nntp_pool' const TICK_MS = 2000 @@ -13,8 +14,8 @@ const MERGE_BUDGET_MS = 30 * 1000 /** * The coordinator's background work (REBUILD.md, part 1): the prefetcher, - * merging fragments of finished ranges every minute, and planning ranges - * for new articles every hour. The API (leases, stream, + * merging fragments of finished ranges every minute, and every hour + * planning ranges for new articles and settling quiet releases. The API (leases, stream, * batches) runs in the web server. */ export default class CoordinatorRun extends BaseCommand { @@ -42,6 +43,8 @@ export default class CoordinatorRun extends BaseCommand { plannedAt = Date.now() const { groups, ranges } = await Planner.planAll() if (ranges > 0) this.logger.info(`Planned ${ranges} ranges over ${groups} groups.`) + const settled = await Settler.run() + if (settled > 0) this.logger.info(`Settled ${settled} releases.`) } if (Date.now() - mergedAt >= MERGE_EVERY_MS) { mergedAt = Date.now() diff --git a/database/migrations/1770000001000_releases_unsettled_index.ts b/database/migrations/1770000001000_releases_unsettled_index.ts new file mode 100644 index 0000000..e65a0d2 --- /dev/null +++ b/database/migrations/1770000001000_releases_unsettled_index.ts @@ -0,0 +1,12 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** Releases still to settle (REBUILD.md, part 8), by when they last changed. */ +export default class extends BaseSchema { + async up() { + this.schema.raw(`CREATE INDEX IF NOT EXISTS releases_unsettled_idx ON releases (last_seen_at) WHERE settled_at IS NULL`) + } + + async down() { + this.schema.raw(`DROP INDEX IF EXISTS releases_unsettled_idx`) + } +} diff --git a/tests/functional/settler.spec.ts b/tests/functional/settler.spec.ts new file mode 100644 index 0000000..1a4adc5 --- /dev/null +++ b/tests/functional/settler.spec.ts @@ -0,0 +1,19 @@ +import { test } from '@japa/runner' +import db from '@adonisjs/lucid/services/db' +import testUtils from '@adonisjs/core/services/test_utils' +import Settler from '#services/settler' + +test.group('Settler', (group) => { + group.each.setup(() => testUtils.db().withGlobalTransaction()) + + test('settles quiet releases: complete by the [i/N] counter, unknown without one', async ({ assert }) => { + const old = new Date(Date.now() - 7 * 3600 * 1000) + const row = (key: string, files: number, fileTotal: number | null, lastSeen: Date) => ({ + release_key: key.padEnd(40, '0'), name: key, source: 'subject', groups: '[]', files, file_total: fileTotal, last_seen_at: lastSeen, + }) + await db.table('releases').multiInsert([row('a', 10, 10, old), row('b', 8, 10, old), row('c', 3, null, old), row('d', 10, 10, new Date())]) + assert.equal(await Settler.run(), 3) + const byName = Object.fromEntries((await db.from('releases').select('name', 'complete', 'settled_at')).map((r) => [r.name, [r.complete, r.settled_at !== null]])) + assert.deepEqual(byName, { a: [true, true], b: [false, true], c: [null, true], d: [null, false] }) + }) +})