diff --git a/.changeset/fix-restarted-subscription-rows.md b/.changeset/fix-restarted-subscription-rows.md new file mode 100644 index 0000000000..6efbddabe1 --- /dev/null +++ b/.changeset/fix-restarted-subscription-rows.md @@ -0,0 +1,5 @@ +--- +'@tanstack/db': patch +--- + +Preserve restored rows for direct `subscribeChanges()` consumers after Collection cleanup and restart, even when their values are unchanged. This prevents incorrect delete events when the new sync run restores the same rows. diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md index 5e430487e0..40165b1772 100644 --- a/docs/contributing/oracle-coverage.md +++ b/docs/contributing/oracle-coverage.md @@ -2349,6 +2349,37 @@ The following related oracles own distinct production paths or observation cuts: Limit: the generated grammars reach a failed retirement followed by another retired replay only rarely (the replay grammar releases one demand, on the last attempt), so the named histories own that sequence. + The cleanup/restart equality crossing (`subscription-lifecycle.restart-publication`) + retains equal rows actually restored into a direct subscriber's private + replacement while deleting omitted retained rows. It runs the shared model, + driver, and exact per-command batch/source/subscriber comparisons over two + delivery paths (acquisition writes and independent source writes), six demand + lifetimes (pending, settled, two live demands, aborted, released, absent), + three prior-row relationships (absent, equal, changed), and resolve/reject + commands for live acquisitions. Acquisition histories without a live attempt + distinguish absent from retained rows and measure the settlement as a no-op; + they do not claim restored equality. The single scenario grammar has 24 + acquisition and 27 source cases. Each case reaches its measured owner lifetime, + prior/delivered row relation, current outcome, and restarted acquisition count + 12 times in each fixed/random campaign. Explicit generated row values and + zero to three prior retired acquisitions remove the fixture-attempt-ID premise. + Additional direct, filtered, ordered-snapshot and snapshot-before-source-events + drivers each measure absent/equal/changed rows 12 times per campaign and reuse + the owner's Map diff at held and settled publication cuts. The predicate + excludes a real nonmatching row; ordered selection covers the complete bounded + two-row replacement, not arbitrary windows. The exact issue #2310 history + remains a shared replay. Original suppression fails every restart campaign; + the earlier acquisition-only and retention-of-omitted-rows mutants remain + recorded. [Evidence and limits](oracle-reviews/2026-10-30-pr-2313-review.md). + This crossing concerns direct subscriptions under the controlled sync boundary. + Eager sources follow a separate accepted first-public-batch reconciliation law + in ARCHITECTURE's cleanup/restart section. The supplemental + `collection-restart-publication-boundaries.test.ts` observes its before-ready + deletion and later reinsertion, and checks actual compiled-query cleanup and + restart in eager/on-demand modes. Public cleanup refuses active dependents; + internal cleanup crashes them; restarting the query builds a fresh result. + This evidence does not claim source-only restart revives a query, nor does it + establish delegated includes, arbitrary ordered windows, or external providers. Bounded three-row histories now compare a subscription's selected keys with callback key/type batches across full, explicit, filtered, and empty snapshots, two later ordered selections, both initial-state modes, and callback reentry. diff --git a/docs/contributing/oracle-reviews/2026-10-30-pr-2313-review.md b/docs/contributing/oracle-reviews/2026-10-30-pr-2313-review.md new file mode 100644 index 0000000000..5d1967e291 --- /dev/null +++ b/docs/contributing/oracle-reviews/2026-10-30-pr-2313-review.md @@ -0,0 +1,168 @@ +# PR #2313 external review evaluation + +Reviewed head: `ef6316a910c24e60cbcfe8e3e613b0bddc821958`. +Source: the maintainer's ten-item external review, in source order below. +The repeated source numbers receive stable IDs P1–P3, O1–O6, and D1. +This follow-up changes tests and documentation. Production remains byte-for-byte +identical to the reviewed head. No product-law change is required. + +## Laws, experiments, and inferences + +| Law and authority | Existing enforcement | Decisive experiment | Inference and repair | +| --- | --- | --- | --- | +| A successful direct on-demand restart publishes restored rows, deletes omitted rows, suppresses equal public updates, and keeps partial or failed replacements private. Authority: `subscription.ts`'s public change-message boundary, the lifecycle grammar, and ARCHITECTURE's cleanup/restart and replay sections. | The publication model independently accumulates source and subscriber rows and compares exact per-command batches. Its old reach labels did not establish restored equality independently of fixture numbering. | On reviewed production, prepend one request/resolve/release of the other demand to the old grammar. The old fixed acquisition campaign passes, but a labeled equal case retains `{id:'a',value:1}` and actually restores `{id:'a',value:2}`. Restoring the original equality-suppression defect makes all 12 expanded restart campaigns fail at the shared publication comparison. | The production fix remains correct. Explicit input row values, observed retained/delivered values, actual settlements, owner lifetimes, and acquisition counts now establish the premises. | +| An eager restart's next public batch reconciles retained subscriber rows with installed state. An empty ready batch reconciles an empty replacement. Authority: ARCHITECTURE, “Cleanup, restart, and detached waiters.” | The existing restart fixture marks an empty eager run ready immediately. It does not promise whole-run atomicity for a source that streams several commits. | The new ordinary boundary test starts with a/b/c, cleans up, restarts, commits a while loading, then commits b/c and marks ready. It observes delete b/c in the first batch and insert b/c in the second. | P1's observation is real and is accepted design under the existing contract. Waiting until ready would change that contract and delay streamed eager updates. This PR does not make that change. | +| Cleanup ends a dependent graph. Public cleanup rejects active live queries, dependent cleanup ends them first, and internal cleanup crashes them. Source-only restart does not revive a crashed graph. Authority: public `cleanup()` documentation, glossary, ARCHITECTURE laws 14/16 and its explicit direct-restart limitation. | The cleanup oracle already exercises admission and dependent invalidation. The publication owner explicitly excludes compiled includes. | Six real compiled-query witnesses cross eager/on-demand with public, dependent, and internal cleanup. Public cleanup rejects. Internal cleanup keeps b/d in a crashed query after source-only restart. Cleaning up and preloading the query again publishes only b. | P2 conflates retained terminal rows with a live replacement. A supported query restart removes d. No delegated-publication production repair is justified by this trace. | +| Ordinary subscription admission is closed during active cleanup. Initial listener registration owns no demand, replay, or public rows. Authority: glossary's admission rule and AGENTS.md's distinction between valid operations and contradictory collaborators. | The cleanup oracle has 18 admission cells. Actual `Collection.on` only registers a callback. The only synchronous cleanup-start callback during construction has no user-visible work to retire. | An executable registration probe calls the real public subscription boundary while adapter cleanup is held. It throws before either listener-registration method runs. Ordinary registration and repeated unsubscribe succeed. The unchanged 18-cell admission owner passes. | P3's arbitrary throwing-registration conditional can leak resources, but the proposed throwing/reentrant collaborator does not occur at this constructor boundary. Do not add defensive recovery for substituted internal methods. | + +A whole-run eager atomicity law would be a product proposal. Its rival is the +accepted first-public-batch rule above. Atomicity would keep older rows visible +until readiness, while the existing rule exposes each committed eager batch. +The review supplies no authority to replace the accepted rule. No decision is +needed to complete this enforcement repair. + +## Oracle repair and measured reach + +The governing row model and exact comparisons are unchanged. Settle commands +can now supply explicit row data, independently of the fixture's attempt ID. +Generated values range from -100,000 to 100,000. Zero to three preliminary +retired acquisitions perturb numbering before the tested restart. + +The single scenario grammar supplies both history construction and declared +reach. A separate observer classifies executed owner operations, adapter +acquisition counts, actual settlement, and callback-applied versus written rows. +Rejected acquisitions establish no rows. Their successful retry supplies the +row-relation witness, after the actual rejection is observed. A canceled or +absent acquisition claims neither restored equality nor a rejection. + +- Acquisition delivery: 24 cases, each measured 12 times per campaign. +- Independent source delivery: 27 cases, each measured 12 times per campaign. +- Direct, filtered, ordered-snapshot, and snapshot-before-source-events receivers: + three row relations each, measured 12 times per campaign and receiving path. +- Fixed and random campaigns share the generator, budget, recorder and comparison. + Together these campaigns execute 1,512 restart histories at multiplier 1. + +The old 72-label count included observationally redundant canceled acquisition +relationships and no-op outcomes. Removing those duplicate categories does not +remove a valid publication distinction. Every generated-input budget remains +12. The original broad fixed/random budgets remain 60 each. O5's recommended +24-input enumeration would itself run 24 × 72 = 1,728 histories, the old total. +It would improve finite input enumeration, but would not reduce that work. +The revised grammar spends its budget on distinct structural cases and a much +larger payload domain. This is no elapsed-performance claim. + +The receiving driver uses the existing independent Map diff. A real predicate +excludes b. An ordered snapshot selects the whole bounded two-row replacement. +A held source publication installs rows before `requestSnapshot()` reads them, +then releases its source events before acquisition settlement. The held-cut +comparison checks unchanged subscriber rows and no callbacks. The settled-cut +comparison checks exact batches, values, omitted-row deletion and readiness. +No test discards installed source events to invent an unsupported public path. + +## Lossless item ledger + +| ID / original location | Claim | Evidence and technical verdict | PR action | Durable value | +| --- | --- | --- | --- | --- | +| P1 / production 1, subscription.ts:1921 | Eager a/b/c restart deletes b/c after an early a-only batch, before readiness. Older behavior, traced but not run. | Observation confirmed by same-path execution. Its claimed defect contradicts the explicit eager first-batch contract. | accepted-design | Ordinary eager boundary witness and the law distinction above. | +| P2 / production 2, subscription.ts:787 | Delegated restart restores b from b/d but leaves d visible because there is no privateRows buffer. Older behavior, traced but not run. | Refuted as an active-query restart failure by real compiled paths. Internal cleanup deliberately leaves terminal rows in a crashed graph. Public cleanup rejects it, or dependent cleanup ends it. Supported query restart yields b. | refuted | Six compiled receiving witnesses and explicit terminal-versus-live scope. | +| P3 / production 3, subscription.ts:300 | A later registration throw leaks earlier listeners. Reentrant unsubscribe spreads an unassigned array. | The conditional about substituted throwing methods is true, but current registrations cannot expose or unsubscribe the partly constructed subscription. Active-cleanup admission fails before registration. No reachable production counterexample was found at the stated seam. | refuted as a current product defect | Keep construction assumptions explicit. Existing admission oracle and the executable probe establish the supported boundary. Future fallible collaborators would require a new construction contract and witness. | +| O1 / oracle 3, :2606 | Reach records grammar labels, and inactive equal/changed acquisition histories count indistinguishable states. | Confirmed in part. Other model comparisons and restart/no-op assertions were real, so the old test was not vacuous. Its row-relation reach claim was unsupported. The fixture perturbation stays green with false equal labels. | fixed-now | Observed reach classifier and 51 distinct structural cases replace the 72-label claim. | +| O2 / oracle 4, :637 | Equality depends on hard-coded fixture acquisition IDs, so reordered or added attempts silently change the premise. | Confirmed by the green perturbed old campaign and retained 1/restored 2 trace. | fixed-now | Explicit generated values and preliminary retired-acquisition dimension. | +| O3 / oracle 5, :2585 | Default-only campaign omits eager, filtered, ordered, and snapshot delivery, which hides P1. | The path omissions are real. P1 is a different accepted law, and snapshots have their own delivery branch, so the blanket implication of a defect is too broad. | fixed-now for the coverage gap | Four shared-model receiving paths, eager and compiled ordinary witnesses, and precise coverage limits. | +| O4 / oracle 6, :2568 | Issue seed 2310 appears in the test name. | Confirmed naming-policy violation in the interpolated runtime name. | fixed-now | Names now say fixed seed or random/replayed seed. Seed data and replay stay unchanged. | +| O5 / oracle 7, :2565 | The 1,728 histories repeat 24 inputs. Enumerating those inputs once gives equal coverage faster. | Finite payload sampling was wasteful. The work conclusion is incorrect: each input drives 72 histories, so enumerating 24 still costs 1,728. Fixed/random parity is also required. | fixed-now for the useful grammar improvement | Broader payloads, numbering churn and removal of redundant structural cases. No run budget is lowered. | +| O6 / oracle 8, :587 | Reach product, repeated outcome arrays and duplicated input type can drift. | Confirmed source-maintenance issue. | fixed-now | One scenario grammar, one outcomes constant and an inferred arbitrary input type. Observed reach remains separate. | +| D1 / docs 6, :5/:184/:223 | Current-head and uncommitted/no-PR claims are stale, raw evidence needs inaccessible scratch, and hiring prose is misplaced. | Confirmed misleading presentation. Later entries already recorded publication, but the opening still claimed current status. | fixed-now | Historical checkpoints are explicit, this versioned record preserves decisive evidence, and hiring recommendations remain conversation output. | + +No raw item is dropped or merged. No sibling production violation was confirmed. +The initial receiver experiments had setup failures (an invalid IR function +name and an unnormalized expected batch), which are not production RED results. +The final valid receiving paths pass unchanged production. A discarded-event +experiment was removed because it did not establish a supported public history. + +## Verification and calibration + +The nine-suite command below passes **661 tests**. TypeScript, +formatting, ESLint, `pnpm lint:changed`, and `git diff --check` pass. + +```sh +pnpm --filter @tanstack/db exec vitest run \ + tests/collection-subscription-lifecycle-publication-oracle.property.test.ts \ + tests/collection-subscription-lifecycle-history-oracle.property.test.ts \ + tests/collection-subscription-lifecycle-oracle.test.ts \ + tests/collection-subscription-replay-oracle.property.test.ts \ + tests/collection-subscription-lifecycle.test.ts \ + tests/collection-subscription-lifecycle-publication.test.ts \ + tests/collection-subscription-reentrancy-oracle.test.ts \ + tests/collection-cleanup-restart-oracle.test.ts \ + tests/collection-restart-publication-boundaries.test.ts \ + --maxWorkers=2 --coverage.enabled=false +``` + +Calibration commands use the same owner with `-t 'matches restarted row +publication'`. The original-defect mutant removes only +`this.truncateReplayState?.privateRows ||` from stale-row reconciliation. It +fails all 12 campaigns at the publication comparison, rather than timeout or +setup. Restoring the exact reviewed production file restores GREEN. +The old-fixture perturbation uses `git show ef6316a91:`, prepends +request/resolve/release of the other demand to each restart history, and runs +`-t 'matches restarted row publication acquisition \(2310\)'`. It passes +while an actual write of 2 replaces retained 1 in a labeled equal history. +These results directly calibrate O1/O2 without treating a temporary mutant's +GREEN result as a repaired branch result. + +Direct replay remains: + +```sh +TANSTACK_DB_ORACLE_SEED=2310 TANSTACK_DB_ORACLE_PATH=0:0:0 \ +TANSTACK_DB_ORACLE_PROPERTY=subscription-lifecycle.restart-publication \ +pnpm --filter @tanstack/db exec node --import tsx tests/oracle-replay.ts \ + tests/collection-subscription-lifecycle-publication-oracle.property.test.ts \ + -t 'matches restarted row publication' --maxWorkers=2 --coverage.enabled=false +``` + +The guarded replay passes six campaigns and reports six successful witnesses +with property, seed, and path. + +The evidence block is: **Law**, direct replacement membership and existing +neighboring lifecycle rules above. **Reach**, 51 structural cases plus 12 +receiver relations, each 12 times in each normal campaign. **RED**, the old +oracle accepts a false equality premise, and the original-defect mutant fails +all expanded campaigns. **GREEN**, the same comparisons on restored reviewed +production. **Mutant**, original equality suppression. **Limits**, finite +integer rows, a controlled sync boundary, bounded ordered snapshot selection, +and the receiving paths above. These checks establish no new external-provider, +framework, arbitrary ordered-window, or delegated-includes promise. + +## Oracle rubric audit and loss audit + +ORC-001/002: existing authority and independent Map diff are unchanged. +ORC-003: executable prose describes the law, grammar, receiving seams and cuts. +ORC-004: the exact issue replay remains. Equality, omission, delivery and live +outcomes distinguish suppression, keep-all and acquisition-only designs. +Numbering churn is a marginal case. Inactive acquisitions cannot claim restored +rows. Transactions remain well formed. ORC-005/006: actual receiving writes, +held/settled public observations and the original-defect assertion failures +establish path and checker sensitivity. ORC-007: fixed/random parity and direct +seed/path replay remain. ORC-008: removing duplicate inactive labels changes no +model state or legal next action. ORC-009: row relation and owner reach labels +are observations, not new production states. ORC-010: failures and secondary +cleanup errors remain ordered, with the primary as cause. ORC-011 does not +require another formulation: no shared semantic fault in the Map diff was +identified. ORC-012: this record links the reviewed head to every raw item and +its evidence. ORC-013: absent/equal/changed and omitted rows discriminate nearby +wrong membership rules. ORC-014: claims remain at the controlled core boundary. +Compiled cleanup is exercised in its real core receiving path, with no external +provider claim. ORC-015: each path asserts measured reach with minimum 12. +ORC-016: ordinary boundary examples stay outside the oracle file, and all +oracle histories use the shared judgment. + +**10 raw items = 7 fixed-now + 1 accepted-design + 2 refuted.** No item is +confirmed-open, deferred, stale, duplicate, or awaiting a product decision. +Item accounting, local validation, and bounded enforcement repairs are complete. CI status and PR readiness require separate final-head evidence. + +The reviewed production SHA-256 remains +`f2ee13496359611fa5fe22066df96d80dec88a77fa18693af2fcfeff4ce29bc4`. +The repaired publication-owner SHA-256 is +`ff91a783388ceee1d3189f0b105cdfdae555e2ac5b15f5a50937ecd5aaa51112`. +These identify the exact source and oracle used for the final local checks. diff --git a/docs/contributing/oracle-reviews/issue-2310-restart-publication.md b/docs/contributing/oracle-reviews/issue-2310-restart-publication.md new file mode 100644 index 0000000000..66a2095115 --- /dev/null +++ b/docs/contributing/oracle-reviews/issue-2310-restart-publication.md @@ -0,0 +1,450 @@ +# Issue #2310: equal rows in restarted publication + +Reviewed source: [issue #2310](https://github.com/TanStack/db/issues/2310), +including its complete body and no comments, retrieved 2026-10-30. +Original investigation base: `c351fc04d7f70dc59bf92ecbed310f899b9511cd`. +At that checkpoint, the repair was an uncommitted diff on `codex/issue-2310-publication`. +Production and oracle repairs were made. No product-law change is needed. + +This document is a chronology. Later sections supersede earlier branch and +measurement claims. PR [#2313](https://github.com/TanStack/db/pull/2313) published +the repair. The [external review evaluation](2026-10-30-pr-2313-review.md) +supersedes the original 72-state reach claim and records the current evidence. + +## Law, experiment, and inference + +A direct subscription survives Collection cleanup. Cleanup discards source rows +without sending subscriber deletes. After restart, successful reacquisition +publishes the new source replacement: an equal row actually restored by that +source remains present, an omitted row leaves, and a changed row updates. +An equal restored row needs no redundant callback. Failed replacement keeps +the prior subscriber snapshot; an obsolete acquisition cannot authorize it. + +Authority: the public `Collection.subscribeChanges` change-message contract, +the existing publication owner's opening contract and independent row model, +and the shared lifecycle grammar's cleanup, restart, cancellation and settlement +rules. The coverage map assigns these observations to the Collection lifecycle +publication owner. This is enforcement of existing replacement membership, not +a new policy for source completeness or retention time. + +| Law | Existing instrument | Decisive experiment | Inference and repair | +| --- | --- | --- | --- | +| Successful replacement includes every restored row, even one equal to the retained subscriber row | The shared lifecycle model selects current attempts; the independent publication model uses Maps to compute source and subscriber membership. The driver compares exact callback batches, source rows and callback-applied rows after each command. | Exact issue seed/path fails at command 5. Current attempt 1 in sync run 1 writes `{id:'b', value:1}`; attempt 0 belongs to the ended run. Source rows contain b, but the subscriber receives delete b. | Production drops the equal delta before adding it to private replacement rows. Retain that delta while a direct private replacement exists; its final diff suppresses a redundant public update. | +| Ordinary source writes obey the same replacement law | Same owner, model and comparison; no new expected-result computation | Before production edits, the new source-delivery lane restores equal independent row c while demand a holds publication. Settlement deletes c unexpectedly, alongside the legitimate deletion of omitted d. | A patch restricted to acquisition keys would miss the class. The predicate covers both real source delivery paths. | +| Omitted rows leave, failed replacement stays private, obsolete settlements are inert, and nonbuffered equal writes stay silent | Same model and observation cuts | Generated neighbors cross missing/equal/changed rows, six demand lifetimes, current resolve/reject and obsolete outcomes. A mutant retaining omitted rows fails; canceled and absent current attempts are measured as no-ops. | Preserve the existing final diff, stale-row cleanup and cancellation semantics. No model or comparison was relaxed. | + +The repair-location judgment is production. Neither the driver nor the model +needs a changed expectation. The actual source write supplies membership before +the disputed publication cut, so there is no missing-provider-information +hypothesis at this controlled boundary. + +## Original reach and grammar (superseded) + +The original general grammar already generated the counterexample; the defect +was not a false-green model. Its finite random budget did not reliably exercise +cleanup/restart with an equal restored row. The added constructive crossing +makes that distinction durable and measures reach. + +`subscription-lifecycle.restart-publication` runs both delivery paths: +acquisition writes and ordinary independent source writes. Each generated input +runs every combination of: + +- six demand histories: pending, settled, two live demands, aborted, released, + and no owner; +- three prior-row relations: absent, equal, changed; +- current resolve and reject commands, with a successful truncate replay after + rejection. + +The demand key (a/b), independent value (0–5), and obsolete resolve/reject outcome +vary. The fixture's acquisition rows use attempt IDs, so the grammar derives +which initial value will equal the restarted write. Omitted independent d +prevents a keep-all-rows implementation from passing. + +There are 72 declared history states. Each is reached **12 times per campaign**, +separately for acquisition and ordinary-source paths, in the normal fixed-seed +and random runs: 1,728 histories total. The reach counter and an exact-count +assertion enforce these counts. For aborted, released and absent owners, the +current-settlement command is explicitly checked as unexecuted; its label is +not evidence of a successful or failed acquisition. Both actual current +outcomes are structurally required for the live-owner histories. + +Reconstruction: the exact nine-command issue history is retained as input to +`runPublicationHistory`, without separate expected values. The general grammar +and original seed/path remain unchanged. Ablation: removing equality loses the +reported defect; removing ordinary source delivery admits an acquisition-only +repair; removing omission admits stale-row retention. Range: both demand keys, +all six independent values, old pending/settled/canceled demand, zero/one/two +live owners, and late obsolete outcomes. Exclusion: source operations always +commit valid complete transactions; the fixture fences writes with their +acquisition signal. An aborted attempt cannot legally supply replacement rows. + +## RED, GREEN, and calibration + +All RED results below preceded production edits. The original model and shared +comparison remained unchanged throughout. + +```sh +TANSTACK_DB_ORACLE_SEED=-1439698414 \ +TANSTACK_DB_ORACLE_PATH=1:22 \ +TANSTACK_DB_ORACLE_PROPERTY=subscription-lifecycle.publication-history \ +pnpm --filter @tanstack/db exec vitest run \ + tests/collection-subscription-lifecycle-publication-oracle.property.test.ts \ + -t 'matches row publications for a random or replayed seed' \ + --maxWorkers=2 --coverage.enabled=false +``` + +Original RED: one assertion fails at `assertPublications`, command 5: +observed delete b(value 1), expected no batch. Repaired GREEN: one test passes. + +The extended owner's fixed seed 2310 and random campaigns also failed before +production edits on both acquisition and ordinary-source paths. The +ordinary-source mismatch at command 7 incorrectly deleted restored c. The +legitimate deletion of omitted d and insertion of demand a remained visible, +so this is a publication assertion failure, not setup or missing execution. + +The new property's direct replay is also GREEN: + +```sh +TANSTACK_DB_ORACLE_SEED=2310 TANSTACK_DB_ORACLE_PATH=0:0:0 \ +TANSTACK_DB_ORACLE_PROPERTY=subscription-lifecycle.restart-publication \ +pnpm --filter @tanstack/db exec vitest run \ + tests/collection-subscription-lifecycle-publication-oracle.property.test.ts \ + -t 'matches restarted row publication' \ + --maxWorkers=2 --coverage.enabled=false +``` + +This selects only the replayed campaign and runs both delivery paths: two tests +pass. Normal campaigns use the same grammar, driver, model, comparisons and +12-input budget; one uses seed 2310 and one chooses a random seed. + +| Temporary production mutant | Result at the intended comparison | +| --- | --- | +| Original equal-delta suppression, removing the private-replacement predicate | Original regression and all four generated tests fail at publication assertions. | +| Preserve equal deltas only for acquisition keys a/b | The original b regression passes, but both source-delivery campaigns fail: restored c is wrongly deleted. This rejects a one-trace repair. | +| Put omitted stale rows back into the final replacement | All four campaigns fail publication assertions because omitted d survives. | + +Each mutant was restored before the next experiment. None survives, times out, +or fails during setup. Mutant passes are not reported as repaired-branch GREEN. + +The eight-suite surrounding run passed **646 tests**, covering lifecycle +publication, lifecycle histories, lifecycle oracle, replay oracle, ordinary +lifecycle/publication examples, reentrancy, and cleanup/restart. The final +publication owner alone passed **143 tests** after stronger reach assertions. +`pnpm --filter @tanstack/db exec tsc --noEmit` passed. Final lint, formatting, +code-weight measurements and any later verification are appended below. + +## Limits and oracle-guide audit + +The closure claim is direct Collection subscriptions × the bounded histories +above × the public Collection sync boundary × per-command exact callbacks, +source rows and callback-applied subscriber rows. The changed predicate applies +only when a direct private-rows map exists. Delegated graph publication keeps +its prior path. These experiments do not establish compiled includes behavior, +a framework's scheduling, or an external provider's completeness or cancellation. +The controlled rows have string keys and numeric values; virtual fields, +arbitrary object equality, optimistic writes and reentrant adapter callbacks +remain with their existing owners. No known reachable violation of this bounded +law remains; generated passing histories are not a universal proof. + +| Requirement | Outcome | +| --- | --- | +| ORC-001 | Existing API/publication contract and limits stated above; no new product law. | +| ORC-002 | Existing independent Map model unchanged; production supplies only observations. | +| ORC-003 | Opening law and adjacent grammar prose explain membership, source delivery, histories, observations and checkpoints. | +| ORC-004 | Reconstruction, ablation, range and exclusion recorded above. | +| ORC-005 | Real Collection API and sync writes; exact callback/source/subscriber comparisons at every command. Actual restart and current/no-op settlement execution are checked. No missing source fact identified. | +| ORC-006 | Three hostile production mutants fail at the intended publication assertions. | +| ORC-007 | Fixed/random campaigns have equal budgets and mechanics; original and new seed/path replay commands were executed. No `fc.commands` replay path applies. | +| ORC-008 | No reference-model state added, removed or merged. | +| ORC-009 | Existing glossary terms retained. Demand-history labels describe grammar premises; canceled/absent current attempts are not classified as settlements. | +| ORC-010 | Existing driver preserves primary mismatch and secondary cleanup failures with `AggregateError`; its cleanup is unchanged. | +| ORC-011 | No new shared semantic-fault hypothesis requiring a second formulation. Counterexample comes from independent source/subscriber disagreement, not merely production/model agreement. | +| ORC-012 | This record identifies the reviewed HEAD and final code/test diff; applicable requirements are accounted for. | +| ORC-013 | Equal/changed/absent and omitted-row neighbors distinguish suppression from membership; narrow-fix and retain-omitted mutants are rejected. | +| ORC-014 | No real-provider or native-host claim is made. This is a controlled Collection boundary. | +| ORC-015 | Every declared history state reaches the driver and comparisons 12 times per normal campaign; each receiving path has separate reach. | +| ORC-016 | Regression and all generated histories replay the shared model/driver/comparison; no separate expected answer was added. | + +## Lossless finding ledger + +Source-order IDs preserve the issue's separate claims and requirements. No +comments, footnotes or capped findings were present. The linked issue supplies the source. This versioned record preserves the +decisive traces, commands, results, and limits without requiring local scratch. + +| ID | Original claim or action | Evidence / verdict | Action | Durable value and destination | +| --- | --- | --- | --- | --- | +| R1 | Cleanup/restart publication deletes retained b; reproduced on main c351fc04d and PR #2307 1d619bb; cause explicitly undiagnosed | Confirmed production defect on the reported main commit with exact seed/path. PR head replay is reported evidence from the issue, not independently rerun here. | fixed-now | Exact shared replay plus generated membership crossing in the primary publication owner. | +| R2 | Identify the selected attempt and establish authority before changing implementation or model | Attempt 1, new sync run 1; source b is installed before settlement publication. Existing replacement law authorizes no delete. | fixed-now | Law, root cause and repair-location judgment above. | +| R3 | Preserve the nine-command history; cross rows, demand lifetimes and current/obsolete outcomes; require RED/GREEN/mutant | Both receiving paths fail unchanged production; all 72 constructed states are measured; original and narrow/broad mutants are rejected. | fixed-now | Primary oracle and coverage-map addition. No expectations, exclusions or existing budgets changed. | +| R4 | Keep separate from Solid PR #2307; #2121 retired-truncate, #1921 deferral handles, #2294 ordered replay and #2302 ordered retirement/provenance are distinct; maintainer requested a separate issue | Approved scope in #2310. The final production diff touches only Collection subscription reconciliation, not Solid, an ordered loader or the model. Historical related-PR claims were context, not independently audited. | accepted-design | This scope record. No GitHub issue or PR was changed. | + +Sibling S1: ordinary source delivery of an equal independent retained row also +vanishes. This is confirmed and fixed under the same law, with its own generated +receiving path and pre-edit RED. It is separate from the four raw items. + +Loss audit: **4 raw items = 3 fixed-now + 1 accepted-design**. No confirmed-open, +deferred, stale, refuted, design-decision or duplicate items remain. S1 is also +fixed. Item accounting and bounded implementation/oracle enforcement are +complete. PR readiness is **not assessed**: no PR was requested, created or +pushed with the repair, and whole-repository CI was not evaluated. + +## Final verification and code weight + +Code/test diff SHA-256, computed from `git diff --` for `subscription.ts`, the +publication owner and `oracle-config.ts`: +`2544ec5a5e5e2727a22be470ae53d3ca4ef4e6d31900cc326501cea60b877e27`. +This identifies the uncommitted repair in addition to its base HEAD. + +`pnpm size:compare --files` completed successfully against the same merge-base. +The final Collection subscription module grows **24 minified bytes**. The db +package grows **5 gzip bytes** and shrinks **4 Brotli bytes**. The typical app +grows **7 gzip bytes / 76 Brotli bytes**; all other package bundles are unchanged. +The production change reuses the existing private-row map and final diff; +it introduces no state, recovery path or helper. Refactoring the predicate from +an explicit undefined comparison to its equivalent Map truthiness reduced +module growth from 33 to 24 minified bytes, db gzip growth from 7 to 5 bytes, +and typical-app gzip growth from 9 to 7 bytes. Retaining the final diff and +stale-row removal is necessary: the omitted-row mutant demonstrates why removing +that work is incorrect. The remaining predicate growth pays for distinguishing +replacement membership from public delta equality. Tests and evidence documentation +grow separately and do not ship in the package. + +Formatting and `git diff --check` pass. The original 60-input broad fixed/random +campaign budgets are unchanged; the new 12-input structural campaigns add +coverage rather than replacing those budgets. + +Worktree setup: `.worktrees/codex-issue-2310-publication`, branch +`codex/issue-2310-publication`, based on freshly fetched `origin/main` at the +recorded HEAD. Initial publication to +`origin/codex/issue-2310-publication` succeeded. `pnpm install` succeeded without +local overrides. At that original checkpoint, the repair and evidence were uncommitted and no PR existed. + +Exact surrounding verification command: + +```sh +pnpm --filter @tanstack/db exec vitest run \ + tests/collection-subscription-lifecycle-publication-oracle.property.test.ts \ + tests/collection-subscription-lifecycle-history-oracle.property.test.ts \ + tests/collection-subscription-lifecycle-oracle.test.ts \ + tests/collection-subscription-replay-oracle.property.test.ts \ + tests/collection-subscription-lifecycle.test.ts \ + tests/collection-subscription-lifecycle-publication.test.ts \ + tests/collection-subscription-reentrancy-oracle.test.ts \ + tests/collection-cleanup-restart-oracle.test.ts \ + --maxWorkers=2 --coverage.enabled=false +``` + +Final TypeScript checking and focused ESLint both pass with no diagnostics. +`pnpm lint:changed` also passes with no reports; no oracle run budget was lowered +or removed. The final eight-suite run remains GREEN at 646 tests. + +## Prep-pr follow-up on current main + +The preceding `c351fc04d` entries describe the original investigation. During +prep-pr, `origin/main` advanced to +`bfd1da1a7e752d66fd667adfd3cdf54f60732b22` with the separate Effect membership +repair (#2301). This branch fast-forwarded with its uncommitted changes +preserved. The code/test diff SHA-256 remains +`2544ec5a5e5e2727a22be470ae53d3ca4ef4e6d31900cc326501cea60b877e27`. +No source or oracle repair was added during quality review. + +At that prep-pr checkpoint, the same eight-suite verification command above passes +**646 tests** and `pnpm --filter @tanstack/db exec tsc --noEmit` is clean. +An independent reviewer additionally ran the shared issue replay and four new +campaigns (**5 passed**) and the original exact seed/path replay (**1 passed**). +Both the independent correctness review and a separate simplification review +recommend no code changes. They agree that equal restored membership belongs +in the private replacement and the final diff owns redundant-update suppression. +The simplifier considered delivery-parameter and typed-owner-metadata alternatives +but declined them because the existing bounded grammar remains clear and small. + +The preceding paragraphs record the review conclusions and validation. There are no behavioral findings or law decisions to +resolve. Two declined style alternatives were evaluated, and the duplicated +request for current-head evidence is closed by this append. The review did not +change any law, exclusion, comparison, generator or run budget. This supplements +the earlier oracle-rubric audit; its finite direct-subscription limits remain. + +The required size comparison also succeeds against `bfd1da1a7`: subscription +**+24 minified bytes**, db **+5 gzip / +39 Brotli bytes**, typical app +**+7 gzip / +76 Brotli bytes**. Other package bundles are unchanged. These are +the current-base measurements; the earlier measurements remain attributed to +the original base. + +Local quality review is complete. PR readiness remains unproven until the +changeset, committed/pushed head and PR checks have been prepared and verified. + + +## CI replay registration follow-up + +PR #2313 first ran at `ebdd2140d6a25efdc6c747070f1385d4b367449d`. +The replay shard found a missing manifest owner for the newly registered +`subscription-lifecycle.restart-publication` property. The property's direct +Vitest replay was valid, but the guarded replay portfolio could not discover +its owner. This was a test-registration omission, not a changed product law +or a mismatch in the independent publication model. + +The existing `named oracle replay manifest` tests reproduce RED locally: +one failure names that property as `no-named-owner`. Adding its suffix beside +`publication-history` in `oracle-replay-manifest.ts` gives GREEN: all six +manifest tests pass. The same check rejects the omitted mapping, so it also +calibrates the registration repair. No production, model, history, comparison, +exclusion or run budget changed. + +```sh +pnpm --filter @tanstack/db exec vitest run tests/oracle-replay.test.ts \ + -t 'named oracle replay manifest' --maxWorkers=2 --coverage.enabled=false + +TANSTACK_DB_ORACLE_SEED=2310 TANSTACK_DB_ORACLE_PATH=0:0:0 \ +TANSTACK_DB_ORACLE_PROPERTY=subscription-lifecycle.restart-publication \ +pnpm --filter @tanstack/db exec node --import tsx tests/oracle-replay.ts \ + tests/collection-subscription-lifecycle-publication-oracle.property.test.ts \ + -t 'matches restarted row publication' --maxWorkers=2 --coverage.enabled=false +``` + +The guarded replay now passes both delivery paths and reports two successful +witnesses with the requested property, seed and path. This supplements the +earlier direct replay and closes the missing portfolio-discovery boundary. + +## Production-weight refactor + +The maintainer requested broader refactoring after PR #2313 reached +`e453634da7339e79d892a2bf198f0e5556828b12`. This follow-up preserves behavior: + +- Update `publishedRows` and `sentKeys` in one pass before calling the consumer. + Delete the two single-use tracking methods and the callback alias. +- Keep the three source listener removal callbacks in one ordered array. + Registration and removal retain their order. Unsubscription detaches the + array before invoking callbacks through `runAllCallbacks`, which still attempts + every callback and rethrows the first exact failure value. +- Initialize that array at registration, removing an unused initial allocation. + Cleanup-start registration can call synchronously, but the initial subscription + has no demands, rows or replay, and that path does not read the listener array + or expose the subscription to a consumer. + +The equal-row production predicate remains intact. No contract, model, grammar, +comparison, exclusion, replay registration or run budget changed. The existing +owners exercise the unchanged contract, so this internal refactor needs no new +oracle law or examples. The final production file's SHA-256 is +`f2ee13496359611fa5fe22066df96d80dec88a77fa18693af2fcfeff4ce29bc4`. + +### Measured shipped bytes + +`pnpm size:compare --files` compares the complete PR with merge-base +`bfd1da1a7e752d66fd667adfd3cdf54f60732b22`, after fetching `origin/main`. +Negative values mean fewer shipped bytes. Test and documentation additions +remain outside these production bundles. + +| Build | Gzip delta | Brotli delta | +| --- | ---: | ---: | +| Full DB package | -28 | -43 | +| Typical app | -32 | -8 | +| App with Effect | -33 | +4 | +| App with groupBy | -38 | +22 | +| App with includes | -33 | +92 | +| App with join | -33 | +76 | +| App with local storage | -32 | -110 | +| App with optimistic action | -33 | +54 | +| App with paced mutations | -33 | +26 | + +The subscription module shrinks **73 minified bytes** against the base. Other +package bundles remain unchanged. Every measured app shrinks under gzip. +Brotli results vary across feature scenarios, so the claim is negative weight +for the typical app and full DB bundle, not for every compressed app. + +The original PR grew 24 minified bytes, 5 DB gzip bytes and 7 typical-app gzip +bytes. Fusing the loops while retaining a single helper reduced the module but +still grew both gzip bundles by 2 bytes. A generic Map-fold helper also left +the DB gzip bundle 2 bytes larger and hid simple bookkeeping behind an +abstraction. Inlining the loop reduced DB gzip by 5 bytes but grew DB Brotli +by 88 bytes. The ordered listener array removes parallel optional fields and +wrapper callbacks, paying for the fix with simpler existing machinery. A +redundant callback alias and initial array allocation were then removed for +clarity and fewer allocations, without tuning names, comments or formatting. + +### Review and verification + +Independent correctness review found no defect or optional suggestion. It +checked tracking before reentry, repeated keys, callback throws, listener handle +ownership, synchronous cleanup registration, idempotence, cleanup order and +exact first-error propagation. Its selected final-wrapper checks passed 20 tests. +The simplification reviewer proposed one P3 item: remove the unused initial +listener array. That proposal was accepted under the maintainer's refactoring +request and verified with TypeScript and the surrounding suites. + +Lossless evaluation: **one raw suggestion = one fixed-now item**. No +confirmed-open, deferred, refuted, stale, design-decision or duplicate item +remains. The source-order suggestion and its rationale appear above. + +The exact eight-suite command recorded earlier passes **646 tests** on the +final refactor, with no type errors. `pnpm --filter @tanstack/db exec tsc --noEmit`, +ESLint, Prettier, `pnpm lint:changed`, and `git diff --check` pass. The latter +reports no lowered or removed oracle budget. + +The final-head sensitivity check temporarily restored the original equality +suppression while retaining the refactor. All four restart campaigns failed +at the shared publication comparison. Restoring the exact production file +made all four campaigns pass (139 unrelated owner tests deselected): + +```sh +pnpm --filter @tanstack/db exec vitest run \ + tests/collection-subscription-lifecycle-publication-oracle.property.test.ts \ + -t 'matches restarted row publication' \ + --maxWorkers=2 --coverage.enabled=false +``` + +This is renewed sensitivity evidence after an internal refactor. The earlier +oracle-first RED, two distinguishing mutants and direct-subscription limits +still govern the original repair. No temporary mutant remains. Local review +and implementation work are complete; final pushed-head CI is checked separately. + +## Private-member build-cache follow-up + +CI at `1aadaaedc1a2159fc4000a36867bd023896b409a` rejected five obsolete +private-member cache entries before the minified consumer test ran. Local +`pnpm run check:mangle` reproduced the same RED. Running +`node scripts/mangle-private-members.mjs --write` removes those five entries +and adds `sourceListenerCleanups`. Every surviving assignment stays unchanged. +The generated cache now passes its safety check with 437 names. + +`pnpm --filter @tanstack/db run build:minified` and +`pnpm run test:minified-db` then pass: 113 error names, index metadata, +query rows and live updates. This repairs build metadata, with no source, +contract, oracle or budget change. The source hash and 646-test result above +remain current. The earlier failed local consumer-test invocation rejected +stale dist output after temporary mutant restoration, before testing behavior. +The explicit fresh build resolves that setup condition. + +The corrected cache changes shipped bytes. These `pnpm size:compare --files` +results **supersede the provisional weight table above** at the same base: + +| Build | Gzip delta | Brotli delta | +| --- | ---: | ---: | +| Full DB package | -44 | +5 | +| Typical app | -45 | +20 | +| App with Effect | -45 | +37 | +| App with groupBy | -49 | -30 | +| App with includes | -45 | +64 | +| App with join | -45 | +37 | +| App with local storage | -44 | -14 | +| App with optimistic action | -46 | -65 | +| App with paced mutations | -46 | +47 | + +The subscription module shrinks **133 minified bytes**. Other package bundles +remain unchanged. The production source diff is 28 additions and 54 deletions +against the base. Tests and documentation grow separately. The final claim is +negative minified production weight and negative gzip across every measured +app. Brotli remains mixed, including +20 bytes for the typical app and +5 bytes +for DB. No names or formatting were tuned to change compression results. + + +## External review correction + +The external review evaluated published head `ef6316a91`. Its +[versioned evaluation](2026-10-30-pr-2313-review.md) records all ten findings. +The original 72 labels were constructed history categories, not 72 measured +production states. Fixture-dependent equality made that claim too strong. +The updated owner measures actual retained and delivered values, executed +settlement and owner lifetime, and replacement acquisition counts. It uses +51 distinct structural cases plus four three-relation receiving paths. +The production repair and the final shipped-byte measurements above remain +unchanged. diff --git a/packages/db/mangle-cache.json b/packages/db/mangle-cache.json index b69634e937..b636738f25 100644 --- a/packages/db/mangle-cache.json +++ b/packages/db/mangle-cache.json @@ -106,7 +106,6 @@ "retainedWindow": "b8", "setReadyIfIdle": "b9", "cachedCollectionRevision": "ca", - "collectionRestartCleanup": "cb", "pendingHydration": "cd", "removeFromBucket": "cf", "restorePublishedSnapshotTracking": "cg", @@ -178,7 +177,6 @@ "requireFullSourceRepair": "dy", "trackOrderedPublication": "dz", "canGarbageCollect": "dA", - "collectionCleanup": "dB", "extendPipelineWithChangeProcessing": "dC", "startSubsetDemand": "dD", "virtualPropsCache": "dE", @@ -212,7 +210,6 @@ "restoreInitialWindow": "d9", "settleTruncateReplay": "ea", "takeInternal": "eb", - "truncateCleanup": "ec", "_createRefForSource": "ee", "compileBasePipeline": "ef", "scheduleIdleCleanup": "eg", @@ -265,7 +262,6 @@ "publishEvents": "e5", "releaseDemand": "e6", "rollbackLease": "e7", - "trackSentKeys": "e9", "updateTimeout": "fa", "flushTruncateReplay": "fb", "accumulateActivity": "fc", @@ -275,7 +271,6 @@ "getOrCreateContext": "fg", "sentToD2Rows": "fh", "subscribeToChanges": "fi", - "trackPublishedRows": "fj", "_status": "fk", "clearSyncRunState": "fl", "retireEmptyReplay": "fm", @@ -439,5 +434,6 @@ "mergesOverRow": "b_", "awaitsSiblingReplays": "gN", "grownPrefixRows": "gZ", - "ownPrefixEnd": "he" + "ownPrefixEnd": "he", + "sourceListenerCleanups": "cb" } diff --git a/packages/db/src/collection/subscription.ts b/packages/db/src/collection/subscription.ts index de491b3325..428dd2fac8 100644 --- a/packages/db/src/collection/subscription.ts +++ b/packages/db/src/collection/subscription.ts @@ -225,10 +225,8 @@ export class CollectionSubscription promise: Promise }>() - // Cleanup function for truncate event listener - private truncateCleanup: (() => void) | undefined - private collectionCleanup: (() => void) | undefined - private collectionRestartCleanup: (() => void) | undefined + // Source listeners retire together on unsubscription. + private sourceListenerCleanups: Array<() => void> // One replay state owns the publication baseline, overlapping attempts, // and buffered changes until every attempt settles. @@ -273,16 +271,20 @@ export class CollectionSubscription ensureIndexForExpression(options.whereExpression, this.collection) } - const callbackWithSentKeysTracking = ( - changes: Array>, - ) => { - this.trackPublishedRows(changes) - this.trackSentKeys(changes) + this.callback = (changes) => { + // Complete both publication records before a callback can reenter. + for (const change of changes) { + if (change.type === `delete`) { + this.publishedRows.delete(change.key) + this.sentKeys.delete(change.key) + } else { + this.publishedRows.set(change.key, change.value) + this.sentKeys.add(change.key) + } + } callback(changes) } - this.callback = callbackWithSentKeysTracking - // Create a filtered callback if where clause is provided this.filteredCallback = options.whereExpression ? createFilteredCallback(this.callback, options) @@ -295,15 +297,10 @@ export class CollectionSubscription // row, so every one re-requests the loaded subsets, also one that a // reload's new run sends after the reload's replay began or settled: the // replacement deleted what that replay loaded. - this.truncateCleanup = this.collection.on(`truncate`, () => - this.handleTruncate(), - ) - this.collectionCleanup = this.collection._onCleanupStart(() => { - this.handleCollectionCleanup() - }) - this.collectionRestartCleanup = this.collection.on( - `status:change`, - ({ status }) => { + this.sourceListenerCleanups = [ + this.collection.on(`truncate`, () => this.handleTruncate()), + this.collection._onCleanupStart(() => this.handleCollectionCleanup()), + this.collection.on(`status:change`, ({ status }) => { if (status !== `loading` && status !== `ready`) return const syncRunGeneration = this.collection._sync.getSyncRunGeneration() const replayState = this.truncateReplayState @@ -319,8 +316,8 @@ export class CollectionSubscription this.restartDetachedDemands(syncRunGeneration) } }) - }, - ) + }), + ] } /** Whether this subscription still defers provider work. */ @@ -1906,7 +1903,12 @@ export class CollectionSubscription value: previous, previousValue: undefined, }) - } else if (!deepEquals(previous, change.value)) { + } else if ( + this.truncateReplayState?.privateRows || + !deepEquals(previous, change.value) + ) { + // An equal retained row still belongs in the private replacement. + // Its final diff suppresses the redundant public update. reconciled.push({ ...change, type: `update`, @@ -1929,28 +1931,6 @@ export class CollectionSubscription return reconciled } - private trackPublishedRows( - changes: Array>, - ): void { - for (const change of changes) { - if (change.type === `delete`) { - this.publishedRows.delete(change.key) - } else { - this.publishedRows.set(change.key, change.value) - } - } - } - - private trackSentKeys(changes: Array>) { - for (const change of changes) { - if (change.type === `delete`) { - this.sentKeys.delete(change.key) - } else { - this.sentKeys.add(change.key) - } - } - } - /** * Mark that the subscription should not filter any changes. * This is used when includeInitialState is explicitly set to false, @@ -1966,17 +1946,11 @@ export class CollectionSubscription // Stop any status listener set already being iterated. Clearing the // emitter's map cannot invalidate that captured Set by itself. this.statusRevision++ - const sourceListenerCleanups = [ - this.truncateCleanup, - this.collectionCleanup, - this.collectionRestartCleanup, - ] - this.truncateCleanup = undefined - this.collectionCleanup = undefined - this.collectionRestartCleanup = undefined + const sourceListenerCleanups = this.sourceListenerCleanups + this.sourceListenerCleanups = [] runAllCallbacks([ - ...sourceListenerCleanups.map((cleanup) => () => cleanup?.()), + ...sourceListenerCleanups, () => { // Stop any buffered replay from publishing after unsubscription. this.discardTruncateReplay() diff --git a/packages/db/tests/collection-restart-publication-boundaries.test.ts b/packages/db/tests/collection-restart-publication-boundaries.test.ts new file mode 100644 index 0000000000..e42cf0902b --- /dev/null +++ b/packages/db/tests/collection-restart-publication-boundaries.test.ts @@ -0,0 +1,122 @@ +import { expect, it, vi } from 'vitest' +import { + CollectionHasSubscribersError, + createCollection, + createLiveQueryCollection, +} from '../src/index.js' +import { flushPromises } from './utils.js' +import type { SyncConfig } from '../src/types.js' + +type Row = { id: string; value: number } +type Ops = Parameters[`sync`]>[0] + +// ARCHITECTURE's cleanup/restart contract gives an eager source's first batch +// replacement authority. Readiness is a separate checkpoint. +it(`reconciles retained eager rows on the first restarted public batch`, async () => { + let ops!: Ops + const source = createCollection({ + getKey: (row) => row.id, + sync: { + sync: (next) => { + ops = next + }, + }, + }) + const visible = new Map() + const batches: Array> = [] + const sub = source.subscribeChanges( + (changes) => { + batches.push(changes.map((change) => `${change.type}:${change.key}`)) + for (const change of changes) { + if (change.type === `delete`) visible.delete(change.key) + else visible.set(change.key, change.value.value) + } + }, + { includeInitialState: false }, + ) + const write = (keys: Array) => { + ops.begin() + keys.forEach((id) => ops.write({ type: `insert`, value: { id, value: 1 } })) + ops.commit() + } + try { + write([`a`, `b`, `c`]) + ops.markReady() + await source.cleanup() + source.startSyncImmediate() + batches.length = 0 + write([`a`]) + expect(source.status).toBe(`loading`) + expect([...visible.keys()]).toEqual([`a`]) + expect(batches).toEqual([[`delete:b`, `delete:c`]]) + write([`b`, `c`]) + ops.markReady() + expect([...visible.keys()]).toEqual([`a`, `b`, `c`]) + } finally { + sub.unsubscribe() + await source.cleanup() + } +}) + +// Cleanup ends a dependent graph. Source restart alone cannot revive it. +// These exact b/d observations supplement the cleanup admission oracle. +for (const syncMode of [`eager`, `on-demand`] as const) { + for (const cleanup of [`public`, `dependents`, `internal`] as const) { + it(`${syncMode} compiled queries obey ${cleanup} cleanup before restart`, async () => { + let ops!: Ops + let run = 0 + const reports = vi.spyOn(console, `error`).mockImplementation(() => {}) + const source = createCollection({ + getKey: (row) => row.id, + syncMode, + sync: { + sync: (next) => { + ops = next + run++ + ops.begin() + for (const id of run === 1 ? [`b`, `d`] : [`b`]) + ops.write({ type: `insert`, value: { id, value: 1 } }) + ops.commit() + ops.markReady() + return { loadSubset: () => true } + }, + }, + }) + const live = createLiveQueryCollection((q) => + q.from({ row: source }).select(({ row }) => row), + ) + const sub = live.subscribeChanges(() => {}) + try { + await live.preload() + expect([...live.keys()].sort()).toEqual([`b`, `d`]) + if (cleanup === `public`) { + await expect(source.cleanup()).rejects.toBeInstanceOf( + CollectionHasSubscribersError, + ) + expect(run).toBe(1) + expect(live.status).toBe(`ready`) + } else { + if (cleanup === `dependents`) + await source.cleanup({ dependents: true }) + else await source._lifecycle.cleanup() + if (cleanup === `internal`) { + expect(live.status).toBe(`crashed`) + source.startSyncImmediate() + await flushPromises() + expect(live.status).toBe(`crashed`) + expect([...live.keys()].sort()).toEqual([`b`, `d`]) + await live.cleanup() + } + await live.preload() + expect(live.status).toBe(`ready`) + expect([...live.keys()]).toEqual([`b`]) + } + } finally { + sub.unsubscribe() + await live.cleanup() + await source.cleanup() + reports.mockRestore() + } + }) + } +} diff --git a/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts b/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts index 882042c34f..1221fce236 100644 --- a/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts +++ b/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts @@ -15,6 +15,7 @@ import { } from './collection-subscription-lifecycle-grammar-oracle.js' import { oracleRandomParameters, readOracleRunConfig } from './oracle-config.js' import { declareOracle } from './oracle-declaration.js' +import { createOracleReach } from './oracle-reach.js' import { flushPromises } from './utils.js' import { assertTerminalPublication, @@ -102,7 +103,8 @@ type SourceMutation = { } type PublicationCommand = - | Exclude + | Exclude + | (Extract & { value?: number }) | { type: `truncate`; replacement?: Row } | SourceMutation @@ -161,6 +163,9 @@ type PublicationObservation = { unloads: number syncRuns: number collectionStatus: string + visibleBefore: Array + writtenRows: Array + currentAttempts: number } type PublicationMismatch = { @@ -437,7 +442,7 @@ function projectPublication( ({ attemptId }) => attemptId === attempt.id, ) if (command.outcome === `resolve` && isCurrent && !attempt.aborted) { - const row = { id: attempt.demand, value: attempt.id } + const row = { id: attempt.demand, value: command.value ?? attempt.id } const replacement = publication.replacement if ( replacement && @@ -557,6 +562,343 @@ const publicationCommandHistoryArbitrary: fc.Arbitrary< }), ) +/** + * Cleanup retains subscriber rows, not source rows. A successful restarted + * acquisition must publish the new source, including an equal retained row. + * Equality suppresses a public delta; it cannot erase replacement membership. + * This follows the opening publication law and the cleanup/restart grammar. + * + * Cross row equality with demand lifetime and delivery path. An ordinary source + * write can restore an independent retained row while acquisitions hold the + * replacement. Another retained row is deliberately omitted: keeping every old + * row would violate replacement just as deleting an equal restored row does. + * Current and obsolete resolve/reject outcomes distinguish authority from + * transport settlement. A failed attempt is followed by an authoritative retry. + * Every source operation is a complete transaction; malformed writes are excluded. + * Removing equality loses the reported defect, removing source delivery admits + * an acquisition-only patch, and removing omission admits a keep-all-rows patch. + */ +const restartOwnerStates = [ + `pending`, + `settled`, + `peer`, + `aborted`, + `released`, + `none`, +] as const +const restartRowStates = [`absent`, `equal`, `changed`] as const +const restartDeliveries = [`acquisition`, `source`] as const +const restartOutcomes = [`resolve`, `reject`] as const +const restartPublicationInputArbitrary = fc.record({ + demand: fc.constantFrom(`a` as const, `b` as const), + value: fc.integer({ min: -100_000, max: 100_000 }), + obsoleteOutcome: fc.constantFrom(...restartOutcomes), + priorAcquisitions: fc.integer({ min: 0, max: 3 }), +}) +type RestartPublicationInput = + typeof restartPublicationInputArbitrary extends fc.Arbitrary + ? Input + : never + +// Only live acquisitions distinguish success from failure or restored equality. +// An inactive acquisition has no delivered row: its retained value is one regime. +const restartScenarios = restartDeliveries.flatMap((delivery) => + restartOwnerStates.flatMap((owner) => { + const live = owner === `pending` || owner === `settled` || owner === `peer` + const rows = + delivery === `acquisition` && !live + ? ([`absent`, `retained`] as const) + : restartRowStates + return rows.flatMap((row) => + (live ? restartOutcomes : ([`noop`] as const)).map((outcome) => ({ + delivery, + owner, + row, + outcome, + state: `${delivery}:${owner}:${row}:${outcome}`, + })), + ) + }), +) + +function restartPublicationHistory( + { + demand, + value, + obsoleteOutcome, + priorAcquisitions, + }: RestartPublicationInput, + scenario: (typeof restartScenarios)[number], +) { + const { delivery, owner, row: rowState, outcome } = scenario + const other = demand === `a` ? `b` : `a` + const settle = ( + target: DemandName, + scope: `current` | `obsolete`, + result: `resolve` | `reject`, + ): PublicationCommand => ({ + type: `settle`, + demand: target, + scope, + age: `newest`, + outcome: result, + // Input data supplies row content. Neither model nor grammar predicts a + // fixture's attempt counter to manufacture equality. + value, + }) + const history: Array = [] + // Retired preliminary loads change attempt numbering without changing the + // target row. Equality must survive that fixture perturbation. + for (let index = 0; index < priorAcquisitions; index++) + history.push( + { type: `request`, demand: other }, + settle(other, `current`, `resolve`), + { type: `release`, demand: other }, + ) + if (owner !== `none`) history.push({ type: `request`, demand }) + if (owner === `peer`) history.push({ type: `request`, demand: other }) + if (owner === `settled`) history.push(settle(demand, `current`, `resolve`)) + const target = delivery === `source` ? `c` : demand + if (rowState !== `absent`) + history.push({ + type: `source`, + key: target, + action: `upsert`, + value: value + (rowState === `changed` ? 1 : 0), + }) + else if (owner === `settled` && delivery === `acquisition`) + history.push({ type: `source`, key: target, action: `delete`, value: 0 }) + history.push({ type: `source`, key: `d`, action: `upsert`, value }) + if (owner === `aborted`) history.push({ type: `abort`, demand }) + if (owner === `released`) history.push({ type: `release`, demand }) + const cleanupIndex = history.length + history.push({ type: `cleanup` }, { type: `restart` }) + const restartIndex = history.length - 1 + history.push(settle(demand, `obsolete`, obsoleteOutcome)) + if (delivery === `source`) + history.push({ type: `source`, key: target, action: `upsert`, value }) + const focalIndex = history.length - (delivery === `source` ? 1 : 0) + history.push( + settle(demand, `current`, outcome === `noop` ? `resolve` : outcome), + ) + if (owner === `peer`) history.push(settle(other, `current`, `resolve`)) + if (outcome === `reject`) { + history.push({ type: `truncate` }, settle(demand, `current`, `resolve`)) + if (owner === `peer`) history.push(settle(other, `current`, `resolve`)) + } + history.push({ type: `release`, demand }) + if (owner === `peer`) history.push({ type: `release`, demand: other }) + return { history, target, cleanupIndex, restartIndex, focalIndex } +} + +// Reach is conditional on facts recorded by the real driver, before comparison +// can hide a missing premise. A successful acquisition records its actual write; +// failure records its settled outcome. Retained equality compares that write +// with the callback-applied row before cleanup, never with an input label. +function observeRestartRow( + prior: Row | undefined, + written: Row | undefined, +): `absent` | `retained` | `equal` | `changed` { + if (!prior) return `absent` + if (!written) return `retained` + return prior.value === written.value ? `equal` : `changed` +} + +// Owner lifetime is observed from executed operations and adapter callbacks. +// This is a reach classifier, not the reference publication computation. +function observeRestartOwner( + observations: ReadonlyArray, + history: ReadonlyArray, + demand: DemandName, +): (typeof restartOwnerStates)[number] { + const executed = observations.flatMap((item) => + item.executed ? [history[item.index]!] : [], + ) + if ( + !executed.some((item) => item.type === `request` && item.demand === demand) + ) + return `none` + if ( + executed.some((item) => item.type === `release` && item.demand === demand) + ) + return `released` + if (executed.some((item) => item.type === `abort` && item.demand === demand)) + return `aborted` + if (observations.at(-1)!.currentAttempts === 2) return `peer` + return executed.some( + (item) => + item.type === `settle` && + item.demand === demand && + item.scope === `current`, + ) + ? `settled` + : `pending` +} + +/** + * A restarted direct subscription publishes the current replacement at applied + * acquisition settlement (ARCHITECTURE, "Cleanup, restart, and detached waiters"). + * Subscription predicates restrict membership; order and limit select a local + * snapshot, not the order of later callback batches. The same Map diff above + * judges these receiving paths, including a snapshot of installed rows whose + * source events were withheld. Source event deferral is an internal controlled + * seam; this makes no claim about a provider's delivery policy. + */ +const restartReceiverPaths = [ + `direct`, + `filtered`, + `ordered`, + `snapshot`, +] as const +async function runRestartReceiver( + input: RestartPublicationInput, + path: (typeof restartReceiverPaths)[number], + relation: (typeof restartRowStates)[number], +): Promise { + const gate = createDeferred() + let operations: SyncOperations | undefined + let restarting = false + let loads = 0 + const source = createCollection({ + getKey: ({ id }) => id, + syncMode: `on-demand`, + sync: { + sync: (ops) => { + operations = ops + ops.markReady() + return { + loadSubset: () => { + loads++ + return restarting ? gate.promise : true + }, + } + }, + }, + }) + const shown = new Map() + const batches: Array> = [] + const subscription = source.subscribeChanges( + (changes) => { + const batch = changes.map((change): PublicationChange => ({ + type: change.type, + key: change.key, + value: cloneRow(change.value), + ...(change.previousValue + ? { previousValue: cloneRow(change.previousValue) } + : {}), + })) + for (const change of batch) { + if (change.type === `delete`) shown.delete(change.key) + else shown.set(change.key, cloneRow(change.value)) + } + batches.push(batch) + }, + { + includeInitialState: false, + ...(path === `filtered` + ? { + whereExpression: new Func(`not`, [ + new Func(`eq`, [new PropRef([`id`]), new Value(`b`)]), + ]), + } + : {}), + }, + ) + const selected = (rows: ReadonlyArray) => + new Map( + rows + .filter(({ id }) => path !== `filtered` || id !== `b`) + .map((row) => [row.id, cloneRow(row)] as const), + ) + const write = (rows: ReadonlyArray) => { + if (!operations) throw new Error(`missing receiver sync run`) + operations.begin() + for (const row of rows) + operations.write({ type: `insert`, value: cloneRow(row) }) + const receipt = operations.commit() + expect(receipt).toBe(true) + } + const priorRows: Array = [ + ...(relation === `absent` + ? [] + : [ + { + id: `a` as const, + value: input.value + (relation === `changed` ? 1 : 0), + }, + ]), + { id: `b`, value: input.value }, + { id: `d`, value: input.value }, + ] + const nextRows: Array = [ + { id: `a`, value: input.value }, + { id: `b`, value: input.value + 2 }, + ] + const failures: Array = [] + let reached = `` + try { + write(priorRows) + subscription.requestSnapshot() + await flushPromises() + expect(shown).toEqual(selected(priorRows)) + const prior = new Map(shown) + await source.cleanup() + restarting = true + source.startSyncImmediate() + await flushPromises() + expect(loads).toBe(2) + expect(subscription.status).toBe(`loadingSubset`) + const start = batches.length + // A snapshot must be able to supply replacement membership independently + // of source callbacks. The held events still publish before settlement. + const held = path === `snapshot` ? source._deferPublication() : undefined + write(nextRows) + if (path === `snapshot` || path === `ordered`) { + subscription.requestSnapshot( + path === `ordered` ? { orderBy: valueOrderBy(), limit: 3 } : {}, + ) + } + held?.publish() + expect([...source.values()].map(cloneRow)).toEqual(nextRows) + expect(shown).toEqual(prior) + expect(batches.slice(start)).toEqual([]) + const observedRelation = observeRestartRow( + prior.get(`a`), + cloneRow(source.get(`a`)!), + ) + expect(observedRelation).toBe(relation) + reached = `${path}:${observedRelation}` + gate.resolve() + await flushPromises() + const next = selected(nextRows) + expect(normalizePublicationOrder(batches.slice(start))).toEqual( + normalizePublicationOrder([publicationDiff(prior, next)]), + ) + expect(shown).toEqual(next) + expect(subscription.status).toBe(`ready`) + } catch (error) { + failures.push(error) + } finally { + gate.resolve() + try { + subscription.unsubscribe() + } catch (error) { + failures.push(error) + } + try { + await source.cleanup() + } catch (error) { + failures.push(error) + } + } + if (failures.length === 1) throw failures[0] + if (failures.length > 1) + throw new AggregateError(failures, `receiver and cleanup failed`, { + cause: failures[0], + }) + return reached +} + async function runPublicationHistory( history: ReadonlyArray, runOptions: PublicationRunOptions = {}, @@ -711,13 +1053,16 @@ async function runPublicationHistory( syncRuns: operationsBySyncRun.size, }) - const writeAttempt = async (attempt: RuntimeAttempt): Promise => { + const writeAttempt = async ( + attempt: RuntimeAttempt, + rowValue = attempt.id, + ): Promise => { // Cancellation fences request-scoped writes at the adapter boundary. // Transport may settle later; it must not publish canceled snapshot rows. if (attempt.signal?.aborted) return const rows = sourceRows.get(attempt.syncRunGeneration) const previous = rows?.get(attempt.demand) - const value = { id: attempt.demand, value: attempt.id } + const value = { id: attempt.demand, value: rowValue } attempt.operations.begin() attempt.operations.write({ type: previous ? `update` : `insert`, @@ -727,6 +1072,7 @@ async function runPublicationHistory( const receipt = attempt.operations.commit() if (receipt !== true) await receipt rows?.set(attempt.demand, value) + return cloneRow(value) } const assertPublications = ( @@ -788,6 +1134,8 @@ async function runPublicationHistory( const expectedPublicationCount = publication.batches.length const unloadCount = unloads.length const syncRunCount = operationsBySyncRun.size + const visibleBefore = [...visible.values()].map(cloneRow) + const writtenRows: Array = [] let executed = false let sourceEffect: SourceEffect | undefined let settlement: `resolve` | `reject` | undefined @@ -840,6 +1188,7 @@ async function runPublicationHistory( ...(previous ? { previousValue: previous } : {}), }) rows?.set(command.key, value) + writtenRows.push(cloneRow(value)) } const receipt = operations?.commit() if (receipt !== true) await receipt @@ -877,7 +1226,8 @@ async function runPublicationHistory( effect.attemptId ] as LifecycleAttempt if (command.outcome === `resolve`) { - await writeAttempt(runtimeAttempt) + const written = await writeAttempt(runtimeAttempt, command.value) + if (written) writtenRows.push(written) runtimeAttempt.deferred.resolve() } else { runtimeAttempt.deferred.reject(expected.failure) @@ -981,6 +1331,11 @@ async function runPublicationHistory( unloads: unloads.length - unloadCount, syncRuns: operationsBySyncRun.size - syncRunCount, collectionStatus: collection.status, + visibleBefore, + writtenRows, + currentAttempts: [...attempts.values()].filter( + ({ current, signal }) => current && !signal?.aborted, + ).length, }) } if (runOptions.terminal) { @@ -1458,8 +1813,10 @@ function expectedSelection( // and unloads with the model. The model predicts the publication step by // step, and the selection model predicts which keys local snapshots select. declareOracle({ - check: [runPublicationHistory, runPublicationProduct], + check: [runPublicationHistory, runPublicationProduct, runRestartReceiver], model: [ + observeRestartRow, + observeRestartOwner, projectPublication, publicationDiff, publishIfChanged, @@ -1467,7 +1824,12 @@ declareOracle({ finishReplacement, expectedSelection, ], - grammar: [deferralCleanupHistories, publicationProductCases], + grammar: [ + deferralCleanupHistories, + publicationProductCases, + restartPublicationInputArbitrary, + restartPublicationHistory, + ], }) describe(`CollectionSubscription lifecycle publication oracle`, () => { @@ -2387,7 +2749,151 @@ describe(`CollectionSubscription lifecycle publication oracle`, () => { { type: `unsubscribe` }, ]) }) + it(`keeps an equal row restored by the restarted acquisition`, async () => { + // Original generated failure: seed -1439698414, path 1:22. + await runPublicationHistory([ + { type: `source`, key: `b`, action: `upsert`, value: 1 }, + { type: `request`, demand: `b` }, + { type: `cleanup` }, + { + type: `settle`, + demand: `a`, + scope: `current`, + age: `oldest`, + outcome: `resolve`, + }, + { type: `restart` }, + { + type: `settle`, + demand: `b`, + scope: `current`, + age: `newest`, + outcome: `resolve`, + }, + { type: `abort`, demand: `a` }, + { + type: `settle`, + demand: `a`, + scope: `current`, + age: `oldest`, + outcome: `reject`, + }, + { type: `release`, demand: `b` }, + ]) + }) + const { multiplier, ...replay } = readOracleRunConfig() + const restartProperty = `subscription-lifecycle.restart-publication` + // Each generated input runs every structural lane. Reach never depends on + // the random seed choosing equality, a live owner, or ordinary source delivery. + // Both campaigns use the same budget, grammar, model, driver and comparisons. + for (const seed of replay.replaySeed === undefined + ? [2310, undefined] + : [undefined]) { + for (const delivery of restartDeliveries) { + it(`matches restarted row publication ${delivery} (${seed === undefined ? `random or replayed seed` : `fixed seed`})`, async () => { + const scenarios = restartScenarios.filter( + (item) => item.delivery === delivery, + ) + const reach = createOracleReach( + restartProperty, + scenarios.map(({ state }) => state), + { minimum: 12 * multiplier }, + ) + await fc.assert( + fc.asyncProperty(restartPublicationInputArbitrary, async (input) => { + for (const scenario of scenarios) { + reach.beginHistory() + const { + history, + target, + cleanupIndex, + restartIndex, + focalIndex, + } = restartPublicationHistory(input, scenario) + const observations = await runPublicationHistory(history) + expect(observations[restartIndex]?.executed).toBe(true) + expect(observations[restartIndex]?.syncRuns).toBe(1) + const focal = observations[focalIndex]! + const prior = observations[cleanupIndex]!.visibleBefore.find( + ({ id }) => id === target, + ) + const written = focal.writtenRows.find(({ id }) => id === target) + const current = observations.find((item) => { + const command = history[item.index] + return ( + item.index > restartIndex && + command?.type === `settle` && + command.scope === `current` + ) + })! + const actualOwner = observeRestartOwner( + observations.slice(0, cleanupIndex), + history, + input.demand, + ) + expect(actualOwner).toBe(scenario.owner) + const actualOutcome = current.settlement ?? `noop` + // Failed acquisitions write nothing. Observe the row relation on + // their authoritative retry, after proving the actual rejection. + const delivered = + written ?? + (actualOutcome === `reject` + ? observations + .slice(focalIndex + 1) + .flatMap((item) => item.writtenRows) + .find(({ id }) => id === target) + : undefined) + const actualRow = observeRestartRow(prior, delivered) + expect(actualOutcome).toBe(scenario.outcome) + expect(actualRow).toBe(scenario.row) + const live = + scenario.owner === `pending` || + scenario.owner === `settled` || + scenario.owner === `peer` + expect(observations[restartIndex]!.currentAttempts).toBe( + live ? (scenario.owner === `peer` ? 2 : 1) : 0, + ) + reach.reach( + `${delivery}:${actualOwner}:${actualRow}:${actualOutcome}`, + ) + } + }), + { + ...oracleRandomParameters(12 * multiplier, replay, restartProperty), + ...(seed === undefined ? {} : { seed }), + }, + ) + reach.assertReached() + }, 120_000) + } + } + for (const seed of replay.replaySeed === undefined + ? [2310, undefined] + : [undefined]) { + for (const path of restartReceiverPaths) { + it(`matches restarted row publication receiver ${path} (${seed === undefined ? `random or replayed seed` : `fixed seed`})`, async () => { + const reach = createOracleReach( + restartProperty, + restartRowStates.map((row) => `${path}:${row}`), + { minimum: 12 * multiplier }, + ) + await fc.assert( + fc.asyncProperty(restartPublicationInputArbitrary, async (input) => { + for (const row of restartRowStates) { + reach.beginHistory() + reach.reach(await runRestartReceiver(input, path, row)) + } + }), + { + ...oracleRandomParameters(12 * multiplier, replay, restartProperty), + ...(seed === undefined ? {} : { seed }), + }, + ) + reach.assertReached() + }, 120_000) + } + } const runs = 60 * multiplier fcTest.prop([publicationCommandHistoryArbitrary], { numRuns: runs, diff --git a/packages/db/tests/oracle-config.ts b/packages/db/tests/oracle-config.ts index 4c2f38468a..edf37d033c 100644 --- a/packages/db/tests/oracle-config.ts +++ b/packages/db/tests/oracle-config.ts @@ -320,6 +320,7 @@ const staticOracleProperties = [ `subscription-replay.same-tick`, `subscription-lifecycle.history-statistics`, `subscription-lifecycle.publication-history`, + `subscription-lifecycle.restart-publication`, `subscription-lifecycle.sync-history`, `subscription-lifecycle.async-history`, `subscription-lifecycle.demand-change`, diff --git a/packages/db/tests/oracle-replay-manifest.ts b/packages/db/tests/oracle-replay-manifest.ts index 222631c23a..6129f2f86c 100644 --- a/packages/db/tests/oracle-replay-manifest.ts +++ b/packages/db/tests/oracle-replay-manifest.ts @@ -195,7 +195,7 @@ const ownerGroups: ReadonlyArray< [ `db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts`, `subscription-lifecycle`, - `publication-history`, + `publication-history restart-publication`, ], [ `db/tests/collection-subscription-lifecycle-history-oracle.property.test.ts`,