Prober: claim only candidates of settled releases, without the join
The settler marks a settled release's candidates ready (and unreadable when it had no readable name). Claims took the candidates by joining releases from random points of the queue; in its newer half nothing had settled, so a slice scanned to the end (0.3 s each, 20 slices and 5 tiers a claim) and the prober stood still: 2.4M waiting, 0 probed a minute. Now an index over the claimable rows only, the slices drawn within them. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
1866e38c25
commit
cdc1cdc0cc
5 files changed
+63
-11
No files matched your search
@@ -166,15 +166,19 @@ export class Prober {
|
|||||||
const want = coordinatorConfig.probe.batch - this.queue.length
|
const want = coordinatorConfig.probe.batch - this.queue.length
|
||||||
if (want <= 0) return
|
if (want <= 0) return
|
||||||
// The starting points chosen here: in the query Postgres drew random() once for all slices.
|
// 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 starts = Array.from({ length: SLICES }, () => Number(lo) + Math.floor(Math.random() * (Number(hi) - Number(lo) + 1)))
|
||||||
const { rows } = await db.rawQuery(
|
const { rows } = await db.rawQuery(
|
||||||
`UPDATE probe_candidates SET status = 'taken' WHERE id IN (
|
`UPDATE probe_candidates SET status = 'taken' WHERE id IN (
|
||||||
SELECT pc.id FROM unnest(?::bigint[]) AS o(start)
|
SELECT pc.id FROM unnest(?::bigint[]) AS o(start)
|
||||||
CROSS JOIN LATERAL (
|
CROSS JOIN LATERAL (
|
||||||
SELECT pc.id FROM probe_candidates pc JOIN releases r ON r.id = pc.release_id
|
SELECT pc.id FROM probe_candidates pc
|
||||||
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
|
WHERE pc.status = 'waiting' AND pc.ready AND pc.kind = ? AND pc.unreadable >= ? AND pc.id >= o.start
|
||||||
ORDER BY pc.id LIMIT ? FOR UPDATE OF pc SKIP LOCKED) pc)
|
ORDER BY pc.kind, pc.unreadable, pc.id LIMIT ? FOR UPDATE SKIP LOCKED) pc)
|
||||||
RETURNING id, release_id, kind, message_id`,
|
RETURNING id, release_id, kind, message_id`,
|
||||||
[starts, kind, unreadable, Math.ceil(want / SLICES)]
|
[starts, kind, unreadable, Math.ceil(want / SLICES)]
|
||||||
)
|
)
|
||||||
|
|||||||
+12
-6
@@ -21,14 +21,20 @@ class Settler {
|
|||||||
async run(): Promise<number> {
|
async run(): Promise<number> {
|
||||||
let settled = 0
|
let settled = 0
|
||||||
while (true) {
|
while (true) {
|
||||||
const { rowCount } = await db.rawQuery(
|
// Their probe candidates become claimable (Prober.claim), the unreadable ones first.
|
||||||
`UPDATE releases SET settled_at = now(), complete = CASE WHEN file_total IS NULL THEN NULL ELSE files >= file_total END
|
const { rows } = await db.rawQuery(
|
||||||
WHERE id IN (SELECT id FROM releases WHERE settled_at IS NULL AND last_seen_at < now() - make_interval(mins => ?)
|
`WITH settled AS (
|
||||||
AND (last_seen_at < now() - make_interval(hours => ?) OR coalesce(last_posted_at, posted_at) < now() - make_interval(hours => ?)) LIMIT ?)`,
|
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]
|
[QUIET_HISTORY_MINUTES, QUIET_HOURS, QUIET_HOURS, BATCH]
|
||||||
)
|
)
|
||||||
settled += rowCount ?? 0
|
const n = rows[0].n
|
||||||
if ((rowCount ?? 0) < BATCH) return settled
|
settled += n
|
||||||
|
if (n < BATCH) return settled
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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`)
|
||||||
|
}
|
||||||
|
}
|
||||||
+5
-1
@@ -169,7 +169,7 @@ export class ProbeCacheSchema extends BaseModel {
|
|||||||
}
|
}
|
||||||
|
|
||||||
export class ProbeCandidateSchema 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
|
$columns = ProbeCandidateSchema.$columns
|
||||||
@column.dateTime({ autoCreate: true })
|
@column.dateTime({ autoCreate: true })
|
||||||
declare createdAt: DateTime
|
declare createdAt: DateTime
|
||||||
@@ -182,9 +182,13 @@ export class ProbeCandidateSchema extends BaseModel {
|
|||||||
@column()
|
@column()
|
||||||
declare messageId: string
|
declare messageId: string
|
||||||
@column()
|
@column()
|
||||||
|
declare ready: boolean
|
||||||
|
@column()
|
||||||
declare releaseId: bigint | number
|
declare releaseId: bigint | number
|
||||||
@column()
|
@column()
|
||||||
declare status: string
|
declare status: string
|
||||||
|
@column()
|
||||||
|
declare unreadable: boolean
|
||||||
}
|
}
|
||||||
|
|
||||||
export class RangeSchema extends BaseModel {
|
export class RangeSchema extends BaseModel {
|
||||||
|
|||||||
@@ -24,7 +24,15 @@ test.group('Settler', (group) => {
|
|||||||
})
|
})
|
||||||
const minutesAgo = (m: number) => new Date(Date.now() - m * 60_000)
|
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))])
|
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: '<a@x>', file: 'a' },
|
||||||
|
{ release_id: ids.fresh, kind: 'lone', message_id: '<b@x>', file: 'b' },
|
||||||
|
])
|
||||||
assert.equal(await Settler.run(), 1)
|
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)
|
const settled = (await db.from('releases').whereNotNull('settled_at').select('name')).map((r) => r.name)
|
||||||
assert.deepEqual(settled, ['old'])
|
assert.deepEqual(settled, ['old'])
|
||||||
})
|
})
|
||||||
|
|||||||
Reference in new issue
Block a user