Rebuild: a batch locks its range with an advisory lock, not the row
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 <noreply@anthropic.com>
This commit is contained in:
1 parent
99fffe30f4
commit
fc86dd00d1
1 file changed
+10
-2
@@ -23,7 +23,11 @@ class BatchStore {
|
|||||||
async store(rangeId: number, nodeId: number, batch: EdgeBatch): Promise<StoredBatch> {
|
async store(rangeId: number, nodeId: number, batch: EdgeBatch): Promise<StoredBatch> {
|
||||||
await ChunkPartitions.ensure()
|
await ChunkPartitions.ensure()
|
||||||
return db.transaction(async (trx) => {
|
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)
|
if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId)
|
||||||
throw new StaleBatch(`Range ${rangeId} is not leased to this node`)
|
throw new StaleBatch(`Range ${rangeId} is not leased to this node`)
|
||||||
if (Number(range.next_id) !== batch.from)
|
if (Number(range.next_id) !== batch.from)
|
||||||
@@ -80,14 +84,18 @@ class BatchStore {
|
|||||||
await trx.insertQuery().table('fragments').multiInsert(rows)
|
await trx.insertQuery().table('fragments').multiInsert(rows)
|
||||||
}
|
}
|
||||||
|
|
||||||
await trx
|
const moved = await trx
|
||||||
.from('ranges')
|
.from('ranges')
|
||||||
.where('id', rangeId)
|
.where('id', rangeId)
|
||||||
|
.where('status', 'leased')
|
||||||
|
.where('node_id', nodeId)
|
||||||
|
.where('next_id', batch.from)
|
||||||
.update({
|
.update({
|
||||||
next_id: batch.next,
|
next_id: batch.next,
|
||||||
headers: db.raw('headers + ?', [batch.headers]),
|
headers: db.raw('headers + ?', [batch.headers]),
|
||||||
heartbeat_at: new Date(),
|
heartbeat_at: new Date(),
|
||||||
})
|
})
|
||||||
|
if (Number(moved) !== 1 && !(Array.isArray(moved) && moved.length === 1)) throw new StaleBatch(`Range ${rangeId} moved on meanwhile`)
|
||||||
return result
|
return result
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in new issue
Block a user