diff --git a/app/services/prober.ts b/app/services/prober.ts index 87add1d..d87759d 100644 --- a/app/services/prober.ts +++ b/app/services/prober.ts @@ -19,6 +19,8 @@ type Outcome = { candidate: Candidate; status: 'done' | 'missing' | 'error' | 'w /** Kinds in the order they're probed: a random name's real name first, then par2 sets, then archives. */ const KINDS = ['lone', 'par2', 'archive'] const FLUSH_MS = 2000 +/** Slices a claim takes from random points of the queue. */ +const SLICES = 20 const REPORT_MS = 60_000 /** @@ -114,18 +116,27 @@ export class Prober { return this.queue.shift() } - /** Waiting candidates of settled releases, by kind, claimed so another round doesn't take them. */ + /** + * Waiting candidates of settled releases, by kind, claimed so another round + * doesn't take them: SLICES slices from random points of the queue. In id + * order it went group by group, and some groups' articles take 0.6–3 s + * where most take 20–60 ms (a.b.zogg 2016: 7 articles/s on 6 connections). + */ private async claim() { for (const kind of KINDS) { const want = coordinatorConfig.probe.batch - this.queue.length if (want <= 0) return const { rows } = await db.rawQuery( `UPDATE probe_candidates SET status = 'taken' WHERE id IN ( - 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 - ORDER BY pc.id LIMIT ? FOR UPDATE OF pc SKIP LOCKED) + SELECT pc.id FROM (SELECT min(id) AS lo, max(id) AS hi FROM probe_candidates) b + CROSS JOIN LATERAL generate_series(1, ?) AS s + 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 pc.id >= b.lo + floor(random() * (b.hi - b.lo + 1))::bigint + s * 0 + ORDER BY pc.id LIMIT ? FOR UPDATE OF pc SKIP LOCKED) pc) RETURNING id, release_id, kind, message_id`, - [kind, want] + [SLICES, kind, Math.ceil(want / SLICES)] ) for (const r of rows) this.queue.push({ id: Number(r.id), releaseId: Number(r.release_id), kind: r.kind, messageId: r.message_id }) } diff --git a/tools/body_timing.ts b/tools/body_timing.ts new file mode 100644 index 0000000..98c10c5 --- /dev/null +++ b/tools/body_timing.ts @@ -0,0 +1,37 @@ +/** + * BODY timing by article age: candidates' first articles, oldest and newest + * posted, fetched one at a time on one connection (status line, then the + * rest), with the release's posting date. + * + * node --env-file=.env --import=@poppinss/ts-exec tools/body_timing.ts [n] + */ +// @ts-ignore pg has no types here +import pg from 'pg' +import { XoverClient } from '#lib/nntp/index' + +const n = Number(process.argv[2] ?? 8) +const env = process.env +const db = new pg.Client({ host: env.DB_HOST, port: Number(env.DB_PORT ?? 5432), user: env.DB_USER, password: env.DB_PASSWORD, database: env.DB_DATABASE }) +await db.connect() +const pick = async (order: string) => + (await db.query(`SELECT pc.message_id, r.posted_at FROM probe_candidates pc JOIN releases r ON r.id = pc.release_id WHERE pc.kind = 'lone' AND r.posted_at IS NOT NULL ORDER BY r.posted_at ${order} LIMIT $1`, [n])).rows +const client = await XoverClient.connect({ host: env.NNTP_HOST!, port: Number(env.NNTP_PORT ?? 563), user: env.NNTP_USER, password: env.NNTP_PASSWORD, secure: (env.NNTP_SECURE ?? 'true') !== 'false', compress: false, timeoutMs: 120_000 }) +for (const [label, rows] of [['oldest', await pick('ASC')], ['newest', await pick('DESC')]] as const) { + let total = 0 + for (const r of rows) { + const at = performance.now() + let size = 0 + try { + size = (await client.body(r.message_id)).length + } catch (e: any) { + size = -1 + console.log(` ${e.message}`) + } + const ms = performance.now() - at + total += ms + console.log(`${label} ${new Date(r.posted_at).toISOString().slice(0, 10)} ${(ms / 1000).toFixed(2)} s ${(size / 1000).toFixed(0)} kB`) + } + console.log(`${label}: ${(total / rows.length / 1000).toFixed(2)} s an article`) +} +client.close() +await db.end()