fix(crdt): make flush cost proportional to what changed - #10
Open
mkhairi wants to merge 4 commits into
Open
Conversation
Every flush re-exported and rewrote the full Loro snapshot of every collection, with no check for whether the document had moved. A snapshot export costs O(document), so the work an idle store did per tick was the size of its whole state. Two consequences followed from that one assumption. The file grew by a full snapshot copy per `auto_flush_ms` with no writes behind it, and the superseded pages are only reclaimed by auto-compact, which is off by default. And once a document grew large enough that its export outlasted the flush interval, the flush task held the reader-visible `crdt` lock for essentially all wall time, so CRDT reads never got scheduled. Track the frontier each collection had when its snapshot was last written, and export only those whose `oplog_version_vector()` has moved since. The mark is recorded after the batch commits, so a failed write leaves the collection dirty and is retried rather than skipped. History compaction rewrites the document without advancing its frontier, so it drops the marks it invalidates. `crdt_snapshot_export_count()` exposes the export count, so the property can be asserted directly instead of inferred from a timing.
Exporting only changed collections removes the idle cost, but the store's actual workload is sustained writes, and under those every flush still rewrote each dirty collection in full. One row changing in a 77 MB collection cost 77 MB of export and 77 MB of superseded pages, once per `auto_flush_ms`. Between checkpoints a flush now exports the operations since the last persisted frontier and stores them under `loro_delta:<collection>:<seq>`, so per-flush cost is O(new operations). The base snapshot is rewritten once the accumulated updates reach a quarter of it, which bounds what open has to replay and amortises the O(document) export over the writes that made it necessary. A checkpoint deletes the updates it supersedes in the same batch, so no restore ever replays operations the base already contains. These updates are durability, not sync: they are deleted when the base that contains them is rewritten, whereas `crdt:delta:<mutation_id>` entries are the queue to Origin and are deleted on acknowledgement. Reusing that queue as the durability log would drop the state of every row Origin had already acknowledged. Restore replays the updates in key order after importing the base and seeds the checkpoint accounting from what it found, so the first flush after an open does not rewrite a base that is already current. An update that fails its CRC32C check is an error rather than a warning: opening without it would silently roll the collection back to the checkpoint.
Tokio's default interval behaviour replays every missed tick as soon as the consumer comes back. For a task whose work can outlast its own period that turns one slow pass into back-to-back passes with no gap: each flush takes the reader-visible `crdt` lock, so a burst of them is a stretch of wall time in which no CRDT read is scheduled at all. Both periodic tasks now measure the next period from the end of the previous pass. This is already the WASM behaviour, so it only changes native. The remaining half of the lock problem is that the snapshot export itself runs under `self.crdt`. Moving it out needs a `CrdtState` handle that can be exported while the engine lock is released, which `nodedb-crdt` deliberately prevents today — its `_single_owner` marker makes sharing one a compile error. That is a change to its contract, not to this crate, so it is not folded in here.
mkhairi
force-pushed
the
fix/crdt-flush-amplification
branch
from
August 2, 2026 04:36
98283df to
108cbd2
Compare
Member
Author
|
CI here stops at dependency resolution, before reaching this branch's code: The workspace pins all fourteen Locally on the rebased branch: |
The unsent-delta queue is append-only and each entry is stored under its own `crdt:delta:<mutation_id>` key, but every flush re-wrote every entry and then wrote the same content again as the legacy bulk blob. The cost is the length of the queue, not what changed — and a replica with no Origin never has a delta acknowledged, so the queue only grows. Measured on a 26,409-object store: an idle flush, with no collection dirty and zero snapshot exports, still spent 1282 of its 1286 seconds inside one `batch_write` of 67,931 puts totalling 236 MB — 67,885 queue entries rewritten byte-identical, plus a 115 MB bulk blob duplicating them. That is 100% CPU with no reads and nothing changed to show for it, and it leaves 287 MB of superseded pages behind per pass. Track which entries are not known to match their stored form — the ones appended since the last flush, and any whose `seq` a send has since assigned — and write only those. The bulk blob duplicates them all, so it is rewritten only when the queue actually changed rather than every tick. Restoring from the bulk blob marks every entry unpersisted, since that path is the only copy they came from; restoring from the individual entries marks none, since each was just read from its own key. `crdt_delta_write_count()` exposes the count, so the property is a counter assertion rather than a timing. The bulk blob remains a full rewrite whenever the queue does change, which under sustained writes is every flush. It duplicates data that the per-entry keys already hold and that restore prefers over it, so it looks retireable — but that is an on-disk format decision, so it is left alone here.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Commits 1–3 of the shape agreed in NodeDB-Lab/nodedb#230: flush no longer treats a collection's full snapshot as a cheap value to rewrite on every tick. An idle store now does no snapshot work, and a store under sustained writes pays per flush in new operations rather than in document size.
Commit 4 of that plan — separating local restore from peer admission — is
nodedb-crdtcode and follows as a separate PR on the nodedb repo. This branch contains no changes there, and that one contains none here.Changes
fix(crdt): export a collection's snapshot only when it has changedCrdtEnginerecords the frontier each collection had when its snapshot was last written;flush()exports only collections whoseoplog_version_vector()has moved. The mark is recorded after the batch commits, so a failed write leaves the collection dirty and is retried rather than skipped.compact_historyandcompact_at_versionrewrite the document without advancing the frontier, so they drop the marks they invalidate.crdt_snapshot_export_count()exposes the export count so the property can be asserted directly instead of inferred from a timing.feat(crdt): write incremental updates between snapshot checkpointsBetween checkpoints a flush exports the operations since the last persisted frontier under
loro_delta:<collection>:<seq>, and rewrites the base snapshot only once accumulated updates reach a quarter of it. A checkpoint deletes the updates it supersedes in the same batch, so no restore replays operations the new base already contains. Restore replays the updates in key order after importing the base and seeds the checkpoint accounting from what it found, so the first flush after an open does not rewrite a base that is already current. An update failing its CRC32C check is an error rather than a warning: opening without it would silently roll the collection back to the checkpoint.fix(core): let a slow flush delay its next tick instead of burstingTokio replays missed ticks immediately by default, so a flush that outlasted its own interval was followed by the ticks it missed with no gap between them, each taking the
crdtlock. Both periodic tasks now measure the next period from the end of the previous pass. This is already the WASM behaviour, so it changes native only.Design notes
A separate key space for the durability updates. The issue points at
crdt:delta:{mutation_id}andrestore_pending_deltas_incrementalas the existing incremental path. Those entries are the sync outbox: they are deleted when Origin acknowledges them, not when a base snapshot absorbs them. Using that queue as the durability log would drop the state of every row Origin had already acknowledged. Henceloro_delta:<collection>:<seq>inNamespace::LoroState, replayed on open and deleted by the checkpoint that contains it. Happy to rename the keys if a different scheme fits better.Half of the lock work is not in this branch. Bounding the tick cadence is here; moving the export itself out from under
self.crdtis not. It needs aCrdtStatehandle that can be exported while the engine lock is released, and_single_owner: PhantomData<Cell<()>>makes sharing one a compile error by design. That isnodedb-crdt's contract to change, so it belongs with the decision about how such a handle should be given out rather than with a workaround here.Tests
nodedb-lite/tests/crdt_flush_dirty_tracking.rs, all counter assertions rather than timings, so they fail for the reason they name on any machine:Each was confirmed to fail with the fix disabled.
A fourth test — file size within a bound of live-state size — was written and dropped. At test scale pagedb reclaims the superseded pages when nothing pins them, so 128 full rewrites left the file under 64 KB and the test passed against the unfixed code. A test that goes green on the bug is worse than no test; the export counter is the form that discriminates.
Validation
cargo fmt --all— cleancargo clippy --workspace --all-targets -- -D warnings— cleancargo nextest run -p nodedb-lite— 24 failures, all reproducing on unmodifiedmain: 4 inauto_flush(reopen storage: already open, pagedb releasing its advisory writer lock asynchronously) and 20 insync_interop_*(connect to Origin pgwire: password missing, no Origin available in this environment)Acceptance
The preserved store that reproduces the original fault needs commit 4 before it will open, so it cannot exercise this branch yet. It is kept, and I will report the result once that PR lands.