From 2ddc3f199edf30afdecf46fa08b537e824c4adb1 Mon Sep 17 00:00:00 2001 From: Lann Martin Date: Mon, 7 Sep 2026 14:47:45 -0400 Subject: [PATCH] A joiner serves the history it adopted as a fragment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit adopt_us installs the adder's group document as an automerge snapshot with an empty sedimentree tree, so the joiner held history it could not serve as items — to a device that syncs only with it, or to the store. A member holding applied history serves it the way compaction does: it builds a fragment. After adoption the joiner gives the document a single head (a signed merge anchor if needed) and builds one plaintext fragment over the whole adopted history, boundary empty, through the same install path compact uses. A third device enrolled by the joiner learns the first era from that fragment without ever meeting the founder. --- runtime/crates/engine/src/document.rs | 17 ++ runtime/crates/engine/src/lib.rs | 248 +++++++++++++++++------- runtime/crates/engine/src/us.rs | 16 ++ runtime/crates/engine/tests/converge.rs | 95 ++++++++- runtime/crates/kernel/src/drive.rs | 28 ++- runtime/crates/kernel/tests/kernel.rs | 67 ++++++- 6 files changed, 388 insertions(+), 83 deletions(-) diff --git a/runtime/crates/engine/src/document.rs b/runtime/crates/engine/src/document.rs index 1f549c5d..c2260241 100644 --- a/runtime/crates/engine/src/document.rs +++ b/runtime/crates/engine/src/document.rs @@ -232,6 +232,23 @@ impl Document { self.doc.fragments(1..) } + /// The document's current heads, and every change hash in its history. + /// + /// For the one caller that has to describe a whole document as a single + /// sedimentree fragment rather than take automerge's own partition of it + /// (`crate::Engine::adopt_fragment`). + pub fn heads(&self) -> Vec { + self.doc.get_heads() + } + + pub fn change_hashes(&self) -> Vec { + self.doc + .get_changes(&[]) + .iter() + .map(automerge::Change::hash) + .collect() + } + /// 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 diff --git a/runtime/crates/engine/src/lib.rs b/runtime/crates/engine/src/lib.rs index 0ea84aa8..e3a5a18c 100644 --- a/runtime/crates/engine/src/lib.rs +++ b/runtime/crates/engine/src/lib.rs @@ -477,7 +477,90 @@ impl + 'static> Engine { .await .map_err(|e| e.to_string())?; } - Ok(()) + self.adopt_fragment().await + } + + /// Make the adopted group document servable: one sedimentree fragment + /// over the whole of it. + /// + /// [`Engine::adopt_us`] installs an automerge *snapshot* — the adder's + /// document, whole — and empties the tree that used to back it. That + /// leaves this device holding history it can read and cannot hand to + /// anyone: no loose commit of the group's past is in its tree, so a third + /// device that syncs only with this one, or a store this one pushes to, + /// gets nothing of the era before the pairing. The pull will not fetch + /// those objects back either, and should not — the changes are already in + /// the document (`Engine::read_not_held`). + /// + /// A fragment is exactly the item sedimentree has for this: a range of + /// history carried as one blob by a member that can read the whole range. + /// This device can — it just adopted it — so it builds one, with an empty + /// boundary because the range reaches the root, and every change as a + /// checkpoint because the fragment covers all of them + /// (`Fragment::supports_block` answers coverage from head, checkpoints + /// and boundary; `Fragment::new` truncates the checkpoints to 12 bytes). + /// + /// Identity is head plus boundary, so the adder — holding the same graph + /// — builds the same fragment if it ever takes this path, and a deeper + /// one automerge draws later subsumes neither: both are correct items + /// over the same changes. + async fn adopt_fragment(&self) -> Result<(), String> { + let tree = us_tree(); + // A sedimentree fragment has one head; an automerge document may + // have several. Where it does, the document is given one the way the + // absorb path gives a partitioned app tree one: an empty change + // depending on every current head. It goes in as a real signed `us` + // commit rather than a bare automerge change, because that is what + // every other device will read it as. + if self.with_us(UsDoc::heads).len() > 1 { + let anchor = { + let mut cell = self.us.borrow_mut(); + let doc = cell + .as_mut() + .ok_or_else(|| "this device has no group document".to_string())?; + doc.merge_anchor() + }; + self.push_us_commit(anchor).await?; + } + let (heads, members) = self.with_us(|doc| (doc.heads(), doc.change_hashes())); + // One head, or none at all: a document with no changes has no + // history to serve, which is not this function's problem to report. + let (Some(head), 1) = (heads.first().copied(), heads.len()) else { + return Ok(()); + }; + if self.storage.holds_fragment(tree, CommitId::new(head.0)) { + return Ok(()); + } + let fragment = automerge::Fragment { + head, + // What `ChangeHash::fragment_level` would say — leading zero + // bytes, the same metric sedimentree's `CountLeadingZeroBytes` + // uses (automerge types.rs:680). It is not the *stratum* this + // fragment sits at, which sedimentree computes from the head + // itself; the one thing automerge reads it for is whether a + // one-member fragment may be encoded as a bare change rather + // than a bundle, and both decode through `load_incremental`. + level: head.0.iter().take_while(|byte| **byte == 0).count(), + boundary: Vec::new(), + checkpoints: members + .iter() + .filter(|hash| **hash != head) + .copied() + .collect(), + members: members.clone(), + }; + let Some(bundle) = self + .with_us(|doc| doc.bundle(vec![fragment.clone()])) + .into_iter() + .next() + else { + return Ok(()); + }; + // Plaintext, like every other item of this tree (`crate::vault` + // module docs: the group document is what tells a device who its + // group is, and it cannot be sealed to a key that knowledge is + // needed to derive). + self.install_fragment(tree, &fragment, bundle, false).await } // -- connections --------------------------------------------------------- @@ -1202,80 +1285,101 @@ impl + 'static> Engine { 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 + self.install_fragment(tree, &fragment, bundle, enveloped) + .await?; + } + Ok(()) + } + + /// Put one fragment into the tree: seal it if the tree is enveloped, hand + /// it to the driver, wait for it to be durable, move the frontier, and + /// drop the loose commits it now carries. + /// + /// Shared by [`Engine::compact`], which builds fragments over ranges + /// automerge closed on their own, and by [`Engine::adopt_fragment`], + /// which builds one over a whole adopted history. The two differ only in + /// where the `automerge::Fragment` came from; everything after that — + /// including the ordering, which is what makes the drop safe — is this. + async fn install_fragment( + &self, + tree: SedimentreeId, + fragment: &automerge::Fragment, + bundle: Vec, + enveloped: bool, + ) -> Result<(), String> { + 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(|hash| CommitId::new(hash.0)) + .map(|id| { + if self.storage.holds_fragment(tree, *id) { + fragment_cref(tree, *id) + } else { + *id.as_bytes() + } + }) .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); + 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(()) } diff --git a/runtime/crates/engine/src/us.rs b/runtime/crates/engine/src/us.rs index 3a197a96..40f8b2fe 100644 --- a/runtime/crates/engine/src/us.rs +++ b/runtime/crates/engine/src/us.rs @@ -134,6 +134,22 @@ impl UsDoc { self.core.applied_ids() } + /// This document's heads, and every change hash in its history. See + /// `crate::document::Document::heads`. + pub fn heads(&self) -> Vec { + self.core.heads() + } + + pub fn change_hashes(&self) -> Vec { + self.core.change_hashes() + } + + /// An empty change depending on every current head — automerge's own + /// merge commit. See `crate::document::Document::merge_anchor`. + pub fn merge_anchor(&mut self) -> Option { + self.core.merge_anchor() + } + /// The bundle bytes for those fragments. See /// `crate::document::Document::bundle`. pub fn bundle(&self, fragments: Vec) -> Vec> { diff --git a/runtime/crates/engine/tests/converge.rs b/runtime/crates/engine/tests/converge.rs index 23ceb3f7..6e790969 100644 --- a/runtime/crates/engine/tests/converge.rs +++ b/runtime/crates/engine/tests/converge.rs @@ -1392,13 +1392,106 @@ fn a_chain_of_fragments_is_still_one_entry_point() { written, "the joiner read both ranges, walking from the newest fragment down", ); + // Scoped to the app tree: B also holds a fragment over the group + // document it adopted at enrollment (`Engine::adopt_fragment`). + let app = *tasks_tree(APP).as_bytes(); assert_eq!( eb.items() .iter() - .filter(|item| item.kind == ItemKind::Fragment) + .filter(|item| item.tree == app && item.kind == ItemKind::Fragment) .count(), 2, "and holds them as fragments, not as the ranges unrolled", ); }); } + +// -- the adopted group document, as an item ----------------------------------- + +/// The `us` tree's items, as (loose commits, fragments). +fn us_items(engine: &TestEngine) -> (usize, usize) { + let tree = *polyvisor_engine::us_tree().as_bytes(); + let mut commits = 0; + let mut fragments = 0; + for item in engine.items().iter().filter(|item| item.tree == tree) { + match item.kind { + ItemKind::Commit => commits += 1, + ItemKind::Fragment => fragments += 1, + } + } + (commits, fragments) +} + +#[test] +fn a_joiner_can_serve_the_history_it_adopted() { + // `adopt_us` installs the adder's document whole and empties the tree + // that used to back it, so without this the joiner would hold the + // group's whole past as something it can read and cannot hand to anyone + // — no item of that era is in its tree, and the pull will not fetch the + // objects back because the changes are already in its document. + let mut pool = LocalPool::new(); + let a = device(&pool, 30, None); + let b = device(&pool, 31, None); + let (ea, eb) = (Rc::clone(&a.engine), Rc::clone(&b.engine)); + + let (snapshot, before) = pool.run_until(async move { + // Some group history to adopt: a second member, and the keyhive + // pointer the founder writes. + wire(&ea, &eb).await; + let (_commits, fragments) = us_items(&eb); + assert_eq!(fragments, 1, "the joiner rolled the adopted history up"); + (eb.snapshot().await.unwrap(), members(&eb).await) + }); + assert_eq!(before.len(), 2, "the adopted group is the adder's, plus us"); + + // The document blanked, the tree kept: what a device holds if it has the + // items and nothing else. If the fragment's bundle is a real automerge + // bundle of the adopted history, the group comes back from it alone. + let mut naked = snapshot; + if let Some(us) = naked.us.as_mut() { + us.doc.clear(); + } + let mut pool = LocalPool::new(); + let restored = device(&pool, 31, Some(naked)); + let engine = Rc::clone(&restored.engine); + let after = pool.run_until(async move { members(&engine).await }); + assert_eq!( + after, before, + "the group came back out of the fragment, with no document to help", + ); +} + +#[test] +fn a_third_device_learns_the_first_era_from_the_second() { + // A enrols B; B enrols C; C never meets A. Everything C learns of the + // group's first era — the era before B existed — comes from B, and B + // holds it as the one fragment it built when it adopted. + let mut pool = LocalPool::new(); + let a = device(&pool, 32, None); + let b = device(&pool, 33, None); + let c = device(&pool, 34, None); + let (ea, eb, ec) = ( + Rc::clone(&a.engine), + Rc::clone(&b.engine), + Rc::clone(&c.engine), + ); + + pool.run_until(async move { + wire(&ea, &eb).await; + // B is now a member and has A's history as a fragment. C pairs with + // B, and B — not A — is the only device it is ever wired to. + wire(&eb, &ec).await; + + let seen = until(|| async { + let seen = members(&ec).await; + (seen.len() == 3).then_some(seen) + }) + .await; + assert_eq!(seen.len(), 3, "C sees the whole group, A included"); + let (_commits, fragments) = us_items(&ec); + assert!( + fragments >= 1, + "and holds that history as a fragment of its own", + ); + }); +} diff --git a/runtime/crates/kernel/src/drive.rs b/runtime/crates/kernel/src/drive.rs index ac219c8b..5dbe20d4 100644 --- a/runtime/crates/kernel/src/drive.rs +++ b/runtime/crates/kernel/src/drive.rs @@ -270,20 +270,30 @@ 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"). + /// How many sedimentree fragments this device holds over one tree — the + /// app's, or (`None`) the group document's. A fragment is a commit range + /// carried as 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. + /// 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. + /// + /// Per-tree because the two have nothing to do with each other: an app's + /// fragments arrive on the hash's schedule, while the group document gets + /// exactly one the moment a device adopts it (`Engine::adopt_fragment`). #[must_use] - pub fn holds_a_fragment(&self) -> bool { - self.engine().is_ok_and(|engine| { + pub fn fragments_held(&self, app: Option<&str>) -> usize { + let tree = match app { + Some(app) => *polyvisor_engine::tasks_tree(app).as_bytes(), + None => *polyvisor_engine::us_tree().as_bytes(), + }; + self.engine().map_or(0, |engine| { engine .items() .iter() - .any(|item| item.kind == ItemKind::Fragment) + .filter(|item| item.tree == tree && item.kind == ItemKind::Fragment) + .count() }) } diff --git a/runtime/crates/kernel/tests/kernel.rs b/runtime/crates/kernel/tests/kernel.rs index 7b989214..dd6c5a8c 100644 --- a/runtime/crates/kernel/tests/kernel.rs +++ b/runtime/crates/kernel/tests/kernel.rs @@ -775,6 +775,22 @@ impl FakeDrive { self.refresh.borrow_mut().clear(); } + /// How many objects in the store are sedimentree *fragments*. Decoded + /// rather than matched by name: an object's name is an HMAC of the item + /// id, and a fragment's id is whatever automerge's change hash happened + /// to be. + fn fragments_in_store(&self) -> usize { + self.files + .borrow() + .values() + .filter(|file| !file.folder) + .filter(|file| { + serde_json::from_slice::(&file.body) + .is_ok_and(|item| item.kind == polyvisor_engine::ItemKind::Fragment) + }) + .count() + } + /// The names of every object in the one folder this group writes to. fn objects(&self) -> Vec { self.files @@ -3624,7 +3640,7 @@ fn a_compacted_range_reaches_the_store_as_one_object() { for n in 1..=4096 { block_on(a.tasks_add(sa, format!("task {n}"))).unwrap(); written = n; - if a.holds_a_fragment() { + if a.fragments_held(Some("todomvc")) > 0 { break; } } @@ -3650,3 +3666,52 @@ fn a_compacted_range_reaches_the_store_as_one_object() { "B read the whole range out of the store" ); } + +#[test] +fn the_joiner_publishes_the_history_it_adopted() { + // Pairing hands the joiner the group document whole and leaves its tree + // empty, so the era before the joiner existed is history it can read + // and — until it rolls that history into a fragment of its own — cannot + // hand to anybody. Both "anybodies" are here: the adder, which takes the + // fragment over the wire, and the store, where the joiner's push is what + // puts the group's first era into Drive under a name the joiner derived. + 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(); + + let _sas = pair(&b, &a); + settle(); + assert_eq!( + b.fragments_held(None), + 1, + "the joiner rolled the history it adopted into one fragment", + ); + // The other half of the problem: it is an *item*, so it syncs. The adder + // — which built nothing, having adopted nothing — ends up holding the + // joiner's fragment beside its own loose commits. + assert_eq!( + a.fragments_held(None), + 1, + "and it reached the adder as an item, over the wire", + ); + + connect_store(&b); + settle(); + // Found by decoding rather than by name: an object's name is an HMAC of + // the item id, and a fragment's id is whatever automerge's change hash + // happened to be. + assert_eq!( + drive.fragments_in_store(), + 1, + "and the joiner published it, so the store holds that era too", + ); + + // Both devices still see the same two-device group, which is the thing + // all of this is carrying. + assert_eq!(block_on(a.sync_members()).unwrap().len(), 2); + assert_eq!(block_on(b.sync_members()).unwrap().len(), 2); +}