Read-back as a head set; merge anchors; Drive pulls before pushing

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

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<u8>)>) -> bool {
+    /// Apply decrypted automerge changes.
+    pub fn apply(&mut self, items: Vec<(CommitId, Vec<u8>)>) -> Absorbed {
         self.core.apply(items)
     }
 
+    pub fn applied_ids(&self) -> std::collections::BTreeSet<CommitId> {
+        self.core.applied_ids()
+    }
+
+    pub fn diverged(&self) -> bool {
+        self.core.diverged()
+    }
+
+    pub fn merge_anchor(&mut self) -> Option<NewCommit> {
+        self.core.merge_anchor()
+    }
+
     fn put_field(&mut self, id: &str, field: &str, value: ScalarValue) -> Result<(), String> {
         let item = self.require(id)?;
         self.core

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<CommitId> {
+        self.applied.clone()
     }
 
     /// The stored blobs of this tree the document has not applied yet, raw.
@@ -135,33 +153,70 @@ 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<NewCommit> {
+        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<u8>)>) -> bool {
+    pub fn apply(&mut self, items: Vec<(CommitId, Vec<u8>)>) -> 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;
             }
             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

runtime/crates/engine/src/lib.rs

@@ -403,6 +403,14 @@ impl<T: Transport<Local> + 'static> Engine<T> {
         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<usize, String> {
+        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<Vec<u8>, String> {
         self.open_us().await?;
@@ -1004,11 +1012,14 @@ impl<T: Transport<Local> + 'static> Engine<T> {
             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<T: Transport<Local> + 'static> Engine<T> {
     async fn seal(
         &self,
         commit: subduction_protocol::command::NewCommit,
-    ) -> Result<subduction_protocol::command::NewCommit, String> {
+    ) -> 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<T: Transport<Local> + 'static> Engine<T> {
                 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<T: Transport<Local> + 'static> Engine<T> {
             // 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<T: Transport<Local> + 'static> Engine<T> {
                     .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: