From fc86dd00d1bcd43593832c1bc362db70bc41590b Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Fri, 9 Oct 2026 18:05:00 +0200 Subject: [PATCH] Rebuild: a batch locks its range with an advisory lock, not the row MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BatchStore.store held the range's row (SELECT … FOR UPDATE) for the whole batch transaction (15-20 s on the HDD), so the prefetcher's spooled_chunks updates and the heartbeats of that range waited on it (0.5 s on average): the prefetcher's NNTP connections sat idle. Now an advisory lock per range serializes its batches, and the resume point is checked again in the final, conditional update. Co-Authored-By: Claude Opus 5.5 --- app/services/batch_store.ts | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index 4a910e8..14c9673 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -23,7 +23,11 @@ class BatchStore { async store(rangeId: number, nodeId: number, batch: EdgeBatch): Promise { await ChunkPartitions.ensure() return db.transaction(async (trx) => { - const range = await trx.from('ranges').where('id', rangeId).forUpdate().first() + // One batch of a range at a time, without holding the range's row for the whole transaction: the + // prefetcher's and heartbeats' updates of it waited on it (0.5 s on average, NNTP connections idle; + // LOG.md 2026-10-09 (13)). The resume point is checked again when it moves, at the end. + await trx.rawQuery(`SELECT pg_advisory_xact_lock(hashtextextended('range:' || ?::text, 0))`, [rangeId]) + const range = await trx.from('ranges').where('id', rangeId).first() if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId) throw new StaleBatch(`Range ${rangeId} is not leased to this node`) if (Number(range.next_id) !== batch.from) @@ -80,14 +84,18 @@ class BatchStore { await trx.insertQuery().table('fragments').multiInsert(rows) } - await trx + const moved = await trx .from('ranges') .where('id', rangeId) + .where('status', 'leased') + .where('node_id', nodeId) + .where('next_id', batch.from) .update({ next_id: batch.next, headers: db.raw('headers + ?', [batch.headers]), heartbeat_at: new Date(), }) + if (Number(moved) !== 1 && !(Array.isArray(moved) && moved.length === 1)) throw new StaleBatch(`Range ${rangeId} moved on meanwhile`) return result }) }