From 35dd8e7773116f11f8a3c42c81840f93290afc7b Mon Sep 17 00:00:00 2001 From: McHersheys <285552078+McHersheys@users.noreply.github.com> Date: Thu, 22 Oct 2026 23:20:49 +0000 Subject: [PATCH] fix(relay): finish pending cache reads before yielding [H15,H17] --- src/relay-page-cache.js | 15 +++- test/relay-page-cache.test.js | 122 +++++++++++++++++++++++++++++++++ test/relay-page-client.test.js | 24 +++++++ 3 files changed, 158 insertions(+), 3 deletions(-) create mode 100644 test/relay-page-cache.test.js diff --git a/src/relay-page-cache.js b/src/relay-page-cache.js index 1d977ed4..32073949 100644 --- a/src/relay-page-cache.js +++ b/src/relay-page-cache.js @@ -140,9 +140,18 @@ export function createRelayPageCache(state) { /** Stream only objects not yet acknowledged by this relay, without an aggregate request body. */ function* pending(source) { - // Node 22.13 may finalize an unreferenced statement while its iterator is suspended across a yield. - const statement = db.prepare('SELECT object.id FROM object JOIN job_object USING(id) WHERE source=? AND ack=0 ORDER BY object.rowid'); - for (const row of statement.iterate(source)) yield object(row.id); + // Finish each read before yielding: an awaited PUT may let another connection commit, + // and a suspended SQLite iterator would make the subsequent ACK upgrade a stale snapshot. + // CROSS JOIN keeps the rowid seek outermost instead of sorting the whole job for each page. + const statement = db.prepare('SELECT object.rowid AS position,object.id FROM object CROSS JOIN job_object USING(id) WHERE source=? AND ack=0 AND object.rowid>? ORDER BY object.rowid LIMIT 1'); + statement.setReadBigInts(true); + let position = 0n; + for (;;) { + const row = statement.get(source, position); + if (!row) return; + position = row.position; + yield object(row.id); + } } /** A lost ACK safely retries the same immutable object; only an actual ACK marks it done. */ diff --git a/test/relay-page-cache.test.js b/test/relay-page-cache.test.js new file mode 100644 index 00000000..b1fc3f07 --- /dev/null +++ b/test/relay-page-cache.test.js @@ -0,0 +1,122 @@ +/** Pending cipher pages finish each SQLite read before awaiting a remote ACK [H15,H17]. */ +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { DatabaseSync } from 'node:sqlite'; +import { test } from 'node:test'; +import { createRelayPageCache, snapshotSourceId } from '../src/relay-page-cache.js'; + +/** Prepare a small child-first graph using only native positive rowid inserts. */ +function graph(cache, label = 'pending') { + const source = snapshotSourceId(label), values = []; + for (let index = 0; index < 4; index++) { + const bytes = Buffer.from(label + index); + const value = { id: snapshotSourceId(bytes), sequence: 7, bytes, children: values.map(child => child.id) }; + cache.put(value); values.push(value); + } + cache.prepare(source, { id: values.at(-1).id, sequence: 7, bytes: 123 }); + return { source, values }; +} + +/** Open two real WAL connections in a fresh private device home, never an existing cache. */ +function fixture(t) { + const root = mkdtempSync(join(tmpdir(), 'pullboard-pending-cache-')); + const old = process.env.PULLBOARD_HOME; + process.env.PULLBOARD_HOME = join(root, 'device'); + const state = { board: 'a'.repeat(32), url: 'https://synthetic-relay.example' }; + const cache = createRelayPageCache(state), seeded = graph(cache), writer = createRelayPageCache(state); + t.after(() => { + writer.close(); cache.close(); + if (old === undefined) delete process.env.PULLBOARD_HOME; else process.env.PULLBOARD_HOME = old; + rmSync(root, { recursive: true, force: true }); + }); + return { state, cache, writer, ...seeded }; +} + +/** Commit after crossing the same asynchronous boundary as the client's object PUT. */ +async function competingCommit(writer) { + await new Promise(resolve => setImmediate(resolve)); + assert.equal(writer.firstNotice(snapshotSourceId('competing-writer')), true); +} + +test('pending ACK after an awaited PUT tolerates a concurrent WAL commit [H15,H17]', async t => { + const box = fixture(t), iterator = box.cache.pending(box.source); + try { + const first = iterator.next().value; + await competingCommit(box.writer); + box.cache.acknowledge(first.id); + assert.equal(box.cache.object(first.id).acknowledged, true); + } finally { iterator.return(); } +}); + +test('pending causal control: no competing writer before ACK [H15,H17]', async t => { + const box = fixture(t), iterator = box.cache.pending(box.source); + try { + const first = iterator.next().value; + await Promise.resolve(); box.cache.acknowledge(first.id); + assert.equal(box.cache.object(first.id).acknowledged, true); + } finally { iterator.return(); } +}); + +test('pending causal control: finalize the read before competing commit and ACK [H15,H17]', async t => { + const box = fixture(t), iterator = box.cache.pending(box.source); + const first = iterator.next().value; iterator.return(); + await competingCommit(box.writer); box.cache.acknowledge(first.id); + assert.equal(box.cache.object(first.id).acknowledged, true); +}); + +test('pending causal control: competing commit precedes the first read [H15,H17]', async t => { + const box = fixture(t); await competingCommit(box.writer); + for (const value of box.cache.pending(box.source)) box.cache.acknowledge(value.id); + assert.deepEqual([...box.cache.pending(box.source)], []); +}); + +test('pending stays lazy, source-bound and ordered while prior ACKs survive reopen [H15,H17]', t => { + const box = fixture(t), other = graph(box.cache, 'other'), raw = new DatabaseSync(box.cache.file); + try { + raw.prepare('UPDATE object SET payload=? WHERE id=?').run(Buffer.from('damaged'), box.values[3].id); + const iterator = box.cache.pending(box.source); + try { + const first = iterator.next().value; + assert.equal(first.id, box.values[0].id, 'a later damaged page is not loaded eagerly'); + box.cache.acknowledge(first.id); + assert.equal(iterator.next().value.id, box.values[1].id); + assert.equal(iterator.next().value.id, box.values[2].id); + assert.throws(() => iterator.next(), { code: 'RELAY_PAGE_CACHE' }); + } finally { iterator.return(); } + const cold = createRelayPageCache(box.state); + try { + assert.equal(cold.object(box.values[0].id).acknowledged, true); + assert.deepEqual([...cold.pending(other.source)].map(value => value.id), other.values.map(value => value.id)); + } finally { cold.close(); } + } finally { raw.close(); } +}); + +test('pending observes another uploader ACK between yields [H15,H17]', t => { + const box = fixture(t), iterator = box.cache.pending(box.source); + try { + const first = iterator.next().value; + box.writer.acknowledge(box.values[1].id); box.cache.acknowledge(first.id); + assert.deepEqual([...iterator].map(value => value.id), box.values.slice(2).map(value => value.id)); + } finally { iterator.return(); } +}); + +test('pending rowid cursor keeps integer precision beyond the Number safe range [H15,H17]', t => { + const box = fixture(t), raw = new DatabaseSync(box.cache.file); + try { + for (const [index, value] of box.values.entries()) raw.prepare('UPDATE object SET rowid=? WHERE id=?').run(9007199254741000n + BigInt(index), value.id); + assert.deepEqual([...box.cache.pending(box.source)].map(value => value.id), box.values.map(value => value.id)); + } finally { raw.close(); } +}); + +test('pending seeks by rowid without sorting the whole job for each page [H15,H17]', t => { + const box = fixture(t), raw = new DatabaseSync(box.cache.file); + try { + const sql = /db\.prepare\('([^']+)'\)/u.exec(box.cache.pending.toString())[1]; + const plan = raw.prepare('EXPLAIN QUERY PLAN ' + sql).all(box.source, 0); + assert.ok(plan.some(row => /INTEGER PRIMARY KEY \(rowid>\?\)/u.test(row.detail))); + assert.ok(plan.every(row => !/TEMP B-TREE/u.test(row.detail))); + t.diagnostic(JSON.stringify({ pendingQueryPlan: plan })); + } finally { raw.close(); } +}); diff --git a/test/relay-page-client.test.js b/test/relay-page-client.test.js index d13d3de9..1a25a298 100644 --- a/test/relay-page-client.test.js +++ b/test/relay-page-client.test.js @@ -82,6 +82,30 @@ async function fixture(t) { loseObjectAck() { loseObjectAck = true; }, corruptRootAck() { corruptRootAck = true; } }; } +test('paged upload ACK tolerates a concurrent cache commit during an awaited PUT [H15,H17]', async t => { + const box = await fixture(t), writer = createRelayPageCache(box.state); + t.after(() => writer.close()); + const request = box.adapter.request; + let committed = false; + /** Return the real page ACK only after a second connection commits across the await. */ + box.adapter.request = async (path, options) => { + const response = await request(path, options); + if (!committed && /\/objects\/[0-9a-f]{64}$/u.test(path)) { + await new Promise(resolve => setImmediate(resolve)); + assert.equal(writer.firstNotice(snapshotSourceId('concurrent-upload-notice')), true); + committed = true; + } + return response; + }; + const pending = { sequence: 0, sealed: 'synthetic-concurrent-upload' }; + const result = await pagedSnapshotRequest(box.state, { method: 'PUT', body: pending }, box.adapter); + assert.equal(committed, true); + assert.equal(result.state.sealed, pending.sealed); + assert.ok(box.calls.filter(call => call.method === 'PUT').length > 8, 'the real encrypted graph is streamed'); + assert.deepEqual([...writer.pending(snapshotSourceId(pending.sealed))], []); + assert.deepEqual(JSON.parse((await pagedSnapshotRequest(box.state, {}, box.adapter)).state.sealed), box.document); +}); + test('a lost object ACK resumes the exact durable graph and native restore preserves all rows [H15,H17]', async t => { const box = await fixture(t); const pending = { sequence: 0, sealed: 'synthetic-original-pending-cipher' };