diff --git a/.changeset/cold-electric-mutation-admission.md b/.changeset/cold-electric-mutation-admission.md new file mode 100644 index 0000000000..738d351d1c --- /dev/null +++ b/.changeset/cold-electric-mutation-admission.md @@ -0,0 +1,7 @@ +--- +"@tanstack/db": patch +"@tanstack/db-sqlite-persistence-core": patch +"@tanstack/electric-db-collection": patch +--- + +Establish on-demand Electric stream positions before dispatching optimistic mutation handlers, including explicit multi-collection transactions. Revalidate positions after resets and prevent dispatch after cancellation or source retirement without preloading rows or retrying backend writes. diff --git a/docs/collections/electric-collection.md b/docs/collections/electric-collection.md index 210615386c..91157af5e9 100644 --- a/docs/collections/electric-collection.md +++ b/docs/collections/electric-collection.md @@ -64,6 +64,21 @@ Each handler should call `await collection.utils.awaitTxId(txid)` to wait for sy Handlers persist mutations to the backend and wait for Electric to sync the changes back. This prevents UI glitches where optimistic updates would be removed and then re-added. TanStack DB blocks sync data until the mutation is confirmed, ensuring smooth user experience. +For on-demand collections, optimistic changes appear immediately, but the +mutation handler waits until each participating Electric stream has established +its continuation position. You do not need to query or preload a collection +before inserting into it. This also applies to explicit transactions spanning +several collections. A stream reset before dispatch requires a fresh position; +rollback or a terminal source failure before dispatch prevents the write. + +This preparation waits for stream evidence, not for source rows to become +visible. A handler must still await its transaction ID or match, and must not +wait for an on-demand query whose publication is held by its own transaction. +Persisted collections also wait for persistence startup to finish before +dispatch. This does not wait for row visibility. Cold writes require sync +connectivity. If confirmation fails after the handler has sent a write, the backend may already have committed it; the library does +not automatically repeat the handler. + ### 1. Using Txid (Recommended) The recommended approach uses PostgreSQL transaction IDs (txids) for precise matching. The backend returns a txid, and the client waits for that specific txid to appear in the Electric stream. diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md index fbea9cd98c..abffb8bdb4 100644 --- a/docs/contributing/oracle-coverage.md +++ b/docs/contributing/oracle-coverage.md @@ -6627,6 +6627,37 @@ retains exact source/evidence hashes and all ORC outcomes. Actual production bundle erasure, Typedoc generation, tooling GREEN and restored-old-tooling mutants remain required receiving checks. Generated docs remain CI-owned. +### Electric cold mutation admission + +The `electric.sdk-mutation-startup` property in +`packages/electric-db-collection/tests/electric-sdk-delivery-oracle.property.test.ts` +checks the proposed stream-position admission contract through the installed +Electric SDK and controlled HTTP. Every generated history exercises direct and +manual inserts, a derived materialized descriptor after its original owner +retires, two participating collections in either release order, cancellation +before readiness, external abort, a prepared participant that resets while its +sibling waits, and retirement of a prepared participant. Three receiving paths +cross the real persistence wrapper with controlled managed storage: held startup +metadata followed by success, rejection with the same storage error, and cleanup +before the source enters. A twelfth path retires the run from its readiness +callback after an active batch was accepted: its transaction evidence remains +available, but its optimistic write is never dispatched. This is distinct from +a callback entering after retirement, which has no publication authority. +Every history reaches all thirteen declared paths, including a coordinator +index held after metadata and the Electric baseline arrive. Mutation dispatch +waits for that remaining persistence startup work to settle. +At each held baseline the mutation handler has not run; after all live source +positions are established it runs once and stays persisting until its txid echo. +The payload and release order vary in equal fixed/random campaigns. This owner +does not establish offline delivery, automatic write retries, native SQLite I/O, +prompt cancellation of held storage I/O, GC ownership, or arbitrary delayed +callback/restart interleavings. The persisted cleanup observation follows storage +release and cleanup completion; it promises no subsequent backend dispatch. +The same persisted-cleanup path records cache-claim requests and live claims, +plus coordinator subscriptions. Late metadata settlement cannot restart storage +or retain either resource. The generated release-order flag also varies whether +cleanup settles before metadata is released. This is controlled managed storage; +it does not establish native SQLite claim release or timer behavior. ### Replay successful-retirement ownership, October 30 diff --git a/docs/contributing/oracle-reviews/2026-10-30-electric-mutation-admission.md b/docs/contributing/oracle-reviews/2026-10-30-electric-mutation-admission.md new file mode 100644 index 0000000000..e9cff0cb63 --- /dev/null +++ b/docs/contributing/oracle-reviews/2026-10-30-electric-mutation-admission.md @@ -0,0 +1,254 @@ +# Electric mutation admission evidence + +Reviewed implementation and tests: `10fe53ec642e738625daf6dba8a2672eb0f63fc6`. +Parent: `9cfa11fc712866f8f276e85762865ebb8bd49cbe`. + +This proposes the on-demand admission contract described in the Electric +collection guide: optimism is immediate; dispatch waits for all current stream +positions. This is a new ordering guarantee, not a claim that prior documentation +already required it. Handler confirmation remains a separate obligation. + +## RED, GREEN, and mutant + +Run from `packages/electric-db-collection`: + +```sh +pnpm exec vitest run tests/electric-sdk-delivery-oracle.property.test.ts \ + -t 'admits cold mutations' --coverage.enabled=false --typecheck.enabled=false +``` + +The generated extension ran before production edits. On the parent both +campaigns fail `no write before baseline: direct`: expected0, observed1. +The smallest history has an empty name and `reverse=false`. + +After the initial repair, review added a persisted receiving witness before +changing the wrapper. It failed `admission remains live: persisted` with +`StreamAbortedError` while metadata was still held. The final composition waits +for source entry and propagates the exact storage failure. + +The final fixed and random campaigns pass. Each runs8 histories through all11 +paths: direct, explicit transaction, multi-collection, reset, retirement, +rollback, abort, persisted success/failure/cleanup, and materialized descriptor +reuse after the original owner retires. Each path reaches8 times per campaign; +`createOracleReach` enforces reach. Payloads are strings of length0–8; release +order varies. Removing the second source loses the all-participants rule; +removing the held echo loses the admission/confirmation distinction. Existing +row admission, invalid missing-key updates, backend retries, native SQLite, +and arbitrary schedules are outside this grammar. + +Replacing the final `guards.every(...)` with unconditional dispatch fails the +`replacement cursor still held` assertion (observed1, expected0). It is an +assertion failure at the intended checkpoint, not a timeout or setup failure. +Direct replay was verified against that mutant and restored production: + +```sh +TANSTACK_DB_ORACLE_PROPERTY=electric.sdk-mutation-startup \ +TANSTACK_DB_ORACLE_SEED=20261030 TANSTACK_DB_ORACLE_PATH=0:0 \ +pnpm exec vitest run tests/electric-sdk-delivery-oracle.property.test.ts \ + --coverage.enabled=false --typecheck.enabled=false +``` + +The mutant fails one replay; restored production passes one replay. Other +campaigns stand aside. The final startup campaigns also pass after the loop's +lint-only change from `while (true)` to `for (;;)`. No budgets were lowered. + +## Review against the oracle rubric + +| Requirements | Evidence or limit | +| --- | --- | +| ORC-001–003 | Proposed authority and limits are stated above and beside the shared checker. The independent rule relates dispatched writes to supplied stream boundaries, without inspecting production preparation state. | +| ORC-004 | Eleven legal paths reconstruct the reported cold trace, multi-source reset, persistence failure, and descriptor reuse. Payload and release order vary; comments describe the axes and exclusions. | +| ORC-005 | Real `createCollection`, explicit transactions, installed ShapeStream, and managed-persistence wrapper receive controlled HTTP/storage. Dispatch, transaction settlement, error identity, and final rows are observed at held boundaries. | +| ORC-006 | Original premature-dispatch failure and stale-guard mutant are rejected at their intended assertions. | +| ORC-007 | Same property and budget run fixed/random; seed/path replay verified as above. No command-model replayPath applies. | +| ORC-008–009 | No second production lifecycle model is introduced. The checker uses external admission boundaries; Collection, sync run, retirement, and settlement retain glossary meanings. | +| ORC-010 | Existing `sameViolation` and `withElectricCleanup` preserve the first mismatch and secondary cleanup failures. Resources are released for each history. | +| ORC-011 | No additional shared-semantic-fault hypothesis requiring a second independent formulation was identified. Generated success alone is not a universal proof. | +| ORC-012 | This versioned record identifies the reviewed source/test commit and bounded evidence. | +| ORC-013 | Held versus released baseline and reset versus replacement-baseline boundaries distinguish early dispatch. A held echo distinguishes admission from confirmation. | +| ORC-014 | Claim ends at installed SDK plus controlled HTTP and managed persistence. It does not establish live-service framing, native SQLite, prompt cancellation of held storage I/O, or arbitrary reconnect schedules. | +| ORC-015–016 | Declared reaches cover every path. The existing SDK oracle owns one shared generated checker; no standalone example expectations were added as separate oracle tests. | + +## Broader validation and production weight + +- 38 adapter test files:1,132 tests pass, no type errors. Four files using + `node:sqlite` fail to load under local default jsdom/Vite; the same setup + failure occurs on unchanged parent. Those files were excluded from this run. +- Eight focused core transaction/ownership/GC suites:534 tests pass. +- Adapter and persistence package `tsc --noEmit` pass. +- Oracle lint, formatting, and the repository pre-commit checks pass. +- Size comparison: gzip +116B core, +185B Electric, +31B persistence; typical + app scenario +115B. The final loop spelling has identical control flow. + +The implementation reuses the existing transaction and Electric lifecycles, +current descriptor binding, and persistence source-start promise. It adds no +write retry queue or GC ownership layer. Waiting for Collection readiness was +rejected because publication can wait on the same optimistic transaction; +preloading rows would also defeat on-demand behavior. The added bytes implement +stream-position preparation and its invalidation, plus wrapper composition. + + +## Review follow-up: accepted evidence across readiness cleanup + +Review base: `006b65bac772c020de657c872a49922743986697`. +CodeRabbit comment4421557210 identified that the added late lifecycle fence +also discarded transaction evidence from a batch admitted while active. The +parent adapter published that evidence after the ready callback. The correction +preserves that behavior while fencing mutation readiness and match publication +against a retired run. Callbacks entering after retirement remain rejected by +the existing entry fence. + +Before the source edit, the generated startup checker was extended with +`ready-cleanup`: insert an optimistic row, hold the first HTTP response, then +supply a source batch containing txid100 whose ready callback cleans up the +Collection. The admitted batch's txid must remain queryable after cleanup; +the pending optimistic mutation must reject without dispatch. Both campaigns +failed with `TimeoutWaitingForTxIdError` for100 on the review base. This is the +specific acknowledgement timeout under test, not a generic harness timeout. + +The repair guards only `resolveMutationReady` and `commitMatches`. It publishes +and clears the accepted batch's evidence regardless of callback retirement. +The existing shared checker now runs twelve paths in each of8 fixed and8 random +histories, with declared reach. Payload and participant-release order remain the +same generated dimensions. The contract and evidence limits above remain; +this adds one synchronous reentry boundary, not arbitrary stale callbacks. + + +## November 2: upstream merge and complete persisted startup + +Merged upstream main `d9d0a8e` in `4911f7b`, preserving both coverage-map +additions. The only merge conflict was that shared documentation append. + +- Law: the proposed mutation-preparation contract waits for persistence startup + as well as current Electric stream positions, without waiting for row visibility. + The Electric collection guide now states this composition explicitly. +- Reach: the existing startup grammar adds `persisted-index`. A real Collection + index causes coordinator startup work to remain held after metadata and the + installed SDK's baseline arrive. All thirteen paths reach eight times in each + fixed/random campaign; payload and participant order still vary. +- RED: the focused startup command above fails both campaigns before the source + change. Seed20261030, path0:0 observes one handler invocation where zero is + required at `no write before persisted startup completes`. +- Repair: after source entry, mutation preparation also awaits the runtime's + existing `ensureStarted()` promise. Source entry keeps its existing metadata + boundary; no startup queue or readiness state is added. +- GREEN: the same campaigns pass. The four affected adapter suites pass408 + runtime/type tests with no type errors; persistence `tsc --noEmit` passes. +- Mutant: removing only the added startup wait reproduces the same assertion + failure in direct seed20261030/path0:0 replay. Restoring it passes that replay. +- Limits: the new receiving path uses controlled coordinator scheduling and + managed storage plus the installed Electric SDK. It establishes held-versus- + released startup ordering, not a new native-SQLite or arbitrary-failure claim. + +The original oracle missed this boundary because its persistence paths held +metadata only; none left coordinator startup pending after source entry. +CodeRabbit's broader claim about late index failures was not reproduced: index +request failures are currently caught by persistence. This repair addresses the +verified premature-dispatch boundary and adopts full startup promise rejection +without changing index failure policy. + +Self-review: ORC-001–007 and ORC-013–016 retain the prior contract, shared checker, +installed-SDK path, campaign parity, replay, and cleanup limits. The new held +coordinator request distinguishes source-entry-only admission from complete +startup admission. ORC-008 adds no model state; ORC-009 uses the existing terms. +ORC-010 cleanup releases the new held index before cleaning up the Collection. +ORC-011 identified no separate shared-semantic-fault hypothesis. This record +supplies ORC-012 evidence for the bounded added path; it is not an exhaustive +audit of unrelated upstream changes. + +Reviewed packages/db-sqlite-persistence-core/src/persisted.ts SHA256: `7445fdf60921cd04feb1f1aa22415b3f637545b084223ba2b7af9f19328194ab`. + +Reviewed packages/electric-db-collection/tests/electric-sdk-delivery-oracle.property.test.ts SHA256: `cf58a63b38f465d3c558d55e3b62daf5984a7c6035615c26fc44803e84e781e7`. + + +## November 3: stale persistence preparation and upstream merge + +Source baseline: PR head `8f6abecd5c7430e09ced81f7304292cb48673f79`, merged with +upstream `36f879727d594601e28329e177c802dc179bf9e9`. This entry describes the +source and oracle changes in the commit containing this entry. Self-review. + +- **Law:** Collection cleanup invalidates pending source preparation. Late + metadata settlement must not restart the retired persistence runtime or retain + a cache claim/coordinator subscription. Authority: the cleanup lifecycle in + the glossary and the proposed mutation-admission contract in the Electric + guide. Successful startup must still dispatch after metadata and index startup. +- **Reach:** the existing `persisted-cleanup` receiving path now observes a live + cache claim before cleanup, then completed cleanup and transaction settlement. + All thirteen paths run eight times in each fixed/random campaign. The release + flag varies metadata release before/after cleanup settlement; payloads remain + strings of length 0–8. No run budget was reduced. +- **RED:** before the persistence repair, both generated startup campaigns fail + `retired preparation cannot restart storage`: two claim requests versus one. + The previous checker observed only blocked dispatch and missed resource + acquisition after cleanup. The extension ran before the production repair. +- **GREEN:** the lifecycle-generation check before `ensureStarted()` makes both + startup campaigns pass. Resource expectations come from the cleanup law; + claim/subscription recorders observe external collaborator calls, without + inspecting the runtime's generation or preparation machinery. +- **Mutant:** removing only the new generation check fails direct replay with + seed `20261030`, path `0:0`, property `electric.sdk-mutation-startup`, at the same + claim-request assertion. Restoring it passes the same replay. This is an + assertion failure at the intended checkpoint, not a setup error or timeout. +- **Limits:** the installed Electric SDK receives controlled HTTP and the real + persistence wrapper receives controlled managed storage. No live-service, + native SQLite resource-release, renewal-timer, arbitrary restart scheduling, + or prompt storage-I/O cancellation claim follows. + +Commands from `packages/electric-db-collection` (Node available on PATH): + +```sh +node ../../node_modules/vitest/vitest.mjs run tests/electric-sdk-delivery-oracle.property.test.ts \ + -t 'admits cold mutations' --coverage.enabled=false --typecheck.enabled=false +TANSTACK_DB_ORACLE_PROPERTY=electric.sdk-mutation-startup \ +TANSTACK_DB_ORACLE_SEED=20261030 TANSTACK_DB_ORACLE_PATH=0:0 \ +node ../../node_modules/vitest/vitest.mjs run tests/electric-sdk-delivery-oracle.property.test.ts \ + --coverage.enabled=false --typecheck.enabled=false +``` + +ORC-001–005: the existing contract, driver, grammar and independent admission +judgment remain; cleanup adds external resource observations at settlement. +Removing the cleanup path loses the stale-continuation witness; removing the +successful persisted/index paths loses the adjacent valid-startup witnesses. +Cleanup precedes source entry; cleanup without an admitted Collection operation +is outside this grammar. ORC-006–007: RED and restored direct replay above, +with unchanged fixed/random budgets. ORC-008–009: no model state or vocabulary +was renamed or merged. ORC-010: existing first-violation reporting and ordered +cleanup remain. ORC-011: no shared semantic classifier is used for the zero-live- +resources rule, and no concrete second-formulation hypothesis was identified. +ORC-012: this append-only entry is versioned with the source/test changes. +ORC-013: live startup versus retired startup distinguishes the guard from an +unconditional abort. ORC-014: the boundary stays explicitly controlled. +ORC-015–016: existing declared per-path reach is retained, with cleanup reach +moved to the settlement boundary; no standalone examples were added. + +The merge retains upstream's retry health and retired-evidence policy alongside +mutation-gate invalidation. In particular, retirement no longer advances the +Electric epoch: upstream failure reporting may still belong to that retired run, +and the next start advances the epoch. External abort only rejects preparation +for its originating active run. + + +Validation: 42 adapter files / 1,219 runtime tests pass (the command excluded +`electric-sdk-delivery.test.ts` and `electric-descriptor-isolation-oracle.test.ts`; +two other exclude patterns matched no files). The focused four-file run passes +400 runtime tests. Those initial runs also reported three missing local build +artifacts during typechecking; production package builds were then generated +before rerunning typechecking. Persistence `tsc --noEmit` passes after linking +its Browser workspace dependency. Changed-source ESLint, oracle lint, formatting, +and `git diff --check` pass. No generated build output is committed. + +Production size against upstream `36f879727`: core +117B gzip, Electric +196B, +persistence +59B; total +372B gzip (+0.2%), +295B brotli. This runs the +repository size comparator with `--exact-base --no-scenarios`; no new app-scenario +claim follows. The local runtime lacks `npx`, so a temporary shell shim routes +its `vite build` calls to the installed workspace binary. The lifecycle check +uses the existing runtime generation and abort error; no extra lifecycle, +retry path, or queue is introduced. Diff cleanup against current upstream finds +no formatting-only churn or unused mechanism; all eleven existing per-file +justifications still apply. + +The final focused type-only run passes all four files (126 checks, no type +errors or unhandled source errors). Expo example tsconfig warnings remain from +the filtered dependency installation and do not fail these checks. The merge +commit skips lint-staged's whole incoming-upstream formatting pass; changed +source ESLint, oracle lint, formatting, type checks and diff checks ran directly. diff --git a/packages/db-sqlite-persistence-core/src/persisted.ts b/packages/db-sqlite-persistence-core/src/persisted.ts index 14c902414e..19e191fc62 100644 --- a/packages/db-sqlite-persistence-core/src/persisted.ts +++ b/packages/db-sqlite-persistence-core/src/persisted.ts @@ -6231,8 +6231,20 @@ function createWrappedSyncConfig< runtime: PersistedCollectionRuntime, persistedReadiness?: PersistedReadinessTracker, ): SyncConfig { + let sourceStartup: Promise | undefined return { ...sourceSyncConfig, + beforeMutation: sourceSyncConfig.beforeMutation + ? async (collection) => { + const lifecycleGeneration = runtime.getLifecycleGeneration() + await sourceStartup + if (lifecycleGeneration !== runtime.getLifecycleGeneration()) { + throw new SyncTransactionAbortedError() + } + await runtime.ensureStarted() + return sourceSyncConfig.beforeMutation!(collection) + } + : undefined, sync: (params) => { const transactionStack: Array> = [] const pendingPublicationTransactions: Array< @@ -7479,6 +7491,7 @@ function createWrappedSyncConfig< ) return sourceResult })() + sourceStartup = sourceResultPromise resolveSourceResultAssigned() void sourceResultPromise.catch((error) => { reportUpstreamError(error) diff --git a/packages/db/src/transactions.ts b/packages/db/src/transactions.ts index 0e882a9525..44d19421ca 100644 --- a/packages/db/src/transactions.ts +++ b/packages/db/src/transactions.ts @@ -755,6 +755,38 @@ class Transaction> { if (this.mutations.length > 0) { // Run mutationFn try { + const collections = [ + ...new Set(this.mutations.map((mutation) => mutation.collection)), + ] + const prepareCollections = collections.filter( + (collection) => collection.config.sync.beforeMutation, + ) + if (prepareCollections.length > 0) { + const retired = this.when(`settled`).then( + () => null, + () => null, + ) + for (;;) { + const guards = await Promise.race([ + Promise.all( + prepareCollections.map((collection) => + Promise.resolve().then(() => { + collection._resumeSyncStart() + return collection.config.sync.beforeMutation!(collection) + }), + ), + ), + retired, + ]) + if ( + guards === null || + (this.state as TransactionState) !== `persisting` + ) + return this + // Validate every source together; no await between this check and dispatch. + if (guards.every((isCurrent) => isCurrent())) break + } + } // At this point we know there's at least one mutation // We've already verified mutations is non-empty, so this cast is safe // Use a direct type assertion instead of object spreading to preserve the original type diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index 8d17b2ae8e..a0138c0ee6 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -447,6 +447,10 @@ export interface SyncConfig< T extends object = Record, TKey extends string | number = string | number, > { + /** Prepare source acknowledgement before dispatch; the guard must still hold at dispatch. */ + beforeMutation?: ( + collection: Collection, + ) => Promise<() => boolean> sync: (params: { collection: Collection /** diff --git a/packages/db/tests/oracle-config.ts b/packages/db/tests/oracle-config.ts index 635bfa2149..4bc073ba77 100644 --- a/packages/db/tests/oracle-config.ts +++ b/packages/db/tests/oracle-config.ts @@ -275,6 +275,7 @@ const staticOracleProperties = [ `electric.sdk-snapshot-delivery`, `electric.sdk-dnf-membership`, `electric.sdk-source-release`, + `electric.sdk-mutation-startup`, `electric.sdk-rotation-release`, `electric-recovery.publication-stream-convergence`, `sqlite-resume.startup-generation`, diff --git a/packages/electric-db-collection/src/electric.ts b/packages/electric-db-collection/src/electric.ts index e91a767448..a832ed0df4 100644 --- a/packages/electric-db-collection/src/electric.ts +++ b/packages/electric-db-collection/src/electric.ts @@ -942,6 +942,30 @@ class ElectricLifecycle> { // The current run's failure: no evidence can arrive after it (EVID-2). private failure: { error: unknown } | undefined + mutationReady: Promise<() => boolean> = Promise.resolve(() => false) + resolveMutationReady: () => void = () => {} + rejectMutationReady: (error: unknown) => void = () => {} + + resetMutationReady(): void { + const ready = new Promise<() => boolean>((resolve, reject) => { + this.resolveMutationReady = () => + resolve(() => this.active && this.mutationReady === ready) + this.rejectMutationReady = (error) => { + reject(error) + // Rejection must invalidate an already-resolved preparation too. + this.mutationReady = Promise.reject(error) + void this.mutationReady.catch(() => undefined) + } + }) + this.mutationReady = ready + void ready.catch(() => undefined) + } + + prepareMutation(): Promise<() => boolean> { + if (!this.active) throw new StreamAbortedError(this.collectionId) + return this.mutationReady + } + constructor(private readonly collectionId?: string) {} readonly utils: ElectricCollectionUtils = { @@ -952,6 +976,7 @@ class ElectricLifecycle> { start(): number { if (this.active) this.retire() this.active = true + this.resetMutationReady() this.runReady = false this.failure = undefined this.epoch++ @@ -966,6 +991,7 @@ class ElectricLifecycle> { retire(epoch?: number): void { if (epoch !== undefined && !this.isActive(epoch)) return this.active = false + this.rejectMutationReady(new StreamAbortedError(this.collectionId)) this.rejectWaits(new StreamAbortedError(this.collectionId)) this.matchBuffer = { committedMessages: [], pendingMessages: [] } for (const txid of this.evidence.seenTxids) this.retired.seenTxids.add(txid) @@ -1312,6 +1338,14 @@ export function electricCollectionOptions>( const syncTemplate = withCollectionSyncConfigFactory( { ...sync, + beforeMutation: + internalSyncMode === `on-demand` + ? (collection) => { + const lifecycle = boundLifecycles.get(collection) + if (!lifecycle) throw new StreamAbortedError(collection.id) + return lifecycle.prepareMutation() + } + : undefined, exportSyncMeta: () => descriptorLifecycle.exportMeta(), importSyncMeta: (meta) => descriptorLifecycle.importMeta(meta), mergeSyncMeta: mergeElectricSyncMeta, @@ -1994,6 +2028,8 @@ function createElectricSync>( // decision). Cleanup aborts internally instead. const forwardExternalAbort = () => { const active = isActiveLifecycle() + if (active) + lifecycle.rejectMutationReady(new StreamAbortedError(collectionId)) abortController.abort() if (active) { const error = new StreamAbortedError(collectionId) @@ -2091,6 +2127,8 @@ function createElectricSync>( // settle on the snapshot it establishes. // Waiting loads and readiness waits reject with the same cause. const failWaits = (cause: unknown) => { + if (!isActiveLifecycle()) return + lifecycle.rejectMutationReady(cause) rejectFullSnapshot(cause) if (!ready) failInitialSync(cause) // No retry follows, so the open transaction can never commit. A @@ -2596,7 +2634,10 @@ function createElectricSync>( processMoveInEvent(message.headers.patterns) } } else if (isMustRefetchMessage(message)) { - if (hasReceivedUpToDate) resetFullSnapshot() + if (hasReceivedUpToDate) { + resetFullSnapshot() + lifecycle.resetMutationReady() + } debug( `${collectionId ? `[${collectionId}] ` : ``}Received must-refetch message, starting transaction with truncate`, ) @@ -2799,10 +2840,12 @@ function createElectricSync>( snapshot, ), ) + if (commitPoint === `up-to-date` && isActiveLifecycle()) + lifecycle.resolveMutationReady() lifecycle.publishEvidence(newTxids, newSnapshots) newTxids.clear() newSnapshots.length = 0 - lifecycle.commitMatches() + if (isActiveLifecycle()) lifecycle.commitMatches() } } @@ -2818,6 +2861,7 @@ function createElectricSync>( // The SDK gave up: it rejected the retry decision or exhausted its // retry loop. No provider batch will settle the gate. rejectFullSnapshot(error) + if (isActiveLifecycle()) lifecycle.rejectMutationReady(error) if (isActiveLifecycle() && !abortController.signal.aborted) { if (!ready) failInitialSync(error) else giveUp(error) 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 9dd9b856e5..be9b32f893 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 @@ -1,8 +1,17 @@ import fc from 'fast-check' import { expect, it, vi } from 'vitest' -import { IR, createCollection } from '@tanstack/db' +import { + BasicIndex, + IR, + createCollection, + createTransaction, +} from '@tanstack/db' import { ShapeStream } from '@electric-sql/client' -import { persistedCollectionOptions } from '../../db-sqlite-persistence-core/src' +import { createDeferred } from '../../db/src/deferred' +import { + SingleProcessCoordinator, + persistedCollectionOptions, +} from '../../db-sqlite-persistence-core/src' import { electricCollectionOptions } from '../src/electric' import { oraclePropertyOptions, @@ -28,7 +37,10 @@ import { createRetryModel, expectedRetryObservation, } from './electric-retry-decision-model' -import { storageTargetId } from './electric-persistence-fixture' +import { + managedTagPersistence, + storageTargetId, +} from './electric-persistence-fixture' import type { Item, Request } from './electric-sdk-delivery-support' import type { ElectricCollectionUtils } from '../src/electric' import type { PersistenceAdapter } from '../../db-sqlite-persistence-core/src' @@ -133,6 +145,7 @@ declareOracle({ runClientRetryCell, checkSourceRelease, checkManagedRotation, + checkMutationStartup, ], model: [createRetryModel, applyRetryModel, expectedRetryObservation], }) @@ -1917,3 +1930,403 @@ sdkCampaigns( releaseReach.assertReached() }, ) + +// Proposed mutation-admission contract: an on-demand source must establish its +// changes-only cursor before a mutation can reach the server. This is stream +// evidence, not applied-row readiness (application may be parked by optimism). +// Grammar varies payload and release order; every history crosses direct, +// manual, derived-descriptor, persisted startup, multi-collection reset/retirement, +// rollback, and abort paths. The persisted driver holds metadata before source +// entry, then succeeds, fails, or cleans up. Cleanup must release its cache +// claim and coordinator subscription without stale preparation acquiring new +// resources (Collection cleanup contract). A coordinator index can remain +// pending after metadata and the Electric baseline arrive. Its storage is controlled, not native +// SQLite; cleanup is observed after releasing storage, not as prompt I/O abort. +// Removing the second collection loses the all-participants law; removing the held echo loses the +// distinction between admitting a write and confirming it. It excludes backend +// retry, persisted cache rotation, and missing update/delete target admission. +const mutationStartupStates = [ + `direct`, + `ready-cleanup`, + `manual`, + `materialized`, + `persisted`, + `persisted-failure`, + `persisted-index`, + `persisted-cleanup`, + `multi`, + `reset`, + `retire`, + `rollback`, + `abort`, +] as const +let startupSource = 0 +async function checkMutationStartup( + name: string, + reverse: boolean, + reach: ReturnType< + typeof createOracleReach<(typeof mutationStartupStates)[number]> + >, +) { + for (const mode of mutationStartupStates) { + const sources = Array.from( + { length: [`multi`, `reset`, `retire`].includes(mode) ? 2 : 1 }, + (_, index) => { + const http = controlledHttp() + const signal = new AbortController() + const invoked = deferred() + let writes = 0 + const metadataEntered = deferred() + const releaseMetadata = createDeferred() + const persistence = managedTagPersistence() + const claims = new Set() + let claimRequests = 0 + const claimCacheGeneration = persistence.adapter.claimCacheGeneration! + persistence.adapter.claimCacheGeneration = async (...args) => { + claimRequests++ + const claim = await claimCacheGeneration(...args) + claims.add(claim.claimId) + return claim + } + const releaseClaim = persistence.adapter.releaseCacheGenerationClaim! + persistence.adapter.releaseCacheGenerationClaim = async (...args) => { + await releaseClaim(...args) + claims.delete(args[0]) + } + const indexEntered = deferred() + const releaseIndex = createDeferred() + const coordinator = new SingleProcessCoordinator() + let subscriptions = 0 + coordinator.subscribe = () => { + subscriptions++ + return () => { + subscriptions-- + } + } + const ensureIndex = + coordinator.requestEnsurePersistedIndex.bind(coordinator) + coordinator.requestEnsurePersistedIndex = async (...args) => { + if (mode === `persisted-index`) { + indexEntered.resolve() + await releaseIndex.promise + } + return ensureIndex(...args) + } + const loadResumeSnapshot = persistence.adapter.loadResumeSnapshot + persistence.adapter.loadResumeSnapshot = async (...args) => { + metadataEntered.resolve() + await releaseMetadata.promise + return loadResumeSnapshot(...args) + } + const options = electricCollectionOptions({ + id: `startup-${mode}-${index}`, + syncMode: `on-demand`, + getKey: (row) => row.id, + shapeOptions: { + url: `https://example.com/startup-${++startupSource}`, + signal: signal.signal, + fetchClient: http.fetchClient, + }, + onInsert: async ({ collection: target }) => { + writes++ + invoked.resolve() + await target.utils.awaitTxId(101 + index, 1000) + }, + }) + const original = + mode === `materialized` ? createCollection(options) : undefined + const collection = createCollection( + original + ? { ...original.config, id: `${options.id}-derived` } + : mode.startsWith(`persisted`) + ? persistedCollectionOptions({ + ...options, + persistence: { adapter: persistence.adapter, coordinator }, + }) + : options, + ) + if (mode === `persisted-index`) { + collection.createIndex((row) => row.name, { indexType: BasicIndex }) + } + const retiredFromReady = createDeferred() + if (mode === `ready-cleanup`) { + collection.onFirstReady(() => { + void collection + .cleanup() + .then(retiredFromReady.resolve, retiredFromReady.reject) + }) + } + return { + claims, + claimRequests: () => claimRequests, + subscriptions: () => subscriptions, + retiredFromReady, + indexEntered, + releaseIndex, + original, + metadataEntered, + releaseMetadata, + http, + signal, + invoked, + collection, + writes: () => writes, + index, + } + }, + ) + await Promise.all(sources.map((source) => source.original?.cleanup())) + let calls = 0 + const tx = [`manual`, `multi`, `reset`, `retire`].includes(mode) + ? createTransaction({ + mutationFn: async () => { + calls++ + sources.forEach((source) => source.invoked.resolve()) + await Promise.all( + sources.map((source) => + source.collection.utils.awaitTxId(101 + source.index, 1000), + ), + ) + }, + }) + : sources[0]!.collection.insert({ id: 1, name }) + if ([`manual`, `multi`, `reset`, `retire`].includes(mode)) { + tx.mutate(() => + sources.forEach((source) => source.collection.insert({ id: 1, name })), + ) + } + const settled = tx.when(`settled`).then( + () => undefined, + (error) => error, + ) + await withElectricCleanup(async () => { + if (mode.startsWith(`persisted`)) { + const source = sources[0]! + await atCheckpoint( + source.metadataEntered.promise, + `held persistence metadata`, + ) + if (mode !== `persisted-cleanup`) reach.reach(mode) + expect(tx.state, `metadata loading is not source retirement`).toBe( + `persisting`, + ) + if (mode === `persisted-failure`) { + const failure = new Error(`metadata unavailable`) + source.releaseMetadata.reject(failure) + expect( + await atCheckpoint(settled, `failed persistence startup`), + ).toBe(failure) + expect(source.writes()).toBe(0) + return + } + if (mode === `persisted-cleanup`) { + expect(source.claims.size, `startup owns a cache claim`).toBe(1) + const claimRequests = source.claimRequests() + const cleanup = source.collection.cleanup() + if (reverse) await cleanup + source.releaseMetadata.resolve() + await cleanup + expect( + await atCheckpoint(settled, `cleanup before source entry`), + ).toBeInstanceOf(Error) + expect(source.writes()).toBe(0) + reach.reach(mode) + expect( + source.claimRequests(), + `retired preparation cannot restart storage`, + ).toBe(claimRequests) + expect(source.claims.size, `cleanup releases every cache claim`).toBe( + 0, + ) + expect( + source.subscriptions(), + `cleanup releases coordinator subscription`, + ).toBe(0) + return + } + source.releaseMetadata.resolve() + } + const first = await Promise.all( + sources.map((source) => + atCheckpoint(source.http.take(), `cold baseline`), + ), + ) + const writeCount = () => + calls + sources.reduce((count, source) => count + source.writes(), 0) + reach.reach(mode) + expect(writeCount(), `no write before baseline: ${mode}`).toBe(0) + // A batch admitted while active keeps its transaction evidence even if + // its ready callback retires the run. A callback arriving after retirement + // is different: the existing processMessages entry fence rejects it. + // This receives the parent adapter's evidence-retention behavior while + // still requiring the new admission gate to reject a retired source. + if (mode === `ready-cleanup`) { + first[0]!.respond( + new Response( + JSON.stringify([ + { + key: `99`, + value: { id: 99, name }, + headers: { operation: `insert`, txids: [100] }, + }, + { headers: { control: `up-to-date` } }, + ]), + { headers: headers(1) }, + ), + ) + await atCheckpoint( + sources[0]!.retiredFromReady.promise, + `cleanup from ready`, + ) + expect(await atCheckpoint(settled, `retired admission`)).toBeInstanceOf( + Error, + ) + expect(writeCount()).toBe(0) + expect(await sources[0]!.collection.utils.awaitTxId(100, 50)).toBe(true) + return + } + if (mode === `abort`) { + sources[0]!.signal.abort() + expect(await atCheckpoint(settled, `aborted startup`)).toBeInstanceOf( + Error, + ) + expect(writeCount()).toBe(0) + return + } + if (mode === `rollback`) tx.rollback() + const live: Array = [] + const order = reverse ? [...sources].reverse() : sources + let replacement: Request | undefined + for (const [position, source] of order.entries()) { + first[source.index]!.respond( + new Response( + JSON.stringify([{ headers: { control: `up-to-date` } }]), + { headers: headers(1) }, + ), + ) + live[source.index] = await atCheckpoint( + source.http.take(), + `live poll after baseline`, + ) + if (position < sources.length - 1) expect(writeCount()).toBe(0) + if (position === 0 && mode === `reset`) { + live[source.index]!.respond( + new Response( + JSON.stringify([{ headers: { control: `must-refetch` } }]), + { headers: headers(2) }, + ), + ) + replacement = await atCheckpoint( + source.http.take(), + `replacement baseline`, + ) + } + if (position === 0 && mode === `retire`) source.signal.abort() + } + if (mode === `retire`) { + expect( + await atCheckpoint(settled, `retired participant`), + ).toBeInstanceOf(Error) + expect(writeCount()).toBe(0) + return + } + if (mode === `persisted-index`) { + const source = sources[0]! + await atCheckpoint( + source.indexEntered.promise, + `coordinator index startup`, + ) + // Source position alone cannot authorize a write while persisted startup + // is still pending. The same generated payload must dispatch after it. + expect( + writeCount(), + `no write before persisted startup completes`, + ).toBe(0) + source.releaseIndex.resolve() + } + if (replacement) { + expect(writeCount(), `replacement cursor still held`).toBe(0) + replacement.respond( + new Response( + JSON.stringify([{ headers: { control: `up-to-date` } }]), + { headers: headers(3) }, + ), + ) + live[order[0]!.index] = await atCheckpoint( + order[0]!.http.take(), + `replacement live poll`, + ) + } + if (mode === `rollback`) { + await settled + expect(writeCount(), `rollback prevents late write`).toBe(0) + return + } + expect( + tx.state, + `admission remains live: ${mode}; ${tx.error?.message ?? ``}`, + ).toBe(`persisting`) + await atCheckpoint( + Promise.all(sources.map((source) => source.invoked.promise)), + `admitted mutation ${mode}`, + ) + expect(tx.state, `baseline is not confirmation`).toBe(`persisting`) + for (const source of sources) { + live[source.index]!.respond( + new Response( + JSON.stringify([ + { + key: `1`, + value: { id: 1, name }, + headers: { operation: `insert`, txids: [101 + source.index] }, + }, + { headers: { control: `up-to-date` } }, + ]), + { headers: headers(mode === `reset` ? 4 : 2) }, + ), + ) + } + expect(await atCheckpoint(settled, `confirmed mutation`)).toBeUndefined() + for (const source of sources) + expect(source.collection.get(1)?.name).toBe(name) + }, [ + ...sources.map((source) => () => source.releaseMetadata.resolve()), + ...sources.map((source) => () => source.releaseIndex.resolve()), + ...sources.map((source) => () => source.collection.cleanup()), + ...sources.map((source) => () => source.http.close()), + async () => { + await settled + }, + ]) + } +} +const startupViolation = sameViolation() +sdkCampaigns( + `electric.sdk-mutation-startup`, + `admits cold mutations only after every participant establishes its stream`, + 20261030, + startupViolation, + async (seed) => { + const reach = createOracleReach( + `electric.sdk-mutation-startup`, + mutationStartupStates, + ) + await fc.assert( + fc.asyncProperty( + fc.string({ maxLength: 8 }), + fc.boolean(), + async (name, reverse) => { + reach.beginHistory() + await startupViolation.check({ name, reverse }, () => + checkMutationStartup(name, reverse, reach), + ) + }, + ), + { + ...oraclePropertyOptions(8, `electric.sdk-mutation-startup`), + ...(seed === undefined ? {} : { seed }), + }, + ) + reach.assertReached() + }, +) diff --git a/packages/electric-db-collection/tests/electric.test.ts b/packages/electric-db-collection/tests/electric.test.ts index f9dbc6423c..a79f68ae93 100644 --- a/packages/electric-db-collection/tests/electric.test.ts +++ b/packages/electric-db-collection/tests/electric.test.ts @@ -3443,6 +3443,7 @@ describe(`Electric Integration`, () => { }) try { + subscriber([{ headers: { control: `up-to-date` } }]) transaction.mutate(() => testCollection.insert({ id: 3, name: `Local row` }), )