WIP feat(client): experimental support for streaming SSE in live mode

[samwillis]

Probably needs some tests

[thruflo]

Is it on the live SSE response as a header? I’m thinking that’s fine, as in, the client only needs to update the cursor when the response is empty, so it’s fine to get it from the header.

[KyleAMathews]

Calling this accept to mimic http could make sense

[balegas]

We want to merge experimental SSE support so that we can use them in our official demos.
One thing we want to experiment is to modify server/client to see if we can do live streaming just with regular http anf fetchclient instead of EventSource.

[thruflo]

I think we did want to keep using SSE (just with the slightly different client behaviour) rather than switching to an alternative server response protocol, because using SSE proper triggers the desired request collapsing behaviour from CDNs.

[balegas]

Did you check that keeping the regular http response open for a bit would mess with the CDN request collapsing?

[kevin-dp]

Looking into the implementation it seems that it does not properly handle corner cases such as 409. Any non 200 status code would basically be thrown whereas in our long polling implementation we handle those status codes appropriately.

Also this currently only works in the browser in other environments like NodeJS the EventSource API is not defined and we have to polyfill it. However, as pointed out by @samwillis the browser's event source API is very limited and we would want to use the fetch-event-source library instead. That will solve the polyfill issue.

As a final remark, i don't like the complexity of the ShapeStream.start method. It was already complex and is now becoming even more complex. I have ideas to refactor this code to make it more comprehensible.

[KyleAMathews]

Shipped in #2776

[thruflo]

One of the nuances here that I was thinking about recently is how the SSE client handles interruptibility.

I.e.: we have the ability for the backend to interrupt a long poll request. But for an SSE request, if it's interrupted, how does the SSE client handle this and is there potential for some kind of duplicate message delivery? Or is the SSE client advancing its offset with every message it receives?

[kevin-dp]

But for an SSE request, if it's interrupted, how does the SSE client handle this and is there potential for some kind of duplicate message delivery?

The SSE approach should behave the same way as the long poll when the frontend or the backend interrupt the request. When that happens, the fetch (note that both long poll and SSE use fetch) is aborted and an error is thrown. That error is caught here:

electric/packages/typescript-client/src/client.ts

Lines 444 to 460 in 572a935

	try {
	await this.#requestShape({
	fetchUrl,
	requestAbortController,
	headers: requestHeaders,
	})
	} catch (e) {
	// Handle abort error triggered by refresh
	if (
	(e instanceof FetchError || e instanceof FetchBackoffAbortError) &&
	requestAbortController.signal.aborted &&
	requestAbortController.signal.reason ===
	FORCE_DISCONNECT_AND_REFRESH
	) {
	// Loop back to the top of the while loop to start a new request
	continue
	}

We will continue to the next iteration of the loop which will try resuming from where we left it.

Or is the SSE client advancing its offset with every message it receives?

Yes, each time we receive a message it is processed by this method:

async #onMessages(messages: string, schema: Schema, isSseMessage = false) {
    const batch = this.#messageParser.parse(messages, schema)

    // Update isUpToDate
    if (batch.length > 0) {
      const lastMessage = batch[batch.length - 1]
      if (isUpToDateMessage(lastMessage)) {
        if (isSseMessage) {
          // Only use the offset from the up-to-date message if this was an SSE message.
          // If we would use this offset from a regular fetch, then it will be wrong
          // and we will get an "offset is out of bounds for this shape" error
          const offset = getOffset(lastMessage)
          if (offset) {
            this.#lastOffset = offset
          }
        }
        this.#lastSyncedAt = Date.now()
        this.#isUpToDate = true
      }

      await this.#publish(batch)
    }
  }

So we're using the offset that is provided in the up-to-date message to advance the offset.

[thruflo]

@kevin-dp my concern is:

• accumulate some messages (no uptodate)
• error, reconnect
• accumulate the same messages again
• uptodate -> commit with duplicate data

Seeing the message processing code, I think this can be a possibility, if the SSE client maintains the accumulated messages when handling an interrupt.

Maybe this isn’t an issue because the client clears the accumulator each time it re-connects on error? If so, it’s worth knowing that it’s important to interrupt with an error.

I had considered a 200 response with @magnetised for the Phoenix.Sync interrupt. We’ve moved past and we’re not doing it any more but for a while it could have snuck in there!

[kevin-dp]

@thruflo right, that's a valid concern. So as you described, if we receive an update and then before we get the up to date message the connection breaks, we will reconnect and start from the old offset and get the update a second time. So the shapestream consumer will receive the update twice (or more, depending on how many times this happens). This happens because we immediately publish every incoming message. So, if the connection breaks before it gets uptodate, we will try again with the same offset and get some updates we already got before. Even though this is not 100% right, i think it's fine in most cases because the consumer would usually just materialize the stream into a map and applying an insert/update/delete twice or more is idempotent. But if the consumer does any additional side effects like counting the number of updates then it will be wrong.

We can solve this problem by changing the SSE onmessage callback such that it buffers messages until we get the uptodate message. That way we process only complete batches of operations like we do in regular mode (by virtue of Electric sending out complete batches that end with an up to date message).

[KyleAMathews]

Yeah we should keep the same semantics so buffering until up-to-date makes sense