Skip to content

fix(blob): never gate apply commits on wire-carried save flags; gate local writes on blob durability #1641

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kriszyp merged 4 commits into from
Jul 7, 2026
Merged

fix(blob): never gate apply commits on wire-carried save flags; gate local writes on blob durability #1641

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e8803c9
fix(blob): gate local-write record commit on blob durability
Jul 3, 2026
7f6cec4
fix(blob): never gate apply commits on wire-carried save flags
ldt1996 Jul 7, 2026
6d74ff6
fix(blob): exempt already-saved blobs from the local-write gate, upda…
ldt1996 Jul 7, 2026
f6b1b96
fix(blob): gate on still-streaming manual saves, strip wire flags via…
ldt1996 Jul 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 38 additions & 4 deletions resources/blob.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ type StorageInfo = {
start?: number;
end?: number;
saving?: Promise<void>;
saved?: boolean; // saving settled successfully; distinguishes durable from still-streaming when fileId is already assigned
asString?: string;
deleteOnFailure?: boolean;
};
Expand Down Expand Up @@ -1150,6 +1151,14 @@ function writeBlobWithStream(blob: Blob, stream: Readable, storageInfo: StorageI
}
}
});
// Mark durable on settle: fileId is assigned as soon as a save STARTS, so pre-commit gating
// needs this to tell a durable blob from one still streaming (review on the local-write gate)
storageInfo.saving.then(
() => {
storageInfo.saved = true;
},
() => {}
);
return blob;
}

Expand Down Expand Up @@ -1583,14 +1592,37 @@ export function startPreCommitBlobsForRecord(
// retained-fileId guard in cleanupUnusedBlobs additionally protects any blob the committed record
// still references.
if (value instanceof FileBackedBlob) {
if (saveInRecord || value.saveBeforeCommit) {
if (trackPersistedBlobs) {
// Replication-apply path: the blob was received out-of-band and its durability
// is tracked separately via outstandingBlobsToFinish. Awaiting it here would
// deadlock a paused WS (see the block comment above). This branch must come
// FIRST: `saveBeforeCommit` rides the wire (pack spreads own properties, unpack
// Object.assigns them back), so an origin-side flag would otherwise gate this
// apply's commit on an in-flight receive stream whose chunks are queued behind
// the paused socket - the copy-leg deadlock (record commits stall, the queue
// never drains, the sender's copy watchdog kills the leg every 300s).
if (storageInfoForBlob.get(value)?.fileId != null) {
blobsToTrackOnly.push(value);
}
} else if (saveInRecord || value.saveBeforeCommit) {
currentStore = store;
if (saveInRecord) {
value.saveInRecord = true;
}
blobsNeedingSaving.push(value);
} else if (trackPersistedBlobs && storageInfoForBlob.get(value)?.fileId != null) {
blobsToTrackOnly.push(value);
} else if (
storageInfoForBlob.get(value)?.fileId == null ||
(storageInfoForBlob.get(value)?.saving != null && !storageInfoForBlob.get(value)?.saved)
) {
// Local-write path with a fresh blob: gate the record commit on the blob's
// durable landing. Without this the record commits before the async save has
// fsync'd, and a peer that pulls the record first sees an orphan ref and marks
// the record permanently diverged (a variant of the harper-pro#481 pattern
// reachable via the write path rather than the receive path). Not a deadlock
// risk like the trackPersistedBlobs branch: local blob sources are the HTTP
// body or an in-memory buffer, not a paused WS receive stream.
currentStore = store;
blobsNeedingSaving.push(value);
}
}
}
Expand Down Expand Up @@ -1708,7 +1740,9 @@ addExtension({
throw new Error('Cannot use the same blob in two different records');
}
}
const options = { ...blob };
// saveBeforeCommit/saveInRecord are origin-local write instructions, not blob data;
// shipping them makes the receiver's apply commit gate on an in-flight receive stream
const { saveBeforeCommit: _sbc, saveInRecord: _sir, ...options } = blob as any;
if (blob.type) options.type = blob.type;
if (blob.size !== undefined) options.size = blob.size;
if (storageInfo) {
Expand Down
68 changes: 31 additions & 37 deletions unitTests/resources/blob.test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -399,7 +399,7 @@ describe('Blob test', () => {
let packResult = encodeBlobsAsBuffers(() => {
return pack(record);
});
assert(packResult.then); // shouldn't be resolved yet
assert(!packResult.then); // put gates the commit on the blob's durable save, so pack resolves synchronously
let retrievedText = await record.blob.text();
assert.equal(retrievedText, expectedResults);
assert.equal(await streamResults, expectedResults);
Expand Down Expand Up @@ -511,15 +511,15 @@ describe('Blob test', () => {
}
}
let blob = createBlob(new BadStream());
await BlobTest.put({ id: 5, blob });
await assert.rejects(async () => {
await BlobTest.put({ id: 5, blob });
}, /test error/);
let eventError, thrownError;
blob.on('error', (err) => {
console.log('received error event');
eventError = err;
});
try {
await blob.written;
} catch {}
await assert.rejects(blob.written, /test error/);
try {
for await (let _entry of blob.stream()) {
console.log('got entry');
Expand All @@ -529,27 +529,16 @@ describe('Blob test', () => {
}
assert(thrownError);
assert(eventError);
thrownError = null;
eventError = null;
let record = await BlobTest.get(5);
record.blob.on('error', (err) => {
eventError = err;
});
try {
for await (let _entry of record.blob.stream()) {
}
} catch (err) {
thrownError = err;
}
assert(thrownError);
assert(eventError);
assert.equal(await BlobTest.get(5), undefined);
});
it('Error before streaming', async () => {
let pt = new PassThrough();
pt.on('error', () => {}); // ignore the uncaught error
pt.destroy(new Error('test error'));
let blob = createBlob(pt);
await BlobTest.put({ id: 6, blob });
await assert.rejects(async () => {
await BlobTest.put({ id: 6, blob });
}, /test error/);
let eventError, thrownError;
blob.on('error', (err) => {
eventError = err;
Expand All @@ -563,21 +552,7 @@ describe('Blob test', () => {
}
assert(thrownError);
assert(eventError);
thrownError = null;
eventError = null;

let record = await BlobTest.get(6);
record.blob.on('error', (err) => {
eventError = err;
});
try {
for await (let _entry of record.blob.stream()) {
}
} catch (err) {
thrownError = err;
}
assert(thrownError);
assert(eventError);
assert.equal(await BlobTest.get(6), undefined);
});
it('invalid blob attempts', async () => {
assert.throws(() => {
Expand Down Expand Up @@ -661,6 +636,25 @@ describe('Blob test', () => {
const existing = await BlobTest.get(250);
assert.equal(await existing.blob.text(), 'first');
});
it('gates a local write on a manually started, still-streaming blob save', async () => {
// saveBlob assigns the fileId as soon as the save STARTS, so a fileId check alone would
// exempt a mid-save blob from the local-write gate and reopen the orphan-reference window
const slow = new PassThrough();
const blob = createBlob(slow);
saveBlob(blob);
slow.write('partial content');
const store = BlobTest.primaryStore.rootStore;
const preCommit = startPreCommitBlobsForRecord({ id: 8, blob }, store, false, false);
assert(preCommit && preCommit.blobs.includes(blob), 'mid-save blob must gate the commit');
let completed = false;
const completion = preCommit.complete().then(() => (completed = true));
await delay(20);
assert.equal(completed, false, 'commit must wait for the save to settle');
slow.end(' done');
await completion;
assert.equal(completed, true);
unlinkSync(getFilePathForBlob(blob)); // not referenced by any record; keep cleanupOrphans at zero
});
it('#406: startPreCommitBlobsForRecord tracks an already-saved blob only when trackPersistedBlobs is set', async () => {
// A replication-received blob is saved out-of-band by receiveBlobs BEFORE the apply, so at pre-commit
// it has a fileId but no saveBeforeCommit flag. It must be tracked (so a superseded apply's cleanup
Expand Down Expand Up @@ -718,7 +712,7 @@ describe('Blob test', () => {
// the write aborts and retries.
const failStream = new PassThrough();
const failBlob = await createBlob(failStream, { saveBeforeCommit: true });
const rejectPc = startPreCommitBlobsForRecord({ id: 1, blob: failBlob }, store, false, true);
const rejectPc = startPreCommitBlobsForRecord({ id: 1, blob: failBlob }, store, false, false);
failStream.destroy(new Error('disk full')); // no sourceBlobUnavailable marker
await assert.rejects(
rejectPc.complete(),
Expand All @@ -729,7 +723,7 @@ describe('Blob test', () => {
// so the record still commits with a diverged reference.
const goneStream = new PassThrough();
const goneBlob = await createBlob(goneStream, { saveBeforeCommit: true });
const toleratePc = startPreCommitBlobsForRecord({ id: 2, blob: goneBlob }, store, false, true);
const toleratePc = startPreCommitBlobsForRecord({ id: 2, blob: goneBlob }, store, false, false);
goneStream.destroy(Object.assign(new Error('Blob error: ENOENT'), { sourceBlobUnavailable: true }));
await assert.doesNotReject(toleratePc.complete(), 'a source-unavailable blob must not abort the commit');

Expand Down
Loading