fix(jsx): Add catch to async function's promise

[mwilkins91]

The .then(async function()) returns a new promise, that the above .catch() does not catch, resulting in rapid refreshing of the page to cause a fatal uncatchable error. Adding this second .catch fixes that.

The problem:

If you're streaming a response, via renderToReadableStream from hono/jsx/streaming, and you rapidly refresh your page, you can trigger an error:

TypeError: Invalid state: Controller is already closed
    at new NodeError (node:internal/errors:405:5)
    at ReadableStreamDefaultController.enqueue (node:internal/webstreams/readablestream:1038:13)
    at /Users/markwilkins/projects/wilkins-software-nx-v2/node_modules/hono/dist/cjs/jsx/streaming.js:137:24
    at processTicksAndRejections (node:internal/process/task_queues:95:5)

which is uncatchable as far as I can tell.

The fix:

Since the catch block above is applied to the promise variable, but not to the promise returned by the async function in the .then(), the call to enqueue on line 150 can fail due to a closed connection.

Fixing it is as simple as adding another .catch, as funny as it looks at first glance.

[usualoma]

Hi @mwilkins91

Thanks for making the pull request. There are indeed some issues you point out, which I think will be fixed in this PR. Please wait a bit so I can look at the whole thing and comment.

[mwilkins91]

@usualoma Thanks! Let me know if I can help.

[usualoma]

Hi @mwilkins91 !

I consider the following

• Errors in controller.enqueue() are no longer expected to be recovered, so I think we should handle them with an outer try clause.
• If the client disconnects first, it is generally not a fatal error on the server side, so it would be nice to be able to ignore the log.
  • On node-server, this is output in info : https://github.com/honojs/node-server/blob/e76c687178f403c524b13651c16580f66b7edeb4/src/listener.ts#L26
  • Elsewhere in the project, it can be controlled by onError.

    hono/deno_dist/helper/streaming/sse.ts

    Line 41 in e428a05

    onError?: (e: Error, stream: SSEStreamingApi) => Promise<void>

Given these assumptions, I would suggest the following patches to this matter. If you have no objection, I would be happy to have this change incorporated into the PR.

diff --git a/src/jsx/streaming.test.tsx b/src/jsx/streaming.test.tsx
index 0c9ba6e1..c5cd5579 100644
--- a/src/jsx/streaming.test.tsx
+++ b/src/jsx/streaming.test.tsx
@@ -390,10 +390,12 @@ d.replaceWith(c.content)
       return <p>{content}</p>
     }
 
+    const onError = vi.fn()
     const stream = renderToReadableStream(
       <Suspense fallback={<p>Loading...</p>}>
         <Content />
-      </Suspense>
+      </Suspense>,
+      onError
     )
 
     const chunks = []
@@ -402,6 +404,8 @@ d.replaceWith(c.content)
       chunks.push(textDecoder.decode(chunk))
     }
 
+    expect(onError).toBeCalledTimes(1)
+
     expect(chunks).toEqual([
       `<template id="H:${suspenseCounter}"></template><p>Loading...</p><!--/$-->`,
       '',
@@ -412,6 +416,55 @@ d.replaceWith(c.content)
     )
   })
 
+  it('closed()', async () => {
+    const Content = async () => {
+      await new Promise<void>((resolve) =>
+        setTimeout(() => {
+          vi.spyOn(ReadableStreamDefaultController.prototype, 'enqueue').mockImplementation(() => {
+            throw new Error('closed')
+          })
+          resolve()
+        }, 10)
+      )
+      return <p>content</p>
+    }
+
+    const onError = vi.fn()
+    const stream = renderToReadableStream(
+      <>
+        <Suspense fallback={<p>Loading...</p>}>
+          <Content />
+        </Suspense>
+        <Suspense fallback={<p>Loading...</p>}>
+          <Content />
+        </Suspense>
+      </>,
+      onError
+    )
+
+    const chunks = []
+    const textDecoder = new TextDecoder()
+    for await (const chunk of stream as any) {
+      chunks.push(textDecoder.decode(chunk))
+    }
+
+    expect(onError).toBeCalledTimes(1)
+
+    expect(chunks).toEqual([
+      `<template id="H:${suspenseCounter}"></template><p>Loading...</p><!--/$--><template id="H:${
+        suspenseCounter + 1
+      }"></template><p>Loading...</p><!--/$-->`,
+    ])
+
+    expect(replacementResult(`<html><body>${chunks.join('')}</body></html>`)).toEqual(
+      '<p>Loading...</p><!--/$--><p>Loading...</p><!--/$-->'
+    )
+
+    suspenseCounter++
+    await new Promise((resolve) => setTimeout(resolve, 10))
+    vi.restoreAllMocks()
+  })
+
   it('Multiple "await" call', async () => {
     const delayedContent = new Promise<HtmlEscapedString>((resolve) =>
       setTimeout(() => resolve(<h1>Hello</h1>), 10)
diff --git a/src/jsx/streaming.ts b/src/jsx/streaming.ts
index fac77193..5df7a3e4 100644
--- a/src/jsx/streaming.ts
+++ b/src/jsx/streaming.ts
@@ -116,48 +116,60 @@ const textEncoder = new TextEncoder()
  * The API might be changed.
  */
 export const renderToReadableStream = (
-  str: HtmlEscapedString | Promise<HtmlEscapedString>
+  str: HtmlEscapedString | Promise<HtmlEscapedString>,
+  onError: (e: unknown) => void = console.trace
 ): ReadableStream<Uint8Array> => {
   const reader = new ReadableStream<Uint8Array>({
     async start(controller) {
-      const tmp = str instanceof Promise ? await str : await str.toString()
-      const context = typeof tmp === 'object' ? tmp : {}
-      const resolved = await resolveCallback(
-        tmp,
-        HtmlEscapedCallbackPhase.BeforeStream,
-        true,
-        context
-      )
-      controller.enqueue(textEncoder.encode(resolved))
-
-      let resolvedCount = 0
-      const callbacks: Promise<void>[] = []
-      const then = (promise: Promise<string>) => {
-        callbacks.push(
-          promise
-            .catch((err) => {
-              console.trace(err)
-              return ''
-            })
-            .then(async (res) => {
-              res = await resolveCallback(res, HtmlEscapedCallbackPhase.BeforeStream, true, context)
-              ;(res as HtmlEscapedString).callbacks
-                ?.map((c) => c({ phase: HtmlEscapedCallbackPhase.Stream, context }))
-                // eslint-disable-next-line @typescript-eslint/no-explicit-any
-                .filter<Promise<string>>(Boolean as any)
-                .forEach(then)
-              resolvedCount++
-              controller.enqueue(textEncoder.encode(res))
-            })
+      try {
+        const tmp = str instanceof Promise ? await str : await str.toString()
+        const context = typeof tmp === 'object' ? tmp : {}
+        const resolved = await resolveCallback(
+          tmp,
+          HtmlEscapedCallbackPhase.BeforeStream,
+          true,
+          context
         )
-      }
-      ;(resolved as HtmlEscapedString).callbacks
-        ?.map((c) => c({ phase: HtmlEscapedCallbackPhase.Stream, context }))
-        // eslint-disable-next-line @typescript-eslint/no-explicit-any
-        .filter<Promise<string>>(Boolean as any)
-        .forEach(then)
-      while (resolvedCount !== callbacks.length) {
-        await Promise.all(callbacks)
+        controller.enqueue(textEncoder.encode(resolved))
+
+        let resolvedCount = 0
+        const callbacks: Promise<void>[] = []
+        const then = (promise: Promise<string>) => {
+          callbacks.push(
+            promise
+              .catch((err) => {
+                console.log(err)
+                onError(err)
+                return ''
+              })
+              .then(async (res) => {
+                res = await resolveCallback(
+                  res,
+                  HtmlEscapedCallbackPhase.BeforeStream,
+                  true,
+                  context
+                )
+                ;(res as HtmlEscapedString).callbacks
+                  ?.map((c) => c({ phase: HtmlEscapedCallbackPhase.Stream, context }))
+                  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+                  .filter<Promise<string>>(Boolean as any)
+                  .forEach(then)
+                resolvedCount++
+                controller.enqueue(textEncoder.encode(res))
+              })
+          )
+        }
+        ;(resolved as HtmlEscapedString).callbacks
+          ?.map((c) => c({ phase: HtmlEscapedCallbackPhase.Stream, context }))
+          // eslint-disable-next-line @typescript-eslint/no-explicit-any
+          .filter<Promise<string>>(Boolean as any)
+          .forEach(then)
+        while (resolvedCount !== callbacks.length) {
+          await Promise.all(callbacks)
+        }
+      } catch (e) {
+        // maybe the connection was closed
+        onError(e)
       }
 
       controller.close()

[mwilkins91]

That seems to make sense to me @usualoma! When you say If you have no objection, I would be happy to have this change incorporated into the PR. do you mean that you'll make that change? or would you like me to make it?

[usualoma]

Hi @mwilkins91!

Thanks for the confirmation.
I am thinking that you can have the content of this PR changed to the diff I made above and add it as your commits.
We can have @yusukebe review it and merge it into the main branch.

[yusukebe]

Yes.

@mwilkins91, can you apply the diff to this PR? If it's done, I'll merge it into main with @usualoma as a co-author.

[mwilkins91]

Sounds perfect, I'll do that tonight!

[mwilkins91]

@yusukebe done!

[yusukebe]

@mwilkins91 Thanks!

One thing. Can you run bun run denoify or yarn denoify and git add&commit&push the generated file.

[mwilkins91]

@yusukebe like so? 3ec79b3

[yusukebe]

@mwilkins91 Yup. That's right!

I'll merge this now. Thanks!