diff --git a/knot2/crates/knot-record/Cargo.toml b/knot2/crates/knot-record/Cargo.toml index 4098f7c02..5bb36c5f3 100644 --- a/knot2/crates/knot-record/Cargo.toml +++ b/knot2/crates/knot-record/Cargo.toml @@ -22,6 +22,7 @@ thiserror = { workspace = true } tracing = { workspace = true } iroh-car = { workspace = true } [dev-dependencies] +proptest = { workspace = true } serde_json = { workspace = true } tempfile = { workspace = true } tokio = { workspace = true } diff --git a/knot2/crates/knot-record/src/chain.rs b/knot2/crates/knot-record/src/chain.rs index db9acd391..7648d12e7 100644 --- a/knot2/crates/knot-record/src/chain.rs +++ b/knot2/crates/knot-record/src/chain.rs @@ -1,3 +1,4 @@ +use std::cmp::Ordering; use std::collections::{BTreeMap, BTreeSet}; use std::str::FromStr; use std::sync::Arc; @@ -6,7 +7,7 @@ use crate::blocks::{self, GitBlocks}; use crate::mst; use bytes::Bytes; use cid::Cid; -use futures::StreamExt; +use futures::{StreamExt, TryStreamExt}; use jacquard_common::types::did::Did; use jacquard_common::types::tid::{Ticker, Tid}; use jacquard_repo::Mst; @@ -15,21 +16,25 @@ use jacquard_repo::mst::VerifiedWriteOp; use jacquard_repo::storage::BlockStore; use knot_git::{Identity, NewCommit, RefUpdate, Repo}; use knot_runtime::Signer; -use knot_types::{AuthorName, Email, Oid, RefName, UnixSeconds}; +use knot_types::{ + AtUri, AuthorName, Email, MstKey, Oid, RecordAddress, RecordBody, RecordCid, RecordCollection, + RefName, RepoDid, UnixSeconds, +}; pub const CHAIN_REF: &str = "refs/atproto/commit"; const COMMIT_BLOB: &str = "commit"; pub const FRAME_BLOB: &str = "frame"; pub const MAX_OPS_PER_COMMIT: usize = 200; +pub const MAX_OPS_PER_APPLY: usize = 2_000; pub const MAX_COMMIT_BLOCKS_BYTES: usize = 2_000_000; -pub const MAX_RECORD_BLOCK_BYTES: usize = 1_000_000; pub const MAX_FRAME_BYTES: usize = 5_000_000; const OP_FRAME_OVERHEAD_BYTES: usize = 256; pub trait Emit { fn rev(&mut self, prev: Option<&Tid>) -> Tid; - fn frame(&mut self, chunk: &CommitChunk) -> Option; + fn frame(&self, chunk: &CommitChunk) -> Option; + fn committed(&mut self, frame: Option); } impl Emit for Ticker { @@ -37,24 +42,55 @@ impl Emit for Ticker { self.next(prev.cloned()) } - fn frame(&mut self, _chunk: &CommitChunk) -> Option { + fn frame(&self, _chunk: &CommitChunk) -> Option { None } + + fn committed(&mut self, _frame: Option) {} } + #[derive(Debug, thiserror::Error)] pub enum ChainError { #[error("record layer error: {0}")] Record(#[from] crate::RecordError), #[error("atproto block error: {0}")] Block(#[from] jacquard_repo::RepoError), - #[error("a record block of {0} bytes is over the {MAX_RECORD_BLOCK_BYTES} byte limit")] - RecordTooLarge(usize), #[error("a single record still exceeds the {MAX_COMMIT_BLOCKS_BYTES} byte blocks limit")] ChunkTooLarge, #[error("the chain at {CHAIN_REF} is already materialized")] AlreadyMaterialized, #[error("the chain at {CHAIN_REF} isn't materialized")] NotEnabled, + #[error("{uri} already has record at {current}, please update it instead of creating")] + RecordExists { + uri: AtUri, + current: RecordCid, + }, + #[error("{uri} has no record to write over")] + RecordAbsent { uri: AtUri }, + #[error("{uri} is at {current}, and write was made from {named}, please rebase from {current}")] + RecordMoved { + uri: AtUri, + current: RecordCid, + named: RecordCid, + }, + #[error("{address} takes one op per chain write, and this one names op twice")] + RecordContested { address: RecordAddress }, + #[error("Chain write takes {limit} ops at a time and this one has {ops}")] + TooManyOps { ops: usize, limit: usize }, + #[error("{address} has {found}, which isn't dag-cbor record CID")] + LeafForeign { + address: RecordAddress, + found: Box, + }, + #[error( + "{written} of {planned} commits were written, before this chain write stopped: {reason}" + )] + PartlyApplied { + written: usize, + planned: usize, + reason: Box, + }, #[error("git error: {0}")] Git(#[from] knot_git::GitError), #[error("malformed atproto chain: {0}")] @@ -275,20 +311,13 @@ fn raw_commit_bytes(repo: &Repo, git_oid: Oid) -> Result, ChainError> { )) } -fn record_too_large(records: &BTreeMap>) -> Option { - records - .values() - .map(Vec::len) - .find(|len| *len > MAX_RECORD_BLOCK_BYTES) -} - -fn did_of(did: &str) -> Result, ChainError> { - Did::::from_str(did).map_err(|error| ChainError::Malformed(error.to_string())) +fn did_of(did: &RepoDid) -> Result, ChainError> { + Did::::from_str(did.as_str()).map_err(|error| ChainError::Malformed(error.to_string())) } pub async fn initialize( repo: &Repo, - did: &str, + did: &RepoDid, signer: &dyn Signer, emit: &mut dyn Emit, now: UnixSeconds, @@ -312,21 +341,26 @@ pub async fn initialize( }; let frame = emit.frame(&chunk); let blocks_tree = blocks::merge_tree(repo, None, &chunk.blocks)?; - write_chain_commit(repo, None, &chunk.commit, now, Some(blocks_tree), frame)?; + write_chain_commit( + repo, + None, + &chunk.commit, + now, + Some(blocks_tree), + frame.clone(), + )?; + emit.committed(frame); Ok(chunk) } pub async fn materialize( repo: &Repo, - did: &str, - records: &BTreeMap>, + did: &RepoDid, + records: &BTreeMap, signer: &dyn Signer, emit: &mut dyn Emit, now: UnixSeconds, ) -> Result, ChainError> { - if let Some(len) = record_too_large(records) { - return Err(ChainError::RecordTooLarge(len)); - } if tip(repo)?.is_some() { return Err(ChainError::AlreadyMaterialized); } @@ -339,13 +373,13 @@ pub async fn materialize( tree: mst::empty(), }; let ops: Vec = records - .keys() - .map(|key| VerifiedWriteOp::Create { + .iter() + .map(|(key, body)| VerifiedWriteOp::Create { key: key.as_str().into(), - cid: crate::cid_of(&records[key]), + cid: crate::cid_of(body.as_bytes()), }) .collect(); - let chunks = plan_chunks(&mut plan, records, ops).await?; + let chunks = plan_chunks(&mut plan, records, records.keys().cloned().collect(), ops).await?; append_commits(repo, None, None, &chunks, now, emit)?; Ok(chunks) } @@ -358,37 +392,48 @@ fn append_commits( now: UnixSeconds, emit: &mut dyn Emit, ) -> Result<(), ChainError> { - chunks - .iter() - .try_fold((parent, blocks_tree), |(parent, tree), chunk| { - let grown = blocks::merge_tree(repo, tree, &chunk.blocks)?; - let commit = write_chain_commit( - repo, - parent, - &chunk.commit, - now, - Some(grown), - emit.frame(chunk), - )?; - Ok::<_, ChainError>((Some(commit), Some(grown))) - })?; - Ok(()) + let written = chunks.iter().try_fold( + (parent, blocks_tree, 0usize), + |(parent, tree, written), chunk| { + let frame = emit.frame(chunk); + let step = blocks::merge_tree(repo, tree, &chunk.blocks).and_then(|grown| { + write_chain_commit(repo, parent, &chunk.commit, now, Some(grown), frame.clone()) + .map(|commit| (Some(commit), Some(grown), written + 1)) + }); + match step { + Ok(next) => { + emit.committed(frame); + Ok(next) + } + Err(error) => Err((written, error)), + } + }, + ); + match written { + Ok(_) => Ok(()), + Err((0, error)) => Err(error), + Err((written, error)) => Err(ChainError::PartlyApplied { + written, + planned: chunks.len(), + reason: Box::new(error), + }), + } } pub async fn sync( repo: &Repo, - did: &str, - collection: &str, - records: &BTreeMap>, + did: &RepoDid, + collection: &RecordCollection, + records: &BTreeMap, signer: &dyn Signer, emit: &mut dyn Emit, now: UnixSeconds, ) -> Result, ChainError> { - if let Some(len) = record_too_large(records) { - return Err(ChainError::RecordTooLarge(len)); - } - let prefix = format!("{collection}/"); - if let Some(key) = records.keys().find(|key| !key.starts_with(&prefix)) { + let prefix = format!("{}/", collection.as_str()); + if let Some(key) = records + .keys() + .find(|key| !key.as_str().starts_with(&prefix)) + { return Err(ChainError::Malformed(format!( "the record key {key} sits outside the {collection} collection this chain syncs" ))); @@ -397,16 +442,17 @@ pub async fn sync( return Err(ChainError::NotEnabled); }; let before = load_tree(repo, &tip); - let leaves: BTreeMap = before + let leaves: BTreeMap = before .leaves() .await? .into_iter() - .map(|(key, cid)| (key.as_str().to_owned(), cid)) - .collect(); + .map(|(key, cid)| MstKey::new(key.as_str().to_owned()).map(|key| (key, cid))) + .collect::, _>>() + .map_err(|error| ChainError::Malformed(error.to_string()))?; let mut ops: Vec = records .iter() - .filter_map(|(key, bytes)| { - let cid = crate::cid_of(bytes); + .filter_map(|(key, body)| { + let cid = crate::cid_of(body.as_bytes()); match leaves.get(key) { None => Some(VerifiedWriteOp::Create { key: key.as_str().into(), @@ -424,7 +470,7 @@ pub async fn sync( ops.extend( leaves .iter() - .filter(|(key, _)| key.starts_with(&prefix) && !records.contains_key(*key)) + .filter(|(key, _)| key.as_str().starts_with(&prefix) && !records.contains_key(*key)) .map(|(key, prev)| VerifiedWriteOp::Delete { key: key.as_str().into(), prev: *prev, @@ -441,7 +487,132 @@ pub async fn sync( prev_data: Some(tip.data), tree: before, }; - let chunks = plan_chunks(&mut plan, records, ops).await?; + let chunks = plan_chunks(&mut plan, records, records.keys().cloned().collect(), ops).await?; + append_commits(repo, Some(tip.git), Some(tip.blocks), &chunks, now, emit)?; + Ok(chunks) +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RecordOp { + Create { record: RecordBody }, + Update { record: RecordBody, from: RecordCid }, + Delete { from: RecordCid }, +} + +impl RecordOp { + fn record(&self) -> Option<&RecordBody> { + match self { + Self::Create { record } | Self::Update { record, .. } => Some(record), + Self::Delete { .. } => None, + } + } +} + +fn adopt(address: &RecordAddress, current: Cid) -> Result { + RecordCid::new(current).ok_or_else(|| ChainError::LeafForeign { + address: address.clone(), + found: Box::new(current), + }) +} + +fn write_op( + repo: &RepoDid, + address: &RecordAddress, + op: &RecordOp, + current: Option, +) -> Result { + let key = address.mst_key(); + match (op, current) { + (RecordOp::Create { record }, None) => Ok(VerifiedWriteOp::Create { + key: key.as_str().into(), + cid: record.cid().cid(), + }), + (RecordOp::Create { .. }, Some(current)) => Err(ChainError::RecordExists { + uri: address.at_uri(repo), + current: adopt(address, current)?, + }), + (RecordOp::Update { .. } | RecordOp::Delete { .. }, None) => { + Err(ChainError::RecordAbsent { + uri: address.at_uri(repo), + }) + } + (RecordOp::Update { from, .. } | RecordOp::Delete { from }, Some(current)) + if from.cid() != current => + { + Err(ChainError::RecordMoved { + uri: address.at_uri(repo), + current: adopt(address, current)?, + named: *from, + }) + } + (RecordOp::Update { record, .. }, Some(current)) => Ok(VerifiedWriteOp::Update { + key: key.as_str().into(), + cid: record.cid().cid(), + prev: current, + }), + (RecordOp::Delete { .. }, Some(current)) => Ok(VerifiedWriteOp::Delete { + key: key.as_str().into(), + prev: current, + }), + } +} + +pub async fn apply( + repo: &Repo, + did: &RepoDid, + ops: &[(RecordAddress, RecordOp)], + signer: &dyn Signer, + emit: &mut dyn Emit, + now: UnixSeconds, +) -> Result, ChainError> { + if ops.len() > MAX_OPS_PER_APPLY { + return Err(ChainError::TooManyOps { + ops: ops.len(), + limit: MAX_OPS_PER_APPLY, + }); + } + let all_keys = ops + .iter() + .try_fold(BTreeSet::::new(), |mut seen, (address, _)| { + seen.insert(address.mst_key()) + .then_some(seen) + .ok_or_else(|| ChainError::RecordContested { + address: address.clone(), + }) + })?; + let Some(tip) = tip(repo)? else { + return Err(ChainError::NotEnabled); + }; + let before = load_tree(repo, &tip); + let verified = futures::stream::iter(ops) + .then(|(address, op)| { + let before = &before; + async move { + let current = before.get(address.mst_key().as_str()).await?; + write_op(did, address, op, current) + } + }) + .try_collect::>() + .await?; + if verified.is_empty() { + return Ok(Vec::new()); + } + let records: BTreeMap = ops + .iter() + .filter_map(|(address, op)| { + op.record() + .map(|record| (address.mst_key(), record.clone())) + }) + .collect(); + let mut plan = Plan { + did: did_of(did)?, + signer, + emit, + prev_rev: Some(tip.rev.clone()), + prev_data: Some(tip.data), + tree: before, + }; + let chunks = plan_chunks(&mut plan, &records, all_keys, verified).await?; append_commits(repo, Some(tip.git), Some(tip.blocks), &chunks, now, emit)?; Ok(chunks) } @@ -492,10 +663,10 @@ impl ChunkDraft { async fn plan_chunks( plan: &mut Plan<'_, S>, - records: &BTreeMap>, + records: &BTreeMap, + all_keys: BTreeSet, ops: Vec, ) -> Result, ChainError> { - let all_keys: BTreeSet = records.keys().cloned().collect(); let mut out = Vec::new(); futures::stream::iter(ops) .chunks(MAX_OPS_PER_COMMIT) @@ -517,8 +688,8 @@ async fn plan_chunks( async fn emit_chunk( plan: &mut Plan<'_, S>, - records: &BTreeMap>, - all_keys: &BTreeSet, + records: &BTreeMap, + all_keys: &BTreeSet, slice: Vec, out: &mut Vec, ) -> Result<(), ChainError> { @@ -537,16 +708,16 @@ async fn emit_chunk( async fn draft_chunk( plan: &Plan<'_, S>, - records: &BTreeMap>, - all_keys: &BTreeSet, + records: &BTreeMap, + all_keys: &BTreeSet, ops: &[VerifiedWriteOp], ) -> Result, ChainError> { let applied = mst::apply(&plan.tree, ops).await?; let mut blocks = crate::proof::covering_blocks(&applied.tree, &applied.diff, all_keys).await?; ops.iter().for_each(|op| match op { VerifiedWriteOp::Create { key, cid } | VerifiedWriteOp::Update { key, cid, .. } => { - if let Some(bytes) = records.get(key.as_str()) { - blocks.insert(*cid, Bytes::from(bytes.clone())); + if let Some(body) = records.get(key.as_str()) { + blocks.insert(*cid, Bytes::copy_from_slice(body.as_bytes())); } } VerifiedWriteOp::Delete { .. } => {} @@ -623,7 +794,16 @@ fn write_chain_commit( filename: COMMIT_BLOB.into(), oid: blob, }); - entries.sort_by(|a, b| a.filename.cmp(&b.filename)); + + // The essence of trying to fit the cylinderical atproto into the square git hole. + // Every commit on atproto has a given tree w/ a blocks subtree in it, a commit blob, + // and a frame blob in which there's a firehose frame we must store. + // To make it fit, especially given that gix just orders exactly however we give it, + // we just have to be really careful to not mess up the ordering or else atproto-wise + // we'll get a wrong hash out the top. Hence the appended '/' to trick git a bit into thinking + // `blocks/` is above '.' but below '0'. + // There's a test at a_tree_entry_sorts_as_git_sorts_it_with_a_subtree_slash. + entries.sort_by(git_entry_order); let tree = git .write_object(gix::objs::Tree { entries }) .map_err(|error| ChainError::Malformed(error.to_string()))? @@ -651,9 +831,30 @@ fn write_chain_commit( Ok(id) } +fn git_entry_order(left: &gix::objs::tree::Entry, right: &gix::objs::tree::Entry) -> Ordering { + let (a, b) = (left.filename.as_slice(), right.filename.as_slice()); + let shared = a.len().min(b.len()); + a[..shared].cmp(&b[..shared]).then_with(|| { + past_shared(a, shared, left.mode.is_tree()).cmp(&past_shared( + b, + shared, + right.mode.is_tree(), + )) + }) +} + +fn past_shared(name: &[u8], shared: usize, tree: bool) -> Option { + match (name.get(shared), tree) { + (Some(byte), _) => Some(*byte), + (None, true) => Some(b'/'), + (None, false) => None, + } +} + #[cfg(test)] mod tests { use super::*; + use jacquard_repo::BlockStore; use knot_runtime::K256Signer; use knot_types::AccountDid; @@ -661,6 +862,10 @@ mod tests { const DID: &str = "did:web:knot.example"; + fn repo_did() -> RepoDid { + RepoDid::new(DID).unwrap() + } + fn repo() -> (tempfile::TempDir, Repo) { let dir = tempfile::tempdir().unwrap(); let layout = knot_git::Layout::new(dir.path()); @@ -668,7 +873,7 @@ mod tests { (dir, layout.create(&did).unwrap()) } - fn records(count: usize) -> BTreeMap> { + fn records(count: usize) -> BTreeMap { (0..count) .map(|n| { let record = crate::InviteRecord::new( @@ -678,8 +883,12 @@ mod tests { ) .unwrap(); ( - format!("sh.tangled.knot.memberInvite/did:plc:subject{n}"), - record.bytes(), + MstKey::new(format!( + "{}/did:plc:subject{n}", + crate::MEMBER_INVITE_COLLECTION + )) + .unwrap(), + record.body().unwrap(), ) }) .collect() @@ -687,12 +896,12 @@ mod tests { async fn enable( repo: &Repo, - records: &BTreeMap>, + records: &BTreeMap, signer: &K256Signer, ) -> Result, ChainError> { materialize( repo, - DID, + &repo_did(), records, signer, &mut Ticker::new(), @@ -726,7 +935,7 @@ mod tests { } #[tokio::test] - async fn five_hundred_roster_entries_chunk_into_limited_commits() { + async fn five_hundred_roll_entries_chunk_into_limited_commits() { let (_dir, repo) = repo(); let chunks = enable(&repo, &records(500), &key()).await.unwrap(); assert_eq!( @@ -763,19 +972,9 @@ mod tests { .await .unwrap() .is_empty(), - "an empty roster materializes no commit at all" + "empty roll doesn't materialize any commit at all" ); assert!(tip(&repo).unwrap().is_none()); - let mut fat = records(1); - fat.insert( - "sh.tangled.knot.memberInvite/did:plc:wide".to_owned(), - vec![0u8; MAX_RECORD_BLOCK_BYTES + 1], - ); - assert!(matches!( - enable(&repo, &fat, &key()).await, - Err(ChainError::RecordTooLarge(_)) - )); - assert!(tip(&repo).unwrap().is_none()); let first = enable(&repo, &records(1), &key()).await.unwrap(); assert!(matches!( enable(&repo, &records(1), &key()).await, @@ -822,11 +1021,14 @@ mod tests { #[tokio::test] async fn over_wide_chunks_split_in_half_until_they_fit() { let (_dir, repo) = repo(); - let records: BTreeMap> = (0..3) + let records: BTreeMap = (0..3) .map(|n| { ( - format!("sh.tangled.knot.memberInvite/did:plc:wide{n}"), - vec![n as u8 + 1; 900_000], + MstKey::new(format!("sh.tangled.knot.memberInvite/did:plc:wide{n}")).unwrap(), + RecordBody::new( + serde_ipld_dagcbor::to_vec(&format!("{n}{}", "x".repeat(900_000))).unwrap(), + ) + .unwrap(), ) }) .collect(); @@ -849,16 +1051,16 @@ mod tests { .collect(); records .values() - .for_each(|bytes| assert!(served.contains(&crate::cid_of(bytes)))); + .for_each(|body| assert!(served.contains(&crate::cid_of(body.as_bytes())))); } - async fn leaves_of(repo: &Repo) -> BTreeMap { + async fn leaves_of(repo: &Repo) -> BTreeMap { load_tree(repo, &tip(repo).unwrap().unwrap()) .leaves() .await .unwrap() .into_iter() - .map(|(key, cid)| (key.as_str().to_owned(), cid)) + .map(|(key, cid)| (MstKey::new(key.as_str().to_owned()).unwrap(), cid)) .collect() } @@ -872,19 +1074,20 @@ mod tests { records.remove("sh.tangled.knot.memberInvite/did:plc:subject0"); records.insert( - "sh.tangled.knot.memberInvite/did:plc:subject3".to_owned(), + MstKey::new("sh.tangled.knot.memberInvite/did:plc:subject3".to_owned()).unwrap(), crate::InviteRecord::new( crate::MEMBER_INVITE_COLLECTION, UnixSeconds::new(99), AccountDid::new("did:plc:3fwecdnvtcscjnrx2p4n7alz").unwrap(), ) .unwrap() - .bytes(), + .body() + .unwrap(), ); let chunks = sync( &repo, - DID, - crate::MEMBER_INVITE_COLLECTION, + &repo_did(), + &crate::member_invite_collection(), &records, &signer, &mut Ticker::new(), @@ -919,8 +1122,8 @@ mod tests { ); let leaves = leaves_of(&repo).await; assert_eq!(leaves.len(), 3); - records.iter().for_each(|(key, bytes)| { - assert_eq!(leaves.get(key), Some(&crate::cid_of(bytes))); + records.iter().for_each(|(key, body)| { + assert_eq!(leaves.get(key), Some(&crate::cid_of(body.as_bytes()))); }); } @@ -933,8 +1136,8 @@ mod tests { let records = records(450); let chunks = sync( &repo, - DID, - crate::MEMBER_INVITE_COLLECTION, + &repo_did(), + &crate::member_invite_collection(), &records, &signer, &mut Ticker::new(), @@ -992,8 +1195,8 @@ mod tests { assert!(matches!( sync( &repo, - DID, - crate::MEMBER_INVITE_COLLECTION, + &repo_did(), + &crate::member_invite_collection(), &records(1), &key(), &mut Ticker::new(), @@ -1009,8 +1212,8 @@ mod tests { assert!( sync( &repo, - DID, - crate::MEMBER_INVITE_COLLECTION, + &repo_did(), + &crate::member_invite_collection(), &records, &signer, &mut Ticker::new(), @@ -1030,15 +1233,15 @@ mod tests { let mut mixed = records(1); let foreign = "other.collection/did:plc:subject9"; mixed.insert( - foreign.to_owned(), + MstKey::new(foreign.to_owned()).unwrap(), records(2).values().next().unwrap().clone(), ); enable(&repo, &mixed, &signer).await.unwrap(); let chunks = sync( &repo, - DID, - crate::MEMBER_INVITE_COLLECTION, + &repo_did(), + &crate::member_invite_collection(), &records(2), &signer, &mut Ticker::new(), @@ -1061,14 +1264,14 @@ mod tests { let mut outside = records(1); outside.insert( - foreign.to_owned(), + MstKey::new(foreign.to_owned()).unwrap(), records(2).values().next().unwrap().clone(), ); assert!(matches!( sync( &repo, - DID, - crate::MEMBER_INVITE_COLLECTION, + &repo_did(), + &crate::member_invite_collection(), &outside, &signer, &mut Ticker::new(), @@ -1086,20 +1289,21 @@ mod tests { let mut records = records(2); let mut chunks = enable(&repo, &records, &signer).await.unwrap(); records.insert( - "sh.tangled.knot.memberInvite/did:plc:late".to_owned(), + MstKey::new("sh.tangled.knot.memberInvite/did:plc:late".to_owned()).unwrap(), crate::InviteRecord::new( crate::MEMBER_INVITE_COLLECTION, UnixSeconds::new(7), AccountDid::new("did:plc:3fwecdnvtcscjnrx2p4n7alz").unwrap(), ) .unwrap() - .bytes(), + .body() + .unwrap(), ); chunks.extend( sync( &repo, - DID, - crate::MEMBER_INVITE_COLLECTION, + &repo_did(), + &crate::member_invite_collection(), &records, &signer, &mut Ticker::new(), @@ -1135,9 +1339,15 @@ mod tests { #[tokio::test] async fn a_fresh_account_initializes_with_an_empty_commit_and_one_more_tip() { let (_dir, repo) = repo(); - let chunk = initialize(&repo, DID, &key(), &mut Ticker::new(), UnixSeconds::new(1)) - .await - .unwrap(); + let chunk = initialize( + &repo, + &repo_did(), + &key(), + &mut Ticker::new(), + UnixSeconds::new(1), + ) + .await + .unwrap(); assert!(chunk.ops.is_empty()); assert_eq!(chunk.prev_rev, None); assert_eq!(chunk.prev_data, None); @@ -1146,17 +1356,24 @@ mod tests { assert_eq!(tip.data, chunk.data); assert!( leaves_of(&repo).await.is_empty(), - "the genesis commit holds no records" + "Genesis commit has no records" ); assert!(matches!( - initialize(&repo, DID, &key(), &mut Ticker::new(), UnixSeconds::new(2)).await, + initialize( + &repo, + &repo_did(), + &key(), + &mut Ticker::new(), + UnixSeconds::new(2) + ) + .await, Err(ChainError::AlreadyMaterialized) )); assert_eq!( sync( &repo, - DID, - crate::MEMBER_INVITE_COLLECTION, + &repo_did(), + &crate::member_invite_collection(), &records(1), &key(), &mut Ticker::new(), @@ -1180,10 +1397,12 @@ mod tests { Tid::from_time(1_700_000_000_000_000 + self.frames.len() as u64, 0) } - fn frame(&mut self, chunk: &CommitChunk) -> Option { - let frame = Bytes::from(chunk.rev.to_string()); - self.frames.push(Some(frame.clone())); - Some(frame) + fn frame(&self, chunk: &CommitChunk) -> Option { + Some(Bytes::from(chunk.rev.to_string())) + } + + fn committed(&mut self, frame: Option) { + self.frames.push(frame); } } @@ -1193,7 +1412,7 @@ mod tests { let mut emit = Collecting::default(); let enabled = materialize( &repo, - DID, + &repo_did(), &records(2), &key(), &mut emit, @@ -1203,8 +1422,8 @@ mod tests { .unwrap(); let appended = sync( &repo, - DID, - crate::MEMBER_INVITE_COLLECTION, + &repo_did(), + &crate::member_invite_collection(), &records(4), &key(), &mut emit, @@ -1243,4 +1462,363 @@ mod tests { "a walk that reaches the bottom of the chain names no cap" ); } + + fn issue_collection() -> RecordCollection { + RecordCollection::new("sh.tangled.repo.issue").unwrap() + } + + fn issue_address(n: u64) -> RecordAddress { + RecordAddress::new( + knot_types::RecordCollection::new("sh.tangled.repo.issue").unwrap(), + knot_types::RecordRkey::minted(Tid::from_time(1_700_000_000_000_000 + n, 0)), + ) + } + + fn issue_body(revision: u64) -> RecordBody { + RecordBody::new(issue_record(revision)).expect("Issue record is dag-cbor") + } + + fn issue_record(revision: u64) -> Vec { + serde_ipld_dagcbor::to_vec(&BTreeMap::from([ + ("$type", "sh.tangled.repo.issue".to_owned()), + ("title", format!("kelp {revision}")), + ])) + .unwrap() + } + + fn issues(count: u64) -> BTreeMap { + (0..count) + .map(|n| (issue_address(n).mst_key(), issue_body(n))) + .collect() + } + + fn cid_at(records: &BTreeMap, address: &RecordAddress) -> RecordCid { + records[address.mst_key().as_str()].cid() + } + + #[tokio::test] + async fn apply_appends_one_op_and_leaves_every_other_record_where_it_was() { + let (_dir, repo) = repo(); + let base = issues(4); + enable(&repo, &base, &key()).await.unwrap(); + let before = tip(&repo).unwrap().unwrap(); + + let chunks = apply( + &repo, + &repo_did(), + &[( + issue_address(9), + RecordOp::Create { + record: issue_body(9), + }, + )], + &key(), + &mut Ticker::new(), + UnixSeconds::new(2), + ) + .await + .unwrap(); + + assert_eq!(chunks.len(), 1, "one op is one commit"); + assert_eq!( + chunks[0].ops, + vec![Op::Create { + key: issue_address(9).mst_key().as_str().to_owned(), + cid: crate::cid_of(&issue_record(9)), + }], + "Frame names exactly the op the caller asked for" + ); + assert_eq!(chunks[0].prev_data, Some(before.data)); + assert_eq!( + crate::proof::cold_invert(&chunks[0].blocks, chunks[0].data, &verified(&chunks[0].ops)) + .await + .unwrap(), + before.data, + "the consumer inverts the chunk, back to the previous root, from the frame alone" + ); + + let leaves = leaves_of(&repo).await; + assert_eq!( + leaves.len(), + 5, + "four records were live and one arrived, with no resync in between" + ); + base.keys().for_each(|key| { + assert!(leaves.contains_key(key), "{key} went missing under apply"); + }); + } + + #[tokio::test] + async fn apply_names_the_current_cid_when_the_record_isnt_where_the_writer_left_it() { + let (_dir, repo) = repo(); + let base = issues(2); + enable(&repo, &base, &key()).await.unwrap(); + let chain_tip = tip(&repo).unwrap().unwrap().git; + let live = cid_at(&base, &issue_address(0)); + let stale = + RecordCid::new(crate::cid_of(&issue_record(77))).expect("Fixture record is dag-cbor"); + + let moved = apply( + &repo, + &repo_did(), + &[( + issue_address(0), + RecordOp::Update { + record: issue_body(1), + from: stale, + }, + )], + &key(), + &mut Ticker::new(), + UnixSeconds::new(2), + ) + .await; + match moved { + Err(ChainError::RecordMoved { + uri, + current, + named, + }) => { + assert_eq!( + current, live, + "Moved record answers with the current version" + ); + assert_eq!(named, stale); + assert_eq!( + uri, + issue_address(0).at_uri(&repo_did()), + "and with at-uri to rebase onto" + ); + } + other => panic!("Edit from version that moved on must be refused, got {other:?}"), + } + + let existing = apply( + &repo, + &repo_did(), + &[( + issue_address(0), + RecordOp::Create { + record: issue_body(3), + }, + )], + &key(), + &mut Ticker::new(), + UnixSeconds::new(2), + ) + .await; + assert!( + matches!(existing, Err(ChainError::RecordExists { .. })), + "Create never writes over a live record, got {existing:?}" + ); + + let absent = apply( + &repo, + &repo_did(), + &[(issue_address(5), RecordOp::Delete { from: live })], + &key(), + &mut Ticker::new(), + UnixSeconds::new(2), + ) + .await; + assert!( + matches!(absent, Err(ChainError::RecordAbsent { .. })), + "there's no record at that address to delete, got {absent:?}" + ); + + let twice = apply( + &repo, + &repo_did(), + &[ + ( + issue_address(0), + RecordOp::Update { + record: issue_body(4), + from: live, + }, + ), + (issue_address(0), RecordOp::Delete { from: live }), + ], + &key(), + &mut Ticker::new(), + UnixSeconds::new(2), + ) + .await; + assert!( + matches!(twice, Err(ChainError::RecordContested { .. })), + "two ops on one address in one write have no defined order, got {twice:?}" + ); + + assert_eq!( + tip(&repo).unwrap().unwrap().git, + chain_tip, + "every write above was turned away and left chain alone" + ); + } + + fn op_map(chunks: &[CommitChunk]) -> BTreeMap, Option)> { + chunks + .iter() + .flat_map(|chunk| chunk.ops.iter()) + .map(|op| (op.key().to_owned(), (op.cid(), op.prev()))) + .collect() + } + + async fn enabled_root(repo: &Repo, base: &BTreeMap) -> Cid { + enable(repo, base, &key()).await.unwrap(); + tip(repo).unwrap().unwrap().data + } + + fn runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("Current-thread runtime builds") + } + + proptest::proptest! { + #![proptest_config(proptest::prelude::ProptestConfig { cases: 24, ..proptest::prelude::ProptestConfig::default() })] + + #[test] + fn a_run_of_apply_reaches_the_root_one_sync_reaches( + actions in proptest::collection::vec(0u8..3, 9) + ) { + let base = issues(6); + let (_sync_dir, sync_repo) = repo(); + let (_apply_dir, apply_repo) = repo(); + let runtime = runtime(); + let (synced, applied) = runtime.block_on(async { + let start = enabled_root(&sync_repo, &base).await; + proptest::prop_assert_eq!( + start, + enabled_root(&apply_repo, &base).await, + "two chains over one record set start from one root" + ); + + let desired: BTreeMap = (0..9u64) + .zip(&actions) + .filter_map(|(n, action)| { + let key = issue_address(n).mst_key(); + match (base.contains_key(&key), action) { + (true, 0) => Some((key, issue_body(n))), + (true, 1) => Some((key, issue_body(100))), + (true, _) | (false, 0) => None, + (false, _) => Some((key, issue_body(n))), + } + }) + .collect(); + let ops: Vec<(RecordAddress, RecordOp)> = (0..9u64) + .zip(&actions) + .filter_map(|(n, action)| { + let address = issue_address(n); + let existing = base + .contains_key(address.mst_key().as_str()) + .then(|| cid_at(&base, &address)); + match (existing, action) { + (_, 0) => None, + (Some(from), 1) => Some(( + address, + RecordOp::Update { + record: issue_body(100), + from, + }, + )), + (Some(from), _) => Some((address, RecordOp::Delete { from })), + (None, _) => Some(( + address, + RecordOp::Create { + record: issue_body(n), + }, + )), + } + }) + .collect(); + + let synced = sync( + &sync_repo, + &repo_did(), + &issue_collection(), + &desired, + &key(), + &mut Ticker::new(), + UnixSeconds::new(2), + ) + .await + .unwrap(); + let applied = apply( + &apply_repo, + &repo_did(), + &ops, + &key(), + &mut Ticker::new(), + UnixSeconds::new(2), + ) + .await + .unwrap(); + Ok((synced, applied)) + })?; + + proptest::prop_assert_eq!( + op_map(&applied), + op_map(&synced), + "Run of apply names the same ops the whole-collection diff derives" + ); + proptest::prop_assert_eq!( + tip(&apply_repo).unwrap().unwrap().data, + tip(&sync_repo).unwrap().unwrap().data, + "MST, a pure function of the record set, gives both paths one root" + ); + runtime.block_on(async { + proptest::prop_assert!( + futures::stream::iter(&applied) + .then(|chunk| async move { + crate::proof::cold_invert( + &chunk.blocks, + chunk.data, + &verified(&chunk.ops), + ) + .await + .ok() + == chunk.prev_data + }) + .all(|inverted| async move { inverted }) + .await, + "every chunk apply emits inverts cold out of its own blocks" + ); + Ok(()) + })?; + } + } + + #[test] + fn a_tree_entry_sorts_as_git_sorts_it_with_a_subtree_slash() { + let entry = |name: &str, kind: gix::objs::tree::EntryKind| gix::objs::tree::Entry { + mode: kind.into(), + filename: name.into(), + oid: gix::ObjectId::empty_blob(gix::hash::Kind::Sha1), + }; + let blob = |name: &str| entry(name, gix::objs::tree::EntryKind::Blob); + let tree = |name: &str| entry(name, gix::objs::tree::EntryKind::Tree); + let mut entries = [ + blob("blocks."), + tree("blocks"), + blob("commit"), + blob("blocks"), + blob("frame"), + ]; + entries.sort_by(git_entry_order); + let sorted: Vec = entries + .iter() + .map(|entry| { + let slash = if entry.mode.is_tree() { "/" } else { "" }; + format!("{}{slash}", entry.filename) + }) + .collect(); + assert_eq!( + sorted, + vec!["blocks", "blocks.", "blocks/", "commit", "frame"], + "git compares subtree as though its name ended in slash. It rejects \ + trees whose entries arrive in any other order" + ); + } } diff --git a/knot2/crates/knot-record/src/git_ref.rs b/knot2/crates/knot-record/src/git_ref.rs index 2a8561560..44181011b 100644 --- a/knot2/crates/knot-record/src/git_ref.rs +++ b/knot2/crates/knot-record/src/git_ref.rs @@ -4,9 +4,10 @@ use std::sync::Arc; use crate::blocks::GitBlocks; use crate::chain::{self, ChainError}; use futures::StreamExt; +use futures::TryStreamExt as _; use jacquard_repo::storage::BlockStore; use knot_git::{RefUpdate, Repo}; -use knot_types::{AccountDid, Oid, PushOption}; +use knot_types::{AccountDid, MstKey, Oid, PushOption, RecordBody}; use serde::{Deserialize, Serialize}; pub const GIT_REF_COLLECTION: &str = "sh.tangled.git.ref"; @@ -90,6 +91,10 @@ impl RefRecord { pub fn bytes(&self) -> Vec { serde_ipld_dagcbor::to_vec(self).expect("text fields always encode as dag-cbor") } + + pub fn body(&self) -> Result { + RecordBody::new(self.bytes()) + } } pub fn sha_of(bytes: &[u8]) -> Option { @@ -107,7 +112,7 @@ pub fn desired_records( applied: &[RefUpdate], editor: Option<&AccountDid>, push_options: &[PushOption], -) -> Result>, ChainError> { +) -> Result, ChainError> { let prefix = format!("{GIT_REF_COLLECTION}/"); let stored = stored_records(repo, &prefix)?; let served = repo @@ -125,7 +130,7 @@ pub fn desired_records( }) .map(|update| update.name().as_str()) .collect::>(); - Ok(served + let served = served .into_iter() .filter_map(|record| match escape(record.name.as_str()) { Some(rkey) => Some((format!("{prefix}{rkey}"), record)), @@ -138,34 +143,50 @@ pub fn desired_records( } }) .map(|(key, record)| { - let kept = stored - .get(&key) - .filter(|bytes| sha_of(bytes).is_some_and(|sha| sha == record.target.to_hex())); - let bytes = match touched.contains(record.name.as_str()) { - true => RefRecord::new(record.target, editor, push_options).bytes(), - false => kept - .cloned() - .unwrap_or_else(|| RefRecord::new(record.target, None, &[]).bytes()), + let kept = stored.get(key.as_str()).filter(|body| { + sha_of(body.as_bytes()).is_some_and(|sha| sha == record.target.to_hex()) + }); + let body = match touched.contains(record.name.as_str()) { + true => RefRecord::new(record.target, editor, push_options).body(), + false => match kept.cloned() { + Some(body) => Ok(body), + None => RefRecord::new(record.target, None, &[]).body(), + }, }; - (key, bytes) + body.map(|body| { + ( + MstKey::new(key).expect("Escaped ref key fits mst"), + body, + ) + }) + .map_err(|error| ChainError::Malformed(error.to_string())) }) - .collect()) + .collect::, _>>()?; + Ok(served) } -fn stored_records(repo: &Repo, prefix: &str) -> Result>, ChainError> { +fn stored_records(repo: &Repo, prefix: &str) -> Result, ChainError> { let tip = match chain::tip(repo)? { Some(tip) => tip, None => return Ok(BTreeMap::new()), }; let blocks = Arc::new(GitBlocks::at(repo, tip.blocks)); futures::executor::block_on(async { - let records = futures::stream::iter(chain::load_tree(repo, &tip).leaves().await?) + futures::stream::iter(chain::load_tree(repo, &tip).leaves().await?) .filter(|(key, _)| std::future::ready(key.as_str().starts_with(prefix))) .then(|(key, cid)| { let blocks = Arc::clone(&blocks); async move { match blocks.get(&cid).await.ok().flatten() { - Some(bytes) => Some((key.as_str().to_owned(), bytes.to_vec())), + Some(bytes) => Some( + RecordBody::new(bytes.to_vec()).map(|body| { + ( + MstKey::new(key.as_str().to_owned()) + .expect("Stored ref key fits mst"), + body, + ) + }), + ), None => { tracing::warn!( key = key.as_str(), @@ -177,15 +198,18 @@ fn stored_records(repo: &Repo, prefix: &str) -> Result> } }) .filter_map(std::future::ready) - .collect::>() - .await; - Ok(records) + .try_collect::>() + .await + .map_err(|error: knot_types::RecordBodyError| { + ChainError::Malformed(error.to_string()) + }) }) } #[cfg(test)] mod tests { use super::*; + use knot_types::{RecordCollection, RepoDid}; const PUSHED: &str = "a4637368617828616261626162616261626162616261626162616261626162616261626162616261626162616261626524747970657273682e74616e676c65642e6769742e7265666d782d746e676c2d656469746f726e6469643a706c633a6c696d70657473782d746e676c2d707573682d6f7074696f6e738167736b69702d6369"; const BARE: &str = "a2637368617840636463646364636463646364636463646364636463646364636463646364636463646364636463646364636463646364636463646364636463646364636463646524747970657273682e74616e676c65642e6769742e726566"; @@ -219,7 +243,7 @@ mod tests { assert_eq!( bare, frozen(BARE), - "a catch-up record names no editor and holds no options, and a sha256 oid fits" + "Catch-up record leaves editor out and options empty, and sha256 oid fits" ); assert_eq!( sha_of(&bare), @@ -298,7 +322,7 @@ mod tests { let signer = knot_runtime::K256Signer::from_slice(&[7u8; 32]).unwrap(); futures::executor::block_on(crate::chain::initialize( &repo, - "did:web:knot.example", + &RepoDid::new("did:web:knot.example").unwrap(), &signer, &mut jacquard_common::types::tid::Ticker::new(), knot_types::UnixSeconds::new(1), @@ -318,9 +342,12 @@ mod tests { ); futures::executor::block_on(chain::sync( &repo, - "did:web:knot.example", - GIT_REF_COLLECTION, - &BTreeMap::from([(key.clone(), stored.clone())]), + &RepoDid::new("did:web:knot.example").unwrap(), + &RecordCollection::new(GIT_REF_COLLECTION).unwrap(), + &BTreeMap::from([( + MstKey::new(key.clone()).unwrap(), + RecordBody::new(stored.clone()).unwrap(), + )]), &signer, &mut jacquard_common::types::tid::Ticker::new(), knot_types::UnixSeconds::new(2), @@ -333,7 +360,8 @@ mod tests { }]; let desired = desired_records(&repo, &applied, None, &[]).unwrap(); assert_eq!( - desired[&key], stored, + desired[key.as_str()].as_bytes(), + stored, "the branch reappeared between the delete and the projection, so the push that owns this sha stays named" ); let applied = [RefUpdate::Update { @@ -343,7 +371,8 @@ mod tests { }]; let desired = desired_records(&repo, &applied, None, &[]).unwrap(); assert_eq!( - desired[&key], stored, + desired[key.as_str()].as_bytes(), + stored, "only a sha that landed can replace the stored provenance" ); } diff --git a/knot2/crates/knot-record/src/lib.rs b/knot2/crates/knot-record/src/lib.rs index d36958d76..feda3ff96 100644 --- a/knot2/crates/knot-record/src/lib.rs +++ b/knot2/crates/knot-record/src/lib.rs @@ -6,7 +6,7 @@ pub mod proof; use bytes::Bytes; use cid::Cid; -use knot_types::{AccountDid, UnixSeconds}; +use knot_types::{AccountDid, RecordBody, RecordBodyError, UnixSeconds}; use serde::Serialize; #[derive(Debug, thiserror::Error)] pub enum RecordError { @@ -24,6 +24,16 @@ pub const MEMBER_INVITE_COLLECTION: &str = "sh.tangled.knot.memberInvite"; pub const COLLABORATOR_INVITE_COLLECTION: &str = "sh.tangled.repo.collaboratorInvite"; +pub fn member_invite_collection() -> knot_types::RecordCollection { + knot_types::RecordCollection::new(MEMBER_INVITE_COLLECTION) + .expect("Member invite collection is valid nsid") +} + +pub fn collaborator_invite_collection() -> knot_types::RecordCollection { + knot_types::RecordCollection::new(COLLABORATOR_INVITE_COLLECTION) + .expect("Collaborator invite collection is valid nsid") +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct InviteRecord { #[serde(rename = "$type")] @@ -52,11 +62,14 @@ impl InviteRecord { pub fn bytes(&self) -> Vec { serde_ipld_dagcbor::to_vec(self).expect("three text fields always encode as dag-cbor") } + + pub fn body(&self) -> Result { + RecordBody::new(self.bytes()) + } } -pub fn cid_of(bytes: &[u8]) -> Cid { - jacquard_repo::mst::util::compute_cid(bytes) - .expect("sha2-256 over dag-cbor always builds a cid") +pub fn cid_of(block: &[u8]) -> Cid { + knot_types::dag_cbor_cid(block) } pub fn car_bytes(root: Cid, blocks: &std::collections::BTreeMap) -> Vec { diff --git a/knot2/crates/knot-record/src/proof.rs b/knot2/crates/knot-record/src/proof.rs index f37ba38ed..943575241 100644 --- a/knot2/crates/knot-record/src/proof.rs +++ b/knot2/crates/knot-record/src/proof.rs @@ -7,12 +7,14 @@ use jacquard_repo::{Mst, MstDiff}; use std::collections::{BTreeMap, BTreeSet}; use std::ops::Bound::{Excluded, Unbounded}; +use knot_types::MstKey; + use crate::{RecordError, Result}; pub async fn covering_blocks( after: &Mst, diff: &MstDiff, - all_keys: &BTreeSet, + all_keys: &BTreeSet, ) -> Result> { let mut blocks = diff.new_mst_blocks.clone(); futures::stream::iter( @@ -25,7 +27,7 @@ pub async fn covering_blocks BTreeSet { .collect() } -fn neighbors(all_keys: &BTreeSet, key: &str) -> Vec { +fn neighbors(all_keys: &BTreeSet, key: &str) -> Vec { let before = all_keys .range::((Unbounded, Excluded(key))) .next_back() @@ -76,7 +78,7 @@ fn neighbors(all_keys: &BTreeSet, key: &str) -> Vec { .range::((Excluded(key), Unbounded)) .next() .cloned(); - [before, after, Some(key.to_owned())] + [before, after, MstKey::new(key.to_owned()).ok()] .into_iter() .flatten() .collect() diff --git a/knot2/crates/knot-record/tests/proof_fixtures.rs b/knot2/crates/knot-record/tests/proof_fixtures.rs index f9e408b4a..49db6def3 100644 --- a/knot2/crates/knot-record/tests/proof_fixtures.rs +++ b/knot2/crates/knot-record/tests/proof_fixtures.rs @@ -1,3 +1,4 @@ +use knot_types::MstKey; use std::collections::{BTreeMap, BTreeSet}; use std::path::PathBuf; @@ -90,10 +91,10 @@ async fn the_covering_set_is_a_superset_that_inverts_cold_with_a_bounded_size() vector["rootAfterCommit"].as_str().unwrap(), "{comment}: applying the vector's ops must reach rootAfterCommit" ); - let all_keys: BTreeSet = keys + let all_keys: BTreeSet = keys .iter() .chain(strings(vector, "adds").iter()) - .cloned() + .map(|key| MstKey::new(key.to_owned()).unwrap()) .collect(); let covering = proof::covering_blocks(&after.tree, &after.diff, &all_keys) .await