diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index efce2ec..cc5b2bc 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -190,9 +190,11 @@ class BatchStore { list.push({ kind: c.kind, messageId: c.messageId, file: c.file, bytes: c.bytes }) targets.set(key, list) } - let written = 0 - for (let i = 0; i < releases.length; i += 1000) { - const rows = releases.slice(i, i + 1000).map((r) => { + // In the record's lock order (LOG.md (136)): all rows sorted, then all classes sorted, in one transaction. The + // upsert locks a finished row too, before its WHERE passes it by; a chunk's classes before the next chunk's rows + // crossed the ladder's record(), which holds its rows and then takes its classes. + const rows = lockOrder( + releases.map((r) => { const probes = targets.get(slot(r.key, r.upload)) ?? [] return { release_key: releaseKey(r.key, groupId, r.firstArticle).toString('hex'), @@ -214,7 +216,12 @@ class BatchStore { // The first target's article (by effort): the cheapest probe first within a class (LOG.md (130)). target_bytes: byEffort(probes)[0]?.bytes ?? null, } - }) + }), + (r) => r.release_key + ) + let written = 0 + const classes = new Map() + for (let i = 0; i < rows.length; i += 1000) { const { rows: done } = await trx.rawQuery( `INSERT INTO pending_releases (release_key, identity_key, group_id, name, poster, posted_at, last_posted_at, first_article, last_article, files, segments, bytes, partial, finished, file_total, tool, candidates, klass, target_bytes) @@ -232,29 +239,27 @@ class BatchStore { target_bytes = coalesce(pending_releases.target_bytes, excluded.target_bytes) WHERE NOT pending_releases.finished RETURNING release_key, tool, klass, (xmax = 0) AS inserted`, - // In the record's lock order: the upsert locks a finished row too, before its WHERE passes it by. - [groupId, JSON.stringify(lockOrder(rows, (r) => r.release_key))] + [groupId, JSON.stringify(rows.slice(i, i + 1000))] ) written += done.length // Each class seen gets its row in the ladder's record, so the claim explores it (pending_classes). - const classes = new Map() for (const d of done) { const key = `${d.tool}\t${d.klass}` const entry = classes.get(key) ?? { tool: d.tool, klass: Number(d.klass), waiting: 0 } if (d.inserted) entry.waiting++ classes.set(key, entry) } - if (classes.size > 0) { - await trx.rawQuery( - `INSERT INTO pending_classes (tool, group_id, klass, waiting) - SELECT v.tool, ?, v.klass, v.waiting FROM jsonb_to_recordset(?::jsonb) AS v(tool text, klass int, waiting int) - ON CONFLICT (tool, group_id, klass) DO UPDATE SET waiting = pending_classes.waiting + excluded.waiting`, - [ - groupId, - JSON.stringify(lockOrder(classes.values(), (c) => `${c.tool}\t${groupId}\t${c.klass}`)), - ] - ) - } + } + if (classes.size > 0) { + await trx.rawQuery( + `INSERT INTO pending_classes (tool, group_id, klass, waiting) + SELECT v.tool, ?, v.klass, v.waiting FROM jsonb_to_recordset(?::jsonb) AS v(tool text, klass int, waiting int) + ON CONFLICT (tool, group_id, klass) DO UPDATE SET waiting = pending_classes.waiting + excluded.waiting`, + [ + groupId, + JSON.stringify(lockOrder(classes.values(), (c) => `${c.tool}\t${groupId}\t${c.klass}`)), + ] + ) } return written } diff --git a/database/migrations/1770000007300_probe_candidates_on_nvme.ts b/database/migrations/1770000007300_probe_candidates_on_nvme.ts new file mode 100644 index 0000000..ea2a671 --- /dev/null +++ b/database/migrations/1770000007300_probe_candidates_on_nvme.ts @@ -0,0 +1,34 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * probe_candidates, with its indexes, in the catalog's tablespace (default_tablespace: nvme on homelab), out of + * CHUNK_TABLESPACE where migration 6000 put it: the prober's catalog claim updates it from random points every + * minute, and on the HDD, beside the pending ladder's sweep and autovacuum, a claim took 2–3 minutes with all the + * lanes waiting (LOG.md 2026-11-03 (136)). Homelab moved it by hand (edge, node-api and the workers stopped; ~8 GB); + * a no-op where it's there already, or without a default tablespace (tests). + */ +export default class extends BaseSchema { + async up() { + this.defer(async (db) => { + const { rows: setting } = await db.rawQuery( + `SELECT current_setting('default_tablespace') AS target` + ) + const target = setting[0].target + if (!target) return + const { rows } = await db.rawQuery( + `SELECT c.relname, c.relkind FROM pg_class c LEFT JOIN pg_tablespace t ON t.oid = c.reltablespace + WHERE (c.relname = 'probe_candidates' OR c.oid IN (SELECT indexrelid FROM pg_index WHERE indrelid = 'probe_candidates'::regclass)) + AND c.relkind IN ('r', 'i') AND coalesce(t.spcname, 'pg_default') <> ? + ORDER BY c.relkind DESC, c.relname`, + [target] + ) + for (const { relname, relkind } of rows) { + await db.rawQuery( + `ALTER ${relkind === 'i' ? 'INDEX' : 'TABLE'} ${relname} SET TABLESPACE ${target}` + ) + } + }) + } + + async down() {} +}