From 504fe438a54acdc5f049f30cb0b894957e534be2 Mon Sep 17 00:00:00 2001 From: Lann Martin Date: Mon, 7 Sep 2026 11:29:40 -0400 Subject: [PATCH] Read-back as a head set; merge anchors; Drive pulls before pushing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per keyhive's causal-encryption record, the entry point to a document is a set of (head, content key) pairs, not every key ever held: chunk_keys becomes that head set, moved by seal and by open — parents' keys ride inside their descendants' envelopes, so an entry is pruned once a descendant this device holds carries it. Keys learned for commits that have not arrived yet are retained (one entry per out-of-order arrival) because nothing else would carry them. The frontier advances only after the commit is in the sedimentree. A concurrent branch sealed under an older epoch is a partition, not a loss: when a batch lands content that leaves the document diverged, the engine writes one empty automerge commit — a merge anchor whose envelope carries both branches' keys under the current epoch — so members who joined after that epoch read it without anyone writing again. An anchor carries no operations and so never triggers another; concurrent anchors rest as two heads until the next real write. A linear old-epoch write (no divergence) stays dark for a later member until the next local mutation; recorded as the known gap. Drive syncs pull → ingest → push. Design record: 'Read-back and partitions'. --- docs/design.md | 38 ++++ runtime/crates/engine/src/doc.rs | 18 +- runtime/crates/engine/src/document.rs | 67 +++++- runtime/crates/engine/src/lib.rs | 85 ++++++-- runtime/crates/engine/src/vault.rs | 263 +++++++++++++++++++----- runtime/crates/engine/tests/converge.rs | 237 ++++++++++++++++++++- runtime/crates/kernel/src/drive.rs | 98 +++++---- 7 files changed, 694 insertions(+), 112 deletions(-) diff --git a/docs/design.md b/docs/design.md index eb36d11c..54b94d97 100644 --- a/docs/design.md +++ b/docs/design.md @@ -202,6 +202,44 @@ keyhive is pinned to git `main` for the CGKA transitive-authority fix (66a6632: relay access no longer grants CGKA membership) that the released 0.5.0 lacks; the pull ≠ read tier separation depends on it. +## Read-back and partitions + +App-tree envelopes are causal: keyhive's premise is that granting an +entry point to a document at a point in history reveals the whole +history behind it, so each envelope carries the content keys of its own +direct causal ancestors (keyhive `design/causal_encryption.md`, §"Key +Management"). What a device therefore keeps — and checkpoints, and hands +to the next device it enrols — is not a key per commit but a *set of +heads*: one ⟨pointer, key⟩ pair per readable branch, from which +everything prior is discovered by following the ancestor keys inside +each envelope (§"Decryption Head"). The design doc is blunt that the +full map is "possible, but fragile and unwieldy"; the head set grows +with live concurrency instead of with history, so a linear history of +any length is one pair. A commit whose key a device does not hold is not +lost, it is *under partition* — latency, in keyhive's framing — and it +is connected "by supplying a new head for it" (§"Multiple Heads"). That +is what the engine does on absorb: when a batch of content lands +concurrent with what a device already had, it writes an automerge merge +commit — an empty change whose dependencies are every current head — +whose envelope names both branches' keys and is sealed under the group's +current epoch. A member enrolled after the branch was written, who could +decrypt neither it nor anything automerge buffers behind it, walks in +from that anchor. Anchors are content-free, and a batch that is nothing +but somebody else's anchor is not a reason to author another, which is +what keeps two devices from anchoring each other forever. For the same +reason the Drive pass pulls before it pushes: a device coming back +online learns the group's current epoch and merges before it publishes, +so writes that are concurrent with the group's go up alongside the +anchor that names their keys. That does not close the linear case — a +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. + ## Storage The store is dumb and untrusted: it holds ciphertext at unguessable diff --git a/runtime/crates/engine/src/doc.rs b/runtime/crates/engine/src/doc.rs index 6760c96c..105a7f3d 100644 --- a/runtime/crates/engine/src/doc.rs +++ b/runtime/crates/engine/src/doc.rs @@ -21,7 +21,7 @@ use sedimentree_core::{id::SedimentreeId, loose_commit::id::CommitId}; use serde::{Deserialize, Serialize}; use subduction_protocol::command::NewCommit; -use crate::document::{Document, actor}; +use crate::document::{Absorbed, Document, actor}; use crate::storage::SnapshotStorage; /// `polyvisor:app/tasks.todo-item`. @@ -213,11 +213,23 @@ impl AppDoc { self.core.unapplied(storage) } - /// Apply decrypted automerge changes. Returns whether anything landed. - pub fn apply(&mut self, items: Vec<(CommitId, Vec)>) -> bool { + /// Apply decrypted automerge changes. + pub fn apply(&mut self, items: Vec<(CommitId, Vec)>) -> Absorbed { self.core.apply(items) } + pub fn applied_ids(&self) -> std::collections::BTreeSet { + self.core.applied_ids() + } + + pub fn diverged(&self) -> bool { + self.core.diverged() + } + + pub fn merge_anchor(&mut self) -> Option { + self.core.merge_anchor() + } + fn put_field(&mut self, id: &str, field: &str, value: ScalarValue) -> Result<(), String> { let item = self.require(id)?; self.core diff --git a/runtime/crates/engine/src/document.rs b/runtime/crates/engine/src/document.rs index 7e4d1afa..ed933e4a 100644 --- a/runtime/crates/engine/src/document.rs +++ b/runtime/crates/engine/src/document.rs @@ -26,6 +26,16 @@ use subduction_protocol::command::NewCommit; use crate::storage::SnapshotStorage; +/// What one batch of [`Document::apply`] did. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct Absorbed { + /// Anything at all landed, so the kernel must checkpoint. + pub landed: bool, + /// At least one landed change carried operations, as opposed to being a + /// merge anchor. Only content is worth anchoring for. + pub content: bool, +} + /// One automerge document and the tree its changes travel in. pub struct Document { doc: Automerge, @@ -122,7 +132,15 @@ impl Document { /// engine decrypts them before calling [`Document::apply`]. pub fn absorb(&mut self, storage: &SnapshotStorage) -> bool { let items = self.unapplied(storage); - self.apply(items) + self.apply(items).landed + } + + /// The commits this document has already applied. The vault needs them to + /// tell "an ancestor key I should keep, because that commit has not + /// arrived" from "an ancestor key I can drop, because I already read that + /// commit and its key rides in a descendant I hold". + pub fn applied_ids(&self) -> BTreeSet { + self.applied.clone() } /// The stored blobs of this tree the document has not applied yet, raw. @@ -135,15 +153,46 @@ impl Document { .collect() } - /// Apply decoded changes to the document. Returns whether anything landed. + /// Whether the document has more than one head — concurrent branches that + /// nothing has merged yet. See [`Document::merge_anchor`]. + pub fn diverged(&self) -> bool { + self.doc.get_heads().len() > 1 + } + + /// An empty change whose dependencies are every current head: automerge's + /// own merge commit (`Automerge::empty_commit` — "the main reason to do + /// this is if you want to create a merge commit"). + /// + /// It carries no operations, so it changes nothing anyone reads. What it + /// carries is its *envelope*: sealed under the group's current epoch with + /// the content keys of both branches inside it + /// (`crate::vault::Vault::seal`), it is the "new head" that + /// `design/causal_encryption.md` §"Multiple Heads" says connects a branch + /// no current member holds a key for. + pub fn merge_anchor(&mut self) -> Option { + let _hash = self + .doc + .empty_commit(automerge::transaction::CommitOptions::default()); + let change = self.doc.get_last_local_change()?; + let head = CommitId::new(change.hash().0); + let _known = self.applied.insert(head); + Some(NewCommit { + head, + parents: change.deps().iter().map(|h| CommitId::new(h.0)).collect(), + blob: Blob::new(change.raw_bytes().to_vec()), + }) + } + + /// Apply decoded changes to the document. /// /// Changes may arrive before their dependencies (sync is a set /// reconciliation, not a topological replay); automerge queues a change /// whose deps are missing and applies it when they arrive, so the whole /// batch goes in as one call and order does not matter. - pub fn apply(&mut self, items: Vec<(CommitId, Vec)>) -> bool { + pub fn apply(&mut self, items: Vec<(CommitId, Vec)>) -> Absorbed { let mut ids = Vec::new(); let mut changes = Vec::new(); + let mut content = false; for (id, blob) in items { if self.applied.contains(&id) { continue; @@ -151,17 +200,23 @@ impl Document { let Ok(change) = automerge::Change::from_bytes(blob) else { continue; }; + // A change with no operations is a merge anchor, somebody else's + // or an older one of ours. It is not a reason to author another. + content |= !change.is_empty(); ids.push(id); changes.push(change); } if changes.is_empty() { - return false; + return Absorbed::default(); } if self.doc.apply_changes(changes).is_err() { - return false; + return Absorbed::default(); } self.applied.extend(ids); - true + Absorbed { + landed: true, + content, + } } /// Run one transaction and, if it produced a change, record the commit diff --git a/runtime/crates/engine/src/lib.rs b/runtime/crates/engine/src/lib.rs index 9f9329b1..d960b80c 100644 --- a/runtime/crates/engine/src/lib.rs +++ b/runtime/crates/engine/src/lib.rs @@ -403,6 +403,14 @@ impl + 'static> Engine { Ok(()) } + /// How many entry points into the group's document this device holds — + /// the size of the head set it checkpoints and hands to the next device it + /// enrols. One per unmerged readable branch; a linear history is one. + pub async fn entry_points(&self) -> Result { + self.open_us().await?; + Ok(self.require_vault()?.entry_points()) + } + /// The user-system document's bytes, for the adder to put in ENROLL. pub async fn us_save(&self) -> Result, String> { self.open_us().await?; @@ -1004,11 +1012,14 @@ impl + 'static> Engine { Ok((answer, doc.tree(), doc.last_local_commit())) })?; if let Some(commit) = commit { - let commit = self.seal(commit).await?; + let (commit, sealed) = self.seal(commit).await?; self.handle .add_commits(tree, vec![commit]) .await .map_err(|e| e.to_string())?; + // Only now: until the driver has taken the commit, the parents + // whose keys are inside it must stay on the frontier. + self.require_vault()?.confirm(&sealed); // A durability barrier, and the reason the kernel may checkpoint // the moment this returns. `add_commits` only queues a command; // the driver signs and persists it inside `drain_effects`, which @@ -1034,7 +1045,7 @@ impl + 'static> Engine { async fn seal( &self, commit: subduction_protocol::command::NewCommit, - ) -> Result { + ) -> Result<(subduction_protocol::command::NewCommit, vault::Sealed), String> { let vault = self.require_vault()?; let preds: Vec<[u8; 32]> = commit.parents.iter().map(|id| *id.as_bytes()).collect(); let sealed = vault @@ -1044,10 +1055,13 @@ impl + 'static> Engine { commit.blob.as_slice().to_vec(), ) .await?; - Ok(subduction_protocol::command::NewCommit { - blob: Blob::new(sealed), - ..commit - }) + Ok(( + subduction_protocol::command::NewCommit { + blob: Blob::new(sealed.blob.clone()), + ..commit + }, + sealed, + )) } /// Make sure this device holds a document for `app`, and that every live @@ -1139,12 +1153,18 @@ impl + 'static> Engine { // stay in storage and are tried again on the next event. return false; }; - let wanted = { + let (wanted, known) = { let apps = self.apps.borrow(); let Some(doc) = apps.get(app) else { return false; }; - doc.unapplied(&self.storage) + ( + doc.unapplied(&self.storage), + doc.applied_ids() + .into_iter() + .map(|id| *id.as_bytes()) + .collect(), + ) }; if wanted.is_empty() { return false; @@ -1155,21 +1175,52 @@ impl + 'static> Engine { .into_iter() .map(|(id, blob)| (*id.as_bytes(), blob)) .collect(), + &known, ) .await else { return false; }; - let mut apps = self.apps.borrow_mut(); - let Some(doc) = apps.get_mut(app) else { - return false; + let (absorbed, tree, anchor) = { + let mut apps = self.apps.borrow_mut(); + let Some(doc) = apps.get_mut(app) else { + return false; + }; + let absorbed = doc.apply( + opened + .into_iter() + .map(|(id, change)| (CommitId::new(id), change)) + .collect(), + ); + // 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 + // may not hold — a device enrolled after that branch was written + // can decrypt neither it nor anything automerge buffers behind it. + // A merge anchor republishes the branch: its envelope names both + // heads' content keys and is sealed under the current epoch, so + // every current member walks in from it. + // + // Only for content, and only on divergence. A batch that is itself + // nothing but somebody else's anchor is not a reason to author + // one, which is what stops two devices anchoring each other + // forever. + let anchor = (absorbed.content && doc.diverged()) + .then(|| doc.merge_anchor()) + .flatten(); + (absorbed, doc.tree(), anchor) }; - doc.apply( - opened - .into_iter() - .map(|(id, change)| (CommitId::new(id), change)) - .collect(), - ) + if let Some(anchor) = anchor + && let Ok((anchor, sealed)) = self.seal(anchor).await + { + let pushed = self.handle.add_commits(tree, vec![anchor]).await; + if pushed.is_ok() { + let _heads = self.handle.tree_heads(tree).await; + vault.confirm(&sealed); + let _published = self.publish_keyhive().await; + } + } + absorbed.landed } /// Ingest the group's keyhive operations, then retry every app document: diff --git a/runtime/crates/engine/src/vault.rs b/runtime/crates/engine/src/vault.rs index 340beab1..9a18ce0d 100644 --- a/runtime/crates/engine/src/vault.rs +++ b/runtime/crates/engine/src/vault.rs @@ -52,6 +52,10 @@ //! one person's own devices — which is the entire membership model here — that //! is the intent. It would be a policy decision to revisit for shared //! documents, which do not exist yet. +//! +//! Because the ancestry rides in the envelopes, the state a device keeps is a +//! *set of heads* — one `⟨pointer, key⟩` pair per readable branch — and not a +//! key per commit. See [`Vault::advance`]. use std::cell::{Cell, RefCell}; use std::collections::{HashMap, HashSet}; @@ -105,10 +109,31 @@ pub struct VaultState { pub archive: Vec, pub group: [u8; 32], pub doc: [u8; 32], - /// The content keys this device has learned, which is what lets it seal a - /// commit whose parents it did not author. Secret, and sealed with the - /// rest of the checkpoint. - pub chunk_keys: Vec<(Cref, [u8; 32])>, + /// This device's entry points into the group's document: one + /// `(commit, content key)` pair per head of the readable frontier + /// (`design/causal_encryption.md` §"Decryption Head"). Secret, and sealed + /// with the rest of the checkpoint. + /// + /// `alias`: an M3c checkpoint wrote the whole key map under `chunk_keys`. + /// Loading one is harmless — the full map is a superset of the head set, + /// and the extra entries are pruned the first time their descendants are + /// opened or sealed. + #[serde(alias = "chunk_keys", default)] + pub heads: Vec<(Cref, [u8; 32])>, +} + +/// A sealed commit, and what the frontier owes it once it has landed. +/// +/// Separate from [`Vault::seal`] because the frontier must not move until +/// the commit is in storage: `advance` prunes the parents whose keys went +/// into this envelope, and if the write then failed those keys would be +/// gone with nothing carrying them — the device would hold a head naming a +/// commit no storage has. +pub struct Sealed { + pub blob: Vec, + cref: Cref, + key: SymmetricKey, + embedded: Vec, } /// This device's keyhive: its own identity, the device group, and the one @@ -121,8 +146,9 @@ pub struct Vault { /// engine holds the vault behind an `Rc` it clones before every await. group: Cell, doc: Cell, - /// Content keys, by commit id. See [`VaultState::chunk_keys`]. - chunk_keys: RefCell>, + /// The readable frontier: a content key per head commit, and nothing + /// below them. See [`VaultState::heads`] and [`Vault::advance`]. + heads: RefCell>, /// Digests of the static events already carried into the keyhive-events /// tree, so republishing is a no-op rather than a re-commit of the whole /// op graph on every turn. @@ -162,7 +188,7 @@ impl Vault { kh, group: Cell::new(group_id), doc: Cell::new(doc_id), - chunk_keys: RefCell::new(HashMap::new()), + heads: RefCell::new(HashMap::new()), published: RefCell::new(HashSet::new()), }) } @@ -190,9 +216,9 @@ impl Vault { store, group: Cell::new(GroupId::new(identifier(state.group)?)), doc: Cell::new(DocumentId::from(identifier(state.doc)?)), - chunk_keys: RefCell::new( + heads: RefCell::new( state - .chunk_keys + .heads .iter() .map(|(cref, key)| (*cref, SymmetricKey::from(*key))) .collect(), @@ -208,8 +234,8 @@ impl Vault { archive: bincode::serialize(&archive).map_err(|e| format!("keyhive archive: {e}"))?, group: self.group.get().to_bytes(), doc: self.doc.get().to_bytes(), - chunk_keys: self - .chunk_keys + heads: self + .heads .borrow() .iter() .map(|(cref, key)| { @@ -299,7 +325,8 @@ impl Vault { self.static_events().await } - /// The content keys this device holds, for the device it is enrolling. + /// This device's entry points into the group's document, for the device it + /// is enrolling. /// /// BeeKEM hands a new member the *current* epoch key and nothing earlier, /// so a freshly paired device sees the group's whole history as noise. The @@ -319,14 +346,17 @@ impl Vault { /// and it is what the e2e pairing ceremony expects. It would be a policy /// question for shared documents, which do not exist. /// - /// The map only grows, and it grows with every commit anyone in the group - /// ever writes. That is the standing cost of total read-back: it is - /// checkpointed on every device and copied to every device enrolled after - /// it. Bounding it *is* the chain-cut policy decision, which this milestone - /// does not make. + /// What is handed over is the *frontier*, not a key per commit: one + /// `⟨pointer, key⟩` pair per readable branch, from which everything + /// causally prior is discovered by following the ancestor keys inside each + /// envelope. `design/causal_encryption.md` §"Key Management" is explicit + /// that keeping them all is "possible, but fragile and unwieldy", and + /// §"Decryption Head" gives this shape instead. So this grows with live + /// concurrency and out-of-order delivery, not with history: a linear + /// history delivered in order hands over one pair however long it is. pub fn export_content_keys(&self) -> Result, String> { let keys: Vec<(Cref, [u8; 32])> = self - .chunk_keys + .heads .borrow() .iter() .map(|(cref, key)| { @@ -341,7 +371,7 @@ impl Vault { fn import_content_keys(&self, bytes: &[u8]) -> Result<(), String> { let keys: Vec<(Cref, [u8; 32])> = bincode::deserialize(bytes).map_err(|e| format!("bad content keys: {e}"))?; - self.chunk_keys.borrow_mut().extend( + self.heads.borrow_mut().extend( keys.into_iter() .map(|(cref, key)| (cref, SymmetricKey::from(key))), ); @@ -355,7 +385,9 @@ impl Vault { /// its contact card is what the adder just enrolled. What is replaced is /// which group and document it seals to; the group of one it generated at /// first boot stays in the op graph, unreferenced, which is what keeps its - /// own pre-pairing content readable *to itself*. + /// own pre-pairing content readable *to itself* — and readable to the group + /// as soon as this device writes once, since that write's envelope names + /// its old frontier as ancestors. pub async fn adopt( &self, events: &[u8], @@ -423,6 +455,16 @@ impl Vault { Ok(out) } + /// How many entry points into the document this device holds: one per + /// branch of the readable frontier it cannot reach from another + /// (`design/causal_encryption.md` §"Multiple Heads"). This is the size of + /// what the checkpoint carries and what the next device enrolled is + /// handed, so it is the number this milestone exists to keep small. + #[must_use] + pub fn entry_points(&self) -> usize { + self.heads.borrow().len() + } + // -- content ------------------------------------------------------------- /// Seal one automerge change as the sedimentree blob that carries it. @@ -431,7 +473,7 @@ impl Vault { cref: Cref, preds: &[Cref], change: Vec, - ) -> Result, String> { + ) -> Result { let doc = self .kh .get_document(self.doc.get()) @@ -443,12 +485,13 @@ impl Vault { // authoring on top of it merely means the chain is cut there — for // everyone, equally, which is what we would want if it happened. let ancestors: HashMap = { - let keys = self.chunk_keys.borrow(); + let keys = self.heads.borrow(); preds .iter() .filter_map(|parent| keys.get(parent).map(|key| (*parent, *key))) .collect() }; + let embedded: Vec = ancestors.keys().copied().collect(); let plaintext = bincode::serialize(&Envelope { plaintext: change, ancestors, @@ -460,11 +503,28 @@ impl Vault { .try_encrypt_content_keyed(doc, &cref, &preds.to_vec(), &plaintext) .await .map_err(|e| format!("encrypt: {e:?}"))?; - let _replaced = self.chunk_keys.borrow_mut().insert(cref, key); self.store .insert(Arc::new(sealed.encrypted_content().clone())) .await; - bincode::serialize(sealed.encrypted_content()).map_err(|e| format!("envelope: {e}")) + let blob = + bincode::serialize(sealed.encrypted_content()).map_err(|e| format!("envelope: {e}"))?; + Ok(Sealed { + blob, + cref, + key, + embedded, + }) + } + + /// The sealed commit is in storage. The frontier moves forward: it becomes + /// a head, and the parents whose keys went INTO its envelope stop being + /// ones. They are not lost — they are one hop below a head, which is where + /// `design/causal_encryption.md` says a key belongs. + pub fn confirm(&self, sealed: &Sealed) { + self.advance( + &[(sealed.cref, sealed.key)], + sealed.embedded.iter().copied(), + ); } /// Open as many of `blobs` as this device can. @@ -480,7 +540,16 @@ impl Vault { /// Undecryptable commits are dropped rather than reported: a device that /// has not yet ingested the epoch material sees them again on the next /// absorb, which is what the keyhive-events tree exists to fix. - pub async fn open(&self, blobs: Vec<(Cref, Vec)>) -> Result)>, String> { + /// `known` is the set of commits the caller has already applied. It is + /// what makes an ancestor key droppable: a key for a commit this device has + /// read is carried by the descendant that named it, while a key for a + /// commit that has *not* arrived is the only way that commit will ever be + /// opened, so it is kept. + pub async fn open( + &self, + blobs: Vec<(Cref, Vec)>, + known: &HashSet, + ) -> Result)>, String> { let Some(doc) = self.kh.get_document(self.doc.get()).await else { // Commits are here but the document's keyhive state is not. Not an // error — the keyhive-events tree has not caught up. @@ -498,6 +567,13 @@ impl Vault { let mut opened: Vec<(Cref, Vec)> = Vec::new(); let mut dark: Vec<&(Cref, Ciphertext)> = Vec::new(); + // Commits that become heads, and commits whose key now rides inside + // one — collected across the whole batch, applied once at the end. + let mut reached: Vec<(Cref, SymmetricKey)> = Vec::new(); + let mut covered: HashSet = HashSet::new(); + // Ancestor keys read out of the envelopes this batch opened, decided on + // once at the end. + let mut ancestors: HashMap = HashMap::new(); for item in &wanted { let (cref, encrypted) = item; match self @@ -508,7 +584,15 @@ impl Vault { Ok((plain, key)) => { let envelope: Envelope> = bincode::deserialize(&plain).map_err(|e| format!("chunk envelope: {e}"))?; - self.remember(*cref, key, &envelope); + // Deferred to one `advance` at the end of the batch: + // applying insert-then-prune commit by commit would let a + // parent opened later in the same batch re-enter the + // frontier after its child had already pruned it. + reached.push((*cref, key)); + // Not covered yet: whether an ancestor's key can be dropped + // depends on whether that ancestor is reachable, which is + // not known until the walk below has run. + ancestors.extend(envelope.ancestors.iter().map(|(a, k)| (*a, *k))); opened.push((*cref, envelope.plaintext)); } Err(_) => dark.push(item), @@ -521,15 +605,33 @@ impl Vault { // which is what the keys handed over at enrollment are for. This is // the same walk `Keyhive::try_causal_decrypt_content` runs, entered // one level down so the seed can be a key rather than an epoch. + // Seeded from every entry point that can reach into the dark: a + // held key for a dark commit itself, and — the case that actually + // carries a partition — a commit just opened under the current + // epoch whose envelope names dark ancestors. The second is the + // whole of `design/causal_encryption.md` §"Multiple Heads": a new + // head supplied for a branch is what connects it, and here the new + // head arrived as ordinary content. + let by_cref: HashMap = + wanted.iter().map(|(cref, enc)| (*cref, enc)).collect(); let mut seeds: Vec<(Arc, SymmetricKey)> = { - let held = self.chunk_keys.borrow(); + let held = self.heads.borrow(); dark.iter() .filter_map(|(cref, encrypted)| { held.get(cref) .map(|key| (Arc::new(encrypted.clone()), *key)) }) + .chain(reached.iter().filter_map(|(cref, key)| { + by_cref + .get(cref) + .map(|enc| (Arc::new((*enc).clone()), *key)) + })) .collect() }; + let seeded: HashSet = seeds + .iter() + .map(|(encrypted, _)| encrypted.content_ref) + .collect(); if !seeds.is_empty() { let walked: CausalDecryptionState> = match CiphertextStoreExt::>::try_causal_decrypt( @@ -545,39 +647,102 @@ impl Vault { // `complete` is already the envelopes' payloads: the walk // unwraps each `Envelope` itself (keyhive_core // store/ciphertext.rs:222). - let reached: HashMap> = walked.complete.into_iter().collect(); - let mut learned: Vec<(Cref, SymmetricKey)> = walked.next.into_iter().collect(); + let plaintexts: HashMap> = walked.complete.into_iter().collect(); + // Everything the walk touched below its seeds is covered by + // definition: it got there by reading a seed's envelope, and so + // will anyone else who holds that seed. `next` — ancestors whose + // ciphertext has not arrived yet — is covered for the same + // reason, which is why a key for a commit we do not hold is not + // worth keeping either. The seeds themselves stay: they are the + // frontier the walk descended from. + // Only what actually opened. `walked.keys` also records the key + // of a ciphertext whose decrypt FAILED — a corrupt or + // wrongly-sealed blob — and pruning on one of those would drop a + // key nothing carries. + covered.extend( + plaintexts + .keys() + .filter(|cref| !seeded.contains(*cref)) + .copied(), + ); + // Ancestors the walk could not reach because their ciphertext is + // not here yet. Same decision as the envelopes' own ancestors, + // below. + ancestors.extend(walked.next.iter().map(|(cref, key)| (*cref, *key))); for (cref, _) in &dark { - if let Some(plain) = reached.get(cref) { - // Keys only for chunks that actually opened: `keys` also - // records the key of a ciphertext whose decrypt FAILED, - // and keeping one of those would let a later seal name a - // parent key that opens nothing. - if let Some(key) = walked.keys.get(cref) { - learned.push((*cref, *key)); - } + if let Some(plain) = plaintexts.get(cref) { opened.push((*cref, plain.clone())); } } - self.chunk_keys.borrow_mut().extend(learned); } } + // The ancestor keys, decided. An ancestor this device can already reach + // — because it opened in this batch, because it is on the frontier, or + // because the caller has long since applied it — is covered: its key + // rides inside the descendant that just named it. An ancestor that has + // simply not arrived is none of those, and dropping its key would strand + // it: when its ciphertext turns up in a later batch there would be + // nothing to seed a walk with, and every descendant already applied + // would sit in automerge's buffer behind it forever. + // + // Keeping it costs one frontier entry per out-of-order arrival, and + // that entry outlives the arrival: once the commit opens as a seed it + // stays a head, because the descendant that would cover it was read + // before it existed here. Bounded by how often delivery runs backwards, + // not by history. + { + let held: HashSet = self.heads.borrow().keys().copied().collect(); + let opened_now: HashSet = opened.iter().map(|(cref, _)| *cref).collect(); + for (cref, key) in ancestors { + if opened_now.contains(&cref) || held.contains(&cref) || known.contains(&cref) { + let _covered = covered.insert(cref); + } else { + reached.push((cref, key)); + } + } + } + self.advance(&reached, covered.into_iter()); Ok(opened) } // -- internals ----------------------------------------------------------- - /// Record a commit's own content key and the ancestor keys its envelope - /// carried. This map is the read-back state: it only grows, one entry per - /// commit this device has ever opened, and it is both checkpointed and - /// handed to the next device enrolled. That growth is the standing cost of - /// total read-back (see `export_content_keys`) — 64 bytes per commit, and - /// nothing prunes it, because pruning it is the chain-cut policy decision - /// this milestone does not make. - fn remember(&self, cref: Cref, key: SymmetricKey, envelope: &Envelope>) { - let mut keys = self.chunk_keys.borrow_mut(); - let _replaced = keys.insert(cref, key); - keys.extend(envelope.ancestors.iter().map(|(a, k)| (*a, *k))); + /// Move the readable frontier: `arrived` become heads, `covered` stop + /// being ones. + /// + /// This is what keeps the entry point a *set of heads* rather than a key + /// per commit ever seen. `design/causal_encryption.md` §"Key Management" + /// is blunt that the latter is "possible, but fragile and unwieldy", and + /// §"Decryption Head" gives the shape that replaces it: a + /// `⟨pointer, key⟩` pair is an entry point, and everything causally prior + /// is discovered by following the ancestor keys inside each envelope. + /// + /// Pruning is only sound because a covered commit's key demonstrably rides + /// inside a head's envelope: `seal` prunes exactly the parents it wrote + /// into `ancestors`, and `open` prunes exactly the crefs it read back out + /// of one. A parent whose key this device did not hold is therefore never + /// pruned — there is nothing carrying it. + /// + /// Inserts run before removals so that a batch containing both a parent + /// and its child leaves the child, whatever order they were opened in. + /// + /// The frontier is therefore exact for in-order delivery and slightly + /// conservative for out-of-order delivery: a commit whose key was kept + /// because it had not arrived stays a head once it does, since the + /// descendant that would have covered it was read before it got here. One + /// extra entry per out-of-order arrival, and nothing that is not a + /// legitimate entry point. + fn advance(&self, arrived: &[(Cref, SymmetricKey)], covered: impl Iterator) { + let mut heads = self.heads.borrow_mut(); + for (cref, key) in arrived { + let _replaced = heads.insert(*cref, *key); + } + let fresh: HashSet = arrived.iter().map(|(cref, _)| *cref).collect(); + for cref in covered { + if !fresh.contains(&cref) { + let _dropped = heads.remove(&cref); + } + } } async fn all_events(&self) -> Vec> { diff --git a/runtime/crates/engine/tests/converge.rs b/runtime/crates/engine/tests/converge.rs index 4d326d13..320d23d9 100644 --- a/runtime/crates/engine/tests/converge.rs +++ b/runtime/crates/engine/tests/converge.rs @@ -12,7 +12,7 @@ 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, + StoreItem, TaskSnapshot, TreeState, tasks_tree, }; use subduction_protocol::event::Direction; use subduction_runtime::memory::transport::MemoryTransport; @@ -200,6 +200,17 @@ async fn enroll(adder: &TestEngine, joiner: &TestEngine) { joiner.adopt_keyhive(&keyhive, &read_back).await.unwrap(); } +/// Wire two engines together without enrolling: the group is already shared. +async fn wire_only(a: &TestEngine, b: &TestEngine) { + let (ta, tb) = MemoryTransport::pair(); + let b_key = b.verifying_key(); + let inbound = RefCell::new(None); + let _ = futures::future::join(a.connect(ta, Direction::Outbound, Some(b_key)), async { + *inbound.borrow_mut() = Some(b.connect(tb, Direction::Inbound, None).await); + }) + .await; +} + fn titles(snapshot: &TaskSnapshot) -> Vec { snapshot.items.iter().map(|i| i.title.clone()).collect() } @@ -869,3 +880,227 @@ fn a_checkpoint_written_before_the_store_existed_still_mints_a_name_key() { ); }); } + +/// The frontier is a set, and divergence is what makes it bigger than one. +/// +/// Two devices write while apart, so the document has two branches off one +/// root; each device then holds an entry point per branch it cannot reach from +/// the other (`design/causal_encryption.md` §"Multiple Heads"). What it does +/// *not* hold is a key per commit: the root's key rides inside its children's +/// envelopes and is dropped from the frontier as soon as one of them is read. +#[test] +fn concurrent_branches_leave_one_entry_point_each() { + let mut pool = LocalPool::new(); + let a = device(&pool, 40, None); + let b = device(&pool, 41, None); + let (ea, eb) = (Rc::clone(&a.engine), Rc::clone(&b.engine)); + + pool.run_until(async move { + ea.tasks_add(APP, "root".into()).await.unwrap(); + enroll(&ea, &eb).await; + assert_eq!( + ea.entry_points().await.unwrap(), + 1, + "a linear history is one entry point" + ); + + // Apart: B has never synced, so its write branches from nothing while + // A's extends the root. + ea.tasks_add(APP, "branch a".into()).await.unwrap(); + eb.tasks_add(APP, "branch b".into()).await.unwrap(); + assert_eq!( + ea.entry_points().await.unwrap(), + 1, + "A extended its own history and stayed at one entry point" + ); + assert_eq!( + eb.entry_points().await.unwrap(), + 2, + "B holds the root it was enrolled with and its own concurrent branch" + ); + + wire_only(&ea, &eb).await; + let merged = until(|| async { + let items = ea.tasks_items(APP).await.unwrap(); + (items.items.len() == 3).then_some(items) + }) + .await; + let mut seen = titles(&merged); + seen.sort(); + assert_eq!(seen, vec!["branch a", "branch b", "root"]); + for _ in 0..500 { + yield_now().await; + } + + // Both anchored the merge, and those two anchors are themselves + // concurrent — two branches, two entry points, and no third anchor: + // an anchor is not content, and only content is worth anchoring for. + // Bounded by concurrency, not by history length, which is the whole + // claim. + for engine in [&ea, &eb] { + assert_eq!( + engine.entry_points().await.unwrap(), + 2, + "the frontier grew past the number of live branches" + ); + } + }); +} + +/// A device enrolled while another was offline still reads that device's +/// branch, without waiting for anyone to write again. +/// +/// This is the partition case: B's writes were sealed under an epoch that +/// predates C's enrolment, so C can decrypt neither them nor anything automerge +/// buffers behind them. The device that *can* read both — A — republishes the +/// branch by anchoring the merge, and C walks in from there +/// (`design/causal_encryption.md` §"Multiple Heads": a branch is connected "by +/// supplying a new head for it"). +#[test] +fn a_branch_written_before_a_joiner_existed_reaches_it_through_the_anchor() { + let mut pool = LocalPool::new(); + let a = device(&pool, 30, None); + let b = device(&pool, 31, None); + let c = device(&pool, 32, None); + let (ea, eb, ec) = ( + Rc::clone(&a.engine), + Rc::clone(&b.engine), + Rc::clone(&c.engine), + ); + + pool.run_until(async move { + ea.tasks_add(APP, "from a".into()).await.unwrap(); + enroll(&ea, &eb).await; + // B never connects: it writes into a partition. It has never synced, + // so its app tree holds exactly this one commit. + eb.tasks_add(APP, "from b offline".into()).await.unwrap(); + let b_commit = { + let mut own: Vec<[u8; 32]> = eb + .items() + .iter() + .filter(|item| item.tree == *tasks_tree(APP).as_bytes()) + .map(|item| item.commit) + .collect(); + assert_eq!(own.len(), 1, "B wrote more than the one commit"); + own.pop().expect("B's own commit") + }; + // C is enrolled meanwhile, so its epoch begins after B's write. + enroll(&ea, &ec).await; + + wire_only(&ea, &eb).await; + let _merged = until(|| async { + let items = ea.tasks_items(APP).await.unwrap(); + (items.items.len() == 2).then_some(items) + }) + .await; + + // Delivered to C through the store, in two batches with the anchor + // first — so the claim "without anyone writing again" is pinned across + // separate deliveries and not just within one sync. + assert!(ec.tasks_items(APP).await.unwrap().items.is_empty()); + let (branch, rest): (Vec<_>, Vec<_>) = ea + .items() + .into_iter() + .partition(|item| item.commit == b_commit); + assert_eq!(branch.len(), 1, "B's offline commit is not in what A holds"); + + let _landed = ec.ingest_items(rest).await.unwrap(); + for _ in 0..200 { + yield_now().await; + } + let _landed = ec.ingest_items(branch).await.unwrap(); + let seen = until(|| async { + let items = ec.tasks_items(APP).await.unwrap(); + (items.items.len() == 2).then_some(items) + }) + .await; + let mut seen = titles(&seen); + seen.sort(); + assert_eq!( + seen, + vec!["from a", "from b offline"], + "the joiner never reached the branch written before it existed" + ); + }); +} + +/// Out-of-order delivery: the child arrives in one batch and its parents in +/// the next, and the parents still open. +/// +/// This is the store path rather than the peer path, because a store hands back +/// whatever it happened to list — there is no causal order in a bucket (keyhive +/// `design/causal_encryption.md` §"Crypt Store": "there is no dependency on +/// ordering between encrypted blobs"). +/// +/// The joiner holds one entry point, the head. Batch one gives it that head and +/// nothing below; the walk opens it and reads out the keys of ancestors whose +/// ciphertext has not arrived (`CausalDecryptionState::next`). Those keys are +/// the only way those commits will ever be opened: they predate the joiner's +/// enrolment, so no epoch of its own reaches them, and once the head is applied +/// no later batch can re-derive them. Batch two delivers them and nothing else, +/// with nobody writing anything. +#[test] +fn parents_delivered_after_their_child_still_open() { + let mut pool = LocalPool::new(); + let a = device(&pool, 50, None); + let c = device(&pool, 52, None); + let (ea, ec) = (Rc::clone(&a.engine), Rc::clone(&c.engine)); + + pool.run_until(async move { + let app_tree = *tasks_tree(APP).as_bytes(); + let app_commits = |engine: &TestEngine| -> Vec<[u8; 32]> { + engine + .items() + .iter() + .filter(|item| item.tree == app_tree) + .map(|item| item.commit) + .collect() + }; + + ea.tasks_add(APP, "first".into()).await.unwrap(); + ea.tasks_add(APP, "second".into()).await.unwrap(); + let older = app_commits(&ea); + assert_eq!(older.len(), 2, "two commits so far"); + ea.tasks_add(APP, "third".into()).await.unwrap(); + + // C is enrolled now, so every one of those three commits predates its + // epoch: the head's key, handed over at enrolment, is its only way in. + enroll(&ea, &ec).await; + + let (parents, first): (Vec<_>, Vec<_>) = ea + .items() + .into_iter() + .partition(|item| item.tree == app_tree && older.contains(&item.commit)); + assert_eq!( + parents.len(), + 2, + "exactly the two older commits are held back" + ); + + // C opens the app before either delivery: without a document the + // items would just sit in storage and both batches would be opened + // together on the first read, which is not what this is testing. + assert!(ec.tasks_items(APP).await.unwrap().items.is_empty()); + + let _landed = ec.ingest_items(first).await.unwrap(); + for _ in 0..200 { + yield_now().await; + } + assert!( + ec.tasks_items(APP).await.unwrap().items.is_empty(), + "the head alone should materialize nothing: its parents are missing" + ); + // Batch two: the parents, alone. Nothing writes. + let _landed = ec.ingest_items(parents).await.unwrap(); + let seen = until(|| async { + let items = ec.tasks_items(APP).await.unwrap(); + (items.items.len() == 3).then_some(items) + }) + .await; + assert_eq!( + titles(&seen), + vec!["first", "second", "third"], + "the parents delivered after their child were never opened" + ); + }); +} diff --git a/runtime/crates/kernel/src/drive.rs b/runtime/crates/kernel/src/drive.rs index d4b68dc1..c4391b86 100644 --- a/runtime/crates/kernel/src/drive.rs +++ b/runtime/crates/kernel/src/drive.rs @@ -500,7 +500,25 @@ impl Kernel { } } - /// Push, then pull. Answers whether the pull landed anything. + /// Pull, then push. Answers whether the pull landed anything. + /// + /// The order matters, and it is the read-back order + /// (`docs/design.md` §"Read-back and partitions"). A device coming back + /// online holds writes sealed under whatever epoch it last knew. If it + /// pushed first, those objects would reach the store ahead of any head that + /// connects them, and a member enrolled since would find them + /// undecryptable. Pulling first means this device ingests the group's + /// current keyhive state and merges before it publishes, so where its + /// writes are *concurrent* with the group's they go up alongside the merge + /// anchor that names their keys. + /// + /// It does not close the linear case, and that is worth naming rather than + /// implying: a device that was merely behind — its writes sit on top of + /// what the group already had, so the ingest produces no divergence and no + /// anchor — pushes commits that a later-enrolled member still cannot open + /// until somebody writes on top of them. That write happens on the next + /// local mutation, whose envelope names this frontier; until then the + /// branch is latency, not loss. async fn sync_attempt(self: &Rc, name_key: &[u8; 32]) -> Result { let folder = self.drive_folder(name_key).await?; let engine = self @@ -508,25 +526,6 @@ impl Kernel { .map_err(|e| Trouble::Hiccup(e.message.clone()))?; let remote = self.drive_list(&folder).await?; - let present: BTreeSet<&str> = remote.iter().map(|(_, name)| name.as_str()).collect(); - - // Push: one object per item the store lacks, and never a second - // upload of a name it has — the name *is* the item's digest pair, so - // an object that exists is already these bytes. - let mut pushed = false; - for item in engine.items() { - let name = object_name(name_key, &item.tree, &item.commit); - if present.contains(name.as_str()) { - continue; - } - let body = serde_json::to_vec(&item) - .map_err(|e| Trouble::Hiccup(format!("an item could not be written: {e}")))?; - self.drive_create(&folder, &name, body).await?; - pushed = true; - } - if pushed { - self.drive.borrow_mut().last_push = self.seams.clock.now_ms(); - } // Pull: everything the store holds under a name this device cannot // account for. `mine` is derived rather than listed, so an item this @@ -556,27 +555,54 @@ impl Kernel { } } } - if fetched.is_empty() { - return Ok(false); - } // The group this pass was reading for must still be this device's // group. Pairing replaces both at once — the group document and the - // name key with it (`Engine::adopt_us`) — so a pass that started - // before that landed is holding objects named under the group this - // device has just *stopped* being: its own group of one. Installing - // them would put the commits `adopt_us` deliberately removed back - // into the adopted document's tree, which is the one thing that - // function exists to prevent, and the device would end up in neither - // group cleanly. The pass is dropped; the next one reads the new - // group's names. + // name key with it (`Engine::adopt_us`) — so a pass that started before + // that landed is holding objects named under the group this device has + // just *stopped* being: its own group of one. Installing them would put + // the commits `adopt_us` deliberately removed back into the adopted + // document's tree, which is the one thing that function exists to + // prevent, and the device would end up in neither group cleanly. + // + // Checked before the push as well as the pull, and for the mirror + // reason: the push writes objects under `name_key`, and a device that + // has just joined a group must not scatter its abandoned group-of-one's + // commits into that group's folder under the old names. The pass is + // dropped; the next one reads and writes the new group's names. if engine.name_key().as_ref() != Some(name_key) { return Ok(false); } - let landed = engine - .ingest_items(fetched) - .await - .map_err(Trouble::Hiccup)?; - self.drive.borrow_mut().last_pull = self.seams.clock.now_ms(); + + let mut landed = false; + if !fetched.is_empty() { + landed = engine + .ingest_items(fetched) + .await + .map_err(Trouble::Hiccup)?; + self.drive.borrow_mut().last_pull = self.seams.clock.now_ms(); + } + + // Push: one object per item the store lacks, and never a second + // upload of a name it has — the name *is* the item's digest pair, so + // an object that exists is already these bytes. Read after the pull, + // so a merge anchor the ingest just authored goes up in this same + // pass rather than waiting for the next one. + 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); + if present.contains(name.as_str()) { + continue; + } + let body = serde_json::to_vec(&item) + .map_err(|e| Trouble::Hiccup(format!("an item could not be written: {e}")))?; + self.drive_create(&folder, &name, body).await?; + pushed = true; + } + if pushed { + self.drive.borrow_mut().last_push = self.seams.clock.now_ms(); + } + Ok(landed) }