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

Fixes subset snapshot filtering when PostgreSQL 32-bit transaction IDs
wrap across an xid8 epoch. Clients now keep live collections current
instead of discarding valid changes after wraparound, including on quiet
SSE streams that have not received another up-to-date boundary.

## Root Cause

Change messages carry 32-bit transaction IDs, while subset snapshot
metadata uses 64-bit epoch-aware IDs. Comparing them directly loses the
epoch and can classify a new wrapped transaction as visible in an old
snapshot.

Reconstructing the epoch also has a half-range constraint: the 32-bit
xid and its 64-bit reference must be within `2^31` transactions. A quiet
stream could retain a snapshot past that bound because its next change
may arrive before another `up-to-date` message.

## Approach

- Resolve every wire xid into the epoch nearest the snapshot `xmax`,
then use the latest resolved xid for snapshot eviction and duplicate
filtering.
- Advance snapshot retirement in message order from both
`global_last_seen_lsn` control messages and each change message `lsn`.
- Expand the model-based tests across xid8 epochs, wrap boundaries, and
multi-xid messages.
- Add regressions for wrapped xid eviction and for a quiet stream whose
first later change passes the snapshot WAL position.

This mirrors the xid reconstruction strategy already used by the sync
service in [#2320](#2320).

## Key Invariants

- An active snapshot never outlives the point where the stream WAL
position passes its `database_lsn`.
- Changes before that boundary still receive snapshot duplicate
filtering; later changes do not.
- Each wire xid is resolved before the latest contributing transaction
is chosen.

## Non-goals

- No sync-service or wire-protocol changes.
- No changes to subset snapshot semantics outside transaction ID
resolution and retirement.

## Trade-offs

Using `xmax` as the epoch reference keeps the fix local and matches
PostgreSQL wraparound arithmetic. Its half-range requirement is safe
because LSN-based retirement bounds each snapshot lifetime without
adding protocol state.

## Verification

```bash
pnpm --dir packages/typescript-client test --run
pnpm --dir packages/typescript-client typecheck
pnpm --dir packages/typescript-client stylecheck
```

The full TypeScript client unit suite passed (440 tests), along with
focused regression tests, type checking, linting, and formatting checks.

## Files changed

- `packages/typescript-client/src/client.ts` — retires snapshots from
control and change LSNs in stream order.
- `packages/typescript-client/src/snapshot-tracker.ts` — reconstructs
epoch-aware transaction IDs before filtering.
- `packages/typescript-client/test/pbt-micro.test.ts` — adds cross-epoch
property coverage and stream retirement regression.
- `packages/typescript-client/test/snapshot-tracker.test.ts` — adds a
wrapped xid retirement regression.
- `.changeset/tidy-xids-wrap.md` — records the client patch.

---

## Related

Related: [#2320](#2320)

_AI disclosure: Codex helped investigate, test, and prepare this
change._

---------

Co-authored-by: Kyle Mathews <mathews.kyle@gmail.com>

Author: neoantox

.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 {