From 1ac77eb372026dff74c09bb0fa707d5cd012071d Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Tue, 25 Aug 2026 14:56:02 -0600 Subject: [PATCH 1/6] fix(electric): cancel bounded refresh waits --- .../electric-db-collection/src/electric.ts | 30 ++++- .../tests/electric.test.ts | 106 ++++++++++++++++++ 2 files changed, 134 insertions(+), 2 deletions(-) diff --git a/packages/electric-db-collection/src/electric.ts b/packages/electric-db-collection/src/electric.ts index 39e6540a6..1066be217 100644 --- a/packages/electric-db-collection/src/electric.ts +++ b/packages/electric-db-collection/src/electric.ts @@ -625,10 +625,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 +661,7 @@ function createLoadSubsetDedupe>({ FORCE_DISCONNECT_AND_REFRESH_TIMEOUT_MS, ) }), + aborted, ]) } catch (error) { if (handleSnapshotError(error, `forceDisconnectAndRefresh`)) { @@ -647,11 +672,12 @@ function createLoadSubsetDedupe>({ error, ) } finally { + removeAbortListeners() clearTimeout(timeoutId) } } - if (opts.signal?.aborted) return + if (signal.aborted || opts.signal?.aborted) 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 c913f8d97..6afe302d0 100644 --- a/packages/electric-db-collection/tests/electric.test.ts +++ b/packages/electric-db-collection/tests/electric.test.ts @@ -2896,6 +2896,112 @@ describe(`Electric Integration`, () => { } }) + it(`should cancel a pending refresh wait when the collection is cleaned up`, async () => { + vi.useFakeTimers() + let resolveRefresh: () => void = () => {} + const refresh = new Promise((resolve) => { + resolveRefresh = resolve + }) + + try { + mockStream.isUpToDate = true + mockForceDisconnectAndRefresh.mockReturnValueOnce(refresh) + + const testCollection = createCollection( + electricCollectionOptions({ + id: `on-demand-refresh-cleanup-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 Promise.resolve() + await testCollection.cleanup() + await vi.advanceTimersByTimeAsync(0) + + expect(loadSettled).toBe(true) + expect(mockRequestSnapshot).not.toHaveBeenCalled() + expect(vi.getTimerCount()).toBe(0) + + resolveRefresh() + await refresh + await load + expect(mockRequestSnapshot).not.toHaveBeenCalled() + } finally { + resolveRefresh() + await vi.runOnlyPendingTimersAsync() + vi.useRealTimers() + } + }) + + it(`should retry a refresh wait after the requesting demand is aborted`, async () => { + vi.useFakeTimers() + let resolveRefresh: () => void = () => {} + const refresh = new Promise((resolve) => { + resolveRefresh = resolve + }) + + try { + mockStream.isUpToDate = true + mockForceDisconnectAndRefresh.mockReturnValueOnce(refresh) + + const testCollection = createCollection( + electricCollectionOptions({ + id: `on-demand-refresh-abort-retry-test`, + shapeOptions: { + url: `http://test-url`, + params: { table: `test_table` }, + }, + syncMode: `on-demand`, + getKey: (item: Row) => item.id as number, + startSync: true, + }), + ) + 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 { + resolveRefresh() + await vi.runOnlyPendingTimersAsync() + vi.useRealTimers() + } + }) + it(`should clear the refresh timeout when refresh settles early`, async () => { vi.useFakeTimers() try { From a14459bd47fc24c45a03dfb8469aba6a554bd31f Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Tue, 25 Aug 2026 15:02:40 -0600 Subject: [PATCH 2/6] test(electric): simplify refresh lifecycle setup --- .../tests/electric.test.ts | 62 ++++++++----------- 1 file changed, 26 insertions(+), 36 deletions(-) diff --git a/packages/electric-db-collection/tests/electric.test.ts b/packages/electric-db-collection/tests/electric.test.ts index 6afe302d0..3abd7ccbf 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() @@ -2898,26 +2912,14 @@ describe(`Electric Integration`, () => { it(`should cancel a pending refresh wait when the collection is cleaned up`, async () => { vi.useFakeTimers() - let resolveRefresh: () => void = () => {} - const refresh = new Promise((resolve) => { - resolveRefresh = resolve - }) + const refresh = createDeferred() try { mockStream.isUpToDate = true - mockForceDisconnectAndRefresh.mockReturnValueOnce(refresh) + mockForceDisconnectAndRefresh.mockReturnValueOnce(refresh.promise) - const testCollection = createCollection( - electricCollectionOptions({ - id: `on-demand-refresh-cleanup-test`, - shapeOptions: { - url: `http://test-url`, - params: { table: `test_table` }, - }, - syncMode: `on-demand`, - getKey: (item: Row) => item.id as number, - startSync: true, - }), + const testCollection = createOnDemandCollection( + `on-demand-refresh-cleanup-test`, ) let loadSettled = false @@ -2935,12 +2937,12 @@ describe(`Electric Integration`, () => { expect(mockRequestSnapshot).not.toHaveBeenCalled() expect(vi.getTimerCount()).toBe(0) - resolveRefresh() - await refresh + refresh.resolve() + await refresh.promise await load expect(mockRequestSnapshot).not.toHaveBeenCalled() } finally { - resolveRefresh() + refresh.resolve() await vi.runOnlyPendingTimersAsync() vi.useRealTimers() } @@ -2948,26 +2950,14 @@ describe(`Electric Integration`, () => { it(`should retry a refresh wait after the requesting demand is aborted`, async () => { vi.useFakeTimers() - let resolveRefresh: () => void = () => {} - const refresh = new Promise((resolve) => { - resolveRefresh = resolve - }) + const refresh = createDeferred() try { mockStream.isUpToDate = true - mockForceDisconnectAndRefresh.mockReturnValueOnce(refresh) + mockForceDisconnectAndRefresh.mockReturnValueOnce(refresh.promise) - const testCollection = createCollection( - electricCollectionOptions({ - id: `on-demand-refresh-abort-retry-test`, - shapeOptions: { - url: `http://test-url`, - params: { table: `test_table` }, - }, - syncMode: `on-demand`, - getKey: (item: Row) => item.id as number, - startSync: true, - }), + const testCollection = createOnDemandCollection( + `on-demand-refresh-abort-retry-test`, ) const abortController = new AbortController() let abortedLoadSettled = false @@ -2996,7 +2986,7 @@ describe(`Electric Integration`, () => { await testCollection.cleanup() await abortedLoad } finally { - resolveRefresh() + refresh.resolve() await vi.runOnlyPendingTimersAsync() vi.useRealTimers() } From 37e69fb1eb9858956c00ce1c737786cd33cf2921 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Tue, 25 Aug 2026 15:03:16 -0600 Subject: [PATCH 3/6] chore: add Electric refresh cancellation changeset --- .changeset/cancel-electric-refresh-wait.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/cancel-electric-refresh-wait.md diff --git a/.changeset/cancel-electric-refresh-wait.md b/.changeset/cancel-electric-refresh-wait.md new file mode 100644 index 000000000..311ce222a --- /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. From 01f187b4d3372450063a9ae4a1894cfae96718b4 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Wed, 26 Aug 2026 07:32:02 -0600 Subject: [PATCH 4/6] fix(electric): skip aborted refresh startup --- .../electric-db-collection/src/electric.ts | 6 ++-- .../tests/electric.test.ts | 35 ++++++++++++++++++- 2 files changed, 38 insertions(+), 3 deletions(-) diff --git a/packages/electric-db-collection/src/electric.ts b/packages/electric-db-collection/src/electric.ts index 1066be217..e0686a8b0 100644 --- a/packages/electric-db-collection/src/electric.ts +++ b/packages/electric-db-collection/src/electric.ts @@ -579,7 +579,9 @@ 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) @@ -677,7 +679,7 @@ function createLoadSubsetDedupe>({ } } - if (signal.aborted || 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 3abd7ccbf..c804bcdf1 100644 --- a/packages/electric-db-collection/tests/electric.test.ts +++ b/packages/electric-db-collection/tests/electric.test.ts @@ -2659,7 +2659,15 @@ describe(`Electric Integration`, () => { // Tests for syncMode configuration describe(`syncMode configuration`, () => { - const createOnDemandCollection = (id: string) => + const createOnDemandCollection = ( + id: string, + ): Collection< + Row, + number, + ElectricCollectionUtils, + StandardSchemaV1, + Row + > => createCollection( electricCollectionOptions({ id, @@ -2948,6 +2956,31 @@ describe(`Electric Integration`, () => { } }) + 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() From eac96d1abd84f3a99fda24a0811ea1ef41978d7e Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Wed, 26 Aug 2026 07:41:19 -0600 Subject: [PATCH 5/6] test(electric): preserve inferred collection key type --- packages/electric-db-collection/tests/electric.test.ts | 10 +--------- 1 file changed, 1 insertion(+), 9 deletions(-) diff --git a/packages/electric-db-collection/tests/electric.test.ts b/packages/electric-db-collection/tests/electric.test.ts index c804bcdf1..96f549fa0 100644 --- a/packages/electric-db-collection/tests/electric.test.ts +++ b/packages/electric-db-collection/tests/electric.test.ts @@ -2659,15 +2659,7 @@ describe(`Electric Integration`, () => { // Tests for syncMode configuration describe(`syncMode configuration`, () => { - const createOnDemandCollection = ( - id: string, - ): Collection< - Row, - number, - ElectricCollectionUtils, - StandardSchemaV1, - Row - > => + const createOnDemandCollection = (id: string) => createCollection( electricCollectionOptions({ id, From 1bfba4e1cd10f1caae2cca2f34c0d235c8cec3cd Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Wed, 26 Aug 2026 09:22:46 -0600 Subject: [PATCH 6/6] fix(electric): stop snapshots after cleanup --- .../electric-db-collection/src/electric.ts | 5 +- .../tests/electric.test.ts | 53 +++++++++++++++++++ 2 files changed, 56 insertions(+), 2 deletions(-) diff --git a/packages/electric-db-collection/src/electric.ts b/packages/electric-db-collection/src/electric.ts index e0686a8b0..c478fe59e 100644 --- a/packages/electric-db-collection/src/electric.ts +++ b/packages/electric-db-collection/src/electric.ts @@ -587,12 +587,13 @@ function createLoadSubsetDedupe>({ 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({ @@ -605,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 } diff --git a/packages/electric-db-collection/tests/electric.test.ts b/packages/electric-db-collection/tests/electric.test.ts index 96f549fa0..6d3955943 100644 --- a/packages/electric-db-collection/tests/electric.test.ts +++ b/packages/electric-db-collection/tests/electric.test.ts @@ -2948,6 +2948,59 @@ describe(`Electric Integration`, () => { } }) + 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()