fix: Add automatic response cleanup via AbortSignal (opennextjs#952)

* feat(dev-overrides): Add automatic response cleanup via onClose callback

changeset

changeset

* lint

* fix changeset

* review

* refactor cloudflare part

* review

* lint

* add comment

Author: sommeeeer

.changeset/eleven-moose-heal.md

@@ -0,0 +1,7 @@
+---
+"@opennextjs/aws": patch
+---
+
+fix: Add automatic response cleanup via AbortSignal
+
+These changes will make `request.signal.onabort` work in route handlers for `node`, `cloudflare-node` and `express-dev` wrappers.

packages/open-next/src/http/openNextResponse.ts

@@ -84,6 +84,13 @@ export class OpenNextNodeResponse extends Transform implements ServerResponse {
     ) {
       this.statusCode = statusCode;
     }
+
+    // https://github.com/vercel/next.js/blob/ea08bf2/packages/next/src/server/web/spec-extension/adapters/next-request.ts#L46-L54
+    // We want to destroy this response when the original response/request is closed. (i.e when the client disconnects)
+    // This is to support `request.signal.onabort` in route handlers
+    streamCreator?.abortSignal?.addEventListener("abort", () => {
+      this.destroy();
+    });
   }
 
   // Necessary for next 12

packages/open-next/src/overrides/wrappers/cloudflare-node.ts

@@ -16,9 +16,9 @@ const handler: WrapperHandler<InternalEvent, InternalResult> =
     request: Request,
     env: Record<string, string>,
     ctx: any,
+    abortSignal: AbortSignal,
   ): Promise<Response> => {
     globalThis.process = process;
-
     // Set the environment variables
     // Cloudflare suggests to not override the process.env object but instead apply the values to it
     for (const [key, value] of Object.entries(env)) {
@@ -66,6 +66,10 @@ const handler: WrapperHandler<InternalEvent, InternalResult> =
 
         return Writable.fromWeb(writable);
       },
+      // This is for passing along the original abort signal from the initial Request you retrieve in your worker
+      // Ensures that the response we pass to NextServer is aborted if the request is aborted
+      // By doing this `request.signal.onabort` will work in route handlers
+      abortSignal: abortSignal,
     };
 
     ctx.waitUntil(

packages/open-next/src/overrides/wrappers/express-dev.ts

@@ -41,6 +41,9 @@ const wrapper: WrapperHandler = async (handler, converter) => {
       req.headers["x-forwarded-proto"] = req.protocol;
     }
     const internalEvent = await converter.convertFrom(req);
+
+    const abortController = new AbortController();
+
     const streamCreator: StreamCreator = {
       writeHeaders: (prelude) => {
         res.setHeader("Set-Cookie", prelude.cookies);
@@ -49,7 +52,13 @@ const wrapper: WrapperHandler = async (handler, converter) => {
         return res;
       },
       onFinish: () => {},
+      abortSignal: abortController.signal,
     };
+
+    res.on("close", () => {
+      abortController.abort();
+    });
+
     await handler(internalEvent, { streamCreator });
   });
 

packages/open-next/src/overrides/wrappers/node.ts

@@ -8,21 +8,32 @@ import { debug, error } from "../../adapters/logger";
 const wrapper: WrapperHandler = async (handler, converter) => {
   const server = createServer(async (req, res) => {
     const internalEvent = await converter.convertFrom(req);
+
+    const abortController = new AbortController();
+
     const streamCreator: StreamCreator = {
       writeHeaders: (prelude) => {
         res.setHeader("Set-Cookie", prelude.cookies);
         res.writeHead(prelude.statusCode, prelude.headers);
         res.flushHeaders();
         return res;
       },
+      abortSignal: abortController.signal,
     };
+
+    res.on("close", () => {
+      abortController.abort();
+    });
+
     if (internalEvent.rawPath === "/__health") {
       res.writeHead(200, {
         "Content-Type": "text/plain",
       });
       res.end("OK");
     } else {
-      await handler(internalEvent, { streamCreator });
+      await handler(internalEvent, {
+        streamCreator,
+      });
     }
   });
 

packages/open-next/src/types/open-next.ts

@@ -49,6 +49,7 @@ export interface StreamCreator {
   // Just to fix an issue with aws lambda streaming with empty body
   onWrite?: () => void;
   onFinish?: (length: number) => void;
+  abortSignal?: AbortSignal;
 }
 
 export type WaitUntil = (promise: Promise<void>) => void;