eventStream doesn't connect properly after browser refresh, on node server setup

[hilja]

Describe the bug

It works good until I refresh the browser and issue a event.

TypeError: The stream is not in a state that permits enqueue
    at ReadableStreamDefaultController2.enqueue (/Users/bob/web/remix-run-sse/build/server.js:3136:21)
    at send (/Users/bob/web/remix-run-sse/node_modules/.pnpm/remix-utils@7.1.0_@remix-run+node@2.1.0_@remix-run+react@2.1.0_react@18.2.0/node_modules/remix-utils/build/server/event-stream.js:12:28)
    at handle (/Users/bob/web/remix-run-sse/app/routes/report.subscribe.ts:11:9)
    at EventEmitter.<anonymous> (/Users/bob/web/remix-run-sse/app/routes/report.subscribe.ts:18:7)
    at EventEmitter.emit (node:events:525:35)
    at doAsyncWork (/Users/bob/web/remix-run-sse/app/statusEmitter.ts:11:13)
    at statusEmitter (/Users/bob/web/remix-run-sse/app/statusEmitter.ts:14:3) Error at the `send()`

Your Example Website or App

https://stackblitz.com/edit/remix-run-remix-p98wmt?file=server.ts,remix.config.js,app%2FstatusEmitter.ts,app%2Froutes%2Freport.subscribe.ts

Steps to Reproduce the Bug or Issue

NOTE: the error won't appear in StackBlitz, you need to clone the repo and run it locally https://github.com/hilja/remix-run-sse

1. Restart the server and don't refresh
2. Navigate to the /report/foo page
3. Click the button and wait for the events to trigger, should work
4. Then refresh and try again and the errors should appear

Expected behavior

No errors, the stream should stay open.

Screenshots or Videos

No response

Platform

• Remix 2.1.0
• remix-utils 7.1.0
• Browser: Brave/Chrome

Additional context

I don't know if this is just a development environment thing or what, haven't deployed it to a real server yet. But I've tried to run the built project, same thing.

There was a header flush update on Remix just remix-run/remix#7619. It made it better, the stream is not in pending state at first. But didn't fix my issue.

[sergiodxa]

If the app is refreshed on the browser the connection is closed and a new one will be open once the app loads again, there's no way the browser will keep the connection open.

[hilja]

I phrased it wrong, it doesn't connect properly after the reload and is not in the state where it can enqueue.

[hilja]

This timer based example works perfectly. But not the one where the send is triggered by an event.

import type { LoaderFunctionArgs } from '@remix-run/node'
import { eventStream } from 'remix-utils/sse/server'
import { interval } from 'remix-utils/timers'

export async function loader({ request }: LoaderFunctionArgs) {
  return eventStream(request.signal, send => {
    async function run() {
      for await (const _ of interval(2000, { signal: request.signal })) {
        send({ event: 'time', data: new Date().toISOString() })
      }
    }

    run()

    return () => {}
  })
}

[hilja]

So when the events are fired continuously with a timer, the ReadableStream is open when the reload happens and it has no trouble reconnecting. But in the case of the events which end before the reload, and I assume the stream closes after the last event? Then after the reload it tries to interact with the closed ReadableStream and fails. Is that what's going on?

[hilja]

If I add a clog in here:

function send({ event = 'message', data }: SendFunctionArgs) {
  console.log('🔴 signal aborted:', signal.aborted)
  controller.enqueue(encoder.encode(`event: ${event}\n`))
  controller.enqueue(encoder.encode(`data: ${data}\n\n`))
}

I see the signal.aborted flip-flopping:

active
🔴 signal aborted: true
Error at the `send()` method:
  TypeError: The stream is not in a state that permits enqueue
working
🔴 signal aborted: false
Error at the `send()` method:
  TypeError: The stream is not in a state that permits enqueue
completed
🔴 signal aborted: true
Error at the `send()` method:
  TypeError: The stream is not in a state that permits enqueue

So it changes after every event. But if I ditch the requst.signal and have my own it seems to work:

export async function loader() {
  const controller = new AbortController()
  controller.signal.addEventListener('abort', () => controller.abort())

  return eventStream(controller.signal, send => {
    function handle(state: string) {
      try {
        send({ event: 'reportStatus', data: state })
      } catch (error: any) {
        logger.error(error, 'Error at the `send()` method:')
      }
    }

    emitter.once('active', state => {
      console.log(state)
      handle(state)
    })

    emitter.once('stalled', state => {
      console.log(state)
      handle(state)
    })

    emitter.once('completed', state => {
      console.log(state)
      handle(state)
      return controller.abort()
    })

    return () => {
      emitter.off('completed', handle)
      emitter.off('stalled', handle)
      emitter.off('active', handle)
    }
  })
}

Wonder if that has some repercussions down the line?

[hilja]

Okay I got it, nothing wrong with this lib :) The problem was I don't revalidate the route because the background job returns nothing, so the React component also wasn't re-rendering, because SSE data won't trigger revalidation and I wasn't putting my SSE data into a state. That caused the useEventSource to be out-of-sync basically. Also the dev server was adding to the confusion, after a reload the stream works erratically, have to restart it manually after every change.

[janmonschke]

@hilja I was running into similar issues and found out that it was a bug in my loader function. I did not properly unsubscribe from the event emitter. This meant that when one browser refresehed, the old listener was still trying to send but that stream was already ended.

In your example above, the same thing can be observed

export async function loader() {
  const controller = new AbortController()
  controller.signal.addEventListener('abort', () => controller.abort())

  return eventStream(controller.signal, send => {
    function handle(state: string) {
      try {
        send({ event: 'reportStatus', data: state })
      } catch (error: any) {
        logger.error(error, 'Error at the `send()` method:')
      }
    }

    emitter.once('active', state => {
      console.log(state)
      handle(state)
    })
 
    // (...)

    return () => {
.      // (...)
      emitter.off('active', handle)
    }
  })
}

The cleanup function is unsubscribing handle but the actual listener on active is an anonymous function that will still listen to events even after cleanup has been called. Changing the listener to a named function should solve that issue 👍