diff --git a/knot2/crates/knot-cob/src/lib.rs b/knot2/crates/knot-cob/src/lib.rs index 578df86aa..75454b3da 100644 --- a/knot2/crates/knot-cob/src/lib.rs +++ b/knot2/crates/knot-cob/src/lib.rs @@ -6,17 +6,22 @@ mod graph; pub mod instrument; mod object; -pub use change::{Change, ChangePayload, CobHome, Payload}; -pub use error::{CobError, PayloadError}; +pub use change::{Change, ChangePayload, CobHome, Emits, Payload, Record, decode_payload}; +pub use error::{AddressMove, CobError, PayloadError}; pub use graph::{ChangeGraph, History}; pub use knot_types::{ActorId, ChangeId, CobId, TypeName}; -pub use object::{Checkpoint, Evaluate, HistoryModel, Object, SnapshotStride, StateSize}; +pub use object::{ + Append, Authored, Authorize, Checkpoint, Evaluate, HistoryModel, KnotAuthored, Object, + SnapshotStride, StateSize, WriteContext, +}; pub use backend::parse_cob_ref; -use knot_git::{RefUpdate, Repo}; +use std::collections::BTreeSet; + +use knot_git::{Haves, RefUpdate, Repo, Wants}; use knot_runtime::Signer; -use knot_types::{Oid, UnixSeconds}; +use knot_types::{Decision, Oid, RecordAddress, UnixSeconds}; use serde::Serialize; use serde::de::DeserializeOwned; @@ -40,16 +45,42 @@ pub struct Created { #[derive(Debug, Clone)] pub struct Materialized

{ pub change: P, - pub record: Option>, + pub record: Record, } impl

Materialized

{ pub fn bare(change: P) -> Self { Self { change, - record: None, + record: Record::Bare, + } + } +} + +pub struct Approved<'cx, C> { + cx: &'cx WriteContext<'cx>, + materialized: Materialized, +} + +impl<'cx, C: Authored> Approved<'cx, C> { + pub fn signed_by( + cx: &'cx WriteContext<'cx>, + materialized: Materialized, + ) -> Result { + let author = materialized.change.author(); + if author == cx.actor() { + Ok(Self { cx, materialized }) + } else { + Err(CobError::Impersonation { + actor: cx.actor().clone(), + author: author.clone(), + }) } } + + pub const fn context(&self) -> &'cx WriteContext<'cx> { + self.cx + } } #[derive(Debug)] @@ -58,7 +89,7 @@ pub struct Delta { pub tip: ChangeId, } -const CHECKPOINT_FORMAT: u16 = 4; +const CHECKPOINT_FORMAT: u16 = 5; #[derive(Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] struct CheckpointDigest(Oid); @@ -113,7 +144,7 @@ impl<'r> CobStore<'r> { self.repo } - pub fn create( + pub fn create( &self, home: &CobHome, payload: &P, @@ -121,10 +152,28 @@ impl<'r> CobStore<'r> { timestamp: UnixSeconds, ) -> Result { let bytes = payload.encode()?; - self.create_from_bytes(home, &P::type_name(), &bytes, None, signer, timestamp) + Self::check_payload(payload, &Record::Bare, timestamp)?; + self.create_from_bytes( + home, + &P::type_name(), + &bytes, + &Record::Bare, + signer, + timestamp, + ) } - pub fn create_materialized( + pub fn create_materialized( + &self, + home: &CobHome, + payload: &Materialized

, + signer: &dyn Signer, + timestamp: UnixSeconds, + ) -> Result { + self.open(home, payload, signer, timestamp) + } + + fn open( &self, home: &CobHome, payload: &Materialized

, @@ -132,22 +181,80 @@ impl<'r> CobStore<'r> { timestamp: UnixSeconds, ) -> Result { let bytes = payload.change.encode()?; + Self::check_payload(&payload.change, &payload.record, timestamp)?; self.create_from_bytes( home, &P::type_name(), &bytes, - payload.record.as_deref(), + &payload.record, signer, timestamp, ) } + pub fn create_approved( + &self, + home: &CobHome, + approved: &Approved<'_, E::Change>, + signer: &dyn Signer, + timestamp: UnixSeconds, + ) -> Result + where + E: Authorize, + E::Change: ChangePayload + Authored, + { + let cx = approved.context(); + match E::may_create(cx) { + Decision::Allow => self.open(home, &approved.materialized, signer, timestamp), + Decision::Deny => Err(CobError::CreateRejected { + actor: cx.actor().clone(), + type_name: E::Change::type_name(), + permission: cx.permission(), + }), + } + } + + pub fn append_approved( + &self, + home: &CobHome, + object: CobId, + approved: &Approved<'_, E::Change>, + signer: &dyn Signer, + timestamp: UnixSeconds, + ) -> Result + where + E: Authorize + Checkpoint, + E::State: Serialize + DeserializeOwned, + E::Change: ChangePayload + Authored + Clone, + { + let cx = approved.context(); + self.retrying_append::(home, object, signer, timestamp, |state| { + match E::may_append(cx, state, &approved.materialized.change) { + Append::Allow => Ok(Some(approved.materialized.clone())), + Append::Deny => Err(CobError::AppendRejected { + actor: cx.actor().clone(), + object, + permission: cx.permission(), + }), + Append::Stale { current } => Err(CobError::StaleRecord { + object, + uri: self + .root_address::(object)? + .ok_or(CobError::Unaddressed { object })? + .at_uri(cx.repo()), + current, + }), + } + }) + .map(|tip| tip.expect("Approved append always yields its change")) + } + fn create_from_bytes( &self, home: &CobHome, type_name: &TypeName, bytes: &[u8], - record: Option<&[u8]>, + record: &Record, signer: &dyn Signer, timestamp: UnixSeconds, ) -> Result { @@ -164,14 +271,24 @@ impl<'r> CobStore<'r> { )?; let object = CobId::new(tip.oid()); let name = backend::cob_ref_name(type_name, object)?; - self.repo.update_ref(&RefUpdate::Create { - name, - new: tip.oid(), - })?; - Ok(Created { object, tip }) + match backend::resolve_tip(self.repo, type_name, object)? { + Some(current) if current == tip.oid() => Ok(Created { object, tip }), + Some(_) => Err(CobError::StaleTip { + object, + expected: tip, + }), + None => self + .repo + .update_ref(&RefUpdate::Create { + name, + new: tip.oid(), + }) + .map(|()| Created { object, tip }) + .map_err(CobError::from), + } } - pub fn update( + pub fn update( &self, home: &CobHome, object: CobId, @@ -188,7 +305,7 @@ impl<'r> CobStore<'r> { ) } - pub fn extend<'p, P: ChangePayload + 'p>( + pub fn extend<'p, P: KnotAuthored + 'p>( &self, home: &CobHome, object: CobId, @@ -206,7 +323,7 @@ impl<'r> CobStore<'r> { expected, changes .into_iter() - .map(|(payload, timestamp)| (payload, None, timestamp)), + .map(|(payload, timestamp)| (payload, &Record::Bare, timestamp)), signer, ) } @@ -227,7 +344,7 @@ impl<'r> CobStore<'r> { object, type_name, expected, - std::iter::once((payload, None, timestamp)), + std::iter::once((payload, &Record::Bare, timestamp)), signer, ) .map(|tip| tip.expect("one change chains onto one tip")) @@ -249,19 +366,96 @@ impl<'r> CobStore<'r> { object, type_name, expected, - std::iter::once((&payload.change, payload.record.as_deref(), timestamp)), + std::iter::once((&payload.change, &payload.record, timestamp)), signer, ) .map(|tip| tip.expect("one change chains onto one tip")) } + fn check_payload( + payload: &P, + record: &Record, + timestamp: UnixSeconds, + ) -> Result<(), CobError> { + Self::check_collection(payload)?; + Self::check_stamp(payload, timestamp)?; + Self::check_claim(payload, record) + } + + fn check_collection(payload: &P) -> Result<(), CobError> { + match (payload.address(), P::collection()) { + (Some(found), Some(collection)) if *found.collection() != collection => { + Err(CobError::AddressForeign { + type_name: P::type_name(), + found, + }) + } + _ => Ok(()), + } + } + + fn check_stamp(payload: &P, timestamp: UnixSeconds) -> Result<(), CobError> { + match payload.stamped_at() { + Some(stamped) if stamped != timestamp => Err(CobError::StampMoved { + stamped, + signed: timestamp, + }), + _ => Ok(()), + } + } + + fn check_claim(payload: &P, record: &Record) -> Result<(), CobError> { + match (payload.emits(), record) { + (Emits::Version(declared), Record::Version(body)) => { + let found = body.cid(); + if found == declared { + Ok(()) + } else { + Err(CobError::RecordMisclaimed { declared, found }) + } + } + (Emits::Version(declared), Record::Bare | Record::Tombstone) => { + Err(CobError::RecordUnwritten { declared }) + } + (Emits::Tombstone, Record::Tombstone) => Ok(()), + (Emits::Tombstone, Record::Bare | Record::Version(_)) => { + Err(CobError::TombstoneUnwritten) + } + (Emits::Unclaimed, Record::Tombstone) => Err(CobError::TombstoneUnclaimed), + (Emits::Unclaimed, Record::Bare | Record::Version(_)) => Ok(()), + } + } + + fn root_address( + &self, + object: CobId, + ) -> Result, CobError> { + let root = backend::read_change(self.repo, ChangeId::new(object.oid()))?; + Ok(P::decode(root.payload())?.address()) + } + + fn check_address(&self, object: CobId, payload: &P) -> Result<(), CobError> { + let Some(found) = payload.address() else { + return Ok(()); + }; + match self.root_address::

(object)? { + Some(expected) if expected == found => Ok(()), + Some(expected) => Err(CobError::AddressMoved(Box::new(error::AddressMove { + object, + expected, + found, + }))), + None => Err(CobError::AddressAdded { object, found }), + } + } + fn chain<'p, P: ChangePayload + 'p>( &self, home: &CobHome, object: CobId, type_name: &TypeName, expected: ChangeId, - changes: impl IntoIterator, UnixSeconds)>, + changes: impl IntoIterator, signer: &dyn Signer, ) -> Result, CobError> { let written = match changes.into_iter().try_fold( @@ -269,6 +463,8 @@ impl<'r> CobStore<'r> { |mut written, (payload, record, timestamp)| { let parent = written.last().copied().unwrap_or(expected); let write = || -> Result { + Self::check_payload(payload, record, timestamp)?; + self.check_address::

(object, payload)?; let bytes = payload.encode()?; backend::write_change_with_record( home, @@ -287,15 +483,14 @@ impl<'r> CobStore<'r> { written.push(tip); Ok(written) } - Err(error) => Err((written, error)), + Err(error) => Err(Box::new((written, error))), } }, ) { Ok(written) => written, - Err((partial, error)) => { - partial - .iter() - .try_for_each(|change| self.repo.remove_loose_object(change.oid()))?; + Err(boxed) => { + let (partial, error) = *boxed; + self.discard(type_name, object, expected, &partial)?; return Err(error); } }; @@ -322,11 +517,7 @@ impl<'r> CobStore<'r> { Ok(()) => Ok(tip), Err(error) => { let actual = backend::resolve_tip(self.repo, type_name, object)?; - if actual != Some(tip.oid()) { - written - .iter() - .try_for_each(|change| self.repo.remove_loose_object(change.oid()))?; - } + self.discard(type_name, object, expected, written)?; match actual { actual if actual != Some(expected.oid()) => { Err(CobError::StaleTip { object, expected }) @@ -337,6 +528,46 @@ impl<'r> CobStore<'r> { } } + fn discard( + &self, + type_name: &TypeName, + object: CobId, + expected: ChangeId, + written: &[ChangeId], + ) -> Result<(), CobError> { + if written.is_empty() { + return Ok(()); + } + // Originally I was gonna just delete `written` entirely but + // a simulatenous writing loser has the same literal OId(s) as the winner by nature, + // so let's not rip those out of what *is* actually valid history. + let published = self.published_since(type_name, object, expected)?; + Ok(written + .iter() + .filter(|change| !published.contains(&change.oid())) + .try_for_each(|change| self.repo.remove_loose_object(change.oid()))?) + } + + fn published_since( + &self, + type_name: &TypeName, + object: CobId, + expected: ChangeId, + ) -> Result, CobError> { + let Some(current) = backend::resolve_tip(self.repo, type_name, object)? else { + return Ok(BTreeSet::new()); + }; + if current == expected.oid() { + return Ok(BTreeSet::new()); + } + let have = [expected.oid()]; + Ok(self + .repo + .rev_walk(Wants::new(&[current]), Haves::new(&have))? + .into_iter() + .collect()) + } + pub fn update_with( &self, home: &CobHome, @@ -347,6 +578,7 @@ impl<'r> CobStore<'r> { ) -> Result where E: Evaluate, + E::Change: KnotAuthored, D: From, { self.update_maybe::(home, object, signer, timestamp, |state| { @@ -365,6 +597,7 @@ impl<'r> CobStore<'r> { ) -> Result, D> where E: Evaluate, + E::Change: KnotAuthored, D: From, { let type_name = E::Change::type_name(); @@ -452,23 +685,38 @@ impl<'r> CobStore<'r> { backend::list_objects(self.repo, &E::Change::type_name()) } - pub fn changes_since( + pub fn cobs(&self) -> Result, CobError> { + backend::list_cobs(self.repo) + } + + pub fn changes_of( &self, + type_name: &TypeName, object: CobId, + history: HistoryModel, since: Option, ) -> Result { - let type_name = E::Change::type_name(); - let tip = backend::resolve_tip(self.repo, &type_name, object)? + let tip = backend::resolve_tip(self.repo, type_name, object)? .map(ChangeId::new) .ok_or(CobError::NoSuchObject(object))?; let collected = backend::collect(self.repo, tip, object, backend::MAX_GRAPH_CHANGES, since)?; match since { - None => backend::check_full_shape(&collected, object, E::HISTORY)?, - Some(since) => backend::check_delta_shape(&collected, object, since, E::HISTORY)?, + None => backend::check_full_shape(&collected, object, history)?, + Some(since) => backend::check_delta_shape(&collected, object, since, history)?, } - let changes = ChangeGraph::new(object, collected).into_ordered(); - Ok(Delta { changes, tip }) + Ok(Delta { + changes: ChangeGraph::new(object, collected).into_ordered(), + tip, + }) + } + + pub fn changes_since( + &self, + object: CobId, + since: Option, + ) -> Result { + self.changes_of(&E::Change::type_name(), object, E::HISTORY, since) } pub fn update_with_checkpointed( @@ -481,6 +729,7 @@ impl<'r> CobStore<'r> { ) -> Result where E: Checkpoint, + E::Change: KnotAuthored, E::State: Serialize + DeserializeOwned, D: From, { @@ -500,10 +749,11 @@ impl<'r> CobStore<'r> { ) -> Result, D> where E: Checkpoint, + E::Change: KnotAuthored, E::State: Serialize + DeserializeOwned, D: From, { - self.update_materialized_checkpointed::(home, object, signer, timestamp, |state| { + self.retrying_append::(home, object, signer, timestamp, |state| { decide(state).map(|change| change.map(Materialized::bare)) }) } @@ -516,6 +766,23 @@ impl<'r> CobStore<'r> { timestamp: UnixSeconds, decide: impl Fn(&E::State) -> Result>, D>, ) -> Result, D> + where + E: Checkpoint, + E::Change: KnotAuthored, + E::State: Serialize + DeserializeOwned, + D: From, + { + self.retrying_append::(home, object, signer, timestamp, decide) + } + + fn retrying_append( + &self, + home: &CobHome, + object: CobId, + signer: &dyn Signer, + timestamp: UnixSeconds, + decide: impl Fn(&E::State) -> Result>, D>, + ) -> Result, D> where E: Checkpoint, E::State: Serialize + DeserializeOwned, @@ -587,9 +854,19 @@ impl<'r> CobStore<'r> { let folded = object::fold_changes::(state, &delta.changes, &type_name)?; Ok((folded, tip, delta.changes.len())) } - Err(_) => self - .full_state::(object) - .map(|state| (state, tip, usize::MAX)), + Err( + unusable @ (CobError::DivergedTip { .. } + | CobError::DeltaNotDescended { .. }), + ) => { + tracing::warn!( + %object, + error = %unusable, + "checkpoint no longer bounds this object's history, fold starts over from its root change" + ); + self.full_state::(object) + .map(|state| (state, tip, usize::MAX)) + } + Err(error) => Err(error), } } None => self @@ -662,7 +939,10 @@ mod tests { use knot_git::{Layout, Repo}; use knot_runtime::{K256Signer, SeededEntropy, Signature, Signer}; - use knot_types::{RepoDid, UnixSeconds}; + use knot_types::{ + Blocked, ContributionPermission, ContributionPolicy, RecordBody, RecordCid, RepoDid, + RepoRole, UnixSeconds, + }; use proptest::prelude::*; use serde::{Deserialize, Serialize}; use tempfile::TempDir; @@ -680,6 +960,8 @@ mod tests { const TYPE: &'static str = "sh.tangled.test.tag"; } + impl KnotAuthored for Tag {} + #[test] fn checkpoint_encoding_stays_byte_stable() { let object = CobId::new(Oid::from_hex("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa").unwrap()); @@ -690,7 +972,7 @@ mod tests { let bytes = encode_checkpoint(object, tip, &state).unwrap(); assert_eq!( knot_types::lowercase_hex(&bytes), - "850478286161616161616161616161616161616161616161616161616161616161616161616161616161616178286262626262626262626262626262626262626262626262626262626262626262626262626262626278283861623531376164633234616530313138383962643634343930663836633562646333333632333782646b656c70657371756964" + "850578286161616161616161616161616161616161616161616161616161616161616161616161616161616178286262626262626262626262626262626262626262626262626262626262626262626262626262626278283861623531376164633234616530313138383962643634343930663836633562646333333632333782646b656c70657371756964" ); } @@ -729,6 +1011,8 @@ mod tests { const TYPE: &'static str = "sh.tangled.test.name"; } + impl KnotAuthored for NameChange {} + struct Names; impl Evaluate for Names { @@ -775,13 +1059,13 @@ mod tests { object: CobId, key: &K256Signer, who: &str, - when: i64, + when: UnixSeconds, ) -> Result, NameError> { store.update_maybe_checkpointed::( &cob_home(), object, key, - at(when), + when, |state| match state.contains(who) { true => Err(NameError::Taken), false => Ok(Some(NameChange::Claim(who.to_string()))), @@ -819,7 +1103,7 @@ mod tests { UnixSeconds::new(seconds) } - fn tag_root(repo: &Repo, key: &K256Signer, subject: &str, when: i64) -> ChangeId { + fn tag_root(repo: &Repo, key: &K256Signer, subject: &str, when: UnixSeconds) -> ChangeId { backend::write_change( &cob_home(), repo, @@ -828,7 +1112,7 @@ mod tests { &[], None, key, - at(when), + when, ) .unwrap() } @@ -839,7 +1123,7 @@ mod tests { subject: &str, parents: &[ChangeId], object: CobId, - when: i64, + when: UnixSeconds, ) -> ChangeId { backend::write_change( &cob_home(), @@ -849,7 +1133,7 @@ mod tests { parents, Some(object), key, - at(when), + when, ) .unwrap() } @@ -870,24 +1154,24 @@ mod tests { right: &str, merge: &str, ) -> CobId { - let root = tag_root(repo, key, "base", 1); + let root = tag_root(repo, key, "base", at(1)); let object = CobId::new(root.oid()); - let left = tag_child(repo, key, left, &[root], object, 2); - let right = tag_child(repo, key, right, &[root], object, 3); - let merge = tag_child(repo, key, merge, &[left, right], object, 4); + let left = tag_child(repo, key, left, &[root], object, at(2)); + let right = tag_child(repo, key, right, &[root], object, at(3)); + let merge = tag_child(repo, key, merge, &[left, right], object, at(4)); publish(repo, object, merge); object } fn linear_chain(repo: &Repo, key: &K256Signer, subjects: &[&str]) -> CobId { let (first, rest) = subjects.split_first().expect("chain needs a root subject"); - let root = tag_root(repo, key, first, 1); + let root = tag_root(repo, key, first, at(1)); let object = CobId::new(root.oid()); let tip = rest .iter() .enumerate() .fold(root, |parent, (index, subject)| { - tag_child(repo, key, subject, &[parent], object, index as i64 + 2) + tag_child(repo, key, subject, &[parent], object, at(index as i64 + 2)) }); publish(repo, object, tip); object @@ -970,7 +1254,7 @@ mod tests { object, &key, &format!("name{index:04}"), - index as i64, + at(index as i64), )); }); @@ -1035,7 +1319,7 @@ mod tests { object, &key, &format!("name{index:04}"), - index as i64, + at(index as i64), ) .unwrap(); }); @@ -1092,7 +1376,14 @@ mod tests { .unwrap() .object; (1..=Names::SNAPSHOT_STRIDE.get()).for_each(|step| { - claim(&store, object, &key, &format!("name{step:04}"), step as i64).unwrap(); + claim( + &store, + object, + &key, + &format!("name{step:04}"), + at(step as i64), + ) + .unwrap(); }); let real_state = store.get::(object).unwrap().state().clone(); @@ -1111,12 +1402,12 @@ mod tests { }) .unwrap(); - let duplicate = claim(&store, object, &key, "squid", 100); + let duplicate = claim(&store, object, &key, "squid", at(100)); assert!( matches!(duplicate, Err(NameError::Taken)), "corrupt checkpoint falls back to real state instead of waving a duplicate through" ); - appended(claim(&store, object, &key, "intruder", 101)); + appended(claim(&store, object, &key, "intruder", at(101))); let healed = repo.find_ref(&snapshot).unwrap().unwrap(); let bytes = repo.read_blob(healed).unwrap(); @@ -1219,7 +1510,6 @@ mod tests { "on a 1 GiB host the fold budget tracks the measured per-change cost near the serving limit, was {one_gib}" ); let roomy = backend::rebuild_change_limit_for(available(64 * 1024 * 1024 * 1024)); - assert_eq!(roomy, 64 * 1024 * 1024 * 1024 / 4 / 2560); assert!( roomy > backend::MAX_GRAPH_CHANGES, "a roomy host rebuilds far past the serving limit, was {roomy}" @@ -1275,10 +1565,6 @@ mod tests { let reloaded = store.get::(created.object).unwrap(); assert_eq!(object.state(), reloaded.state()); - let graph = store.graph::(created.object).unwrap(); - let again = store.graph::(created.object).unwrap(); - assert_eq!(graph.len(), again.len()); - assert_eq!(graph.causal_order(), again.causal_order()); let head = backend::read_change(&repo, tip).unwrap(); assert!(head.verify(&cob_home(), &head.author, Some(created.object))); @@ -1495,7 +1781,7 @@ mod tests { author: actor.clone(), signature: Signature::from_bytes(Vec::new()), payload: Payload::new(Vec::new()), - record: None, + record: Record::Bare, timestamp: UnixSeconds::new(index as i64), }; (id, change) @@ -1966,7 +2252,7 @@ mod tests { let (_other_dir, other_repo) = fixture(); let key = signer(53); let foreign = TypeName::new("sh.tangled.test.other").unwrap(); - let root = tag_root(&other_repo, &key, "nel", 1); + let root = tag_root(&other_repo, &key, "nel", at(1)); let object = CobId::new(root.oid()); let child = backend::write_change( &cob_home(), @@ -2022,7 +2308,7 @@ mod tests { let (dir, repo) = fixture(); let key = signer(500); let nsid = Tag::type_name(); - let root = tag_root(&repo, &key, "base", 1); + let root = tag_root(&repo, &key, "base", at(1)); let object = CobId::new(root.oid()); let child = |index: usize| { let (is_add, subject) = ops[index]; @@ -2086,19 +2372,825 @@ mod tests { prop_assert_eq!(&canonical, &model); prop_assert_eq!(&permuted, &model); } + } + + #[derive(Debug, Clone, Serialize, Deserialize)] + struct Opening { + address: knot_types::RecordAddress, + author: knot_types::AccountDid, + produces: RecordCid, + text: String, + } + + #[derive(Debug, Clone, Serialize, Deserialize)] + struct Editing { + author: knot_types::AccountDid, + from: RecordCid, + produces: RecordCid, + text: String, + } + + #[derive(Debug, Clone, Serialize, Deserialize)] + struct Erasure { + author: knot_types::AccountDid, + from: RecordCid, + } + + #[derive(Debug, Clone, Serialize, Deserialize)] + #[serde(tag = "op", content = "data", rename_all = "snake_case")] + enum NoteChange { + Open(Opening), + Edit(Editing), + Erase(Erasure), + } + + impl ChangePayload for NoteChange { + const TYPE: &'static str = "sh.tangled.test.note"; + + fn address(&self) -> Option { + match self { + Self::Open(opening) => Some(opening.address.clone()), + Self::Edit(_) | Self::Erase(_) => None, + } + } + + fn emits(&self) -> Emits { + match self { + Self::Open(opening) => Emits::Version(opening.produces), + Self::Edit(editing) => Emits::Version(editing.produces), + Self::Erase(_) => Emits::Tombstone, + } + } + + fn collection() -> Option { + knot_types::RecordCollection::new(Self::TYPE).ok() + } + } + + impl Authored for NoteChange { + fn author(&self) -> &knot_types::AccountDid { + match self { + Self::Open(opening) => &opening.author, + Self::Edit(editing) => &editing.author, + Self::Erase(erasure) => &erasure.author, + } + } + } + + #[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)] + struct Note { + author: Option, + version: Option, + text: Vec, + } + + struct Notes; + + impl Evaluate for Notes { + type State = Note; + type Change = NoteChange; + + const HISTORY: HistoryModel = HistoryModel::Convergent; + + fn initial() -> Self::State { + Note::default() + } + + fn apply(mut state: Self::State, change: Self::Change, _author: &ActorId) -> Self::State { + match change { + NoteChange::Open(opening) => { + state.author = Some(opening.author); + state.version = Some(opening.produces); + state.text.push(opening.text); + } + NoteChange::Edit(editing) => { + state.version = Some(editing.produces); + state.text.push(editing.text); + } + NoteChange::Erase(_) => { + state.version = None; + state.text.clear(); + } + } + state + } + } + + impl Checkpoint for Notes { + const SNAPSHOT_STRIDE: SnapshotStride = SnapshotStride::new(4); + fn checkpoint_size(state: &Self::State) -> StateSize { + StateSize::new(state.text.len()) + } + } + + impl Authorize for Notes { + fn may_create(cx: &WriteContext<'_>) -> Decision { + cx.permission().contributes() + } + + fn may_append(cx: &WriteContext<'_>, state: &Self::State, change: &Self::Change) -> Append { + let owns = state.author.as_ref() == Some(cx.actor()); + match ( + change, + owns || cx.permission().moderates().is_allowed(), + state.version, + ) { + (NoteChange::Open(_), ..) | (_, false, _) => Append::Deny, + (_, true, None) => Append::Deny, + ( + NoteChange::Edit(Editing { from, .. }) + | NoteChange::Erase(Erasure { from, .. }), + true, + Some(current), + ) if *from != current => Append::Stale { current }, + (NoteChange::Edit(_) | NoteChange::Erase(_), true, Some(_)) => Append::Allow, + } + } + } + + fn note_address(rkey: &str) -> knot_types::RecordAddress { + knot_types::RecordAddress::new( + knot_types::RecordCollection::new("sh.tangled.test.note").unwrap(), + knot_types::RecordRkey::new(rkey).unwrap(), + ) + } + + fn who(suffix: &str) -> knot_types::AccountDid { + knot_types::AccountDid::new(format!("did:plc:{suffix}")).unwrap() + } + + fn note_record(text: &str) -> (RecordBody, RecordCid) { + let body = RecordBody::new( + serde_ipld_dagcbor::to_vec(&std::collections::BTreeMap::from([( + "text", + text.to_owned(), + )])) + .unwrap(), + ) + .unwrap(); + let cid = body.cid(); + (body, cid) + } + + fn opened(author: &str, rkey: &str, text: &str) -> Materialized { + let (body, produces) = note_record(text); + Materialized { + change: NoteChange::Open(Opening { + address: note_address(rkey), + author: who(author), + produces, + text: text.to_owned(), + }), + record: Record::version(body), + } + } + + fn edited(author: &str, from: RecordCid, text: &str) -> Materialized { + let (body, produces) = note_record(text); + Materialized { + change: NoteChange::Edit(Editing { + author: who(author), + from, + produces, + text: text.to_owned(), + }), + record: Record::version(body), + } + } + + fn erased(author: &str, from: RecordCid) -> Materialized { + Materialized { + change: NoteChange::Erase(Erasure { + author: who(author), + from, + }), + record: Record::Tombstone, + } + } + fn with_record( + materialized: Materialized, + record: Record, + ) -> Materialized { + Materialized { + change: materialized.change, + record, + } + } + + fn squid() -> RepoDid { + RepoDid::new("did:plc:squid").unwrap() + } + + fn context<'a>( + actor: &'a knot_types::AccountDid, + repo: &'a RepoDid, + role: RepoRole, + ) -> WriteContext<'a> { + WriteContext::new( + actor, + repo, + ContributionPermission::new(Blocked::No, role, ContributionPolicy::Anyone), + ) + } + + fn note_append( + store: &CobStore, + object: CobId, + key: &K256Signer, + materialized: Materialized, + when: UnixSeconds, + ) -> Result, CobError> { + store.retrying_append::(&cob_home(), object, key, when, |_state| { + Ok(Some(materialized.clone())) + }) + } + fn opened_note(seed: u64) -> (TempDir, Repo, Created, RecordCid) { + let (dir, repo) = fixture(); + let first = opened("nel", "3lubrptx57d22", "kelp"); + let live = first.change.emits().version().unwrap(); + let created = CobStore::new(&repo) + .open(&cob_home(), &first, &signer(seed), at(1)) + .unwrap(); + (dir, repo, created, live) + } + + #[test] + fn a_bare_change_and_a_tombstoned_record_stay_apart_through_git() { + let (_dir, repo, created, opened_cid) = opened_note(11); + let key = signer(11); + let store = CobStore::new(&repo); + note_append( + &store, + created.object, + &key, + erased("nel", opened_cid), + at(2), + ) + .unwrap(); + + let records: Vec = store + .changes_since::(created.object, None) + .unwrap() + .changes + .into_iter() + .map(|change| change.record) + .collect(); + assert_eq!( + records, + vec![Record::version(note_record("kelp").0), Record::Tombstone,], + "version this change wrote, and tombstone that erased it, live apart in git" + ); + + let tags = linear_chain(&repo, &key, &["base"]); + let bare: Vec = store + .changes_since::(tags, None) + .unwrap() + .changes + .into_iter() + .map(|change| change.record) + .collect(); + assert_eq!( + bare, + vec![Record::Bare], + "change that doesn't emit records, and change that erases one, are different facts" + ); + } + + #[test] + fn a_change_cant_move_or_invent_the_address_its_object_was_opened_at() { + let (_dir, repo, created, live) = opened_note(12); + let key = signer(12); + let store = CobStore::new(&repo); + + let moved = note_append( + &store, + created.object, + &key, + opened("nel", "3lubrptx57d33", "elsewhere"), + at(2), + ); + assert!( + matches!(moved, Err(CobError::AddressMoved { .. })), + "Object materializes at one address for its whole life, got {moved:?}" + ); + assert!( + note_append( + &store, + created.object, + &key, + opened("nel", "3lubrptx57d22", "same place"), + at(3), + ) + .is_ok(), + "Root's address is the one later change may name" + ); + + let bare = store + .open(&cob_home(), &erased("nel", live), &key, at(1)) + .unwrap(); + let invented = note_append( + &store, + bare.object, + &key, + opened("nel", "3lubrptx57d44", "late"), + at(2), + ); + assert!( + matches!(invented, Err(CobError::AddressAdded { .. })), + "Object with no record can't grow one after its root, got {invented:?}" + ); + } + + #[test] + fn a_declared_record_and_the_stored_one_are_one_or_the_write_stops() { + let (_dir, repo) = fixture(); + let key = signer(13); + let store = CobStore::new(&repo); + let (_, claimed) = note_record("kelp"); + let mismatched = store.open( + &cob_home(), + &with_record( + opened("nel", "3lubrptx57d22", "kelp"), + Record::version(note_record("someone else's words").0), + ), + &key, + at(1), + ); + assert!( + matches!(mismatched, Err(CobError::RecordMisclaimed { .. })), + "named version and stored version differ, got {mismatched:?}" + ); + let unwritten = store.open( + &cob_home(), + &with_record(opened("nel", "3lubrptx57d22", "kelp"), Record::Bare), + &key, + at(1), + ); + assert!( + matches!(unwritten, Err(CobError::RecordUnwritten { .. })), + "Change that names its record has to store said record, got {unwritten:?}" + ); + let erasing = store.open( + &cob_home(), + &with_record( + erased("nel", claimed), + Record::version(note_record("kelp").0), + ), + &key, + at(1), + ); + assert!( + matches!(erasing, Err(CobError::TombstoneUnwritten)), + "Change that erases its record stores tombstone, not version, got {erasing:?}" + ); + } + + #[test] + fn a_write_the_actor_didnt_author_never_reaches_git() { + let nel = who("nel"); + let repo_did = squid(); + let cx = context(&nel, &repo_did, RepoRole::Stranger); + let impersonation = Approved::signed_by(&cx, opened("olaren", "3lubrptx57d22", "kelp")); + assert!( + matches!(impersonation, Err(CobError::Impersonation { .. })), + "Engine, not handler, is where stranger's byline gets turned away" + ); + assert!(Approved::signed_by(&cx, opened("nel", "3lubrptx57d22", "kelp")).is_ok()); + } + + #[test] + fn the_policy_opens_an_object_and_the_author_or_a_moderator_edits_it() { + let (_dir, repo) = fixture(); + let key = signer(14); + let store = CobStore::new(&repo); + let (nel, olaren) = (who("nel"), who("olaren")); + + let repo_did = squid(); + let closed = WriteContext::new( + &nel, + &repo_did, + ContributionPermission::new( + Blocked::No, + RepoRole::Stranger, + ContributionPolicy::Collaborators, + ), + ); + let refused = store.create_approved::( + &cob_home(), + &Approved::signed_by(&closed, opened("nel", "3lubrptx57d22", "kelp")).unwrap(), + &key, + at(1), + ); + let sentence = refused.as_ref().unwrap_err().to_string(); + assert!( + matches!(refused, Err(CobError::CreateRejected { .. })) + && sentence.contains("stranger, where contribution policy is collaborators"), + "Stranger can't open objects on a collaborator-only repository, \ + and 403 sentence names policy, got {sentence}" + ); + + let author = context(&nel, &repo_did, RepoRole::Stranger); + let first = opened("nel", "3lubrptx57d22", "kelp"); + let live = first.change.emits().version().unwrap(); + let created = store + .create_approved::( + &cob_home(), + &Approved::signed_by(&author, first).unwrap(), + &key, + at(1), + ) + .unwrap(); + + let stranger = context(&olaren, &repo_did, RepoRole::Stranger); + let intruded = store.append_approved::( + &cob_home(), + created.object, + &Approved::signed_by(&stranger, edited("olaren", live, "not yours")).unwrap(), + &key, + at(2), + ); + assert!( + matches!(intruded, Err(CobError::AppendRejected { .. })), + "Edit is author's, got {intruded:?}" + ); + + let maintainer = context(&olaren, &repo_did, RepoRole::Collaborator); + assert!( + store + .append_approved::( + &cob_home(), + created.object, + &Approved::signed_by(&maintainer, erased("olaren", live)).unwrap(), + &key, + at(3), + ) + .is_ok(), + "Maintainer override is what makes the operator able to moderate" + ); + } + + #[test] + fn two_edits_from_one_folded_state_end_as_one_write_and_one_conflict() { + let dir = tempfile::tempdir().unwrap(); + let layout = Layout::new(dir.path()); + let did = squid(); + let repo = layout.create(&did).unwrap(); + let key = signer(15); + let nel = who("nel"); + let cx = context(&nel, &did, RepoRole::Stranger); + let first = opened("nel", "3lubrptx57d22", "kelp"); + let live = first.change.emits().version().unwrap(); + let object = CobStore::new(&repo) + .create_approved::( + &cob_home(), + &Approved::signed_by(&cx, first).unwrap(), + &key, + at(1), + ) + .unwrap() + .object; + drop(repo); + + let outcomes: Vec> = std::thread::scope(|scope| { + let handles: Vec<_> = ["one", "two"] + .into_iter() + .map(|text| { + let layout = layout.clone(); + let did = &did; + let key = &key; + let nel = &nel; + scope.spawn(move || { + let repo = layout.open(did).unwrap(); + let cx = context(nel, did, RepoRole::Stranger); + CobStore::new(&repo).append_approved::( + &cob_home(), + object, + &Approved::signed_by(&cx, edited("nel", live, text)).unwrap(), + key, + at(2), + ) + }) + }) + .collect(); + handles + .into_iter() + .map(|handle| handle.join().expect("Edit thread runs to its answer")) + .collect() + }); + + assert_eq!( + outcomes.iter().filter(|outcome| outcome.is_ok()).count(), + 1, + "one edit from a shared version is stored, got {outcomes:?}" + ); + let expected_uri = note_address("3lubrptx57d22").at_uri(&did); + assert_eq!( + outcomes + .iter() + .filter(|outcome| matches!( + outcome, + Err(CobError::StaleRecord { current, uri, .. }) + if *current != live && *uri == expected_uri + )) + .count(), + 1, + "loser reads at-uri and current version, for rebasing, got {outcomes:?}" + ); + } + + fn note_change( + repo: &Repo, + key: &K256Signer, + materialized: &Materialized, + parents: &[ChangeId], + object: Option, + when: UnixSeconds, + ) -> ChangeId { + backend::write_change_with_record( + &cob_home(), + repo, + &NoteChange::type_name(), + &materialized.change.encode().unwrap(), + &materialized.record, + parents, + object, + key, + when, + ) + .unwrap() + } + + #[test] + fn a_forked_social_history_folds_through_the_convergent_arm() { + let (_dir, repo) = fixture(); + let key = signer(16); + let store = CobStore::new(&repo); + let root_change = opened("nel", "3lubrptx57d22", "kelp"); + let live = root_change.change.emits().version().unwrap(); + let root = note_change(&repo, &key, &root_change, &[], None, at(1)); + let object = CobId::new(root.oid()); + + let left_change = edited("nel", live, "left"); + let right_change = edited("nel", live, "right"); + let left = note_change(&repo, &key, &left_change, &[root], Some(object), at(2)); + let right = note_change(&repo, &key, &right_change, &[root], Some(object), at(3)); + let merge_change = edited("nel", live, "merged"); + let merge = note_change( + &repo, + &key, + &merge_change, + &[left, right], + Some(object), + at(4), + ); + repo.update_ref(&RefUpdate::Create { + name: backend::cob_ref_name(&NoteChange::type_name(), object).unwrap(), + new: merge.oid(), + }) + .unwrap(); + + let folded = store.get::(object).unwrap(); + assert_eq!( + folded.state().text, + vec!["kelp", "left", "right", "merged"], + "Causal order puts fork's branches in timestamp order under merge" + ); + assert_eq!( + folded.state().version, + merge_change.change.emits().version(), + "the current version is the merge's, which fold order reaches last" + ); + let again = store.get::(object).unwrap(); + assert_eq!( + serde_ipld_dagcbor::to_vec(folded.state()).unwrap(), + serde_ipld_dagcbor::to_vec(again.state()).unwrap(), + "Second fold of one forked history reaches byte-identical state" + ); + } + + fn note_fan(creation_order: &[usize]) -> (TempDir, Repo, CobId) { + let (dir, repo) = fixture(); + let key = signer(600); + let root_change = opened("nel", "3lubrptx57d22", "kelp"); + let live = root_change.change.emits().version().unwrap(); + let root = note_change(&repo, &key, &root_change, &[], None, at(1)); + let object = CobId::new(root.oid()); + let edit = |index: usize| { + let change = edited("nel", live, &format!("edit{index}")); + note_change( + &repo, + &key, + &change, + &[root], + Some(object), + at(index as i64 + 2), + ) + }; + creation_order.iter().for_each(|&index| { + edit(index); + }); + let parents: Vec = (0..creation_order.len()).map(edit).collect(); + let merge = note_change( + &repo, + &key, + &edited("nel", live, "merged"), + &parents, + Some(object), + at(creation_order.len() as i64 + 2), + ); + repo.update_ref(&RefUpdate::Create { + name: backend::cob_ref_name(&NoteChange::type_name(), object).unwrap(), + new: merge.oid(), + }) + .unwrap(); + (dir, repo, object) + } + + proptest! { + #![proptest_config(ProptestConfig { cases: 32, ..ProptestConfig::default() })] #[test] - fn prop_rematerialization_is_idempotent((ops, keys) in ops_and_perm()) { - let (_dir, repo, object) = build_repo(&ops, &permutation(&keys)); - let store = CobStore::new(&repo); - let first = store.get::(object).unwrap(); - let second = store.get::(object).unwrap(); - prop_assert_eq!(first.state(), second.state()); - prop_assert_eq!(first.history().len(), second.history().len()); + fn prop_a_social_fork_folds_the_same_whatever_order_its_changes_arrived( + keys in prop::collection::vec(any::(), 1..8usize) + ) { + let identity: Vec = (0..keys.len()).collect(); + let (_canon_dir, canon_repo, canon_object) = note_fan(&identity); + let canonical = CobStore::new(&canon_repo).get::(canon_object).unwrap(); + let (_perm_dir, perm_repo, perm_object) = note_fan(&permutation(&keys)); + let permuted = CobStore::new(&perm_repo).get::(perm_object).unwrap(); + + let expected: Vec = std::iter::once("kelp".to_owned()) + .chain((0..keys.len()).map(|index| format!("edit{index}"))) + .chain(std::iter::once("merged".to_owned())) + .collect(); + prop_assert_eq!(&canonical.state().text, &expected); prop_assert_eq!( - store.graph::(object).unwrap().causal_order(), - store.graph::(object).unwrap().causal_order() + serde_ipld_dagcbor::to_vec(canonical.state()).unwrap(), + serde_ipld_dagcbor::to_vec(permuted.state()).unwrap(), + "Fold order is causal sort, not the order git took objects in" ); } } + + #[test] + fn a_fork_of_one_edit_folds_root_edit_and_merge_once_each() { + let (_dir, repo, object) = note_fan(&[0]); + let folded = CobStore::new(&repo).get::(object).unwrap(); + assert_eq!( + folded.state().text, + vec!["kelp", "edit0", "merged"], + "Fold order is causal sort, not the order git took objects in" + ); + } + + #[test] + fn a_merge_that_forks_below_a_checkpoint_folds_from_the_root_instead_of_double_counting() { + let (_dir, repo, created, mut live) = opened_note(17); + let key = signer(17); + let store = CobStore::new(&repo); + let object = created.object; + let below = (2..=Notes::SNAPSHOT_STRIDE.get() as i64).fold(created.tip, |_, step| { + let edit = edited("nel", live, &format!("step{step}")); + live = edit.change.emits().version().unwrap(); + note_append(&store, object, &key, edit, at(step)) + .unwrap() + .expect("Edit is stored") + }); + let snapshot = backend::checkpoint_ref_name(&NoteChange::type_name(), object).unwrap(); + assert!( + repo.find_ref(&snapshot).unwrap().is_some(), + "stride has to have written its checkpoint, or this test doesn't prove anything" + ); + + let root = ChangeId::new(object.oid()); + let side_change = edited("nel", live, "side"); + let side = note_change(&repo, &key, &side_change, &[root], Some(object), at(20)); + let merge_change = edited("nel", live, "merged"); + let merge = note_change( + &repo, + &key, + &merge_change, + &[below, side], + Some(object), + at(21), + ); + repo.update_ref(&RefUpdate::Update { + name: backend::cob_ref_name(&NoteChange::type_name(), object).unwrap(), + old: below.oid(), + new: merge.oid(), + }) + .unwrap(); + + let (materialized, tip) = store.materialize::(object).unwrap(); + let folded = store.get::(object).unwrap(); + assert_eq!(tip, merge); + assert_eq!( + materialized, + *folded.state(), + "checkpoint is abandoned, rather than trusted, delta under that mr checkpoint would count its prefix twice" + ); + assert_eq!( + materialized.text.len(), + Notes::SNAPSHOT_STRIDE.get() + 2, + "Fold counts each change once: root, its edits, side and merge" + ); + } + + #[test] + fn a_failed_chain_leaves_the_prefix_another_writer_already_published() { + let (_dir, repo, created, live) = opened_note(18); + let key = signer(18); + let store = CobStore::new(&repo); + + let shared = edited("nel", live, "shared"); + let published = note_append(&store, created.object, &key, shared.clone(), at(2)) + .unwrap() + .expect("First writer stores its edit"); + + let refused = store.chain( + &cob_home(), + created.object, + &NoteChange::type_name(), + created.tip, + [ + (&shared.change, &shared.record, at(2)), + ( + &edited("nel", live, "unwritten").change, + &Record::Bare, + at(3), + ), + ], + &key, + ); + assert!( + matches!(refused, Err(CobError::RecordUnwritten { .. })), + "Second change of chain names record, which said change doesn't store, got {refused:?}" + ); + + assert_eq!( + backend::resolve_tip(&repo, &NoteChange::type_name(), created.object).unwrap(), + Some(published.oid()), + "Loser's cleanup left winner's tip alone" + ); + assert!( + store.get::(created.object).is_ok(), + "two chains sharing one prefix are content-addressed to one commit. Loser \ + mustn't pull it out from under winner" + ); + } + + #[test] + fn a_change_past_the_byte_limit_doesnt_reach_git() { + let (_dir, repo) = fixture(); + let key = signer(19); + let store = CobStore::new(&repo); + let fat = "x".repeat(knot_types::MAX_RECORD_BYTES + 1); + let refused = store.create(&cob_home(), &Tag::Add(fat), &key, at(1)); + assert!( + matches!( + refused, + Err(CobError::ChangeTooLarge { + blob: "payload", + .. + }) + ), + "Payload past limit is refused before blob is written, got {refused:?}" + ); + assert!( + loose_commits(&repo).is_empty(), + "Change store turned away didn't reach any commit" + ); + } + + #[test] + fn a_replayed_create_answers_with_the_object_it_already_made() { + let (_dir, repo) = fixture(); + let key = signer(20); + let store = CobStore::new(&repo); + let first = store + .create(&cob_home(), &Tag::Add("base".into()), &key, at(1)) + .unwrap(); + let replayed = store + .create(&cob_home(), &Tag::Add("base".into()), &key, at(1)) + .unwrap(); + assert_eq!( + replayed, first, + "content-addressed create makes client's retry, after timeout, return same object, rather than error about ref that already exists" + ); + + store + .update( + &cob_home(), + first.object, + &Tag::Add("later".into()), + &key, + at(2), + ) + .unwrap(); + let grown = store.create(&cob_home(), &Tag::Add("base".into()), &key, at(1)); + assert!( + matches!(grown, Err(CobError::StaleTip { .. })), + "Replay of root whose object has moved on conflicts, got {grown:?}" + ); + } }