add txid tracking tests

Author: samwillis

packages/db-collection-e2e/src/suites/progressive.suite.ts

@@ -1,5 +1,5 @@
 /**
- * Progressive Mode Test Suite (Electric only)
+ * Progressive Mode Test Suite
  *
  * Tests progressive sync mode behavior including:
  * - Snapshot loading during initial sync
@@ -11,12 +11,14 @@
 import { describe, expect, it } from "vitest"
 import { createLiveQueryCollection, eq, gt } from "@tanstack/db"
 import { waitFor, waitForQueryData } from "../utils/helpers"
-import type { E2ETestConfig } from "../types"
+import type { E2ETestConfig, User } from "../types"
+import type { Collection } from "@tanstack/db"
+import type { ElectricCollectionUtils } from "@tanstack/electric-db-collection"
 
 export function createProgressiveTestSuite(
   getConfig: () => Promise<E2ETestConfig>
 ) {
-  describe(`Progressive Mode Suite (Electric only)`, () => {
+  describe(`Progressive Mode Suite`, () => {
     describe(`Basic Progressive Mode`, () => {
       it(`should explicitly validate snapshot phase and atomic swap transition`, async () => {
         const config = await getConfig()
@@ -406,6 +408,145 @@ export function createProgressiveTestSuite(
       })
     })
 
+    describe(`Txid Tracking Behavior (Electric only)`, () => {
+      it(`should not track txids during snapshot phase but track them after atomic swap`, async () => {
+        const config = await getConfig()
+        if (
+          !config.collections.progressive ||
+          !config.mutations?.insertUser ||
+          !config.getTxid
+        ) {
+          return // Skip if progressive collections, mutations, or getTxid not available
+        }
+        const progressiveUsers = config.collections.progressive
+          .users as Collection<User, string, ElectricCollectionUtils>
+
+        // awaitTxId is guaranteed to exist on ElectricCollectionUtils
+        // This test is Electric-only via the describe block name
+
+        // Start sync but don't release yet (stay in snapshot phase)
+        progressiveUsers.startSyncImmediate()
+        await new Promise((resolve) => setTimeout(resolve, 100))
+
+        // Should be in loading state (snapshot phase)
+        if (progressiveUsers.status !== `loading`) {
+          console.log(
+            `Collection already ready, cannot test snapshot phase txid behavior`
+          )
+          return
+        }
+
+        // === PHASE 1: INSERT DURING SNAPSHOT PHASE ===
+        const snapshotPhaseUser = {
+          id: crypto.randomUUID(),
+          name: `Snapshot Phase User`,
+          email: `[email protected]`,
+          age: 28,
+          isActive: true,
+          createdAt: new Date(),
+          metadata: null,
+          deletedAt: null,
+        }
+
+        // Insert user and track when awaitTxId completes
+        let txidResolved = false
+
+        // Start the insert
+        await config.mutations.insertUser(snapshotPhaseUser)
+
+        // Get the txid from postgres
+        const txid = await config.getTxid()
+
+        if (!txid) {
+          console.log(`Could not get txid, skipping txid tracking validation`)
+          config.progressiveTestControl?.releaseInitialSync()
+          return
+        }
+
+        // Start awaiting the txid (should NOT resolve during snapshot phase)
+        progressiveUsers.utils.awaitTxId(txid, 60000).then(() => {
+          txidResolved = true
+        })
+
+        // Wait a moment for sync to process
+        await new Promise((resolve) => setTimeout(resolve, 500))
+
+        // Txid should NOT have resolved yet (snapshot phase, txids not tracked)
+        expect(txidResolved).toBe(false)
+
+        // Query for the user (triggers fetchSnapshot with this user)
+        const query = createLiveQueryCollection((q) =>
+          q
+            .from({ user: progressiveUsers })
+            .where(({ user }) => eq(user.id, snapshotPhaseUser.id))
+        )
+
+        await query.preload()
+        await waitForQueryData(query, { minSize: 1, timeout: 10000 })
+
+        // User should be in snapshot data
+        expect(query.size).toBe(1)
+        expect(query.get(snapshotPhaseUser.id)).toBeDefined()
+
+        // But collection is still in snapshot phase
+        expect(progressiveUsers.status).toBe(`loading`)
+
+        // Txid should STILL not have resolved (snapshot doesn't track txids)
+        expect(txidResolved).toBe(false)
+
+        // === PHASE 2: TRIGGER ATOMIC SWAP ===
+        if (config.progressiveTestControl) {
+          config.progressiveTestControl.releaseInitialSync()
+        }
+
+        // Wait for atomic swap to complete
+        await waitFor(() => progressiveUsers.status === `ready`, {
+          timeout: 30000,
+          message: `Progressive collection did not complete sync`,
+        })
+
+        // NOW txid should resolve (buffered messages include txids)
+        await waitFor(() => txidResolved, {
+          timeout: 5000,
+          message: `Txid did not resolve after atomic swap`,
+        })
+
+        expect(txidResolved).toBe(true)
+
+        // === PHASE 3: VERIFY TXID TRACKING POST-SWAP ===
+        // User should still be present after atomic swap
+        expect(progressiveUsers.get(snapshotPhaseUser.id)).toBeDefined()
+
+        // Now insert another user and verify txid tracking works
+        const postSwapUser = {
+          id: crypto.randomUUID(),
+          name: `Post Swap User`,
+          email: `[email protected]`,
+          age: 29,
+          isActive: true,
+          createdAt: new Date(),
+          metadata: null,
+          deletedAt: null,
+        }
+
+        await config.mutations.insertUser(postSwapUser)
+
+        // Wait for incremental update (txid tracking should work now)
+        if (config.hasReplicationLag) {
+          await waitFor(() => progressiveUsers.has(postSwapUser.id), {
+            timeout: 10000,
+            message: `Post-swap user not synced via incremental update`,
+          })
+        }
+
+        // Both users should be present
+        expect(progressiveUsers.get(snapshotPhaseUser.id)).toBeDefined()
+        expect(progressiveUsers.get(postSwapUser.id)).toBeDefined()
+
+        await query.cleanup()
+      })
+    })
+
     describe(`Progressive Mode Resilience`, () => {
       it(`should handle cleanup and restart during snapshot phase`, async () => {
         const config = await getConfig()

packages/db-collection-e2e/src/types.ts

@@ -76,6 +76,9 @@ export interface E2ETestConfig {
     insertPost: (post: Post) => Promise<void>
   }
 
+  // Helper to get txid for Electric txid tracking tests (Electric only)
+  getTxid?: () => Promise<number | null>
+
   // Indicates if the backend has replication lag (e.g., Electric sync)
   // When true, tests will wait for mutations to propagate before proceeding
   hasReplicationLag?: boolean

packages/electric-db-collection/e2e/electric.e2e.test.ts

@@ -431,6 +431,16 @@ describe(`Electric Collection E2E Tests`, () => {
           commentsUpToDateControl.current?.()
         },
       },
+      getTxid: async () => {
+        // Get the current transaction ID from the last operation
+        // This uses pg_current_xact_id_if_assigned() which returns the txid
+        // Note: This gets the CURRENT transaction's ID, so must be called
+        // immediately after an insert in the same transaction context
+        const result = await dbClient.query(
+          `SELECT pg_current_xact_id_if_assigned()::text::bigint as txid`
+        )
+        return result.rows[0]?.txid || null
+      },
       mutations: {
         // Use direct SQL for Electric tests (simulates external changes)
         // This tests that Electric sync picks up database changes