SSE catch-up loses earlier chunks when a finite response ends before up-to-date #4835

Disclaimer: This issue has been generated completely by GPT-6.1 Sol. I understand the issue only on a high level and haven't studied whether the suggested patch is reasonable.

Problem

The JavaScript client can silently lose changes while catching up over SSE. If the outstanding backlog spans several server chunks, subscribers may receive only the final chunk, even though all earlier chunks arrived successfully with HTTP 200. The rows remain in PostgreSQL.

The problem is in @electric-sql/client, reproduced with the unmodified npm release 1.5.28. The same request-local buffering code is present in upstream main at 3f9809fb0, inspected on 2026-10-24.

For example, consider an initially empty chat shape:

1. The client finishes its initial JSON load and switches to SSE for changes.
2. While it is disconnected or before its next SSE request reaches the server, enough changes accumulate to span several stored chunks. This is a backlog at request time, rather than the cumulative number of changes since initial loading.
3. The SSE request receives the next complete chunk. Electric deliberately finishes the response at that chunk boundary. It does not send up-to-date, because more changes are waiting.
4. The client holds that chunk's messages until it sees up-to-date. However, its buffer belongs to this one HTTP request, so finishing the response discards the unpublished messages.
5. The client requests the next chunk using the offset from the preceding response headers. That offset correctly identifies the next position, but the discarded rows are now behind it.
6. The final response includes up-to-date. Only the messages buffered during that last response reach subscribers. Earlier chunks are never replayed by continuing from the advanced offset.

With 80 distinct inserts in chunks of 39 + 39 + 2, this can leave an initially empty subscriber with only 2 rows. This does not require a timeout race, a truncated event, a proxy, or rate limiting: the server ends the early responses cleanly after complete chunks. Chunk sizes are based on stored log size; 39 rows is specific to the fixture.

An SSE response that is already waiting for new changes normally remains open after sending them and waits for more until its lifetime limit or another termination condition. That established-stream path differs from the finite catch-up response above and from #4833.

Relevant code

• The server chooses a stored chunk boundary and omits up-to-date when that boundary precedes the available head.
• When existing changes are available, it returns a bounded response instead of entering the long-lived wait.
• The client's ShapeStream method that reads one SSE response is named #requestShapeSSE (# means a private JavaScript method). It creates a fresh buffer for each request and publishes it only when an up-to-date message arrives.
• Before reading those messages, the client accepts the response's offset header, advancing the resume position despite the still-unpublished body.

Missing inserts were directly reproduced. The same path carries updates and deletes, so skipped messages can also leave stale values or rows that should have disappeared. Applications using the stream to replace optimistic state can remove that state after confirmation or a confirmation wait expires without ever receiving the authoritative row. These application consequences are separate from the minimal client reproduction below.

Minimal regression

This standalone failing test uses only the unmodified npm package and Node's assertion library. No Electric server, application code, TanStack adapter, browser persistence, or local patches are involved. The mocked fetch supplies the clean finite-response sequence described above; it does not reproduce or replace the client's implementation.

In an isolated directory, install the package, save the test as repro.mjs, and run it:

pnpm init
pnpm add @electric-sql/client@1.5.28
node repro.mjs

import assert from 'node:assert/strict';
import { ShapeStream } from '@electric-sql/client';

const abort = new AbortController();
const delivered = [];
const requests = []; // Captures resume offsets; no real network request is made.
const marker = (lsn) => ({ headers: { control: 'up-to-date', global_last_seen_lsn: lsn } });
const row = (id) => ({ key: id, value: { id }, headers: { operation: 'insert', lsn: '1' } });
const deadline = setTimeout(() => abort.abort(), 3000);

const stream = new ShapeStream({
  url: 'https://example.invalid/' + crypto.randomUUID(),
  liveSse: true,
  liveRequestTimeoutMs: false,
  signal: abort.signal,
  onError: (error) => console.error(error.name),
  fetchClient: async (input) => {
    const url = new URL(String(input));
    const index = requests.length;
    requests.push(url.searchParams.get('offset'));
    if (index >= 3) {
      abort.abort();
      throw new DOMException('Reproduction complete', 'AbortError');
    }
    const headers = {
      'electric-handle': 'test-handle',
      'electric-offset': ['0_0', '1_0', '1_2'][index],
      'electric-cursor': String(index),
      'electric-schema': JSON.stringify({ id: { type: 'text' } }),
    };
    if (index !== 1) {
      headers['electric-up-to-date'] = 'true';
    }
    const messages = index === 0 ? [marker('0')]
      : index === 1 ? [row('A')]
      : [row('B'), marker('1')];
    return url.searchParams.get('live_sse') === 'true'
      ? new Response(messages.map((message) => 'data: ' + JSON.stringify(message) + '\n\n').join(''), {
          headers: { ...headers, 'content-type': 'text/event-stream' },
        })
      : Response.json(messages, { headers });
  },
});
stream.subscribe((messages) => {
  delivered.push(...messages.filter((message) => message.key).map((message) => message.key));
});
await new Promise((resolve) => abort.signal.addEventListener('abort', resolve, { once: true }));
clearTimeout(deadline);
await new Promise((resolve) => setTimeout(resolve, 20));
assert.deepEqual(delivered, ['A', 'B']);
process.exit(0);

Expected: both A and B reach the subscriber together with the actual completion marker. Observed: the assertion fails with actual: ['B']; A is missing. This is an assertion failure, not expiration of the safety deadline. The unpatched client also has the separate offset-regression behavior discussed in #4806/#4807; fixing that alone does not restore A.

A separate real-server check corroborated the response pattern against Electric 1.8.2: JSON reads found all 80 rows in chunks of 39 + 39 + 2, while SSE subscribers received 2/80 in two runs. JSON polling delivered 80/80. That server had the local established-stream and empty-move-in patches related to #4833/#4834; its finite-response path was unchanged. The independent failing test above requires none of those patches and isolates this client defect.

Proposed fix

Retain unpublished messages on the ShapeStream instance across clean finite responses, and publish them only when the actual up-to-date boundary arrives. Preserve that pending batch if the existing short-connection heuristic switches the transport to JSON polling. Clear pending rows when the shape resets, rather than mixing data from different shape generations.

For interrupted SSE responses, restore the request's starting resume state and discard the partial body of that failed request. Earlier clean chunks remain retained. Completed messages from the failed request can replay (at-least-once delivery), but unread changes cannot be skipped simply because response headers arrived. Preserve pause state and stale-response/reset handling.

Source file: packages/typescript-client/src/client.ts. This diff is against the unpatched 1.5.28 source; it does not include our independent short-SSE fallback or monotonic-offset patches.

--- a/packages/typescript-client/src/client.ts
+++ b/packages/typescript-client/src/client.ts
@@ -671,6 +671,9 @@
 
   #started = false
   #syncState: ShapeStreamState
+  // Finite SSE responses may end before up-to-date. Keep their unpublished
+  // messages until the real completion boundary, including through JSON fallback.
+  #pendingSseMessages: Array<Message<T>> = []
   #connected: boolean = false
   #mode: LogMode
   #onError?: ShapeStreamErrorHandler
@@ -1649,7 +1652,20 @@
       return this.#requestShapeSSE(opts)
     }
 
-    return this.#requestShapeLongPoll(opts)
+    const initialState = this.#syncState
+    const hadPendingMessages = this.#pendingSseMessages.length > 0
+    try {
+      return await this.#requestShapeLongPoll(opts)
+    } catch (error) {
+      // A failed fallback body must not skip the rest of a retained batch.
+      // Preserve resets performed while rejecting stale response metadata.
+      if (hadPendingMessages && this.#syncState.handle === initialState.handle) {
+        this.#syncState = this.#syncState instanceof PausedState
+          ? initialState.pause()
+          : initialState
+      }
+      throw error
+    }
   }
 
   async #withRequestTimeout<T>(
@@ -1712,7 +1728,7 @@
     const schema = this.#syncState.schema! // we know that it is not undefined because it is set by `this.#onInitialResponse`
     const res = await response.text()
     const messages = res || `[]`
-    const batch = this.#messageParser.parse<Array<Message<T>>>(messages, schema)
+    let batch = this.#messageParser.parse<Array<Message<T>>>(messages, schema)
 
     if (!Array.isArray(batch)) {
       const preview = bigintSafeStringify(batch)?.slice(0, 200)
@@ -1727,6 +1743,14 @@
       )
     }
 
+    if (this.#pendingSseMessages.length > 0) {
+      batch = [...this.#pendingSseMessages, ...batch]
+      this.#pendingSseMessages = batch
+      if (!isUpToDateMessage(batch[batch.length - 1])) {
+        return
+      }
+      this.#pendingSseMessages = []
+    }
     await this.#onMessages(batch)
   }
 
@@ -1747,9 +1771,11 @@
       Accept: `text/event-stream`,
     }
 
+    const initialState = this.#syncState
+    let cleanClose = false
+    let buffer = [...this.#pendingSseMessages]
     let ignoredStaleResponse = false
     try {
-      let buffer: Array<Message<T>> = []
       await fetchEventSource(fetchUrl.toString(), {
         headers: sseHeaders,
         fetch,
@@ -1759,6 +1785,10 @@
           if (!shouldProcessBody) {
             ignoredStaleResponse = true
             throw new Error(`stale response ignored`)
+          }
+          if (this.#syncState.handle !== initialState.handle) {
+            buffer = []
+            this.#pendingSseMessages = []
           }
         },
         onmessage: (event: EventSourceMessage) => {
@@ -1775,9 +1805,13 @@
               // Flush the buffer on up-to-date message.
               // Ensures that we only process complete batches of operations.
               this.#onMessages(buffer, true)
+              this.#pendingSseMessages = []
               buffer = []
             }
           }
+        },
+        onclose: () => {
+          cleanClose = true
         },
         onerror: (error: Error) => {
           // rethrow to close the SSE connection
@@ -1814,6 +1848,19 @@
       // is misconfigured for SSE and closes the connection immediately
       const connectionDuration = Date.now() - this.#lastSseConnectionStartTime!
       const wasAborted = requestAbortController.signal.aborted
+
+      if (!ignoredStaleResponse) {
+        if (cleanClose && !wasAborted) {
+          this.#pendingSseMessages = buffer
+        } else {
+          // Response headers can be ahead of an unread or truncated body.
+          // Retry this response from its starting checkpoint; already published
+          // messages can replay, but unpublished changes must never be skipped.
+          this.#syncState = this.#syncState instanceof PausedState
+            ? initialState.pause()
+            : initialState
+        }
+      }
 
       const transition = this.#syncState.handleSseConnectionClosed({
         connectionDuration,
@@ -2126,6 +2173,7 @@
    * shape handle
    */
   #reset(handle?: string) {
+    this.#pendingSseMessages = []
     this.#syncState = this.#syncState.markMustRefetch(handle)
     this.#connected = false
     // releaseAllMatching intentionally doesn't fire onReleased — every caller

The standalone regression fails against the unmodified client 1.5.28 and passes with this diff alone. Additional regressions cover JSON fallback, interrupted responses, intentional refresh, and shape reset.

Against Electric 1.8.2, the patched client also received all 80 rows across three finite responses and all 200 rows across six chunks, including fallback from SSE to JSON.

This preserves the completion boundary: it does not fabricate up-to-date messages or publish partial batches on clean EOF. No server change or larger timeout is needed for this client fix.

Related reports

• Live SSE can stop at a completed shape-log chunk until reconnect #4833 concerns draining stored chunks on an established SSE connection, which remains open. This report concerns unpublished messages discarded between separate finite responses.
• Live SSE short-connection fallback triggered by CDN cache-collapsed responses #4137 concerns short-SSE fallback. Retaining messages must survive that fallback; changing the heuristic alone does not recover discarded rows.
• TypeScript client: SSE up-to-date offset ${lsn}_0 regresses behind header electric-offset when the LSN carries multiple ops, causing an immediate-replay loop #4806 / fix(client): never move SSE offset backwards on up-to-date #4807 concern offsets moving backwards. This report still reproduces with that fix backported and requires retaining rows before advancing past them.
• TanStack DB #2244 concerns scoped snapshot confirmation. An adapter cannot apply messages discarded before its subscription is notified.