Sedimentree fragment compaction

docs/design.md

@@ -235,10 +235,42 @@ device that was merely behind produces no divergence and so no anchor,
 and its commits stay unreadable to a later-enrolled member until the
 next local mutation, whose envelope names this frontier, is written on
 top of them; latency again, not loss.
-Wire and store growth is bounded later by sedimentree fragments — a
-roll-up of a commit range into one item — which only a member that can
-open the whole range can build; that is an M-later item, not a gap in
-this one.
+
+**Fragments.** Wire and store growth is bounded by sedimentree
+fragments: a closed range of commits rolled up into one item whose
+payload is an automerge *bundle* of every change in it. Automerge draws
+the ranges, and its metric is sedimentree's — a commit heads a level-1
+fragment when its hash starts with a zero byte, about one in 256 — so
+the two agree on head, boundary and checkpoints with nothing in between
+to disagree. Compaction runs after a local mutation and after an absorb
+that landed; only a device that can read the whole range can build one,
+which falls out of the construction rather than being enforced (a
+commit it could not open was never applied, so no fragment was drawn
+over it). The roll-up is sealed like any commit, and its envelope names
+the key of whatever *carries* its boundary — the fragment below it,
+under that fragment's own reference, since the boundary commit's
+envelope went with its range and its key left the frontier when that
+range was covered. So the fragments form a chain a reader walks down:
+a member enrolled afterwards opens the newest, reads its whole range
+out of the bundle, and follows the embedded key to the one below.
+Sealing a fragment also retires the entry it names, so the head set
+stays one pair per readable branch rather than growing one per
+fragment. The loose commits the fragment carries are then dropped from
+local storage —
+sedimentree's own `minimize` decides which, so a commit concurrent with
+the range is never one of them — while the automerge document keeps its
+full history. Identity is head plus boundary, both functions of the
+change graph, so two devices build the same fragment; the second
+arrival is a no-op locally, and on the store the two land under one
+name and the last write wins, harmlessly, because they carry the same
+range. Nothing deletes from the store, so a device that pushed a range
+before compacting leaves those objects behind: correct, and no smaller
+— the saving is in what is written from then on, and in what a device
+that compacts before publishing sends at all. The pull skips those
+names on the strength of the document rather than a ledger: a change
+the document has applied and the tree no longer holds as an item is one
+a fragment carries, and fetching it back would undo the compaction on
+every pass.
 
 ## Storage
 

runtime/crates/engine/Cargo.toml

@@ -11,6 +11,13 @@ publish.workspace = true
 [dependencies]
 automerge.workspace = true
 beekem.workspace = true
+# The fragment envelope's content reference (`crate::fragment_cref`): a
+# fragment and the commit that heads it are two different plaintexts and must
+# not share one reference in the vault, so the reference is a domain-separated
+# hash of the pair that names the fragment. BLAKE3 rather than a second use of
+# `sha2` because it is already in the workspace for the pairing transcript, and
+# a content reference is not a place to introduce a second hash's habits.
+blake3.workspace = true
 bincode.workspace = true
 ed25519-dalek.workspace = true
 future_form.workspace = true
@@ -29,3 +36,7 @@ subduction_runtime.workspace = true
 [dev-dependencies]
 futures = { workspace = true, features = ["executor"] }
 future_form.workspace = true
+# The fragment-identity test decodes a checkpoint's `Signed<Fragment>` to
+# compare head and boundary; both are already dependencies of the library.
+sedimentree_core.workspace = true
+subduction_crypto.workspace = true

runtime/crates/engine/src/doc.rs

@@ -218,6 +218,28 @@ impl AppDoc {
         self.core.apply(items)
     }
 
+    /// The tree's stored fragment envelopes this document has not opened.
+    pub fn unapplied_fragments(&self, storage: &SnapshotStorage) -> Vec<(CommitId, Vec<u8>)> {
+        self.core.unapplied_fragments(storage)
+    }
+
+    /// Apply decrypted automerge bundles.
+    pub fn apply_bundles(&mut self, bundles: Vec<(CommitId, Vec<u8>)>) -> Absorbed {
+        self.core.apply_bundles(bundles)
+    }
+
+    /// The fragments automerge draws over this document at level 1 and
+    /// deeper. See `crate::document::Document::fragments`.
+    pub fn fragments(&self) -> Vec<automerge::Fragment> {
+        self.core.fragments()
+    }
+
+    /// The bundle bytes for those fragments. See
+    /// `crate::document::Document::bundle`.
+    pub fn bundle(&self, fragments: Vec<automerge::Fragment>) -> Vec<Vec<u8>> {
+        self.core.bundle(fragments)
+    }
+
     pub fn applied_ids(&self) -> std::collections::BTreeSet<CommitId> {
         self.core.applied_ids()
     }

runtime/crates/engine/src/document.rs

@@ -130,9 +130,16 @@ impl Document {
     /// Returns whether anything landed. The plaintext path, for the
     /// user-system document — an app document's blobs are envelopes, and the
     /// engine decrypts them before calling [`Document::apply`].
+    ///
+    /// Fragments first. Automerge buffers a change whose dependencies are
+    /// missing either way, so the order is not required for correctness; it
+    /// is cheaper, because a bundle that lands first makes every loose commit
+    /// it carries a no-op instead of a second decode.
     pub fn absorb(&mut self, storage: &SnapshotStorage) -> bool {
+        let bundles = self.unapplied_fragments(storage);
+        let fragments = self.apply_bundles(bundles);
         let items = self.unapplied(storage);
-        self.apply(items).landed
+        fragments.landed | self.apply(items).landed
     }
 
     /// The commits this document has already applied. The vault needs them to
@@ -153,6 +160,86 @@ impl Document {
             .collect()
     }
 
+    /// The stored *fragment* blobs of this tree the document has not applied.
+    ///
+    /// A fragment is skipped once its head is applied, and that is exact
+    /// rather than approximate: the head is a member of the fragment
+    /// (automerge `change_graph.rs:1661` — `members` is the section the head
+    /// closes), so a document that has the head has been through this bundle.
+    pub fn unapplied_fragments(&self, storage: &SnapshotStorage) -> Vec<(CommitId, Vec<u8>)> {
+        storage
+            .fragment_blobs(self.tree)
+            .into_iter()
+            .filter(|(head, _)| !self.applied.contains(head))
+            .collect()
+    }
+
+    /// Apply fragment payloads: automerge *bundles*, each carrying every
+    /// change of one commit range.
+    ///
+    /// `load_incremental` takes a bundle exactly as it takes a save or a
+    /// single change (automerge `change_graph.rs:1454`,
+    /// `bundle_fragments_roundtrips_through_load_incremental`), and it
+    /// buffers what it cannot yet apply, so a bundle whose boundary has not
+    /// arrived is not an error.
+    ///
+    /// The `applied` set is re-read from the document afterwards rather than
+    /// predicted from the fragment's member list: the document is the
+    /// authority on what it holds, and a bundle names hundreds of changes
+    /// whose ids we would otherwise be copying out of an envelope nobody has
+    /// checked.
+    pub fn apply_bundles(&mut self, bundles: Vec<(CommitId, Vec<u8>)>) -> Absorbed {
+        let mut loaded = false;
+        for (head, bytes) in bundles {
+            if self.applied.contains(&head) {
+                continue;
+            }
+            if self.doc.load_incremental(&bytes).is_ok() {
+                loaded = true;
+            }
+        }
+        if !loaded {
+            return Absorbed::default();
+        }
+        let mut content = false;
+        let mut landed = false;
+        for change in self.doc.get_changes(&[]) {
+            let id = CommitId::new(change.hash().0);
+            if self.applied.insert(id) {
+                landed = true;
+                // As in `apply`: an empty change is a merge anchor and is not
+                // a reason to author another.
+                content |= !change.is_empty();
+            }
+        }
+        Absorbed { landed, content }
+    }
+
+    /// The fragments automerge would draw over this document's history at
+    /// level 1 and deeper, oldest first.
+    ///
+    /// `#[doc(hidden)]`/EXPERIMENTAL upstream, and used anyway: automerge's
+    /// fragments are co-designed with sedimentree — `ChangeHash`'s
+    /// `fragment_level` counts leading zero *bytes* (automerge
+    /// `types.rs:680`), which is `CountLeadingZeroBytes` exactly
+    /// (sedimentree_core `depth.rs`) — so this is the one decomposition whose
+    /// heads, boundaries and checkpoints line up with the tree the sync
+    /// engine already keeps. Reimplementing it over `get_changes` would be a
+    /// second implementation of the same partition, free to disagree.
+    /// Ink & Switch's own adapter does exactly this mapping
+    /// (`legacy/automerge_subduction_ingest/src/main.rs`, `ingest_automerge`).
+    pub fn fragments(&self) -> Vec<automerge::Fragment> {
+        self.doc.fragments(1..)
+    }
+
+    /// The bundle bytes for each fragment, in the order given. Separate from
+    /// [`Document::fragments`] because bundling re-encodes every member of
+    /// every fragment handed to it, and the caller drops all but the ones it
+    /// has not already stored.
+    pub fn bundle(&self, fragments: Vec<automerge::Fragment>) -> Vec<Vec<u8>> {
+        self.doc.bundle_fragments(fragments)
+    }
+
     /// Whether the document has more than one head — concurrent branches that
     /// nothing has merged yet. See [`Document::merge_anchor`].
     pub fn diverged(&self) -> bool {