diff --git a/Cargo.lock b/Cargo.lock index fdb57488..cf3bf747 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1694,6 +1694,7 @@ dependencies = [ "automerge", "beekem", "bincode", + "blake3", "ed25519-dalek", "future_form", "futures", diff --git a/docs/design.md b/docs/design.md index eecb3e59..cfe0b7d8 100644 --- a/docs/design.md +++ b/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 diff --git a/runtime/crates/engine/Cargo.toml b/runtime/crates/engine/Cargo.toml index 0b3c19c0..306ebe2b 100644 --- a/runtime/crates/engine/Cargo.toml +++ b/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` to +# compare head and boundary; both are already dependencies of the library. +sedimentree_core.workspace = true +subduction_crypto.workspace = true diff --git a/runtime/crates/engine/src/doc.rs b/runtime/crates/engine/src/doc.rs index 105a7f3d..737a8505 100644 --- a/runtime/crates/engine/src/doc.rs +++ b/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)> { + self.core.unapplied_fragments(storage) + } + + /// Apply decrypted automerge bundles. + pub fn apply_bundles(&mut self, bundles: Vec<(CommitId, Vec)>) -> 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 { + self.core.fragments() + } + + /// The bundle bytes for those fragments. See + /// `crate::document::Document::bundle`. + pub fn bundle(&self, fragments: Vec) -> Vec> { + self.core.bundle(fragments) + } + pub fn applied_ids(&self) -> std::collections::BTreeSet { self.core.applied_ids() } diff --git a/runtime/crates/engine/src/document.rs b/runtime/crates/engine/src/document.rs index ed933e4a..1f549c5d 100644 --- a/runtime/crates/engine/src/document.rs +++ b/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)> { + 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)>) -> 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 { + 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) -> Vec> { + 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 { diff --git a/runtime/crates/engine/src/lib.rs b/runtime/crates/engine/src/lib.rs index d960b80c..0ea84aa8 100644 --- a/runtime/crates/engine/src/lib.rs +++ b/runtime/crates/engine/src/lib.rs @@ -12,7 +12,7 @@ //! `RefCell`. use std::cell::RefCell; -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use std::future::Future; use std::pin::Pin; use std::rc::Rc; @@ -29,7 +29,7 @@ mod vault; pub use clock::EngineClock; pub use doc::{TaskSnapshot, TodoItem}; pub use ed25519_dalek::VerifyingKey; -pub use storage::{AppState, Item, Snapshot, StoreItem, TreeState}; +pub use storage::{AppState, Item, ItemKind, Snapshot, StoreItem, TreeState}; pub use subduction_protocol::peer_id::PeerId; pub use transport::{DynTransport, EngineTransport}; pub use us::{Member, us_tree}; @@ -600,6 +600,59 @@ impl + 'static> Engine { self.storage.all_items() } + /// Every `(tree, commit)` this device has *read* but does not hold as an + /// item: a change that is in the document and whose loose commit is not + /// in the tree, because a fragment carries it instead. + /// + /// The store deletes nothing, so a commit pruned by compaction is still + /// there under its name, and to a pull that only knows [`Engine::items`] + /// it looks exactly like a commit some other device wrote and this one + /// has never seen — so every pass would fetch the whole compacted range + /// back, and `accept` would refuse it, forever. This is the set that + /// closes that loop. + /// + /// Derived from the documents rather than remembered: the document *is* + /// the record of what this device has read, rebuilt from its own changes + /// on every load (`crate::document::Document`), and it answers for a + /// range that arrived as somebody else's fragment just as well as for one + /// this device compacted itself. + #[must_use] + pub fn read_not_held(&self) -> Vec<([u8; 32], [u8; 32])> { + let mut found = Vec::new(); + let mut walk = |tree: SedimentreeId, applied: std::collections::BTreeSet| { + for id in applied { + if !self.storage.holds(tree, id) { + found.push((*tree.as_bytes(), *id.as_bytes())); + } + } + }; + for doc in self.apps.borrow().values() { + walk(doc.tree(), doc.applied_ids()); + } + if let Some(doc) = self.us.borrow().as_ref() { + walk(us_tree(), doc.applied_ids()); + } + found + } + + /// Whether `tree`'s document has already applied `commit` — the change is + /// in this device's history whether or not the commit that carried it is + /// still an item. See [`Engine::read_not_held`]. + fn read(&self, tree: SedimentreeId, commit: CommitId) -> bool { + if tree == us_tree() { + return self + .us + .borrow() + .as_ref() + .is_some_and(|doc| doc.applied_ids().contains(&commit)); + } + self.apps + .borrow() + .values() + .find(|doc| doc.tree() == tree) + .is_some_and(|doc| doc.applied_ids().contains(&commit)) + } + /// Install items a *store* handed back — another device of this group /// pushed them — and apply whatever they unlock. Answers whether anything /// was new, which is the kernel's cue to checkpoint. @@ -617,7 +670,7 @@ impl + 'static> Engine { pub async fn ingest_items(&self, items: Vec) -> Result { self.hydrate().await?; self.open_us().await?; - let mut by_tree: BTreeMap> = BTreeMap::new(); + let mut by_tree: BTreeMap, Vec)> = BTreeMap::new(); let mut fresh = false; // The member keys, read once: the store is not a peer, so nothing // else has checked who authored what it hands back. @@ -625,13 +678,15 @@ impl + 'static> Engine { for item in items { if self.accept(&members, &item) { fresh = true; - by_tree - .entry(SedimentreeId::new(item.tree)) - .or_default() - .push(Item { - signed: item.signed, - blob: item.blob, - }); + let bucket = by_tree.entry(SedimentreeId::new(item.tree)).or_default(); + let landing = match item.kind { + ItemKind::Commit => &mut bucket.0, + ItemKind::Fragment => &mut bucket.1, + }; + landing.push(Item { + signed: item.signed, + blob: item.blob, + }); } } if !fresh { @@ -648,8 +703,8 @@ impl + 'static> Engine { } }); for tree in order { - let items = by_tree.remove(&tree).unwrap_or_default(); - self.storage.restore(tree, items, Vec::new()); + let (commits, fragments) = by_tree.remove(&tree).unwrap_or_default(); + self.storage.restore(tree, commits, fragments); let (commits, fragments) = self.storage.metadata(tree); // Merged, not replaced: `Command::HydrateTree` adds each commit to // the resident tree (subduction_protocol/src/core_machine.rs:284), @@ -685,34 +740,57 @@ impl + 'static> Engine { /// before the app trees, so that pass is usually the same one); /// - the **tree** the object was filed under is the one the commit names /// (`sedimentree_id`), so an item cannot be moved between trees; - /// - the **commit id** the object was named by is the commit's own head; - /// - the **blob** is the one the commit committed to (`BlobMeta`), so the + /// - the **item id** the object was named by is the item's own head; + /// - the **blob** is the one the item committed to (`BlobMeta`), so the /// signed metadata and the bytes beside it cannot be from two different /// items. /// + /// The same five checks for a fragment, against `Signed` — a + /// fragment is a signed sedimentree item like any other, and a store that + /// could hand back an unverified one would be handing back a whole range + /// of forged history in a single object rather than one commit's worth. + /// /// A failing item is skipped, not fatal: the folder is the user's own /// Drive and one bad object must not stop the rest from landing. fn accept(&self, members: &std::collections::BTreeSet<[u8; 32]>, item: &StoreItem) -> bool { - let Ok(signed) = subduction_crypto::signed::Signed::< - sedimentree_core::loose_commit::LooseCommit, - >::try_decode(&item.signed) else { - return false; - }; - let Ok(verified) = signed.try_verify() else { - return false; - }; - if !members.contains(&verified.issuer().to_bytes()) { - return false; - } - let payload = verified.payload(); let tree = SedimentreeId::new(item.tree); - if payload.sedimentree_id() != tree - || payload.head() != CommitId::new(item.commit) - || *payload.blob_meta() != BlobMeta::new(&Blob::new(item.blob.clone())) - { - return false; + let id = CommitId::new(item.commit); + let blob = BlobMeta::new(&Blob::new(item.blob.clone())); + match item.kind { + ItemKind::Commit => { + let Some(payload) = + verify::(&item.signed, members) + else { + return false; + }; + if payload.sedimentree_id() != tree + || payload.head() != id + || *payload.blob_meta() != blob + { + return false; + } + // Held is not the whole of "not news": a commit whose change + // the document has already applied was read and then pruned + // (or never held loose at all, having arrived inside somebody + // else's fragment). Reinstating it would undo the compaction + // on every pass. See [`Engine::read_not_held`]. + !self.storage.holds(tree, id) && !self.read(tree, id) + } + ItemKind::Fragment => { + let Some(payload) = + verify::(&item.signed, members) + else { + return false; + }; + if payload.sedimentree_id() != tree + || payload.head() != id + || payload.summary().blob_meta() != blob + { + return false; + } + !self.storage.holds_fragment(tree, id) + } } - !self.storage.holds(tree, payload.head()) } // -- checkpointing ------------------------------------------------------- @@ -956,6 +1034,12 @@ impl + 'static> Engine { }; self.refresh_members(); self.push_us_commit(commit).await?; + // The group document is compacted on the same terms as an app's: + // `Engine::absorb` covers what arrives, this covers what is written + // here. A group that reaches a fragment's worth of membership edits + // is not a case anyone expects, and the cost of saying so is one + // walk of a very short change graph. + let _compacted = self.compact(us_tree()).await; Ok(answer) } @@ -1035,10 +1119,166 @@ impl + 'static> Engine { // Encrypting may have advanced the document's CGKA epoch, and the // update op is what lets the other devices follow. self.publish_keyhive().await?; + // One commit in ~256 closes a level-1 fragment (its hash starts + // with a zero byte); the other 255 times this walks the change + // graph, finds every fragment already held, and stops before + // bundling anything. Not free — `fragments(1..)` is linear in the + // history — but linear in a walk automerge does over its own + // index, not in re-encoding the document. + // + // Not `?`: the mutation is already durable — the barrier above + // saw to that — so a compaction that failed must not report the + // user's write as failed. The roll-up waits for the next one. + let _compacted = self.compact(tree).await; } Ok(answer) } + /// Roll every closed commit range of `tree` up into a sedimentree + /// fragment, and drop the loose commits the roll-up carries. + /// + /// Called after a local mutation and after an absorb that landed + /// (`Engine::mutate`, `Engine::absorb`), which between them cover every + /// way this device's history grows. + /// + /// **Who may build one.** Only a device that can read the whole range: + /// the fragment's payload is an automerge *bundle* of its members' + /// changes, and building it means having those changes in the document. + /// That falls out of the construction rather than being enforced — a + /// device that could not open an envelope never applied it, so automerge + /// never drew a fragment over it. A commit this device could not read is + /// outside the fragment's members and stays loose, and sedimentree + /// decides coverage by head/checkpoints/boundary + /// (`Fragment::supports_block`), so nothing claims to carry it. + /// + /// The keyhive-events tree is skipped: it has no automerge document — + /// its commits *are* the state, unordered and content-addressed (see + /// [`keyhive_tree`]) — so there is no change graph to fragment. + async fn compact(&self, tree: SedimentreeId) -> Result<(), String> { + if tree == keyhive_tree() { + return Ok(()); + } + // Before anything else, and unconditionally: pruning is the last + // step of building a fragment and every step before it can fail + // (`publish_keyhive`, a storage write), which would leave a fragment + // durable with its range still loose beside it. Re-running it here + // costs one `minimize` and catches that on the next turn. + let _pruned = self.storage.prune(tree); + // The group document is plaintext by ruling (`crate::vault` module + // docs), so its fragments are too; an app tree's are envelopes like + // its commits. + let enveloped = tree != us_tree(); + // A fragment whose head we already hold is one we have already built + // or received — identity is head plus boundary + // (`design/sedimentree.md`) and the tree is keyed by head, so two + // devices with the same causal graph produce the same one and the + // second is a no-op. Filtered *before* bundling: `bundle_fragments` + // re-encodes every member of every fragment it is handed, so bundling + // the whole history to throw all but the newest away would make each + // mutation cost the whole document. + let candidates: Vec<(automerge::Fragment, Vec)> = { + let apps = self.apps.borrow(); + let doc: Option<&AppDoc> = enveloped + .then(|| apps.values().find(|doc| doc.tree() == tree)) + .flatten(); + if enveloped && doc.is_none() { + return Ok(()); + } + let fragments = match doc { + Some(doc) => doc.fragments(), + None => self.with_us(UsDoc::fragments), + }; + let fresh: Vec = fragments + .into_iter() + .filter(|f| !self.storage.holds_fragment(tree, CommitId::new(f.head.0))) + .collect(); + if fresh.is_empty() { + return Ok(()); + } + let bundles = match doc { + Some(doc) => doc.bundle(fresh.clone()), + None => self.with_us(|us| us.bundle(fresh.clone())), + }; + fresh.into_iter().zip(bundles).collect() + }; + for (fragment, bundle) in candidates { + let head = CommitId::new(fragment.head.0); + let boundary: BTreeSet = fragment + .boundary + .iter() + .map(|hash| CommitId::new(hash.0)) + .collect(); + let checkpoints: Vec = fragment + .checkpoints + .iter() + .map(|hash| CommitId::new(hash.0)) + .collect(); + let (blob, sealed) = if enveloped { + let vault = self.require_vault()?; + // What keeps the causal walk going below the fragment. The + // boundary names the commits just under it, and for each one + // the *carrier* is what has to be named: if we hold a + // fragment headed at that commit, the thing a later reader + // must be able to open is that fragment, under its own cref — + // the boundary commit's own envelope was pruned along with + // its range, and its content key left the frontier when it + // was covered, so naming the commit would embed nothing at + // all. `Vault::seal` embeds exactly those preds whose keys + // this device still holds, and `Vault::confirm` then drops + // them from the frontier, which is what keeps the head set + // at one entry point per branch instead of one per fragment. + let preds: Vec<[u8; 32]> = boundary + .iter() + .map(|id| { + if self.storage.holds_fragment(tree, *id) { + fragment_cref(tree, *id) + } else { + *id.as_bytes() + } + }) + .collect(); + let sealed = vault + .seal(fragment_cref(tree, head), &preds, bundle) + .await?; + (Blob::new(sealed.blob.clone()), Some(sealed)) + } else { + (Blob::new(bundle), None) + }; + self.handle + .add_fragments( + tree, + vec![subduction_protocol::command::NewFragment { + head, + boundary, + checkpoints, + blob, + }], + ) + .await + .map_err(|e| e.to_string())?; + // The durability barrier `mutate` documents: the fragment must be + // in storage before anything is dropped on the strength of it. + let _heads = self + .handle + .tree_heads(tree) + .await + .map_err(|e| e.to_string())?; + if let Some(sealed) = sealed { + let vault = self.require_vault()?; + // `confirm` makes the fragment an entry point and drops the + // preds it embedded — which is where the *previous* fragment + // stops being one, since this envelope now carries its key. + vault.confirm(&sealed); + // And the members it carries stop being entry points too: + // their changes are in the bundle (`Vault::cover`). + vault.cover(fragment.members.iter().map(|hash| hash.0)); + self.publish_keyhive().await?; + } + let _pruned = self.storage.prune(tree); + } + Ok(()) + } + /// Replace an app commit's plaintext change with its keyhive envelope. /// This is the whole of M3c's claim: what reaches the driver — and so the /// wire, the relay and storage — is ciphertext. @@ -1116,6 +1356,22 @@ impl + 'static> Engine { /// is an async call into the vault. The `RefCell` borrows are taken and /// dropped around each await rather than across one. async fn absorb(&self, tree: SedimentreeId) -> bool { + let landed = self.absorb_items(tree).await; + if landed { + // The second compaction trigger. Absorbing is how a device that + // was behind catches up, and a batch of a few hundred commits is + // exactly the case fragments exist for; a device that only ever + // compacted its own writes would carry a peer's history loose + // forever. Failure is not the caller's business — nothing here + // is lost if the roll-up waits for the next batch. + let _compacted = self.compact(tree).await; + } + landed + } + + /// [`Engine::absorb`] without the compaction step: what actually applies + /// the tree's stored items to its document. + async fn absorb_items(&self, tree: SedimentreeId) -> bool { if tree == us_tree() { let landed = { let mut cell = self.us.borrow_mut(); @@ -1153,22 +1409,54 @@ impl + 'static> Engine { // stay in storage and are tried again on the next event. return false; }; - let (wanted, known) = { + let (wanted, bundles, tree, known) = { let apps = self.apps.borrow(); let Some(doc) = apps.get(app) else { return false; }; ( doc.unapplied(&self.storage), + doc.unapplied_fragments(&self.storage), + doc.tree(), doc.applied_ids() .into_iter() .map(|id| *id.as_bytes()) .collect(), ) }; - if wanted.is_empty() { + if wanted.is_empty() && bundles.is_empty() { return false; } + // Fragments first, as `Document::absorb` does and for the same + // reason: a bundle that lands first turns every loose commit it + // carries into a no-op. Their envelopes are keyed by the fragment + // cref, not by the head — a fragment and its head commit are two + // different plaintexts and may not share one content reference — so + // the walk's answers are mapped back through `by_cref`. + let by_cref: BTreeMap<[u8; 32], CommitId> = bundles + .iter() + .map(|(head, _)| (fragment_cref(tree, *head), *head)) + .collect(); + let opened_bundles = if bundles.is_empty() { + Vec::new() + } else { + match vault + .open( + bundles + .into_iter() + .map(|(head, blob)| (fragment_cref(tree, head), blob)) + .collect(), + &known, + ) + .await + { + Ok(opened) => opened + .into_iter() + .filter_map(|(cref, bundle)| Some((*by_cref.get(&cref)?, bundle))) + .collect(), + Err(_) => Vec::new(), + } + }; let Ok(opened) = vault .open( wanted @@ -1186,12 +1474,15 @@ impl + 'static> Engine { let Some(doc) = apps.get_mut(app) else { return false; }; - let absorbed = doc.apply( + let from_fragments = doc.apply_bundles(opened_bundles); + let mut absorbed = doc.apply( opened .into_iter() .map(|(id, change)| (CommitId::new(id), change)) .collect(), ); + absorbed.landed |= from_fragments.landed; + absorbed.content |= from_fragments.content; // The partition case (`design/causal_encryption.md` §"Multiple // Heads"): what just landed was concurrent with what this device // already had, so it was sealed under an epoch some *other* member @@ -1267,6 +1558,46 @@ impl + 'static> Engine { } } +/// Where a fragment's envelope lives in the vault: `blake3("polyvisor:fragment" +/// ‖ tree ‖ head)`. +/// +/// Not the head itself, which is what a fragment is *named* by. A content +/// reference indexes one plaintext in keyhive's ciphertext store, and the head +/// commit already owns that reference for its own change; a fragment stored +/// under it would collide with the commit whose range it closes — the walk +/// would find one where it wanted the other, and the key it holds would open +/// neither reliably. Domain-separated so the two can never coincide. +fn fragment_cref(tree: SedimentreeId, head: CommitId) -> [u8; 32] { + *blake3::Hasher::new() + .update(b"polyvisor:fragment") + .update(tree.as_bytes()) + .update(head.as_bytes()) + .finalize() + .as_bytes() +} + +/// Decode a store object's envelope, check the signature, and check the +/// issuer is one of `members`. The half of [`Engine::accept`] that does not +/// depend on which item kind it is. +/// +/// `try_verify`, not the trusted-storage decode: the trusted decode reads the +/// fields of an envelope nobody has checked, which is exactly the situation it +/// documents itself as being wrong for. +fn verify(signed: &[u8], members: &std::collections::BTreeSet<[u8; 32]>) -> Option +where + T: sedimentree_core::codec::schema::Schema + + sedimentree_core::codec::encode::EncodeFields + + sedimentree_core::codec::decode::DecodeFields + + Clone, +{ + let signed = subduction_crypto::signed::Signed::::try_decode(signed).ok()?; + let verified = signed.try_verify().ok()?; + if !members.contains(&verified.issuer().to_bytes()) { + return None; + } + Some(verified.payload().clone()) +} + /// A domain-separated 32 bytes from the device seed and this run's /// randomness. fn mix(domain: &[u8], seed: &[u8; 32], entropy: &[u8; 32]) -> [u8; 32] { diff --git a/runtime/crates/engine/src/storage.rs b/runtime/crates/engine/src/storage.rs index aeafd955..34eac7de 100644 --- a/runtime/crates/engine/src/storage.rs +++ b/runtime/crates/engine/src/storage.rs @@ -15,14 +15,16 @@ //! do — and it would quietly relabel their authorship as ours. use std::cell::RefCell; -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use future_form::{FutureForm as _, Local}; use futures::future::LocalBoxFuture; use sedimentree_core::{ + depth::CountLeadingZeroBytes, fragment::Fragment, id::SedimentreeId, loose_commit::{LooseCommit, id::CommitId}, + sedimentree::Sedimentree, }; use serde::{Deserialize, Serialize}; use subduction_crypto::signed::Signed; @@ -82,6 +84,21 @@ pub struct TreeState { pub fragments: Vec, } +/// Which sedimentree item a [`StoreItem`] carries. +/// +/// The store is names plus opaque bytes, but the two item kinds decode into +/// different envelopes (`Signed` vs `Signed`) and are +/// checked differently, so the record has to say which it is. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum ItemKind { + /// One automerge change. + #[default] + Commit, + /// A roll-up of a commit range: one automerge bundle. + Fragment, +} + /// A stored sedimentree item: its signed envelope and its payload blob. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct Item { @@ -89,7 +106,7 @@ pub struct Item { pub blob: Vec, } -/// One sedimentree item as the durable store carries it: the tree and commit +/// One sedimentree item as the durable store carries it: the tree and item id /// that name it, and the bytes. /// /// Ids are raw bytes rather than `SedimentreeId`/`CommitId` so the kernel can @@ -99,9 +116,15 @@ pub struct Item { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct StoreItem { pub tree: [u8; 32], + /// The commit id for a commit, the head for a fragment. pub commit: [u8; 32], pub signed: Vec, pub blob: Vec, + /// `#[serde(default)]` — `Commit` — because objects written before + /// fragments existed carry no such field, and a device must still read + /// its group's older objects. + #[serde(default)] + pub kind: ItemKind, } /// Map-backed item storage. Interior mutability is a `RefCell`: the driver @@ -133,34 +156,64 @@ impl SnapshotStorage { .unwrap_or_default() } - /// Every stored commit of every tree, as the durable store carries them. + /// Every stored fragment of `tree` as `(head, blob)`. The blob is an + /// automerge *bundle* (or, on an app tree, an envelope around one) — + /// see `crate::document::Document::apply_bundles`. + pub fn fragment_blobs(&self, tree: SedimentreeId) -> Vec<(CommitId, Vec)> { + self.trees + .borrow() + .get(&tree) + .map(|t| { + t.fragments + .iter() + .map(|(head, (_signed, blob))| (*head, blob.clone())) + .collect() + }) + .unwrap_or_default() + } + + /// Whether `tree` already holds a fragment headed at `head`. What + /// compaction asks before building one: a fragment is identified by its + /// head, so a head we hold is a fragment we have already built or + /// received. + pub fn holds_fragment(&self, tree: SedimentreeId, head: CommitId) -> bool { + self.trees + .borrow() + .get(&tree) + .is_some_and(|t| t.fragments.contains_key(&head)) + } + + /// Every item of every tree, as the durable store carries them. /// /// What the durable store pushes (docs/design.md "Storage"): one object - /// per item, content-addressed by the pair that names it. Commits only — - /// this engine authors no fragments (nothing calls `add_fragments`, and - /// compaction is not implemented), so a fragment could only arrive from a - /// peer that had one, and there is none to have. + /// per item, content-addressed by the pair that names it. Fragments go up + /// beside commits — a device that pulls one gets the whole range in a + /// single object, which is the saving compaction exists for. pub fn all_items(&self) -> Vec { self.trees .borrow() .iter() .flat_map(|(tree, t)| { - t.commits - .iter() - .map(|(id, (signed, blob))| StoreItem { - tree: *tree.as_bytes(), - commit: *id.as_bytes(), - signed: signed.as_bytes().to_vec(), - blob: blob.clone(), - }) - .collect::>() + let commits = t.commits.iter().map(|(id, (signed, blob))| StoreItem { + tree: *tree.as_bytes(), + commit: *id.as_bytes(), + signed: signed.as_bytes().to_vec(), + blob: blob.clone(), + kind: ItemKind::Commit, + }); + let fragments = t.fragments.iter().map(|(head, (signed, blob))| StoreItem { + tree: *tree.as_bytes(), + commit: *head.as_bytes(), + signed: signed.as_bytes().to_vec(), + blob: blob.clone(), + kind: ItemKind::Fragment, + }); + commits.chain(fragments).collect::>() }) .collect() } - /// Whether `tree` already holds `commit`. What the push path asks before - /// uploading: the store is addressed by name, and a name it already has - /// is an object already written. + /// Whether `tree` holds `commit`. pub fn holds(&self, tree: SedimentreeId, commit: CommitId) -> bool { self.trees .borrow() @@ -168,6 +221,61 @@ impl SnapshotStorage { .is_some_and(|t| t.commits.contains_key(&commit)) } + /// Drop every loose commit of `tree` that a fragment we hold carries. + /// + /// The decision is sedimentree's own, not ours: + /// `Sedimentree::minimize(&CountLeadingZeroBytes)` keeps a loose commit + /// unless every range it belongs to is covered by a kept fragment + /// (sedimentree_core/src/sedimentree/commit_dag.rs:128). Anything it + /// keeps stays, so a commit concurrent with the range — a branch this + /// device could not see when it built the fragment — is never dropped. + /// + /// **The invariant this adds on top.** `minimize` is asked only about + /// *anchored* fragments: those whose every boundary id is the head of + /// another fragment we hold, plus those whose boundary is empty (they + /// reach the root). A fragment whose boundary hangs on a commit we hold + /// nothing for is a fragment nobody can walk below, and letting it + /// authorise a drop would strand the commits under it — readable in the + /// document, gone from the tree, and with no item left naming a way down. + /// That is not hypothetical once fragments arrive from peers, where a + /// middle one of a chain can turn up before the one under it. + /// + /// Fragments themselves are only ever added here, never removed. + /// `minimize` would also drop a level-1 fragment subsumed by a level-2 + /// one, and level-2 fragments do exist (one commit in 65 536) — not + /// dropping them is the conservative side of that trade: a redundant + /// item costs storage and a little sync chatter, and dropping one on a + /// judgement this code did not make could cost the range it carried. + /// + /// Answers how many commits went. + pub fn prune(&self, tree: SedimentreeId) -> usize { + let (commits, fragments) = self.metadata(tree); + let heads: BTreeSet = fragments.iter().map(Fragment::head).collect(); + let anchored: Vec = fragments + .into_iter() + .filter(|f| f.boundary().iter().all(|id| heads.contains(id))) + .collect(); + if anchored.is_empty() { + return 0; + } + let keep = Sedimentree::new(anchored, commits).minimize(&CountLeadingZeroBytes); + let kept: BTreeSet = keep.loose_commits().map(LooseCommit::head).collect(); + let mut trees = self.trees.borrow_mut(); + let Some(entry) = trees.get_mut(&tree) else { + return 0; + }; + let doomed: Vec = entry + .commits + .keys() + .filter(|id| !kept.contains(id)) + .copied() + .collect(); + for id in &doomed { + let _dropped = entry.commits.remove(id); + } + doomed.len() + } + /// The tree's decoded metadata, for `Handle::hydrate_tree` after a /// restore. Items that no longer decode are dropped: a resident tree /// missing an item re-syncs it, where a decode panic would lose the diff --git a/runtime/crates/engine/src/us.rs b/runtime/crates/engine/src/us.rs index 3c6b1f1b..3a197a96 100644 --- a/runtime/crates/engine/src/us.rs +++ b/runtime/crates/engine/src/us.rs @@ -20,7 +20,7 @@ //! that founder's, shared by construction. use automerge::{ObjType, ROOT, ReadDoc, transaction::Transactable}; -use sedimentree_core::id::SedimentreeId; +use sedimentree_core::{id::SedimentreeId, loose_commit::id::CommitId}; use sha2::{Digest as _, Sha256}; use subduction_protocol::command::NewCommit; @@ -122,6 +122,24 @@ impl UsDoc { self.core.absorb(storage) } + /// The fragments automerge draws over this document at level 1 and + /// deeper. See `crate::document::Document::fragments`. + pub fn fragments(&self) -> Vec { + self.core.fragments() + } + + /// The commits this document has applied — its own history, whether or + /// not the tree still holds the items that carried it. + pub fn applied_ids(&self) -> std::collections::BTreeSet { + self.core.applied_ids() + } + + /// The bundle bytes for those fragments. See + /// `crate::document::Document::bundle`. + pub fn bundle(&self, fragments: Vec) -> Vec> { + self.core.bundle(fragments) + } + pub fn last_local_commit(&mut self) -> Option { self.core.last_local_commit() } diff --git a/runtime/crates/engine/src/vault.rs b/runtime/crates/engine/src/vault.rs index 9a18ce0d..908eeec8 100644 --- a/runtime/crates/engine/src/vault.rs +++ b/runtime/crates/engine/src/vault.rs @@ -527,6 +527,20 @@ impl Vault { ); } + /// A fragment this device built and stored carries `members` bodily: its + /// bundle *is* their changes. So their individual keys stop being entry + /// points, exactly as a parent's does when a child's envelope names it — + /// the carrier here is the fragment rather than a descendant commit. + /// + /// Only sound because the fragment's own envelope is a head (its + /// [`Vault::confirm`] runs first) and names the *boundary* keys, so the + /// walk below the fragment continues where the members' envelopes would + /// have taken it. Called with the members and nothing else: a commit + /// outside the fragment is not carried by it and must keep its key. + pub fn cover(&self, members: impl Iterator) { + self.advance(&[], members); + } + /// Open as many of `blobs` as this device can. /// /// Two mechanisms, in order. The document's current epoch key opens diff --git a/runtime/crates/engine/tests/converge.rs b/runtime/crates/engine/tests/converge.rs index 320d23d9..23ceb3f7 100644 --- a/runtime/crates/engine/tests/converge.rs +++ b/runtime/crates/engine/tests/converge.rs @@ -11,8 +11,8 @@ use future_form::Local; use futures::future::LocalBoxFuture; use futures::{executor::LocalPool, task::LocalSpawnExt as _}; use polyvisor_engine::{ - AppState, Engine, EngineClock, EngineEvent, EngineNotify, LocalFuture, Snapshot, Spawner, - StoreItem, TaskSnapshot, TreeState, tasks_tree, + AppState, Engine, EngineClock, EngineEvent, EngineNotify, ItemKind, LocalFuture, Snapshot, + Spawner, StoreItem, TaskSnapshot, TreeState, tasks_tree, }; use subduction_protocol::event::Direction; use subduction_runtime::memory::transport::MemoryTransport; @@ -1104,3 +1104,301 @@ fn parents_delivered_after_their_child_still_open() { ); }); } + +// -- compaction --------------------------------------------------------------- + +/// Add tasks until automerge closes a level-1 fragment over the app tree. +/// +/// A commit heads a level-1 fragment when its hash starts with a zero byte, +/// so this is a geometric draw with p = 1/256 — about 256 mutations, and the +/// bound is generous rather than tuned. It is also the only way to reach the +/// case: the threshold is the hash's own, there is no knob, and picking +/// titles to hit it would be testing a rigged document. +async fn until_compacted(engine: &TestEngine) -> usize { + until_fragments(engine, 1).await +} + +/// Add tasks until the tree holds `n` fragments — a chain of ranges, each +/// one's boundary the previous one's head. +async fn until_fragments(engine: &TestEngine, n: usize) -> usize { + for written in 1..=8192 { + let _id = engine + .tasks_add(APP, format!("task {written}")) + .await + .unwrap(); + if engine + .items() + .iter() + .filter(|item| item.kind == ItemKind::Fragment) + .count() + >= n + { + return written; + } + } + panic!("8192 commits without {n} level-1 fragments; the depth metric moved"); +} + +fn kinds(engine: &TestEngine) -> (usize, usize) { + let items = engine.items(); + let fragments = items + .iter() + .filter(|item| item.kind == ItemKind::Fragment) + .count(); + (items.len() - fragments, fragments) +} + +#[test] +fn a_closed_range_becomes_one_fragment_and_the_commits_it_carries_go() { + let mut pool = LocalPool::new(); + let a = device(&pool, 20, None); + let ea = Rc::clone(&a.engine); + + pool.run_until(async move { + let before = ea.entry_points().await.unwrap(); + // A few of the commits this run will later prune, captured as the + // durable store would have them. + let _first = ea.tasks_add(APP, "first".into()).await.unwrap(); + let early = ea.items(); + let written = 1 + until_compacted(&ea).await; + let (_commits, fragments) = kinds(&ea); + assert_eq!(fragments, 1, "one level-1 fragment closed"); + assert!( + !ea.read_not_held().is_empty(), + "the pruned commits are still changes this device has read", + ); + // Every task is still readable — the document keeps its full history + // whatever the tree drops. + assert_eq!(ea.tasks_items(APP).await.unwrap().items.len(), written); + + // The app tree is now the fragment and whatever was written after it + // closed; the loose commits left in `items()` belong to the group and + // keyhive trees, which have no fragment. + // The saving, stated as the inequality it is rather than as a + // predicted number: how many commits one fragment covers is the hash + // draw's business. + let tree = *tasks_tree(APP).as_bytes(); + let loose = ea + .items() + .iter() + .filter(|item| item.tree == tree && item.kind == ItemKind::Commit) + .count(); + assert!( + loose < written, + "the fragment's range left storage: {loose} loose commits from {written} mutations", + ); + + // The store deletes nothing, so those objects are still under their + // names and the next pull will hand them straight back. A commit this + // device pruned on purpose is not news, and reinstating it would undo + // the compaction on every pass, forever. + assert!( + !ea.ingest_items(early).await.unwrap(), + "commits pruned by compaction are not reinstalled from the store", + ); + + let after = ea.entry_points().await.unwrap(); + eprintln!("PROBE2 written={written} before={before} after={after}"); + assert!( + after <= before + 1, + "compaction adds at most the fragment's own entry point: {before} -> {after}", + ); + }); +} + +#[test] +fn a_compacted_tree_restores_from_its_checkpoint() { + // The checkpoint carries the fragment and the loose commits that survived + // it, and nothing else — the covered range's bytes are gone. Restoring + // has to read the whole list back out of the bundle. + let mut pool = LocalPool::new(); + let a = device(&pool, 21, None); + let ea = Rc::clone(&a.engine); + let (snapshot, written) = pool.run_until(async move { + let written = until_compacted(&ea).await; + (ea.snapshot().await.unwrap(), written) + }); + + let mut pool = LocalPool::new(); + let restored = device(&pool, 21, Some(snapshot)); + let engine = Rc::clone(&restored.engine); + let items = pool.run_until(async move { engine.tasks_items(APP).await.unwrap() }); + assert_eq!( + items.items.len(), + written, + "the restored device reads the compacted range", + ); +} + +#[test] +fn a_device_enrolled_after_compaction_reads_the_range_from_the_fragment() { + // The read-back case, at range scale. B is enrolled *after* the fragment + // was sealed, so it never held the epoch keys the covered commits were + // written under — and their envelopes are not in A's storage to send any + // more. What reaches B is the fragment: one envelope, sealed under the + // group's current epoch, carrying every change of the range. + let mut pool = LocalPool::new(); + let a = device(&pool, 22, None); + let b = device(&pool, 23, None); + let (ea, eb) = (Rc::clone(&a.engine), Rc::clone(&b.engine)); + + pool.run_until(async move { + let written = until_compacted(&ea).await; + assert!(eb.tasks_items(APP).await.unwrap().items.is_empty()); + wire(&ea, &eb).await; + + let seen = until(|| async { + let items = eb.tasks_items(APP).await.unwrap(); + (items.items.len() == written).then_some(items) + }) + .await; + assert_eq!(seen.items.len(), written); + assert!( + eb.items() + .iter() + .any(|item| item.kind == ItemKind::Fragment), + "B holds the fragment itself, not the range unrolled into commits", + ); + }); +} + +#[test] +fn two_devices_build_the_same_fragment_from_the_same_history() { + // Identity is head + boundary (`design/sedimentree.md`), and both are + // functions of the change graph — so a second device holding the same + // history builds a fragment the first one's tree already has, and adding + // it is a no-op locally. On the store the two land under one name and the + // last write wins; what makes that harmless is asserted here, that they + // are the same fragment. + let mut pool = LocalPool::new(); + let a = device(&pool, 24, None); + let ea = Rc::clone(&a.engine); + let snapshot = pool.run_until(async move { + let _written = until_compacted(&ea).await; + ea.snapshot().await.unwrap() + }); + let mine = snapshot + .apps + .iter() + .find(|app| app.app == APP) + .expect("the app was compacted") + .state + .fragments + .clone(); + assert_eq!(mine.len(), 1); + + // The same document, on a device whose tree has never seen the fragment: + // the commits it covered are gone with it, which is exactly the state a + // rebuild has to work from. + let mut naked = snapshot.clone(); + for app in &mut naked.apps { + app.state.fragments.clear(); + } + + let mut pool = LocalPool::new(); + let rebuilt = device(&pool, 24, Some(naked)); + let engine = Rc::clone(&rebuilt.engine); + let items = pool.run_until(async move { + // Compaction runs on a mutation; one more task is the cheapest + // trigger, and a level-0 commit on top does not move the level-1 + // fragment underneath it. + let _id = engine.tasks_add(APP, "one more".into()).await.unwrap(); + engine.items() + }); + let theirs: Vec<&StoreItem> = items + .iter() + .filter(|item| item.kind == ItemKind::Fragment) + .collect(); + assert_eq!(theirs.len(), 1, "the same one fragment, rebuilt"); + let signed = + subduction_crypto::signed::Signed::::try_decode( + &mine[0].signed, + ) + .expect("the checkpoint's fragment decodes"); + let original = signed + .try_decode_trusted_payload() + .expect("and its payload does"); + let signed = + subduction_crypto::signed::Signed::::try_decode( + &theirs[0].signed, + ) + .expect("the rebuilt fragment decodes"); + let rebuilt = signed + .try_decode_trusted_payload() + .expect("and its payload does"); + assert_eq!(original.head(), rebuilt.head(), "same head"); + assert_eq!( + original.boundary(), + rebuilt.boundary(), + "same boundary — the two are the same fragment", + ); + // The envelopes are byte-identical here, which the design did not + // predict: keyhive derives the content key and nonce from the group's + // epoch key and the payload rather than from fresh randomness, so two + // devices of one group seal one plaintext to one ciphertext. It is not + // asserted, because nothing in the contract promises it — the sealed + // plaintext is a `bincode` `Envelope` whose `ancestors` is a `HashMap`, + // and two devices with the same ancestors in a different iteration order + // would produce different bytes. Both cases are fine, and for the same + // reason: the two objects decrypt to the same range. +} + +#[test] +fn a_chain_of_fragments_is_still_one_entry_point() { + // Two ranges, so the second fragment's boundary is the first fragment's + // head — the case where naming the boundary *commit* would embed nothing + // (its envelope was pruned with its range and its key left the frontier + // when the first fragment covered it). What the second envelope names is + // the first *fragment*, and that is what a joiner walks down. + // + // So: the head set does not grow one entry per fragment, and a device + // enrolled after both were sealed reads both ranges from the single + // entry point it was handed. + let mut pool = LocalPool::new(); + let a = device(&pool, 25, None); + let b = device(&pool, 26, None); + let (ea, eb) = (Rc::clone(&a.engine), Rc::clone(&b.engine)); + + pool.run_until(async move { + // One entry point before any compaction: a linear history is one + // readable branch (docs/design.md §"Read-back and partitions"). + let _first = ea.tasks_add(APP, "first".into()).await.unwrap(); + let before = ea.entry_points().await.unwrap(); + let written = 1 + until_fragments(&ea, 2).await; + let (_commits, fragments) = kinds(&ea); + assert_eq!(fragments, 2, "two fragments, chained"); + // The loop stops on the commit that closed the second fragment, so + // that commit is a member and its key is covered: what is left is the + // fragment chain's single newest entry. Naming the boundary *commit* + // rather than the boundary *fragment* would leave the first fragment + // a head too, and this reads 2. + let after = ea.entry_points().await.unwrap(); + assert!( + after <= before, + "the head set does not grow one entry per fragment: {before} -> {after}", + ); + + // B is enrolled now — after both ranges were sealed, under epochs it + // never held. + assert!(eb.tasks_items(APP).await.unwrap().items.is_empty()); + wire(&ea, &eb).await; + let seen = until(|| async { + let items = eb.tasks_items(APP).await.unwrap(); + (items.items.len() == written).then_some(items) + }) + .await; + assert_eq!( + seen.items.len(), + written, + "the joiner read both ranges, walking from the newest fragment down", + ); + assert_eq!( + eb.items() + .iter() + .filter(|item| item.kind == ItemKind::Fragment) + .count(), + 2, + "and holds them as fragments, not as the ranges unrolled", + ); + }); +} diff --git a/runtime/crates/kernel/src/drive.rs b/runtime/crates/kernel/src/drive.rs index c4391b86..ac219c8b 100644 --- a/runtime/crates/kernel/src/drive.rs +++ b/runtime/crates/kernel/src/drive.rs @@ -57,7 +57,7 @@ use std::rc::Rc; use data_encoding::{BASE64URL_NOPAD, HEXLOWER}; use hmac::{Hmac, Mac as _}; -use polyvisor_engine::StoreItem; +use polyvisor_engine::{ItemKind, StoreItem}; use serde::{Deserialize, Serialize}; use sha2::{Digest as _, Sha256}; @@ -270,6 +270,23 @@ impl Drive { // -- the exports ------------------------------------------------------------- impl Kernel { + /// Whether this device holds a sedimentree fragment yet — a commit range + /// rolled up into one item (`docs/design.md` §"Read-back and partitions"). + /// + /// Test introspection, in the shape of `Engine::live_connections`: which + /// commit closes a fragment is the hash's decision, so a test that wants + /// the compacted case has to write until one appears and cannot predict + /// the number. Nothing in the WIT world reads this. + #[must_use] + pub fn holds_a_fragment(&self) -> bool { + self.engine().is_ok_and(|engine| { + engine + .items() + .iter() + .any(|item| item.kind == ItemKind::Fragment) + }) + } + /// `storage.status`. pub fn storage_status(&self) -> Result { self.open()?; @@ -533,7 +550,18 @@ impl Kernel { let mine: BTreeSet = engine .items() .iter() - .map(|item| object_name(name_key, &item.tree, &item.commit)) + .map(|item| object_name(name_key, &item.tree, &item.commit, item.kind)) + // And the objects for changes this device has read but does not + // hold as items — a range a fragment carries instead + // (`Engine::read_not_held`). Nothing deletes them from the store, + // so without this the pull would fetch the whole compacted range + // back on every single pass, to be refused every time. + .chain( + engine + .read_not_held() + .iter() + .map(|(tree, commit)| object_name(name_key, tree, commit, ItemKind::Commit)), + ) .collect(); let mut fetched = Vec::new(); for (id, name) in &remote { @@ -590,7 +618,7 @@ impl Kernel { let present: BTreeSet<&str> = remote.iter().map(|(_, name)| name.as_str()).collect(); let mut pushed = false; for item in engine.items() { - let name = object_name(name_key, &item.tree, &item.commit); + let name = object_name(name_key, &item.tree, &item.commit, item.kind); if present.contains(name.as_str()) { continue; } @@ -1030,7 +1058,8 @@ fn json(response: &HttpResponse, what: &str) -> Result; -/// `hex(HMAC-SHA256(name_key, tree ‖ commit))` — the object's name. +/// `hex(HMAC-SHA256(name_key, tree ‖ item id [‖ "fragment"]))` — the object's +/// name. /// /// Derived rather than descriptive, and that is the point (docs/design.md /// "Storage"): a tree id is global and stable, so a plain name would tell @@ -1041,11 +1070,26 @@ type HmacSha256 = Hmac; /// It is also the deduplication: the name is a function of the item's own /// digests, so an object that exists under it is already these bytes and the /// push skips it. +/// +/// Which is exactly why the kind is mixed in for a fragment. A fragment is +/// named by its *head*, and that head is also a commit — two different +/// payloads for one `(tree, id)` pair. Left undistinguished they would race +/// for one name and the dedup would silently keep whichever landed first. The +/// commit case is left byte-for-byte as it was, so objects a group wrote +/// before fragments existed keep their names. #[must_use] -pub fn object_name(name_key: &[u8; 32], tree: &[u8; 32], commit: &[u8; 32]) -> String { +pub fn object_name( + name_key: &[u8; 32], + tree: &[u8; 32], + commit: &[u8; 32], + kind: ItemKind, +) -> String { let mut mac = HmacSha256::new_from_slice(name_key).expect("HMAC takes a key of any length"); mac.update(tree); mac.update(commit); + if kind == ItemKind::Fragment { + mac.update(b"fragment"); + } HEXLOWER.encode(&mac.finalize().into_bytes()) } @@ -1110,19 +1154,22 @@ mod tests { let key = [3u8; 32]; let other = [4u8; 32]; let (tree, commit) = ([1u8; 32], [2u8; 32]); - let name = object_name(&key, &tree, &commit); + let name = object_name(&key, &tree, &commit, ItemKind::Commit); assert_eq!(name.len(), 64, "hex of an HMAC-SHA256"); assert_eq!( name, - object_name(&key, &tree, &commit), + object_name(&key, &tree, &commit, ItemKind::Commit), "derived, not drawn" ); // A different group derives a different name for the same item, which // is what keeps two accounts' stores uncorrelatable. - assert_ne!(name, object_name(&other, &tree, &commit)); + assert_ne!(name, object_name(&other, &tree, &commit, ItemKind::Commit)); // The tree and the commit are separate inputs, not one concatenated // blob a shift could confuse. - assert_ne!(name, object_name(&key, &commit, &tree)); + assert_ne!(name, object_name(&key, &commit, &tree, ItemKind::Commit)); + // A fragment is named by its head, and that head is also a commit: + // the two must not land on one object. + assert_ne!(name, object_name(&key, &tree, &commit, ItemKind::Fragment)); assert_ne!(folder_name(&key), folder_name(&other)); assert!(folder_name(&key).starts_with("polyvisor-")); assert_eq!(folder_name(&key).len(), "polyvisor-".len() + 16); diff --git a/runtime/crates/kernel/tests/kernel.rs b/runtime/crates/kernel/tests/kernel.rs index 7f215a68..013f7a9b 100644 --- a/runtime/crates/kernel/tests/kernel.rs +++ b/runtime/crates/kernel/tests/kernel.rs @@ -3529,3 +3529,70 @@ fn a_device_that_joins_a_group_stops_reading_its_old_folder() { }); assert!(seen.contains(&"from the adder".to_string())); } + +#[test] +fn a_compacted_range_reaches_the_store_as_one_object() { + // Compaction seen from the store. A writes — with no store bound, so + // nothing has been pushed yet — until automerge closes a level-1 fragment + // (about one commit in 256; the threshold is the hash's own, so the loop + // is a draw and the bound is generous). Only then does it bind the store, + // and what goes up is the fragment plus the handful of commits left loose + // after it, not one object per mutation. + // + // Which is the whole shape of the saving, and the reason it is tested + // this way round: nothing deletes from the store (docs/design.md + // "Read-back and partitions"), so a device that had been pushing all + // along would leave its covered objects behind — correct, and no smaller. + let drive = FakeDrive::shared(); + let here = World::default().with_drive(&drive); + let there = here.peer().with_drive(&drive); + let a = here.boot(); + let b = there.boot(); + let (sa, sb) = (session(&a), session(&b)); + settle(); + + // Paired, so B holds A's group and A's store-name key. + let _sas = pair(&b, &a); + settle(); + // Cut the network: everything B learns from here came through the store. + let (ida, idb) = ( + a.device_status().unwrap().endpoint_id, + b.device_status().unwrap().endpoint_id, + ); + here.net.unplug(&ida); + here.net.unplug(&idb); + settle(); + + // A fragment is one commit in 256, so the run is bounded well above the + // mean rather than at it; a run that reached the bound would mean the + // depth metric moved. + let mut written = 0usize; + for n in 1..=4096 { + block_on(a.tasks_add(sa, format!("task {n}"))).unwrap(); + written = n; + if a.holds_a_fragment() { + break; + } + } + assert!(written < 4096, "no level-1 fragment in 4096 commits"); + + connect_store(&a); + settle(); + let objects = drive.objects().len(); + assert!( + objects < written, + "the range went up as one object, not {written}: {objects} in the folder", + ); + + connect_store(&b); + let seen = settle_until(|| async { + let _synced = b.sync_now().await; + let seen = titles(&b, sb).await; + (seen.len() == written).then_some(seen) + }); + assert_eq!( + seen.len(), + written, + "B read the whole range out of the store" + ); +}