Skip to content

Commit ce88577

Browse files
kriszypkriszyp
authored
Merge pull request #1444 from HarperFast/fix/blob-receive-idle-watchdog
fix(blob): idle watchdog on writeBlobWithStream source to unwedge stalled replication receives
2 parents 6721253 + d288517 commit ce88577

3 files changed

Lines changed: 247 additions & 3 deletions

File tree

‎resources/blob.ts‎

Lines changed: 64 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@ type StorageInfo = {
5151
filePath?: string;
5252
recordId?: number;
5353
contentBuffer?: any;
54-
source?: NodeJS.ReadableStream;
54+
source?: Readable;
5555
storageBuffer?: Buffer;
5656
compress?: boolean;
5757
flush?: boolean;
@@ -753,7 +753,40 @@ export function saveBlob(blob: FileBackedBlob, deleteOnFailure = false) {
753753
/**
754754
* Create a blob from a readable stream
755755
*/
756-
function writeBlobWithStream(blob: Blob, stream: NodeJS.ReadableStream, storageInfo: StorageInfo): Blob {
756+
// Source-stream idle timeout for writeBlobWithStream. When > 0, a stream that goes this long without
757+
// delivering data is force-destroyed so the save promise rejects in finite time. Without it,
758+
// pipeline(source, writeStream, finished) sits forever on a sender that never sends the closing
759+
// `finished:true` BLOB_CHUNK, saveBlob.saving never settles, outstandingBlobsToFinish keeps the
760+
// stuck promise, and the per-database apply consumer's drain await wedges — surfaced in production
761+
// as a (sender, receiver, database) tuple pinned at lastReceivedStatus="Receiving" indefinitely
762+
// (JJill preprod 5.1.7; harper-pro#453).
763+
//
764+
// OFF by default. writeBlobWithStream is the generic primitive for every blob write — HTTP upload,
765+
// origin-fetch cache fill, replication receive — and bounding a source's liveness is the owning
766+
// caller's responsibility, not this primitive's: a blanket timeout here would destroy a legitimately
767+
// slow non-replication source. The owning caller arms it per-write by setting `blobStreamIdleTimeoutMs`
768+
// on the source stream (the replication receive path does this on its PassThrough). A process-wide
769+
// HARPER_BLOB_STREAM_IDLE_TIMEOUT_MS env var, when set, overrides the per-stream value for every write
770+
// (ops escape hatch / kill switch — set 0 to force-disable). Read at call time so tests and operators
771+
// can change it without a restart. The timer re-arms while the source is paused (pipeline backpressure,
772+
// not a real stall) so a slow writeStream never trips a false destroy.
773+
// Largest delay setTimeout accepts; a larger value (or Infinity/NaN) is silently coerced to 1ms, which
774+
// would fire the watchdog almost immediately and destroy a healthy stream — so clamp/reject instead.
775+
const MAX_SET_TIMEOUT_MS = 2147483647; // 2^31 - 1
776+
function getBlobStreamIdleTimeoutMs(stream: Readable): number {
777+
const configured = process.env.HARPER_BLOB_STREAM_IDLE_TIMEOUT_MS;
778+
// env override (process-wide kill switch) when set, else the per-stream value the owning caller armed.
779+
const raw =
780+
configured != null
781+
? Number(configured)
782+
: Number((stream as { blobStreamIdleTimeoutMs?: number }).blobStreamIdleTimeoutMs ?? 0);
783+
// A NaN/negative/zero value means off; cap a too-large value at the setTimeout max so it doesn't
784+
// collapse to 1ms and instantly destroy the source.
785+
if (!Number.isFinite(raw) || raw <= 0) return 0;
786+
return Math.min(raw, MAX_SET_TIMEOUT_MS);
787+
}
788+
789+
function writeBlobWithStream(blob: Blob, stream: Readable, storageInfo: StorageInfo): Blob {
757790
const { filePath, fileId, store, compress, flush } = storageInfo;
758791
storageInfo.saving = new Promise((resolve, reject) => {
759792
// pipe the stream to the file
@@ -768,6 +801,27 @@ function writeBlobWithStream(blob: Blob, stream: NodeJS.ReadableStream, storageI
768801
writeStream.write(createHeader(blob.size)); // write the default header
769802
wroteSize = true;
770803
}
804+
// Source-idle watchdog: destroys the source if no 'data' arrives for the threshold so pipeline
805+
// rejects cleanly. Off unless the owning caller armed this source (or the env override is set);
806+
// see getBlobStreamIdleTimeoutMs. On expiry, re-arm if the stream is paused (pipeline backpressure,
807+
// not a real stall) so a slow writeStream doesn't trip a false destroy.
808+
let idleTimer: NodeJS.Timeout | undefined;
809+
let armIdleTimer: (() => void) | undefined;
810+
const idleTimeoutMs = getBlobStreamIdleTimeoutMs(stream);
811+
if (idleTimeoutMs > 0) {
812+
armIdleTimer = () => {
813+
if (idleTimer) clearTimeout(idleTimer);
814+
idleTimer = setTimeout(() => {
815+
if (stream.isPaused()) {
816+
armIdleTimer?.();
817+
return;
818+
}
819+
stream.destroy(new Error(`Blob source stream idle for ${idleTimeoutMs}ms (fileId=${fileId})`));
820+
}, idleTimeoutMs).unref();
821+
};
822+
stream.on('data', armIdleTimer);
823+
armIdleTimer();
824+
}
771825
let compressedStream: any;
772826
if (compress) {
773827
if (!wroteSize) writeStream.write(COMPRESS_HEADER); // write the default header to the file
@@ -787,6 +841,14 @@ function writeBlobWithStream(blob: Blob, stream: NodeJS.ReadableStream, storageI
787841
}
788842
// when the stream is finished, we may need to flush, and then close the handle and resolve the promise
789843
function finished(error?: Error) {
844+
if (idleTimer) {
845+
clearTimeout(idleTimer);
846+
idleTimer = undefined;
847+
}
848+
if (armIdleTimer) {
849+
stream.removeListener('data', armIdleTimer);
850+
armIdleTimer = undefined;
851+
}
790852
const fd = (writeStream as any).fd;
791853
if (error) {
792854
store.unlock(lockKey);

‎unitTests/resources/blob.test.js‎

Lines changed: 176 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -854,6 +854,182 @@ describe('Blob test', () => {
854854
setDeletionDelay(500); // restore original
855855
});
856856
});
857+
858+
describe('saveBlob with idle source stream (replication wedge regression)', () => {
859+
let WedgeTable;
860+
let savedIdleTimeoutEnv;
861+
before(function () {
862+
setupTestDBPath();
863+
// Enable the source-stream idle timeout for these tests so the wedge case has a finite
864+
// settle deadline. The value must be short enough that the 'never-ended' test settles
865+
// inside its 3s wait.
866+
savedIdleTimeoutEnv = process.env.HARPER_BLOB_STREAM_IDLE_TIMEOUT_MS;
867+
process.env.HARPER_BLOB_STREAM_IDLE_TIMEOUT_MS = '1500';
868+
WedgeTable = table({
869+
table: 'WedgeTable',
870+
database: 'test',
871+
attributes: [
872+
{ name: 'id', isPrimaryKey: true },
873+
{ name: 'blob', type: 'Blob' },
874+
],
875+
});
876+
});
877+
after(function () {
878+
if (savedIdleTimeoutEnv === undefined) delete process.env.HARPER_BLOB_STREAM_IDLE_TIMEOUT_MS;
879+
else process.env.HARPER_BLOB_STREAM_IDLE_TIMEOUT_MS = savedIdleTimeoutEnv;
880+
});
881+
882+
it('settles saveBlob.saving when the source PassThrough was destroyed before save started', async () => {
883+
// Mirrors the replication-receive race: the BLOB_CHUNK handler creates a PassThrough in
884+
// blobsInFlight; a later chunk with `finished:true, error:"..."` calls stream.destroy(err).
885+
// When the audit entry then arrives, receiveBlobs retrieves the destroyed stream and
886+
// saveBlob's pipeline runs over an already-destroyed source. Without the idle watchdog,
887+
// pipeline() may not observe the destroy, saveBlob.saving never settles, and the per-
888+
// (sender, receiver, database) replication tuple wedges at status "Receiving".
889+
const stream = new PassThrough();
890+
stream.on('error', () => {}); // suppress 'unhandled error' from the manual destroy
891+
stream.destroy(new Error('Blob error: simulated upstream tear-down'));
892+
const blob = await createBlob(stream);
893+
const info = decodeFromDatabase(() => saveBlob(blob), WedgeTable.primaryStore.rootStore);
894+
895+
let state = 'pending';
896+
// eslint-disable-next-line promise/catch-or-return
897+
(info.saving ?? Promise.resolve())
898+
.then(() => {
899+
state = 'resolved';
900+
})
901+
.catch(() => {
902+
state = 'rejected';
903+
});
904+
905+
await delay(2000);
906+
assert.notStrictEqual(
907+
state,
908+
'pending',
909+
'saveBlob.saving never settled; in replication this wedges the per-database receive consumer indefinitely'
910+
);
911+
});
912+
913+
it('settles saveBlob.saving when the source stream has chunks but is never ended', async () => {
914+
// Production scenario: a sender's BLOB_CHUNK frames arrive partial. Some content lands but
915+
// the closing `finished:true` (or error) frame never does. The PassThrough sits idle:
916+
// neither ended nor destroyed. Without the idle watchdog, pipeline waits forever and the
917+
// tracked saveBlob.saving promise pins outstandingBlobsToFinish, stalling the apply
918+
// consumer's drain await with no log signature.
919+
const stream = new PassThrough();
920+
stream.write(Buffer.from('chunk-but-no-finish'));
921+
// NO destroy, NO end: prod-observed state of an abandoned blob stream.
922+
923+
const blob = await createBlob(stream);
924+
const info = decodeFromDatabase(() => saveBlob(blob), WedgeTable.primaryStore.rootStore);
925+
926+
let state = 'pending';
927+
// eslint-disable-next-line promise/catch-or-return
928+
(info.saving ?? Promise.resolve())
929+
.then(() => {
930+
state = 'resolved';
931+
})
932+
.catch(() => {
933+
state = 'rejected';
934+
});
935+
936+
await delay(3000);
937+
assert.notStrictEqual(
938+
state,
939+
'pending',
940+
'saveBlob.saving did not settle within 3s for an idle source stream; pipeline waits forever and wedges the per-database replication apply consumer (production: lastReceivedStatus stuck on "Receiving")'
941+
);
942+
});
943+
944+
it('settles when a mid-stream chunk arrives, then a destroy, then no further chunks', async () => {
945+
// More faithful repro of the receive path: PassThrough is created in blobsInFlight, some
946+
// chunks arrive, the stream is destroyed (e.g. by a sender-side error frame), then
947+
// saveBlob is started by the audit-record receive. No further chunks ever land. In the
948+
// production receiver this leaves pipeline() waiting on a torn-down source that never
949+
// ends nor errors from this side, holding outstandingBlobsToFinish forever.
950+
const stream = new PassThrough();
951+
stream.on('error', () => {});
952+
953+
stream.write(Buffer.from('partial-blob-payload-'));
954+
stream.destroy(new Error('Blob error: simulated tear-down mid-stream'));
955+
956+
const blob = await createBlob(stream);
957+
const info = decodeFromDatabase(() => saveBlob(blob), WedgeTable.primaryStore.rootStore);
958+
959+
let state = 'pending';
960+
// eslint-disable-next-line promise/catch-or-return
961+
(info.saving ?? Promise.resolve())
962+
.then(() => {
963+
state = 'resolved';
964+
})
965+
.catch(() => {
966+
state = 'rejected';
967+
});
968+
969+
await delay(3000);
970+
assert.notStrictEqual(
971+
state,
972+
'pending',
973+
'saveBlob.saving never settled with a partially-written-then-destroyed source: replication wedge'
974+
);
975+
});
976+
});
977+
978+
describe('saveBlob source-idle watchdog is opt-in (off by default, per-stream arm)', () => {
979+
let OptInTable;
980+
let savedIdleTimeoutEnv;
981+
before(function () {
982+
setupTestDBPath();
983+
// Deliberately NO HARPER_BLOB_STREAM_IDLE_TIMEOUT_MS: the watchdog must be OFF unless the owning
984+
// caller arms the specific source. writeBlobWithStream is the generic primitive for every blob
985+
// write (HTTP upload, origin-fetch cache fill, replication receive); bounding a source is the
986+
// caller's job, not the primitive's. (The process-wide env override is exercised in the block above.)
987+
savedIdleTimeoutEnv = process.env.HARPER_BLOB_STREAM_IDLE_TIMEOUT_MS;
988+
delete process.env.HARPER_BLOB_STREAM_IDLE_TIMEOUT_MS;
989+
OptInTable = table({
990+
table: 'OptInTable',
991+
database: 'test',
992+
attributes: [
993+
{ name: 'id', isPrimaryKey: true },
994+
{ name: 'blob', type: 'Blob' },
995+
],
996+
});
997+
});
998+
after(function () {
999+
if (savedIdleTimeoutEnv === undefined) delete process.env.HARPER_BLOB_STREAM_IDLE_TIMEOUT_MS;
1000+
else process.env.HARPER_BLOB_STREAM_IDLE_TIMEOUT_MS = savedIdleTimeoutEnv;
1001+
});
1002+
1003+
it('does NOT destroy an unarmed idle source (a slow non-replication write is left alone)', async () => {
1004+
const stream = new PassThrough();
1005+
stream.write(Buffer.from('slow-source-no-arm')); // chunk lands, never ended, never armed
1006+
const blob = await createBlob(stream);
1007+
const info = decodeFromDatabase(() => saveBlob(blob), OptInTable.primaryStore.rootStore);
1008+
let state = 'pending';
1009+
// eslint-disable-next-line promise/catch-or-return
1010+
(info.saving ?? Promise.resolve()).then(() => (state = 'resolved')).catch(() => (state = 'rejected'));
1011+
await delay(1500);
1012+
assert.strictEqual(state, 'pending', 'an unarmed idle source must NOT be force-destroyed by the watchdog');
1013+
stream.destroy(); // clean up the deliberately-stuck write so the blob lock is released
1014+
await delay(50);
1015+
});
1016+
1017+
it('settles when the owning caller arms the source via stream.blobStreamIdleTimeoutMs', async () => {
1018+
// How the replication receive path opts in: it sets this on its PassThrough; other callers stay off.
1019+
const stream = new PassThrough();
1020+
stream.blobStreamIdleTimeoutMs = 800;
1021+
stream.on('error', () => {});
1022+
stream.write(Buffer.from('armed-but-never-finished')); // chunk lands, then idle, never ended
1023+
const blob = await createBlob(stream);
1024+
const info = decodeFromDatabase(() => saveBlob(blob), OptInTable.primaryStore.rootStore);
1025+
let state = 'pending';
1026+
// eslint-disable-next-line promise/catch-or-return
1027+
(info.saving ?? Promise.resolve()).then(() => (state = 'resolved')).catch(() => (state = 'rejected'));
1028+
await delay(2500);
1029+
assert.notStrictEqual(state, 'pending', 'an armed idle source should be destroyed within its timeout and settle');
1030+
});
1031+
});
1032+
8571033
function delay(ms) {
8581034
return new Promise((resolve) => setTimeout(resolve, ms)); // wait for audit log removal and deletion
8591035
}

‎unitTests/resources/txn-tracking.test.js‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,13 @@ describe('Txn Expiration', () => {
5757
}
5858
await Promise.race([delay(50), result]);
5959
assert(performedDBInteractions);
60-
assert.equal(trackedTxns.size, existingTxns);
60+
// Check the specific txn we started was expired and removed. Counting against
61+
// existingTxns is unreliable: other tests' transactions can expire concurrently and
62+
// shift the count underneath us during the 50ms window.
63+
assert.ok(
64+
!trackedTxns.has(lastTxn),
65+
'expected the slow transaction to have been expired and removed from trackedTxns'
66+
);
6167
});
6268
after(function () {
6369
setTxnExpiration(30000);

0 commit comments

Comments
 (0)