diff --git a/.changeset/fresh-cursors-keep-rows.md b/.changeset/fresh-cursors-keep-rows.md new file mode 100644 index 0000000000..51a8986058 --- /dev/null +++ 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 index b9b3ea5217..78151d1eea 100644 --- 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 index 4e2504c9e9..cb96152256 100644 --- a/packages/typescript-client/src/client.ts +++ b/packages/typescript-client/src/client.ts @@ -1554,11 +1554,7 @@ export class ShapeStream = 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 = 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 index 8606548fa1..c756245c6a 100644 --- 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 index c85560fb34..8380f395f8 100644 --- 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 index 2628373ef9..09fdf60f36 100644 --- 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> = [] + + 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((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[] = []