fix(blob): tolerate source-unavailable blobs in pre-commit so a missing blob can't wedge replication

DESIGN.md

@@ -42,6 +42,8 @@ The mitigations live in three places:
 
 When adding a new commit-handler early-return path: reset `write.skipped = false` at the top of the handler if you don't already, then set `write.skipped = true` immediately before the `return`. Decide first whether the audit log will reference the blob (via `auditRecordToStore`) — if it does, leave `skipped` unset. `cleanupOrphans` is the periodic safety net; don't rely on it for transactional correctness.
 
+**Source-unavailable blobs must not abort the commit.** `startPreCommitBlobsForRecord().complete()` awaits each blob's `saving` promise; a rejection there propagates up and aborts the record's apply (the replication subscription loop catches and logs it as `error in subscription handler`). For a blob the replication source can no longer provide — evicted/expired at the origin, the receiver having flagged the rejection `sourceBlobUnavailable` (harper-pro#403) — that abort permanently wedged a replication copy stream on an expiration cache table whose TTL-evicted blobs are gone everywhere: every orphaned record's apply re-threw, the copy never advanced, and backpressure pinned at ~100%. `complete()` therefore tolerates a `sourceBlobUnavailable` rejection (`isSourceBlobUnavailable`): the record commits with a diverged blob reference, left for proactive backfill (harper-pro#388). Local/transient save faults stay unmarked and still reject, so the write aborts and a reconnect retries it — no silent loss. This is the apply/commit-side complement to the replication receiver's resume-cursor advance (harper-pro#403/#405), which handles the durability-watermark side of the same missing blob.
+
 ## Opening a source LMDB DBI for migration must thread through `compression`
 
 When `migrateOnStart` opens a source LMDB primary store to read records out for the RocksDB copy, it constructs an `OpenDBIObject` and calls `sourceRootStore.openDB(key, dbiInit)`. Critically, the per-attribute `compression` setting from the corresponding `__dbis__` entry must be assigned onto `dbiInit` before that call — `dbiInit.compression = attribute.compression`. Without it, lmdb-js doesn't install its decompression layer; every read on the DBI returns raw compressed bytes. msgpackr then misreads bytes in the `0x40–0x7F` range as shared-structure refs, calls `loadStructures` → decodes the (also compressed) structures buffer → finds more bytes in that range → recurses → stack overflow.

resources/blob.ts

@@ -1020,6 +1020,22 @@ export interface PreCommitBlobs {
 	complete: () => Promise<void[]>;
 }
 
+/**
+ * Whether a blob-save rejection denotes a blob the replication SOURCE can no longer provide — evicted or
+ * expired at the origin, with the receiver having flagged the error `sourceBlobUnavailable` (set by the
+ * replication layer's `markSourceBlobUnavailable`, harper-pro#403). Such a blob is unrecoverable —
+ * re-streaming reproduces the ENOENT — so the pre-commit path tolerates it and lets the record commit with
+ * a diverged blob reference (left for proactive backfill, harper-pro#388) instead of aborting the apply.
+ * Local/transient save faults are NOT this and must still abort the write so it is retried (no silent loss).
+ */
+export function isSourceBlobUnavailable(error: unknown): boolean {
+	return (
+		typeof error === 'object' &&
+		error !== null &&
+		(error as { sourceBlobUnavailable?: boolean }).sourceBlobUnavailable === true
+	);
+}
+
 /**
  * Do a shallow/fast search for blobs on the record and start saving them if they are supposed to be saved before a commit
  * @param record
@@ -1065,7 +1081,15 @@ export function startPreCommitBlobsForRecord(
 				currentStore = store;
 				return Promise.all(
 					blobsNeedingSaving.map((blob) => {
-						return saveBlob(blob as any, true).saving ?? Promise.resolve();
+						const saving = saveBlob(blob as any, true).saving ?? Promise.resolve();
+						// Tolerate a blob the replication source can no longer provide so the record still commits
+						// with a diverged reference (backfilled later, harper-pro#388) instead of aborting the apply
+						// and wedging the copy stream. Local/transient faults still reject → the write aborts and
+						// retries (no silent loss). See isSourceBlobUnavailable / harper-pro#403.
+						return saving.catch((error) => {
+							if (isSourceBlobUnavailable(error)) return;
+							throw error;
+						});
 					})
 				);
 			},

unitTests/resources/blob.test.js

@@ -19,6 +19,7 @@ const {
 	saveBlob,
 	decodeFromDatabase,
 	startPreCommitBlobsForRecord,
+	isSourceBlobUnavailable,
 } = require('#src/resources/blob');
 const { existsSync, unlinkSync } = require('fs');
 const { pack } = require('msgpackr');
@@ -513,6 +514,42 @@ describe('Blob test', () => {
 		assert(preCommit && preCommit.blobs.includes(receivedBlob), 'received blob tracked on source apply');
 		unlinkSync(getFilePathForBlob(receivedBlob)); // not referenced by any record; remove so it isn't counted as an orphan
 	});
+	it('isSourceBlobUnavailable: only the replication source-missing marker, not local/transient faults', () => {
+		// The classification gate for pre-commit tolerance: the replication receiver flags an unrecoverable
+		// source-missing blob with `sourceBlobUnavailable` (markSourceBlobUnavailable, harper-pro#403). A
+		// local/transient save fault (disk full, a local ENOENT) is unmarked and must NOT be tolerated.
+		assert.equal(
+			isSourceBlobUnavailable(Object.assign(new Error('Blob error: ENOENT'), { sourceBlobUnavailable: true })),
+			true
+		);
+		assert.equal(isSourceBlobUnavailable(Object.assign(new Error('ENOENT'), { code: 'ENOENT' })), false);
+		assert.equal(isSourceBlobUnavailable(new Error('disk full')), false);
+		assert.equal(isSourceBlobUnavailable({ sourceBlobUnavailable: false }), false);
+		assert.equal(isSourceBlobUnavailable(null), false);
+		assert.equal(isSourceBlobUnavailable(undefined), false);
+	});
+	it('startPreCommitBlobsForRecord.complete() tolerates a source-missing blob but still rejects a transient fault', async () => {
+		// harper-pro#403/#388: a replicated record whose blob is gone at the source must still commit (blob
+		// diverged, backfilled later) rather than aborting the apply and wedging the copy stream. A genuine
+		// local/transient save fault must still abort the write so it is retried (no silent data loss).
+		const store = BlobTest.primaryStore.rootStore;
+
+		// source-unavailable: destroy the receive stream with the replication marker → saving rejects → tolerated
+		const goneStream = new PassThrough();
+		const goneBlob = await createBlob(goneStream);
+		decodeFromDatabase(() => saveBlob(goneBlob, true), store); // start the save (assigns fileId, begins the pipeline)
+		const toleratePc = startPreCommitBlobsForRecord({ id: 1, blob: goneBlob }, store, false, true);
+		goneStream.destroy(Object.assign(new Error('Blob error: ENOENT'), { sourceBlobUnavailable: true }));
+		await assert.doesNotReject(toleratePc.complete(), 'source-unavailable blob must not abort the commit');
+
+		// transient/local fault: unmarked rejection must still propagate so the write aborts and retries
+		const failStream = new PassThrough();
+		const failBlob = await createBlob(failStream);
+		decodeFromDatabase(() => saveBlob(failBlob, true), store);
+		const rejectPc = startPreCommitBlobsForRecord({ id: 2, blob: failBlob }, store, false, true);
+		failStream.destroy(new Error('disk full')); // no sourceBlobUnavailable marker
+		await assert.rejects(rejectPc.complete(), 'a local/transient save fault must still abort the commit');
+	});
 	it('#406: cleanupUnusedBlobs deletes non-retained blobs but keeps retained ones', async () => {
 		// The retained-fileId guard: a skipped/aborted write may carry a blob whose fileId the surviving
 		// record still references; deleting it would corrupt that record.