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 <noreply@anthropic.com>
This commit is contained in:
1 parent
fde1e6cc90
commit
80c920f678
4 files changed
+64
-2
No files matched your search
@@ -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<number> {
|
||||||
|
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()
|
||||||
@@ -4,6 +4,7 @@ import { setTimeout as sleep } from 'node:timers/promises'
|
|||||||
import { Prefetcher } from '#services/prefetcher'
|
import { Prefetcher } from '#services/prefetcher'
|
||||||
import Planner from '#services/planner'
|
import Planner from '#services/planner'
|
||||||
import FragmentMerger from '#services/fragment_merger'
|
import FragmentMerger from '#services/fragment_merger'
|
||||||
|
import Settler from '#services/settler'
|
||||||
import NntpPool from '#services/nntp_pool'
|
import NntpPool from '#services/nntp_pool'
|
||||||
|
|
||||||
const TICK_MS = 2000
|
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,
|
* The coordinator's background work (REBUILD.md, part 1): the prefetcher,
|
||||||
* merging fragments of finished ranges every minute, and planning ranges
|
* merging fragments of finished ranges every minute, and every hour
|
||||||
* for new articles every hour. The API (leases, stream,
|
* planning ranges for new articles and settling quiet releases. The API (leases, stream,
|
||||||
* batches) runs in the web server.
|
* batches) runs in the web server.
|
||||||
*/
|
*/
|
||||||
export default class CoordinatorRun extends BaseCommand {
|
export default class CoordinatorRun extends BaseCommand {
|
||||||
@@ -42,6 +43,8 @@ export default class CoordinatorRun extends BaseCommand {
|
|||||||
plannedAt = Date.now()
|
plannedAt = Date.now()
|
||||||
const { groups, ranges } = await Planner.planAll()
|
const { groups, ranges } = await Planner.planAll()
|
||||||
if (ranges > 0) this.logger.info(`Planned ${ranges} ranges over ${groups} groups.`)
|
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) {
|
if (Date.now() - mergedAt >= MERGE_EVERY_MS) {
|
||||||
mergedAt = Date.now()
|
mergedAt = Date.now()
|
||||||
|
|||||||
@@ -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`)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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] })
|
||||||
|
})
|
||||||
|
})
|
||||||
Reference in new issue
Block a user