Electric: scoped move-in snapshots can prematurely resolve awaitTxId and remove optimistic rows

Start with packages/electric-db-collection/src/electric.ts and trace how move-in announcements, snapshot-end messages, and transaction IDs reach the confirmation machinery. Reproduce the reported sequence with the Vitest regression using eager and progressive sync modes. Done means awaitTxId(150) remains pending after A's scoped snapshot and B's announcement and row, then resolves after B's snapshot-end, while full snapshots retain visibility confirmation.

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

An Electric collection using a membership subquery can confirm an optimistic write before the stream delivers that write's rows. The mutation handler returns successfully, TanStack DB removes the optimistic state, and an inserted row disappears until its real synced row arrives. For an optimistic update, this can show the previous value again.

For example, a chat shape contains chats in which the current user is a participant:

SELECT * FROM chats
WHERE id IN (
  SELECT chat_id FROM chat_participants WHERE user_id = 'current-user'
);

Creating a chat and its membership in PostgreSQL makes that chat newly eligible for the shape. Electric processes these dependency changes and fetches the newly eligible rows in scoped move-in snapshots. Separate changes need not be fetched together. A move-in query can observe a newer PostgreSQL database state than the dependency events Electric has processed so far.

Consider this sequence:

1. PostgreSQL transaction 149 creates chat A and its membership.
2. PostgreSQL transaction 150 creates chat B and its membership. The client shows B optimistically and its mutation handler waits on awaitTxId(150).
3. Electric handles A's membership change and fetches A's move-in rows. B's membership change has not yet been processed.
4. A's query runs after both transactions committed. It returns A only, but its snapshot-end carries PostgreSQL visibility metadata such as xmin: "151", xmax: "151", xip_list: [].
5. The adapter treats that scoped snapshot as collection-wide transaction evidence. isVisibleInSnapshot(150, snapshot) returns true, so awaitTxId(150) resolves even though B's row has never arrived.
6. B's optimistic insert is removed. B's actual move-in arrives later.

The number 151 is a database visibility boundary, not a transaction ID assigned to the move-in query and not a claim that every earlier transaction has been delivered. Being visible to a query restricted to A does not mean B was included in its result.

Illustrative wire sequence, omitting unrelated fields:

move-in       txids=[149], scope=A
insert        A, is_move_in=true
snapshot-end  xmin=151, xmax=151, xip_list=[]
up-to-date

-- awaitTxId(150) must still be pending here --

move-in       txids=[150], scope=B
insert        B, is_move_in=true
snapshot-end  xmin=152, xmax=152, xip_list=[]
up-to-date

-- now transaction 150 can be confirmed --

This is a successful but premature acknowledgement, not an API write failure or an acknowledgement timeout. It reproduces without SQLite persistence, a UI framework, or a real backend.

Minimal regression

Save this as a Vitest test in a project with @tanstack/db, @tanstack/electric-db-collection, and vitest. It uses a controlled fetch implementation to replay the relevant protocol messages; no Electric or PostgreSQL service is needed.

It fails against the original published adapter 0.6.3: the first expect(settled).toBe(false) after A's delivery receives true. Both eager and progressive variants pass with the source change below. Later assertions also prevent confirming B from its move-in announcement alone, before its snapshot finishes.

import { createCollection } from '@tanstack/db';
import { electricCollectionOptions } from '@tanstack/electric-db-collection';
import { expect, it } from 'vitest';

const upToDate = { headers: { control: 'up-to-date' } };
const snapshotEnd = (boundary: number) => ({
  headers: {
    control: 'snapshot-end',
    xmin: String(boundary),
    xmax: String(boundary),
    xip_list: [],
  },
});
const moveIn = (txid: number, id: string) => ({
  headers: {
    event: 'move-in',
    txids: [txid],
    patterns: [{ pos: 0, value: id }],
  },
});
const row = (id: string) => ({
  key: id,
  value: { id },
  headers: {
    operation: 'insert',
    is_move_in: true,
    tags: [id],
    active_conditions: [true],
  },
});

it.each(['eager', 'progressive'] as const)(
  'does not confirm B from A\'s move-in snapshot (%s)',
  async (syncMode) => {
    let respond: ((response: Response) => void) | undefined;
    let requestCount = 0;
    let offset = 0;
    const response = (messages: unknown[]) => Response.json(messages, {
      headers: {
        'electric-handle': 'chats',
        'electric-offset': `1_${++offset}`,
        'electric-cursor': String(offset),
        'electric-schema': JSON.stringify({ id: { type: 'text' } }),
      },
    });
    const chats = createCollection(electricCollectionOptions<{ id: string }>({
      getKey: (chat) => chat.id,
      syncMode,
      shapeOptions: {
        url: 'https://example.test/shape',
        fetchClient: async () => {
          if (++requestCount === 1) {
            return response([snapshotEnd(149), upToDate]);
          }
          return new Promise<Response>((resolve) => { respond = resolve; });
        },
      },
      // Simulate the mutation API returning B's PostgreSQL transaction ID.
      onInsert: async ({ collection }) => {
        await collection.utils.awaitTxId(150);
      },
    }));
    const deliver = async (messages: unknown[]) => {
      await expect.poll(() => respond !== undefined).toBe(true);
      const resolve = respond!;
      respond = undefined;
      resolve(response(messages));
      // The next request starts after the adapter processes this batch.
      await expect.poll(() => respond !== undefined).toBe(true);
    };
    try {
      await chats.preload();
      const mutation = chats.insert({ id: 'B' });
      let settled = false;
      void mutation.when('settled').then(() => { settled = true; });
      expect(chats.has('B')).toBe(true);

      // A's query sees transaction 150, but only fetches A's rows.
      await deliver([moveIn(149, 'A'), row('A'), snapshotEnd(151), upToDate]);
      expect(settled).toBe(false); // FAILS on unpatched 0.6.3: true
      expect(chats.has('B')).toBe(true);

      // An announcement alone must not acknowledge B either.
      await deliver([moveIn(150, 'B'), upToDate]);
      expect(settled).toBe(false);
      await deliver([row('B'), upToDate]);
      expect(settled).toBe(false);

      await deliver([snapshotEnd(152), upToDate]);
      await mutation.when('settled');
      expect(chats.has('B')).toBe(true);
    } finally {
      await chats.cleanup();
    }
  },
);

Verified with @tanstack/electric-db-collection 0.6.3, @tanstack/db 0.15.2, @electric-sql/client 1.5.28, and Vitest 5.0.0. The local Electric client has unrelated SSE transport patches; this reproduction uses finite JSON responses rather than SSE. The original adapter source matches the upstream main source fetched on 2026-10-24. The source diff below applies cleanly to that file.

Proposed fix

The affected file is packages/electric-db-collection/src/electric.ts.

Keep PostgreSQL visibility confirmation for full snapshots, but do not publish a move-in snapshot's visibility metadata as collection-wide evidence. Instead:

• Stage the explicit transaction IDs on the move-in announcement.
• Release those IDs into the existing confirmation machinery only after the matching snapshot-end; retain the existing collection commit boundary before publishing evidence.
• Keep the staged state across message batches. An up-to-date message can arrive between a move-in announcement, its rows, and its closing snapshot boundary.
• Recognize is_move_in rows as scoped even when a resumed stream starts after the announcement.
• Preserve snapshot boundary messages during progressive buffering and process evidence in order during the atomic swap.
• Clear staged evidence on must-refetch.

This does not wait for a persisted collection's final row application: the existing commit/evidence publication mechanism remains in place, avoiding a deadlock with the optimistic mutation's own handler.

The following is the source-only diff from our locally verified patch. The published ESM and CommonJS bundles were also patched for the runtime verification.

diff --git a/packages/electric-db-collection/src/electric.ts b/packages/electric-db-collection/src/electric.ts
index bb5f8b604826dbb73ddd3aa2b3284b08793d28a1..84b0bca8f0f0b6818fcde3fff0454a37066f6991 100644
--- a/packages/electric-db-collection/src/electric.ts
+++ b/packages/electric-db-collection/src/electric.ts
@@ -2032,6 +2032,35 @@ function createElectricSync<T extends Row<unknown>>(
     let transactionStarted = false
     const newTxids = new Set<Txid>()
     const newSnapshots: Array<PostgresSnapshot> = []
+    // A move-in snapshot covers dependency tags, not the whole shape.
+    // Confirm only its explicit txids, after its rows and snapshot-end arrive.
+    // Remove this patch when upstream scopes incremental snapshot evidence.
+    let inMoveInSnapshot = false
+    const moveInTxids = new Set<Txid>()
+    const observeTxidEvidence = (message: Message<T>) => {
+      if (isMustRefetchMessage(message)) {
+        inMoveInSnapshot = false
+        moveInTxids.clear()
+        newTxids.clear()
+        newSnapshots.length = 0
+      } else if (isMoveInMessage(message)) {
+        inMoveInSnapshot = true
+        if (hasTxids(message)) message.headers.txids?.forEach((txid) => moveInTxids.add(txid))
+      } else if (isChangeMessage(message) && message.headers.is_move_in) {
+        inMoveInSnapshot = true
+      } else if (isSnapshotEndMessage(message)) {
+        if (inMoveInSnapshot) {
+          moveInTxids.forEach((txid) => newTxids.add(txid))
+          moveInTxids.clear()
+          inMoveInSnapshot = false
+        } else {
+          newSnapshots.push(parseSnapshotMessage(message))
+        }
+      } else if (hasTxids(message)) {
+        message.headers.txids?.forEach((txid) => newTxids.add(txid))
+      }
+    }
+
     // Track if we've completed initial sync in progressive mode. A persisted
     // resume starts from an already-committed stream offset, so the next
     // up-to-date message must not run the initial atomic swap again.
@@ -2385,11 +2414,8 @@ function createElectricSync<T extends Row<unknown>>(
         // Skip during buffered initial sync in progressive mode (txids will be extracted during atomic swap)
         // EXCEPTION: If a transaction is already started (e.g., from must-refetch), track txids
         // to avoid losing them when messages are written to the existing transaction.
-        if (
-          hasTxids(message) &&
-          (!isBufferingInitialSync() || transactionStarted)
-        ) {
-          message.headers.txids?.forEach((txid) => newTxids.add(txid))
+        if (!isBufferingInitialSync() || transactionStarted) {
+          observeTxidEvidence(message)
         }
 
         if (isChangeMessage(message)) {
@@ -2449,8 +2475,8 @@ function createElectricSync<T extends Row<unknown>>(
           // Skip during buffered initial sync (will be extracted during atomic swap)
           // EXCEPTION: If a transaction is already started (e.g., from must-refetch), track snapshots
           // to avoid losing them when messages are written to the existing transaction.
-          if (!isBufferingInitialSync() || transactionStarted) {
-            newSnapshots.push(parseSnapshotMessage(message))
+          if (isBufferingInitialSync() && !transactionStarted) {
+            bufferedMessages.push(message)
           }
         } else if (isUpToDateMessage(message)) {
           // up-to-date takes precedence - also triggers progressive mode atomic swap
@@ -2565,16 +2591,9 @@ function createElectricSync<T extends Row<unknown>>(
 
           // Apply all buffered change messages and extract txids/snapshots
           for (const bufferedMsg of bufferedMessages) {
+            observeTxidEvidence(bufferedMsg)
             if (isChangeMessage(bufferedMsg)) {
               processChangeMessage(bufferedMsg)
-
-              // Extract txids from buffered messages (will be committed to store after transaction)
-              if (hasTxids(bufferedMsg)) {
-                bufferedMsg.headers.txids?.forEach((txid) => newTxids.add(txid))
-              }
-            } else if (isSnapshotEndMessage(bufferedMsg)) {
-              // Extract snapshots from buffered messages (will be committed to store after transaction)
-              newSnapshots.push(parseSnapshotMessage(bufferedMsg))
             } else if (isMoveOutMessage(bufferedMsg)) {
               // Process buffered move-out messages during atomic swap
               processMoveOutEvent(

Additional regression coverage should retain full initial-snapshot confirmation and cover scoped snapshots during progressive initial buffering, resumed move-in rows without their announcement, reset during a move-in, and empty move-in snapshots.

There is a related server edge case: Electric 1.8.2 omits snapshot-end for zero-row move-in queries. A client that correctly waits for this boundary needs the server to emit it for empty results too. We have a separate server fix for that case; it does not justify treating scoped visibility metadata as global confirmation.

Related work

• #1561 already describes snapshot-based awaitTxId resolution before a row arrives. Its reproduction and proposed direction concern core sync buffering behind another persisting mutation. This report isolates an adapter-level cause: unrelated scoped snapshot metadata can acknowledge a write before its row is in the incoming stream at all. It requires no second pending mutation.
• #648 introduced snapshot visibility confirmation for full initial snapshots. That behavior should remain supported.
• #924 fixes commits before the first up-to-date. Here, premature confirmation also happens after initial loading has completed.
• #2120 concerns delayed acknowledgement during progressive background loading, rather than acknowledgement from an unrelated move-in snapshot.
• #2201 / #2208 concern a row already received by the source but waiting behind a persistence write. Here the row has not been received, and persistence is not required.

Is treating scoped move-in snapshots separately from full-snapshot confirmation the intended approach? If this belongs under the earlier #1561 discussion, the reproduction and source diff can serve as a focused follow-up there.