diff --git a/.changeset/cancel-electric-refresh-wait.md b/.changeset/cancel-electric-refresh-wait.md new file mode 100644 index 0000000000..311ce222a5 --- /dev/null +++ b/.changeset/cancel-electric-refresh-wait.md @@ -0,0 +1,5 @@ +--- +'@tanstack/electric-db-collection': patch +--- + +Cancel an on-demand refresh wait when its request or collection is cleaned up, preventing snapshots from starting after teardown. diff --git a/packages/electric-db-collection/src/electric.ts b/packages/electric-db-collection/src/electric.ts index 39e6540a6a..c478fe59e5 100644 --- a/packages/electric-db-collection/src/electric.ts +++ b/packages/electric-db-collection/src/electric.ts @@ -579,18 +579,21 @@ function createLoadSubsetDedupe>({ const loadSubset = async (opts: LoadSubsetOptions) => { const commitCursor = getCommitCursor() - if (opts.signal?.aborted) return + const isAborted = (): boolean => + signal.aborted || opts.signal?.aborted === true + if (isAborted()) return if (isBufferingInitialSync()) { const snapshotParams = compileSQL(opts, compileOptions) try { const { data: rows } = await stream.fetchSnapshot(snapshotParams) - if (opts.signal?.aborted || !isBufferingInitialSync()) { + if (isAborted() || !isBufferingInitialSync()) { debug(`${logPrefix}Ignoring snapshot - sync completed while fetching`) return } if (rows.length > 0) { + if (isAborted()) return begin() for (const row of rows) { write({ @@ -603,7 +606,7 @@ function createLoadSubsetDedupe>({ debug(`${logPrefix}Applied snapshot with ${rows.length} rows`) } } catch (error) { - if (opts.signal?.aborted) return + if (isAborted()) return if (handleSnapshotError(error, `fetchSnapshot`)) { return } @@ -625,10 +628,34 @@ function createLoadSubsetDedupe>({ // 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. + // still works. Cleanup or request cancellation ends the wait without starting + // a snapshot that no current demand can use. if (stream.isUpToDate) { let timeoutId: ReturnType | undefined + let removeAbortListeners = () => {} try { + const abortSignals = new Set( + [signal, opts.signal].filter( + (candidate): candidate is AbortSignal => candidate !== undefined, + ), + ) + const aborted = new Promise((resolve) => { + const onAbort = () => resolve() + if (Array.from(abortSignals).some((candidate) => candidate.aborted)) { + resolve() + return + } + + abortSignals.forEach((candidate) => + candidate.addEventListener(`abort`, onAbort, { once: true }), + ) + removeAbortListeners = () => { + abortSignals.forEach((candidate) => + candidate.removeEventListener(`abort`, onAbort), + ) + } + }) + await Promise.race([ stream.forceDisconnectAndRefresh(), new Promise((resolve) => { @@ -637,6 +664,7 @@ function createLoadSubsetDedupe>({ FORCE_DISCONNECT_AND_REFRESH_TIMEOUT_MS, ) }), + aborted, ]) } catch (error) { if (handleSnapshotError(error, `forceDisconnectAndRefresh`)) { @@ -647,11 +675,12 @@ function createLoadSubsetDedupe>({ error, ) } finally { + removeAbortListeners() clearTimeout(timeoutId) } } - if (opts.signal?.aborted) return + if (isAborted()) return // Upstream limitation: ShapeStream.requestSnapshot() publishes its rows // through the stream callback before its Promise resolves. It accepts no diff --git a/packages/electric-db-collection/tests/electric.test.ts b/packages/electric-db-collection/tests/electric.test.ts index c913f8d973..6d39559437 100644 --- a/packages/electric-db-collection/tests/electric.test.ts +++ b/packages/electric-db-collection/tests/electric.test.ts @@ -2659,6 +2659,20 @@ describe(`Electric Integration`, () => { // Tests for syncMode configuration describe(`syncMode configuration`, () => { + const createOnDemandCollection = (id: string) => + createCollection( + electricCollectionOptions({ + id, + shapeOptions: { + url: `http://test-url`, + params: { table: `test_table` }, + }, + syncMode: `on-demand`, + getKey: (item: Row) => item.id as number, + startSync: true, + }), + ) + it(`should not request snapshots during subscription in eager mode`, () => { vi.clearAllMocks() @@ -2896,6 +2910,166 @@ describe(`Electric Integration`, () => { } }) + it(`should cancel a pending refresh wait when the collection is cleaned up`, async () => { + vi.useFakeTimers() + const refresh = createDeferred() + + try { + mockStream.isUpToDate = true + mockForceDisconnectAndRefresh.mockReturnValueOnce(refresh.promise) + + const testCollection = createOnDemandCollection( + `on-demand-refresh-cleanup-test`, + ) + + let loadSettled = false + const load = Promise.resolve( + testCollection._sync.loadSubset({ limit: 10 }), + ).then(() => { + loadSettled = true + }) + + await Promise.resolve() + await testCollection.cleanup() + await vi.advanceTimersByTimeAsync(0) + + expect(loadSettled).toBe(true) + expect(mockRequestSnapshot).not.toHaveBeenCalled() + expect(vi.getTimerCount()).toBe(0) + + refresh.resolve() + await refresh.promise + await load + expect(mockRequestSnapshot).not.toHaveBeenCalled() + } finally { + refresh.resolve() + await vi.runOnlyPendingTimersAsync() + vi.useRealTimers() + } + }) + + it(`does not start buffered snapshot publication after adapter cleanup`, async () => { + const snapshot = createDeferred<{ + data: Array<{ + key: string + value: Row + headers: { operation: `insert` } + }> + }>() + mockFetchSnapshot.mockReturnValueOnce(snapshot.promise) + const options = electricCollectionOptions({ + id: `progressive-snapshot-cleanup-test`, + shapeOptions: { + url: `http://test-url`, + params: { table: `test_table` }, + }, + syncMode: `progressive`, + getKey: (item: Row) => item.id as number, + startSync: true, + }) + const begin = vi.fn() + const write = vi.fn() + const commit = vi.fn(() => true as const) + const controls = options.sync.sync({ + collection: { id: options.id, status: `loading` }, + begin, + write, + commit, + markReady: vi.fn(), + markError: vi.fn(), + truncate: vi.fn(), + } as never) + if (!controls || typeof controls === `function` || !controls.loadSubset) { + throw new Error(`Expected progressive sync controls`) + } + + const load = controls.loadSubset({ limit: 10 }) + controls.cleanup?.() + snapshot.resolve({ + data: [ + { + key: `1`, + value: { id: 1, name: `Late snapshot user` }, + headers: { operation: `insert` }, + }, + ], + }) + if (load !== true) await load + + expect(begin).not.toHaveBeenCalled() + expect(write).not.toHaveBeenCalled() + expect(commit).not.toHaveBeenCalled() + }) + + it(`does not start a refresh when the collection signal is already aborted`, async () => { + mockStream.isUpToDate = true + const abortController = new AbortController() + abortController.abort() + const testCollection = createCollection( + electricCollectionOptions({ + id: `on-demand-refresh-already-aborted-test`, + shapeOptions: { + url: `http://test-url`, + params: { table: `test_table` }, + signal: abortController.signal, + }, + syncMode: `on-demand`, + getKey: (item: Row) => item.id as number, + startSync: true, + }), + ) + + await testCollection._sync.loadSubset({ limit: 10 }) + + expect(mockForceDisconnectAndRefresh).not.toHaveBeenCalled() + expect(mockRequestSnapshot).not.toHaveBeenCalled() + await testCollection.cleanup() + }) + + it(`should retry a refresh wait after the requesting demand is aborted`, async () => { + vi.useFakeTimers() + const refresh = createDeferred() + + try { + mockStream.isUpToDate = true + mockForceDisconnectAndRefresh.mockReturnValueOnce(refresh.promise) + + const testCollection = createOnDemandCollection( + `on-demand-refresh-abort-retry-test`, + ) + const abortController = new AbortController() + let abortedLoadSettled = false + const abortedLoad = Promise.resolve( + testCollection._sync.loadSubset({ + limit: 10, + signal: abortController.signal, + }), + ).then(() => { + abortedLoadSettled = true + }) + + await Promise.resolve() + abortController.abort() + await vi.advanceTimersByTimeAsync(0) + + expect(abortedLoadSettled).toBe(true) + expect(mockRequestSnapshot).not.toHaveBeenCalled() + expect(vi.getTimerCount()).toBe(0) + + mockForceDisconnectAndRefresh.mockResolvedValueOnce(undefined) + await testCollection._sync.loadSubset({ limit: 10 }) + + expect(mockForceDisconnectAndRefresh).toHaveBeenCalledTimes(2) + expect(mockRequestSnapshot).toHaveBeenCalledTimes(1) + await testCollection.cleanup() + await abortedLoad + } finally { + refresh.resolve() + await vi.runOnlyPendingTimersAsync() + vi.useRealTimers() + } + }) + it(`should clear the refresh timeout when refresh settles early`, async () => { vi.useFakeTimers() try {