unhandledRejection: ResponseAborted error when using SSE and App Router

[DiskCrasher]

Summary

I'm using server side events (SSE) in my route.ts module. It's working as expected upon first connection from a client, but when the client/browser refreshes or reconnects, the below error is seen on the server:

unhandledRejection: ResponseAborted
    at ServerResponse.<anonymous> (Z:\Documents\Programming\redis-app-nextjs\node_modules\next\dist\server\web\spec-extension\adapters\next-request.js:52:26)
    at Object.onceWrapper (node:events:632:28)
    at ServerResponse.emit (node:events:530:35)
    at emitCloseNT (node:_http_server:1023:10)
    at Socket.onServerResponseClose (node:_http_server:278:5)
    at Socket.emit (node:events:530:35)
    at TCP.<anonymous> (node:net:337:12)
    at TCP.callbackTrampoline (node:internal/async_hooks:130:17)

The new connection receives data as expected. It appears that the original connection is still open on the server resulting in this error. I'm not sure how (or if) there's a way to correct this, or whether it's a bug in Next.js itself. Since none of my modules are listed in the stack trace I'm leaning towards the latter.

Sample code (/src/app/api/events/route.ts):

export async function GET(req: NextRequest) {
  const responseStream = new TransformStream();
  const writer = responseStream.writable.getWriter();
  const encoder = new TextEncoder();

//  ... redacted code that periodically updates the message variable ...

    const eventData = { message: `Received "${message}"` };
    const responseMsg = `data: ${JSON.stringify(eventData)}\n\n`;
    writer.write(encoder.encode(responseMsg));

  return new Response(responseStream.readable, {
    status: 200,
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      "Connection": "keep-alive",
    },
  });

Additional information

No response

Example

No response

[PabloG6]

Figured out a fix for this?

[evakkuri]

Hi! Thanks to this reply in another discussion I was able to solve this.

You just need to add a listener to the request object like in the comment:

export async function GET(req: NextRequest) {
  const stream = new TransformStream();
  const writer = stream.writable.getWriter();
  const encoder = new TextEncoder();

  req.signal.onabort = () => {
    console.log('abort');
   (...other operations like closing connections to where your data comes from)
    writer.close();
  };

  (...other logic)

  return new NextResponse(stream.readable, {
    headers: {
      'Content-Type': 'text/event-stream',
      Connection: 'keep-alive',
      'Cache-Control': 'no-cache, no-transform',
    },
  });
}

[DiskCrasher]

Using the code above I receive the following error on page refresh/reload when writer.write() is called in the "other logic" section:

TypeError [ERR_INVALID_STATE]: Invalid state: WritableStream is closed

[joeschoe]

Yeah same. My program continues to work though. I don't know how route handlers work really but I think on refresh a second instance of the route handler is created and the first one is just hanging out erroring like crazy. So maybe the first route handler needs to be killed altogether? Not sure how to do that. I tried returning a new non SSE response but that did not work.

[DiskCrasher]

Yes, it appears a connection/object is left hanging which could be a huge issue (memory leak) for a web server with thousands of connections a day.

[evakkuri]

Sorry to hear it's not working for you. In my case I'm using this to read data from Kafka with KafkaJS and req.signal.abort solved it for me.

If you're still getting "WritableStream is closed" errors, in my mind that could mean that the process that tries to write to the stream does not close properly in the req.signal.abort block, that used to be the issue for me during development.

[DiskCrasher]

This was the issue. In my code above I'm also subscribing to Redis pub/sub events (not shown). Once I unsubscribed from those events and removed the Redis listener in addition to closing the writer stream all errors went away. Thanks!

    req.signal.onabort = async () => {
      // Close connections
      console.log("Browser disconnected. Unsubscribing from Redis and closing writer.");
      redis.removeListener("message", redisListener); // Unregister Redis event listener (created using redis.on(...))
      await writer.ready;
      await writer.close();
      await unSubscribeRedis("first", "second"); // Unsubscribe from Redis channels (calls redis.unsubscribe(...))
    };

[evakkuri]

Great to hear!

[joeschoe]

That was indeed it, just had to unsubscribe from my firestore listener, thank you kindly.

  req.signal.onabort = async () => {
    console.log("abort");
    await writer.ready;
    await writer.close();
    unsubscribe();
  };

  const unsubscribe = db
    .collection("users")
    .doc(authUser.uid)
    .collection("notifications")
    .where("read", "==", false)
    .onSnapshot((snapshot) => {...

[mozeryansky]

@evakkuri req.signal.onabort only works for me on localhost, but when deployed to vercel I never get the abort signal. Were you able to confirm this works when deployed?

[joeschoe]

Edit: I don't think what I wrote below is totally right. Aborting doesn't close the writer it just aborts it , so it's open and aborted. Closing it just gives me that TypeError [ERR_INVALID_STATE]: Invalid state: WritableStream is closed error which I'm not sure how to handle. My solution below kind of works except I think the aborted writer/stream is still running just in an aborted state, which seems like it could be a memory leak. Also you don't need to use the writer.closed method, it's just erroring because the writer is not in fact closed. The good news is writer.ready does error aborted requests, so you can use it to skip writing to the aborted stream.

I think what's happening is the route handler is attempting to return the stream but it can't because it's been aborted. Then when you reload the client it creates a new stream, but the old one is still active. Not sure how to deal with it.

Seems like the TypeError [ERR_INVALID_STATE]: Invalid state: WritableStream is closed error could be a bug there's lots of other similar issues around #55608

---

What I came up with is basically just to handle the errors. I think the req.signal.onabort = () => {.. listener is unnecessary as aborting closes the stream anyway, so I'm not going to use that. Writer has two methods .ready and .closed which you can use to catch the errors. And I'll write inside the ready block since writer has to be ready to write.

writer.closed.catch((e) => {
  console.log("--->writer.closed error", e);
});

writer.ready
    .then(() => {
      writer.write(encoder.encode(message));
    })
    .catch((e) => {
      console.log("--->writer.ready error", e);
    });

When the client aborts it throws ResponseAborted at ServerResponse... and we catch it with writer.closed no big deal makes sense. But then when I write after an abort writer.ready catches ResponseAborted at ServerResponse... which I don't really understand because we're back up and running now, the writer is ready which it shouldn't be if we're dealing with an abort because abort closes the writer. Maybe some day I'll understand but for now, since my program is working, I'll just handle the errors and move on with my life.

I looked at these docs https://developer.mozilla.org/en-US/docs/Web/API/WritableStreamDefaultWriter

[evakkuri]

In my case, I got WritableStream is closed errors during development when I did not remember to close the process that writes to the stream in the req.signal.abort block. I'm using KafkaJS to read data from Kafka and I did not close the consumer process that was writing to the stream. You probably have double-checked that already, don't know if I can help but maybe if you share some more of your code you would get more eyes on it?

[joeschoe]

Good stuff thanks, replied a lil more in depth to the thread above.