mary.my.id
streaming 7-Zip archive extractor
fix: terminate the Blob reader stream once the slice is delivered
author
Mary
date
(Jun 3, 2026, 8:24 PM +0700)
commit
9d07d503
9d07d5036ebc9249b4b9f91a68080e1bfbbfa5a2
parent
b5abaa0d
b5abaa0d7dbfdf80dce7ed503a581114de380e71
+100
-3
+36
-1
lib/reader/common.ts
+36
-1
lib/reader/common.ts
···
58
58
const endOffset = length !== undefined ? offset + length : totalLength;
59
59
60
60
const slice = blob.slice(offset, endOffset);
61
-
return slice.stream();
61
+
const expected = slice.size;
62
+
63
+
// Bun's Blob-backed slice streams can fail to emit a final
64
+
// `{ done: true }` (empty slices never close; others stall a
65
+
// read-until-close consumer), so re-emit the bytes through a stream
66
+
// we close ourselves once the slice length has been delivered,
67
+
// without issuing the trailing read that would block.
68
+
const source = slice.stream().getReader();
69
+
let delivered = 0;
70
+
71
+
return new ReadableStream<Uint8Array<ArrayBuffer>>({
72
+
type: 'bytes',
73
+
async pull(rawController) {
74
+
const controller = rawController as ReadableByteStreamController;
75
+
76
+
if (delivered >= expected) {
77
+
controller.close();
78
+
source.cancel().catch(() => {});
79
+
return;
80
+
}
81
+
82
+
const { done, value } = await source.read();
83
+
if (done) {
84
+
controller.close();
85
+
return;
86
+
}
87
+
88
+
if (value.byteLength > 0) {
89
+
delivered += value.byteLength;
90
+
controller.enqueue(value as Uint8Array<ArrayBuffer>);
91
+
}
92
+
},
93
+
cancel(reason) {
94
+
return source.cancel(reason);
95
+
},
96
+
}, { highWaterMark: 64 * 1024 });
62
97
},
63
98
};
64
99
}
+64
-2
tests/reader.test.ts
+64
-2
tests/reader.test.ts
···
3
3
import { ReaderError, TruncatedArchiveError } from '../lib/errors.ts';
4
4
import { concatReaders } from '../lib/reader/concat.ts';
5
5
import { fromFetch, fromFile, fromUint8Array } from '../lib/reader/common.ts';
6
-
import { readBytes } from '../lib/utils/buffer.ts';
6
+
import { concat, readBytes } from '../lib/utils/buffer.ts';
7
7
8
8
const sample = new Uint8Array(256);
9
9
for (let i = 0; i < sample.length; i++) {
···
60
60
});
61
61
62
62
Deno.test('readBytes: BYOB path used when fromFile delivers a byte stream and size >= 4096', async () => {
63
-
// fromFile delegates to Blob.stream(), which is byte-typed and therefore
63
+
// fromFile re-emits Blob.stream() through a byte-typed stream, so it still
64
64
// supports BYOB readers. readBytes prefers BYOB at size>=4096, so this
65
65
// exercises the branch fromUint8Array can't reach.
66
66
const big = new Uint8Array(8192);
···
70
70
const reader = fromFile(new Blob([big]));
71
71
const got = await readBytes(reader, 0, big.length);
72
72
assertEquals(got, big);
73
+
});
74
+
75
+
// a Blob whose slice streams deliver their bytes but never emit a final
76
+
// `{ done: true }` — the Bun behaviour fromFile has to defend against.
77
+
const neverClosingBlob = (payload: Uint8Array<ArrayBuffer>): Blob =>
78
+
({
79
+
size: payload.byteLength,
80
+
slice(start = 0, end = payload.byteLength) {
81
+
const part = payload.subarray(start, end);
82
+
return {
83
+
size: part.byteLength,
84
+
stream: () =>
85
+
new ReadableStream<Uint8Array<ArrayBuffer>>({
86
+
start(controller) {
87
+
if (part.byteLength > 0) {
88
+
// fresh buffer per chunk, like a real Blob.stream(); a byte
89
+
// stream detaches whatever it enqueues.
90
+
controller.enqueue(part.slice() as Uint8Array<ArrayBuffer>);
91
+
}
92
+
// deliberately never close()
93
+
},
94
+
}),
95
+
};
96
+
},
97
+
}) as unknown as Blob;
98
+
99
+
// drains a stream to completion, failing fast instead of hanging if it never
100
+
// closes (the symptom this guards against).
101
+
const collectBounded = async (stream: ReadableStream<Uint8Array>): Promise<Uint8Array<ArrayBuffer>> => {
102
+
let timer: ReturnType<typeof setTimeout> | undefined;
103
+
const timeout = new Promise<never>((_, reject) => {
104
+
timer = setTimeout(() => reject(new Error('stream did not close within 5s')), 5000);
105
+
});
106
+
const drain = (async () => {
107
+
const chunks: Uint8Array[] = [];
108
+
for await (const chunk of stream) {
109
+
chunks.push(chunk);
110
+
}
111
+
return concat(chunks);
112
+
})();
113
+
try {
114
+
return await Promise.race([drain, timeout]);
115
+
} finally {
116
+
clearTimeout(timer);
117
+
}
118
+
};
119
+
120
+
Deno.test('fromFile: a non-empty range ends even when the source never closes', async () => {
121
+
const payload = new Uint8Array(1000);
122
+
for (let i = 0; i < payload.length; i++) {
123
+
payload[i] = (i * 3) & 0xff;
124
+
}
125
+
const reader = fromFile(neverClosingBlob(payload));
126
+
const got = await collectBounded(await reader.read(100, 300));
127
+
assertEquals(got, payload.subarray(100, 400));
128
+
});
129
+
130
+
Deno.test('fromFile: an empty range ends with no bytes', async () => {
131
+
const payload = new Uint8Array(1000);
132
+
const reader = fromFile(neverClosingBlob(payload));
133
+
const got = await collectBounded(await reader.read(100, 0));
134
+
assertEquals(got.byteLength, 0);
73
135
});
74
136
75
137
Deno.test('concatReaders: single reader passes through unchanged', () => {