Release completeness check without reading every binary; SEED_RANGES
release_complete() matched each par2-listed file through a correlated subquery that scanned all binaries per listed file, run twice per inserted release_files row: one insert of a few thousand rows ran for 20+ minutes. It now reads the release's files once by release id, and release_files refreshes once per INSERT statement. seed:node's SEED_CONNECTIONS/--connections is now SEED_RANGES/--ranges. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
2dbca9b152
commit
4d17c5a8e8
5 files changed
+135
-11
No files matched your search
@@ -33,7 +33,7 @@ interface Lease {
|
||||
* connection(s) (D37), and posts the results back in gzip-compressed
|
||||
* batches. Outgoing requests only: no ports, database or Redis needed.
|
||||
*
|
||||
* SEED_URL=https://seed.example SEED_TOKEN=sn_… NNTP_HOST=… node ace seed:node --connections 4
|
||||
* SEED_URL=https://seed.example SEED_TOKEN=sn_… node ace seed:node --ranges 4 --inspectors 2
|
||||
*/
|
||||
export default class SeedNode extends BaseCommand {
|
||||
public static commandName = 'seed:node'
|
||||
@@ -53,10 +53,10 @@ export default class SeedNode extends BaseCommand {
|
||||
|
||||
@flags.number({
|
||||
description:
|
||||
'Ranges worked on at once, one NNTP connection each; 0 for an inspection-only node (default: SEED_CONNECTIONS or 1)',
|
||||
default: process.env.SEED_CONNECTIONS ? Number(process.env.SEED_CONNECTIONS) : 1,
|
||||
'Ranges seeded at once (with SEED_FETCH=direct, one NNTP connection each); 0 for an inspection-only node (default: SEED_RANGES or 1)',
|
||||
default: process.env.SEED_RANGES ? Number(process.env.SEED_RANGES) : 1,
|
||||
})
|
||||
declare connections: number
|
||||
declare ranges: number
|
||||
|
||||
@flags.number({
|
||||
description:
|
||||
@@ -100,11 +100,11 @@ export default class SeedNode extends BaseCommand {
|
||||
|
||||
const api = new SeedApi(base, token, (message) => this.logger.warning(message))
|
||||
this.logger.info(
|
||||
`Seeding node for ${base}: ${this.connections} range(s) at once, ${this.fetch === 'direct' ? 'own NNTP connections' : 'headers through the seed API'}` +
|
||||
`Seeding node for ${base}: ${this.ranges} range(s) at once, ${this.fetch === 'direct' ? 'own NNTP connections' : 'headers through the seed API'}` +
|
||||
(this.inspectors > 0 ? `; ${this.inspectors} inspection(s) at once, articles through the seed API.` : '.')
|
||||
)
|
||||
await Promise.all([
|
||||
...Array.from({ length: this.connections }, (_, i) => this.work(api, i + 1)),
|
||||
...Array.from({ length: this.ranges }, (_, i) => this.work(api, i + 1)),
|
||||
...Array.from({ length: this.inspectors }, (_, i) => this.inspect(api, i + 1)),
|
||||
])
|
||||
this.logger.info('Seeding node stopped.')
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
import { BaseSchema } from '@adonisjs/lucid/schema'
|
||||
|
||||
/**
|
||||
* Release completeness (D24) without cubic cost (LOG.md 2026-10-06 (4)). A
|
||||
* par2 set listing n files was inserted into release_files row by row, each
|
||||
* row's trigger ran release_complete() twice, and that matched every listed
|
||||
* file against all binaries in the database: n × 2 × n × (all binaries). One
|
||||
* insert of a few thousand rows ran for over 20 minutes at full CPU.
|
||||
*
|
||||
* - release_complete(): the release's files are read once, by release id,
|
||||
* and each listed file is matched against those by size or by name. The
|
||||
* correlated subquery read every binary in the database per listed file
|
||||
* (binaries_size_index), filtering by release only afterwards.
|
||||
* - refresh_release_complete(): computes it once.
|
||||
* - release_files: one refresh per release per INSERT statement.
|
||||
*/
|
||||
export default class extends BaseSchema {
|
||||
async up() {
|
||||
await this.db.rawQuery(`
|
||||
CREATE OR REPLACE FUNCTION release_complete(rid bigint) RETURNS boolean LANGUAGE sql STABLE AS $$
|
||||
WITH r AS (SELECT file_total, file_count FROM releases WHERE id = rid),
|
||||
-- The release's files, once, by index (read only when there's a par2 listing to check).
|
||||
have AS MATERIALIZED (
|
||||
SELECT b.size, f.real_name FROM files f JOIN binaries b ON b.file_id = f.id WHERE f.release_id = rid)
|
||||
SELECT coalesce((
|
||||
SELECT CASE
|
||||
WHEN r.file_total IS NOT NULL THEN r.file_count >= r.file_total
|
||||
WHEN EXISTS (SELECT 1 FROM release_files rf WHERE rf.release_id = rid) THEN NOT EXISTS (
|
||||
SELECT 1 FROM release_files rf
|
||||
WHERE rf.release_id = rid AND rf.size >= 65536
|
||||
AND NOT EXISTS (SELECT 1 FROM have h WHERE h.size = rf.size)
|
||||
AND NOT EXISTS (SELECT 1 FROM have h WHERE h.real_name = rf.name))
|
||||
ELSE false END
|
||||
FROM r), false)
|
||||
$$`)
|
||||
await this.db.rawQuery(`
|
||||
CREATE OR REPLACE FUNCTION refresh_release_complete(rid bigint) RETURNS void LANGUAGE sql AS $$
|
||||
UPDATE releases r SET complete = c.complete
|
||||
FROM (SELECT release_complete(rid) AS complete) c
|
||||
WHERE r.id = rid AND r.complete IS DISTINCT FROM c.complete
|
||||
$$`)
|
||||
await this.db.rawQuery(`
|
||||
CREATE OR REPLACE FUNCTION release_files_release_complete_statement() RETURNS trigger LANGUAGE plpgsql AS $$
|
||||
BEGIN
|
||||
PERFORM refresh_release_complete(n.release_id) FROM (SELECT DISTINCT release_id FROM new_rows) n;
|
||||
RETURN NULL;
|
||||
END $$`)
|
||||
// Swapping the trigger locks release_files: inserts still running the old cubic check
|
||||
// (minutes) are cancelled rather than waited for; their jobs are retried, and run fast.
|
||||
await this.db.rawQuery(`SET LOCAL lock_timeout = '2min'`)
|
||||
await this.db.rawQuery(`
|
||||
SELECT pg_cancel_backend(pid) FROM pg_stat_activity
|
||||
WHERE pid <> pg_backend_pid() AND state = 'active' AND query ILIKE 'insert into "release_files"%'
|
||||
AND now() - query_start > interval '30 seconds'`)
|
||||
await this.db.rawQuery(`DROP TRIGGER IF EXISTS release_files_release_complete ON release_files`)
|
||||
await this.db.rawQuery(`
|
||||
CREATE TRIGGER release_files_release_complete AFTER INSERT ON release_files
|
||||
REFERENCING NEW TABLE AS new_rows FOR EACH STATEMENT
|
||||
EXECUTE FUNCTION release_files_release_complete_statement()`)
|
||||
}
|
||||
|
||||
async down() {
|
||||
await this.db.rawQuery(`DROP TRIGGER IF EXISTS release_files_release_complete ON release_files`)
|
||||
await this.db.rawQuery(`
|
||||
CREATE TRIGGER release_files_release_complete AFTER INSERT ON release_files
|
||||
FOR EACH ROW EXECUTE FUNCTION release_files_release_complete()`)
|
||||
await this.db.rawQuery(`DROP FUNCTION IF EXISTS release_files_release_complete_statement()`)
|
||||
}
|
||||
}
|
||||
+4
-4
@@ -26,7 +26,7 @@ the NNTP lines in `node.env`).
|
||||
- Outgoing HTTPS to `seed.exp.sforum.nl` (the seed API) and
|
||||
`gitea.exp.sforum.nl` (to build the image), and outgoing NNTP to the news
|
||||
server (port 563 for TLS).
|
||||
- Per connection: roughly 0.4 CPU core and 1–3 MB/s download (headers come
|
||||
- Per range seeded at once: roughly 0.4 CPU core and 1–3 MB/s download (headers come
|
||||
gzip-compressed); a few hundred MB of RAM per node. Disk: about 1.5 GB for
|
||||
the image. Per inspector: up to a core while decoding, and a few MB/s
|
||||
download (whole articles, ~700 KB each).
|
||||
@@ -62,7 +62,7 @@ for the same name replaces the token; `--disable` revokes it.
|
||||
|---|---|
|
||||
| `SEED_URL` | `https://seed.exp.sforum.nl` (leave as is) |
|
||||
| `SEED_TOKEN` | the token from step 1 |
|
||||
| `SEED_CONNECTIONS` | ranges seeded at once. Start with 2–4; `0` for no seeding. |
|
||||
| `SEED_RANGES` | ranges seeded at once. Start with 2–4; `0` for no seeding. |
|
||||
| `SEED_INSPECTORS` | inspections and archive walks at once (default `0`). Try 2–4 on a machine with spare cores. |
|
||||
| `SEED_FETCH=direct` and `NNTP_*` | optional: seed with this node's own NNTP account (see the commented lines). `NNTP_TLS_SERVERNAME` only through a tunnel: the news server's real hostname, for the TLS certificate check. |
|
||||
|
||||
@@ -76,7 +76,7 @@ needs are filled in for a node automatically.
|
||||
|
||||
The first build takes a few minutes. A working node logs:
|
||||
|
||||
[ info ] Seeding node for https://seed.exp.sforum.nl with 2 connection(s).
|
||||
[ info ] Seeding node for https://seed.exp.sforum.nl: 2 range(s) at once, headers through the seed API; 2 inspection(s) at once, articles through the seed API.
|
||||
[ info ] [1] alt.binaries.ath 49834000000–49835999999
|
||||
[ info ] [1] done: 2,000,000 headers in 237 s (8,450/s)
|
||||
|
||||
@@ -111,7 +111,7 @@ articles reached, and per node its leases and headers.
|
||||
|---|---|
|
||||
| `The seed API rejected this node token` | wrong or revoked `SEED_TOKEN`; the node stops |
|
||||
| `Seed API …: 404 …; retrying` | the seed API is restarting (a deploy on homelab); the node retries |
|
||||
| `NNTP login failed: …` | the node didn't take a range and tries again in a minute (5 minutes for the IP-address limit); `481`/`502`: wrong account, too many connections (lower `SEED_CONNECTIONS`) or too many IP addresses |
|
||||
| `NNTP login failed: …` | the node didn't take a range and tries again in a minute (5 minutes for the IP-address limit); `481`/`502`: wrong account, too many connections (lower `SEED_RANGES`) or too many IP addresses |
|
||||
| `Hostname/IP does not match certificate's altnames: Host: localhost` | TLS through a tunnel: set `NNTP_TLS_SERVERNAME` to the news server's hostname |
|
||||
| `Range … failed: …` | the range is given back and retried later, possibly by another node |
|
||||
| `Range … passed to another node; dropping it.` | this node was too slow to report (15 minutes); harmless |
|
||||
|
||||
@@ -3,7 +3,7 @@ SEED_URL=https://seed.exp.sforum.nl
|
||||
SEED_TOKEN=sn_replace_me
|
||||
|
||||
# Ranges seeded at once (about 0.4 CPU core each). 0: no seeding.
|
||||
SEED_CONNECTIONS=4
|
||||
SEED_RANGES=4
|
||||
# Content inspections and archive walks at once (CPU for decoding and parsing; articles come through the seed API).
|
||||
SEED_INSPECTORS=0
|
||||
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
import { test } from '@japa/runner'
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
import testUtils from '@adonisjs/core/services/test_utils'
|
||||
|
||||
/** A release with `n` files (each with a binary of size 100000 + i), named f<i>.bin. */
|
||||
async function release(n: number) {
|
||||
const now = new Date()
|
||||
const [row] = await db
|
||||
.table('releases')
|
||||
.insert({ release_key: `complete-${n}`.padEnd(40, '0'), name: `Complete.Test.${n}`, source: 'stem', groups: '[]', first_seen_at: now, last_seen_at: now })
|
||||
.returning('id')
|
||||
const releaseId = Number(row.id)
|
||||
await db.rawQuery(
|
||||
`WITH f AS (
|
||||
INSERT INTO files (filename, poster, date, parts, groups, segments, release_id, file_key, created_at, updated_at)
|
||||
SELECT 'f' || i || '.bin', 'x', 0, 1, '[]', '\\x'::bytea, ?::bigint, 'complete-' || ?::int || '-' || i, now(), now()
|
||||
FROM generate_series(1, ?::int) i
|
||||
RETURNING id, filename)
|
||||
INSERT INTO binaries (file_key, poster, subject, name, total_parts, groups, first_seen_at, last_seen_at, size, file_id, release_id)
|
||||
SELECT 'complete-b-' || ?::int || '-' || f.id, 'x', 'x', f.filename, 1, '[]', now(), now(), 100000 + substr(f.filename, 2, length(f.filename) - 5)::int, f.id, ?::bigint
|
||||
FROM f`,
|
||||
[releaseId, n, n, n, releaseId]
|
||||
)
|
||||
return releaseId
|
||||
}
|
||||
|
||||
/** The par2 listing of `count` files f1..f<count> (sizes as in release()), in one INSERT, as Par2Service.attach does. */
|
||||
async function listing(releaseId: number, count: number, offset = 0) {
|
||||
await db.rawQuery(
|
||||
`INSERT INTO release_files (release_id, par2_file_id, name, size, md5_16k, set_id)
|
||||
SELECT ?::bigint, md5(i::text), 'f' || i || '.bin', 100000 + i, md5(i::text), 'set' FROM generate_series(1 + ?::int, ?::int + ?::int) i`,
|
||||
[releaseId, offset, count, offset]
|
||||
)
|
||||
return Boolean((await db.from('releases').where('id', releaseId).first()).complete)
|
||||
}
|
||||
|
||||
test.group('release completeness', (group) => {
|
||||
group.each.setup(() => testUtils.db().withGlobalTransaction())
|
||||
|
||||
test('complete when every listed file of 64 KiB or more is there, by size or name', async ({ assert }) => {
|
||||
const id = await release(3)
|
||||
assert.isTrue(await listing(id, 3))
|
||||
const other = await release(2)
|
||||
// f3 is listed but missing.
|
||||
assert.isFalse(await listing(other, 3))
|
||||
})
|
||||
|
||||
test('a par2 listing of thousands of files is checked in one go', async ({ assert }) => {
|
||||
const id = await release(3000)
|
||||
const started = Date.now()
|
||||
assert.isTrue(await listing(id, 3000))
|
||||
// Was cubic: minutes for a few thousand rows.
|
||||
assert.isBelow(Date.now() - started, 5000)
|
||||
}).timeout(60_000)
|
||||
})
|
||||
Reference in new issue
Block a user