fix(client): don't skip live changes after transaction ID wraparound

.changeset/tidy-xids-wrap.md

@@ -0,0 +1,6 @@
+---
+"@electric-sql/client": patch
+---
+
+Fix subset snapshot filtering after PostgreSQL transaction ID wraparound and
+retire filters once the stream passes each snapshot's database LSN.

packages/typescript-client/src/client.ts

@@ -1571,8 +1571,26 @@ export class ShapeStream<T extends Row<unknown> = Row>
     // Filter messages using snapshot tracker
     const messagesToProcess = batch.filter((message) => {
       if (isChangeMessage(message)) {
+        const changeLsn = message.headers.lsn
+        if (typeof changeLsn === `string` && changeLsn) {
+          // A quiet SSE stream may deliver its first later change before the
+          // next up-to-date boundary. Retire snapshots against that change's
+          // WAL position before resolving its wrapped transaction ID.
+          this.#snapshotTracker.lastSeenUpdate(BigInt(changeLsn))
+        }
         return !this.#snapshotTracker.shouldRejectMessage(message)
       }
+
+      if (isUpToDateMessage(message)) {
+        const lastSeenLsn = message.headers.global_last_seen_lsn
+        if (typeof lastSeenLsn === `string` && lastSeenLsn) {
+          // Process this in message order: changes before the up-to-date
+          // boundary still need snapshot deduplication, while later changes do
+          // not once the database has passed the snapshot's LSN.
+          this.#snapshotTracker.lastSeenUpdate(BigInt(lastSeenLsn))
+        }
+      }
+
       return true // Always process control messages
     })
 

packages/typescript-client/src/snapshot-tracker.ts

@@ -91,6 +91,31 @@ export class SnapshotTracker {
     if (set.size === 0) map.delete(key)
   }
 
+  /**
+   * Resolves a 32-bit xid against an epoch-aware xid8.
+   *
+   * This signed modulo-2^32 calculation requires the reference and xid to be
+   * within 2^31 transactions. ShapeStream enforces that lifetime bound by
+   * retiring snapshots as global_last_seen_lsn passes their database_lsn.
+   *
+   * Mirrors `Electric.Postgres.Xid`
+   * (`packages/sync-service/lib/electric/postgres/xid.ex`).
+   */
+  #toXid8(xid: number, referenceXid8: bigint): bigint {
+    return referenceXid8 + BigInt.asIntN(32, BigInt(xid) - referenceXid8)
+  }
+
+  /**
+   * Resolves each contributing xid into the epoch nearest `referenceXid8` and
+   * returns the latest. Snapshot callers use `xmax` as the reference.
+   */
+  #resolveLatestXid8(xids: number[], referenceXid8: bigint): bigint {
+    return xids.reduce((latest, xid) => {
+      const xid8 = this.#toXid8(xid, referenceXid8)
+      return xid8 > latest ? xid8 : latest
+    }, BigInt(-1))
+  }
+
   /**
    * Check if a change message should be filtered because its already in an active snapshot
    * Returns true if the message should be filtered out (not processed)
@@ -99,19 +124,20 @@ export class SnapshotTracker {
     const txids = message.headers.txids || []
     if (txids.length === 0) return false
 
-    const xid = Math.max(...txids) // Use the maximum transaction ID
-
     for (const [xmax, snapshots] of this.xmaxSnapshots.entries()) {
-      if (xid >= xmax) {
+      const xid8 = this.#resolveLatestXid8(txids, xmax)
+      if (xid8 >= xmax) {
         for (const snapshot of snapshots) {
           this.removeSnapshot(snapshot)
         }
       }
     }
 
-    return [...this.activeSnapshots.values()].some(
-      (x) => x.keys.has(message.key) && isVisibleInSnapshot(xid, x)
-    )
+    return [...this.activeSnapshots.values()].some((snapshot) => {
+      if (!snapshot.keys.has(message.key)) return false
+      const xid8 = this.#resolveLatestXid8(txids, snapshot.xmax)
+      return isVisibleInSnapshot(xid8, snapshot)
+    })
   }
 
   lastSeenUpdate(newDatabaseLsn: bigint): void {