diff --git a/app/services/prober.ts b/app/services/prober.ts index 4d633e6..d097b44 100644 --- a/app/services/prober.ts +++ b/app/services/prober.ts @@ -166,15 +166,19 @@ export class Prober { const want = coordinatorConfig.probe.batch - this.queue.length if (want <= 0) return // The starting points chosen here: in the query Postgres drew random() once for all slices. - const [{ lo, hi }] = (await db.rawQuery(`SELECT min(id) AS lo, max(id) AS hi FROM probe_candidates`)).rows + // Over the claimable ones of this tier only: the rest of the queue isn't settled yet. + const [{ lo, hi }] = ( + await db.rawQuery(`SELECT min(id) AS lo, max(id) AS hi FROM probe_candidates WHERE status = 'waiting' AND ready AND kind = ? AND unreadable >= ?`, [kind, unreadable]) + ).rows + if (lo === null) continue const starts = Array.from({ length: SLICES }, () => Number(lo) + Math.floor(Math.random() * (Number(hi) - Number(lo) + 1))) const { rows } = await db.rawQuery( `UPDATE probe_candidates SET status = 'taken' WHERE id IN ( SELECT pc.id FROM unnest(?::bigint[]) AS o(start) CROSS JOIN LATERAL ( - SELECT pc.id FROM probe_candidates pc JOIN releases r ON r.id = pc.release_id - WHERE pc.status = 'waiting' AND pc.kind = ? AND r.settled_at IS NOT NULL AND (NOT ? OR NOT r.readable) AND pc.id >= o.start - ORDER BY pc.id LIMIT ? FOR UPDATE OF pc SKIP LOCKED) pc) + SELECT pc.id FROM probe_candidates pc + WHERE pc.status = 'waiting' AND pc.ready AND pc.kind = ? AND pc.unreadable >= ? AND pc.id >= o.start + ORDER BY pc.kind, pc.unreadable, pc.id LIMIT ? FOR UPDATE SKIP LOCKED) pc) RETURNING id, release_id, kind, message_id`, [starts, kind, unreadable, Math.ceil(want / SLICES)] ) diff --git a/app/services/settler.ts b/app/services/settler.ts index 0892aae..7d83548 100644 --- a/app/services/settler.ts +++ b/app/services/settler.ts @@ -21,14 +21,20 @@ 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(mins => ?) - AND (last_seen_at < now() - make_interval(hours => ?) OR coalesce(last_posted_at, posted_at) < now() - make_interval(hours => ?)) LIMIT ?)`, + // Their probe candidates become claimable (Prober.claim), the unreadable ones first. + const { rows } = await db.rawQuery( + `WITH settled AS ( + 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(mins => ?) + AND (last_seen_at < now() - make_interval(hours => ?) OR coalesce(last_posted_at, posted_at) < now() - make_interval(hours => ?)) LIMIT ?) + RETURNING id, readable), + marked AS (UPDATE probe_candidates pc SET ready = true, unreadable = NOT s.readable FROM settled s WHERE pc.release_id = s.id) + SELECT count(*)::int AS n FROM settled`, [QUIET_HISTORY_MINUTES, QUIET_HOURS, QUIET_HOURS, BATCH] ) - settled += rowCount ?? 0 - if ((rowCount ?? 0) < BATCH) return settled + const n = rows[0].n + settled += n + if (n < BATCH) return settled } } } diff --git a/database/migrations/1770000002500_probe_candidates_ready.ts b/database/migrations/1770000002500_probe_candidates_ready.ts new file mode 100644 index 0000000..9a8cf15 --- /dev/null +++ b/database/migrations/1770000002500_probe_candidates_ready.ts @@ -0,0 +1,30 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * Candidates the prober may claim (Prober.claim): `ready` once their release + * settled (the settler sets it), `unreadable` when that release had no + * readable name then. The claims joined releases for both and started at + * random points of the queue; in its newer half nothing had settled, so a + * slice scanned to the end (0.3 s a slice, 20 slices, 5 tiers: the prober + * stood still for minutes per claim; LOG.md 2026-10-12 (20)). With an index + * over the claimable rows only. + */ +export default class extends BaseSchema { + static disableTransactions = true + + async up() { + this.schema.raw(`ALTER TABLE probe_candidates ADD COLUMN IF NOT EXISTS ready boolean NOT NULL DEFAULT false`) + this.schema.raw(`ALTER TABLE probe_candidates ADD COLUMN IF NOT EXISTS unreadable boolean NOT NULL DEFAULT false`) + this.schema.raw( + `UPDATE probe_candidates pc SET ready = true, unreadable = NOT r.readable FROM releases r + WHERE r.id = pc.release_id AND r.settled_at IS NOT NULL AND NOT pc.ready` + ) + this.schema.raw(`CREATE INDEX CONCURRENTLY IF NOT EXISTS probe_candidates_ready_idx ON probe_candidates (kind, unreadable, id) WHERE status = 'waiting' AND ready`) + this.schema.raw(`DROP INDEX CONCURRENTLY IF EXISTS probe_candidates_kind_waiting_idx`) + } + + async down() { + this.schema.raw(`DROP INDEX CONCURRENTLY IF EXISTS probe_candidates_ready_idx`) + this.schema.raw(`ALTER TABLE probe_candidates DROP COLUMN IF EXISTS ready, DROP COLUMN IF EXISTS unreadable`) + } +} diff --git a/database/schema.ts b/database/schema.ts index 52d0fd3..f836854 100644 --- a/database/schema.ts +++ b/database/schema.ts @@ -169,7 +169,7 @@ export class ProbeCacheSchema extends BaseModel { } export class ProbeCandidateSchema extends BaseModel { - static $columns = ['createdAt', 'file', 'id', 'kind', 'messageId', 'releaseId', 'status'] as const + static $columns = ['createdAt', 'file', 'id', 'kind', 'messageId', 'ready', 'releaseId', 'status', 'unreadable'] as const $columns = ProbeCandidateSchema.$columns @column.dateTime({ autoCreate: true }) declare createdAt: DateTime @@ -182,9 +182,13 @@ export class ProbeCandidateSchema extends BaseModel { @column() declare messageId: string @column() + declare ready: boolean + @column() declare releaseId: bigint | number @column() declare status: string + @column() + declare unreadable: boolean } export class RangeSchema extends BaseModel { diff --git a/tests/functional/settler.spec.ts b/tests/functional/settler.spec.ts index 132a3d5..21611ef 100644 --- a/tests/functional/settler.spec.ts +++ b/tests/functional/settler.spec.ts @@ -24,7 +24,15 @@ test.group('Settler', (group) => { }) const minutesAgo = (m: number) => new Date(Date.now() - m * 60_000) await db.table('releases').multiInsert([row('old', minutesAgo(31), posted), row('fresh', minutesAgo(5), posted), row('tail', minutesAgo(31), minutesAgo(40))]) + const ids = Object.fromEntries((await db.from('releases').select('id', 'name')).map((r) => [r.name, r.id])) + await db.table('probe_candidates').multiInsert([ + { release_id: ids.old, kind: 'lone', message_id: '', file: 'a' }, + { release_id: ids.fresh, kind: 'lone', message_id: '', file: 'b' }, + ]) assert.equal(await Settler.run(), 1) + // The settled release's candidates are claimable now, as unreadable (readable defaults to false). + const ready = await db.from('probe_candidates').orderBy('message_id').select('ready', 'unreadable') + assert.deepEqual(ready.map((r) => [r.ready, r.unreadable]), [[true, true], [false, false]]) const settled = (await db.from('releases').whereNotNull('settled_at').select('name')).map((r) => r.name) assert.deepEqual(settled, ['old']) })