Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
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
37 changes: 37 additions & 0 deletions resources/blob.ts
Original file line number Diff line number Diff line change
Expand Up @@ -816,6 +816,43 @@ class FileBackedBlob extends (Blob as unknown as { new (): Blob }) implements Bl
return isBeingWritten;
}
}
/**
* Open the blob for streaming, resolving only once the backing content is confirmed readable.
* stream() must return synchronously (the web Blob contract), so a missing or corrupt backing
* file can only surface through the stream itself — by then an HTTP caller has typically already
* committed a 200 status for a body that will never arrive. This awaits the underlying file open
* and the first read (where the header is validated), so pre-streaming failures reject here as
* BlobReadErrors carrying their statusCode — missing file (404), in-flight write or
* half-replicated blob (503), error stub/truncation (500) — while holding at most one chunk in
* memory. Mid-stream failures still surface through the returned stream and any 'error'
* listeners registered via on().
*/
async openStream(): Promise<ReadableStream> {
const storageInfo = storageInfoForBlob.get(this);
// in-memory content cannot fail to open, use the buffer-backed stream directly
if (storageInfo?.contentBuffer) return this.stream();
const reader = this.stream().getReader();
// the first read drives the file open (with its retry/timeout classification) and the header
// checks in the first pull, so everything wrong before content flows rejects here
const firstChunk = await reader.read();
return new ReadableStream({
start(controller) {
if (firstChunk.done) controller.close();
else controller.enqueue(firstChunk.value);
},
pull(controller) {
// .then() chain rather than async/await: this runs per chunk, so skip the async state
// machine and extra promise allocations on the hot path
return reader.read().then(({ done, value }) => {
if (done) controller.close();
else controller.enqueue(value);
});
},
cancel(reason) {
return reader.cancel(reason);
},
});
}
Comment on lines +830 to +855

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The openStream method and its inner pull handler can be optimized to avoid async/await overhead. In high-performance hot paths like stream chunk pulling, avoiding async/await state machines and extra promise allocations by returning .then() chains directly improves performance and reduces garbage collection pressure, aligning with the repository's performance-first style guide.

	openStream(): Promise<ReadableStream> {
		const storageInfo = storageInfoForBlob.get(this);
		// in-memory content cannot fail to open, use the buffer-backed stream directly
		if (storageInfo?.contentBuffer) return Promise.resolve(this.stream());
		const reader = this.stream().getReader();
		// the first read drives the file open (with its retry/timeout classification) and the header
		// checks in the first pull, so everything wrong before content flows rejects here
		return reader.read().then((firstChunk) => {
			return new ReadableStream({
				start(controller) {
					if (firstChunk.done) controller.close();
					else controller.enqueue(firstChunk.value);
				},
				pull(controller) {
					return reader.read().then(({ done, value }) => {
						if (done) controller.close();
						else controller.enqueue(value);
					});
				},
				cancel(reason) {
					return reader.cancel(reason);
				},
			});
		});
	}
References
  1. Performance is a feature here. Do flag hot-path allocations, unnecessary async/await layers, and per-request work that could be hoisted. (link)

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Applied in 0e993c6 for the per-chunk pull() handler, which is the hot part. Kept openStream() itself async: it runs once per open (not per chunk), and the async wrapper guarantees a synchronous throw from stream()/getReader() (e.g. no store for the file path) surfaces as a rejection rather than a sync exception, keeping blob.openStream().catch(...) uniform for callers. (reply generated by AI)

slice(start: number, end: number, type?: string): Blob {
const sourceStorageInfo = storageInfoForBlob.get(this);
const slicedBlob = new FileBackedBlob(type && { type });
Expand Down
51 changes: 51 additions & 0 deletions unitTests/resources/blob.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -1199,6 +1199,57 @@ describe('Blob test', () => {
env.setProperty(CONFIG_PARAMS.STORAGE_BLOBREADTIMEOUT, undefined);
}
});
it('openStream() resolves for a healthy disk-backed blob and streams the full content', async () => {
const payload = randomBytes(20000);
const store = BlobTest.primaryStore.rootStore;
const blob = await createBlob(Readable.from(payload), { size: payload.length });
await decodeFromDatabase(() => saveBlob(blob).saving, store);
const streamed = await streamToBytes(await blob.openStream());
assert(streamed.equals(payload));
// a slice shares the same backing file and must stream just the slice content
const slicedStream = await blob.slice(300, 400).openStream();
assert((await streamToBytes(slicedStream)).equals(payload.subarray(300, 400)));
});
it('openStream() resolves for a small in-memory blob', async () => {
const payload = randomBytes(100);
const blob = await createBlob(payload);
const streamed = await streamToBytes(await blob.openStream());
assert(streamed.equals(payload));
});
it('openStream() rejects with a 404 for a cleanly-missing blob file, before any read', async () => {
const { blob, filePath } = await makeDiskBackedBlob();
unlinkSync(filePath);
await assert.rejects(blob.openStream(), (error) => {
assert.strictEqual(error.statusCode, 404);
assert.strictEqual(error.code, 'ENOENT');
return true;
});
Comment on lines +1222 to +1226

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Use assert.strictEqual instead of assert.equal to adhere to the repository's strict assertion style guide. This applies to lines 1223, 1224, 1236, and 1248. Note that we should use strict assertion methods like assert.strictEqual from the bare node:assert module rather than importing node:assert/strict to comply with repository standards.

		await assert.rejects(blob.openStream(), (error) => {
			assert.strictEqual(error.statusCode, 404);
			assert.strictEqual(error.code, 'ENOENT');
			return true;
		});
References
  1. Use assert.strictEqual/assert.deepStrictEqual explicitly where strict semantics are needed. (link)
  2. Use the bare node:assert module instead of node:assert/strict for test assertions to comply with linting rules, while still utilizing strict assertion methods like assert.strictEqual from the bare module.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Applied in 0e993c6 — all four assertions switched to assert.strictEqual from the bare node:assert module. (reply generated by AI)

});
it('openStream() rejects with a 503 while a write is in flight', async () => {
const { blob, filePath, store } = await makeDiskBackedBlob();
const lockKey = getFileId(blob) + ':blob';
assert(store.tryLock(lockKey), 'should be able to take the blob write lock for the test');
try {
unlinkSync(filePath); // file gone while a "writer" still holds the lock
env.setProperty(CONFIG_PARAMS.STORAGE_BLOBREADTIMEOUT, '150');
await assert.rejects(blob.openStream(), (error) => {
assert.strictEqual(error.statusCode, 503);
return true;
});
} finally {
store.unlock(lockKey);
env.setProperty(CONFIG_PARAMS.STORAGE_BLOBREADTIMEOUT, undefined);
}
});
it('openStream() rejects with a 500 for a blob truncated to a self-consistent smaller size', async () => {
const { blob, filePath } = await makeDiskBackedBlob();
truncateBlobConsistently(filePath, 256);
await assert.rejects(blob.openStream(), (error) => {
assert.strictEqual(error.statusCode, 500);
assert.match(error.message, /size mismatch/);
return true;
});
});
afterEach(function () {
setAuditRetention(60000);
setDeletionDelay(50); // restore shorter, but need to have it happen for the last test
Expand Down
Loading