From 4c9ebcd43179737f7aed4e264b3e6e1708bbbb5e Mon Sep 17 00:00:00 2001 From: Tim Haasdyk Date: Fri, 4 Sep 2026 12:03:02 +0200 Subject: [PATCH 1/3] Add failing tests for snapshot loss after a late commit Adding several commits at once only keeps a snapshot for every other one. A commit dated between a commit with no snapshot and the next one that has one makes the replay start after its own parent, while the entity resumes from the older snapshot it still has, so the commits in between are applied by nobody. Both tests fail: one loses an edit, the other revives a cascade-deleted definition whose word is still deleted and breaks the foreign key. Co-Authored-By: Claude Opus 5 --- src/SIL.Harmony.Tests/LateCommitTests.cs | 67 ++++++++++++++++++++++++ 1 file changed, 67 insertions(+) create mode 100644 src/SIL.Harmony.Tests/LateCommitTests.cs 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(); + } +} From a553c73a897419bef2b09bbdc51379cf771059c1 Mon Sep 17 00:00:00 2001 From: Tim Haasdyk Date: Fri, 4 Sep 2026 17:05:09 +0200 Subject: [PATCH 2/3] Document the snapshot checkpoint design Notes from the investigation into #105: the completeness invariant, why a hole is an interval rather than a commit, the rules that make checkpoints safe, and the dead ends so nobody repeats them. Records the measurements too, since most of the arguments here turn on numbers. Co-Authored-By: Claude Opus 5 --- docs/snapshot-checkpoints.md | 325 +++++++++++++++++++++++++++++++++++ 1 file changed, 325 insertions(+) create mode 100644 docs/snapshot-checkpoints.md diff --git a/docs/snapshot-checkpoints.md b/docs/snapshot-checkpoints.md new file mode 100644 index 0000000..6d662d9 --- /dev/null +++ b/docs/snapshot-checkpoints.md @@ -0,0 +1,325 @@ +# 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. + +## 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. From 62bf84a1f3a8d3e50fa3dfc302a4f126fe37a18f Mon Sep 17 00:00:00 2001 From: Tim Haasdyk Date: Fri, 4 Sep 2026 17:50:08 +0200 Subject: [PATCH 3/3] Resume snapshot replays from checkpoints Fixes #105. A commit dated before commits already in the database made the replay resume each entity from whatever snapshot survived, which can predate edits whose snapshots were pruned, so nothing re-applied them. One symptom lost an edit, the other revived a cascade-deleted entity and broke the FK to its deleted parent. Commits now carry IsSnapshotCheckpoint: a position where every entity's newest snapshot at or before it is that entity's state there, so a replay can resume from it. SnapshotCheckpointPolicy picks every 8th commit of a replayed batch plus its last, and the pruner keeps whatever snapshots that choice needs, which replaces the CommitIndex % 2 rule. Reading state at an old commit resumes from a checkpoint too, so a change no longer reads a neighbour's state as of the commit being asked about rather than the position being replayed. Flags are only ever written for commits inside a window being replayed. When there is no checkpoint before the late commit, nothing safe exists to resume from, so everything is dropped and regenerated: that is both the repair and the bootstrap for databases written before this existed. Consumers need a migration for the new Commits column. +semver: minor Co-Authored-By: Claude Opus 5 --- docs/snapshot-checkpoints.md | 15 + ...eChanges.WriteMultipleCommits.verified.txt | 2 + ...hangesAtOnceWithMergedHistory.verified.txt | 43 +-- ...DoesNotEffectTheFirstSnapshot.verified.txt | 2 + ....WritingAChangeMakesASnapshot.verified.txt | 1 + ...ommitWithMultipleChangesWorks.verified.txt | 1 + src/SIL.Harmony.Tests/DataModelTestBase.cs | 6 +- .../DbContextTests.VerifyModel.verified.txt | 3 +- src/SIL.Harmony.Tests/DbContextTests.cs | 1 + src/SIL.Harmony.Tests/ModelSnapshotTests.cs | 19 +- src/SIL.Harmony.Tests/RepositoryTests.cs | 42 ++- .../SnapshotCheckpointPolicyTests.cs | 58 ++++ .../SnapshotCheckpointTests.cs | 272 ++++++++++++++++++ src/SIL.Harmony.Tests/SnapshotTests.cs | 3 +- src/SIL.Harmony/Commit.cs | 9 + src/SIL.Harmony/DataModel.cs | 75 ++--- src/SIL.Harmony/Db/CrdtRepository.cs | 83 ++++-- .../Db/EntityConfig/CommitEntityConfig.cs | 4 + src/SIL.Harmony/SnapshotCheckpointPolicy.cs | 41 +++ src/SIL.Harmony/SnapshotWorker.cs | 41 ++- 20 files changed, 592 insertions(+), 129 deletions(-) create mode 100644 src/SIL.Harmony.Tests/SnapshotCheckpointPolicyTests.cs create mode 100644 src/SIL.Harmony.Tests/SnapshotCheckpointTests.cs create mode 100644 src/SIL.Harmony/SnapshotCheckpointPolicy.cs diff --git a/docs/snapshot-checkpoints.md b/docs/snapshot-checkpoints.md index 6d662d9..29bb24e 100644 --- a/docs/snapshot-checkpoints.md +++ b/docs/snapshot-checkpoints.md @@ -163,6 +163,21 @@ 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. 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/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; - } }