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 }) }