diff --git a/docs/snapshot-checkpoints.md b/docs/snapshot-checkpoints.md new file mode 100644 index 0000000..29bb24e --- /dev/null +++ b/docs/snapshot-checkpoints.md @@ -0,0 +1,340 @@ +# Snapshot checkpoints + +Design notes for fixing sillsdev/harmony#105. Written after a long investigation with a lot of +dead ends; the point of this file is so nobody repeats them. Everything under "Measurements" was +produced by running code, not reasoned about. + +## The problem + +Commits are the source of truth, snapshots are derived state. When a commit arrives dated before +commits already in the database (a "late commit", L), `AddNewCommits` sets the replay window to all +commits after `parent(L)`, `UpdateSnapshots` deletes snapshots after L, and the replay resumes each +entity from whatever snapshot survived. + +That resume is the bug. An entity's newest surviving snapshot can be *older* than `parent(L)`, +because the snapshot at its last touch was pruned by the `CommitIndex % 2 == 0` rule in +`SnapshotWorker.GenerateSnapshotForEntity`. The commits between that snapshot and `parent(L)` are +not in the window, so nothing re-applies them. Two symptoms, both with repro tests in +`src/SIL.Harmony.Tests/LateCommitTests.cs`: + +1. Silent loss of an edit. Recoverable by `RegenerateSnapshots`. +2. A cascade-deleted entity comes back to life, its projected row is re-inserted, and the FK to its + deleted parent fails: `SQLite Error 19: 'FOREIGN KEY constraint failed'`. This wedged a + production project. Sync fails hard until repaired. + +The cut and the window already share a boundary (no commit lies strictly between `parent(L)` and L), +so this is not a cut/window mismatch. It is purely about what the resume snapshot is. + +## The three concepts + +**Completeness at P.** For every entity E, E's newest snapshot at or before P is at or after E's +most recent touch at or before P. "Touch" means a change to E, or a cascade from a delete in that +commit. This is a property of P alone, not of the prefix before it. + +**Checkpoint.** A commit at which completeness holds, so it is safe to resume a replay from. +Critically, a checkpoint does **not** mean a snapshot row dated at that commit for every entity. +The project-wide state at a checkpoint is virtual: the union over entities of each one's newest +snapshot at or before it. That is the whole economy of the idea. An entity untouched for 10,000 +commits contributes its old row and costs nothing. + +**Hole.** When E's snapshot at commit `t` is dropped and E's next surviving snapshot is at `n`, +every position in `[t, n)` is unsafe for E. A hole is an interval, not a point. Getting this wrong +is what made the first attempt lose data (see Corrections). + +## What we are building + +- **Explicit checkpoints, marked with a bool on `Commit`.** Local-only bookkeeping: `[JsonIgnore]`, + never synced, not part of the commit hash (the hash is `f(Id, parentHash)` only, so a new column + is safe). Different devices will legitimately have different checkpoint sets. +- **A bool is enough.** A generation/level int would make tiered thinning declarative, but thinning + can pick which flags to clear by any criterion at the time (for example keep every 4th flagged + commit in commit order), so the extra column buys nothing now. Add it later if thinning wants it. + A datestamp buys nothing at all: commits already carry their own date, which is what you would + thin by. +- **The flag is a decision, not a record.** Choose which commits will be checkpoints, then make the + pruner respect that choice: never drop a snapshot if the resulting hole would span a chosen + checkpoint. This is the opposite direction of causality from the first attempt, which recorded + after the fact which commits happened to be safe. +- **Many checkpoints per `SnapshotWorker` run, not one.** Density is the only dial in the design + (see below). Any policy works because the choice is recorded: every Nth commit by position in the + batch is the simplest, and unlike `% 2` it is entirely under our control. +- **Consumers need a migration** for the new `Commits` column. Harmony ships no migrations. + +### Density is the only dial + +Once checkpoints exist, a snapshot earns its keep only by being some entity's resume state at some +checkpoint. Everything else is dead weight. So there is no separate pruning policy to design: + +| checkpoint density | what it is | +|---|---| +| every commit | never prune anything | +| one in K | the design here | +| dense at the head, sparse with age | thinning, deferred | + +Storage scales with density. Rollback distance and point-in-time read cost scale inversely with it. +Both costs move the same way against storage, so there is one number to choose, not two. + +The only exception to "worthless unless a checkpoint needs it" is each entity's newest snapshot, +which is its current state and feeds `CurrentSnapshots`, `GetLatest`, and the projected row that +references it by id. That is the same rule read forwards: it is the resume state for every +checkpoint after it, including ones that do not exist yet. + +Roots are worth keeping unconditionally, but as insurance rather than necessity. By the rule a root +whose interval spans no checkpoint is worthless; deleting one makes a later resume apply an edit +against a null snapshot, which throws instead of degrading. One row per entity is cheap insurance. + +## Rules that must not be broken + +1. **Never add a flag retroactively.** Claiming safety at a commit that pruning has already holed is + silent data loss. Flags may only be set on commits inside a window being replayed, since that + window is re-pruned under the current policy and completeness there is under our control. + Removing flags is always safe. There are no exceptions to this; see "Existing projects" for the + one that was considered and dropped. +2. **No checkpoint lookup may guess.** If there is no flagged commit before L, that means replay + everything: `DeleteSnapshotsAfter(null)` and replay all of history. Do not derive, infer, or + approximate a checkpoint. The first attempt did and it picked unsafe commits. +3. **Deleting a snapshot is no longer free.** This was the assumption before this work. A snapshot + may only be deleted if the hole it creates spans no live checkpoint. +4. **Thinning order.** Drop flags first, then sweep snapshots. The reverse breaks resume points. +5. **Checkpoints may only ever be coarsened.** Going back to a finer density needs a regenerate. + +## Corrections to intuitions that turned out wrong + +- **A hole is an interval.** The first pushed attempt marked only the commit whose snapshot was + dropped. Demonstrated failure: entity A touched at c1/c3/c5 and B at c2/c4, one batch, c3 setting + a note and c5 setting text. A's snapshot at c3 is pruned, flags come out c1=T c2=T c3=F c4=T c5=T, + and a late commit landing between c4 and c5 resumes A from c1 and loses the note. c4 is inside + A's hole `[c3, c5)` but was marked safe. +- **A commit-local predicate cannot detect holes.** "Every change at this commit has a snapshot + here" is local and fails the same way. It is also blind to cascades: a cascade writes a snapshot + for an entity with no `ChangeEntity`, so a pruned cascade snapshot is invisible to any anti-join + over `ChangeEntities`. Verified: `ChangeEntities` for a cascade-deleted definition at its delete + commit is 0. +- **`% 2` is not a checkpoint policy.** It is a retention policy on individual snapshots, and it + gates on the index of the commit *doing the dropping*, which says nothing about where the hole + falls. Safety at a position is a conjunction over all entities, so holes from different entities + union together and cover nearly everything. Simulated: 0.6% to 1.4% of positions safe in synced + batch shapes, 35% when the batch is mostly creates. It does produce checkpoints reliably in one + place, where it never fires: a locally authored commit is its own batch with nothing prunable, so + every one is safe. That is why this bug hides until a project syncs or clones. +- **Position parity in history does not work either.** "Call every second commit a checkpoint" was + disproven: in the A/B history above the hole lands at odd index 3 while both even indices are safe. +- **Changes read other entities far more than assumed.** 15 of 38 change files in FW Lite's + `LcmCrdt/Changes` read at apply time. `IsObjectDeleted` is a default method on `IChangeContext` + that wraps `GetSnapshot`, which is easy to miss when grepping; five sample change files call it, + including `NewDefinitionChange` and `NewExampleChange`. Consequence: per-entity state is not + independently meaningful, so completeness has to be project-wide. This is also why every scheme + that replays only some entities is unsound without tracking read sets. +- **Every replay path is affected, not just the late-commit path.** `GetSnapshotAtCommit` seeds from + the subject's own nearest snapshot and replays the whole range after it, so the subject's own chain + is correct. But the neighbours it reads are seeded from the scoped repository at "newest at or + before X", which is both the wrong position (state as of X, not the replay position) and possibly + stale. Cascades and `IsObjectDeleted` then compute from wrong state and feed the subject. + +## Consequences for the other replay paths + +`GetSnapshotAtCommit` should resume from the newest checkpoint at or before X, seeding every entity +from its newest snapshot at or before that checkpoint (scoped to the checkpoint, not to X as today), +then replay the range. Completeness at the checkpoint makes every seed correct, and every entity +that changes inside the range is replayed alongside the subject so reads land at the right position. +Both the staleness and the position error go away. It does not need a checkpoint at which the +subject itself was touched; any checkpoint works. + +This also means `UpdateSnapshots` and `GetSnapshotAtCommit` share one primitive: "resume from the +newest checkpoint at or before P". + +**Do not filter commits inside that range** without tracking read sets. Skipping commits that do not +touch the subject reintroduces exactly the neighbour-read bug: a neighbour that changed inside the +range would be left at its checkpoint state while the subject's later change reads it. With dense +checkpoints the range is small, so replay it fully. Prefer density over filtering. + +## Pulling selection out of the playback + +Deciding what to persist after the playback instead of during it is the right shape, but holding +every generated snapshot in memory is not viable: a clone or regenerate is one batch of the whole +history, so that is up to one snapshot per touch (hundreds of thousands of entity JSON blobs). + +Checkpoints give the bounded version for free. No hole can span a checkpoint, so no decision about a +snapshot before one can depend on anything after it. Decide and flush at each checkpoint, holding at +most one checkpoint interval's worth of snapshots, O(K) rather than O(batch). At each checkpoint the +rule collapses to something trivial: every entity touched since the previous checkpoint keeps its +latest snapshot, everything else it accumulated is discarded. + +Extract the decision as a pure function over (previous snapshot commit, current commit, checkpoint +set). The same function is what thinning runs later, and it can be tested without driving a replay. + +## What shipped + +`SnapshotCheckpointPolicy` holds both halves of the decision at an interval of 8: `IsCheckpoint` picks every 8th commit +of a replayed batch plus its last, and `MustKeepSnapshot` is the pure function the pruner and, later, thinning share. +`CrdtRepository.SetCheckpoints` writes the flags before the replay starts. `DataModel.ResumeFromCheckpoint` is the shared +primitive, used by `UpdateSnapshots`, `GetSnapshotsAtCommit` and `GetSnapshotAtCommit`. + +Two things from the sections above were left alone: + +- **The playback still decides during the walk**, incrementally, rather than accumulating an interval's snapshots and + deciding at each checkpoint. Same outcome from the same function, and the restructure would not save much: the batch + already holds a snapshot per touched entity in `_pendingSnapshots` regardless, which is the O(batch) part. +- **Reading state at an old commit on a database with no checkpoints replays all of history**, since there is nothing to + resume from. Correct but slow, and it lasts until the first late commit establishes checkpoints. + +## Deferred: thinning + +Not in the first change. Recorded here so we know the room exists. + +Keep snapshot `s` for entity E at commit `c` iff a live checkpoint lies in `[c, next snapshot of E)`. +Two properties make this cheap: + +- **It is one pass, not a fixpoint.** Deleting `s(i)` widens `s(i-1)`'s gap, but if no checkpoint sits + in `[s(i-1), s(i))` and none sits in `[s(i), s(i+1))` then none sits in the union, so every snapshot + can be evaluated independently against its *original* neighbour and deleted in a single sweep. +- **It is the same predicate the pruner uses at write time**, just applied to rows already on disk. + +Query shape: each snapshot paired with the next commit for the same entity, so a `LEAD` window +function over Snapshots joined to Commits in commit order, or per-entity grouping in memory. The +`Snapshots(EntityId)` index supports the ordering. + +Always keep each entity's newest snapshot (current state, referenced by the projected row) and its +root (insurance, see above). + +## Existing projects + +**Decision: do nothing. Ship no migration step, no detection, and do not mark anything.** + +Legacy databases have no flags, so the first late commit finds no checkpoint and replays all of +history, which rebuilds every snapshot correctly and establishes checkpoints throughout. That is +simultaneously the repair and the bootstrap, triggered automatically at the moment it is actually +needed, with no upgrade path to write and nothing to detect. + +The reasoning for not marking the newest commit, which was the obvious alternative: + +- **We cannot know whether a client's snapshots are already broken.** Corruption is per device, + because it depends on the order batches arrived, and it lives in snapshot *content* rather than in + any structural property. Nothing in the database records that a past replay resumed inside a hole, + so there is no signature to look for short of recomputing and diffing. +- **Marking the head would freeze that corruption.** The flag would be structurally honest (no + earlier snapshot is needed to resume at the head) while resuming from wrong values, and narrow + rollbacks would then never reach back past it. Today's wide rollbacks occasionally heal corruption + by accident; marking the head removes even that. +- **Marking the head does not avoid the full replay, it defers it.** Any late commit dated before the + marker still finds no checkpoint before it and replays everything. In a multi-device workflow that + is very likely: as soon as a client has edited locally, a peer's commits from the meantime are + interleaved with its own, and any of them dated before the client's last local edit triggers the + rewind. So the cost arrives anyway, just at an unpredictable moment and without the repair. + +Accepted consequences: + +- The first sync that carries a late commit is slow, once, on the order of seconds to a few minutes + by the measurements below. Afterwards checkpoints exist and rollbacks are narrow. +- A client that never receives a late commit never heals and never gets checkpoints. That is + self-consistent, since it also never rewinds, but any existing corruption stays and can still + escape at the FieldWorks sync boundary, where current snapshot state is authoritative rather than + the commits. Devices that sync to FLEx are therefore the ones worth watching, and + `RegenerateSnapshots` stays the support path for them. +- **Implementation note:** when no checkpoint is found, take the regenerate path rather than the + rewind path. A rewind covering all of history measured about 3x more per commit than a regenerate, + because it replays against a still-populated table, and it leaves the projected tables to be + updated row by row instead of rebuilt. See the crossover rule under Measurements. + +Unapplied changes (an `OpaqueChange` this client cannot apply, or a change that does not support +updating an existing entity) produce no snapshot but also no state change, so they do not affect +completeness. + +## Measurements + +All from probes run during the investigation. Release, .NET 10, SQLite unless noted. + +**Pruning storage saving** +- Nothing at all when an entity is touched twice in a batch: measured byte-identical databases, + because the only edit's previous snapshot is the root and the `!IsRoot` guard fails. +- 500 entities x 11 touches in one batch (5500 commits): 3250 snapshots pruned vs 5500 unpruned + (0.59x), snapshot payload 0.59x, vacuumed file 4.53 MB vs 5.64 MB (0.80x). Marginal disk cost of a + snapshot row is about 490 bytes including indexes; payload alone about 177 bytes. +- Extrapolated to 50,000 entities x 11 touches: about 225,000 extra rows, roughly 110 MB at sample + entity sizes. FW Lite entries are plausibly 1 to 3 KB, which would put it at 300 to 500 MB. + **Nobody has measured a real project.** This query gives the exact number of rows never-prune + would add on any database: + `SELECT count(*) FROM (SELECT DISTINCT CommitId, EntityId FROM ChangeEntities) ce WHERE NOT EXISTS + (SELECT 1 FROM Snapshots s WHERE s.CommitId = ce.CommitId AND s.EntityId = ce.EntityId)` +- Pruning is intra-batch only (`IsNew(prevSnapshot)`), so the device that authored a history already + stores 100% of its snapshots. The saving only exists on devices that received history in bulk: a + fresh clone, and `RegenerateSnapshots`. +- The existing `DataModelPerformanceTests` never exercises pruning at all: every change in it creates + a brand new entity, so the parity branch is unreachable. + +**Replay** +- `RegenerateSnapshots`, 10k commits: 1186 ms projected on, 1013 ms off, so 0.10 to 0.12 ms per + commit with small entities. A second harness measured about 1 ms per commit plus 80 ms fixed; + the difference is consistent with Debug, larger entities, or validation on. Either way 100k commits + is minutes, not hours. +- Split of replay time with projection on: `ApplyCommitChanges` 23%, `AddSnapshots` 51% (of which + `ProjectSnapshot` 15% and `SaveChanges` 32%), loading commits and dropping tables 26%. +- Late commit into a 10k history: 100 back 173 ms, 1000 back 996 ms, 5000 back 2013 ms. +- **Rewinding costs about 3x per commit compared with regenerating** (2013 ms for a 5000-commit + rewind vs 1186 ms for a full 10k regenerate), because rewind replays against a still-populated + table. Practical consequence, independent of this design: when the window covers more than roughly + half of history, drop everything and regenerate instead of rewinding. + +**Simulated checkpoint density** (2000 commits, 20 seeds) +- Under `% 2`: 1.4% of positions safe interleaved, 0.8% clustered, 0.6% two-device interleave, 35% + mostly creates. +- Under a one-in-8 rule that respects holes: mean rollback gap 1 to 3 commits, max 11 clustered and + 54 interleaved; 34% to 38% of snapshots retained where edits cluster, 97% under uniform-random + touching. Humans do not edit uniformly at random. + +**Legacy gap detection queries** (2000 commits, 2050 snapshots) both translate to plain SQLite with +no raw SQL, including `References.Contains` becoming a correlated `json_each` subquery. The ordinary +anti-join runs in about 3 ms worst case, the cascade-aware one about 86 ms, with no false positives +over 50 intact cascade groups. + +## Separate performance findings, worth their own issues + +- **`CurrentSnapshots()` in the delete path.** Every delete of a referenced entity runs + `MakeCurrentSnapshotsQuery`, a raw-SQL window function over the whole Snapshots table, once per + cascade level. About 10 ms per 1000 snapshot rows per call, measured 134 ms for a single delete at + 8k snapshots. Superlinear in project size and live in production today. Wants an indexed + current-snapshot-per-entity lookup. This is probably worth more than the bug fix. +- **No index on `ChangeEntities.EntityId`.** Every "which commits touch this entity" question + currently scans. + +## What to test + +The highest-value test is the invariant itself, as a property over randomized histories: for every +flagged checkpoint b and every entity E, E's newest snapshot at or before b is at E's most recent +touch at or before b. That checks the design rather than a scenario, and it is what would have caught +the interval bug immediately. + +Then: the interleaved history from Corrections (it fails on the first attempt and on `main`); a +density assertion so nobody silently reintroduces a policy that produces one checkpoint per batch; +and a repro for the point-in-time path, which has **no test today** even though we now know it is +wrong. A history where a neighbour's snapshot is pruned and a change reads it via `IsObjectDeleted`, +then `GetAtCommit` for the subject, is enough. + +Note that `LateCommitTests`'s cascade test asserts `AssertSnapshotWasPruned` as a precondition. Any +design that stops pruning that particular snapshot makes the test fail on its precondition rather +than its assertion, which is correct behaviour but means the test needs re-pointing. + +## Rejected + +- **Fixpoint widening** (roll back to the affected entities' oldest surviving snapshot, follow the + reference closure, repeat): unbounded, drags in unrelated entities, and the UNIQUE violations it hit + on `Snapshots(CommitId, EntityId)` were a symptom of widening the replay without widening the + delete to match. +- **One checkpoint per batch** (the batch's last commit): correct but the rollback reaches the start + of the batch, and for a fresh clone that is the whole history. +- **Never prune**: correct and the smallest diff, but it is the most expensive point on the density + dial, and at 50k entities x 11 touches that is 110 MB or more on every cloned or regenerated + database. +- **Choosing checkpoints by a function of the commit id** (`Id.ToByteArray()[0] % K == 0`): elegant, + needs no column, and the invariant is a two-line proof, but it freezes the policy (only ever + coarsenable, since `% 16` boundaries are a subset of `% 8` ones) and the predicate is not + SQL-translatable, so finding the nearest one is an unindexed in-memory walk. Explicit flags give + the same guarantee with a free choice of policy and an indexed lookup. +- **Sparse or partial replay** of only the affected entities: blocked by the read sites above. Its + bail-out condition fires on nearly every real window. +- **Forward merge without rewind**: FW Lite's changes are op-based with arbitrary reads; last-write- + wins would need per-field timestamps in every entity and a rewrite of every change type. +- **Lazy repair**: symptom 2 fails during the narrow replay itself, and wrong state would be visible + to the user in between. +- **Dropping the projected-table FKs**: turns a hard wedge into silent divergence. The FK is the + canary. diff --git a/src/SIL.Harmony.Tests/DataModelSimpleChanges.WriteMultipleCommits.verified.txt b/src/SIL.Harmony.Tests/DataModelSimpleChanges.WriteMultipleCommits.verified.txt index a378761..bd705c5 100644 --- a/src/SIL.Harmony.Tests/DataModelSimpleChanges.WriteMultipleCommits.verified.txt +++ b/src/SIL.Harmony.Tests/DataModelSimpleChanges.WriteMultipleCommits.verified.txt @@ -19,6 +19,7 @@ ], Hash: Hash_1, ParentHash: Hash_Empty, + IsSnapshotCheckpoint: true, ChangeEntities: [ { $type: ChangeEntity, @@ -68,6 +69,7 @@ ], Hash: Hash_2, ParentHash: Hash_1, + IsSnapshotCheckpoint: true, ChangeEntities: [ { $type: ChangeEntity, diff --git a/src/SIL.Harmony.Tests/DataModelSimpleChanges.Writing2ChangesAtOnceWithMergedHistory.verified.txt b/src/SIL.Harmony.Tests/DataModelSimpleChanges.Writing2ChangesAtOnceWithMergedHistory.verified.txt index acd4e0b..a183cbc 100644 --- a/src/SIL.Harmony.Tests/DataModelSimpleChanges.Writing2ChangesAtOnceWithMergedHistory.verified.txt +++ b/src/SIL.Harmony.Tests/DataModelSimpleChanges.Writing2ChangesAtOnceWithMergedHistory.verified.txt @@ -19,6 +19,7 @@ ], Hash: Hash_1, ParentHash: Hash_Empty, + IsSnapshotCheckpoint: true, ChangeEntities: [ { $type: ChangeEntity, @@ -50,29 +51,13 @@ }, { $type: Commit, - Snapshots: [ - { - $type: ObjectSnapshot, - Id: Guid_5, - TypeName: Word, - Entity: { - $type: Word, - Text: first, - Note: a word note, - Id: Guid_2 - }, - EntityId: Guid_2, - EntityIsDeleted: false, - CommitId: Guid_6, - IsRoot: false - } - ], Hash: Hash_2, ParentHash: Hash_1, + IsSnapshotCheckpoint: false, ChangeEntities: [ { $type: ChangeEntity, - CommitId: Guid_6, + CommitId: Guid_5, EntityId: Guid_2, Change: { $type: SetWordNoteChange, @@ -85,9 +70,9 @@ CompareKey: { $type: ValueTuple, - CommitId: Guid_7, + CommitId: Guid_6, EntityId: Guid_2, Change: { $type: SetWordTextChange, @@ -118,9 +104,9 @@ CompareKey: { $type: ValueTuple, - CommitId: Guid_9, + CommitId: Guid_8, EntityId: Guid_2, Change: { $type: SetWordTextChange, @@ -168,9 +155,9 @@ CompareKey: { $type: ValueTuple, @@ -68,6 +69,7 @@ ], Hash: Hash_2, ParentHash: Hash_1, + IsSnapshotCheckpoint: true, ChangeEntities: [ { $type: ChangeEntity, diff --git a/src/SIL.Harmony.Tests/DataModelSimpleChanges.WritingAChangeMakesASnapshot.verified.txt b/src/SIL.Harmony.Tests/DataModelSimpleChanges.WritingAChangeMakesASnapshot.verified.txt index 5db6e99..cef91dd 100644 --- a/src/SIL.Harmony.Tests/DataModelSimpleChanges.WritingAChangeMakesASnapshot.verified.txt +++ b/src/SIL.Harmony.Tests/DataModelSimpleChanges.WritingAChangeMakesASnapshot.verified.txt @@ -19,6 +19,7 @@ ], Hash: Hash_1, ParentHash: Hash_Empty, + IsSnapshotCheckpoint: true, ChangeEntities: [ { $type: ChangeEntity, diff --git a/src/SIL.Harmony.Tests/DataModelSimpleChanges.WritingACommitWithMultipleChangesWorks.verified.txt b/src/SIL.Harmony.Tests/DataModelSimpleChanges.WritingACommitWithMultipleChangesWorks.verified.txt index d69bdc5..2540901 100644 --- a/src/SIL.Harmony.Tests/DataModelSimpleChanges.WritingACommitWithMultipleChangesWorks.verified.txt +++ b/src/SIL.Harmony.Tests/DataModelSimpleChanges.WritingACommitWithMultipleChangesWorks.verified.txt @@ -33,6 +33,7 @@ ], Hash: Hash_1, ParentHash: Hash_Empty, + IsSnapshotCheckpoint: true, ChangeEntities: [ { $type: ChangeEntity, diff --git a/src/SIL.Harmony.Tests/DataModelTestBase.cs b/src/SIL.Harmony.Tests/DataModelTestBase.cs index 8dcf44f..a277aff 100644 --- a/src/SIL.Harmony.Tests/DataModelTestBase.cs +++ b/src/SIL.Harmony.Tests/DataModelTestBase.cs @@ -89,7 +89,7 @@ public async ValueTask WriteChangeBefore(Commit before, IChange change, return await WriteChange(_localClientId, before.DateTime.AddHours(-1), change, add); } - protected async ValueTask WriteChange(Guid clientId, + public async ValueTask WriteChange(Guid clientId, DateTimeOffset dateTime, IChange change, bool add = true) @@ -97,7 +97,7 @@ protected async ValueTask WriteChange(Guid clientId, return await WriteChange(clientId, dateTime, [change], add); } - protected async ValueTask WriteChange(Guid clientId, + public async ValueTask WriteChange(Guid clientId, DateTimeOffset dateTime, IEnumerable changes, bool add = true) @@ -122,7 +122,7 @@ protected async ValueTask WriteChange(Guid clientId, return await DataModel.AddChanges(clientId, changes); } - protected async Task AddCommitsViaSync(IEnumerable commits) + public async Task AddCommitsViaSync(IEnumerable commits) { await ((ISyncable)DataModel).AddRangeFromSync(commits); } diff --git a/src/SIL.Harmony.Tests/DbContextTests.VerifyModel.verified.txt b/src/SIL.Harmony.Tests/DbContextTests.VerifyModel.verified.txt index 3a780f3..7c27acc 100644 --- a/src/SIL.Harmony.Tests/DbContextTests.VerifyModel.verified.txt +++ b/src/SIL.Harmony.Tests/DbContextTests.VerifyModel.verified.txt @@ -4,6 +4,7 @@ Id (Guid) Required PK AfterSave:Throw ValueGenerated.OnAdd ClientId (Guid) Required Hash (string) Required + IsSnapshotCheckpoint (bool) Required Metadata (CommitMetadata) Required Annotations: Relational:ColumnType: jsonb @@ -24,7 +25,7 @@ Keys: Id PK Annotations: - CustomIndex:CompositeIndexes: [{"paths":["HybridDateTime.DateTime","HybridDateTime.Counter","Id"],"unique":false,"name":"IX_Commits_DateTime_Counter_Id"}] + CustomIndex:CompositeIndexes: [{"paths":["HybridDateTime.DateTime","HybridDateTime.Counter","Id"],"unique":false,"name":"IX_Commits_DateTime_Counter_Id"},{"paths":["IsSnapshotCheckpoint","HybridDateTime.DateTime","HybridDateTime.Counter","Id"],"unique":false,"name":"IX_Commits_IsSnapshotCheckpoint_DateTime_Counter_Id"}] Relational:FunctionName: Relational:Schema: Relational:SqlQuery: diff --git a/src/SIL.Harmony.Tests/DbContextTests.cs b/src/SIL.Harmony.Tests/DbContextTests.cs index 8733439..cc3df5b 100644 --- a/src/SIL.Harmony.Tests/DbContextTests.cs +++ b/src/SIL.Harmony.Tests/DbContextTests.cs @@ -53,6 +53,7 @@ await DbContext.Set().ToLinqToDBTable().AsValueInsertable() .Value(c => c.Metadata, new CommitMetadata()) .Value(c => c.Hash, "") .Value(c => c.ParentHash, "") + .Value(c => c.IsSnapshotCheckpoint, false) .InsertAsync(TestContext.Current.CancellationToken); var actualCommit = await DbContext.Commits.SingleOrDefaultAsyncEF(c => c.Id == commitId, TestContext.Current.CancellationToken); actualCommit!.HybridDateTime.DateTime.Should().Be(expectedDateTime, "EF"); diff --git a/src/SIL.Harmony.Tests/LateCommitTests.cs b/src/SIL.Harmony.Tests/LateCommitTests.cs new file mode 100644 index 0000000..64c00fb --- /dev/null +++ b/src/SIL.Harmony.Tests/LateCommitTests.cs @@ -0,0 +1,67 @@ +using Microsoft.EntityFrameworkCore; +using SIL.Harmony.Sample.Changes; +using SIL.Harmony.Sample.Models; + +namespace SIL.Harmony.Tests; + +/// +/// The smallest shape of the late commit bug, on a single client. +/// Adding several commits at once only keeps a snapshot for every other one, so an entity ends up +/// with a commit that has no snapshot. A commit dated between that commit and the next one that +/// does have a snapshot then makes the replay start after its own parent (the commit with no +/// snapshot), while snapshots are only deleted from the late commit onwards. The entity resumes +/// from the older snapshot it still has, and everything between that snapshot and the late +/// commit's parent is applied by nobody. +/// +public class LateCommitTests : DataModelTestBase +{ + private async Task AssertSnapshotWasPruned(Commit commit, Guid entityId) + { + var snapshots = await DbContext.Snapshots.AsNoTracking() + .CountAsync(s => s.CommitId == commit.Id && s.EntityId == entityId); + snapshots.Should().Be(0, "otherwise there's no gap and the test proves nothing"); + } + + [Fact] + public async Task ALateCommitKeepsAnEditWhoseSnapshotWasPruned() + { + var wordId = Guid.NewGuid(); + // add: false builds the commit without applying it, so all three land in one batch below + var create = await WriteNextChange(SetWord(wordId, "word"), add: false); + var setNote = await WriteNextChange(new SetWordNoteChange(wordId, "a note"), add: false); + var rename = await WriteNextChange(new SetWordTextChange(wordId, "renamed word"), add: false); + await AddCommitsViaSync([create, setNote, rename]); + // the batch keeps the word's snapshots at create and rename, but not the one in the middle + await AssertSnapshotWasPruned(setNote, wordId); + + await WriteChangeAfter(setNote, SetWord(Guid.NewGuid(), "written late")); + + // the rename snapshot is gone and the word resumed from create, so nothing re-applied the note + var word = await DataModel.GetLatest(wordId); + word!.Text.Should().Be("renamed word"); + word.Note.Should().Be("a note"); + } + + [Fact] + public async Task ALateCommitKeepsACascadeDeleteWhoseSnapshotWasPruned() + { + var wordId = Guid.NewGuid(); + var definitionId = Guid.NewGuid(); + var create = await WriteNextChange(SetWord(wordId, "word"), add: false); + // only here to shift which snapshots the batch keeps, so the definition loses the one below + var unrelated = await WriteNextChange(SetWord(Guid.NewGuid(), "another word"), add: false); + var newDefinition = await WriteNextChange(NewDefinition(wordId, "a definition", "noun", definitionId: definitionId), add: false); + // deleting the word deletes its definition too, but only as a snapshot: no commit records it + var delete = await WriteNextChange(DeleteWord(wordId), add: false); + var editDefinition = await WriteNextChange(new SetDefinitionPartOfSpeechChange(definitionId, "verb"), add: false); + await AddCommitsViaSync([create, unrelated, newDefinition, delete, editDefinition]); + await AssertSnapshotWasPruned(delete, definitionId); + + // the definition resumes from its creation snapshot, and replaying the delete commit does not + // cascade again, so the edit lands on a live definition whose word is still deleted. + // Projecting that row back in breaks the foreign key to the word. + await WriteChangeAfter(delete, SetWord(Guid.NewGuid(), "written late")); + + (await DataModel.GetLatest(definitionId))!.DeletedAt.Should().NotBeNull(); + } +} diff --git a/src/SIL.Harmony.Tests/ModelSnapshotTests.cs b/src/SIL.Harmony.Tests/ModelSnapshotTests.cs index 951af8a..ae21c49 100644 --- a/src/SIL.Harmony.Tests/ModelSnapshotTests.cs +++ b/src/SIL.Harmony.Tests/ModelSnapshotTests.cs @@ -108,9 +108,15 @@ public async Task CanGetWordForASpecificTime() thirdWord.Text.Should().Be("third"); } - private Task ClearNonRootSnapshots() + /// + /// leaves each entity nothing but its root snapshot, so reading state at a commit has to replay history to get there + /// + private async Task ClearNonRootSnapshots() { - return DbContext.Snapshots.Where(s => !s.IsRoot).ExecuteDeleteAsync(TestContext.Current.CancellationToken); + //the flags go first: a checkpoint claims that the snapshots at or before it hold the state a replay resumes from + await DbContext.Commits.ExecuteUpdateAsync(s => s.SetProperty(c => c.IsSnapshotCheckpoint, false), TestContext.Current.CancellationToken); + await DbContext.Snapshots.Where(s => !s.IsRoot).ExecuteDeleteAsync(TestContext.Current.CancellationToken); + DbContext.ChangeTracker.Clear(); } [Theory] @@ -125,15 +131,16 @@ public async Task CanGetSnapshotFromEarlier(int changeCount) var addNew = new List(changeCount); for (var i = 0; i < changeCount; i++) { - // todo: these commits all have an odd index, so no intermediate snapshots will be persisted i.e. the snapshot count checking is somewhat deceptive changes.Add(await WriteNextChange(SetWord(entityId, $"change {i}"), false).AsTask()); addNew.Add(await WriteNextChange(SetWord(Guid.NewGuid(), $"add {i}"), false).AsTask()); } //adding all via sync means there's sparse snapshots await AddCommitsViaSync(changes.Concat(addNew)); - //there will only be a snapshot for every other commit, but there's change count * 2 commits, plus a first and last change - DbContext.Snapshots.Should().HaveCount(2 + changeCount); + var commitCount = changeCount * 2; + var checkpointCount = Enumerable.Range(1, commitCount).Count(i => SnapshotCheckpointPolicy.Default.IsCheckpoint(i, commitCount)); + //the root from the first change, a root per newly added word, and one snapshot of the edited word per checkpoint + DbContext.Snapshots.Should().HaveCount(1 + changeCount + checkpointCount); for (int i = 0; i < changeCount; i++) { @@ -156,7 +163,7 @@ await AddCommitsViaSync(Enumerable.Range(0, changeCount) var latestSnapshot = await DataModel.GetLatestSnapshotByObjectId(entityId); //delete snapshots so when we get at then we need to re-apply - await DbContext.Snapshots.Where(s => !s.IsRoot).ExecuteDeleteAsync(TestContext.Current.CancellationToken); + await ClearNonRootSnapshots(); var computedModelSnapshots = await DataModel.GetSnapshotsAtCommit(latestSnapshot.Commit); diff --git a/src/SIL.Harmony.Tests/RepositoryTests.cs b/src/SIL.Harmony.Tests/RepositoryTests.cs index 3ad69d3..098d116 100644 --- a/src/SIL.Harmony.Tests/RepositoryTests.cs +++ b/src/SIL.Harmony.Tests/RepositoryTests.cs @@ -245,16 +245,38 @@ await _repository.AddSnapshots([ } [Fact] - public async Task DeleteStaleSnapshots_WithNoSnapshots_DoesNothing() + public async Task DeleteSnapshotsAfter_WithNoSnapshots_DoesNothing() { //the empty-repository branch: nothing to delete, must not throw - await _repository.DeleteStaleSnapshots(Commit(Guid.NewGuid(), Time(1, 0))); + await _repository.DeleteSnapshotsAfter(Commit(Guid.NewGuid(), Time(1, 0))); _crdtDbContext.Snapshots.Should().BeEmpty(); } [Fact] - public async Task DeleteStaleSnapshots_KeepsSnapshotsOlderThanTheCommit() + public async Task DeleteSnapshotsAfter_Null_DeletesEverySnapshot() + { + await _repository.AddSnapshots([ + Snapshot(Guid.NewGuid(), Guid.NewGuid(), Time(1, 0)), + Snapshot(Guid.NewGuid(), Guid.NewGuid(), Time(2, 0)), + ]); + + await _repository.DeleteSnapshotsAfter(null); + + _crdtDbContext.Snapshots.Should().BeEmpty(); + } + + [Fact] + public async Task HasSnapshotsAfter_ComparesTheWholeCommitOrder() + { + await _repository.AddSnapshots([Snapshot(Guid.NewGuid(), Guid.NewGuid(), Time(1, 1))]); + + (await _repository.HasSnapshotsAfter(Commit(Guid.NewGuid(), Time(1, 0)))).Should().BeTrue(); + (await _repository.HasSnapshotsAfter(Commit(Guid.NewGuid(), Time(1, 2)))).Should().BeFalse(); + } + + [Fact] + public async Task DeleteSnapshotsAfter_KeepsSnapshotsOlderThanTheCommit() { await _repository.AddSnapshots([ Snapshot(Guid.NewGuid(), Guid.NewGuid(), Time(1, 0)), @@ -262,39 +284,39 @@ await _repository.AddSnapshots([ ]); //the new commit is newer than every existing snapshot, so none are stale - await _repository.DeleteStaleSnapshots(Commit(Guid.NewGuid(), Time(3, 0))); + await _repository.DeleteSnapshotsAfter(Commit(Guid.NewGuid(), Time(3, 0))); _crdtDbContext.Snapshots.Should().HaveCount(2); } [Fact] - public async Task DeleteStaleSnapshots_DeletesSnapshotsAfterCommitByTime() + public async Task DeleteSnapshotsAfter_DeletesSnapshotsAfterCommitByTime() { await _repository.AddSnapshots([ Snapshot(Guid.NewGuid(), Guid.NewGuid(), Time(1, 0)), Snapshot(Guid.NewGuid(), Guid.NewGuid(), Time(3, 0)), ]); - await _repository.DeleteStaleSnapshots(Commit(Guid.NewGuid(), Time(2, 0))); + await _repository.DeleteSnapshotsAfter(Commit(Guid.NewGuid(), Time(2, 0))); _crdtDbContext.Snapshots.Include(s => s.Commit).Should().ContainSingle() .Which.Commit.HybridDateTime.DateTime.Hour.Should().Be(1); } [Fact] - public async Task DeleteStaleSnapshots_DeletesSnapshotsAfterCommitByCount() + public async Task DeleteSnapshotsAfter_DeletesSnapshotsAfterCommitByCount() { await _repository.AddSnapshots([ Snapshot(Guid.NewGuid(), Guid.NewGuid(), Time(1, 0)), Snapshot(Guid.NewGuid(), Guid.NewGuid(), Time(1, 2)), ]); - await _repository.DeleteStaleSnapshots(Commit(Guid.NewGuid(), Time(1, 1))); + await _repository.DeleteSnapshotsAfter(Commit(Guid.NewGuid(), Time(1, 1))); _crdtDbContext.Snapshots.Include(s => s.Commit).Should().ContainSingle() .Which.Commit.HybridDateTime.Counter.Should().Be(0); } [Fact] - public async Task DeleteStaleSnapshots_DeletesSnapshotsAfterCommitByCommitId() + public async Task DeleteSnapshotsAfter_DeletesSnapshotsAfterCommitByCommitId() { var time = Time(1, 1); var entityId = Guid.NewGuid(); @@ -303,7 +325,7 @@ await _repository.AddSnapshots([ Snapshot(entityId, ids[0], time), Snapshot(entityId, ids[2], time), ]); - await _repository.DeleteStaleSnapshots(Commit(ids[1], time)); + await _repository.DeleteSnapshotsAfter(Commit(ids[1], time)); _crdtDbContext.Snapshots.Should().ContainSingle() .Which.CommitId.Should().Be(ids[0]); diff --git a/src/SIL.Harmony.Tests/SnapshotCheckpointPolicyTests.cs b/src/SIL.Harmony.Tests/SnapshotCheckpointPolicyTests.cs new file mode 100644 index 0000000..5623b3e --- /dev/null +++ b/src/SIL.Harmony.Tests/SnapshotCheckpointPolicyTests.cs @@ -0,0 +1,58 @@ +namespace SIL.Harmony.Tests; + +public class SnapshotCheckpointPolicyTests +{ + private static readonly SnapshotCheckpointPolicy Policy = new(4); + + [Theory] + [InlineData(1, 10, false)] + [InlineData(4, 10, true)] + [InlineData(8, 10, true)] + [InlineData(9, 10, false)] + [InlineData(10, 10, true)] + public void PicksEveryNthCommitOfTheBatchAndItsLast(int commitIndex, int commitCount, bool isCheckpoint) + { + Policy.IsCheckpoint(commitIndex, commitCount).Should().Be(isCheckpoint); + } + + [Theory] + [InlineData(1, 3, false)] + [InlineData(1, 5, true)] + //a checkpoint at the snapshot's own commit counts, that's the position a replay would seed the entity from + [InlineData(4, 5, true)] + [InlineData(5, 8, false)] + [InlineData(5, 9, true)] + public void KeepsASnapshotOnlyWhenACheckpointFallsInTheGapItWouldLeave(int commitIndex, int nextCommitIndex, bool mustKeep) + { + Policy.MustKeepSnapshot(commitIndex, nextCommitIndex).Should().Be(mustKeep); + } + + [Fact] + public void KeepsExactlyTheSnapshotsTheCheckpointsItPicksNeed() + { + const int commitCount = 40; + var policy = new SnapshotCheckpointPolicy(7); + var checkpoints = Enumerable.Range(1, commitCount).Where(i => policy.IsCheckpoint(i, commitCount)).ToHashSet(); + + for (var commitIndex = 1; commitIndex <= commitCount; commitIndex++) + { + for (var nextCommitIndex = commitIndex + 1; nextCommitIndex <= commitCount; nextCommitIndex++) + { + var gapSpansACheckpoint = Enumerable.Range(commitIndex, nextCommitIndex - commitIndex).Any(checkpoints.Contains); + policy.MustKeepSnapshot(commitIndex, nextCommitIndex).Should().Be(gapSpansACheckpoint, + $"the gap [{commitIndex}, {nextCommitIndex}) of a {commitCount} commit batch"); + } + } + } + + [Fact] + public void KeepsEverySnapshotAtTheNeverPruneEndOfTheDensityDial() + { + var everyCommit = new SnapshotCheckpointPolicy(1); + foreach (var commitIndex in Enumerable.Range(1, 5)) + { + everyCommit.IsCheckpoint(commitIndex, 5).Should().BeTrue(); + everyCommit.MustKeepSnapshot(commitIndex, commitIndex + 1).Should().BeTrue(); + } + } +} diff --git a/src/SIL.Harmony.Tests/SnapshotCheckpointTests.cs b/src/SIL.Harmony.Tests/SnapshotCheckpointTests.cs new file mode 100644 index 0000000..936b940 --- /dev/null +++ b/src/SIL.Harmony.Tests/SnapshotCheckpointTests.cs @@ -0,0 +1,272 @@ +using System.Text.Json; +using Microsoft.EntityFrameworkCore; +using SIL.Harmony.Changes; +using SIL.Harmony.Sample.Changes; +using SIL.Harmony.Sample.Models; + +namespace SIL.Harmony.Tests; + +/// +/// Checkpoints are the commits a replay may resume from: every entity's newest snapshot at or before one holds that +/// entity's state there. These tests check that property rather than any single scenario, because the ways it can break +/// all look local and harmless (see docs/snapshot-checkpoints.md). +/// +public class SnapshotCheckpointTests : DataModelTestBase +{ + private sealed record PlannedChange(DateTimeOffset Date, IChange Change); + + /// + /// A history of creates, edits, references and cascading deletes. Randomized so the tests cover the shapes of gap + /// that a hand written history keeps missing, seeded so a failure is reproducible. + /// + private static PlannedChange[] PlanHistory(int commitCount, int seed) + { + var random = new Random(seed); + var date = new DateTimeOffset(2001, 1, 1, 0, 0, 0, TimeSpan.Zero); + List words = []; + List definitions = []; + var plan = new List(); + while (plan.Count < commitCount) + { + date = date.AddDays(1); + plan.Add(new PlannedChange(date, NextChange())); + } + + return [.. plan]; + + IChange NextChange() + { + if (words.Count == 0 || random.Next(4) == 0) return NewWord(); + var wordId = words[random.Next(words.Count)]; + return random.Next(6) switch + { + 0 => new SetWordNoteChange(wordId, $"note {plan.Count}"), + 1 => Antonym(wordId), + 2 => NewDefinitionFor(wordId), + 3 when definitions.Count > 0 => new SetDefinitionPartOfSpeechChange(definitions[random.Next(definitions.Count)], $"part of speech {plan.Count}"), + 4 => new DeleteChange(wordId), + _ => new SetWordTextChange(wordId, $"text {plan.Count}"), + }; + } + + IChange NewWord() + { + var wordId = Guid.NewGuid(); + words.Add(wordId); + return new SetWordTextChange(wordId, $"word {words.Count}"); + } + + IChange Antonym(Guid wordId) + { + var others = words.Where(w => w != wordId).ToArray(); + if (others is []) return new SetWordTextChange(wordId, $"text {plan.Count}"); + //setObject false keeps the snapshots comparable: the whole antonym would otherwise be nested in the word + return new SetAntonymReferenceChange(wordId, others[random.Next(others.Length)], setObject: false); + } + + IChange NewDefinitionFor(Guid wordId) + { + var definitionId = Guid.NewGuid(); + definitions.Add(definitionId); + return new NewDefinitionChange(definitionId) + { + WordId = wordId, + Text = $"definition {definitions.Count}", + PartOfSpeech = "noun", + Order = definitions.Count + }; + } + } + + private async Task AddInOneBatch(DataModelTestBase model, IEnumerable plan) + { + var commits = new List(); + foreach (var planned in plan) + { + commits.Add(await model.WriteChange(_localClientId, planned.Date, planned.Change, add: false)); + } + + await model.AddCommitsViaSync(commits); + return [.. commits]; + } + + private static async Task> CurrentState(DataModelTestBase model) + { + var snapshots = await model.DataModel.GetLatestSnapshots().ToArrayAsync(TestContext.Current.CancellationToken); + return snapshots.ToDictionary(s => s.EntityId, s => Describe(s.Entity.DbObject)); + } + + private async Task> StateFromSnapshotsAtOrBefore(Commit commit) + { + var snapshots = await DbContext.Snapshots.AsNoTracking() + .Include(s => s.Commit) + .ToArrayAsync(TestContext.Current.CancellationToken); + return snapshots + .Where(s => s.Commit.CompareKey.CompareTo(commit.CompareKey) <= 0) + .GroupBy(s => s.EntityId) + .ToDictionary(g => g.Key, g => Describe(g.MaxBy(s => s.Commit.CompareKey)!.Entity.DbObject)); + } + + //comparing json rather than the objects keeps FluentAssertions from comparing them as bare objects, which finds no members at all + private static string Describe(object entity) => JsonSerializer.Serialize(entity, entity.GetType()); + + private async Task CheckpointIds() + { + return await DbContext.Commits.AsNoTracking() + .Where(c => c.IsSnapshotCheckpoint) + .DefaultOrder() + .Select(c => c.Id) + .ToArrayAsync(TestContext.Current.CancellationToken); + } + + [Theory] + [InlineData(1)] + [InlineData(2)] + [InlineData(3)] + public async Task EveryCheckpointHoldsTheStateAReplayWouldResumeFrom(int seed) + { + var plan = PlanHistory(24, seed); + var commits = await AddInOneBatch(this, plan); + var checkpoints = await CheckpointIds(); + checkpoints.Should().HaveCountGreaterThan(1, "otherwise this only checks the end of the batch"); + + foreach (var checkpointId in checkpoints) + { + var commitCount = Array.FindIndex(commits, c => c.Id == checkpointId) + 1; + //a history added from empty in one batch is complete at its last commit by construction, so it can say what the checkpoint should hold + await using var fromScratch = new DataModelTestBase(); + await AddInOneBatch(fromScratch, plan.Take(commitCount)); + + var atCheckpoint = await StateFromSnapshotsAtOrBefore(commits[commitCount - 1]); + atCheckpoint.Should().BeEquivalentTo(await CurrentState(fromScratch), + $"snapshots have to be complete at the checkpoint {commitCount} commits in"); + } + } + + [Theory] + [InlineData(1)] + [InlineData(2)] + public async Task ALateCommitAtAnyPositionKeepsEveryEntitysState(int seed) + { + var plan = PlanHistory(20, seed); + var commits = await AddInOneBatch(this, plan); + var expected = await CurrentState(this); + + for (var position = 0; position < commits.Length; position++) + { + await using var fork = ForkDatabase(); + var late = await fork.WriteChangeAfter(commits[position], fork.SetWord(Guid.NewGuid(), "written late")); + + var state = await CurrentState(fork); + state.Remove(late.ChangeEntities[0].EntityId).Should().BeTrue(); + state.Should().BeEquivalentTo(expected, $"a commit landing after commit {position + 1} only adds a word"); + } + } + + [Fact] + public async Task ALateCommitInsideAGapKeepsTheEditThatGapSpans() + { + //the history that broke the first attempt at this: A is touched at 1, 3 and 5 and B at 2 and 4, so A's snapshot + //at 3 is dropped and the late commit lands at 4, a position that looks safe from B's snapshots alone + var a = Guid.NewGuid(); + var b = Guid.NewGuid(); + var commits = new[] + { + await WriteNextChange(SetWord(a, "a"), add: false), + await WriteNextChange(SetWord(b, "b"), add: false), + await WriteNextChange(new SetWordNoteChange(a, "a note"), add: false), + await WriteNextChange(SetWord(b, "b renamed"), add: false), + await WriteNextChange(SetWord(a, "a renamed"), add: false), + }; + await AddCommitsViaSync(commits); + + await WriteChangeAfter(commits[3], SetWord(Guid.NewGuid(), "written late")); + + var word = await DataModel.GetLatest(a); + word!.Text.Should().Be("a renamed"); + word.Note.Should().Be("a note"); + } + + [Fact] + public async Task ASyncedBatchGetsACheckpointEveryEighthCommitAndOneAtItsEnd() + { + var commits = await AddInOneBatch(this, PlanHistory(20, seed: 4)); + + //pinned on purpose: one checkpoint per batch is also correct, but it rolls a fresh clone back to the start of history + (await CheckpointIds()).Should().Equal(commits[7].Id, commits[15].Id, commits[19].Id); + } + + [Fact] + public async Task EveryLocallyAuthoredCommitIsACheckpoint() + { + var entityId = Guid.NewGuid(); + var first = await WriteNextChange(SetWord(entityId, "first")); + var second = await WriteNextChange(SetWord(entityId, "second")); + + (await CheckpointIds()).Should().Equal(first.Id, second.Id); + } + + [Fact] + public async Task ALateCommitResumesFromTheNewestCheckpointBeforeItRatherThanRebuildingEverything() + { + var commits = await AddInOneBatch(this, PlanHistory(20, seed: 5)); + var checkpoint = commits[7]; + var untouchedSnapshotIds = await DbContext.Snapshots.AsNoTracking() + .Include(s => s.Commit) + .ToArrayAsync(TestContext.Current.CancellationToken); + var keptSnapshotIds = untouchedSnapshotIds + .Where(s => s.Commit.CompareKey.CompareTo(checkpoint.CompareKey) <= 0) + .Select(s => s.Id) + .ToArray(); + keptSnapshotIds.Should().NotBeEmpty(); + + await WriteChangeAfter(commits[9], SetWord(Guid.NewGuid(), "written late")); + + var snapshotIds = await DbContext.Snapshots.AsNoTracking().Select(s => s.Id).ToArrayAsync(TestContext.Current.CancellationToken); + snapshotIds.Should().Contain(keptSnapshotIds); + (await CheckpointIds()).Should().Contain(checkpoint.Id, "a commit before the replay window keeps its flag"); + } + + [Fact] + public async Task ADatabaseWithNoCheckpointsRebuildsEverySnapshotOnTheFirstLateCommit() + { + var plan = PlanHistory(12, seed: 6); + var commits = await AddInOneBatch(this, plan); + var expected = await CurrentState(this); + //what a database written before checkpoints existed looks like + await DbContext.Commits.ExecuteUpdateAsync(s => s.SetProperty(c => c.IsSnapshotCheckpoint, false), TestContext.Current.CancellationToken); + DbContext.ChangeTracker.Clear(); + + var late = await WriteChangeAfter(commits[5], SetWord(Guid.NewGuid(), "written late")); + + (await CheckpointIds()).Should().NotBeEmpty("the repair is also the bootstrap"); + var state = await CurrentState(this); + state.Remove(late.ChangeEntities[0].EntityId).Should().BeTrue(); + state.Should().BeEquivalentTo(expected); + } + + [Fact] + public async Task ReadingStateAtAnOldCommitDoesNotSeeANeighbourItsChangeCouldNotHaveSeen() + { + //the neighbour is deleted at commit 2, its snapshot there is dropped at commit 6, and its next snapshot is the + //revival at 6, so its newest snapshot at commit 4 says it is still alive + var neighbourId = Guid.NewGuid(); + var wordId = Guid.NewGuid(); + var commits = new[] + { + await WriteNextChange(SetWord(neighbourId, "neighbour"), add: false), + await WriteNextChange(DeleteWord(neighbourId), add: false), + await WriteNextChange(SetWord(wordId, "word"), add: false), + await WriteNextChange(new SetAntonymReferenceChange(wordId, neighbourId, setObject: false), add: false), + await WriteNextChange(SetWord(wordId, "word renamed"), add: false), + await WriteNextChange(SetWord(neighbourId, "neighbour revived"), add: false), + }; + await AddCommitsViaSync(commits); + + var word = await DataModel.GetAtCommit(commits[3], wordId); + + //the antonym was deleted when the reference was written, so the change skipped it + word.AntonymId.Should().BeNull(); + word.Text.Should().Be("word"); + } +} diff --git a/src/SIL.Harmony.Tests/SnapshotTests.cs b/src/SIL.Harmony.Tests/SnapshotTests.cs index 617e213..590fa87 100644 --- a/src/SIL.Harmony.Tests/SnapshotTests.cs +++ b/src/SIL.Harmony.Tests/SnapshotTests.cs @@ -41,7 +41,8 @@ public async Task MultipleChangesPreservesSomeIntermediateSnapshots() { var entityId = Guid.NewGuid(); var commits = new List(); - for (var i = 0; i < 6; i++) + //the batch has to be longer than the checkpoint interval, a shorter one needs no snapshot but the root and the latest + for (var i = 0; i < 20; i++) { commits.Add(await WriteChange(_localClientId, new DateTimeOffset(2000, 1, 1, 0, 0, 0, TimeSpan.Zero).AddHours(i), diff --git a/src/SIL.Harmony/Commit.cs b/src/SIL.Harmony/Commit.cs index 53eebbc..2bd2fb6 100644 --- a/src/SIL.Harmony/Commit.cs +++ b/src/SIL.Harmony/Commit.cs @@ -40,4 +40,13 @@ internal Commit() : this(Guid.NewGuid()) [JsonIgnore] public string ParentHash { get; private set; } + + /// + /// Snapshots are complete as of this commit: every entity's newest snapshot at or before it holds that entity's state + /// there, so a replay can resume from here. It does not mean every entity has a snapshot dated at this commit. + /// A commit that arrives out of order rolls snapshots back to the newest checkpoint before it and replays from there. + /// Local bookkeeping: never synced, not part of the hash, and different devices legitimately have different sets of it. + /// + [JsonIgnore] + public bool IsSnapshotCheckpoint { get; internal set; } } diff --git a/src/SIL.Harmony/DataModel.cs b/src/SIL.Harmony/DataModel.cs index 01e5219..66ff3e3 100644 --- a/src/SIL.Harmony/DataModel.cs +++ b/src/SIL.Harmony/DataModel.cs @@ -191,7 +191,27 @@ private async Task UpdateSnapshots(CrdtRepository repo, SortedSet commit { if (commitsToApply.Count == 0) return; var oldestAddedCommit = commitsToApply.First(); - await repo.DeleteStaleSnapshots(oldestAddedCommit); + if (await repo.HasSnapshotsAfter(oldestAddedCommit)) + { + //rolling back to the new commit is not enough: an entity's newest surviving snapshot can predate edits whose + //snapshots were pruned, and nothing in the window would re-apply them. Resume from a checkpoint instead. + var checkpoint = await repo.FindNewestCheckpoint(oldestAddedCommit); + if (checkpoint is null) + { + //no checkpoint to resume from, so every snapshot has to be rebuilt. Replaying all of history against a + //populated table measured about 3x the cost per commit of dropping everything and regenerating. + await repo.DeleteSnapshotsAndProjectedTables(); + //the delete goes around the change tracker, so drop what it holds and read the commits back fresh + repo.ClearChangeTracker(); + commitsToApply = await repo.CurrentCommits().Include(c => c.ChangeEntities).ToSortedSetAsync(); + } + else + { + await repo.DeleteSnapshotsAfter(checkpoint); + commitsToApply = (await repo.GetCommitsAfter(checkpoint)).ToSortedSet(); + } + } + Dictionary snapshotLookup = []; if (commitsToApply.Count > 10) { @@ -300,18 +320,23 @@ public async Task GetBySnapshotId(Guid snapshotId) public async Task> GetSnapshotsAtCommit(Commit commit) { await using var repo = await _crdtRepositoryFactory.CreateRepository(); - var repository = repo.GetScopedRepository(commit); - var (snapshots, pendingCommits) = await repository.GetCurrentSnapshotsAndPendingCommits(); - - if (pendingCommits.Count != 0) - { - snapshots = await SnapshotWorker.ApplyCommitsToSnapshots(snapshots, - repository, - pendingCommits, - _crdtConfig.Value); - } + var (checkpointState, commitsToReplay) = await ResumeFromCheckpoint(commit, repo); + var snapshots = await checkpointState.GetCurrentSnapshots(); + if (commitsToReplay.Count == 0) return snapshots; + return await SnapshotWorker.ApplyCommitsToSnapshots(snapshots, checkpointState, commitsToReplay, _crdtConfig.Value); + } - return snapshots; + /// + /// The primitive every replay shares: the state a replay resumes from, which is the newest checkpoint at or before + /// with each entity seeded from its newest snapshot there, and the commits to replay onto it. + /// + private static async Task<(CrdtRepository checkpointState, SortedSet commitsToReplay)> ResumeFromCheckpoint( + Commit commit, + CrdtRepository repo) + { + var checkpoint = await repo.FindNewestCheckpoint(commit, inclusive: true); + var commitsToReplay = await repo.GetScopedRepository(commit).GetCommitsAfter(checkpoint); + return (repo.GetScopedRepository(checkpoint), commitsToReplay.ToSortedSet()); } public async Task GetAtTime(DateTimeOffset time, Guid entityId) @@ -368,26 +393,12 @@ private async Task GetAtCommit(Commit commit, Guid entityId, CrdtRepositor private async Task GetSnapshotAtCommit(Commit commit, Guid entityId, CrdtRepository repo) { - var repository = repo.GetScopedRepository(commit); - var snapshot = await repository.GetCurrentSnapshotByObjectId(entityId, false); - if (snapshot is null) return null; - var newCommits = await repository.CurrentCommits() - .Include(c => c.ChangeEntities) - .WhereAfter(snapshot.Commit) - .ToSortedSetAsync(); - if (newCommits.Count > 0) - { - var snapshots = await SnapshotWorker.ApplyCommitsToSnapshots( - new Dictionary([ - new KeyValuePair(snapshot.EntityId, snapshot) - ]), - repository, - newCommits, - _crdtConfig.Value); - snapshot = snapshots[snapshot.EntityId]; - } - - return snapshot; + //replaying the whole range rather than only the commits touching this entity is deliberate: changes read each + //other's entities, so a neighbour left at its checkpoint state would feed stale values into this entity's changes. + var (checkpointState, commitsToReplay) = await ResumeFromCheckpoint(commit, repo); + var snapshots = await SnapshotWorker.ApplyCommitsToSnapshots([], checkpointState, commitsToReplay, _crdtConfig.Value); + //an entity untouched since the checkpoint isn't part of the replay, so its snapshot there is already its state here + return snapshots.GetValueOrDefault(entityId) ?? await checkpointState.GetCurrentSnapshotByObjectId(entityId); } public async Task GetSyncState() diff --git a/src/SIL.Harmony/Db/CrdtRepository.cs b/src/SIL.Harmony/Db/CrdtRepository.cs index d7a8a10..82c77cd 100644 --- a/src/SIL.Harmony/Db/CrdtRepository.cs +++ b/src/SIL.Harmony/Db/CrdtRepository.cs @@ -54,15 +54,23 @@ internal class CrdtRepository : IDisposable, IAsyncDisposable private readonly ILogger _logger; public CrdtRepository(ICrdtDbContext dbContext, IOptions crdtConfig, + ILogger logger) : this(dbContext, crdtConfig, logger, scoped: false, null) + { + } + + private CrdtRepository(ICrdtDbContext dbContext, IOptions crdtConfig, ILogger logger, - Commit? ignoreChangesAfter = null) + bool scoped, + Commit? ignoreChangesAfter) { _crdtConfig = crdtConfig; - _dbContext = ignoreChangesAfter is not null ? new ScopedDbContext(dbContext, ignoreChangesAfter) : dbContext; + _dbContext = scoped ? new ScopedDbContext(dbContext, ignoreChangesAfter) : dbContext; _logger = logger; //we can't use the scoped db context is it prevents access to the DbSet for the Snapshots, //but since we're using a custom query, we can use it directly and apply the scoped filters manually - _currentSnapshotsQueryable = MakeCurrentSnapshotsQuery(dbContext, ignoreChangesAfter); + _currentSnapshotsQueryable = scoped && ignoreChangesAfter is null + ? dbContext.Set().Where(_ => false).AsNoTracking() + : MakeCurrentSnapshotsQuery(dbContext, ignoreChangesAfter); _lock = Locks.GetOrAdd(DatabaseIdentifier, _ => new AsyncLock()); } @@ -127,15 +135,40 @@ public async Task HasCommit(Guid commitId) return (oldestChange, newCommits); } - public async Task DeleteStaleSnapshots(Commit oldestChange) + public async Task HasSnapshotsAfter(Commit commit) + { + return await Snapshots.WhereAfter(commit).AnyAsync(); + } + + /// null deletes every snapshot + public async Task DeleteSnapshotsAfter(Commit? commit) + { + await (commit is null ? Snapshots : Snapshots.WhereAfter(commit)).ExecuteDeleteAsync(); + } + + /// + /// The newest commit a replay may resume from, or null when there is none and all of history has to be replayed. + /// + public async Task FindNewestCheckpoint(Commit? before = null, bool inclusive = false) { - //use the oldest commit added to clear any snapshots that are based on a now incomplete history - //this is a performance optimization to avoid deleting snapshots where there are none to delete - var mostRecentCommit = await Snapshots.MaxAsync(s => (DateTimeOffset?)s.Commit.HybridDateTime.DateTime); - if (mostRecentCommit < oldestChange.HybridDateTime.DateTime) return; - await Snapshots - .WhereAfter(oldestChange) - .ExecuteDeleteAsync(); + var checkpoints = Commits.Where(c => c.IsSnapshotCheckpoint); + if (before is not null) checkpoints = checkpoints.WhereBefore(before, inclusive); + return await checkpoints.DefaultOrderDescending().FirstOrDefaultAsync(); + } + + /// + /// Records which of the commits about to be replayed are checkpoints. Has to run before the replay, which keeps + /// whatever snapshots this choice needs, and only ever covers commits being replayed: see . + /// + public async Task SetCheckpoints(SortedSet commitsToReplay, SnapshotCheckpointPolicy policy) + { + var commitIndex = 0; + foreach (var commit in commitsToReplay) + { + commit.IsSnapshotCheckpoint = policy.IsCheckpoint(++commitIndex, commitsToReplay.Count); + } + + await _dbContext.SaveChangesAsync(); } public async Task DeleteSnapshotsAndProjectedTables() @@ -210,18 +243,9 @@ public IAsyncEnumerable CurrenSimpleSnapshots(bool includeDelete return snapshots; } - public async Task<(Dictionary currentSnapshots, SortedSet pendingCommits)> GetCurrentSnapshotsAndPendingCommits() + public async Task> GetCurrentSnapshots() { - var snapshots = await CurrentSnapshots().Include(s => s.Commit).ToDictionaryAsync(s => s.EntityId); - - if (snapshots.Count == 0) return (snapshots, []); - var lastCommit = snapshots.Values.Select(s => s.Commit).MaxBy(c => c.CompareKey); - ArgumentNullException.ThrowIfNull(lastCommit); - var newCommits = await CurrentCommits() - .Include(c => c.ChangeEntities) - .WhereAfter(lastCommit) - .ToSortedSetAsync(); - return (snapshots, newCommits); + return await CurrentSnapshots().Include(s => s.Commit).ToDictionaryAsync(s => s.EntityId); } public async Task FindCommitByHash(string hash) @@ -376,9 +400,10 @@ private async ValueTask ProjectSnapshot(ObjectSnapshot objectSnapshot) return entity is not null ? _dbContext.Entry(entity) : null; } - public CrdtRepository GetScopedRepository(Commit excludeChangesAfterCommit) + /// null hides all of history, which is what resuming a replay from before the first commit sees + public CrdtRepository GetScopedRepository(Commit? excludeChangesAfterCommit) { - return new CrdtRepository(_dbContext, _crdtConfig, _logger, excludeChangesAfterCommit); + return new CrdtRepository(_dbContext, _crdtConfig, _logger, scoped: true, excludeChangesAfterCommit); } /// @@ -485,11 +510,15 @@ public async ValueTask DisposeAsync() } } -internal class ScopedDbContext(ICrdtDbContext inner, Commit ignoreChangesAfter) : ICrdtDbContext +internal class ScopedDbContext(ICrdtDbContext inner, Commit? ignoreChangesAfter) : ICrdtDbContext { - public IQueryable Commits => inner.Commits.WhereBefore(ignoreChangesAfter, inclusive: true); + public IQueryable Commits => ignoreChangesAfter is null + ? inner.Commits.Where(_ => false) + : inner.Commits.WhereBefore(ignoreChangesAfter, inclusive: true); - public IQueryable Snapshots => inner.Snapshots.WhereBefore(ignoreChangesAfter, inclusive: true); + public IQueryable Snapshots => ignoreChangesAfter is null + ? inner.Snapshots.Where(_ => false) + : inner.Snapshots.WhereBefore(ignoreChangesAfter, inclusive: true); public Task SaveChangesAsync(CancellationToken cancellationToken = default) { diff --git a/src/SIL.Harmony/Db/EntityConfig/CommitEntityConfig.cs b/src/SIL.Harmony/Db/EntityConfig/CommitEntityConfig.cs index eb42691..882e67e 100644 --- a/src/SIL.Harmony/Db/EntityConfig/CommitEntityConfig.cs +++ b/src/SIL.Harmony/Db/EntityConfig/CommitEntityConfig.cs @@ -29,6 +29,10 @@ public void Configure(EntityTypeBuilder builder) builder.HasComplexCompositeIndex( c => new { c.HybridDateTime.DateTime, c.HybridDateTime.Counter, c.Id }, indexName: "IX_Commits_DateTime_Counter_Id"); + // finding the newest checkpoint before a commit is on the hot path of every out of order commit + builder.HasComplexCompositeIndex( + c => new { c.IsSnapshotCheckpoint, c.HybridDateTime.DateTime, c.HybridDateTime.Counter, c.Id }, + indexName: "IX_Commits_IsSnapshotCheckpoint_DateTime_Counter_Id"); builder.Property(c => c.Metadata) .HasColumnType("jsonb") .HasConversion( diff --git a/src/SIL.Harmony/SnapshotCheckpointPolicy.cs b/src/SIL.Harmony/SnapshotCheckpointPolicy.cs new file mode 100644 index 0000000..f1e537a --- /dev/null +++ b/src/SIL.Harmony/SnapshotCheckpointPolicy.cs @@ -0,0 +1,41 @@ +namespace SIL.Harmony; + +/// +/// Picks which commits of a replayed batch become checkpoints, and which snapshots that choice forces the replay to keep. +/// +/// +/// A replay resumes at a checkpoint by seeding every entity from its newest snapshot at or before it, so that snapshot +/// has to be the entity's state there. Dropping a snapshot leaves a gap from it up to the entity's next snapshot, and any +/// checkpoint inside that gap would seed the entity from before an edit nothing is going to re-apply. Hence the two halves +/// here: choose the checkpoints first, then keep whatever snapshots they need. +/// +/// Density is the only dial. Storage scales with it; rollback distance and the cost of reading state at an old commit +/// scale inversely. A commit's flag may only ever be cleared, never set outside a window being replayed, otherwise it +/// claims safety at a position an earlier replay already left a gap in. +/// +/// every Nth commit of a batch is a checkpoint +internal sealed record SnapshotCheckpointPolicy(int Interval) +{ + internal static SnapshotCheckpointPolicy Default { get; } = new(8); + + /// 1 based position in the batch + /// size of the batch; its last commit is always a checkpoint, since every entity keeps the snapshot of its last touch + internal bool IsCheckpoint(int commitIndex, int commitCount) + { + return commitIndex % Interval == 0 || commitIndex == commitCount; + } + + /// + /// Whether the snapshot an entity got at has to be kept, given that the entity's next + /// snapshot in the batch is at . + /// + internal bool MustKeepSnapshot(int commitIndex, int nextCommitIndex) + { + return NextCheckpointAtOrAfter(commitIndex) < nextCommitIndex; + } + + private int NextCheckpointAtOrAfter(int commitIndex) + { + return (commitIndex + Interval - 1) / Interval * Interval; + } +} diff --git a/src/SIL.Harmony/SnapshotWorker.cs b/src/SIL.Harmony/SnapshotWorker.cs index 37a8c1e..798eb1b 100644 --- a/src/SIL.Harmony/SnapshotWorker.cs +++ b/src/SIL.Harmony/SnapshotWorker.cs @@ -17,6 +17,9 @@ internal class SnapshotWorker private readonly Dictionary _pendingSnapshots = []; private readonly Dictionary _rootSnapshots = []; private readonly List _newIntermediateSnapshots = []; + /// position in the batch of each snapshot this run generated, which is what decides whether it may be dropped + private readonly Dictionary _newSnapshotCommitIndex = []; + private readonly SnapshotCheckpointPolicy _checkpointPolicy = SnapshotCheckpointPolicy.Default; private SnapshotWorker(Dictionary snapshots, Dictionary snapshotLookup, @@ -37,7 +40,14 @@ internal static async Task> ApplyCommitsToSnaps { //we need to pass in the snapshots because we expect it to be modified, this is intended. //if the constructor makes a copy in the future this will need to be updated - await new SnapshotWorker(snapshots, [], crdtRepository, crdtConfig).ApplyCommitChanges(commits); + var worker = new SnapshotWorker(snapshots, [], crdtRepository, crdtConfig); + await worker.ApplyCommitChanges(commits); + foreach (var (entityId, rootSnapshot) in worker._rootSnapshots) + { + //entities created during the replay only exist as roots, and a caller asking for state at a commit wants them too + snapshots.TryAdd(entityId, rootSnapshot); + } + return snapshots; } @@ -52,6 +62,8 @@ internal SnapshotWorker(Dictionary snapshotLookup, public async Task UpdateSnapshots(SortedSet commits) { + //deciding the checkpoints before the replay is what makes them a decision rather than a record of what happened to be safe + await _crdtRepository.SetCheckpoints(commits, _checkpointPolicy); await ApplyCommitChanges(commits); await _crdtRepository.AddSnapshots([ .._rootSnapshots.Values, @@ -223,18 +235,22 @@ private async Task GenerateSnapshotForEntity(IObjectBase entity, ObjectSnapshot? { //do nothing, will cause prevSnapshot to be overriden in _pendingSnapshots if it exists } - else if (context.CommitIndex % 2 == 0 && !prevSnapshot.IsRoot && IsNew(prevSnapshot)) + else if (!prevSnapshot.IsRoot + && _newSnapshotCommitIndex.TryGetValue(prevSnapshot.EntityId, out var prevCommitIndex) + && _checkpointPolicy.MustKeepSnapshot(prevCommitIndex, context.CommitIndex)) { + //a checkpoint falls between the two, so this snapshot is what a replay resuming there seeds the entity from context.IntermediateSnapshots[prevSnapshot.Entity.Id] = prevSnapshot; } await _crdtConfig.BeforeSaveObject.Invoke(entity.DbObject, newSnapshot); - AddSnapshot(newSnapshot); + AddSnapshot(newSnapshot, context.CommitIndex); } - private void AddSnapshot(ObjectSnapshot snapshot) + private void AddSnapshot(ObjectSnapshot snapshot, int commitIndex) { + _newSnapshotCommitIndex[snapshot.EntityId] = commitIndex; if (snapshot.IsRoot) { _rootSnapshots[snapshot.Entity.Id] = snapshot; @@ -245,21 +261,4 @@ private void AddSnapshot(ObjectSnapshot snapshot) _pendingSnapshots[snapshot.Entity.Id] = snapshot; } } - - /// - /// snapshot is not from the database - /// - private bool IsNew(ObjectSnapshot snapshot) - { - var entityId = snapshot.EntityId; - if (_pendingSnapshots.TryGetValue(entityId, out var pendingSnapshot)) - { - return pendingSnapshot.Id == snapshot.Id; - } - if (_rootSnapshots.TryGetValue(entityId, out var rootSnapshot)) - { - return rootSnapshot.Id == snapshot.Id; - } - return false; - } }