diff --git a/.changeset/electric-cleanup-stream-listeners.md b/.changeset/electric-cleanup-stream-listeners.md new file mode 100644 index 0000000000..4faea3325e --- /dev/null +++ b/.changeset/electric-cleanup-stream-listeners.md @@ -0,0 +1,5 @@ +--- +'@tanstack/electric-db-collection': patch +--- + +Release the shape stream's visibility listener when a sync run is cleaned up. The Electric client registers a `visibilitychange` listener for each `ShapeStream` (an `AppState` subscription on React Native, from `@electric-sql/client` 1.5.23) and removes it only in `unsubscribeAll()`, so aborting the stream left the listener behind, and the listener kept the stream reachable. An app that cleans up and recreates collections, or whose on-demand cache rotates its provider session, gained one retained stream each time. diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md index a085dd75d9..0b63c00a68 100644 --- a/docs/contributing/oracle-coverage.md +++ b/docs/contributing/oracle-coverage.md @@ -636,7 +636,7 @@ generic visitor behavior outside the named collectors. | QueryRef operators and user-value boundaries | `packages/db/tests/query/subquery-user-value-oracle.test.ts`, `packages/db/tests/transactions.test.ts`, [repair record](oracle-reviews/2026-09-30-queryref-publication-repair.md) | A finite array/Set model checks the compiled output bag for a joined DISTINCT subquery with an outer WHERE. Live-query drivers check selected containers, no-select, functional, and predicate-literal user objects with IR-like fields, top-level and aggregate-subquery proxy-shaped fields, arrays of selected references, outer virtual-field filters on joined DISTINCT and nested aggregate QueryRefs, and a renamed no-select source. The transaction suite documents the existing development-browser duplicate-load guard. The bounded cases run in `test:oracles` or the full DB suite. Joined DISTINCT, nested and flat aggregate source-update histories pass. Ordered joined `findOne()` checks initial, singleton, empty, restored, and changed-join-key cuts; an unordered `findOne()` QueryRef on either side of a join checks default-key candidate deletion and reinsertion. A materialized joined `findOne()` supplies a separate receiving formulation. Other `singleResult` forms, join forms, schedules, and cross-copy value handling outside the duplicate-load guard remain unproved. | | Opaque backend pagination | [window oracle](https://github.com/TanStack/db/blob/main/packages/query-db-collection/tests/cursor-pagination.oracle.test.ts), [cache histories](https://github.com/TanStack/db/blob/main/packages/query-db-collection/tests/cursor-pagination.cache-oracle.test.ts), [cache publication](https://github.com/TanStack/db/blob/main/packages/query-db-collection/tests/cursor-pagination.publication-oracle.test.ts), [browser acquisition boundaries](https://github.com/TanStack/db/blob/main/packages/query-db-collection/tests/cursor-pagination.boundary-oracle.test.ts), [QueryCollection integration](https://github.com/TanStack/db/blob/main/packages/query-db-collection/tests/cursor-pagination-oracle.integration.test.ts) | Full filter/sort/slice reference, opaque token transport, actual Query cache expiry/invalidation/GC, forced refresh during growth, protocol failure publication/recovery, bounded slice work, nested cancellation/replacement, reader abort, browser retry defaults, manual-write cache isolation, and production window publications. Stable backend sequences; not snapshot guarantees for changing endpoints. Peek-ahead remains enabled. | | Experimental no-peek continuation | [test-only live-window probe](https://github.com/TanStack/db/blob/main/packages/query-db-collection/tests/cursor-pagination-oracle.no-peek.integration.test.ts) | The candidate uses an authoritative fact only for the same complete output prefix and otherwise requests the extra row. It remains test-only; the shipped opaque-pagination owner keeps peek-ahead. See [experiment limits](https://github.com/TanStack/db/blob/main/packages/query-db-collection/tests/cursor-pagination/NO-PEEK-RESULTS.md). | -| Electric and TrailBase | [Electric histories](https://github.com/TanStack/db/blob/main/packages/electric-db-collection/tests/electric-oracle.property.test.ts), [recovery histories](https://github.com/TanStack/db/blob/main/packages/electric-db-collection/tests/electric-recovery-oracle.test.ts), [held resume snapshots](https://github.com/TanStack/db/blob/main/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts), [PostgreSQL semantics](https://github.com/TanStack/db/blob/main/packages/electric-db-collection/e2e/sql-predicate-semantics-oracle.e2e.test.ts), [TrailBase contract](https://github.com/TanStack/db/blob/main/packages/trailbase-db-collection/tests/ORACLE.md) | Installed SDK delivery/framing, independent predicates, exact subscription arguments, restart/reset lineage, held certification and durability races, source-order publication before durability, and late errors. The queued-presence property runs identical fixed/random generators plus isolated seed-and-path replay across insert, update, delete, and truncate callbacks. The recovery fixtures use a mocked ShapeStream; they do not establish live Electric-service framing or native persistence-host behavior. The TrailBase lifecycle oracle's early-demand lane checks that an on-demand load issued before the change listener runs does not fetch until the listener runs or cannot start, through a real live query and the subset boundary: 16 histories cross listen, subscribe rejection, reader failure, abort, and cleanup with a service change before the subscription; the adapter from before the wait fails all 16. Its page-event lane checks that a page read before a concurrent change event neither reverts nor resurrects that row, and its stream-end lane pins that the adapter does not resubscribe: 17 histories, 6 of which fail on the adapter before the page fence. A generated page lane (`trailbase.page-event`) crosses two-page loads, cursor and offset paging, overlapping on-demand loads, and up to four changes among page releases. A held-commit lane checks that a page never fails a load for an older event that core holds while a user mutation persists; deciding insert or update from visible rows fails it. After a generated history drains, the Collection holds exactly the rows a page returned or an event touched. Both page lanes deliver events for changes made before the list call at once, during the page, or after it applies, and check that the listener survives a late insert for a row a page already holds. The retry-decision property (`electric.retry-decision`) generates histories over full-log and progressive waits, with and without `onError`, crossing synchronous and asynchronous retries and give-ups, a throwing `onError`, a later error after a retry, an Electric client give-up, an external abort before or during a decision, new demands, the first up-to-date, a persisting optimistic transaction that holds the retried snapshot and later settles, and cleanup. The retry-decision grammar also crosses eager and changes-only sources, an external signal already aborted at sync start, and a status listener that cleans up inside the failure; it checks that a readiness wait rejects with the same cause as every waiting demand, and that such a cleanup leaves a rejected demand rejected. It also holds the acceptance of the first up-to-date commit across later errors, retries, give-ups, and aborts, and checks that the Collection becomes ready with the snapshot's rows when the commit is accepted, and fails with the acceptance's cause when it is not. The descriptor-isolation oracle holds that acceptance across a managed-cache rotation and checks that the retired session's acceptance settles nothing, whether it resolves or rejects, and that the replacement session's snapshot decides readiness. A progressive snapshot's truncate applies at once; a demand whose snapshot arrived before an external abort settles when its rows apply, and cleanup after the abort rejects it. An independent model checks the status, every demand's settlement and cause, and row visibility after each step: a source that still tries stays `loading`, and only a give-up fails its waits. `main` before #2127 and mutants that fail before the decision, skip the failure on a throw or give-up, or skip `markError` on a client give-up or an external abort each fail it at an assertion. An installed-client lane in the SDK delivery oracle drives the same model (`electric-retry-decision-model.ts`) through the real Electric client and HTTP: a retried 503, a response without Electric headers, and an exhausted 50-retry budget, in eager, full-log, and progressive modes, with a synchronous or asynchronous retry (18 cells). Skipping `markError` on the client's give-up fails the 12 give-up cells; failing before the decision, or `main` before #2127, fails 15; the 3 survivors give up in the same continuation that decides, so an early failure is not observable there. A mutant that settles a held demand on acceptance instead of visibility fails the held replays. Excluded: stream steps while `onError` decides or after a final failure, which the client cannot produce. | +| Electric and TrailBase | [Electric histories](https://github.com/TanStack/db/blob/main/packages/electric-db-collection/tests/electric-oracle.property.test.ts), [recovery histories](https://github.com/TanStack/db/blob/main/packages/electric-db-collection/tests/electric-recovery-oracle.test.ts), [held resume snapshots](https://github.com/TanStack/db/blob/main/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts), [PostgreSQL semantics](https://github.com/TanStack/db/blob/main/packages/electric-db-collection/e2e/sql-predicate-semantics-oracle.e2e.test.ts), [TrailBase contract](https://github.com/TanStack/db/blob/main/packages/trailbase-db-collection/tests/ORACLE.md) | Installed SDK delivery/framing, independent predicates, exact subscription arguments, restart/reset lineage, held certification and durability races, source-order publication before durability, and late errors. The queued-presence property runs identical fixed/random generators plus isolated seed-and-path replay across insert, update, delete, and truncate callbacks. The recovery fixtures use a mocked ShapeStream; they do not establish live Electric-service framing or native persistence-host behavior. The TrailBase lifecycle oracle's early-demand lane checks that an on-demand load issued before the change listener runs does not fetch until the listener runs or cannot start, through a real live query and the subset boundary: 16 histories cross listen, subscribe rejection, reader failure, abort, and cleanup with a service change before the subscription; the adapter from before the wait fails all 16. Its page-event lane checks that a page read before a concurrent change event neither reverts nor resurrects that row, and its stream-end lane pins that the adapter does not resubscribe: 17 histories, 6 of which fail on the adapter before the page fence. A generated page lane (`trailbase.page-event`) crosses two-page loads, cursor and offset paging, overlapping on-demand loads, and up to four changes among page releases. A held-commit lane checks that a page never fails a load for an older event that core holds while a user mutation persists; deciding insert or update from visible rows fails it. After a generated history drains, the Collection holds exactly the rows a page returned or an event touched. Both page lanes deliver events for changes made before the list call at once, during the page, or after it applies, and check that the listener survives a late insert for a row a page already holds. The retry-decision property (`electric.retry-decision`) generates histories over full-log and progressive waits, with and without `onError`, crossing synchronous and asynchronous retries and give-ups, a throwing `onError`, a later error after a retry, an Electric client give-up, an external abort before or during a decision, new demands, the first up-to-date, a persisting optimistic transaction that holds the retried snapshot and later settles, and cleanup. The retry-decision grammar also crosses eager and changes-only sources, an external signal already aborted at sync start, and a status listener that cleans up inside the failure; it checks that a readiness wait rejects with the same cause as every waiting demand, and that such a cleanup leaves a rejected demand rejected. It also holds the acceptance of the first up-to-date commit across later errors, retries, give-ups, and aborts, and checks that the Collection becomes ready with the snapshot's rows when the commit is accepted, and fails with the acceptance's cause when it is not. The descriptor-isolation oracle holds that acceptance across a managed-cache rotation and checks that the retired session's acceptance settles nothing, whether it resolves or rejects, and that the replacement session's snapshot decides readiness. A progressive snapshot's truncate applies at once; a demand whose snapshot arrived before an external abort settles when its rows apply, and cleanup after the abort rejects it. An independent model checks the status, every demand's settlement and cause, and row visibility after each step: a source that still tries stays `loading`, and only a give-up fails its waits. `main` before #2127 and mutants that fail before the decision, skip the failure on a throw or give-up, or skip `markError` on a client give-up or an external abort each fail it at an assertion. An installed-client lane in the SDK delivery oracle drives the same model (`electric-retry-decision-model.ts`) through the real Electric client and HTTP: a retried 503, a response without Electric headers, and an exhausted 50-retry budget, in eager, full-log, and progressive modes, with a synchronous or asynchronous retry (18 cells). Skipping `markError` on the client's give-up fails the 12 give-up cells; failing before the decision, or `main` before #2127, fails 15; the 3 survivors give up in the same continuation that decides, so an early failure is not observable there. A mutant that settles a held demand on acceptance instead of visibility fails the held replays. Excluded: stream steps while `onError` decides or after a final failure, which the client cannot produce. The source-release property (`electric.sdk-source-release`) checks that cleanup removes every `visibilitychange` listener that the installed client registered for a sync run's streams: one to three runs in eager, on-demand, and progressive modes, each answering up to three requests with an up-to-date or a retried 503, then staying open, giving up, or aborting the external signal before cleanup, with restarts through `preload()`. It compares the live listener count with the number of active runs after each step. `main` before the fix, and mutants that release only while the stream is unaborted or only without a final initial failure, fail it at cleanup; the managed cache rotation case receives the law for a retired session. It observes the browser listener in jsdom, not React Native's AppState adapter. | | PowerSync | [tests](https://github.com/TanStack/db/tree/main/packages/powersync-db-collection/tests), `tests/correctness-oracle.test.ts` | Applied receipt positions crossed with held peers, native SQLite/SDK and cleanup evidence. A metadata-bearing `Collection.update` passes through an observed SDK watcher callback to the public and SQLite rows and an exact queued CRUD patch. Held callback ordering and synthetic zero-field cases remain controlled-only, and remote upload has no receiving fixture here. Run the focused owner with the package's `test:oracles` command. A timeout mutant proves a progress failure, not every value assertion. | | SQLite persistence and native hosts | [persisted histories](https://github.com/TanStack/db/blob/main/packages/db-sqlite-persistence-core/tests/persisted-oracle.test.ts), [reset/resume histories](https://github.com/TanStack/db/blob/main/packages/db-sqlite-persistence-core/tests/sqlite-core-adapter-oracle.test.ts), [setup reopen](https://github.com/TanStack/db/blob/main/packages/db-sqlite-persistence-core/tests/collection-setup-reopen-oracle.test.ts), [reload class](https://github.com/TanStack/db/blob/main/packages/db-sqlite-persistence-core/tests/reload-class-oracle.ts), [dual-adapter resume snapshots](https://github.com/TanStack/db/blob/main/packages/db-sqlite-persistence-core/tests/sqlite-resume-snapshot-oracle.test.ts), [Browser composed-owner histories](https://github.com/TanStack/db/blob/main/packages/browser-db-sqlite-persistence/tests/per-collection-coordinator-oracle.test.ts), [Browser coordinator RPC](https://github.com/TanStack/db/blob/main/packages/browser-db-sqlite-persistence/tests/browser-coordinator-oracle.test.ts), [shared-driver fairness](https://github.com/TanStack/db/blob/main/packages/browser-db-sqlite-persistence/tests/shared-driver-fairness-oracle.test.ts), [driver contracts](https://github.com/TanStack/db/blob/main/packages/db-sqlite-persistence-core/tests/contracts/sqlite-driver-contract.ts), [Node shared-handle scheduling](https://github.com/TanStack/db/blob/main/packages/node-db-sqlite-persistence/tests/node-driver.test.ts), [OP-SQLite shared-handle scheduling](https://github.com/TanStack/db/blob/main/packages/react-native-db-sqlite-persistence/tests/op-sqlite-driver-oracle.test.ts), [browser OPFS lifecycle](https://github.com/TanStack/db/blob/main/packages/browser-db-sqlite-persistence/tests/opfs-page-lifecycle-oracle.test.ts), [worker diagnostics](https://github.com/TanStack/db/blob/main/packages/browser-db-sqlite-persistence/tests/opfs-worker-diagnostics-oracle.test.ts), [request order](https://github.com/TanStack/db/blob/main/packages/browser-db-sqlite-persistence/tests/opfs-request-order-oracle.ts), [Electron IPC and composed owner](https://github.com/TanStack/db/blob/main/packages/electron-db-sqlite-persistence/tests/electron-ipc.test.ts), [113-law manifest](https://github.com/TanStack/db/blob/main/packages/db-collection-e2e/src/fixtures/persisted-conformance-manifest.ts) | Core cache/remote rejection/peer/reopen histories, atomic reset/resume lineage, key-set evidence, dual-adapter races, queued coordinator reload/gap notifications during startup, unscheduled startup/reset overlap, and exact driver results. The setup-reopen owner checks that a fresh adapter's first read repairs an interrupted setup, every subset of missing indexes, triggers, and bookkeeping rows, and a schema change to the fresh-install schema with rows kept or reset; complete setup costs at most one driver call over a cached read. It excludes cache-generation claims and concurrent adapters; the held-registration resume cases cover a collected generation at the setup DDL and at the registry read. The reload class owner keeps the reload law, model, histories, driver loop, and comparison in one module, with four production drivers: the Node Browser driver (real Browser persistence and coordinator, one connection per tab), the Node Electron driver (real renderers and main handler with an in-process `ipcMain` double), the real Chromium OPFS page, and a real Electron app (renderer over real IPC to a main process running the Node persistence over `node:sqlite`). Histories cover 1 to 4 eager or managed Collections with any subset of missing derived setup facts, a schema change, a database missing a migrated column or index, lock grants 1 to 5 tasks after the request, statement latency where the host controls it, a local write during another startup read (issued before or after its grant), a write held in flight at another Collection's grant, a follower write, a write from a third tab, takeover (also while a follower read is held after an earlier follower or third-tab write, and while a follower that knows the departed leader's route has a source commit queued behind a held subset read, which it flushes after the grant or which the departed leader received and never answered), restart, and a tab closed while its terms wait. It checks skipped and repaired Collection and database setup, that no term runs ahead of work queued on its tab at its grant unless a leader need is live in that election (a need that an earlier election served exempts none), that needs elect and settle, that a takeover's queued source commit settles within one RPC timeout (one timeout and a retry when the departed leader received it), and an exact per-reload budget: one registry read per adapter, one row scan per tab, two startup snapshots per eager Collection (metadata before the source starts, rows at the baseline hydrate, both required by the Electric resume laws), no rescan after a commit the tab already knows, and no database setup statement when it is complete. The implementation on origin/main fails every fixed history: the original 24 on all four drivers, and the 5 that cover per-election needs on the Node Browser driver. Production on this branch passes all 29 on all four drivers. The Node drivers reject 14 production mutants by assertion, including a lasting need latch, an earlier served need from this tab or another, a need still waiting at the grant that is ignored, and a grant that keeps the departed leader's route. The Browser coordinator RPC owner's per-election cells also reject those latches and a delayed heartbeat from the departed leader that restores its route during an election. Real Chromium rejects the never-waiting barrier across its histories; the real Electron app rejects it in 10 of the 29 fixed histories; in the others, real IPC finishes the startup reads before any grant. Hold seams patch the adapter in the process that runs it. Not owned here: a barrier that skips queued but not yet running regular work, which driver transaction order and the writer lock hide in reload histories; the shared logical scheduling oracle owns it. The real Electron app uses `node:sqlite` because Electron's ABI cannot load better-sqlite3; a native better-sqlite3 build for Electron is not run. The shared core-adapter contract checks custom local comparator dispatch for ascending, descending, and comparator-equal public-key ordering at `loadSubset` completion. Host packages register this contract, but the local sqlite3 CLI run does not prove native host execution. Browser composes public source commits with per-collection elected-owner routing and covers the complete committed-transaction wire partition through deterministic Node transport seams. Remote-subset histories distinguish logical demand, physical acquisitions, exact acquisition leases, and released replay tombstones. Electron composes source commits with a per-collection renderer owner, IPC adapter, real SQLite, and reopen checks. The shared-driver fairness owner checks K=1 complete-logical-hydrate scheduling with identical fixed, random, and seed-plus-path campaigns. It records public rows, raw dequeue reach, and a persist-first FIFO hostile control. Its Chromium OPFS fixture refines the provider boundary but does not establish a browser matrix, elapsed-time latency, unbounded eventuality, or multi-process coordination. Same-handle Node and OP-SQLite tests cover transaction admission. Controlled OPFS page/worker histories cover ownership and diagnostic-cause retention. The reset/resume owners use sqlite3 CLI and in-memory node:sqlite seams; they do not prove multi-process WAL, mobile/Tauri, or other native-device execution. Distinct database handles rely on SQLite lock admission rather than one in-process queue. React Native hosts without async-context propagation must use the transaction driver supplied to the callback for nested work. Fake workers and synthetic page events do not prove native handle release or real bfcache admission. The Browser composed seams are not real multi-context/OPFS-worker execution; the Electron harness is not an actual Electron process unless its explicit runtime-bridge mode runs. An ownerless elected node suppresses core routing, while a follower may route demand to the elected owner. The follower generation-notice and rotated-reacquisition lanes state the rotation law: a rotated follower holds the new generation's commit notices until the owner has reacquired every active demand, then ends scoped recovery and applies them. Production does not yet end that recovery, so their cells that require it are pinned as failing under the known gap [#2139](https://github.com/TanStack/db/issues/2139); the random generation-notice campaign is omitted until the fix, because some of its histories pass today. Cells that still pass cover failed reacquisitions (scoped recovery stays and the failure is reported), a superseding rotation as leader, and a scoped recovery without rotation, with or without an active demand (untrusted durable rows stay hidden). Other cells cover a follower whose claim expires while its ensure is in flight and an owner reset (the follower requests its demand again). The follower remote-demand lane checks that a follower's load, and a live query over its demand, settle only after the owner's broadcast applies here, for a broadcast that applies at once, one queued behind another subset's hydration, and one while the cache claim is expired; the claim-rotation placement is pinned under #2139. A failed apply rejects the load, errors the query, and fails the source. A renewal that fails transiently at the reply rejects the load but keeps and retries the demand, without releasing the owner's acquisition; a rotation during the reply wait is tracked under #2139. The flush-side expiry check and arrival-order queueing have no killing witness yet. Removing the follower's wait on either ensure path, its terminal check, or the queued flush's terminal marking each fails it. The Electron IPC readiness oracle checks a fresh or reopened renderer, a successful or failed IPC hydration, and an upstream load that succeeds, fails, or is released during hydration, against a model of the three readiness checkpoints; a renderer that replies before the main-process read or treats an IPC failure as empty fails it. The owner is a controlled coordinator double, not two real coordinators; host coordinators retry only classified transport or admission failures while demand remains retained. The manifest excludes progressive and move suites; registration and shim runs are not device execution. The per-collection coordinator oracle's real-tab lane runs a follower and an owner tab with real Browser coordinators over Node's BroadcastChannel and Web Locks and one real SQLite file: 16 histories over the owner's source (custom or Query Collection), the broadcast's placement, the follower's apply outcome, and the consumer, plus a rotation cell. Main fails all 16 histories; an owner tab for a rotated follower generation is not reachable there. The Electron IPC readiness oracle adds a follower and an owner renderer with Electron coordinators: 4 histories over the apply outcome and the consumer, in process and through a real Electron main process; main fails all 4. | | SQLite expression-index planning | [Node expression-index oracle](https://github.com/TanStack/db/blob/main/packages/node-db-sqlite-persistence/tests/expression-index-oracle.test.ts) | RFC #1659 invariant 8 owns identical persisted-index and runtime-expression shapes. Independent expected keys are checked against direct captured SQL, adapter results, and named-index plans. Generated BigInts use SQLite's signed range; one fixed case checks legacy oversized-value reads. Other limits: bounded unqualified JSON paths/scalars, Node BetterSQLite, and no null, arbitrary raw SQL, or native-host planning. Run the package's `test:oracles` campaign. | diff --git a/packages/db/tests/oracle-config.ts b/packages/db/tests/oracle-config.ts index 46e6efb6b2..eba93fa818 100644 --- a/packages/db/tests/oracle-config.ts +++ b/packages/db/tests/oracle-config.ts @@ -40,6 +40,7 @@ const staticOracleProperties = [ `electric.persistence-interleaving`, `electric.sdk-snapshot-delivery`, `electric.sdk-dnf-membership`, + `electric.sdk-source-release`, `electric-recovery.publication-stream-convergence`, `sqlite-resume.startup-generation`, `sqlite-resume.managed-claim-read`, diff --git a/packages/db/tests/oracle-replay-manifest.ts b/packages/db/tests/oracle-replay-manifest.ts index 7bcf5f9337..94f1e1b606 100644 --- a/packages/db/tests/oracle-replay-manifest.ts +++ b/packages/db/tests/oracle-replay-manifest.ts @@ -77,7 +77,7 @@ const ownerGroups: ReadonlyArray = [ [ `electric-db-collection/tests/electric-sdk-delivery-oracle.property.test.ts`, `electric`, - `sdk-snapshot-delivery sdk-dnf-membership`, + `sdk-snapshot-delivery sdk-dnf-membership sdk-source-release`, ], [ `db/tests/collection-sync-reentrancy-oracle.test.ts`, diff --git a/packages/electric-db-collection/src/electric.ts b/packages/electric-db-collection/src/electric.ts index bb5f8b6048..df8a804a3e 100644 --- a/packages/electric-db-collection/src/electric.ts +++ b/packages/electric-db-collection/src/electric.ts @@ -2756,6 +2756,11 @@ function createElectricSync>( shapeOptions.signal?.removeEventListener(`abort`, forwardExternalAbort) // Unsubscribe from the stream unsubscribeStream() + // Aborting the stream does not remove the visibility listener it + // registered when it was constructed, and that listener keeps the + // stream reachable. The stream is private to this session, so + // unsubscribeAll affects no other consumer. + stream.unsubscribeAll() // Abort the abort controller to stop the stream abortController.abort() if (transactionStarted) { diff --git a/packages/electric-db-collection/tests/async-hold-outcomes.test.ts b/packages/electric-db-collection/tests/async-hold-outcomes.test.ts index 42cd1f356b..96b0d9bc48 100644 --- a/packages/electric-db-collection/tests/async-hold-outcomes.test.ts +++ b/packages/electric-db-collection/tests/async-hold-outcomes.test.ts @@ -36,6 +36,7 @@ vi.mock(`@electric-sql/client`, async () => { ...actual, ShapeStream: vi.fn(() => ({ subscribe: mockSubscribe, + unsubscribeAll: vi.fn(), fetchSnapshot: mockFetchSnapshot, requestSnapshot: vi.fn(), })), diff --git a/packages/electric-db-collection/tests/electric-descriptor-isolation-oracle.test.ts b/packages/electric-db-collection/tests/electric-descriptor-isolation-oracle.test.ts index 6a4507b50a..922936f01e 100644 --- a/packages/electric-db-collection/tests/electric-descriptor-isolation-oracle.test.ts +++ b/packages/electric-db-collection/tests/electric-descriptor-isolation-oracle.test.ts @@ -184,6 +184,7 @@ vi.mock(`@electric-sql/client`, async () => { onStreamSubscribed?.() return unsubscribe }, + unsubscribeAll: vi.fn(), // ShapeStream.requestSnapshot is legal only in changes_only mode. // The installed-SDK delivery oracle is the receiving witness. requestSnapshot: vi.fn(() => { diff --git a/packages/electric-db-collection/tests/electric-live-query.test.ts b/packages/electric-db-collection/tests/electric-live-query.test.ts index eb76fa5b91..5e83711db1 100644 --- a/packages/electric-db-collection/tests/electric-live-query.test.ts +++ b/packages/electric-db-collection/tests/electric-live-query.test.ts @@ -81,6 +81,7 @@ function expectNoRepeatedSnapshotRequests() { const mockStream = { subscribe: mockSubscribe, + unsubscribeAll: vi.fn(), fetchSnapshot: mockFetchSnapshot, requestSnapshot: async (...args: any) => { const result = await mockRequestSnapshot(...args) diff --git a/packages/electric-db-collection/tests/electric-oracle.property.test.ts b/packages/electric-db-collection/tests/electric-oracle.property.test.ts index fa62022c93..51bef72249 100644 --- a/packages/electric-db-collection/tests/electric-oracle.property.test.ts +++ b/packages/electric-db-collection/tests/electric-oracle.property.test.ts @@ -162,6 +162,7 @@ const mockSubscribe = vi.fn() const shapeId = `{"params":{"table":"test_table"},"url":"http://test-url"}` const mockStream = { subscribe: mockSubscribe, + unsubscribeAll: vi.fn(), requestSnapshot: vi.fn().mockResolvedValue(undefined), fetchSnapshot: vi.fn().mockResolvedValue({ metadata: {}, data: [] }), forceDisconnectAndRefresh: vi.fn().mockResolvedValue(undefined), diff --git a/packages/electric-db-collection/tests/electric-recovery-oracle.test.ts b/packages/electric-db-collection/tests/electric-recovery-oracle.test.ts index cfeef06173..ecf23f543a 100644 --- a/packages/electric-db-collection/tests/electric-recovery-oracle.test.ts +++ b/packages/electric-db-collection/tests/electric-recovery-oracle.test.ts @@ -135,6 +135,7 @@ vi.mock(`@electric-sql/client`, async () => ({ ...(await vi.importActual(`@electric-sql/client`)), ShapeStream: vi.fn(() => ({ subscribe: mockSubscribe, + unsubscribeAll: vi.fn(), requestSnapshot: vi.fn().mockResolvedValue(undefined), fetchSnapshot: vi.fn().mockResolvedValue({ metadata: {}, data: [] }), forceDisconnectAndRefresh: vi.fn().mockResolvedValue(undefined), diff --git a/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts b/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts index 4d8174da91..d26bfe1696 100644 --- a/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts +++ b/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts @@ -33,6 +33,7 @@ vi.mock(`@electric-sql/client`, async () => ({ ...(await vi.importActual(`@electric-sql/client`)), ShapeStream: vi.fn(() => ({ subscribe: mockSubscribe, + unsubscribeAll: vi.fn(), requestSnapshot: vi.fn(() => snapshotRequest?.() ?? Promise.resolve()), fetchSnapshot: vi.fn().mockResolvedValue({ metadata: {}, data: [] }), forceDisconnectAndRefresh: vi.fn().mockResolvedValue(undefined), diff --git a/packages/electric-db-collection/tests/electric-sdk-delivery-oracle.property.test.ts b/packages/electric-db-collection/tests/electric-sdk-delivery-oracle.property.test.ts index 832b2b8a0d..08e1e0ec64 100644 --- a/packages/electric-db-collection/tests/electric-sdk-delivery-oracle.property.test.ts +++ b/packages/electric-db-collection/tests/electric-sdk-delivery-oracle.property.test.ts @@ -18,6 +18,7 @@ import { oraclePropertyOptions, readOracleRunConfig, } from '../../db/tests/oracle-config' +import { createOracleReach } from '../../db/tests/oracle-reach' import { managedTagPersistence, storageTargetId, @@ -82,6 +83,8 @@ import type { Message } from '@electric-sql/client' * A real node:sqlite cold-launch witness starts with an incompatible partial * cache and no reset marker. It must exclude that row before and after the * installed SDK returns an empty source snapshot for the demanded predicate. + * A source-release lane, documented at its property, checks that cleanup + * removes every listener that a run's streams registered on `document`. * * These finite HTTP fixtures do not prove that a live Electric service emits * the authored snapshot or move frames, nor do they cover multiple row keys, @@ -309,6 +312,29 @@ function controlledHttp() { } } +// Records the `visibilitychange` listeners that code registers on `document`. +function recordVisibilityListeners() { + const add = vi.spyOn(document, `addEventListener`) + const remove = vi.spyOn(document, `removeEventListener`) + const visibility = (calls: Array>) => + calls + .filter(([type]) => type === `visibilitychange`) + .map(([, listener]) => listener) + return { + registered: () => visibility(add.mock.calls).length, + live: () => { + const removed = new Set(visibility(remove.mock.calls)) + return visibility(add.mock.calls).filter( + (listener) => !removed.has(listener), + ).length + }, + restore: () => { + add.mockRestore() + remove.mockRestore() + }, + } +} + // Deliver a coordinator notification through the same persistence wrapper // used by the installed-SDK receiver. The test still controls the provider // response independently of this durable-cache notification. @@ -835,10 +861,14 @@ fixedCase(`does not certify an aborted SDK subset snapshot`, async () => { * its own snapshots. This receives the controlled provider-restart law through * the installed ShapeStream and HTTP path. A Map-backed adapter supplies the * generation boundary; real SQLite and native hosts have separate owners. + * It also receives the stream-listener release law below for the retired + * session: after the rotation only the replacement stream's listener stays + * registered, and none stays after cleanup. */ fixedCase( `discards an old SDK snapshot after managed cache rotation`, async () => { + const listeners = recordVisibilityListeners() const http = controlledHttp() const rows = new Map>([ [`old-generation`, new Map()], @@ -1031,7 +1061,14 @@ fixedCase( expect(collection.get(1)).toBeUndefined() expect(collection.get(2)?.name).toBe(`fresh`) expect([...rows.get(`new-generation`)!.keys()]).toEqual([2]) - }, [() => collection.cleanup(), () => http.close()]) + expect(listeners.registered()).toBe(2) + expect(listeners.live(), `live listeners after rotation`).toBe(1) + }, [ + () => collection.cleanup(), + () => http.close(), + () => expect(listeners.live(), `live listeners after cleanup`).toBe(0), + () => listeners.restore(), + ]) }, ) @@ -2164,3 +2201,244 @@ fixedCase.each(clientRetryCells)( (outcome, mode, decision) => runClientRetryCell(outcome, mode, decision), 60_000, ) + +/** + * # Cleanup releases the installed client's stream listeners + * + * Cleanup ends a sync run and releases its adapter-owned resources + * (`CleanupFn` in `packages/db/src/types.ts`, and "Sync Implementation" in the + * collection options creator guide). The Electric adapter constructs one + * ShapeStream per provider session. The installed client registers a + * `visibilitychange` listener on `document` for each stream, so that the + * stream pauses while the page is hidden. That listener keeps the stream + * reachable until the client releases it. The law: after cleanup, `document` + * holds no `visibilitychange` listener from a stream of the cleaned-up run, + * whatever state the source reached. While a run is active, it holds at most + * the listener of its one stream. + * + * The model is the number of active sync runs, 0 or 1. It reads no adapter or + * client state. A run is active from its start until its cleanup, and each run + * constructs one stream, so the expected number of live listeners is at most + * the number of active runs, and exactly 0 once no run is active. The model + * does not predict when an active stream releases its listener, because the + * client may release it early, for example when its signal aborts. + * + * The history grammar has a sync mode and one to three sync runs. The first + * run starts with the Collection; each later run is a restart of the + * cleaned-up Collection through `preload()`. In each run the provider answers + * zero to three requests, each with a row and an up-to-date, or with a 503 + * that `onError` retries. The run then stays open, gives up on one more 503 + * because `onError` returns no retry, or aborts the external signal. Cleanup + * ends the run. The signal stays aborted, so a run that starts after an abort + * starts with an aborted signal and sends no request. The provider answers + * only a request the client sent. The grammar does not answer a request after + * a give-up or an abort, because the client sends none. + * + * The production driver uses a real Collection, the Electric adapter, the + * installed ShapeStream, and controlled HTTP. The recorder wraps + * `document.addEventListener` and `document.removeEventListener` and counts + * the `visibilitychange` listeners still registered. The comparison runs after + * each provider answer, give-up, abort, and cleanup, once the client has sent + * its next request or decided to send none. A calibration check at the end + * shows that the recorder saw one listener for each run. + * + * Limits: the lane observes the browser listener in jsdom. Client versions + * that observe React Native's AppState instead register through a runtime + * adapter, which this lane does not install. A live Electric service and + * snapshot requests are outside this lane. The managed cache rotation case + * above replaces a provider session within one run; it receives the same law + * for the retired session. + */ +type ReleaseRun = { + answers: Array<`up-to-date` | `retry`> + end: `open` | `give-up` | `abort` +} +type ReleaseHistory = { + mode: `eager` | `on-demand` | `progressive` + runs: Array +} +type ReleaseState = + | `cleanup-while-loading` + | `cleanup-after-ready` + | `cleanup-after-give-up` + | `cleanup-after-abort` + | `cleanup-of-restarted-run` + +const createReleaseReach = () => + createOracleReach(`electric.sdk-source-release`, [ + `cleanup-while-loading`, + `cleanup-after-ready`, + `cleanup-after-give-up`, + `cleanup-after-abort`, + `cleanup-of-restarted-run`, + ]) +// Each campaign declares its own reach. +let releaseReach = createReleaseReach() + +async function checkSourceRelease(history: ReleaseHistory): Promise { + releaseReach.beginHistory() + const listeners = recordVisibilityListeners() + const http = controlledHttp() + const external = new AbortController() + let retries = true + let onErrorCalls = 0 + const collection = createCollection( + electricCollectionOptions({ + id: `sdk-release-${++sequence}`, + shapeOptions: { + url: `http://test-url/release-${sequence}`, + params: { table: `rows` }, + fetchClient: http.fetchClient, + signal: external.signal, + backoffOptions: { + initialDelay: 0, + maxDelay: 0, + multiplier: 1, + maxRetries: 0, + }, + onError: () => { + onErrorCalls++ + return retries ? {} : undefined + }, + }, + syncMode: history.mode, + startSync: true, + getKey: (row) => row.id, + }), + ) + + // The model is `activeRuns`. `sending` and `ready` track the source only to + // choose legal provider steps and to classify reach. + let activeRuns = 1 + let runsStarted = 1 + let sending = true + let ready = false + let offset = 0 + + const check = (label: string) => + expect(listeners.live(), `${label}: live listeners`).toBeLessThanOrEqual( + activeRuns, + ) + const next = (label: string) => + atCheckpoint(http.take(), `${label} HTTP request`) + + const consoleError = vi.spyOn(console, `error`).mockImplementation(() => {}) + const consoleWarn = vi.spyOn(console, `warn`).mockImplementation(() => {}) + await withElectricCleanup(async () => { + for (const [index, run] of history.runs.entries()) { + if (index > 0) { + void collection.preload().catch(() => undefined) + activeRuns = 1 + runsStarted++ + sending = !external.signal.aborted + ready = false + } + let request = sending ? await next(`run ${index} start`) : undefined + for (const answer of run.answers) { + if (!request) break + if (answer === `up-to-date`) { + offset++ + request.respond( + new Response( + JSON.stringify([ + { + key: `1`, + value: { id: `1`, name: `row` }, + headers: { operation: `insert` }, + }, + { + headers: { + control: `up-to-date`, + global_last_seen_lsn: String(offset), + }, + }, + ]), + { headers: headers(offset) }, + ), + ) + ready = true + } else { + request.respond(new Response(`transient`, { status: 503 })) + } + request = await next(`run ${index} ${answer}`) + check(`run ${index} ${answer}`) + } + if (request && run.end === `give-up`) { + const calls = onErrorCalls + retries = false + request.respond(new Response(`transient`, { status: 503 })) + await vi.waitFor(() => expect(onErrorCalls).toBe(calls + 1)) + retries = true + sending = false + check(`run ${index} give-up`) + } else if (run.end === `abort` && !external.signal.aborted) { + external.abort() + sending = false + check(`run ${index} abort`) + } + + releaseReach.reach( + external.signal.aborted + ? `cleanup-after-abort` + : !sending + ? `cleanup-after-give-up` + : ready + ? `cleanup-after-ready` + : `cleanup-while-loading`, + ) + if (index > 0) releaseReach.reach(`cleanup-of-restarted-run`) + await collection.cleanup() + activeRuns = 0 + check(`run ${index} cleanup`) + } + // Calibration: the recorder observed the listener of every run's stream. + expect(listeners.registered()).toBe(runsStarted) + }, [ + () => collection.cleanup(), + () => http.close(), + () => listeners.restore(), + () => consoleError.mockRestore(), + () => consoleWarn.mockRestore(), + ]) +} + +const releaseViolation = sameViolation() +const releaseProperty = fc.asyncProperty( + fc.record({ + mode: fc.constantFrom( + `eager` as const, + `on-demand` as const, + `progressive` as const, + ), + runs: fc.array( + fc.record({ + answers: fc.array( + fc.constantFrom(`up-to-date` as const, `retry` as const), + { maxLength: 3 }, + ), + end: fc.constantFrom( + `open` as const, + `give-up` as const, + `abort` as const, + ), + }), + { minLength: 1, maxLength: 3 }, + ), + }), + (history: ReleaseHistory) => + releaseViolation.check(history, () => checkSourceRelease(history)), +) +sdkCampaigns( + `electric.sdk-source-release`, + `releases every installed-SDK stream listener when a sync run is cleaned up`, + 20261024, + releaseViolation, + async (seed) => { + releaseReach = createReleaseReach() + await fc.assert(releaseProperty, { + ...oraclePropertyOptions(40, `electric.sdk-source-release`), + ...(seed === undefined ? {} : { seed }), + }) + releaseReach.assertReached() + }, +) diff --git a/packages/electric-db-collection/tests/electric.test.ts b/packages/electric-db-collection/tests/electric.test.ts index 3f0aa80c5f..7c30e63c3a 100644 --- a/packages/electric-db-collection/tests/electric.test.ts +++ b/packages/electric-db-collection/tests/electric.test.ts @@ -44,6 +44,7 @@ const mockFetchSnapshot = vi.fn() const mockForceDisconnectAndRefresh = vi.fn() const mockStream = { subscribe: mockSubscribe, + unsubscribeAll: vi.fn(), requestSnapshot: mockRequestSnapshot, fetchSnapshot: mockFetchSnapshot, forceDisconnectAndRefresh: mockForceDisconnectAndRefresh, diff --git a/packages/electric-db-collection/tests/tags.test.ts b/packages/electric-db-collection/tests/tags.test.ts index 48af9c6b15..65cfafc13a 100644 --- a/packages/electric-db-collection/tests/tags.test.ts +++ b/packages/electric-db-collection/tests/tags.test.ts @@ -20,6 +20,7 @@ const mockRequestSnapshot = vi.fn() const mockFetchSnapshot = vi.fn() const mockStream = { subscribe: mockSubscribe, + unsubscribeAll: vi.fn(), requestSnapshot: mockRequestSnapshot, fetchSnapshot: mockFetchSnapshot, }