diff --git a/knot2/crates/knot-cob/src/backend.rs b/knot2/crates/knot-cob/src/backend.rs index b4e7eb918..b11e16175 100644 --- a/knot2/crates/knot-cob/src/backend.rs +++ b/knot2/crates/knot-cob/src/backend.rs @@ -15,7 +15,7 @@ use crate::graph::ChangeGraph; use crate::object::HistoryModel; const COBS_PREFIX: &str = "refs/cobs/"; -const CHECKPOINTS_PREFIX: &str = "refs/cob-checkpoints/"; +pub const CHECKPOINTS_PREFIX: &str = "refs/cob-checkpoints/"; const TYPE_HEADER: &str = "cob-type"; const SIG_HEADER: &str = "cob-sig"; const AUTHOR_HEADER: &str = "cob-author"; diff --git a/knot2/crates/knot-cob/src/lib.rs b/knot2/crates/knot-cob/src/lib.rs index 75454b3da..da23f7197 100644 --- a/knot2/crates/knot-cob/src/lib.rs +++ b/knot2/crates/knot-cob/src/lib.rs @@ -15,7 +15,7 @@ pub use object::{ SnapshotStride, StateSize, WriteContext, }; -pub use backend::parse_cob_ref; +pub use backend::{CHECKPOINTS_PREFIX, parse_cob_ref}; use std::collections::BTreeSet; @@ -2746,7 +2746,7 @@ mod tests { } #[test] - fn a_write_the_actor_didnt_author_never_reaches_git() { + fn impersonated_bylines_never_reach_git() { let nel = who("nel"); let repo_did = squid(); let cx = context(&nel, &repo_did, RepoRole::Stranger); diff --git a/knot2/crates/knot-record/src/blocks.rs b/knot2/crates/knot-record/src/blocks.rs index cd377d128..a805c998c 100644 --- a/knot2/crates/knot-record/src/blocks.rs +++ b/knot2/crates/knot-record/src/blocks.rs @@ -169,6 +169,10 @@ impl GitBlocks { } fn blob(&self, cid: &Cid) -> Result, RepoError> { + self.in_git(cid) + } + + fn in_git(&self, cid: &Cid) -> Result, RepoError> { let Some((first, second)) = shard(cid) else { return Ok(None); }; diff --git a/knot2/crates/knot-record/src/chain.rs b/knot2/crates/knot-record/src/chain.rs index 7648d12e7..267dfdb32 100644 --- a/knot2/crates/knot-record/src/chain.rs +++ b/knot2/crates/knot-record/src/chain.rs @@ -18,7 +18,7 @@ use knot_git::{Identity, NewCommit, RefUpdate, Repo}; use knot_runtime::Signer; use knot_types::{ AtUri, AuthorName, Email, MstKey, Oid, RecordAddress, RecordBody, RecordCid, RecordCollection, - RefName, RepoDid, UnixSeconds, + RefName, RepoDid, UnixMicros, UnixSeconds, }; pub const CHAIN_REF: &str = "refs/atproto/commit"; @@ -28,8 +28,6 @@ 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_FRAME_BYTES: usize = 5_000_000; -const OP_FRAME_OVERHEAD_BYTES: usize = 256; pub trait Emit { fn rev(&mut self, prev: Option<&Tid>) -> Tid; @@ -68,11 +66,13 @@ pub enum ChainError { }, #[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}")] + #[error( + "{uri} is at {current}, and the write was made from {claimed}; please rebase from {current}" + )] RecordMoved { uri: AtUri, current: RecordCid, - named: RecordCid, + claimed: RecordCid, }, #[error("{address} takes one op per chain write, and this one names op twice")] RecordContested { address: RecordAddress }, @@ -134,7 +134,7 @@ impl Op { } } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct CommitChunk { pub commit: Vec, pub rev: Tid, @@ -145,6 +145,37 @@ pub struct CommitChunk { pub blocks: BTreeMap, } +pub trait Records { + fn placed(&self) -> &BTreeMap; + + fn body(&self, key: &MstKey) -> Result; +} + +pub struct InTree { + placed: BTreeMap, + bodies: BTreeMap, +} + +pub fn in_tree(bodies: BTreeMap) -> impl Records { + let placed = bodies + .iter() + .map(|(key, body)| (key.clone(), body.cid())) + .collect(); + InTree { placed, bodies } +} + +impl Records for InTree { + fn placed(&self) -> &BTreeMap { + &self.placed + } + + fn body(&self, key: &MstKey) -> Result { + self.bodies.get(key).cloned().ok_or_else(|| { + ChainError::Malformed(format!("the record key {key} came without its bytes")) + }) + } +} + pub fn blocks_bytes(blocks: &BTreeMap) -> usize { blocks.values().map(Bytes::len).sum() } @@ -177,6 +208,24 @@ pub fn load_tree(repo: &Repo, tip: &ChainTip) -> Mst { Mst::load(Arc::new(GitBlocks::at(repo, tip.blocks)), tip.data, None) } +pub async fn leaves(repo: &Repo) -> Result, ChainError> { + match tip(repo)? { + Some(tip) => keyed_leaves(&load_tree(repo, &tip)).await, + None => Ok(BTreeMap::new()), + } +} + +async fn keyed_leaves( + tree: &Mst, +) -> Result, ChainError> { + tree.leaves() + .await? + .into_iter() + .map(|(key, cid)| MstKey::new(key.as_str().to_owned()).map(|key| (key, cid))) + .collect::, _>>() + .map_err(|error| ChainError::Malformed(error.to_string())) +} + pub struct ChainStep { pub next: Option, pub rev: Tid, @@ -219,18 +268,18 @@ fn step(repo: &Repo, oid: Oid, load_frame: bool) -> Result, + pub found: Vec<(UnixMicros, Oid)>, pub hit_limit: bool, } -pub fn framed_revs(repo: &Repo, start: Oid, after: u64, limit: usize) -> FramedRevs { +pub fn framed_revs(repo: &Repo, start: Oid, after: UnixMicros, limit: usize) -> FramedRevs { let mut found = Vec::new(); let mut hit_limit = false; let mut next = Some(start); while let Some(oid) = next { match step(repo, oid, false) { Ok(step) => { - if step.rev.timestamp() <= after { + if UnixMicros::new(step.rev.timestamp()) <= after { break; } if step.framed { @@ -238,7 +287,7 @@ pub fn framed_revs(repo: &Repo, start: Oid, after: u64, limit: usize) -> FramedR hit_limit = true; break; } - found.push((step.rev.timestamp(), oid)); + found.push((UnixMicros::new(step.rev.timestamp()), oid)); } next = step.next; } @@ -364,95 +413,61 @@ pub async fn materialize( if tip(repo)?.is_some() { return Err(ChainError::AlreadyMaterialized); } + let ops: Vec = records + .iter() + .map(|(key, body)| VerifiedWriteOp::Create { + key: key.as_str().into(), + cid: crate::cid_of(body.as_bytes()), + }) + .collect(); let mut plan = Plan { + repo, did: did_of(did)?, signer, emit, prev_rev: None, prev_data: None, tree: mst::empty(), + parent: None, + blocks_tree: None, + now, }; - let ops: Vec = records - .iter() - .map(|(key, body)| VerifiedWriteOp::Create { - key: key.as_str().into(), - cid: crate::cid_of(body.as_bytes()), - }) - .collect(); - let chunks = plan_chunks(&mut plan, records, records.keys().cloned().collect(), ops).await?; - append_commits(repo, None, None, &chunks, now, emit)?; - Ok(chunks) -} - -fn append_commits( - repo: &Repo, - parent: Option, - blocks_tree: Option, - chunks: &[CommitChunk], - now: UnixSeconds, - emit: &mut dyn Emit, -) -> Result<(), ChainError> { - 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), - }), - } + plan_chunks(&mut plan, &in_tree(records.clone()), ops).await } pub async fn sync( repo: &Repo, did: &RepoDid, - collection: &RecordCollection, - records: &BTreeMap, + collections: &[RecordCollection], + records: &dyn Records, signer: &dyn Signer, emit: &mut dyn Emit, now: UnixSeconds, ) -> Result, ChainError> { - let prefix = format!("{}/", collection.as_str()); - if let Some(key) = records - .keys() - .find(|key| !key.as_str().starts_with(&prefix)) - { + let prefixes: Vec = collections + .iter() + .map(|collection| format!("{}/", collection.as_str())) + .collect(); + let synced = |key: &MstKey| { + prefixes + .iter() + .any(|prefix| key.as_str().starts_with(prefix)) + }; + if let Some(key) = records.placed().keys().find(|key| !synced(key)) { return Err(ChainError::Malformed(format!( - "the record key {key} sits outside the {collection} collection this chain syncs" + "the record key {key} isn't in the collections this chain syncs" ))); } let Some(tip) = tip(repo)? else { return Err(ChainError::NotEnabled); }; let before = load_tree(repo, &tip); - let leaves: BTreeMap = before - .leaves() - .await? - .into_iter() - .map(|(key, cid)| MstKey::new(key.as_str().to_owned()).map(|key| (key, cid))) - .collect::, _>>() - .map_err(|error| ChainError::Malformed(error.to_string()))?; + let leaves = keyed_leaves(&before).await?; let mut ops: Vec = records + .placed() .iter() - .filter_map(|(key, body)| { - let cid = crate::cid_of(body.as_bytes()); + .filter_map(|(key, placed)| { + let cid = placed.cid(); match leaves.get(key) { None => Some(VerifiedWriteOp::Create { key: key.as_str().into(), @@ -470,7 +485,7 @@ pub async fn sync( ops.extend( leaves .iter() - .filter(|(key, _)| key.as_str().starts_with(&prefix) && !records.contains_key(*key)) + .filter(|(key, _)| synced(key) && !records.placed().contains_key(*key)) .map(|(key, prev)| VerifiedWriteOp::Delete { key: key.as_str().into(), prev: *prev, @@ -480,16 +495,18 @@ pub async fn sync( return Ok(Vec::new()); } let mut plan = Plan { + repo, did: did_of(did)?, signer, emit, prev_rev: Some(tip.rev.clone()), prev_data: Some(tip.data), tree: before, + parent: Some(tip.git), + blocks_tree: Some(tip.blocks), + now, }; - 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) + plan_chunks(&mut plan, records, ops).await } #[derive(Debug, Clone, PartialEq, Eq)] @@ -518,15 +535,15 @@ fn adopt(address: &RecordAddress, current: Cid) -> Result fn write_op( repo: &RepoDid, address: &RecordAddress, + key: &MstKey, op: &RecordOp, current: Option, -) -> Result { - let key = address.mst_key(); +) -> Result, ChainError> { match (op, current) { - (RecordOp::Create { record }, None) => Ok(VerifiedWriteOp::Create { + (RecordOp::Create { record }, None) => Ok(Some(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)?, @@ -542,18 +559,21 @@ fn write_op( Err(ChainError::RecordMoved { uri: address.at_uri(repo), current: adopt(address, current)?, - named: *from, + claimed: *from, }) } - (RecordOp::Update { record, .. }, Some(current)) => Ok(VerifiedWriteOp::Update { + (RecordOp::Update { record, .. }, Some(current)) if record.cid().cid() == current => { + Ok(None) + } + (RecordOp::Update { record, .. }, Some(current)) => Ok(Some(VerifiedWriteOp::Update { key: key.as_str().into(), cid: record.cid().cid(), prev: current, - }), - (RecordOp::Delete { .. }, Some(current)) => Ok(VerifiedWriteOp::Delete { + })), + (RecordOp::Delete { .. }, Some(current)) => Ok(Some(VerifiedWriteOp::Delete { key: key.as_str().into(), prev: current, - }), + })), } } @@ -571,53 +591,76 @@ pub async fn apply( limit: MAX_OPS_PER_APPLY, }); } - let all_keys = ops + let keyed: Vec<(MstKey, &RecordAddress, &RecordOp)> = ops + .iter() + .map(|(address, op)| (address.mst_key(), address, op)) + .collect(); + keyed .iter() - .try_fold(BTreeSet::::new(), |mut seen, (address, _)| { - seen.insert(address.mst_key()) + .try_fold(BTreeSet::new(), |mut seen, (key, address, _)| { + seen.insert(key) .then_some(seen) .ok_or_else(|| ChainError::RecordContested { - address: address.clone(), + 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 verified = futures::stream::iter(&keyed) + .then(|(key, address, op)| { let before = &before; async move { - let current = before.get(address.mst_key().as_str()).await?; - write_op(did, address, op, current) + let current = before.get(key.as_str()).await?; + write_op(did, address, key, op, current) } }) - .try_collect::>() - .await?; + .try_collect::>>() + .await? + .into_iter() + .flatten() + .collect::>(); + let (mut writes, mut deletes): (Vec, Vec) = verified + .into_iter() + .partition(|op| !matches!(op, VerifiedWriteOp::Delete { .. })); + writes.sort_by_key(|op| op_key(op).to_owned()); + deletes.sort_by_key(|op| op_key(op).to_owned()); + writes.append(&mut deletes); + let verified = writes; 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 records = in_tree( + keyed + .into_iter() + .filter_map(|(key, _, op)| op.record().map(|record| (key, record.clone()))) + .collect(), + ); let mut plan = Plan { + repo, did: did_of(did)?, signer, emit, prev_rev: Some(tip.rev.clone()), prev_data: Some(tip.data), tree: before, + parent: Some(tip.git), + blocks_tree: Some(tip.blocks), + now, }; - 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) + plan_chunks(&mut plan, &records, verified).await } -fn wire_ops(ops: &[VerifiedWriteOp]) -> Vec { +fn op_key(op: &VerifiedWriteOp) -> &str { + match op { + VerifiedWriteOp::Create { key, .. } + | VerifiedWriteOp::Update { key, .. } + | VerifiedWriteOp::Delete { key, .. } => key.as_str(), + } +} + +fn ops_of(ops: &[VerifiedWriteOp]) -> Vec { ops.iter() .map(|op| match op { VerifiedWriteOp::Create { key, cid } => Op::Create { @@ -638,98 +681,126 @@ fn wire_ops(ops: &[VerifiedWriteOp]) -> Vec { } struct Plan<'a, S> { + repo: &'a Repo, did: Did, signer: &'a dyn Signer, emit: &'a mut dyn Emit, prev_rev: Option, prev_data: Option, tree: Mst, + parent: Option, + blocks_tree: Option, + now: UnixSeconds, } struct ChunkDraft { tree: Mst, data: Cid, ops: Vec, - ops_wire: usize, blocks: BTreeMap, } impl ChunkDraft { fn over_limits(&self) -> bool { - let bytes = blocks_bytes(&self.blocks); - bytes > MAX_COMMIT_BLOCKS_BYTES || bytes + self.ops_wire > MAX_FRAME_BYTES + self.blocks.values().map(Bytes::len).sum::() > MAX_COMMIT_BLOCKS_BYTES } } async fn plan_chunks( plan: &mut Plan<'_, S>, - records: &BTreeMap, - all_keys: BTreeSet, + records: &dyn Records, ops: Vec, ) -> Result, ChainError> { + let intended = ops.len().div_ceil(MAX_OPS_PER_COMMIT); let mut out = Vec::new(); - futures::stream::iter(ops) + let planned = futures::stream::iter(ops) .chunks(MAX_OPS_PER_COMMIT) .map(Vec::from) .fold( Ok::<(&mut Plan<'_, S>, &mut Vec), ChainError>((plan, &mut out)), - |acc, slice| { - let (records, all_keys) = (records, &all_keys); - async move { - let (plan, out) = acc?; - emit_chunk(plan, records, all_keys, slice, out).await?; - Ok((plan, out)) - } + |acc, slice| async move { + let (plan, out) = acc?; + emit_chunk(plan, records, slice, out).await?; + Ok((plan, out)) }, ) - .await?; - Ok(out) + .await + .map(|_| ()); + match (planned, out.is_empty()) { + (Ok(()), _) => Ok(out), + (Err(error), true) => Err(error), + (Err(error), false) => Err(ChainError::PartlyApplied { + written: out.len(), + planned: intended, + reason: Box::new(error), + }), + } } async fn emit_chunk( plan: &mut Plan<'_, S>, - records: &BTreeMap, - all_keys: &BTreeSet, + records: &dyn Records, slice: Vec, out: &mut Vec, ) -> Result<(), ChainError> { - let draft = draft_chunk(plan, records, all_keys, &slice).await?; + let draft = draft_chunk(plan, records, &slice).await?; if !draft.over_limits() { - out.push(sign_chunk(plan, draft).await?); + let chunk = sign_chunk(plan, draft).await?; + write_chunk(plan, &chunk)?; + out.push(chunk); return Ok(()); } if slice.len() == 1 { return Err(ChainError::ChunkTooLarge); } let (head, tail) = slice.split_at(slice.len() / 2); - Box::pin(emit_chunk(plan, records, all_keys, head.to_vec(), out)).await?; - Box::pin(emit_chunk(plan, records, all_keys, tail.to_vec(), out)).await + Box::pin(emit_chunk(plan, records, head.to_vec(), out)).await?; + Box::pin(emit_chunk(plan, records, tail.to_vec(), out)).await +} + +fn write_chunk(plan: &mut Plan<'_, S>, chunk: &CommitChunk) -> Result<(), ChainError> { + let frame = plan.emit.frame(chunk); + let grown = blocks::merge_tree(plan.repo, plan.blocks_tree, &chunk.blocks)?; + let written = write_chain_commit( + plan.repo, + plan.parent, + &chunk.commit, + plan.now, + Some(grown), + frame.clone(), + )?; + plan.parent = Some(written); + plan.blocks_tree = Some(grown); + plan.emit.committed(frame); + Ok(()) } async fn draft_chunk( plan: &Plan<'_, S>, - records: &BTreeMap, - all_keys: &BTreeSet, + records: &dyn Records, 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(body) = records.get(key.as_str()) { - blocks.insert(*cid, Bytes::copy_from_slice(body.as_bytes())); + let mut blocks = + crate::proof::covering_blocks(&plan.tree, &applied.tree, &applied.diff).await?; + for op in ops { + let (key, cid) = match op { + VerifiedWriteOp::Create { key, cid } | VerifiedWriteOp::Update { key, cid, .. } => { + (key, cid) } + VerifiedWriteOp::Delete { .. } => continue, + }; + if !records.placed().contains_key(key.as_str()) { + continue; } - VerifiedWriteOp::Delete { .. } => {} - }); - let wire = wire_ops(ops); + let key = MstKey::new(key.as_str().to_owned()) + .map_err(|error| ChainError::Malformed(error.to_string()))?; + let body = records.body(&key)?; + blocks.insert(*cid, Bytes::copy_from_slice(body.as_bytes())); + } Ok(ChunkDraft { data: mst::root(&applied.tree).await?, - ops_wire: wire - .iter() - .map(|op| OP_FRAME_OVERHEAD_BYTES + op.key().len()) - .sum(), - ops: wire, + ops: ops_of(ops), blocks, tree: applied.tree, }) @@ -858,6 +929,7 @@ mod tests { use jacquard_repo::BlockStore; use knot_runtime::K256Signer; use knot_types::AccountDid; + use proptest::strategy::Strategy; use std::collections::BTreeSet; const DID: &str = "did:web:knot.example"; @@ -876,18 +948,15 @@ mod tests { fn records(count: usize) -> BTreeMap { (0..count) .map(|n| { + let collection = crate::member_invite_collection(); let record = crate::InviteRecord::new( - crate::MEMBER_INVITE_COLLECTION, + crate::InviteCollection::Member, UnixSeconds::new(n as i64), AccountDid::new("did:plc:3fwecdnvtcscjnrx2p4n7alz").unwrap(), ) .unwrap(); ( - MstKey::new(format!( - "{}/did:plc:subject{n}", - crate::MEMBER_INVITE_COLLECTION - )) - .unwrap(), + MstKey::new(format!("{}/did:plc:subject{n}", collection.as_str())).unwrap(), record.body().unwrap(), ) }) @@ -975,6 +1044,10 @@ mod tests { "empty roll doesn't materialize any commit at all" ); assert!(tip(&repo).unwrap().is_none()); + assert!( + leaves_of(&repo).await.is_empty(), + "a chain that never materialized must answer an empty leaf list rather than failing the reader" + ); let first = enable(&repo, &records(1), &key()).await.unwrap(); assert!(matches!( enable(&repo, &records(1), &key()).await, @@ -1055,13 +1128,7 @@ mod tests { } async fn leaves_of(repo: &Repo) -> BTreeMap { - load_tree(repo, &tip(repo).unwrap().unwrap()) - .leaves() - .await - .unwrap() - .into_iter() - .map(|(key, cid)| (MstKey::new(key.as_str().to_owned()).unwrap(), cid)) - .collect() + leaves(repo).await.unwrap() } #[tokio::test] @@ -1076,7 +1143,7 @@ mod tests { records.insert( MstKey::new("sh.tangled.knot.memberInvite/did:plc:subject3".to_owned()).unwrap(), crate::InviteRecord::new( - crate::MEMBER_INVITE_COLLECTION, + crate::InviteCollection::Member, UnixSeconds::new(99), AccountDid::new("did:plc:3fwecdnvtcscjnrx2p4n7alz").unwrap(), ) @@ -1087,8 +1154,8 @@ mod tests { let chunks = sync( &repo, &repo_did(), - &crate::member_invite_collection(), - &records, + std::slice::from_ref(&crate::member_invite_collection()), + &in_tree(records.clone()), &signer, &mut Ticker::new(), UnixSeconds::new(2), @@ -1137,8 +1204,8 @@ mod tests { let chunks = sync( &repo, &repo_did(), - &crate::member_invite_collection(), - &records, + std::slice::from_ref(&crate::member_invite_collection()), + &in_tree(records), &signer, &mut Ticker::new(), UnixSeconds::new(2), @@ -1156,11 +1223,6 @@ mod tests { chunks.iter().for_each(|chunk| { assert!(chunk.ops.len() <= MAX_OPS_PER_COMMIT); assert!(blocks_bytes(&chunk.blocks) <= MAX_COMMIT_BLOCKS_BYTES); - assert!( - blocks_bytes(&chunk.blocks) - + chunk.ops.iter().map(|op| op.key().len()).sum::() - <= MAX_FRAME_BYTES - ); }); chunks.windows(2).for_each(|pair| { assert!(pair[1].rev.newer_than(&pair[0].rev)); @@ -1196,8 +1258,8 @@ mod tests { sync( &repo, &repo_did(), - &crate::member_invite_collection(), - &records(1), + std::slice::from_ref(&crate::member_invite_collection()), + &in_tree(records(1)), &key(), &mut Ticker::new(), UnixSeconds::new(1) @@ -1213,8 +1275,8 @@ mod tests { sync( &repo, &repo_did(), - &crate::member_invite_collection(), - &records, + std::slice::from_ref(&crate::member_invite_collection()), + &in_tree(records), &signer, &mut Ticker::new(), UnixSeconds::new(2) @@ -1241,8 +1303,8 @@ mod tests { let chunks = sync( &repo, &repo_did(), - &crate::member_invite_collection(), - &records(2), + std::slice::from_ref(&crate::member_invite_collection()), + &in_tree(records(2)), &signer, &mut Ticker::new(), UnixSeconds::new(2), @@ -1271,8 +1333,8 @@ mod tests { sync( &repo, &repo_did(), - &crate::member_invite_collection(), - &outside, + std::slice::from_ref(&crate::member_invite_collection()), + &in_tree(outside), &signer, &mut Ticker::new(), UnixSeconds::new(3) @@ -1291,7 +1353,7 @@ mod tests { records.insert( MstKey::new("sh.tangled.knot.memberInvite/did:plc:late".to_owned()).unwrap(), crate::InviteRecord::new( - crate::MEMBER_INVITE_COLLECTION, + crate::InviteCollection::Member, UnixSeconds::new(7), AccountDid::new("did:plc:3fwecdnvtcscjnrx2p4n7alz").unwrap(), ) @@ -1303,8 +1365,8 @@ mod tests { sync( &repo, &repo_did(), - &crate::member_invite_collection(), - &records, + std::slice::from_ref(&crate::member_invite_collection()), + &in_tree(records.clone()), &signer, &mut Ticker::new(), UnixSeconds::new(2), @@ -1373,8 +1435,8 @@ mod tests { sync( &repo, &repo_did(), - &crate::member_invite_collection(), - &records(1), + std::slice::from_ref(&crate::member_invite_collection()), + &in_tree(records(1)), &key(), &mut Ticker::new(), UnixSeconds::new(2), @@ -1423,8 +1485,8 @@ mod tests { let appended = sync( &repo, &repo_did(), - &crate::member_invite_collection(), - &records(4), + std::slice::from_ref(&crate::member_invite_collection()), + &in_tree(records(4)), &key(), &mut emit, UnixSeconds::new(2), @@ -1432,7 +1494,7 @@ mod tests { .await .unwrap(); let start = tip(&repo).unwrap().unwrap().git; - let mut read = framed_revs(&repo, start, 0, usize::MAX) + let mut read = framed_revs(&repo, start, UnixMicros::new(0), usize::MAX) .found .into_iter() .map(|(_micros, oid)| frame_blob(&repo, oid).unwrap().unwrap()) @@ -1447,19 +1509,22 @@ mod tests { .collect::>(), "the walk hands back every stored frame, oldest first" ); - let one_short = framed_revs(&repo, start, 0, 1); + let one_short = framed_revs(&repo, start, UnixMicros::new(0), 1); assert_eq!( one_short.found, - vec![(tip(&repo).unwrap().unwrap().rev.timestamp(), start)], + vec![( + UnixMicros::new(tip(&repo).unwrap().unwrap().rev.timestamp()), + start, + )], "a walk one frame short keeps only the newest commit" ); assert!( one_short.hit_limit, - "the walk names the history it left behind" + "the walk should report the history it left behind" ); assert!( - !framed_revs(&repo, start, 0, 2).hit_limit, - "a walk that reaches the bottom of the chain names no cap" + !framed_revs(&repo, start, UnixMicros::new(0), 2).hit_limit, + "a walk shouldn't hit the limit when it reaches the bottom of the chain" ); } @@ -1496,58 +1561,126 @@ mod tests { records[address.mst_key().as_str()].cid() } + struct Framing; + + impl Emit for Framing { + fn rev(&mut self, prev: Option<&Tid>) -> Tid { + Ticker::new().next(prev.cloned()) + } + + fn frame(&self, chunk: &CommitChunk) -> Option { + Some(Bytes::from( + chunk + .blocks + .keys() + .map(Cid::to_string) + .collect::>() + .join("\n"), + )) + } + + fn committed(&mut self, _frame: Option) {} + } + #[tokio::test] - async fn apply_appends_one_op_and_leaves_every_other_record_where_it_was() { + async fn planned_chunk_keeps_leaf_in_git_and_frame_lists_it() { let (_dir, repo) = repo(); - let base = issues(4); - enable(&repo, &base, &key()).await.unwrap(); - let before = tip(&repo).unwrap().unwrap(); - + enable(&repo, &issues(1), &key()).await.unwrap(); + let body = issue_body(7); let chunks = apply( &repo, &repo_did(), &[( - issue_address(9), + issue_address(7), RecordOp::Create { - record: issue_body(9), + record: body.clone(), }, )], &key(), - &mut Ticker::new(), + &mut Framing, UnixSeconds::new(2), ) .await .unwrap(); - - assert_eq!(chunks.len(), 1, "one op is one commit"); + let [chunk] = chunks.as_slice() else { + panic!("one op must be one commit, found {}", chunks.len()); + }; + let leaf = body.cid().cid(); + assert!( + chunk.blocks.contains_key(&leaf), + "the planned chunk must include the record's bytes" + ); + let tip = tip(&repo).unwrap().unwrap(); + let stored = frame_blob(&repo, tip.git).unwrap().unwrap(); + assert!( + std::str::from_utf8(&stored) + .unwrap() + .contains(&leaf.to_string()), + "the stored frame oughtta list every block it wrote, including the leaf" + ); + assert!( + std::str::from_utf8(&stored) + .unwrap() + .contains(&tip.data.to_string()), + "the stored frame ought to keep the MST proof" + ); + let git_only = GitBlocks::at(&repo, tip.blocks); 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" + git_only.get(&leaf).await.unwrap().as_deref(), + Some(body.as_bytes()), + "the git block tree must answer the leaf directly" ); - 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" + leaves_of(&repo).await.get(&issue_address(7).mst_key()), + Some(&leaf), + "the MST must point at the leaf whichever store answers it" ); + } - let leaves = leaves_of(&repo).await; + #[tokio::test] + async fn chunks_split_at_blocks_byte_limit() { + let (_dir, repo) = repo(); + enable(&repo, &issues(1), &key()).await.unwrap(); + let heft = |n: u64| { + RecordBody::new( + serde_ipld_dagcbor::to_vec(&BTreeMap::from([ + ("$type", "sh.tangled.repo.issue".to_owned()), + ("title", format!("kelp {n}")), + ("body", "k".repeat(900_000)), + ])) + .unwrap(), + ) + .unwrap() + }; + let ops: Vec<(RecordAddress, RecordOp)> = (0..3) + .map(|n| (issue_address(100 + n), RecordOp::Create { record: heft(n) })) + .collect(); + let chunks = apply( + &repo, + &repo_did(), + &ops, + &key(), + &mut Framing, + UnixSeconds::new(3), + ) + .await + .unwrap(); assert_eq!( - leaves.len(), - 5, - "four records were live and one arrived, with no resync in between" + chunks + .iter() + .map(|chunk| chunk.ops.len()) + .collect::>(), + vec![1, 2], + "three records past the blocks byte limit oughtta split, so no commit draft keeps \ + unbounded record bytes in memory" ); - base.keys().for_each(|key| { - assert!(leaves.contains_key(key), "{key} went missing under apply"); + chunks.iter().for_each(|chunk| { + assert!( + blocks_bytes(&chunk.blocks) <= MAX_COMMIT_BLOCKS_BYTES, + "each commit's blocks should stay under the {MAX_COMMIT_BLOCKS_BYTES} limit" + ); }); } - #[tokio::test] async fn apply_names_the_current_cid_when_the_record_isnt_where_the_writer_left_it() { let (_dir, repo) = repo(); @@ -1573,25 +1706,24 @@ mod tests { 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 Err(ChainError::RecordMoved { + uri, + current, + claimed, + }) = &moved + else { + panic!("Edit from version that moved on should be rejected, got {moved:?}") + }; + assert_eq!( + current, &live, + "Moved record ought to answer with the current version" + ); + assert_eq!(claimed, &stale); + assert_eq!( + uri, + &issue_address(0).at_uri(&repo_did()), + "and should include the at-uri to rebase onto" + ); let existing = apply( &repo, @@ -1656,7 +1788,9 @@ mod tests { ); } - fn op_map(chunks: &[CommitChunk]) -> BTreeMap, Option)> { + type OpMap = BTreeMap, Option)>; + + fn op_map(chunks: &[CommitChunk]) -> OpMap { chunks .iter() .flat_map(|chunk| chunk.ops.iter()) @@ -1664,11 +1798,6 @@ mod tests { .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() @@ -1676,122 +1805,297 @@ mod tests { .expect("Current-thread runtime builds") } + const BASE_ISSUES: u64 = 20; + const REACH: u64 = 30; + const OVER_LIMIT_ISSUES: u64 = 240; + const ENABLED_AT: u64 = 1_700_000_000_000_000; + const WRITTEN_AT: u64 = 1_800_000_000_000_000; + + struct Stamped(u64); + + impl Emit for Stamped { + fn rev(&mut self, _prev: Option<&Tid>) -> Tid { + self.0 += 1; + Tid::from_time(self.0, 0) + } + + fn frame(&self, _chunk: &CommitChunk) -> Option { + None + } + + fn committed(&mut self, _frame: Option) {} + } + + fn faces(chunks: &[CommitChunk]) -> Vec<(CommitChunk, OpMap)> { + chunks + .iter() + .map(|chunk| { + ( + CommitChunk { + ops: Vec::new(), + ..chunk.clone() + }, + op_map(std::slice::from_ref(chunk)), + ) + }) + .collect() + } + + fn op_of(existing: Option, action: u8, n: u64) -> Option { + match (existing, action) { + (Some(body), 1) => Some(RecordOp::Update { + from: body.cid(), + record: body, + }), + (Some(body), 2) => Some(RecordOp::Update { + from: body.cid(), + record: issue_body(1_000 + n), + }), + (Some(body), 3) => Some(RecordOp::Delete { from: body.cid() }), + (Some(_), _) | (None, 0) => None, + (None, _) => Some(RecordOp::Create { + record: issue_body(n), + }), + } + } + + fn wanted(base: &BTreeMap, actions: &[u8]) -> BTreeMap { + actions + .iter() + .enumerate() + .filter_map(|(index, action)| { + let n = index as u64; + let key = issue_address(n).mst_key(); + let existing = base.get(&key).cloned(); + match op_of(existing.clone(), *action, n) { + Some(op) => op.record().cloned().map(|record| (key, record)), + None => existing.map(|body| (key, body)), + } + }) + .collect() + } + + fn ops_in_order( + base: &BTreeMap, + actions: &[u8], + order: &[u64], + ) -> Vec<(RecordAddress, RecordOp)> { + order + .iter() + .filter_map(|n| { + let address = issue_address(*n); + let existing = base.get(address.mst_key().as_str()).cloned(); + op_of(existing, actions[*n as usize], *n).map(|op| (address, op)) + }) + .collect() + } + + async fn synced_and_applied( + base: &BTreeMap, + desired: BTreeMap, + ops: &[(RecordAddress, RecordOp)], + ) -> (Vec, Cid, Vec, Cid) { + let (_sync_dir, sync_repo) = repo(); + let (_apply_dir, apply_repo) = repo(); + let enabled = async |repo: &Repo| { + materialize( + repo, + &repo_did(), + base, + &key(), + &mut Stamped(ENABLED_AT), + UnixSeconds::new(1), + ) + .await + .unwrap() + }; + let one_way = enabled(&sync_repo).await; + let other_way = enabled(&apply_repo).await; + assert_eq!( + faces(&one_way), + faces(&other_way), + "two chains over one record set must start from one chain" + ); + let synced = sync( + &sync_repo, + &repo_did(), + std::slice::from_ref(&issue_collection()), + &in_tree(desired), + &key(), + &mut Stamped(WRITTEN_AT), + UnixSeconds::new(2), + ) + .await + .unwrap(); + let applied = apply( + &apply_repo, + &repo_did(), + ops, + &key(), + &mut Stamped(WRITTEN_AT), + UnixSeconds::new(2), + ) + .await + .unwrap(); + ( + synced, + tip(&sync_repo).unwrap().unwrap().data, + applied, + tip(&apply_repo).unwrap().unwrap().data, + ) + } + + async fn inverts_cold(chunks: &[CommitChunk]) -> bool { + futures::stream::iter(chunks) + .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 + } + proptest::proptest! { - #![proptest_config(proptest::prelude::ProptestConfig { cases: 24, ..proptest::prelude::ProptestConfig::default() })] + #![proptest_config(proptest::prelude::ProptestConfig { cases: 16, ..proptest::prelude::ProptestConfig::default() })] #[test] - fn a_run_of_apply_reaches_the_root_one_sync_reaches( - actions in proptest::collection::vec(0u8..3, 9) + fn apply_in_any_order_emits_chunk_that_sync_emits( + drawn in proptest::collection::vec(0u8..5, REACH as usize), + order in Strategy::prop_shuffle(proptest::prelude::Just((0..=REACH).collect::>())) ) { - let base = issues(6); - let (_sync_dir, sync_repo) = repo(); - let (_apply_dir, apply_repo) = repo(); + let actions: Vec = drawn.iter().copied().chain([2]).collect(); + let base = issues(BASE_ISSUES); + let ops = ops_in_order(&base, &actions, &order); 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)) - })?; + let (synced, sync_root, applied, apply_root) = runtime.block_on( + synced_and_applied(&base, wanted(&base, &actions), &ops), + ); proptest::prop_assert_eq!( - op_map(&applied), - op_map(&synced), - "Run of apply names the same ops the whole-collection diff derives" + faces(&applied), + faces(&synced), + "whatever order its ops arrive in, apply must sign the commit the \ + whole-collection diff signs and include the same proof blocks" ); proptest::prop_assert_eq!( - tip(&apply_repo).unwrap().unwrap().data, - tip(&sync_repo).unwrap().unwrap().data, + apply_root, + sync_root, "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(()) - })?; + proptest::prop_assert!( + runtime.block_on(inverts_cold(&applied)), + "every chunk apply emits inverts cold out of its own blocks" + ); } } + #[tokio::test] + async fn one_op_apply_covers_neighbors_whole_collection_sync_covers() { + let base = issues(BASE_ISSUES); + let base = &base; + futures::stream::iter([ + (11u64, 2u8, "an update"), + (7, 3, "a delete"), + (BASE_ISSUES + 2, 2, "a create"), + ]) + .for_each(|(n, action, shape)| async move { + let actions: Vec = (0..BASE_ISSUES + 3) + .map(|index| if index == n { action } else { 0 }) + .collect(); + let ops = ops_in_order(base, &actions, &[n]); + let (synced, sync_root, applied, apply_root) = + synced_and_applied(base, wanted(base, &actions), &ops).await; + assert_eq!( + faces(&applied), + faces(&synced), + "{shape} written through apply must cover the MST neighbors the \ + whole-collection diff covers for the very same op" + ); + assert_eq!( + apply_root, sync_root, + "{shape} oughtta reach one root either way" + ); + assert!( + inverts_cold(&applied).await, + "{shape} ought to invert cold out of the blocks its own frame includes" + ); + }) + .await; + } + + #[tokio::test] + async fn run_over_chunk_limit_emits_commits_that_sync_emits() { + let actions: Vec = (0..320u64) + .map(|n| match n { + 0..70 => 3, + 70..140 => 2, + 140..150 => 1, + 150..OVER_LIMIT_ISSUES => 0, + _ => 2, + }) + .collect(); + let order: Vec = (70..150).chain(240..320).chain(0..70).collect(); + let base = issues(OVER_LIMIT_ISSUES); + let ops = ops_in_order(&base, &actions, &order); + let (synced, sync_root, applied, apply_root) = + synced_and_applied(&base, wanted(&base, &actions), &ops).await; + + assert_eq!( + applied + .iter() + .map(|chunk| chunk.ops.len()) + .collect::>(), + vec![200, 20], + "the ten byte-identical updates among them produce no op, and 220 ops in one apply \ + ought to cross the 200-op limit into two commits" + ); + assert_eq!( + faces(&applied), + faces(&synced), + "both paths oughtta sign the same commits past the limit and cover them with \ + the same blocks" + ); + assert_eq!(apply_root, sync_root); + } + + #[tokio::test] + async fn update_to_unchanged_bytes_skips_commit() { + let (_dir, repo) = repo(); + let base = issues(3); + enable(&repo, &base, &key()).await.unwrap(); + let before = tip(&repo).unwrap().unwrap(); + let chunks = apply( + &repo, + &repo_did(), + &[( + issue_address(1), + RecordOp::Update { + record: issue_body(1), + from: cid_at(&base, &issue_address(1)), + }, + )], + &key(), + &mut Ticker::new(), + UnixSeconds::new(2), + ) + .await + .unwrap(); + assert!( + chunks.is_empty(), + "a write of the bytes already at the address mustn't be a write, and mustn't create \ + a signed commit" + ); + assert_eq!( + tip(&repo).unwrap().unwrap().git, + before.git, + "the ref shouldn't move for it either" + ); + } + #[test] - fn a_tree_entry_sorts_as_git_sorts_it_with_a_subtree_slash() { + fn tree_entry_sorts_where_git_sorts_it_with_subtree_slash() { let entry = |name: &str, kind: gix::objs::tree::EntryKind| gix::objs::tree::Entry { mode: kind.into(), filename: name.into(), diff --git a/knot2/crates/knot-record/src/git_ref.rs b/knot2/crates/knot-record/src/git_ref.rs index 44181011b..372e47f88 100644 --- a/knot2/crates/knot-record/src/git_ref.rs +++ b/knot2/crates/knot-record/src/git_ref.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use crate::blocks::GitBlocks; use crate::chain::{self, ChainError}; +use crate::{dag_cbor, encoded}; use futures::StreamExt; use futures::TryStreamExt as _; use jacquard_repo::storage::BlockStore; @@ -89,11 +90,11 @@ impl RefRecord { } pub fn bytes(&self) -> Vec { - serde_ipld_dagcbor::to_vec(self).expect("text fields always encode as dag-cbor") + dag_cbor(self) } pub fn body(&self) -> Result { - RecordBody::new(self.bytes()) + encoded(self) } } @@ -209,6 +210,7 @@ fn stored_records(repo: &Repo, prefix: &str) -> Result Vec { - hex.as_bytes() - .chunks_exact(2) - .map(|pair| u8::from_str_radix(std::str::from_utf8(pair).unwrap(), 16).unwrap()) - .collect() - } - #[test] fn the_ref_record_bytes_are_frozen() { let pushed = RefRecord::new( @@ -343,11 +338,11 @@ mod tests { futures::executor::block_on(chain::sync( &repo, &RepoDid::new("did:web:knot.example").unwrap(), - &RecordCollection::new(GIT_REF_COLLECTION).unwrap(), - &BTreeMap::from([( + std::slice::from_ref(&RecordCollection::new(GIT_REF_COLLECTION).unwrap()), + &chain::in_tree(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), diff --git a/knot2/crates/knot-record/src/lib.rs b/knot2/crates/knot-record/src/lib.rs index feda3ff96..76cbc353b 100644 --- a/knot2/crates/knot-record/src/lib.rs +++ b/knot2/crates/knot-record/src/lib.rs @@ -1,73 +1,133 @@ pub mod blocks; pub mod chain; pub mod git_ref; +pub mod issue; pub mod mst; pub mod proof; use bytes::Bytes; use cid::Cid; use knot_types::{AccountDid, RecordBody, RecordBodyError, UnixSeconds}; -use serde::Serialize; +use serde::{Serialize, Serializer}; #[derive(Debug, thiserror::Error)] pub enum RecordError { #[error("atproto block error: {0}")] Block(#[from] jacquard_repo::RepoError), #[error("covering proof failed to invert the commit cold")] ColdInversion, + #[error("malformed atproto record: {0}")] + Malformed(String), #[error("createdAt {0} is outside the rfc3339 range")] CreatedAtUnrenderable(i64), } pub type Result = std::result::Result; -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") + InviteCollection::Member.collection() } pub fn collaborator_invite_collection() -> knot_types::RecordCollection { - knot_types::RecordCollection::new(COLLABORATOR_INVITE_COLLECTION) - .expect("Collaborator invite collection is valid nsid") + InviteCollection::Collaborator.collection() } -#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum InviteCollection { + Member, + Collaborator, +} + +impl InviteCollection { + pub const fn as_str(self) -> &'static str { + match self { + Self::Member => "sh.tangled.knot.memberInvite", + Self::Collaborator => "sh.tangled.repo.collaboratorInvite", + } + } + + pub fn collection(self) -> knot_types::RecordCollection { + knot_types::RecordCollection::new(self.as_str()).expect("invite collection is a valid nsid") + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] pub struct InviteRecord { - #[serde(rename = "$type")] - record_type: &'static str, - #[serde(rename = "createdAt")] - created_at: String, - #[serde(rename = "x-tngl-editor")] + kind: InviteCollection, + created_at: CreatedAt, editor: AccountDid, } impl InviteRecord { pub fn new( - collection: &'static str, + kind: InviteCollection, created_at: UnixSeconds, editor: AccountDid, ) -> Result { Ok(Self { - record_type: collection, - created_at: chrono::DateTime::from_timestamp(created_at.get(), 0) - .map(|time| time.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)) - .ok_or(RecordError::CreatedAtUnrenderable(created_at.get()))?, + kind, + created_at: CreatedAt::at(created_at)?, editor, }) } pub fn bytes(&self) -> Vec { - serde_ipld_dagcbor::to_vec(self).expect("three text fields always encode as dag-cbor") + dag_cbor(self) } pub fn body(&self) -> Result { - RecordBody::new(self.bytes()) + encoded(self) } } +#[derive(Serialize)] +struct InviteEncoding<'a> { + #[serde(rename = "$type")] + record_type: knot_types::RecordCollection, + #[serde(rename = "createdAt")] + created_at: &'a CreatedAt, + #[serde(rename = "x-tngl-editor")] + editor: &'a AccountDid, +} + +impl Serialize for InviteRecord { + fn serialize(&self, serializer: S) -> Result { + InviteEncoding { + record_type: self.kind.collection(), + created_at: &self.created_at, + editor: &self.editor, + } + .serialize(serializer) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(transparent)] +pub struct CreatedAt(String); + +impl CreatedAt { + pub fn at(time: UnixSeconds) -> Result { + chrono::DateTime::from_timestamp(time.get(), 0) + .map(|at| Self(at.to_rfc3339_opts(chrono::SecondsFormat::Secs, true))) + .ok_or(RecordError::CreatedAtUnrenderable(time.get())) + } +} + +pub(crate) fn dag_cbor(value: &T) -> Vec { + serde_ipld_dagcbor::to_vec(value).expect("record text fields and lists encode as dag-cbor") +} + +pub(crate) fn encoded(value: &T) -> Result { + RecordBody::new(dag_cbor(value)) +} + +#[cfg(test)] +pub(crate) fn frozen(hex: &str) -> Vec { + hex.as_bytes() + .chunks_exact(2) + .map(|pair| u8::from_str_radix(std::str::from_utf8(pair).unwrap(), 16).unwrap()) + .collect() +} + pub fn cid_of(block: &[u8]) -> Cid { knot_types::dag_cbor_cid(block) } @@ -87,15 +147,88 @@ pub fn car_bytes(root: Cid, blocks: &std::collections::BTreeMap) -> out } -fn varint(out: &mut Vec, mut value: usize) { - loop { - let byte = (value & 0x7f) as u8; - value >>= 7; - out.push(byte | if value == 0 { 0 } else { 0x80 }); - if value == 0 { - return; - } - } +fn varint(out: &mut Vec, value: usize) { + let chunks = std::iter::successors(Some(value), |value| (*value >= 0x80).then(|| value >> 7)); + out.extend(chunks.map(|value| ((value & 0x7f) as u8) | u8::from(value >= 0x80) << 7)); +} + +#[derive(Debug, thiserror::Error)] +pub enum CarError { + #[error("car is cut short at byte {0}")] + Truncated(usize), + #[error("car varint at byte {0} doesn't fit a length")] + Varint(usize), + #[error("car header doesn't decode: {0}")] + Header(String), + #[error("car section at byte {at} is missing its cid: {reason}")] + Cid { at: usize, reason: String }, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Car { + pub roots: Vec, + pub blocks: std::collections::BTreeMap, +} + +struct Varint { + value: usize, + after: usize, +} + +fn read_varint(bytes: &[u8], at: usize) -> Result { + let tail = bytes.get(at..).ok_or(CarError::Truncated(at))?; + let last = tail + .iter() + .take(10) + .position(|byte| byte & 0x80 == 0) + .ok_or(CarError::Truncated(at))?; + let value = tail[..=last] + .iter() + .enumerate() + .try_fold(0usize, |value, (index, byte)| { + let part = usize::from(byte & 0x7f); + let shift = (7 * index) as u32; + part.checked_shl(shift) + .filter(|shifted| shifted >> shift == part) + .map(|shifted| value | shifted) + }) + .ok_or(CarError::Varint(at))?; + Ok(Varint { + value, + after: at + last + 1, + }) +} + +pub fn car_blocks(bytes: &[u8]) -> Result { + let framed = read_varint(bytes, 0)?; + let header = bytes + .get(framed.after..framed.after + framed.value) + .ok_or(CarError::Truncated(framed.after))?; + let roots = iroh_car::CarHeader::decode(header) + .map_err(|error| CarError::Header(error.to_string()))? + .roots() + .to_vec(); + let mut at = framed.after + framed.value; + let blocks = std::iter::from_fn(|| { + (at < bytes.len()).then(|| { + let framed = read_varint(bytes, at)?; + let section = bytes + .get(framed.after..framed.after + framed.value) + .ok_or(CarError::Truncated(framed.after))?; + let mut cursor = std::io::Cursor::new(§ion); + let cid = Cid::read_bytes(&mut cursor).map_err(|error| CarError::Cid { + at: framed.after, + reason: error.to_string(), + })?; + at = framed.after + framed.value; + Ok(( + cid, + Bytes::copy_from_slice(§ion[cursor.position() as usize..]), + )) + }) + }) + .collect::, CarError>>()?; + Ok(Car { roots, blocks }) } #[cfg(test)] @@ -120,6 +253,33 @@ mod car_tests { .unwrap() ); } + + #[test] + fn car_reader_inverts_car_writer() { + let one = cid_of(b"a record body"); + let root = cid_of(b"a commit"); + let blocks = BTreeMap::from([ + (one, Bytes::from_static(b"a record body")), + (root, Bytes::from_static(b"a commit")), + ]); + let written = car_bytes(root, &blocks); + assert_eq!( + car_blocks(&written).unwrap(), + Car { + roots: vec![root], + blocks: blocks.clone() + } + ); + assert_eq!( + car_bytes(root, &car_blocks(&written).unwrap().blocks), + written, + "a read then written car is the same bytes" + ); + assert!(matches!( + car_blocks(&written[..written.len() - 3]), + Err(CarError::Truncated(_)) + )); + } } pub struct CommitSigner<'a>(&'a dyn knot_runtime::Signer); @@ -144,6 +304,8 @@ impl jacquard_repo::commit::SigningKey for CommitSigner<'_> { mod tests { use super::*; + const MEMBER: &str = "a3652474797065781c73682e74616e676c65642e6b6e6f742e6d656d626572496e766974656963726561746564417474323032352d30382d31335431333a30313a33365a6d782d746e676c2d656469746f7278206469643a706c633a3366776563646e76746373636a6e72783270346e37616c7a"; + fn editor() -> AccountDid { AccountDid::new("did:plc:3fwecdnvtcscjnrx2p4n7alz").unwrap() } @@ -151,28 +313,18 @@ mod tests { #[test] fn the_invite_record_bytes_are_frozen() { let member = InviteRecord::new( - MEMBER_INVITE_COLLECTION, + InviteCollection::Member, UnixSeconds::new(1_755_090_096), editor(), ) .unwrap(); assert_eq!( member.bytes(), - [ - 0xa3, 0x65, b'$', b't', b'y', b'p', b'e', 0x78, 0x1c, b's', b'h', b'.', b't', b'a', - b'n', b'g', b'l', b'e', b'd', b'.', b'k', b'n', b'o', b't', b'.', b'm', b'e', b'm', - b'b', b'e', b'r', b'I', b'n', b'v', b'i', b't', b'e', 0x69, b'c', b'r', b'e', b'a', - b't', b'e', b'd', b'A', b't', 0x74, b'2', b'0', b'2', b'5', b'-', b'0', b'8', b'-', - b'1', b'3', b'T', b'1', b'3', b':', b'0', b'1', b':', b'3', b'6', b'Z', 0x6d, b'x', - b'-', b't', b'n', b'g', b'l', b'-', b'e', b'd', b'i', b't', b'o', b'r', 0x78, 0x20, - b'd', b'i', b'd', b':', b'p', b'l', b'c', b':', b'3', b'f', b'w', b'e', b'c', b'd', - b'n', b'v', b't', b'c', b's', b'c', b'j', b'n', b'r', b'x', b'2', b'p', b'4', b'n', - b'7', b'a', b'l', b'z', - ], - "the first emitted record freezes this wire format: $type first, length-first key order, seconds precision, utc z" + frozen(MEMBER), + "the first emitted record freezes these bytes: $type first, length-first key order, seconds precision, utc z" ); let collaborator = InviteRecord::new( - COLLABORATOR_INVITE_COLLECTION, + InviteCollection::Collaborator, UnixSeconds::new(1_755_090_096), editor(), ) @@ -188,7 +340,7 @@ mod tests { fn a_timestamp_outside_the_rfc3339_range_is_refused_at_construction() { assert!(matches!( InviteRecord::new( - MEMBER_INVITE_COLLECTION, + InviteCollection::Member, UnixSeconds::new(i64::MAX), editor() ), diff --git a/knot2/crates/knot-record/src/proof.rs b/knot2/crates/knot-record/src/proof.rs index 943575241..6a941c216 100644 --- a/knot2/crates/knot-record/src/proof.rs +++ b/knot2/crates/knot-record/src/proof.rs @@ -1,40 +1,45 @@ use bytes::Bytes; use cid::Cid; -use futures::StreamExt; -use jacquard_repo::mst::VerifiedWriteOp; -use jacquard_repo::storage::MemoryBlockStore; +use futures::{StreamExt, TryStreamExt}; +use jacquard_repo::mst::{NodeEntry, VerifiedWriteOp}; +use jacquard_repo::storage::{BlockStore, MemoryBlockStore}; use jacquard_repo::{Mst, MstDiff}; use std::collections::{BTreeMap, BTreeSet}; -use std::ops::Bound::{Excluded, Unbounded}; - -use knot_types::MstKey; +use std::pin::Pin; use crate::{RecordError, Result}; +use knot_types::MstKey; -pub async fn covering_blocks( +pub async fn covering_blocks( + before: &Mst, after: &Mst, diff: &MstDiff, - all_keys: &BTreeSet, ) -> Result> { - let mut blocks = diff.new_mst_blocks.clone(); - futures::stream::iter( - mutated_keys(diff) - .into_iter() - .flat_map(|key| neighbors(all_keys, &key)), - ) - .fold( - Ok::<&mut BTreeMap, RecordError>(&mut blocks), - |collected, key| async move { - let blocks = collected?; + let covered = futures::stream::iter(mutated_keys(diff)) + .then(|key| async move { + let (below_before, above_before) = around(before.clone(), &key, None, None).await?; + let (below_after, above_after) = around(after.clone(), &key, None, None).await?; + Ok::<[Option; 3], RecordError>([ + below_before.max(below_after), + nearest(above_before, above_after), + Some(key), + ]) + }) + .try_collect::>() + .await? + .into_iter() + .flatten() + .flatten() + .collect::>(); + futures::stream::iter(covered.into_iter().map(Ok::<_, RecordError>)) + .try_fold(diff.new_mst_blocks.clone(), |mut blocks, key| async move { after - .blocks_for_path(key.as_str(), blocks) + .blocks_for_path(key.as_str(), &mut blocks) .await .map_err(RecordError::Block)?; Ok(blocks) - }, - ) - .await?; - Ok(blocks) + }) + .await } pub async fn cold_invert( @@ -56,30 +61,108 @@ pub async fn cold_invert( tree.root().await.map_err(RecordError::Block) } -fn mutated_keys(diff: &MstDiff) -> BTreeSet { +fn mutated_keys(diff: &MstDiff) -> BTreeSet { diff.creates .iter() - .map(|(key, _)| key.as_str().to_owned()) - .chain( - diff.updates - .iter() - .map(|(key, _, _)| key.as_str().to_owned()), - ) - .chain(diff.deletes.iter().map(|(key, _)| key.as_str().to_owned())) + .map(|(key, _)| key.as_str()) + .chain(diff.updates.iter().map(|(key, _, _)| key.as_str())) + .chain(diff.deletes.iter().map(|(key, _)| key.as_str())) + .flat_map(|key| MstKey::new(key.to_owned())) .collect() } -fn neighbors(all_keys: &BTreeSet, key: &str) -> Vec { - let before = all_keys - .range::((Unbounded, Excluded(key))) - .next_back() - .cloned(); - let after = all_keys - .range::((Excluded(key), Unbounded)) - .next() - .cloned(); - [before, after, MstKey::new(key.to_owned()).ok()] - .into_iter() - .flatten() - .collect() +fn nearest(left: Option, right: Option) -> Option { + match (left, right) { + (Some(left), Some(right)) => Some(left.min(right)), + (left, right) => left.or(right), + } +} + +fn checked(value: Option<&str>) -> Result> { + value + .map(|key| MstKey::new(key.to_owned())) + .transpose() + .map_err(|error| RecordError::Malformed(error.to_string())) +} + +type Picker = for<'e> fn(&'e [NodeEntry]) -> Option<&'e NodeEntry>; + +fn edge( + tree: Mst, + pick: Picker, +) -> Pin>> + Send>> { + Box::pin(async move { + let entries = entries_of(&tree).await?; + match pick(&entries) { + Some(NodeEntry::Tree(under)) => edge(under.clone(), pick).await, + Some(entry) => checked(entry.leaf_key()), + None => Ok(None), + } + }) +} + +async fn toward( + entry: Option<&NodeEntry>, + inherited: Option, + pick: Picker, +) -> Result> { + match entry { + Some(NodeEntry::Tree(under)) => edge(under.clone(), pick).await, + Some(entry) => checked(entry.leaf_key()), + None => Ok(inherited), + } +} + +fn around<'a, S: BlockStore + Sync + 'static>( + tree: Mst, + key: &'a MstKey, + below: Option, + above: Option, +) -> Pin, Option)>> + Send + 'a>> { + Box::pin(async move { + let entries = entries_of(&tree).await?; + let at = entries + .iter() + .position(|entry| entry.leaf_key().is_some_and(|leaf| leaf >= key.as_str())) + .unwrap_or(entries.len()); + let here = entries.get(at).and_then(NodeEntry::leaf_key); + let left = at.checked_sub(1).and_then(|index| entries.get(index)); + let higher = match here.filter(|leaf| *leaf != key.as_str()) { + Some(leaf) => checked(Some(leaf))?.or(above), + None => above, + }; + if here == Some(key.as_str()) { + return Ok(( + toward(left, below, <[NodeEntry]>::last).await?, + toward(entries.get(at + 1), higher, <[NodeEntry]>::first).await?, + )); + } + match left { + Some(NodeEntry::Tree(under)) => { + let lower = checked( + at.checked_sub(2) + .and_then(|index| entries.get(index)) + .and_then(NodeEntry::leaf_key), + )? + .or(below); + around(under.clone(), key, lower, higher).await + } + left => Ok(( + checked(left.and_then(NodeEntry::leaf_key))?.or(below), + higher, + )), + } + }) +} + +async fn entries_of(tree: &Mst) -> Result>> { + futures::stream::unfold(0isize, |index| async move { + tree.at_index(index) + .await + .map_err(RecordError::Block) + .transpose() + .map(|entry| (entry, index + 1)) + }) + .try_collect() + .await } diff --git a/knot2/crates/knot-record/tests/proof_fixtures.rs b/knot2/crates/knot-record/tests/proof_fixtures.rs index 49db6def3..eccbc8be4 100644 --- a/knot2/crates/knot-record/tests/proof_fixtures.rs +++ b/knot2/crates/knot-record/tests/proof_fixtures.rs @@ -1,4 +1,3 @@ -use knot_types::MstKey; use std::collections::{BTreeMap, BTreeSet}; use std::path::PathBuf; @@ -79,24 +78,15 @@ async fn the_covering_set_is_a_superset_that_inverts_cold_with_a_bounded_size() vector["rootBeforeCommit"].as_str().unwrap(), "{comment}: the builder must reproduce rootBeforeCommit from the pairs alone" ); - let after = mst::apply( - &mst::tree_of(&pairs(&keys, leaf)).await.unwrap(), - &ops(vector, leaf), - ) - .await - .unwrap(); + let before = mst::tree_of(&pairs(&keys, leaf)).await.unwrap(); + let after = mst::apply(&before, &ops(vector, leaf)).await.unwrap(); let root = mst::root(&after.tree).await.unwrap(); assert_eq!( root.to_string(), vector["rootAfterCommit"].as_str().unwrap(), "{comment}: applying the vector's ops must reach rootAfterCommit" ); - let all_keys: BTreeSet = keys - .iter() - .chain(strings(vector, "adds").iter()) - .map(|key| MstKey::new(key.to_owned()).unwrap()) - .collect(); - let covering = proof::covering_blocks(&after.tree, &after.diff, &all_keys) + let covering = proof::covering_blocks(&before, &after.tree, &after.diff) .await .unwrap(); let expected: BTreeSet = strings(vector, "blocksInProof")