fix: cancel the body stream when not consumed (#789)

Author: vicb

.changeset/eager-steaks-crash.md

@@ -0,0 +1,5 @@
+---
+"@opennextjs/cloudflare": patch
+---
+
+cancel the body stream when unconsumed

packages/cloudflare/src/api/durable-objects/queue.ts

@@ -111,14 +111,15 @@ export class DOQueueHandler extends DurableObject<CloudflareEnv> {
 	}
 
 	async executeRevalidation(msg: QueueMessage) {
+		let response: Response | undefined;
 		try {
 			debug(`Revalidating ${msg.MessageBody.host}${msg.MessageBody.url}`);
 			const {
 				MessageBody: { host, url },
 			} = msg;
 			const protocol = host.includes("localhost") ? "http" : "https";
 
-			const response = await this.service.fetch(`${protocol}://${host}${url}`, {
+			response = await this.service.fetch(`${protocol}://${host}${url}`, {
 				method: "HEAD",
 				headers: {
 					// This is defined during build
@@ -177,6 +178,12 @@ export class DOQueueHandler extends DurableObject<CloudflareEnv> {
 			error(e);
 		} finally {
 			this.ongoingRevalidations.delete(msg.MessageDeduplicationId);
+			// Cancel the stream when it has not been consumed
+			try {
+				await response?.body?.cancel();
+			} catch {
+				// Ignore errors when the stream was actually consumed
+			}
 		}
 	}
 

packages/cloudflare/src/api/overrides/asset-resolver/index.ts

@@ -36,6 +36,7 @@ const resolver: AssetResolver = {
 		});
 
 		if (response.status === 404) {
+			await response.body?.cancel();
 			return undefined;
 		}
 

packages/cloudflare/src/api/overrides/incremental-cache/static-assets-incremental-cache.ts

@@ -34,7 +34,10 @@ class StaticAssetsIncrementalCache implements IncrementalCache {
 
 		try {
 			const response = await assets.fetch(this.getAssetUrl(key, cacheType));
-			if (!response.ok) return null;
+			if (!response.ok) {
+				await response.body?.cancel();
+				return null;
+			}
 
 			return {
 				value: await response.json(),

packages/cloudflare/src/api/overrides/internal.ts

@@ -54,8 +54,9 @@ export async function internalPurgeCacheByTags(env: CloudflareEnv, tags: string[
 		return "missing-credentials";
 	}
 
+	let response: Response | undefined;
 	try {
-		const response = await fetch(
+		response = await fetch(
 			`https://api.cloudflare.com/client/v4/zones/${env.CACHE_PURGE_ZONE_ID}/purge_cache`,
 			{
 				headers: {
@@ -90,5 +91,12 @@ export async function internalPurgeCacheByTags(env: CloudflareEnv, tags: string[
 	} catch (error) {
 		console.error("Error purging cache by tags:", error);
 		return "purge-failed";
+	} finally {
+		// Cancel the stream when it has not been consumed
+		try {
+			await response?.body?.cancel();
+		} catch {
+			// Ignore errors when the stream was actually consumed
+		}
 	}
 }

packages/cloudflare/src/api/overrides/queue/memory-queue.ts

@@ -29,13 +29,15 @@ export class MemoryQueue implements Queue {
 
 		this.revalidatedPaths.add(MessageDeduplicationId);
 
+		let response: Response | undefined;
+
 		try {
 			const protocol = host.includes("localhost") ? "http" : "https";
 
 			// TODO: Drop the import - https://github.com/opennextjs/opennextjs-cloudflare/issues/361
 			// @ts-ignore
 			const manifest = await import("./.next/prerender-manifest.json");
-			const response = await service.fetch(`${protocol}://${host}${url}`, {
+			response = await service.fetch(`${protocol}://${host}${url}`, {
 				method: "HEAD",
 				headers: {
 					"x-prerender-revalidate": manifest.preview.previewModeId,
@@ -54,6 +56,12 @@ export class MemoryQueue implements Queue {
 			error(e);
 		} finally {
 			this.revalidatedPaths.delete(MessageDeduplicationId);
+			// Cancel the stream when it has not been consumed
+			try {
+				await response?.body?.cancel();
+			} catch {
+				// Ignore errors when the stream was actually consumed
+			}
 		}
 	}
 }