fix(electric): bound refresh wait for on-demand subsets

.changeset/tame-rn-live-poll-refresh.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/electric-db-collection': patch
+---
+
+Bound the wait for Electric stream refreshes before loading on-demand subsets so native fetch implementations that do not promptly abort long polls do not keep live queries loading until the poll times out.

packages/electric-db-collection/src/electric.ts

@@ -64,6 +64,8 @@ export { isChangeMessage, isControlMessage } from '@electric-sql/client'
 
 const debug = DebugModule.debug(`ts/db:electric`)
 
+const FORCE_DISCONNECT_AND_REFRESH_TIMEOUT_MS = 250
+
 /**
  * Symbol for internal test hooks (hidden from public API)
  */
@@ -481,11 +483,23 @@ function createLoadSubsetDedupe<T extends Row<unknown>>({
     // When the stream is already up-to-date, it may be in a long-poll wait.
     // Forcing a disconnect-and-refresh ensures requestSnapshot gets a response
     // from a fresh server round-trip rather than waiting for the current poll to end.
-    // If the refresh fails (e.g., PauseLock held during subscriber processing in
-    // join pipelines), we fall through to requestSnapshot which still works.
+    // Some native fetch implementations (notably React Native/Expo) may not abort
+    // long-poll requests promptly. Bound the wait so on-demand live queries don't
+    // remain loading until the long-poll naturally times out.
+    // If the refresh fails or times out, we fall through to requestSnapshot which
+    // still works.
     if (stream.isUpToDate) {
+      let timeoutId: ReturnType<typeof setTimeout> | undefined
       try {
-        await stream.forceDisconnectAndRefresh()
+        await Promise.race([
+          stream.forceDisconnectAndRefresh(),
+          new Promise<void>((resolve) => {
+            timeoutId = setTimeout(
+              resolve,
+              FORCE_DISCONNECT_AND_REFRESH_TIMEOUT_MS,
+            )
+          }),
+        ])
       } catch (error) {
         if (handleSnapshotError(error, `forceDisconnectAndRefresh`)) {
           return
@@ -494,6 +508,8 @@ function createLoadSubsetDedupe<T extends Row<unknown>>({
           `${logPrefix}forceDisconnectAndRefresh failed, proceeding to requestSnapshot: %o`,
           error,
         )
+      } finally {
+        clearTimeout(timeoutId)
       }
     }
 

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

@@ -2565,6 +2565,122 @@ describe(`Electric Integration`, () => {
       expect(mockRequestSnapshot).toHaveBeenCalledTimes(1)
     })
 
+    it(`should request the snapshot after the refresh timeout and ignore late fulfillment`, async () => {
+      vi.useFakeTimers()
+      try {
+        let resolveRefresh: () => void = () => {}
+        const refresh = new Promise<void>((resolve) => {
+          resolveRefresh = resolve
+        })
+        mockStream.isUpToDate = true
+        mockForceDisconnectAndRefresh.mockReturnValueOnce(refresh)
+
+        const testCollection = createCollection(
+          electricCollectionOptions({
+            id: `on-demand-refresh-timeout-fulfillment-test`,
+            shapeOptions: {
+              url: `http://test-url`,
+              params: { table: `test_table` },
+            },
+            syncMode: `on-demand`,
+            getKey: (item: Row) => item.id as number,
+            startSync: true,
+          }),
+        )
+
+        let loadSettled = false
+        const load = Promise.resolve(
+          testCollection._sync.loadSubset({ limit: 10 }),
+        ).then(() => {
+          loadSettled = true
+        })
+
+        await vi.advanceTimersByTimeAsync(249)
+        expect(mockRequestSnapshot).not.toHaveBeenCalled()
+        expect(loadSettled).toBe(false)
+
+        await vi.advanceTimersByTimeAsync(1)
+        await load
+        expect(mockRequestSnapshot).toHaveBeenCalledTimes(1)
+        expect(loadSettled).toBe(true)
+        await testCollection.cleanup()
+        expect(vi.getTimerCount()).toBe(0)
+
+        resolveRefresh()
+        await refresh
+        await Promise.resolve()
+        expect(mockRequestSnapshot).toHaveBeenCalledTimes(1)
+        expect(vi.getTimerCount()).toBe(0)
+      } finally {
+        vi.useRealTimers()
+      }
+    })
+
+    it(`should handle late refresh rejection after requesting the snapshot`, async () => {
+      vi.useFakeTimers()
+      try {
+        let rejectRefresh: (error: Error) => void = () => {}
+        const refresh = new Promise<void>((_resolve, reject) => {
+          rejectRefresh = reject
+        })
+        mockStream.isUpToDate = true
+        mockForceDisconnectAndRefresh.mockReturnValueOnce(refresh)
+
+        const testCollection = createCollection(
+          electricCollectionOptions({
+            id: `on-demand-refresh-timeout-rejection-test`,
+            shapeOptions: {
+              url: `http://test-url`,
+              params: { table: `test_table` },
+            },
+            syncMode: `on-demand`,
+            getKey: (item: Row) => item.id as number,
+            startSync: true,
+          }),
+        )
+
+        const load = testCollection._sync.loadSubset({ limit: 10 })
+        await vi.advanceTimersByTimeAsync(250)
+        await load
+
+        rejectRefresh(new Error(`late refresh failure`))
+        await expect(refresh).rejects.toThrow(`late refresh failure`)
+        await Promise.resolve()
+        expect(mockRequestSnapshot).toHaveBeenCalledTimes(1)
+        expect(vi.getTimerCount()).toBe(0)
+      } finally {
+        vi.useRealTimers()
+      }
+    })
+
+    it(`should clear the refresh timeout when refresh settles early`, async () => {
+      vi.useFakeTimers()
+      try {
+        mockStream.isUpToDate = true
+        mockForceDisconnectAndRefresh.mockResolvedValueOnce(undefined)
+
+        const testCollection = createCollection(
+          electricCollectionOptions({
+            id: `on-demand-refresh-clears-timeout-test`,
+            shapeOptions: {
+              url: `http://test-url`,
+              params: { table: `test_table` },
+            },
+            syncMode: `on-demand`,
+            getKey: (item: Row) => item.id as number,
+            startSync: true,
+          }),
+        )
+
+        await testCollection._sync.loadSubset({ limit: 10 })
+
+        expect(mockRequestSnapshot).toHaveBeenCalledTimes(1)
+        expect(vi.getTimerCount()).toBe(0)
+      } finally {
+        vi.useRealTimers()
+      }
+    })
+
     it(`should fetch snapshots in progressive mode when loadSubset is called before sync completes`, async () => {
       vi.clearAllMocks()