electric-sql · joshdchang · Jul 15, 2026
diff --git a/.changeset/fresh-cursors-keep-rows.md b/.changeset/fresh-cursors-keep-rows.md
@@ -0,0 +1,5 @@
+---
+'@electric-sql/client': patch
+---
+
+Preserve data messages when suppressing cached up-to-date notifications during replay.
diff --git a/packages/typescript-client/SPEC.md b/packages/typescript-client/SPEC.md
@@ -241,7 +241,7 @@ up-to-date; replay is meaningless.
 
 - ErrorState:
   `handleResponseMetadata` returns `{ action: 'ignored', state: this }`
-  and `handleMessageBatch` returns `{ state: this, suppressBatch: false, becameUpToDate: false }`
+  and `handleMessageBatch` returns `{ state: this, suppressUpToDate: false, becameUpToDate: false }`
 - PausedState:
   `handleMessageBatch` and `handleSseConnectionClosed` are no-ops
   and `handleResponseMetadata` delegates to `previousState`, preserving the paused wrapper for `accepted` and `stale-retry` transitions (`ignored` returns `this`)
@@ -292,6 +292,16 @@ back to Live, SSE state resets to defaults.
 
 **Enforcement**: Dedicated test (`SSE state is preserved through LiveState self-transitions`).
 
+### C9: Replay suppression must not discard data messages
+
+When a non-SSE replay reaches an up-to-date message with the same cursor as the
+previous session, the duplicate `up-to-date` control message is suppressed. Every
+other message in that batch, including inserts, updates, and deletes, is still
+delivered to stream subscribers. A fresh up-to-date message is delivered once the
+cursor advances.
+
+**Enforcement**: Dedicated test (`should preserve change messages while suppressing a cached up-to-date`).
+
 ## Shape notification semantics
 
 The `Shape` class (`shape.ts`) wraps a `ShapeStream` and notifies subscribers
@@ -368,31 +378,32 @@ observing an intermediate empty-rows notification. The
 | C6         | -     | -           | yes            |
 | C7         | -     | yes         | yes            |
 | C8         | -     | -           | yes            |
+| C9         | -     | -           | yes            |
 
 ### Code -> Doc: Is each test derived from the spec?
 
-| Test File / Section            | Spec Reference          |
-| ------------------------------ | ----------------------- |
-| Tier 1: scenario builder tests | I0-I11 (via auto-check) |
-| Tier 2: transition truth table | All 70 cells            |
-| Algebraic property tests       | I3, I4, I10, I11, I8    |
-| Fuzz testing                   | I0-I12 (all invariants) |
-| Mutation testing               | I0-I12 (robustness)     |
-| shouldUseSse guard tests       | LiveState SSE behavior  |
-| SSE connection closed tests    | LiveState SSE fallback  |
-| applyUrlParams tests           | URL construction        |
-| Schema adoption tests          | C4                      |
-| 204/200 lastSyncedAt tests     | C5                      |
-| SSE offset tests               | C6                      |
-| Stale handle tests             | C7                      |
-| ReplayingState suppress tests  | Replay cursor semantics |
+| Test File / Section            | Spec Reference              |
+| ------------------------------ | --------------------------- |
+| Tier 1: scenario builder tests | I0-I11 (via auto-check)     |
+| Tier 2: transition truth table | All 70 cells                |
+| Algebraic property tests       | I3, I4, I10, I11, I8        |
+| Fuzz testing                   | I0-I12 (all invariants)     |
+| Mutation testing               | I0-I12 (robustness)         |
+| shouldUseSse guard tests       | LiveState SSE behavior      |
+| SSE connection closed tests    | LiveState SSE fallback      |
+| applyUrlParams tests           | URL construction            |
+| Schema adoption tests          | C4                          |
+| 204/200 lastSyncedAt tests     | C5                          |
+| SSE offset tests               | C6                          |
+| Stale handle tests             | C7                          |
+| ReplayingState suppress tests  | Replay cursor semantics, C9 |
 
 ### Gaps
 
 | Gap                            | Status | Notes                                         |
 | ------------------------------ | ------ | --------------------------------------------- |
 | SSE fallback to long polling   | Tested | Direct construction only (DSL doesn't expose) |
-| ReplayingState suppressBatch   | Tested | Direct construction only (DSL doesn't expose) |
+| ReplayingState suppression     | Tested | Direct construction only (DSL doesn't expose) |
 | ErrorState.reset()             | Tested | Direct construction (DSL doesn't have reset)  |
 | handleMessageBatch no-messages | Tested | Direct construction (edge case)               |
 

diff --git a/packages/typescript-client/src/client.ts b/packages/typescript-client/src/client.ts
@@ -1554,11 +1554,7 @@ export class ShapeStream<T extends Row<unknown> = Row>
 
     if (hasUpToDateMessage) {
       this.#refreshCatchUpWatchdogActive = false
-      if (transition.suppressBatch) {
-        return
-      }
-
-      if (this.#currentFetchUrl) {
+      if (!transition.suppressUpToDate && this.#currentFetchUrl) {
         const shapeKey = canonicalShapeKey(this.#currentFetchUrl)
         upToDateTracker.recordUpToDate(
           shapeKey,
@@ -1570,12 +1566,17 @@ export class ShapeStream<T extends Row<unknown> = Row>
 
     // Filter messages using snapshot tracker
     const messagesToProcess = batch.filter((message) => {
+      if (transition.suppressUpToDate && isUpToDateMessage(message)) {
+        return false
+      }
       if (isChangeMessage(message)) {
         return !this.#snapshotTracker.shouldRejectMessage(message)
       }
       return true // Always process control messages
     })
 
+    if (messagesToProcess.length === 0) return
+
     await this.#publish(messagesToProcess, {
       allowReentrantBypass: opts.allowReentrantPublishBypass,
     })

diff --git a/packages/typescript-client/src/shape-stream-state.ts b/packages/typescript-client/src/shape-stream-state.ts
@@ -90,7 +90,7 @@ export interface MessageBatchInput {
 
 export interface MessageBatchTransition {
   state: ShapeStreamState
-  suppressBatch: boolean
+  suppressUpToDate: boolean
   becameUpToDate: boolean
 }
 
@@ -196,7 +196,7 @@ export abstract class ShapeStreamState {
   }
 
   handleMessageBatch(_input: MessageBatchInput): MessageBatchTransition {
-    return { state: this, suppressBatch: false, becameUpToDate: false }
+    return { state: this, suppressUpToDate: false, becameUpToDate: false }
   }
 
   // --- Universal transitions ---
@@ -323,7 +323,7 @@ abstract class ActiveState extends ShapeStreamState {
 
   handleMessageBatch(input: MessageBatchInput): MessageBatchTransition {
     if (!input.hasMessages || !input.hasUpToDateMessage) {
-      return { state: this, suppressBatch: false, becameUpToDate: false }
+      return { state: this, suppressUpToDate: false, becameUpToDate: false }
     }
 
     // Has up-to-date message — compute shared fields for the transition
@@ -350,7 +350,7 @@ abstract class ActiveState extends ShapeStreamState {
   ): MessageBatchTransition {
     return {
       state: new LiveState(shared),
-      suppressBatch: false,
+      suppressUpToDate: false,
       becameUpToDate: true,
     }
   }
@@ -546,7 +546,7 @@ export class LiveState extends ActiveState {
   ): MessageBatchTransition {
     return {
       state: new LiveState(shared, this.sseState),
-      suppressBatch: false,
+      suppressUpToDate: false,
       becameUpToDate: true,
     }
   }
@@ -638,13 +638,13 @@ export class ReplayingState extends ActiveState {
     shared: SharedStateFields,
     input: MessageBatchInput
   ): MessageBatchTransition {
-    // Suppress replayed cache data when cursor has not moved since
+    // Suppress the replayed up-to-date notification when the cursor has not moved since
     // the previous session (non-SSE only).
-    const suppressBatch =
+    const suppressUpToDate =
       !input.isSse && this.#replayCursor === input.currentCursor
     return {
       state: new LiveState(shared),
-      suppressBatch,
+      suppressUpToDate,
       becameUpToDate: true,
     }
   }

diff --git a/packages/typescript-client/test/shape-stream-state.test.ts b/packages/typescript-client/test/shape-stream-state.test.ts
@@ -287,7 +287,7 @@ describe(`shape stream state machine`, () => {
       makeMessageBatchInput({ currentCursor: `cursor-1` })
     )
 
-    expect(transition.suppressBatch).toBe(true)
+    expect(transition.suppressUpToDate).toBe(true)
     expect(transition.state).toBeInstanceOf(LiveState)
     expect(transition.state.replayCursor).toBe(undefined)
   })
@@ -303,7 +303,7 @@ describe(`shape stream state machine`, () => {
       makeMessageBatchInput({ currentCursor: `cursor-2` })
     )
 
-    expect(transition.suppressBatch).toBe(false)
+    expect(transition.suppressUpToDate).toBe(false)
     expect(transition.state).toBeInstanceOf(LiveState)
   })
 
@@ -318,7 +318,7 @@ describe(`shape stream state machine`, () => {
       makeMessageBatchInput({ isSse: true, currentCursor: `cursor-1` })
     )
 
-    expect(transition.suppressBatch).toBe(false)
+    expect(transition.suppressUpToDate).toBe(false)
     expect(transition.state).toBeInstanceOf(LiveState)
   })
 
@@ -594,7 +594,7 @@ describe(`shape stream state machine`, () => {
     )
 
     expect(transition.state).toBe(syncing)
-    expect(transition.suppressBatch).toBe(false)
+    expect(transition.suppressUpToDate).toBe(false)
     expect(transition.becameUpToDate).toBe(false)
   })
 

diff --git a/packages/typescript-client/test/up-to-date-tracker.test.ts b/packages/typescript-client/test/up-to-date-tracker.test.ts
@@ -1,5 +1,5 @@
 import { beforeEach, afterEach, describe, expect, it, vi } from 'vitest'
-import { ShapeStream } from '../src'
+import { type Message, ShapeStream } from '../src'
 import { UpToDateTracker, upToDateTracker } from '../src/up-to-date-tracker'
 
 describe(`UpToDateTracker`, () => {
@@ -214,6 +214,102 @@ describe(`UpToDateTracker`, () => {
     expect(tracker.shouldEnterReplayMode(shapeKey2)).toBe(null)
   })
 
+  it(`should preserve change messages while suppressing a cached up-to-date`, async () => {
+    const table = `replayed_rows`
+    const shapeKey = `${shapeUrl}?table=${table}`
+    const rowKey = `dashboard-version-1`
+    const notifications: Array<Array<Message>> = []
+
+    upToDateTracker.recordUpToDate(shapeKey, `cursor-1`)
+
+    fetchMock
+      .mockResolvedValueOnce(
+        new Response(
+          JSON.stringify([
+            {
+              key: rowKey,
+              value: { id: rowKey },
+              headers: { operation: `insert` },
+            },
+            { headers: { control: `up-to-date` } },
+          ]),
+          {
+            status: 200,
+            headers: {
+              'electric-handle': `test-handle-1`,
+              'electric-offset': `0_0`,
+              'electric-schema': `{}`,
+              'electric-cursor': `cursor-1`,
+              'electric-up-to-date': `true`,
+            },
+          }
+        )
+      )
+      .mockResolvedValueOnce(
+        new Response(JSON.stringify([{ headers: { control: `up-to-date` } }]), {
+          status: 200,
+          headers: {
+            'electric-handle': `test-handle-1`,
+            'electric-offset': `0_0`,
+            'electric-schema': `{}`,
+            'electric-cursor': `cursor-2`,
+            'electric-up-to-date': `true`,
+          },
+        })
+      )
+
+    const stream = new ShapeStream({
+      url: shapeUrl,
+      params: { table },
+      signal: aborter.signal,
+      fetchClient: fetchMock,
+      subscribe: true,
+    })
+
+    await new Promise<void>((resolve, reject) => {
+      const timeout = setTimeout(() => {
+        aborter.abort()
+        reject(new Error(`Timed out waiting for fresh up-to-date`))
+      }, 500)
+
+      stream.subscribe(
+        (messages) => {
+          notifications.push(messages)
+          const hasUpToDate = messages.some(
+            (message) =>
+              `control` in message.headers &&
+              message.headers.control === `up-to-date`
+          )
+          if (hasUpToDate) {
+            clearTimeout(timeout)
+            aborter.abort()
+            resolve()
+          }
+        },
+        (error) => {
+          clearTimeout(timeout)
+          reject(error)
+        }
+      )
+    })
+
+    const receivedMessages = notifications.flat()
+    expect(fetchMock).toHaveBeenCalledTimes(2)
+    expect(
+      receivedMessages.some(
+        (message) => `key` in message && message.key === rowKey
+      )
+    ).toBe(true)
+    expect(
+      receivedMessages.some(
+        (message) =>
+          `control` in message.headers &&
+          message.headers.control === `up-to-date`
+      )
+    ).toBe(true)
+    expect(stream.isUpToDate).toBe(true)
+  })
+
   it(`should suppress cached up-to-dates during replay mode`, async () => {
     const notifications: any[] = []