fix(web): centralize stream cancellation logic (#98)

Author: ayuhito

README.md

@@ -76,15 +76,25 @@ await writer.close();
 // When done adding entries, finalize the archive
 controller.finalize();
 
-// `readable` now contains the complete tar archive which can be piped or processed
-const tarStream = readable;
-
-// Create a tar decoder
-const decoder = createTarDecoder();
-const decodedStream = tarStream.pipeThrough(decoder);
+// Pipe the archive right into a decoder
+const decodedStream = readable.pipeThrough(createTarDecoder());
 for await (const entry of decodedStream) {
 	console.log(`Decoded: ${entry.header.name}`);
-	// Process `entry.body` stream as needed
+
+	const shouldSkip = entry.header.name.endsWith(".md");
+	if (shouldSkip) {
+		// You MUST drain the body with cancel() to proceed to the next entry or read it fully,
+		// otherwise the stream will stall.
+		await entry.body.cancel();
+		continue;
+	}
+
+	const reader = entry.body.getReader();
+	while (true) {
+		const { done, value } = await reader.read();
+		if (done) break;
+		processChunk(value);
+	}
 }
 ```
 

REFERENCE.md

@@ -75,7 +75,21 @@ const entriesStream = tarStream.pipeThrough(decoder);
 
 for await (const entry of entriesStream) {
   console.log(`Entry: ${entry.header.name}`);
-  // Process entry.body stream as needed
+
+  const shouldSkip = entry.header.name.endsWith('.md');
+  if (shouldSkip) {
+  	// You MUST drain the body with cancel() to proceed to the next entry or read it fully,
+		// otherwise the stream will stall.
+    await entry.body.cancel();
+    continue;
+  }
+
+  const reader = entry.body.getReader();
+  while (true) {
+    const { done, value } = await reader.read();
+    if (done) break;
+    processChunk(value);
+  }
 }
 ```
 

src/fs/unpack.ts

@@ -8,22 +8,6 @@ import { createFileSink } from "./file-sink";
 import { createPathCache } from "./path-cache";
 import type { UnpackOptionsFS } from "./types";
 
-/**
- * Helper to discard body data for filtered entries.
- * Returns true if complete, false if more data is needed.
- */
-function discardBody(unpacker: ReturnType<typeof createUnpacker>): boolean {
-	while (!unpacker.isBodyComplete()) {
-		// Callback always returns true to discard data.
-		unpacker.streamBody(() => true);
-		if (!unpacker.isBodyComplete()) return false;
-	}
-
-	while (!unpacker.skipPadding()) return false;
-
-	return true;
-}
-
 /**
  * Extract a tar archive to a directory.
  *
@@ -77,7 +61,7 @@ export function unpackTar(
 
 				// If we're in the middle of discarding a body, continue it
 				if (needsDiscardBody) {
-					if (!discardBody(unpacker)) {
+					if (!unpacker.skipEntry()) {
 						// Still need more data
 						cb();
 						return;
@@ -139,7 +123,7 @@ export function unpackTar(
 					const transformedHeader = transformHeader(header, options);
 					// Filtered out.
 					if (!transformedHeader) {
-						if (!discardBody(unpacker)) {
+						if (!unpacker.skipEntry()) {
 							needsDiscardBody = true;
 							cb();
 							return;
@@ -198,7 +182,7 @@ export function unpackTar(
 						opQueue.add(() => fileStream.end());
 					} else {
 						// No body data or already handled.
-						if (!discardBody(unpacker)) {
+						if (!unpacker.skipEntry()) {
 							needsDiscardBody = true;
 							cb();
 							return;

src/tar/unpacker.ts

@@ -199,25 +199,21 @@ export function createUnpacker(options: DecoderOptions = {}) {
 			return true;
 		},
 
-		hasEnded(): boolean {
-			return ended;
-		},
+		/**
+		 * Discards the current entry's body and its padding.
+		 *
+		 * Returns true when the full entry has been skipped, false if more
+		 * data is required.
+		 */
+		skipEntry(): boolean {
+			if (state !== STATE_BODY || !currentEntry) return true;
 
-		/** Check if unpacker likely has enough data to make progress without waiting. */
-		canContinueProcessing(): boolean {
-			// If we're waiting for a header, we need at least BLOCK_SIZE bytes
-			if (state === STATE_HEADER) return available() >= BLOCK_SIZE;
-
-			// If we're in body state, we can process if we have any data or need to skip padding
-			if (state === STATE_BODY && currentEntry) {
-				return currentEntry.remaining === 0
-					? // If body is complete, we just need to skip padding
-						available() >= currentEntry.padding
-					: // If body isn't complete, any data is useful
-						available() > 0;
+			while (!unpacker.isBodyComplete()) {
+				const fed = unpacker.streamBody(() => true);
+				if (fed === 0) return false;
 			}
 
-			return true;
+			return unpacker.skipPadding();
 		},
 
 		validateEOF() {

src/web/unpack.ts

@@ -16,7 +16,21 @@ import type { ParsedTarEntry } from "./types";
  *
  * for await (const entry of entriesStream) {
  *  console.log(`Entry: ${entry.header.name}`);
- *  // Process entry.body stream as needed
+ *
+ *  const shouldSkip = entry.header.name.endsWith('.md');
+ *  if (shouldSkip) {
+ *   // You MUST drain the body with cancel() to proceed to the next entry or read it fully,
+ * 	 // otherwise the stream will stall.
+ *   await entry.body.cancel();
+ *   continue;
+ *  }
+ *
+ *  const reader = entry.body.getReader();
+ *  while (true) {
+ * 	 const { done, value } = await reader.read();
+ * 	 if (done) break;
+ * 	 processChunk(value);
+ *  }
  * }
  */
 export function createTarDecoder(
@@ -65,6 +79,8 @@ export function createTarDecoder(
 			while (!controllerTerminated) {
 				// Look for the next header.
 				if (!bodyController) {
+					if (!unpacker.skipEntry()) break;
+
 					// Respect backpressure on the main stream.
 					if (!force && (controller.desiredSize ?? 0) < 0) break;
 
@@ -83,6 +99,7 @@ export function createTarDecoder(
 						// If the consumer cancels this body stream, clear the controller.
 						cancel: () => {
 							bodyController = null;
+							pump(controller);
 						},
 					});
 

tests/web/extract.test.ts

@@ -158,6 +158,38 @@ describe("createTarDecoder", () => {
 		expect(names).toEqual(["file.txt", "dir/", "dir/nested.txt"]);
 	});
 
+	it("iterate over headers by cancelling", async () => {
+		const archive = await createBaseArchive([
+			{ header: { name: "dir/", type: "directory", size: 0 } },
+			{
+				header: { name: "dir/file.txt", type: "file", size: 5 },
+				body: "hello",
+			},
+			{
+				header: { name: "dir/empty.txt", type: "file", size: 0 },
+				body: new Uint8Array(),
+			},
+		]);
+
+		const stream = new ReadableStream({
+			start(controller) {
+				controller.enqueue(archive);
+				controller.close();
+			},
+		});
+
+		const decoder = createTarDecoder();
+		const entryStream = stream.pipeThrough(decoder);
+
+		const names: string[] = [];
+		for await (const entry of entryStream) {
+			names.push(entry.header.name);
+			await entry.body?.cancel();
+		}
+
+		expect(names).toEqual(["dir/", "dir/file.txt", "dir/empty.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: "" },

tests/web/fixtures/index.ts

@@ -38,3 +38,7 @@ export const LODASH_TGZ = join(__dirname, "lodash-4.17.21.tgz");
 export const NEXT_SWC_TGZ = join(__dirname, "next-swc-linux-14.2.15.tgz");
 export const SHARP_TGZ = join(__dirname, "sharp-0.33.5.tgz");
 export const ELECTRON_TGZ = join(__dirname, "electron-33.0.2.tgz");
+export const NODE_V25_DARWIN_ARM64_TAR_GZ = join(
+	__dirname,
+	"node-v25.2.0-darwin-arm64.tar.gz",
+);

tests/web/fixtures/node-v25.2.0-darwin-arm64.tar.gz

@@ -2,11 +2,18 @@ import * as fs from "node:fs";
 import { describe, expect, it } from "vitest";
 import {
 	createGzipDecoder,
+	createTarDecoder,
 	type ParsedTarEntryWithData,
 	unpackTar,
 } from "../../src/web";
 import { streamToBuffer } from "../../src/web/stream-utils";
-import { ELECTRON_TGZ, LODASH_TGZ, NEXT_SWC_TGZ, SHARP_TGZ } from "./fixtures";
+import {
+	ELECTRON_TGZ,
+	LODASH_TGZ,
+	NEXT_SWC_TGZ,
+	NODE_V25_DARWIN_ARM64_TAR_GZ,
+	SHARP_TGZ,
+} from "./fixtures";
 
 async function extractTgz(filePath: string): Promise<ParsedTarEntryWithData[]> {
 	// @ts-expect-error ReadableStream.from is supported in Node tests
@@ -32,7 +39,7 @@ describe("real world examples", () => {
 			(e) => e.header.name === "package/README.md",
 		);
 		expect(readmeEntry).toBeDefined();
-		expect(readmeEntry?.data.length).toBeGreaterThan(1000);
+		expect(readmeEntry?.data?.length).toBe(1107);
 	});
 
 	it(
@@ -51,7 +58,7 @@ describe("real world examples", () => {
 				(e) => e.header.name === "package/next-swc.linux-x64-gnu.node",
 			);
 			expect(binaryEntry).toBeDefined();
-			expect(binaryEntry?.data.length).toBeGreaterThan(30 * 1024 * 1024); // > 30MB
+			expect(binaryEntry?.data?.length).toBe(131406240);
 
 			// Verify package.json exists
 			expect(
@@ -72,14 +79,14 @@ describe("real world examples", () => {
 		const cppFiles = entries.filter(
 			(e) => e.header.name.endsWith(".cc") || e.header.name.endsWith(".h"),
 		);
-		expect(cppFiles.length).toBeGreaterThan(5);
+		expect(cppFiles.length).toBe(13);
 
 		// Verify a specific C++ file has substantial content
 		const sharpCcEntry = entries.find(
 			(e) => e.header.name === "package/src/sharp.cc",
 		);
 		expect(sharpCcEntry).toBeDefined();
-		expect(sharpCcEntry?.data.length).toBeGreaterThan(1000);
+		expect(sharpCcEntry?.data?.length).toBe(1465);
 	});
 
 	it("extracts a package with installation scripts (electron)", async () => {
@@ -104,6 +111,58 @@ describe("real world examples", () => {
 			(e) => e.header.name === "package/electron.d.ts",
 		);
 		expect(electronDtsEntry).toBeDefined();
-		expect(electronDtsEntry?.data.length).toBeGreaterThan(1000);
+		expect(electronDtsEntry?.data?.length).toBe(987499);
+	});
+
+	it("extracts a Node.js release tarball", async () => {
+		// @ts-expect-error ReadableStream.from is supported in Node tests
+		const fileStream = ReadableStream.from(
+			fs.createReadStream(NODE_V25_DARWIN_ARM64_TAR_GZ),
+		);
+
+		const entryStream = fileStream
+			.pipeThrough(createGzipDecoder())
+			.pipeThrough(createTarDecoder());
+
+		let count = 0;
+		let lastEntry = "";
+		let totalBytes = 0;
+		for await (const entry of entryStream) {
+			count++;
+			lastEntry = entry.header.name;
+
+			const reader = entry.body.getReader();
+			while (true) {
+				const { done, value } = await reader.read();
+				if (done) break;
+				if (value) totalBytes += value.length;
+			}
+		}
+
+		expect(count).toBe(5986);
+		expect(lastEntry).toBe("node-v25.2.0-darwin-arm64/bin/npm");
+		expect(totalBytes).toBe(200544142);
+	});
+
+	it("streams entries from the Node.js release tarball", async () => {
+		// @ts-expect-error ReadableStream.from is supported in Node tests
+		const fileStream = ReadableStream.from(
+			fs.createReadStream(NODE_V25_DARWIN_ARM64_TAR_GZ),
+		);
+
+		const entryStream = fileStream
+			.pipeThrough(createGzipDecoder())
+			.pipeThrough(createTarDecoder());
+
+		let count = 0;
+		let lastEntry = "";
+		for await (const entry of entryStream) {
+			count++;
+			lastEntry = entry.header.name;
+			await entry.body.cancel();
+		}
+
+		expect(count).toBe(5986);
+		expect(lastEntry).toBe("node-v25.2.0-darwin-arm64/bin/npm");
 	});
 });