diff --git a/REBUILD.md b/REBUILD.md index 48eb408..be27605 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -464,6 +464,13 @@ releases, a third of them one poster's image dump (27.7 M rows, LOG.md merger: the carry does the joining. The settler goes: `settled_at` is the finish. The **tail** needs no path of its own: its lease ends leave such provisional releases, the next lease's continuation finishes them. + One exception (2026-10-27): a provisional release that is complete + (every file its counter announced found) and quiet for six hours is + settled by the coordinator (`Settler.settleComplete`, every five + minutes), as the finish would: no counted file can still come, and the + continuation may be hours away or never (the first run's range leases + left 109k provisional rows at their blocks' end edges, 606 of them + complete, shown as still collecting; the user, release 10697). 3. **Albums.** Files of one poster in one owner group whose subjects differ only by a number sequence, or share a directory-like prefix, posted diff --git a/app/services/settler.ts b/app/services/settler.ts new file mode 100644 index 0000000..0268c3a --- /dev/null +++ b/app/services/settler.ts @@ -0,0 +1,36 @@ +import db from '@adonisjs/lucid/services/db' + +/** Hours a complete provisional release stays open for its window's continuation before it is settled here. */ +export const SETTLE_COMPLETE_HOURS = 6 + +/** + * Provisional releases (REBUILD.md, "Voyager 3: window indexing", 2): a release the window hadn't closed on at a + * lease's end is written with `settled_at` null, for the window's continuation to finish. A continuation may + * never come (the range-lease path of the first run left 109k such rows, LOG.md (118); a group's next lease is + * hours away), and the catalog shows the release as still collecting. A complete one — every file its counter + * announced found — can't gain a counted file any more, so once it has been quiet for SETTLE_COMPLETE_HOURS it is + * settled here, as the finish would: `settled_at` set (the judge's trigger fires on it), its waiting probe + * candidates made claimable. Incomplete ones keep waiting: the continuation adds their files. + */ +class Settler { + async settleComplete(quietHours = SETTLE_COMPLETE_HOURS): Promise { + return db.transaction(async (trx) => { + const { rows } = await trx.rawQuery( + `UPDATE releases SET settled_at = now() + WHERE settled_at IS NULL AND complete + AND first_seen_at < now() - make_interval(hours => ?) AND last_posted_at < now() - make_interval(hours => ?) + RETURNING id`, + [quietHours, quietHours] + ) + if (rows.length > 0) + await trx.rawQuery( + `UPDATE probe_candidates pc SET ready = true, unreadable = NOT r.readable FROM releases r + WHERE pc.release_id = ANY(?::bigint[]) AND r.id = pc.release_id AND pc.status = 'waiting' AND NOT pc.ready`, + [rows.map((r: { id: string }) => Number(r.id))] + ) + return rows.length + }) + } +} + +export default new Settler() diff --git a/commands/coordinator_run.ts b/commands/coordinator_run.ts index 4a4eea9..63b81cd 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 Leases from '#services/leases' +import Settler from '#services/settler' import NntpPool from '#services/nntp_pool' import { NntpShare } from '#services/nntp_shares' import coordinatorConfig from '#config/coordinator' @@ -101,6 +102,13 @@ export default class CoordinatorRun extends BaseCommand { } catch (error: any) { this.logger.error(`Finishing abandoned leases failed: ${error.message}`) } + try { + const settled = await Settler.settleComplete() + if (settled > 0) + this.logger.info(`Settled ${settled} complete provisional releases quiet for hours.`) + } catch (error: any) { + this.logger.error(`Settling complete releases failed: ${error.message}`) + } for (let waited = 0; waited < ABANDONED_EVERY_MS && !stopping; waited += TICK_MS) await sleep(TICK_MS) } diff --git a/tests/functional/settler.spec.ts b/tests/functional/settler.spec.ts new file mode 100644 index 0000000..bf9bad0 --- /dev/null +++ b/tests/functional/settler.spec.ts @@ -0,0 +1,112 @@ +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' +import { insertRelease } from '#tests/functional/helpers/releases' + +const hoursAgo = (h: number) => new Date(Date.now() - h * 3600_000) + +test.group('Settler', (group) => { + group.each.setup(() => testUtils.db().withGlobalTransaction()) + + test('a complete provisional release quiet for six hours is settled, its waiting candidates made claimable', async ({ + assert, + }) => { + const base = { + groups: ['alt.binaries.test'], + poster: 'p@x', + first_article: 1, + posted_at: hoursAgo(30), + last_posted_at: hoursAgo(30), + } + // Complete and old: settled. + const old = await insertRelease({ + ...base, + name: 'old', + files: 3, + file_total: 3, + complete: true, + readable: true, + first_seen_at: hoursAgo(7), + }) + // Complete but seen an hour ago: its window may still be carried. + const recent = await insertRelease({ + ...base, + name: 'recent', + files: 3, + file_total: 3, + complete: true, + first_seen_at: hoursAgo(1), + }) + // Complete, old, but its last part posted an hour ago: still arriving. + const posting = await insertRelease({ + ...base, + name: 'posting', + files: 3, + file_total: 3, + complete: true, + first_seen_at: hoursAgo(7), + last_posted_at: hoursAgo(1), + }) + // Incomplete, and without a counter: the continuation's. + const partial = await insertRelease({ + ...base, + name: 'partial', + files: 2, + file_total: 3, + complete: false, + first_seen_at: hoursAgo(7), + }) + const uncounted = await insertRelease({ + ...base, + name: 'uncounted', + files: 2, + file_total: null, + complete: null, + first_seen_at: hoursAgo(7), + }) + // Settled already: left alone. + const settled = await insertRelease({ + ...base, + name: 'settled', + files: 3, + file_total: 3, + complete: true, + first_seen_at: hoursAgo(7), + settled_at: hoursAgo(6), + }) + await db.table('probe_candidates').multiInsert([ + { release_id: old, kind: 'par2', message_id: '', file: 'x.par2', status: 'waiting' }, + { + release_id: partial, + kind: 'par2', + message_id: '', + file: 'x.par2', + status: 'waiting', + }, + ]) + + assert.equal(await Settler.settleComplete(), 1) + const rows = await db + .from('releases') + .whereIn('id', [old, recent, posting, partial, uncounted, settled]) + .select('id', 'settled_at') + const settledIds = rows + .filter((r) => r.settled_at !== null) + .map((r) => Number(r.id)) + .sort() + assert.deepEqual(settledIds, [old, settled].sort()) + const candidates = await db + .from('probe_candidates') + .whereIn('release_id', [old, partial]) + .select('release_id', 'ready', 'unreadable') + assert.deepEqual( + candidates.map((c) => [Number(c.release_id), c.ready, c.unreadable]).sort(), + [ + [old, true, false], + [partial, false, false], + ].sort() + ) + assert.equal(await Settler.settleComplete(), 0) + }) +})