fix(web): decoder should not wait for last chunk (#96)

* fix(web): decoder should not wait for last chunk

* style: format

Author: ayuhito

src/web/unpack.ts

@@ -66,7 +66,7 @@ export function createTarDecoder(
 				// Look for the next header.
 				if (!bodyController) {
 					// Respect backpressure on the main stream.
-					if (!force && (controller.desiredSize ?? 1) <= 0) break;
+					if (!force && (controller.desiredSize ?? 0) < 0) break;
 
 					const header = unpacker.readHeader();
 					if (header === null) break; // Not enough data
@@ -126,16 +126,18 @@ export function createTarDecoder(
 					try {
 						// biome-ignore lint/style/noNonNullAssertion: body processing state.
 						bodyController!.enqueue(chunk);
+
 						// If the body stream's buffer is full, signal to pause the pump.
 						// biome-ignore lint/style/noNonNullAssertion: body processing state.
 						if ((bodyController!.desiredSize ?? 1) <= 0) shouldPause = true;
 					} catch {
 						return true; // Body errored or closed, discard
 					}
-					return true; // Continue processing unpacker's buffer
+
+					return true;
 				});
 
-				// No buffered data was available; wait for the next chunk to resume pumping.
+				// No buffered data is available. Wait for the next chunk to resume pumping.
 				if (fed === 0) break;
 
 				// Check again if the body is now complete after streaming.
@@ -160,46 +162,51 @@ export function createTarDecoder(
 		}
 	};
 
-	return new TransformStream<Uint8Array, ParsedTarEntry>({
-		transform(chunk, controller) {
-			try {
-				// In strict mode, ensure EOF blocks are all zeroes.
-				if (eofReached && options.strict && chunk.some((byte) => byte !== 0))
-					throw new Error("Invalid EOF.");
-
-				// Write incoming data to the unpacker.
-				unpacker.write(chunk);
-				pump(controller);
-			} catch (error) {
-				abortAll(error, controller);
-				throw error;
-			}
-		},
+	return new TransformStream<Uint8Array, ParsedTarEntry>(
+		{
+			transform(chunk, controller) {
+				try {
+					// In strict mode, ensure EOF blocks are all zeroes.
+					if (eofReached && options.strict && chunk.some((byte) => byte !== 0))
+						throw new Error("Invalid EOF.");
+
+					// Write incoming data to the unpacker.
+					unpacker.write(chunk);
+					pump(controller);
+				} catch (error) {
+					abortAll(error, controller);
+					throw error;
+				}
+			},
 
-		flush(controller) {
-			try {
-				unpacker.end();
-				pump(controller, true); // Force pump for remaining data
+			flush(controller) {
+				try {
+					unpacker.end();
+					pump(controller, true); // Force pump for remaining data
+
+					// If a bodyController still exists, the archive was truncated mid-file.
+					if (bodyController) {
+						if (options.strict) throw new Error("Tar archive is truncated.");
 
-				// If a bodyController still exists, the archive was truncated mid-file.
-				if (bodyController) {
-					if (options.strict) {
-						throw new Error("Tar archive is truncated.");
+						// In non-strict mode, just close the partial stream.
+						try {
+							bodyController.close();
+						} catch {}
+						bodyController = null;
 					}
-					// In non-strict mode, just close the partial stream.
-					try {
-						bodyController.close();
-					} catch {}
-					bodyController = null;
-				}
 
-				unpacker.validateEOF();
+					unpacker.validateEOF();
 
-				if (!controllerTerminated) controller.terminate();
-			} catch (error) {
-				abortAll(error, controller);
-				throw error;
-			}
+					if (!controllerTerminated) controller.terminate();
+				} catch (error) {
+					abortAll(error, controller);
+					throw error;
+				}
+			},
+		},
+		undefined,
+		{
+			highWaterMark: 1,
 		},
-	});
+	);
 }

tests/web/extract.test.ts

@@ -19,6 +19,7 @@ const createBaseArchive = (
 ): Promise<Uint8Array> => {
 	return packTar(entries);
 };
+
 describe("unpackTar", () => {
 	it("extracts a single file tar", async () => {
 		const buffer = await fs.readFile(ONE_FILE_TAR);
@@ -97,6 +98,66 @@ describe("unpackTar", () => {
 });
 
 describe("createTarDecoder", () => {
+	it("streams entries as data arrives", async () => {
+		const archive = await createBaseArchive([
+			{ header: { name: "file.txt", type: "file", size: 5 }, body: "hello" },
+			{ header: { name: "dir/", type: "directory", size: 0 } },
+			{
+				header: { name: "dir/nested.txt", type: "file", size: 3 },
+				body: new Uint8Array([97, 98, 99]),
+			},
+		]);
+
+		const decoder = createTarDecoder();
+		const reader = decoder.readable.getReader();
+		const writer = decoder.writable.getWriter();
+
+		const splitPoint = Math.floor((archive.length * 2) / 3);
+		const firstPart = archive.subarray(0, splitPoint);
+
+		const firstResultPromise = Promise.race([
+			reader.read(),
+			new Promise<never>((_, reject) =>
+				setTimeout(
+					() => reject(new Error("Timed out waiting for first entry")),
+					100,
+				),
+			),
+		]);
+
+		await writer.write(firstPart);
+		const firstResult = await firstResultPromise;
+		expect(firstResult.done).toBe(false);
+		const firstEntry = firstResult.value;
+		if (!firstEntry) throw new Error("Expected first entry");
+		expect(firstEntry.header.name).toBe("file.txt");
+
+		await firstEntry.body.cancel();
+
+		const names = [firstEntry.header.name];
+
+		for (let i = 0; i < 2; i++) {
+			const nextResult = await Promise.race([
+				reader.read(),
+				new Promise<never>((_, reject) =>
+					setTimeout(
+						() => reject(new Error("Timed out waiting for next entry")),
+						100,
+					),
+				),
+			]);
+			expect(nextResult.done).toBe(false);
+			const entry = nextResult.value;
+			if (!entry) throw new Error("Expected streamed entry");
+
+			names.push(entry.header.name);
+
+			if (entry.header.size > 0) await entry.body.cancel();
+		}
+
+		expect(names).toEqual(["file.txt", "dir/", "dir/nested.txt"]);
+	});
+
 	it("rejects a stream with an invalid checksum in strict mode", async () => {
 		const archive = await createBaseArchive([
 			{ header: { name: "test.txt", type: "file", size: 0 }, body: "" },