From f639192dc6ccd2c584e0f738fcb834b4f8cf59ed Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Thu, 13 Jul 2023 13:21:15 +0200 Subject: [PATCH] cob: Remove CRDTs from COB state It turns out that the CRDT formed by the union of Git DAGs is enough to guarantee everything we need for COBs. This changes the following things: * COB operations no longer need to be commutative * COB histories are traversed in the same deterministic order on all replicas * It's now possible to implement RSMs on top of COBs, eg. scripting * Lamport clocks have been removed * `radicle-crdt` is no longer a dependency of `radicle` * COBs are no longer instances of `Semilattice` * The `Ops` type was removed in favor of having `Op` contain multiple actions --- radicle-cli/src/commands/id.rs | 8 +- radicle-cli/src/commands/issue.rs | 2 +- radicle-cli/src/commands/patch/list.rs | 2 +- radicle-cob/src/change_graph.rs | 4 +- radicle-cob/src/history.rs | 23 +- radicle-cob/src/history/entry.rs | 12 - radicle-httpd/src/api/json.rs | 11 +- radicle-httpd/src/api/v1/projects.rs | 16 +- radicle-httpd/src/test.rs | 4 +- radicle-node/src/test/simulator.rs | 66 +- radicle-remote-helper/src/push.rs | 4 +- radicle-tui/src/cob/issue.rs | 2 +- radicle-tui/src/cob/patch.rs | 2 +- radicle/Cargo.toml | 4 - radicle/src/cob/common.rs | 5 +- radicle/src/cob/identity.rs | 157 ++-- radicle/src/cob/issue.rs | 369 ++++---- radicle/src/cob/op.rs | 55 +- radicle/src/cob/patch.rs | 864 ++++++------------ radicle/src/cob/store.rs | 93 +- radicle/src/cob/test.rs | 77 +- radicle/src/cob/thread.rs | 338 +++---- .../src/storage/git/transport/remote/mock.rs | 12 + radicle/src/test.rs | 285 +++++- 24 files changed, 1024 insertions(+), 1391 deletions(-) diff --git a/radicle-cli/src/commands/id.rs b/radicle-cli/src/commands/id.rs index aad8efe2..8c1a9bda 100644 --- a/radicle-cli/src/commands/id.rs +++ b/radicle-cli/src/commands/id.rs @@ -379,7 +379,7 @@ pub fn run(options: Options, ctx: impl term::Context) -> anyhow::Result<()> { let mut timestamped = Vec::new(); let mut no_latest = Vec::new(); for result in proposals.all()? { - let (id, proposal, _) = result?; + let (id, proposal) = result?; match proposal.latest() { None => no_latest.push((id, proposal)), Some((_, revision)) => { @@ -490,7 +490,7 @@ fn select<'a>( let revision = proposal .revision(&id) .context(format!("No revision found for {id}"))? - .get() + .as_ref() .context(format!("Revision {id} was redacted"))?; (id, revision) } @@ -524,7 +524,7 @@ fn commit_select<'a>( let revision = proposal .revision(&id) .context(format!("No revision found for {id}"))? - .get() + .as_ref() .context(format!("Revision {id} was redacted"))?; (id, revision) } @@ -623,7 +623,7 @@ fn print( Some(rid) => proposal .revision(rid) .context(format!("No revision found for {rid}"))? - .get() + .as_ref() .context(format!("Revision {rid} was redacted"))?, }; print_meta(proposal.title(), proposal.description(), proposal.state()); diff --git a/radicle-cli/src/commands/issue.rs b/radicle-cli/src/commands/issue.rs index 6aad2654..0dfa10b2 100644 --- a/radicle-cli/src/commands/issue.rs +++ b/radicle-cli/src/commands/issue.rs @@ -381,7 +381,7 @@ fn list( let mut all = Vec::new(); for result in issues.all()? { - let Ok((id, issue, _)) = result else { + let Ok((id, issue)) = result else { // Skip issues that failed to load. continue; }; diff --git a/radicle-cli/src/commands/patch/list.rs b/radicle-cli/src/commands/patch/list.rs index 2108bf28..94ce3acb 100644 --- a/radicle-cli/src/commands/patch/list.rs +++ b/radicle-cli/src/commands/patch/list.rs @@ -22,7 +22,7 @@ pub fn run( let mut all = Vec::new(); for patch in patches.all()? { - let Ok((id, patch, _)) = patch else { + let Ok((id, patch)) = patch else { // Skip patches that failed to load. continue; }; diff --git a/radicle-cob/src/change_graph.rs b/radicle-cob/src/change_graph.rs index 6b724278..d1e0d172 100644 --- a/radicle-cob/src/change_graph.rs +++ b/radicle-cob/src/change_graph.rs @@ -91,19 +91,17 @@ impl ChangeGraph { let manifest = root_node.manifest.clone(); let graph = self .graph - .fold(&root, Dag::new(), |mut graph, _, change, depth| { + .fold(&root, Dag::new(), |mut graph, _, change, _| { // Check the change signatures are valid. if !change.valid_signatures() { return ControlFlow::Break(graph); } - let clock = depth as u64 + 1; let entry = Entry::new( *change.id(), change.signature.key, change.resource, change.contents().clone(), change.timestamp, - clock, ); let id = *entry.id(); diff --git a/radicle-cob/src/history.rs b/radicle-cob/src/history.rs index d02f2844..59fdd5cc 100644 --- a/radicle-cob/src/history.rs +++ b/radicle-cob/src/history.rs @@ -7,7 +7,7 @@ use radicle_crypto::PublicKey; use radicle_dag::Dag; pub mod entry; -pub use entry::{Clock, Contents, Entry, EntryId, Timestamp}; +pub use entry::{Contents, Entry, EntryId, Timestamp}; /// The DAG of changes making up the history of a collaborative object. #[derive(Clone, Debug)] @@ -51,7 +51,6 @@ impl History { resource, contents, timestamp, - clock: 1, }; Self { @@ -60,16 +59,6 @@ impl History { } } - /// Get the current value of the logical clock. - /// This is the maximum value of all tips. - pub fn clock(&self) -> Clock { - self.graph - .tips() - .map(|(_, node)| node.clock) - .max() - .unwrap_or_default() - } - /// Get the current history timestamp. /// This is the latest timestamp of any tip. pub fn timestamp(&self) -> Timestamp { @@ -127,14 +116,8 @@ impl History { { let tips = self.tips(); let new_id = new_id.into(); - let new_entry = Entry::new( - new_id, - new_actor, - new_resource, - new_contents, - new_timestamp, - self.clock() + 1, - ); + let new_entry = Entry::new(new_id, new_actor, new_resource, new_contents, new_timestamp); + self.graph.node(new_id, new_entry); for tip in tips { diff --git a/radicle-cob/src/history/entry.rs b/radicle-cob/src/history/entry.rs index 82e7558f..bb4c6e32 100644 --- a/radicle-cob/src/history/entry.rs +++ b/radicle-cob/src/history/entry.rs @@ -15,9 +15,6 @@ use crate::{object, ObjectId}; /// This is the change payload. pub type Contents = NonEmpty>; -/// Logical clock used to track causality in change graph. -pub type Clock = u64; - /// Local time in seconds since epoch. pub type Timestamp = u64; @@ -93,8 +90,6 @@ pub struct Entry { pub(super) contents: Contents, /// The entry timestamp, as seconds since epoch. pub(super) timestamp: Timestamp, - /// Logical clock. - pub(super) clock: Clock, } impl Entry { @@ -104,7 +99,6 @@ impl Entry { resource: Oid, contents: Contents, timestamp: Timestamp, - clock: Clock, ) -> Self where Id: Into, @@ -115,7 +109,6 @@ impl Entry { resource, contents, timestamp, - clock, } } @@ -143,9 +136,4 @@ impl Entry { pub fn id(&self) -> &EntryId { &self.id } - - /// Logical clock. - pub fn clock(&self) -> Clock { - self.clock - } } diff --git a/radicle-httpd/src/api/json.rs b/radicle-httpd/src/api/json.rs index 3fa310cc..7408d7fa 100644 --- a/radicle-httpd/src/api/json.rs +++ b/radicle-httpd/src/api/json.rs @@ -138,7 +138,7 @@ pub(crate) fn patch( "discussions": rev.discussion().comments() .map(|(id, comment)| Comment::new(id, comment, aliases)) .collect::>(), - "timestamp": rev.timestamp(), + "timestamp": rev.timestamp().as_secs().to_string(), "reviews": rev.reviews().map(|(nid, _review)| review(nid, aliases.alias(nid), _review)).collect::>(), }) }).collect::>(), @@ -165,7 +165,7 @@ fn merge(merge: &Merge, nid: &NodeId, alias: Option) -> Value { "alias": alias, }, "commit": merge.commit, - "timestamp": merge.timestamp, + "timestamp": merge.timestamp.as_secs().to_string(), "revision": merge.revision, }), None => json!({ @@ -173,7 +173,7 @@ fn merge(merge: &Merge, nid: &NodeId, alias: Option) -> Value { "id": nid, }, "commit": merge.commit, - "timestamp": merge.timestamp, + "timestamp": merge.timestamp.as_secs().to_string(), "revision": merge.revision, }), } @@ -190,7 +190,7 @@ fn review(nid: &NodeId, alias: Option, review: &Review) -> Value { "verdict": review.verdict(), "summary": review.summary(), "comments": review.comments().collect::>(), - "timestamp": review.timestamp(), + "timestamp": review.timestamp().as_secs().to_string(), }), None => json!({ "author": { @@ -199,7 +199,7 @@ fn review(nid: &NodeId, alias: Option, review: &Review) -> Value { "verdict": review.verdict(), "summary": review.summary(), "comments": review.comments().collect::>(), - "timestamp": review.timestamp(), + "timestamp": review.timestamp().as_secs().to_string(), }), } } @@ -240,6 +240,7 @@ struct Comment<'a> { author: Value, body: &'a str, reactions: Vec<(&'a ActorId, &'a Reaction)>, + #[serde(with = "radicle::serde_ext::localtime::time")] timestamp: Timestamp, reply_to: Option, } diff --git a/radicle-httpd/src/api/v1/projects.rs b/radicle-httpd/src/api/v1/projects.rs index 67e7bfd0..910d002a 100644 --- a/radicle-httpd/src/api/v1/projects.rs +++ b/radicle-httpd/src/api/v1/projects.rs @@ -460,16 +460,16 @@ async fn issues_handler( let mut issues: Vec<_> = issues .all()? .filter_map(|r| { - let (id, issue, clock) = r.ok()?; - (state.matches(issue.state())).then_some((id, issue, clock)) + let (id, issue) = r.ok()?; + (state.matches(issue.state())).then_some((id, issue)) }) .collect::>(); - issues.sort_by(|(_, a, _), (_, b, _)| b.timestamp().cmp(&a.timestamp())); + issues.sort_by(|(_, a), (_, b)| b.timestamp().cmp(&a.timestamp())); let aliases = &ctx.profile.aliases(); let issues = issues .into_iter() - .map(|(id, issue, _)| api::json::issue(id, issue, aliases)) + .map(|(id, issue)| api::json::issue(id, issue, aliases)) .skip(page * per_page) .take(per_page) .collect::>(); @@ -739,15 +739,15 @@ async fn patches_handler( let mut patches = patches .all()? .filter_map(|r| { - let (id, patch, clock) = r.ok()?; - (state.matches(patch.state())).then_some((id, patch, clock)) + let (id, patch) = r.ok()?; + (state.matches(patch.state())).then_some((id, patch)) }) .collect::>(); - patches.sort_by(|(_, a, _), (_, b, _)| b.timestamp().cmp(&a.timestamp())); + patches.sort_by(|(_, a), (_, b)| b.timestamp().cmp(&a.timestamp())); let aliases = ctx.profile.aliases(); let patches = patches .into_iter() - .map(|(id, patch, _)| api::json::patch(id, patch, &repo, &aliases)) + .map(|(id, patch)| api::json::patch(id, patch, &repo, &aliases)) .skip(page * per_page) .take(per_page) .collect::>(); diff --git a/radicle-httpd/src/test.rs b/radicle-httpd/src/test.rs index 72b11e40..5191520d 100644 --- a/radicle-httpd/src/test.rs +++ b/radicle-httpd/src/test.rs @@ -37,7 +37,7 @@ pub const ISSUE_ID: &str = "5ad77fa3f476beed9a26f49b2b3b844e61bef792"; pub const ISSUE_DISCUSSION_ID: &str = "f1dff128a22e8183a23516dd9812e72e80914c92"; pub const ISSUE_COMMENT_ID: &str = "845218041bf9eb8155bfa4aaa8f0c91ce18e5c13"; pub const SESSION_ID: &str = "u9MGAkkfkMOv0uDDB2WeUHBT7HbsO2Dy"; -pub const TIMESTAMP: u64 = 1671125284; +pub const TIMESTAMP: &str = "1671125284"; pub const CONTRIBUTOR_RID: &str = "rad:z4XaCmN3jLSeiMvW15YTDpNbDHFhG"; pub const CONTRIBUTOR_DID: &str = "did:key:z6Mkk7oqY4pPxhMmGEotDYsFo97vhCj85BLY1H256HrJmjN8"; pub const CONTRIBUTOR_NID: &str = "z6Mkk7oqY4pPxhMmGEotDYsFo97vhCj85BLY1H256HrJmjN8"; @@ -97,7 +97,7 @@ fn seed_with_signer(dir: &Path, profile: radicle::Profile, signer: &G let workdir = dir.join("hello-world"); - env::set_var("RAD_COMMIT_TIME", TIMESTAMP.to_string()); + env::set_var("RAD_COMMIT_TIME", TIMESTAMP); fs::create_dir_all(&workdir).unwrap(); diff --git a/radicle-node/src/test/simulator.rs b/radicle-node/src/test/simulator.rs index 02a37223..37473825 100644 --- a/radicle-node/src/test/simulator.rs +++ b/radicle-node/src/test/simulator.rs @@ -14,12 +14,11 @@ use localtime::{LocalDuration, LocalTime}; use log::*; use crate::crypto::Signer; -use crate::git::raw as git; use crate::prelude::{Address, Id}; use crate::service::io::Io; use crate::service::{DisconnectReason, Event, Message, NodeId}; +use crate::storage::WriteStorage; use crate::storage::{Namespaces, RefUpdate}; -use crate::storage::{WriteRepository, WriteStorage}; use crate::test::peer::Service; use crate::worker::FetchError; use crate::Link; @@ -413,15 +412,19 @@ impl Simulation { } Input::Fetched(rid, nid, result) => { let result = Rc::try_unwrap(result).unwrap(); - let mut repo = match p.storage().repository_mut(rid) { + let repo = match p.storage().repository_mut(rid) { Ok(repo) => repo, Err(e) if e.is_not_found() => p.storage().create(rid).unwrap(), Err(e) => panic!("Failed to open repository: {e}"), }; match &result { Ok((_, remotes)) => { - fetch(&mut repo, &nid, Namespaces::Trusted(remotes.clone())) - .unwrap(); + radicle::test::fetch( + &repo, + &nid, + Namespaces::Trusted(remotes.clone()), + ) + .unwrap(); } Err(err) => panic!("Error fetching: {err}"), } @@ -673,56 +676,3 @@ impl Simulation { self.partitions.contains(&(a, b)) || self.partitions.contains(&(b, a)) } } - -/// Perform a fetch between two local repositories. -/// This has the same outcome as doing a "real" fetch, but suffices for the simulation, and -/// doesn't require running nodes. -fn fetch( - repo: &mut W, - node: &NodeId, - namespaces: impl Into, -) -> Result, radicle::storage::FetchError> { - let namespace = match namespaces.into() { - Namespaces::All => None, - Namespaces::Trusted(trusted) => trusted.into_iter().next(), - }; - let mut updates = Vec::new(); - let mut callbacks = git::RemoteCallbacks::new(); - let mut opts = git::FetchOptions::default(); - let refspec = if let Some(namespace) = namespace { - opts.prune(git::FetchPrune::On); - format!("refs/namespaces/{namespace}/refs/*:refs/namespaces/{namespace}/refs/*") - } else { - opts.prune(git::FetchPrune::Off); - "refs/namespaces/*:refs/namespaces/*".to_owned() - }; - - callbacks.update_tips(|name, old, new| { - if let Ok(name) = radicle::git::RefString::try_from(name) { - if name.to_namespaced().is_some() { - updates.push(RefUpdate::from(name, old, new)); - // Returning `true` ensures the process is not aborted. - return true; - } - } - false - }); - opts.remote_callbacks(callbacks); - - let mut remote = repo.raw().remote_anonymous( - radicle::storage::git::transport::remote::Url { - node: *node, - repo: repo.id(), - namespace, - } - .to_string() - .as_str(), - )?; - remote.fetch(&[refspec], Some(&mut opts), None)?; - drop(opts); - - repo.validate()?; - repo.set_head()?; - - Ok(updates) -} diff --git a/radicle-remote-helper/src/push.rs b/radicle-remote-helper/src/push.rs index 81b3e97c..fe028b98 100644 --- a/radicle-remote-helper/src/push.rs +++ b/radicle-remote-helper/src/push.rs @@ -452,12 +452,12 @@ fn patch_merge( let mut patches = patch::Patches::open(stored)?; for patch in patches.all()? { - let (id, patch, clock) = patch?; + let (id, patch) = patch?; let (revision_id, revision) = patch.latest(); if patch.is_open() && commits.contains(&revision.head()) { let revision_id = *revision_id; - let mut patch = patch::PatchMut::new(id, patch, clock, &mut patches); + let mut patch = patch::PatchMut::new(id, patch, &mut patches); patch.merge(revision_id, new, signer)?; diff --git a/radicle-tui/src/cob/issue.rs b/radicle-tui/src/cob/issue.rs index 04234aec..0e0f1e5c 100644 --- a/radicle-tui/src/cob/issue.rs +++ b/radicle-tui/src/cob/issue.rs @@ -9,7 +9,7 @@ pub fn all(repository: &Repository) -> Result> { Ok(patches .into_iter() - .map(|(id, issue, _)| (id, issue)) + .map(|(id, issue)| (id, issue)) .collect::>()) } diff --git a/radicle-tui/src/cob/patch.rs b/radicle-tui/src/cob/patch.rs index 01dce274..dc73955e 100644 --- a/radicle-tui/src/cob/patch.rs +++ b/radicle-tui/src/cob/patch.rs @@ -10,7 +10,7 @@ pub fn all(repository: &Repository) -> Result> { Ok(patches .into_iter() - .map(|(id, patch, _)| (id, patch)) + .map(|(id, patch)| (id, patch)) .collect::>()) } diff --git a/radicle/Cargo.toml b/radicle/Cargo.toml index 15d07b5d..089ed5e0 100644 --- a/radicle/Cargo.toml +++ b/radicle/Cargo.toml @@ -37,10 +37,6 @@ features = ["vendored-libgit2"] path = "../radicle-cob" version = "0" -[dependencies.radicle-crdt] -path = "../radicle-crdt" -version = "0" - [dependencies.radicle-crypto] path = "../radicle-crypto" version = "0" diff --git a/radicle/src/cob/common.rs b/radicle/src/cob/common.rs index 6bbf74ef..7cd66218 100644 --- a/radicle/src/cob/common.rs +++ b/radicle/src/cob/common.rs @@ -1,12 +1,13 @@ use std::fmt::{self, Display}; use std::str::FromStr; +use localtime::LocalTime; use serde::{Deserialize, Serialize}; use crate::prelude::*; -pub use radicle_crdt::clock; -pub use radicle_crdt::clock::Physical as Timestamp; +/// Timestamp used for COB operations. +pub type Timestamp = LocalTime; /// Author. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] diff --git a/radicle/src/cob/identity.rs b/radicle/src/cob/identity.rs index e4ab03fa..9aa9d1b8 100644 --- a/radicle/src/cob/identity.rs +++ b/radicle/src/cob/identity.rs @@ -1,9 +1,9 @@ +use std::collections::BTreeMap; use std::{ops::Deref, str::FromStr}; use crypto::{PublicKey, Signature}; use once_cell::sync::Lazy; use radicle_cob::{ObjectId, TypeName}; -use radicle_crdt::{clock, GMap, GSet, LWWMap, LWWReg, Max, Redactable, Semilattice}; use radicle_crypto::{Signer, Verified}; use radicle_git_ext::Oid; use serde::{Deserialize, Serialize}; @@ -12,8 +12,8 @@ use thiserror::Error; use crate::{ cob::{ self, - common::Timestamp, store::{self, FromHistory as _, HistoryAction, Transaction}, + Timestamp, }, identity::{doc::DocError, Did, Identity, IdentityError}, prelude::{Doc, ReadRepository}, @@ -25,9 +25,6 @@ use super::{ Author, EntryId, }; -/// The logical clock we use to order operations to proposals. -pub use clock::Lamport as Clock; - /// Type name of an identity proposal. pub static TYPENAME: Lazy = Lazy::new(|| FromStr::from_str("xyz.radicle.id.proposal").expect("type name is valid")); @@ -141,18 +138,18 @@ pub enum Error { /// Once a proposal has reached the quourum threshold for the previous /// [`Identity`] then it may be committed to the person's local /// storage using [`Proposal::commit`]. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Default)] pub struct Proposal { /// Title of the proposal. - title: LWWReg>, + title: String, /// Description of the proposal. - description: LWWReg>, + description: String, /// Current state of the proposal. - state: LWWReg>, + state: State, /// List of revisions for this proposal. - revisions: GMap>, + revisions: BTreeMap>, /// Timeline of events. - timeline: GSet<(clock::Lamport, EntryId)>, + timeline: Vec, } #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] @@ -164,25 +161,6 @@ pub enum State { Committed, } -impl Semilattice for Proposal { - fn merge(&mut self, other: Self) { - self.description.merge(other.description); - self.revisions.merge(other.revisions); - } -} - -impl Default for Proposal { - fn default() -> Self { - Self { - title: LWWReg::initial(Max::from(String::default())), - description: LWWReg::initial(Max::from(String::default())), - state: LWWReg::initial(Max::from(State::default())), - revisions: GMap::default(), - timeline: GSet::default(), - } - } -} - impl Proposal { /// Commit the [`Doc`], found at the given `revision`, to the /// provided `remote`. @@ -213,7 +191,7 @@ impl Proposal { let revision = self .revision(rid) .ok_or_else(|| CommitError::Missing(*rid))? - .get() + .as_ref() .ok_or_else(|| CommitError::Redacted(*rid))?; let doc = &revision.proposed; let previous = Identity::load(signer.public_key(), repo)?; @@ -264,29 +242,29 @@ impl Proposal { /// The most recent title for the proposal. pub fn title(&self) -> &str { - self.title.get().get() + &self.title } /// The most recent description for the proposal, if present. pub fn description(&self) -> Option<&str> { - Some(self.description.get().get()) + Some(self.description.as_str()) } pub fn state(&self) -> &State { - self.state.get().get() + &self.state } /// A specific [`Revision`], that may be redacted. - pub fn revision(&self, revision: &RevisionId) -> Option<&Redactable> { + pub fn revision(&self, revision: &RevisionId) -> Option<&Option> { self.revisions.get(revision) } /// All the [`Revision`]s that have not been redacted. pub fn revisions(&self) -> impl DoubleEndedIterator { - self.timeline.iter().filter_map(|(_, id)| { + self.timeline.iter().filter_map(|id| { self.revisions .get(id) - .and_then(Redactable::get) + .and_then(|o| o.as_ref()) .map(|rev| (id, rev)) }) } @@ -321,45 +299,41 @@ impl store::FromHistory for Proposal { Ok(()) } - fn apply( - &mut self, - ops: impl IntoIterator, - repo: &R, - ) -> Result<(), Self::Error> { - for op in ops { - let id = op.id; - let author = Author::new(op.author); - let timestamp = op.timestamp; + fn apply(&mut self, op: Op, repo: &R) -> Result<(), Self::Error> { + let id = op.id; + let author = Author::new(op.author); + let timestamp = op.timestamp; - self.timeline.insert((op.clock, id)); + debug_assert!(!self.timeline.contains(&op.id)); - match op.action { + self.timeline.push(id); + + for action in op.actions { + match action { Action::Accept { revision, signature, } => match self.revisions.get_mut(&revision) { - Some(Redactable::Present(revision)) => { - revision.accept(op.author, signature, op.clock) - } - Some(Redactable::Redacted) => return Err(ApplyError::Redacted(revision)), + Some(Some(revision)) => revision.accept(op.author, signature), + Some(None) => return Err(ApplyError::Redacted(revision)), None => return Err(ApplyError::Missing(revision)), }, - Action::Close => self.state.set(State::Closed, op.clock), + Action::Close => self.state = State::Closed, Action::Edit { title, description } => { - self.title.set(title, op.clock); - self.description.set(description, op.clock); + self.title = title; + self.description = description; } - Action::Commit => self.state.set(State::Committed, op.clock), + Action::Commit => self.state = State::Committed, Action::Redact { revision } => { if let Some(revision) = self.revisions.get_mut(&revision) { - revision.merge(Redactable::Redacted); + *revision = None; } else { return Err(ApplyError::Missing(revision)); } } Action::Reject { revision } => match self.revisions.get_mut(&revision) { - Some(Redactable::Present(revision)) => revision.reject(op.author, op.clock), - Some(Redactable::Redacted) => return Err(ApplyError::Redacted(revision)), + Some(Some(revision)) => revision.reject(op.author), + Some(None) => return Err(ApplyError::Redacted(revision)), None => return Err(ApplyError::Missing(revision)), }, Action::Revision { current, proposed } => { @@ -371,23 +345,16 @@ impl store::FromHistory for Proposal { } self.revisions.insert( id, - Redactable::Present(Revision::new(author, current, proposed, timestamp)), - ) + Some(Revision::new(author.clone(), current, proposed, timestamp)), + ); } Action::Thread { revision, action } => match self.revisions.get_mut(&revision) { - Some(Redactable::Present(revision)) => revision.discussion.apply( - [cob::Op::new( - op.id, - action, - op.author, - op.timestamp, - op.clock, - op.identity, - )], + Some(Some(revision)) => revision.discussion.apply( + cob::Op::new(op.id, action, op.author, op.timestamp, op.identity), repo, )?, - Some(Redactable::Redacted) => return Err(ApplyError::Redacted(revision)), + Some(None) => return Err(ApplyError::Redacted(revision)), None => return Err(ApplyError::Missing(revision)), }, } @@ -418,7 +385,7 @@ pub struct Revision { /// Discussion thread for this revision. pub discussion: Thread, /// [`Verdict`]s given by the delegates. - pub verdicts: LWWMap>, + pub verdicts: BTreeMap>, /// Physical timestamp of this proposal revision. pub timestamp: Timestamp, } @@ -435,7 +402,7 @@ impl Revision { current, proposed, discussion: Thread::default(), - verdicts: LWWMap::default(), + verdicts: BTreeMap::default(), timestamp, } } @@ -450,7 +417,7 @@ impl Revision { pub fn verdicts(&self) -> impl Iterator { self.verdicts .iter() - .filter_map(|(key, verdict)| verdict.get().map(|verdict| (key, verdict))) + .filter_map(|(key, verdict)| verdict.as_ref().map(|verdict| (key, verdict))) } pub fn accepted(&self) -> Vec { @@ -475,7 +442,7 @@ impl Revision { let votes_for = self .verdicts .iter() - .fold(0, |count, (_, verdict)| match verdict.get() { + .fold(0, |count, (_, verdict)| match verdict { Some(Verdict::Accept(_)) => count + 1, Some(Verdict::Reject) => count, None => count, @@ -483,14 +450,12 @@ impl Revision { votes_for >= previous.doc.threshold } - fn accept(&mut self, key: PublicKey, signature: Signature, clock: Clock) { - self.verdicts - .insert(key, Redactable::Present(Verdict::Accept(signature)), clock); + fn accept(&mut self, key: PublicKey, signature: Signature) { + self.verdicts.insert(key, Some(Verdict::Accept(signature))); } - fn reject(&mut self, key: PublicKey, clock: Clock) { - self.verdicts - .insert(key, Redactable::Present(Verdict::Reject), clock) + fn reject(&mut self, key: PublicKey) { + self.verdicts.insert(key, Some(Verdict::Reject)); } } @@ -565,7 +530,6 @@ pub struct ProposalMut<'a, 'g, R> { pub id: ObjectId, proposal: Proposal, - clock: clock::Lamport, store: &'g mut Proposals<'a, R>, } @@ -573,15 +537,9 @@ impl<'a, 'g, R> ProposalMut<'a, 'g, R> where R: WriteRepository + cob::Store, { - pub fn new( - id: ObjectId, - proposal: Proposal, - clock: clock::Lamport, - store: &'g mut Proposals<'a, R>, - ) -> Self { + pub fn new(id: ObjectId, proposal: Proposal, store: &'g mut Proposals<'a, R>) -> Self { Self { id, - clock, proposal, store, } @@ -597,21 +555,15 @@ where G: Signer, F: FnOnce(&mut Transaction) -> Result<(), store::Error>, { - let mut tx = Transaction::new(*signer.public_key(), self.clock); + let mut tx = Transaction::new(*signer.public_key()); operations(&mut tx)?; - let (ops, clock, commit) = tx.commit(message, self.id, &mut self.store.raw, signer)?; + let (ops, commit) = tx.commit(message, self.id, &mut self.store.raw, signer)?; self.proposal.apply(ops, self.store.as_ref())?; - self.clock = clock; Ok(commit) } - /// Get the internal logical clock. - pub fn clock(&self) -> &clock::Lamport { - &self.clock - } - /// Accept a proposal revision. pub fn accept( &mut self, @@ -715,22 +667,20 @@ where proposed: Doc, signer: &G, ) -> Result, Error> { - let (id, proposal, clock) = + let (id, proposal) = Transaction::initial("Create proposal", &mut self.raw, signer, |tx| { tx.revision(current.into(), proposed)?; tx.edit(title, description)?; Ok(()) })?; - // Just a sanity check that our clock is advancing as expected. - debug_assert_eq!(clock.get(), 1); - Ok(ProposalMut::new(id, proposal, clock, self)) + Ok(ProposalMut::new(id, proposal, self)) } /// Get a proposal. pub fn get(&self, id: &ObjectId) -> Result, store::Error> { - self.raw.get(id).map(|r| r.map(|(p, _)| p)) + self.raw.get(id) } /// Get a proposal mutably. @@ -738,14 +688,13 @@ where &'g mut self, id: &ObjectId, ) -> Result, store::Error> { - let (proposal, clock) = self + let proposal = self .raw .get(id)? .ok_or_else(move || store::Error::NotFound(TYPENAME.clone(), *id))?; Ok(ProposalMut { id: *id, - clock, proposal, store: self, }) diff --git a/radicle/src/cob/issue.rs b/radicle/src/cob/issue.rs index 0254cc6c..295c7e79 100644 --- a/radicle/src/cob/issue.rs +++ b/radicle/src/cob/issue.rs @@ -1,3 +1,4 @@ +use std::collections::BTreeSet; use std::ops::Deref; use std::str::FromStr; @@ -5,9 +6,6 @@ use once_cell::sync::Lazy; use serde::{Deserialize, Serialize}; use thiserror::Error; -use radicle_crdt::clock; -use radicle_crdt::{LWWReg, LWWSet, Max, Semilattice}; - use crate::cob; use crate::cob::common::{Author, Reaction, Tag, Timestamp}; use crate::cob::store::Transaction; @@ -92,42 +90,20 @@ impl State { } /// Issue state. Accumulates [`Action`]. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Default)] pub struct Issue { /// Actors assigned to this issue. - assignees: LWWSet, + assignees: BTreeSet, /// Title of the issue. - title: LWWReg>, + title: String, /// Current state of the issue. - state: LWWReg>, + state: State, /// Associated tags. - tags: LWWSet, + tags: BTreeSet, /// Discussion around this issue. thread: Thread, } -impl Semilattice for Issue { - fn merge(&mut self, other: Self) { - self.assignees.merge(other.assignees); - self.title.merge(other.title); - self.state.merge(other.state); - self.tags.merge(other.tags); - self.thread.merge(other.thread); - } -} - -impl Default for Issue { - fn default() -> Self { - Self { - assignees: LWWSet::default(), - title: LWWReg::initial(Max::from(String::default())), - state: LWWReg::initial(Max::from(State::default())), - tags: LWWSet::default(), - thread: Thread::default(), - } - } -} - impl store::FromHistory for Issue { type Action = Action; type Error = Error; @@ -137,7 +113,7 @@ impl store::FromHistory for Issue { } fn validate(&self) -> Result<(), Self::Error> { - if self.title.get().is_empty() { + if self.title.is_empty() { return Err(Error::Validate("title is empty")); } if self.thread.validate().is_err() { @@ -146,45 +122,34 @@ impl store::FromHistory for Issue { Ok(()) } - fn apply( - &mut self, - ops: impl IntoIterator, - repo: &R, - ) -> Result<(), Error> { - for op in ops { - match op.action { + fn apply(&mut self, op: Op, repo: &R) -> Result<(), Error> { + for action in op.actions { + match action { Action::Assign { add, remove } => { for assignee in add { - self.assignees.insert(assignee, op.clock); + self.assignees.insert(assignee); } for assignee in remove { - self.assignees.remove(assignee, op.clock); + self.assignees.remove(&assignee); } } Action::Edit { title } => { - self.title.set(title, op.clock); + self.title = title; } Action::Lifecycle { state } => { - self.state.set(state, op.clock); + self.state = state; } Action::Tag { add, remove } => { for tag in add { - self.tags.insert(tag, op.clock); + self.tags.insert(tag); } for tag in remove { - self.tags.remove(tag, op.clock); + self.tags.remove(&tag); } } Action::Thread { action } => { self.thread.apply( - [cob::Op::new( - op.id, - action, - op.author, - op.timestamp, - op.clock, - op.identity, - )], + cob::Op::new(op.id, action, op.author, op.timestamp, op.identity), repo, )?; } @@ -200,11 +165,11 @@ impl Issue { } pub fn title(&self) -> &str { - self.title.get().as_str() + self.title.as_str() } pub fn state(&self) -> &State { - self.state.get() + &self.state } pub fn tags(&self) -> impl Iterator { @@ -333,7 +298,6 @@ impl store::Transaction { pub struct IssueMut<'a, 'g, R> { id: ObjectId, - clock: clock::Lamport, issue: Issue, store: &'g mut Issues<'a, R>, } @@ -342,16 +306,21 @@ impl<'a, 'g, R> IssueMut<'a, 'g, R> where R: WriteRepository + cob::Store, { + /// Reload the issue data from storage. + pub fn reload(&mut self) -> Result<(), store::Error> { + self.issue = self + .store + .get(&self.id)? + .ok_or_else(|| store::Error::NotFound(TYPENAME.clone(), self.id))?; + + Ok(()) + } + /// Get the issue id. pub fn id(&self) -> &ObjectId { &self.id } - /// Get the internal logical clock. - pub fn clock(&self) -> &clock::Lamport { - &self.clock - } - /// Assign one or more actors to an issue. pub fn assign( &mut self, @@ -449,12 +418,11 @@ where G: Signer, F: FnOnce(&mut Transaction) -> Result<(), store::Error>, { - let mut tx = Transaction::new(*signer.public_key(), self.clock); + let mut tx = Transaction::new(*signer.public_key()); operations(&mut tx)?; - let (ops, clock, commit) = tx.commit(message, self.id, &mut self.store.raw, signer)?; + let (ops, commit) = tx.commit(message, self.id, &mut self.store.raw, signer)?; self.issue.apply(ops, self.store.as_ref())?; - self.clock = clock; Ok(commit) } @@ -501,19 +469,18 @@ where /// Get an issue. pub fn get(&self, id: &ObjectId) -> Result, store::Error> { - self.raw.get(id).map(|r| r.map(|(i, _clock)| i)) + self.raw.get(id) } /// Get an issue mutably. pub fn get_mut<'g>(&'g mut self, id: &ObjectId) -> Result, store::Error> { - let (issue, clock) = self + let issue = self .raw .get(id)? .ok_or_else(move || store::Error::NotFound(TYPENAME.clone(), *id))?; Ok(IssueMut { id: *id, - clock, issue, store: self, }) @@ -528,21 +495,17 @@ where assignees: &[ActorId], signer: &G, ) -> Result, Error> { - let (id, issue, clock) = - Transaction::initial("Create issue", &mut self.raw, signer, |tx| { - tx.thread(description)?; - tx.assign(assignees.to_owned(), [])?; - tx.edit(title)?; - tx.tag(tags.to_owned(), [])?; + let (id, issue) = Transaction::initial("Create issue", &mut self.raw, signer, |tx| { + tx.thread(description)?; + tx.assign(assignees.to_owned(), [])?; + tx.edit(title)?; + tx.tag(tags.to_owned(), [])?; - Ok(()) - })?; - // Just a sanity check that our clock is advancing as expected. - debug_assert_eq!(clock.get(), 1); + Ok(()) + })?; Ok(IssueMut { id, - clock, issue, store: self, }) @@ -553,7 +516,7 @@ where let all = self.all()?; let state_groups = all.filter_map(|s| s.ok()) - .fold(IssueCounts::default(), |mut state, (_, p, _)| { + .fold(IssueCounts::default(), |mut state, (_, p)| { match p.state() { State::Open => state.open += 1, State::Closed { .. } => state.closed += 1, @@ -610,6 +573,77 @@ mod test { use crate::test; use crate::test::arbitrary; + #[test] + fn test_concurrency() { + let t = test::setup::Network::default(); + let mut issues_alice = Issues::open(&*t.alice.repo).unwrap(); + let mut bob_issues = Issues::open(&*t.bob.repo).unwrap(); + let mut eve_issues = Issues::open(&*t.eve.repo).unwrap(); + + let mut issue_alice = issues_alice + .create("Alice Issue", "Alice's comment", &[], &[], &t.alice.signer) + .unwrap(); + let id = *issue_alice.id(); + + t.bob.repo.fetch(&t.alice); + t.eve.repo.fetch(&t.alice); + + let mut issue_eve = eve_issues.get_mut(&id).unwrap(); + let mut issue_bob = bob_issues.get_mut(&id).unwrap(); + + issue_bob + .comment("Bob's reply", id.into(), &t.bob.signer) + .unwrap(); + issue_alice + .comment("Alice's reply", id.into(), &t.alice.signer) + .unwrap(); + + assert_eq!(issue_bob.comments().count(), 2); + assert_eq!(issue_alice.comments().count(), 2); + + t.bob.repo.fetch(&t.alice); + issue_bob.reload().unwrap(); + assert_eq!(issue_bob.comments().count(), 3); + + t.alice.repo.fetch(&t.bob); + issue_alice.reload().unwrap(); + assert_eq!(issue_alice.comments().count(), 3); + + let bob_comments = issue_bob + .comments() + .map(|(_, c)| c.body()) + .collect::>(); + let alice_comments = issue_alice + .comments() + .map(|(_, c)| c.body()) + .collect::>(); + + assert_eq!(bob_comments, alice_comments); + + t.eve.repo.fetch(&t.alice); + + let eve_reply = issue_eve + .comment("Eve's reply", id.into(), &t.eve.signer) + .unwrap(); + + t.bob.repo.fetch(&t.eve); + t.alice.repo.fetch(&t.eve); + + issue_alice.reload().unwrap(); + issue_bob.reload().unwrap(); + issue_eve.reload().unwrap(); + + assert_eq!(issue_eve.comments().count(), 4); + assert_eq!(issue_bob.comments().count(), 4); + assert_eq!(issue_alice.comments().count(), 4); + + let (first, _) = issue_bob.comments().next().unwrap(); + let (last, _) = issue_bob.comments().last().unwrap(); + + assert_eq!(*first, issue_alice.id.into()); + assert_eq!(*last, eve_reply); + } + #[test] fn test_ordering() { assert!(CloseReason::Solved > CloseReason::Other); @@ -623,11 +657,8 @@ mod test { #[test] fn test_issue_create_and_assign() { - let tmp = tempfile::tempdir().unwrap(); - let test::setup::Context { - signer, project, .. - } = test::setup::Context::new(&tmp); - let mut issues = Issues::open(&project).unwrap(); + let test::setup::NodeWithRepo { node, repo, .. } = test::setup::NodeWithRepo::default(); + let mut issues = Issues::open(&*repo).unwrap(); let assignee: ActorId = arbitrary::gen(1); let assignee_two: ActorId = arbitrary::gen(1); @@ -637,7 +668,7 @@ mod test { "Blah blah blah.", &[], &[assignee], - &signer, + &node.signer, ) .unwrap(); @@ -649,7 +680,7 @@ mod test { assert!(assignees.contains(&Did::from(assignee))); let mut issue = issues.get_mut(&id).unwrap(); - issue.assign([assignee_two], &signer).unwrap(); + issue.assign([assignee_two], &node.signer).unwrap(); let id = issue.id; let issue = issues.get(&id).unwrap().unwrap(); @@ -662,11 +693,8 @@ mod test { #[test] fn test_issue_create_and_reassign() { - let tmp = tempfile::tempdir().unwrap(); - let test::setup::Context { - signer, project, .. - } = test::setup::Context::new(&tmp); - let mut issues = Issues::open(&project).unwrap(); + let test::setup::NodeWithRepo { node, repo, .. } = test::setup::NodeWithRepo::default(); + let mut issues = Issues::open(&*repo).unwrap(); let assignee: ActorId = arbitrary::gen(1); let assignee_two: ActorId = arbitrary::gen(1); @@ -676,12 +704,12 @@ mod test { "Blah blah blah.", &[], &[assignee], - &signer, + &node.signer, ) .unwrap(); - issue.assign([assignee_two], &signer).unwrap(); - issue.assign([assignee_two], &signer).unwrap(); + issue.assign([assignee_two], &node.signer).unwrap(); + issue.assign([assignee_two], &node.signer).unwrap(); let id = issue.id; let issue = issues.get(&id).unwrap().unwrap(); @@ -694,23 +722,18 @@ mod test { #[test] fn test_issue_create_and_get() { - let tmp = tempfile::tempdir().unwrap(); - let test::setup::Context { - signer, project, .. - } = test::setup::Context::new(&tmp); - let mut issues = Issues::open(&project).unwrap(); + let test::setup::NodeWithRepo { node, repo, .. } = test::setup::NodeWithRepo::default(); + let mut issues = Issues::open(&*repo).unwrap(); let created = issues - .create("My first issue", "Blah blah blah.", &[], &[], &signer) + .create("My first issue", "Blah blah blah.", &[], &[], &node.signer) .unwrap(); - assert_eq!(created.clock().get(), 1); - let (id, created) = (created.id, created.issue); let issue = issues.get(&id).unwrap().unwrap(); assert_eq!(created, issue); assert_eq!(issue.title(), "My first issue"); - assert_eq!(issue.author().id, Did::from(signer.public_key())); + assert_eq!(issue.author().id, Did::from(node.signer.public_key())); assert_eq!(issue.description().1, "Blah blah blah."); assert_eq!(issue.comments().count(), 1); assert_eq!(issue.state(), &State::Open); @@ -718,13 +741,10 @@ mod test { #[test] fn test_issue_create_and_change_state() { - let tmp = tempfile::tempdir().unwrap(); - let test::setup::Context { - signer, project, .. - } = test::setup::Context::new(&tmp); - let mut issues = Issues::open(&project).unwrap(); + let test::setup::NodeWithRepo { node, repo, .. } = test::setup::NodeWithRepo::default(); + let mut issues = Issues::open(&*repo).unwrap(); let mut issue = issues - .create("My first issue", "Blah blah blah.", &[], &[], &signer) + .create("My first issue", "Blah blah blah.", &[], &[], &node.signer) .unwrap(); issue @@ -732,7 +752,7 @@ mod test { State::Closed { reason: CloseReason::Other, }, - &signer, + &node.signer, ) .unwrap(); @@ -746,7 +766,7 @@ mod test { } ); - issue.lifecycle(State::Open, &signer).unwrap(); + issue.lifecycle(State::Open, &node.signer).unwrap(); let issue = issues.get(&id).unwrap().unwrap(); assert_eq!(*issue.state(), State::Open); @@ -754,11 +774,8 @@ mod test { #[test] fn test_issue_create_and_unassign() { - let tmp = tempfile::tempdir().unwrap(); - let test::setup::Context { - signer, project, .. - } = test::setup::Context::new(&tmp); - let mut issues = Issues::open(&project).unwrap(); + let test::setup::NodeWithRepo { node, repo, .. } = test::setup::NodeWithRepo::default(); + let mut issues = Issues::open(&*repo).unwrap(); let assignee: ActorId = arbitrary::gen(1); let assignee_two: ActorId = arbitrary::gen(1); @@ -768,11 +785,11 @@ mod test { "Blah blah blah.", &[], &[assignee, assignee_two], - &signer, + &node.signer, ) .unwrap(); - issue.unassign([assignee], &signer).unwrap(); + issue.unassign([assignee], &node.signer).unwrap(); let id = issue.id; let issue = issues.get(&id).unwrap().unwrap(); @@ -784,16 +801,13 @@ mod test { #[test] fn test_issue_edit() { - let tmp = tempfile::tempdir().unwrap(); - let test::setup::Context { - signer, project, .. - } = test::setup::Context::new(&tmp); - let mut issues = Issues::open(&project).unwrap(); + let test::setup::NodeWithRepo { node, repo, .. } = test::setup::NodeWithRepo::default(); + let mut issues = Issues::open(&*repo).unwrap(); let mut issue = issues - .create("My first issue", "Blah blah blah.", &[], &[], &signer) + .create("My first issue", "Blah blah blah.", &[], &[], &node.signer) .unwrap(); - issue.edit("Sorry typo", &signer).unwrap(); + issue.edit("Sorry typo", &node.signer).unwrap(); let id = issue.id; let issue = issues.get(&id).unwrap().unwrap(); @@ -804,17 +818,14 @@ mod test { #[test] fn test_issue_edit_description() { - let tmp = tempfile::tempdir().unwrap(); - let test::setup::Context { - signer, project, .. - } = test::setup::Context::new(&tmp); - let mut issues = Issues::open(&project).unwrap(); + let test::setup::NodeWithRepo { node, repo, .. } = test::setup::NodeWithRepo::default(); + let mut issues = Issues::open(&*repo).unwrap(); let mut issue = issues - .create("My first issue", "Blah blah blah.", &[], &[], &signer) + .create("My first issue", "Blah blah blah.", &[], &[], &node.signer) .unwrap(); issue - .edit_description("Bob Loblaw law blog", &signer) + .edit_description("Bob Loblaw law blog", &node.signer) .unwrap(); let id = issue.id; @@ -826,19 +837,16 @@ mod test { #[test] fn test_issue_react() { - let tmp = tempfile::tempdir().unwrap(); - let test::setup::Context { - signer, project, .. - } = test::setup::Context::new(&tmp); - let mut issues = Issues::open(&project).unwrap(); + let test::setup::NodeWithRepo { node, repo, .. } = test::setup::NodeWithRepo::default(); + let mut issues = Issues::open(&*repo).unwrap(); let mut issue = issues - .create("My first issue", "Blah blah blah.", &[], &[], &signer) + .create("My first issue", "Blah blah blah.", &[], &[], &node.signer) .unwrap(); let (comment, _) = issue.root(); let comment = *comment; let reaction = Reaction::new('🥳').unwrap(); - issue.react(comment, reaction, &signer).unwrap(); + issue.react(comment, reaction, &node.signer).unwrap(); let id = issue.id; let issue = issues.get(&id).unwrap().unwrap(); @@ -851,19 +859,16 @@ mod test { #[test] fn test_issue_reply() { - let tmp = tempfile::tempdir().unwrap(); - let test::setup::Context { - signer, project, .. - } = test::setup::Context::new(&tmp); - let mut issues = Issues::open(&project).unwrap(); + let test::setup::NodeWithRepo { node, repo, .. } = test::setup::NodeWithRepo::default(); + let mut issues = Issues::open(&*repo).unwrap(); let mut issue = issues - .create("My first issue", "Blah blah blah.", &[], &[], &signer) + .create("My first issue", "Blah blah blah.", &[], &[], &node.signer) .unwrap(); let (root, _) = issue.root(); let root = *root; - let c1 = issue.comment("Hi hi hi.", root, &signer).unwrap(); - let c2 = issue.comment("Ha ha ha.", root, &signer).unwrap(); + let c1 = issue.comment("Hi hi hi.", root, &node.signer).unwrap(); + let c2 = issue.comment("Ha ha ha.", root, &node.signer).unwrap(); let id = issue.id; let mut issue = issues.get_mut(&id).unwrap(); @@ -873,10 +878,10 @@ mod test { assert_eq!(reply1.body(), "Hi hi hi."); assert_eq!(reply2.body(), "Ha ha ha."); - issue.comment("Re: Hi.", c1, &signer).unwrap(); - issue.comment("Re: Ha.", c2, &signer).unwrap(); - issue.comment("Re: Ha. Ha.", c2, &signer).unwrap(); - issue.comment("Re: Ha. Ha. Ha.", c2, &signer).unwrap(); + issue.comment("Re: Hi.", c1, &node.signer).unwrap(); + issue.comment("Re: Ha.", c2, &node.signer).unwrap(); + issue.comment("Re: Ha. Ha.", c2, &node.signer).unwrap(); + issue.comment("Re: Ha. Ha. Ha.", c2, &node.signer).unwrap(); let issue = issues.get(&id).unwrap().unwrap(); @@ -891,11 +896,8 @@ mod test { #[test] fn test_issue_tag() { - let tmp = tempfile::tempdir().unwrap(); - let test::setup::Context { - signer, project, .. - } = test::setup::Context::new(&tmp); - let mut issues = Issues::open(&project).unwrap(); + let test::setup::NodeWithRepo { node, repo, .. } = test::setup::NodeWithRepo::default(); + let mut issues = Issues::open(&*repo).unwrap(); let bug_tag = Tag::new("bug").unwrap(); let ux_tag = Tag::new("ux").unwrap(); let wontfix_tag = Tag::new("wontfix").unwrap(); @@ -905,12 +907,12 @@ mod test { "Blah blah blah.", &[ux_tag.clone()], &[], - &signer, + &node.signer, ) .unwrap(); - issue.tag([bug_tag.clone()], [], &signer).unwrap(); - issue.tag([wontfix_tag.clone()], [], &signer).unwrap(); + issue.tag([bug_tag.clone()], [], &node.signer).unwrap(); + issue.tag([wontfix_tag.clone()], [], &node.signer).unwrap(); let id = issue.id; let issue = issues.get(&id).unwrap().unwrap(); @@ -923,26 +925,19 @@ mod test { #[test] fn test_issue_comment() { - let tmp = tempfile::tempdir().unwrap(); - let test::setup::Context { - signer, project, .. - } = test::setup::Context::new(&tmp); - let author = *signer.public_key(); - let mut issues = Issues::open(&project).unwrap(); + let test::setup::NodeWithRepo { node, repo, .. } = test::setup::NodeWithRepo::default(); + let author = *node.signer.public_key(); + let mut issues = Issues::open(&*repo).unwrap(); let mut issue = issues - .create("My first issue", "Blah blah blah.", &[], &[], &signer) + .create("My first issue", "Blah blah blah.", &[], &[], &node.signer) .unwrap(); - assert_eq!(issue.clock.get(), 1); - // The root thread op id is always the same. let (c0, _) = issue.root(); let c0 = *c0; - issue.comment("Ho ho ho.", c0, &signer).unwrap(); - issue.comment("Ha ha ha.", c0, &signer).unwrap(); - - assert_eq!(issue.clock.get(), 3); + issue.comment("Ho ho ho.", c0, &node.signer).unwrap(); + issue.comment("Ha ha ha.", c0, &node.signer).unwrap(); let id = issue.id; let issue = issues.get(&id).unwrap().unwrap(); @@ -976,20 +971,23 @@ mod test { #[test] fn test_issue_all() { - let tmp = tempfile::tempdir().unwrap(); - let test::setup::Context { - signer, project, .. - } = test::setup::Context::new(&tmp); - let mut issues = Issues::open(&project).unwrap(); + let test::setup::NodeWithRepo { node, repo, .. } = test::setup::NodeWithRepo::default(); + let mut issues = Issues::open(&*repo).unwrap(); - issues.create("First", "Blah", &[], &[], &signer).unwrap(); - issues.create("Second", "Blah", &[], &[], &signer).unwrap(); - issues.create("Third", "Blah", &[], &[], &signer).unwrap(); + issues + .create("First", "Blah", &[], &[], &node.signer) + .unwrap(); + issues + .create("Second", "Blah", &[], &[], &node.signer) + .unwrap(); + issues + .create("Third", "Blah", &[], &[], &node.signer) + .unwrap(); let issues = issues .all() .unwrap() - .map(|r| r.map(|(_, i, _)| i)) + .map(|r| r.map(|(_, i)| i)) .collect::, _>>() .unwrap(); @@ -1002,18 +1000,15 @@ mod test { #[test] fn test_issue_multilines() { - let tmp = tempfile::tempdir().unwrap(); - let test::setup::Context { - signer, project, .. - } = test::setup::Context::new(&tmp); - let mut issues = Issues::open(&project).unwrap(); + let test::setup::NodeWithRepo { node, repo, .. } = test::setup::NodeWithRepo::default(); + let mut issues = Issues::open(&*repo).unwrap(); let created = issues .create( "My first issue", "Blah blah blah.\nYah yah yah", &[], &[], - &signer, + &node.signer, ) .unwrap(); @@ -1022,7 +1017,7 @@ mod test { assert_eq!(created, issue); assert_eq!(issue.title(), "My first issue"); - assert_eq!(issue.author().id, Did::from(signer.public_key())); + assert_eq!(issue.author().id, Did::from(node.signer.public_key())); assert_eq!(issue.description().1, "Blah blah blah.\nYah yah yah"); assert_eq!(issue.comments().count(), 1); assert_eq!(issue.state(), &State::Open); diff --git a/radicle/src/cob/op.rs b/radicle/src/cob/op.rs index a17be802..f4747cf7 100644 --- a/radicle/src/cob/op.rs +++ b/radicle/src/cob/op.rs @@ -2,10 +2,9 @@ use nonempty::NonEmpty; use thiserror::Error; use radicle_cob::history::{Entry, EntryId}; -use radicle_crdt::clock; -use radicle_crdt::clock::Lamport; use radicle_crypto::PublicKey; +use crate::cob::Timestamp; use crate::git; /// The author of an [`Op`]. @@ -29,13 +28,11 @@ pub struct Op { /// Id of the entry under which this operation lives. pub id: EntryId, /// The action carried out by this operation. - pub action: A, + pub actions: NonEmpty, /// The author of the operation. pub author: ActorId, - /// Lamport clock. - pub clock: Lamport, /// Timestamp of this operation. - pub timestamp: clock::Physical, + pub timestamp: Timestamp, /// Head of identity document committed to by this operation. pub identity: git::Oid, } @@ -55,17 +52,15 @@ impl Ord for Op { impl Op { pub fn new( id: EntryId, - action: A, + actions: impl Into>, author: ActorId, - timestamp: impl Into, - clock: Lamport, + timestamp: impl Into, identity: git::Oid, ) -> Self { Self { id, - action, + actions: actions.into(), author, - clock, timestamp: timestamp.into(), identity, } @@ -76,9 +71,7 @@ impl Op { } } -pub struct Ops(pub NonEmpty>); - -impl<'a, A> TryFrom<&'a Entry> for Ops +impl<'a, A> TryFrom<&'a Entry> for Op where for<'de> A: serde::Deserialize<'de>, { @@ -87,34 +80,32 @@ where fn try_from(entry: &'a Entry) -> Result { let id = *entry.id(); let identity = entry.resource(); - let ops = entry + let actions: Vec<_> = entry .contents() .iter() - .map(|blob| { - let action = serde_json::from_slice(blob.as_slice())?; - let op = Op { - id, - action, - author: *entry.actor(), - clock: entry.clock().into(), - timestamp: entry.timestamp().into(), - identity, - }; - Ok::<_, Self::Error>(op) - }) + .map(|blob| serde_json::from_slice(blob.as_slice())) .collect::>()?; // SAFETY: Entry is guaranteed to have at least one operation. #[allow(clippy::unwrap_used)] - Ok(Self(NonEmpty::from_vec(ops).unwrap())) + let actions = NonEmpty::from_vec(actions).unwrap(); + let op = Op { + id, + actions, + author: *entry.actor(), + timestamp: Timestamp::from_secs(entry.timestamp()), + identity, + }; + + Ok(op) } } -impl IntoIterator for Ops { - type Item = Op; - type IntoIter = > as IntoIterator>::IntoIter; +impl IntoIterator for Op { + type Item = A; + type IntoIter = as IntoIterator>::IntoIter; fn into_iter(self) -> Self::IntoIter { - self.0.into_iter() + self.actions.into_iter() } } diff --git a/radicle/src/cob/patch.rs b/radicle/src/cob/patch.rs index 4523a238..05d5fa31 100644 --- a/radicle/src/cob/patch.rs +++ b/radicle/src/cob/patch.rs @@ -1,5 +1,5 @@ #![allow(clippy::too_many_arguments)] -use std::collections::HashMap; +use std::collections::{BTreeMap, BTreeSet, HashMap}; use std::fmt; use std::ops::Deref; use std::ops::Range; @@ -11,11 +11,6 @@ use serde::ser::SerializeStruct; use serde::{Deserialize, Serialize}; use thiserror::Error; -use radicle_crdt::clock; -use radicle_crdt::{ - GMap, GSet, Immutable, LWWMap, LWWReg, LWWSet, Lamport, Max, Redactable, Semilattice, -}; - use crate::cob; use crate::cob::common::{Author, Tag, Timestamp}; use crate::cob::store::Transaction; @@ -31,9 +26,6 @@ use crate::identity::doc::DocError; use crate::identity::PayloadError; use crate::prelude::*; -/// The logical clock we use to order operations to patches. -pub use clock::Lamport as Clock; - /// Type name of a patch. pub static TYPENAME: Lazy = Lazy::new(|| FromStr::from_str("xyz.radicle.patch").expect("type name is valid")); @@ -182,18 +174,18 @@ impl MergeTarget { } } -/// Patch CRDT. -#[derive(Debug, Clone, PartialEq, Eq)] +/// Patch state. +#[derive(Debug, Clone, PartialEq, Eq, Default)] pub struct Patch { /// Title of the patch. - title: LWWReg>, + title: String, /// Current state of the patch. - state: LWWReg>, + state: State, /// Target this patch is meant to be merged in. - target: LWWReg>, + target: MergeTarget, /// Associated tags. /// Tags can be added and removed at will. - tags: LWWSet, + tags: BTreeSet, /// Patch merges. /// /// Only one merge is allowed per user. @@ -201,64 +193,34 @@ pub struct Patch { /// Merges can be removed and replaced, but not modified. Generally, once a revision is merged, /// it stays that way. Being able to remove merges may be useful in case of force updates /// on the target branch. - merges: LWWMap>, + merges: BTreeMap, /// List of patch revisions. The initial changeset is part of the /// first revision. /// /// Revisions can be redacted, but are otherwise immutable. - revisions: GMap>, + revisions: BTreeMap>, /// Users assigned to review this patch. - reviewers: LWWSet, + reviewers: BTreeSet, /// Timeline of operations. - timeline: GSet<(Lamport, EntryId)>, + timeline: Vec, /// Reviews index. Keeps track of reviews for better performance. - reviews: GMap>, -} - -impl Semilattice for Patch { - fn merge(&mut self, other: Self) { - self.title.merge(other.title); - self.state.merge(other.state); - self.target.merge(other.target); - self.merges.merge(other.merges); - self.tags.merge(other.tags); - self.revisions.merge(other.revisions); - self.reviewers.merge(other.reviewers); - self.timeline.merge(other.timeline); - self.reviews.merge(other.reviews); - } -} - -impl Default for Patch { - fn default() -> Self { - Self { - title: LWWReg::initial(Max::from(String::default())), - state: LWWReg::initial(Max::from(State::default())), - target: LWWReg::initial(Max::from(MergeTarget::default())), - tags: LWWSet::default(), - merges: LWWMap::default(), - revisions: GMap::default(), - reviewers: LWWSet::default(), - timeline: GSet::default(), - reviews: GMap::default(), - } - } + reviews: BTreeMap>, } impl Patch { /// Title of the patch. pub fn title(&self) -> &str { - self.title.get().get() + self.title.as_str() } /// Current state of the patch. pub fn state(&self) -> &State { - self.state.get().get() + &self.state } /// Target this patch is meant to be merged in. pub fn target(&self) -> MergeTarget { - *self.target.get().get() + self.target } /// Timestamp of the first revision of the patch. @@ -294,16 +256,16 @@ impl Patch { /// /// None is returned if the `Revision` has been redacted (deleted). pub fn revision(&self, id: &RevisionId) -> Option<&Revision> { - self.revisions.get(id).and_then(Redactable::get) + self.revisions.get(id).and_then(|o| o.as_ref()) } /// List of patch revisions. The initial changeset is part of the /// first revision. pub fn revisions(&self) -> impl DoubleEndedIterator { - self.timeline.iter().filter_map(|(_, id)| { + self.timeline.iter().filter_map(|id| { self.revisions .get(id) - .and_then(Redactable::get) + .and_then(|o| o.as_ref()) .map(|rev| (id, rev)) }) } @@ -315,7 +277,7 @@ impl Patch { /// Get the merges. pub fn merges(&self) -> impl Iterator { - self.merges.iter().map(|(a, m)| (a, m.deref())) + self.merges.iter().map(|(a, m)| (a, m)) } /// Reference to the Git object containing the code on the latest revision. @@ -415,32 +377,30 @@ impl store::FromHistory for Patch { Ok(()) } - fn apply( - &mut self, - ops: impl IntoIterator, - repo: &R, - ) -> Result<(), Error> { - for op in ops { - let id = op.id; - let author = Author::new(op.author); - let timestamp = op.timestamp; + fn apply(&mut self, op: Op, repo: &R) -> Result<(), Error> { + let id = op.id; + let author = Author::new(op.author); + let timestamp = op.timestamp; - self.timeline.insert((op.clock, id)); + debug_assert!(!self.timeline.contains(&op.id)); - match op.action { + self.timeline.push(op.id); + + for action in op.actions { + match action { Action::Edit { title, target } => { - self.title.set(title, op.clock); - self.target.set(target, op.clock); + self.title = title; + self.target = target; } Action::Lifecycle { state } => { - self.state.set(state, op.clock); + self.state = state; } Action::Tag { add, remove } => { for tag in add { - self.tags.insert(tag, op.clock); + self.tags.insert(tag); } for tag in remove { - self.tags.remove(tag, op.clock); + self.tags.remove(&tag); } } Action::EditRevision { @@ -449,15 +409,15 @@ impl store::FromHistory for Patch { } => { if let Some(redactable) = self.revisions.get_mut(&revision) { // If the revision was redacted concurrently, there's nothing to do. - if let Redactable::Present(revision) = redactable { - revision.description.set(description, op.clock); + if let Some(revision) = redactable { + revision.description = description; } } else { return Err(Error::Missing(revision)); } } Action::EditReview { review, summary } => { - let Some(Redactable::Present((revision, author))) = + let Some(Some((revision, author))) = self.reviews.get(&review) else { return Err(Error::Missing(review)); }; @@ -466,12 +426,12 @@ impl store::FromHistory for Patch { }; // If the revision was redacted concurrently, there's nothing to do. // Likewise, if the review was redacted concurrently, there's nothing to do. - if let Some(rev) = rev.get_mut() { + if let Some(rev) = rev { let Some(review) = rev.reviews.get_mut(author) else { return Err(Error::Missing(review)); }; - if let Redactable::Present(review) = review { - review.summary.set(summary.map(Max::from), op.clock); + if let Some(review) = review { + review.summary = summary; } } } @@ -482,20 +442,19 @@ impl store::FromHistory for Patch { } => { self.revisions.insert( id, - Redactable::Present(Revision::new( - author, + Some(Revision::new( + author.clone(), description, base, oid, timestamp, - op.clock, )), ); } Action::Redact { revision } => { // Redactions must have observed a revision to be valid. if let Some(revision) = self.revisions.get_mut(&revision) { - revision.merge(Redactable::Redacted); + *revision = None; } else { return Err(Error::Missing(revision)); } @@ -508,21 +467,15 @@ impl store::FromHistory for Patch { let Some(rev) = self.revisions.get_mut(&revision) else { return Err(Error::Missing(revision)); }; - if let Some(rev) = rev.get_mut() { + if let Some(rev) = rev { // Nb. Applying two reviews by the same author is not allowed and // results in the review being redacted. rev.reviews.insert( op.author, - Redactable::Present(Review::new( - verdict, - summary.to_owned(), - timestamp, - op.clock, - )), + Some(Review::new(verdict, summary.to_owned(), timestamp)), ); // Update reviews index. - self.reviews - .insert(op.id, Redactable::Present((revision, op.author))); + self.reviews.insert(op.id, Some((revision, op.author))); } } Action::EditCodeComment { @@ -530,27 +483,31 @@ impl store::FromHistory for Patch { comment, body, } => { - let Some(Redactable::Present((revision, author))) = - self.reviews.get(&review) else { - return Err(Error::Missing(review)); - }; - let Some(rev) = self.revisions.get_mut(revision) else { - return Err(Error::Missing(*revision)); - }; - // If the revision was redacted concurrently, there's nothing to do. - // Likewise, if the review was redacted concurrently, there's nothing to do. - if let Some(rev) = rev.get_mut() { - let Some(review) = rev.reviews.get_mut(author) else { - return Err(Error::Missing(review)); - }; - if let Some(review) = review.get_mut() { - let Some(comment) = review.comments.get_mut(&comment) else { - return Err(Error::Missing(comment)); + match self.reviews.get(&review) { + Some(Some((revision, author))) => { + let Some(rev) = self.revisions.get_mut(revision) else { + return Err(Error::Missing(*revision)); }; - if let Some(comment) = comment.get_mut() { - comment.edit(op.clock, body, timestamp); + // If the revision was redacted concurrently, there's nothing to do. + // Likewise, if the review was redacted concurrently, there's nothing to do. + if let Some(rev) = rev { + let Some(review) = rev.reviews.get_mut(author) else { + return Err(Error::Missing(review)); + }; + if let Some(review) = review { + let Some(comment) = review.comments.get_mut(&comment) else { + return Err(Error::Missing(comment)); + }; + if let Some(comment) = comment { + comment.edit(body, timestamp); + } + } } } + Some(None) => { + // Redacted. + } + None => return Err(Error::Missing(review)), } } Action::CodeComment { @@ -558,34 +515,38 @@ impl store::FromHistory for Patch { body, location, } => { - let Some(Redactable::Present((revision, author))) = - self.reviews.get(&review) else { - return Err(Error::Missing(review)); - }; - let Some(rev) = self.revisions.get_mut(revision) else { - return Err(Error::Missing(*revision)); - }; - // If the revision was redacted concurrently, there's nothing to do. - // Likewise, if the review was redacted concurrently, there's nothing to do. - if let Some(rev) = rev.get_mut() { - let Some(review) = rev.reviews.get_mut(author) else { - return Err(Error::Missing(review)); - }; - if let Redactable::Present(review) = review { - review.comments.insert( - id, - Redactable::Present(CodeComment::new( - op.author, body, location, timestamp, - )), - ); + match self.reviews.get(&review) { + Some(Some((revision, author))) => { + let Some(rev) = self.revisions.get_mut(revision) else { + return Err(Error::Missing(*revision)); + }; + // If the revision was redacted concurrently, there's nothing to do. + // Likewise, if the review was redacted concurrently, there's nothing to do. + if let Some(rev) = rev { + let Some(review) = rev.reviews.get_mut(author) else { + return Err(Error::Missing(review)); + }; + if let Some(review) = review { + review.comments.insert( + id, + Some(CodeComment::new( + op.author, body, location, timestamp, + )), + ); + } + } } + Some(None) => { + // Redacted. + } + None => return Err(Error::Missing(review)), } } Action::Merge { revision, commit } => { let Some(rev) = self.revisions.get_mut(&revision) else { return Err(Error::Missing(revision)); }; - if rev.get().is_some() { + if rev.is_some() { let doc = repo.identity_doc_at(op.identity)?.verified()?; match self.target() { @@ -612,12 +573,11 @@ impl store::FromHistory for Patch { } self.merges.insert( op.author, - Immutable::new(Merge { + Merge { revision, commit, timestamp, - }), - op.clock, + }, ); let mut merges = self.merges.iter().fold( @@ -636,43 +596,32 @@ impl store::FromHistory for Patch { } [(revision, commit)] => { // Patch is merged. - self.state.set( - State::Merged { - revision: *revision, - commit: *commit, - }, - op.clock, - ); + self.state = State::Merged { + revision: *revision, + commit: *commit, + }; } revisions => { // More than one revision met the quorum. - self.state.set( - State::Open { - conflicts: revisions.to_vec(), - }, - op.clock, - ); + self.state = State::Open { + conflicts: revisions.to_vec(), + }; } } } } Action::Thread { revision, action } => { - // TODO(cloudhead): Make sure we can deal with redacted revisions which are added - // to out of order, like in the `Merge` case. - if let Some(Redactable::Present(revision)) = self.revisions.get_mut(&revision) { - revision.discussion.apply( - [cob::Op::new( - op.id, - action, - op.author, - timestamp, - op.clock, - op.identity, - )], - repo, - )?; - } else { - return Err(Error::Missing(revision)); + match self.revisions.get_mut(&revision) { + Some(Some(revision)) => { + revision.discussion.apply( + cob::Op::new(op.id, action, op.author, timestamp, op.identity), + repo, + )?; + } + Some(None) => { + // Redacted. + } + None => return Err(Error::Missing(revision)), } } } @@ -687,7 +636,7 @@ pub struct Revision { /// Author of the revision. author: Author, /// Revision description. - description: LWWReg>, + description: String, /// Base branch commit, used as a merge base. base: git::Oid, /// Reference to the Git object containing the code (revision head). @@ -695,7 +644,7 @@ pub struct Revision { /// Discussion around this revision. discussion: Thread, /// Reviews of this revision's changes (one per actor). - reviews: GMap>, + reviews: BTreeMap>, /// When this revision was created. timestamp: Timestamp, } @@ -707,21 +656,20 @@ impl Revision { base: git::Oid, oid: git::Oid, timestamp: Timestamp, - clock: Clock, ) -> Self { Self { author, - description: LWWReg::new(Max::from(description), clock), + description, base, oid, discussion: Thread::default(), - reviews: GMap::default(), + reviews: BTreeMap::default(), timestamp, } } pub fn description(&self) -> &str { - self.description.get() + self.description.as_str() } /// Author of the revision. @@ -753,12 +701,12 @@ impl Revision { pub fn reviews(&self) -> impl DoubleEndedIterator { self.reviews .iter() - .filter_map(|(author, review)| review.get().map(|r| (author, r))) + .filter_map(|(author, review)| review.as_ref().map(|r| (author, r))) } /// Get a review by author. pub fn review(&self, author: &ActorId) -> Option<&Review> { - self.reviews.get(author).and_then(Redactable::get) + self.reviews.get(author).and_then(|o| o.as_ref()) } } @@ -821,14 +769,6 @@ pub enum Verdict { Reject, } -impl Semilattice for Verdict { - fn merge(&mut self, other: Self) { - if self == &Self::Accept && other == Self::Reject { - *self = other; - } - } -} - impl fmt::Display for Verdict { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { @@ -871,7 +811,7 @@ pub struct CodeComment { /// Code location of the comment. location: CodeLocation, /// Comment edits. - edits: GMap>, + edits: Vec, } impl Serialize for CodeComment { @@ -899,14 +839,13 @@ impl CodeComment { Self { author, location, - edits: GMap::singleton(Lamport::initial(), Max::from(edit)), + edits: vec![edit], } } /// Add an edit. - pub fn edit(&mut self, clock: Lamport, body: String, timestamp: Timestamp) { - self.edits - .insert(clock, thread::Edit { body, timestamp }.into()) + pub fn edit(&mut self, body: String, timestamp: Timestamp) { + self.edits.push(thread::Edit { body, timestamp }); } /// Comment author. @@ -924,7 +863,7 @@ impl CodeComment { // SAFETY: There is always at least one edit. This is guaranteed by [`CodeComment::new`] // constructor. #[allow(clippy::unwrap_used)] - self.edits.values().last().unwrap().get().body.as_str() + self.edits.last().unwrap().body.as_str() } } @@ -938,9 +877,9 @@ pub struct Review { /// Review summary. /// /// Can be edited or set to `None`. - summary: LWWReg>>, + summary: Option, /// Review inline code comments. - comments: GMap>, + comments: BTreeMap>, /// Review timestamp. timestamp: Timestamp, } @@ -963,16 +902,11 @@ impl Serialize for Review { } impl Review { - pub fn new( - verdict: Option, - summary: Option, - timestamp: Timestamp, - clock: Clock, - ) -> Self { + pub fn new(verdict: Option, summary: Option, timestamp: Timestamp) -> Self { Self { verdict, - summary: LWWReg::new(summary.map(Max::from), clock), - comments: GMap::default(), + summary, + comments: BTreeMap::default(), timestamp, } } @@ -986,12 +920,12 @@ impl Review { pub fn comments(&self) -> impl Iterator { self.comments .iter() - .filter_map(|(id, r)| r.get().map(|comment| (id, comment))) + .filter_map(|(id, r)| r.as_ref().map(|comment| (id, comment))) } /// Review general comment. pub fn summary(&self) -> Option<&str> { - self.summary.get().as_ref().map(|m| m.get().as_str()) + self.summary.as_deref() } /// Review timestamp. @@ -1146,7 +1080,6 @@ pub struct PatchMut<'a, 'g, R> { pub id: ObjectId, patch: Patch, - clock: clock::Lamport, store: &'g mut Patches<'a, R>, } @@ -1154,18 +1087,8 @@ impl<'a, 'g, R> PatchMut<'a, 'g, R> where R: ReadRepository + SignRepository + cob::Store, { - pub fn new( - id: ObjectId, - patch: Patch, - clock: clock::Lamport, - store: &'g mut Patches<'a, R>, - ) -> Self { - Self { - id, - clock, - patch, - store, - } + pub fn new(id: ObjectId, patch: Patch, store: &'g mut Patches<'a, R>) -> Self { + Self { id, patch, store } } pub fn transaction( @@ -1178,21 +1101,15 @@ where G: Signer, F: FnOnce(&mut Transaction) -> Result<(), store::Error>, { - let mut tx = Transaction::new(*signer.public_key(), self.clock); + let mut tx = Transaction::new(*signer.public_key()); operations(&mut tx)?; - let (ops, clock, commit) = tx.commit(message, self.id, &mut self.store.raw, signer)?; + let (op, commit) = tx.commit(message, self.id, &mut self.store.raw, signer)?; - self.patch.apply(ops, self.store.as_ref())?; - self.clock = clock; + self.patch.apply(op, self.store.as_ref())?; Ok(commit) } - /// Get the internal logical clock. - pub fn clock(&self) -> &clock::Lamport { - &self.clock - } - /// Edit patch metadata. pub fn edit( &mut self, @@ -1406,7 +1323,7 @@ where let all = self.all()?; let state_groups = all.filter_map(|s| s.ok()) - .fold(PatchCounts::default(), |mut state, (_, p, _)| { + .fold(PatchCounts::default(), |mut state, (_, p)| { match p.state() { State::Draft => state.draft += 1, State::Open { .. } => state.open += 1, @@ -1433,35 +1350,33 @@ where let result = self .all()? .filter_map(|result| result.ok()) - .find_map(|(p_id, p, _)| p.revision(id).map(|r| (p_id, p.clone(), r.clone()))); + .find_map(|(p_id, p)| p.revision(id).map(|r| (p_id, p.clone(), r.clone()))); Ok(result) } /// Get a patch. pub fn get(&self, id: &ObjectId) -> Result, store::Error> { - self.raw.get(id).map(|r| r.map(|(p, _)| p)) + self.raw.get(id) } /// Get proposed patches. - pub fn proposed( - &self, - ) -> Result + '_, Error> { + pub fn proposed(&self) -> Result + '_, Error> { let all = self.all()?; Ok(all .into_iter() .filter_map(|result| result.ok()) - .filter(|(_, p, _)| p.is_open())) + .filter(|(_, p)| p.is_open())) } /// Get patches proposed by the given key. pub fn proposed_by<'b>( &'b self, who: &'b Did, - ) -> Result + '_, Error> { + ) -> Result + '_, Error> { Ok(self .proposed()? - .filter(move |(_, p, _)| p.author().id() == who)) + .filter(move |(_, p)| p.author().id() == who)) } } @@ -1517,14 +1432,13 @@ where /// Get a patch mutably. pub fn get_mut<'g>(&'g mut self, id: &ObjectId) -> Result, store::Error> { - let (patch, clock) = self + let patch = self .raw .get(id)? .ok_or_else(move || store::Error::NotFound(TYPENAME.clone(), *id))?; Ok(PatchMut { id: *id, - clock, patch, store: self, }) @@ -1542,37 +1456,28 @@ where state: State, signer: &G, ) -> Result, Error> { - let (id, patch, clock) = - Transaction::initial("Create patch", &mut self.raw, signer, |tx| { - tx.revision(description, base, oid)?; - tx.edit(title, target)?; - tx.tag(tags.to_owned(), [])?; + let (id, patch) = Transaction::initial("Create patch", &mut self.raw, signer, |tx| { + tx.revision(description, base, oid)?; + tx.edit(title, target)?; + tx.tag(tags.to_owned(), [])?; - if state != State::default() { - tx.lifecycle(state)?; - } - Ok(()) - })?; - // Just a sanity check that our clock is advancing as expected. - debug_assert_eq!(clock.get(), 1); + if state != State::default() { + tx.lifecycle(state)?; + } + Ok(()) + })?; - Ok(PatchMut::new(id, patch, clock, self)) + Ok(PatchMut::new(id, patch, self)) } } #[cfg(test)] mod test { use std::str::FromStr; - use std::{array, iter}; - use radicle_crdt::test::{assert_laws, WeightedGenerator}; - - use nonempty::nonempty; use pretty_assertions::assert_eq; - use qcheck::{Arbitrary, TestResult}; use super::*; - use crate::assert_matches; use crate::cob::test::Actor; use crate::crypto::test::signer::MockSigner; use crate::test; @@ -1580,149 +1485,6 @@ mod test { use crate::test::arbitrary::gen; use crate::test::storage::MockRepository; - #[derive(Clone)] - struct Changes { - permutations: [Vec; N], - } - - impl std::fmt::Debug for Changes { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - for (i, p) in self.permutations.iter().enumerate() { - writeln!( - f, - "{i}: {:#?}", - p.iter().map(|c| &c.action).collect::>() - )?; - } - Ok(()) - } - } - - impl Arbitrary for Changes { - fn arbitrary(g: &mut qcheck::Gen) -> Self { - type State = (Actor, clock::Lamport, Vec, Vec); - - let rng = fastrand::Rng::with_seed(u64::arbitrary(g)); - let oids = iter::repeat_with(|| { - git::Oid::try_from( - iter::repeat_with(|| rng.u8(..)) - .take(20) - .collect::>() - .as_slice(), - ) - .unwrap() - }) - .take(16) - .collect::>(); - - let gen = WeightedGenerator::<(clock::Lamport, Op), State>::new(rng.clone()) - .variant(1, |(actor, clock, _, _), rng| { - Some(( - clock.tick(), - actor.op(Action::Edit { - title: iter::repeat_with(|| rng.alphabetic()).take(8).collect(), - target: MergeTarget::Delegates, - }), - )) - }) - .variant(1, |(actor, clock, revisions, _), rng| { - if revisions.is_empty() { - return None; - } - let revision = revisions[rng.usize(..revisions.len())]; - let commit = oids[rng.usize(..oids.len())]; - - Some((clock.tick(), actor.op(Action::Merge { revision, commit }))) - }) - .variant(1, |(actor, clock, revisions, _), rng| { - if revisions.is_empty() { - return None; - } - let revision = revisions[rng.usize(..revisions.len())]; - - Some((clock.tick(), actor.op(Action::Redact { revision }))) - }) - .variant(1, |(actor, clock, _, tags), rng| { - let add = iter::repeat_with(|| rng.alphabetic()) - .take(rng.usize(0..=3)) - .map(|c| Tag::new(c).unwrap()) - .collect::>(); - let remove = tags - .iter() - .take(rng.usize(0..=tags.len())) - .cloned() - .collect(); - for tag in &add { - tags.push(tag.clone()); - } - Some((clock.tick(), actor.op(Action::Tag { add, remove }))) - }) - .variant(1, |(actor, clock, revisions, _), rng| { - let oid = oids[rng.usize(..oids.len())]; - let base = oids[rng.usize(..oids.len())]; - let description = iter::repeat_with(|| rng.alphabetic()).take(6).collect(); - let op = actor.op(Action::Revision { - description, - base, - oid, - }); - - if rng.bool() { - revisions.push(op.id); - } - Some((*clock, op)) - }); - - let mut changes = Vec::new(); - let mut permutations: [Vec; N] = array::from_fn(|_| Vec::new()); - - for (_, op) in gen.take(g.size()) { - changes.push(op); - } - - for p in &mut permutations { - *p = changes.clone(); - rng.shuffle(&mut changes); - } - - Changes { permutations } - } - } - - #[test] - fn prop_invariants() { - fn property(repo: MockRepository, log: Changes<3>) -> TestResult { - let t = Patch::default(); - let [p1, p2, p3] = log.permutations; - - let mut t1 = t.clone(); - if t1.apply(p1, &repo).is_err() { - return TestResult::discard(); - } - - let mut t2 = t.clone(); - if t2.apply(p2, &repo).is_err() { - return TestResult::discard(); - } - - let mut t3 = t; - if t3.apply(p3, &repo).is_err() { - return TestResult::discard(); - } - - assert_eq!(t1, t2); - assert_eq!(t2, t3); - assert_laws(&t1, &t2, &t3); - - TestResult::passed() - } - - qcheck::QuickCheck::new() - .min_tests_passed(100) - .gen(qcheck::Gen::new(7)) - .quickcheck(property as fn(MockRepository, Changes<3>) -> TestResult); - } - #[test] fn test_json_serialization() { let edit = Action::Tag { @@ -1737,27 +1499,24 @@ mod test { #[test] fn test_patch_create_and_get() { - let tmp = tempfile::tempdir().unwrap(); - let ctx = test::setup::Context::new(&tmp); - let signer = &ctx.signer; - let pr = ctx.branch_with(test::setup::initial_blobs()); - let mut patches = Patches::open(&ctx.project).unwrap(); - let author: Did = signer.public_key().into(); + let alice = test::setup::NodeWithRepo::default(); + let checkout = alice.repo.checkout(); + let branch = checkout.branch_with([("README", b"Hello World!")]); + let mut patches = Patches::open(&*alice.repo).unwrap(); + let author: Did = alice.signer.public_key().into(); let target = MergeTarget::Delegates; let patch = patches .create( "My first patch", "Blah blah blah.", target, - pr.base, - pr.oid, + branch.base, + branch.oid, &[], - signer, + &alice.signer, ) .unwrap(); - assert_eq!(patch.clock.get(), 1); - let patch_id = patch.id; let patch = patches.get(&patch_id).unwrap().unwrap(); @@ -1773,8 +1532,8 @@ mod test { assert_eq!(revision.author.id(), &author); assert_eq!(revision.description(), "Blah blah blah."); assert_eq!(revision.discussion.len(), 0); - assert_eq!(revision.oid, pr.oid); - assert_eq!(revision.base, pr.base); + assert_eq!(revision.oid, branch.oid); + assert_eq!(revision.base, branch.base); let (id, _, _) = patches.find_by_revision(rev_id).unwrap().unwrap(); assert_eq!(id, patch_id); @@ -1782,20 +1541,19 @@ mod test { #[test] fn test_patch_discussion() { - let tmp = tempfile::tempdir().unwrap(); - let ctx = test::setup::Context::new(&tmp); - let signer = &ctx.signer; - let pr = ctx.branch_with(test::setup::initial_blobs()); - let mut patches = Patches::open(&ctx.project).unwrap(); + let alice = test::setup::NodeWithRepo::default(); + let checkout = alice.repo.checkout(); + let branch = checkout.branch_with([("README", b"Hello World!")]); + let mut patches = Patches::open(&*alice.repo).unwrap(); let patch = patches .create( "My first patch", "Blah blah blah.", MergeTarget::Delegates, - pr.base, - pr.oid, + branch.base, + branch.oid, &[], - signer, + &alice.signer, ) .unwrap(); @@ -1804,7 +1562,7 @@ mod test { let (revision_id, _) = patch.revisions().last().unwrap(); assert!( patch - .comment(*revision_id, "patch comment", None, signer) + .comment(*revision_id, "patch comment", None, &alice.signer) .is_ok(), "can comment on patch" ); @@ -1816,26 +1574,25 @@ mod test { #[test] fn test_patch_merge() { - let tmp = tempfile::tempdir().unwrap(); - let ctx = test::setup::Context::new(&tmp); - let signer = &ctx.signer; - let pr = ctx.branch_with(test::setup::initial_blobs()); - let mut patches = Patches::open(&ctx.project).unwrap(); + let alice = test::setup::NodeWithRepo::default(); + let checkout = alice.repo.checkout(); + let branch = checkout.branch_with([("README", b"Hello World!")]); + let mut patches = Patches::open(&*alice.repo).unwrap(); let mut patch = patches .create( "My first patch", "Blah blah blah.", MergeTarget::Delegates, - pr.base, - pr.oid, + branch.base, + branch.oid, &[], - signer, + &alice.signer, ) .unwrap(); let id = patch.id; let (rid, _) = patch.revisions().next().unwrap(); - let _merge = patch.merge(*rid, pr.base, signer).unwrap(); + let _merge = patch.merge(*rid, branch.base, &alice.signer).unwrap(); let patch = patches.get(&id).unwrap().unwrap(); @@ -1843,83 +1600,36 @@ mod test { assert_eq!(merges.len(), 1); let (merger, merge) = merges.first().unwrap(); - assert_eq!(*merger, signer.public_key()); - assert_eq!(merge.commit, pr.base); - } - - #[test] - fn test_patch_merge_and_archive() { - let rid = gen::(1); - let base = git::Oid::from_str("d8711a8d43dc919fe39ae4b7c2f7b24667f5d470").unwrap(); - let commit = git::Oid::from_str("cb18e95ada2bb38aadd8e6cef0963ce37a87add3").unwrap(); - - let mut alice = Actor::::default(); - let mut bob = Actor::::default(); - - let proj = gen::(1); - let doc = Doc::new(proj, nonempty![alice.did(), bob.did()], 1) - .verified() - .unwrap(); - let repo = MockRepository::new(rid, doc); - let patch = alice - .patch("Some changes", "", base, commit, &repo) - .unwrap(); - let (revision, _) = patch.revisions().next().unwrap(); - - // Create two concurrent operations. - let clock = Lamport::from(2); - let identity = repo.identity_head().unwrap(); - let ops = [ - alice.op_with( - Action::Merge { - revision: *revision, - commit, - }, - clock, - identity, - ), - bob.op_with( - Action::Lifecycle { - state: State::Archived, - }, - clock, - identity, - ), - ]; - - let mut patch1 = patch.clone(); - let mut patch2 = patch.clone(); - - // Apply the ops in different orders and expect the patch state to remain the same. - patch1.apply(ops.iter().cloned(), &repo).unwrap(); - patch2.apply(ops.iter().cloned().rev(), &repo).unwrap(); - - assert_matches!(patch1.state(), &State::Merged { .. }); - assert_matches!(patch2.state(), &State::Merged { .. }); + assert_eq!(*merger, alice.signer.public_key()); + assert_eq!(merge.commit, branch.base); } #[test] fn test_patch_review() { - let tmp = tempfile::tempdir().unwrap(); - let ctx = test::setup::Context::new(&tmp); - let signer = &ctx.signer; - let pr = ctx.branch_with(test::setup::initial_blobs()); - let mut patches = Patches::open(&ctx.project).unwrap(); + let alice = test::setup::NodeWithRepo::default(); + let checkout = alice.repo.checkout(); + let branch = checkout.branch_with([("README", b"Hello World!")]); + let mut patches = Patches::open(&*alice.repo).unwrap(); let mut patch = patches .create( "My first patch", "Blah blah blah.", MergeTarget::Delegates, - pr.base, - pr.oid, + branch.base, + branch.oid, &[], - signer, + &alice.signer, ) .unwrap(); let (rid, _) = patch.latest(); patch - .review(*rid, Some(Verdict::Accept), Some("LGTM".to_owned()), signer) + .review( + *rid, + Some(Verdict::Accept), + Some("LGTM".to_owned()), + &alice.signer, + ) .unwrap(); let id = patch.id; @@ -1927,7 +1637,7 @@ mod test { let (_, revision) = patch.latest(); assert_eq!(revision.reviews.len(), 1); - let review = revision.review(signer.public_key()).unwrap(); + let review = revision.review(alice.signer.public_key()).unwrap(); assert_eq!(review.verdict(), Some(Verdict::Accept)); assert_eq!(review.summary(), Some("LGTM")); } @@ -1956,14 +1666,14 @@ mod test { commit: oid, }); - patch.apply([a1], &repo).unwrap(); + patch.apply(a1, &repo).unwrap(); assert!(patch.revisions().next().is_some()); - patch.apply([a2], &repo).unwrap(); + patch.apply(a2, &repo).unwrap(); assert!(patch.revisions().next().is_none()); - patch.apply([a3], &repo).unwrap(); - patch.apply([a4], &repo).unwrap(); + patch.apply(a3, &repo).unwrap(); + patch.apply(a4, &repo).unwrap(); } #[test] @@ -1971,44 +1681,65 @@ mod test { let base = arbitrary::oid(); let oid = arbitrary::oid(); let repo = gen::(1); - let mut alice = Actor::new(MockSigner::default()); - let mut p1 = Patch::default(); - let mut p2 = Patch::default(); + let time = Timestamp::now(); + let alice = MockSigner::default(); + let bob = MockSigner::default(); + let mut h0: cob::test::HistoryBuilder = cob::test::history( + &Action::Revision { + description: String::from("Original"), + base, + oid, + }, + time, + &alice, + ); + h0.commit( + &Action::Edit { + title: String::from("Some patch"), + target: MergeTarget::Delegates, + }, + &alice, + ); - let a1 = alice.op(Action::Revision { - description: String::new(), - base, - oid, - }); - let a2 = alice.op(Action::Redact { revision: a1.id }); - let a3 = alice.op(Action::EditRevision { - revision: a1.id, - description: String::from("Edited"), - }); + let mut h1 = h0.clone(); + h1.commit( + &Action::Redact { + revision: h0.root(), + }, + &alice, + ); - p1.apply([a1.clone(), a2.clone(), a3.clone()], &repo) - .unwrap(); - p2.apply([a1, a3, a2], &repo).unwrap(); + let mut h2 = h0.clone(); + h2.commit( + &Action::EditRevision { + revision: h0.root(), + description: String::from("Edited"), + }, + &bob, + ); - assert_eq!(p1, p2); + h0.merge(h1); + h0.merge(h2); + + let patch = Patch::from_history(&h0, &repo).unwrap(); + assert_eq!(patch.revisions().count(), 0); } #[test] fn test_patch_review_edit() { - let tmp = tempfile::tempdir().unwrap(); - let ctx = test::setup::Context::new(&tmp); - let signer = &ctx.signer; - let pr = ctx.branch_with(test::setup::initial_blobs()); - let mut patches = Patches::open(&ctx.project).unwrap(); + let alice = test::setup::NodeWithRepo::default(); + let checkout = alice.repo.checkout(); + let branch = checkout.branch_with([("README", b"Hello World!")]); + let mut patches = Patches::open(&*alice.repo).unwrap(); let mut patch = patches .create( "My first patch", "Blah blah blah.", MergeTarget::Delegates, - pr.base, - pr.oid, + branch.base, + branch.oid, &[], - signer, + &alice.signer, ) .unwrap(); @@ -2016,56 +1747,60 @@ mod test { let rid = *rid; let review = patch - .review(rid, Some(Verdict::Accept), Some("LGTM".to_owned()), signer) + .review( + rid, + Some(Verdict::Accept), + Some("LGTM".to_owned()), + &alice.signer, + ) .unwrap(); patch - .edit_review(review, Some("Whoops!".to_owned()), signer) + .edit_review(review, Some("Whoops!".to_owned()), &alice.signer) .unwrap(); // Overwrite the comment. // let (_, revision) = patch.latest(); - let review = revision.review(signer.public_key()).unwrap(); + let review = revision.review(alice.signer.public_key()).unwrap(); assert_eq!(review.verdict(), Some(Verdict::Accept)); assert_eq!(review.summary(), Some("Whoops!")); } #[test] fn test_patch_review_comment() { - let tmp = tempfile::tempdir().unwrap(); - let ctx = test::setup::Context::new(&tmp); - let signer = &ctx.signer; - let pr = ctx.branch_with(test::setup::initial_blobs()); - let mut patches = Patches::open(&ctx.project).unwrap(); + let alice = test::setup::NodeWithRepo::default(); + let checkout = alice.repo.checkout(); + let branch = checkout.branch_with([("README", b"Hello World!")]); + let mut patches = Patches::open(&*alice.repo).unwrap(); let mut patch = patches .create( "My first patch", "Blah blah blah.", MergeTarget::Delegates, - pr.base, - pr.oid, + branch.base, + branch.oid, &[], - signer, + &alice.signer, ) .unwrap(); let (rid, _) = patch.latest(); let rid = *rid; let location = CodeLocation { - path: PathBuf::from_str("README.md").unwrap(), + path: PathBuf::from_str("README").unwrap(), old: None, new: Some(5..8), }; - let review = patch.review(rid, None, None, signer).unwrap(); + let review = patch.review(rid, None, None, &alice.signer).unwrap(); patch .code_comment( review, "I like these lines of code", location.clone(), - signer, + &alice.signer, ) .unwrap(); let (_, revision) = patch.latest(); - let review = revision.review(signer.public_key()).unwrap(); + let review = revision.review(alice.signer.public_key()).unwrap(); let (_, comment) = review.comments().next().unwrap(); assert_eq!(comment.body(), "I like these lines of code"); @@ -2074,66 +1809,62 @@ mod test { #[test] fn test_patch_review_remove_summary() { - let tmp = tempfile::tempdir().unwrap(); - let ctx = test::setup::Context::new(&tmp); - let signer = &ctx.signer; - let pr = ctx.branch_with(test::setup::initial_blobs()); - let mut patches = Patches::open(&ctx.project).unwrap(); + let alice = test::setup::NodeWithRepo::default(); + let checkout = alice.repo.checkout(); + let branch = checkout.branch_with([("README", b"Hello World!")]); + let mut patches = Patches::open(&*alice.repo).unwrap(); let mut patch = patches .create( "My first patch", "Blah blah blah.", MergeTarget::Delegates, - pr.base, - pr.oid, + branch.base, + branch.oid, &[], - signer, + &alice.signer, ) .unwrap(); let (rid, _) = patch.latest(); let rid = *rid; let review = patch - .review(rid, None, Some("Nah".to_owned()), signer) + .review(rid, None, Some("Nah".to_owned()), &alice.signer) .unwrap(); - patch.edit_review(review, None, signer).unwrap(); + patch.edit_review(review, None, &alice.signer).unwrap(); let id = patch.id; let patch = patches.get_mut(&id).unwrap(); let (_, revision) = patch.latest(); - let review = revision.review(signer.public_key()).unwrap(); + let review = revision.review(alice.signer.public_key()).unwrap(); assert_eq!(review.summary(), None); } #[test] fn test_patch_update() { - let tmp = tempfile::tempdir().unwrap(); - let ctx = test::setup::Context::new(&tmp); - let signer = &ctx.signer; - let pr = ctx.branch_with(test::setup::initial_blobs()); - let mut patches = Patches::open(&ctx.project).unwrap(); + let alice = test::setup::NodeWithRepo::default(); + let checkout = alice.repo.checkout(); + let branch = checkout.branch_with([("README", b"Hello World!")]); + let mut patches = Patches::open(&*alice.repo).unwrap(); let mut patch = patches .create( "My first patch", "Blah blah blah.", MergeTarget::Delegates, - pr.base, - pr.oid, + branch.base, + branch.oid, &[], - signer, + &alice.signer, ) .unwrap(); - assert_eq!(patch.clock.get(), 1); assert_eq!(patch.description(), "Blah blah blah."); assert_eq!(patch.version(), 0); - let update = ctx.branch_with(test::setup::update_blobs()); + let update = checkout.branch_with([("README", b"Hello Radicle!")]); let _ = patch - .update("I've made changes.", pr.base, update.oid, signer) + .update("I've made changes.", branch.base, update.oid, &alice.signer) .unwrap(); - assert_eq!(patch.clock.get(), 2); let id = patch.id; let patch = patches.get(&id).unwrap().unwrap(); @@ -2158,35 +1889,38 @@ mod test { #[test] fn test_patch_redact() { - let tmp = tempfile::tempdir().unwrap(); - let ctx = test::setup::Context::new(&tmp); - let signer = &ctx.signer; - let pr = ctx.branch_with(test::setup::initial_blobs()); - let mut patches = Patches::open(&ctx.project).unwrap(); + let alice = test::setup::Node::default(); + let repo = alice.project(); + let branch = repo + .checkout() + .branch_with([("README.md", b"Hello, World!")]); + let mut patches = Patches::open(&*repo).unwrap(); let mut patch = patches .create( "My first patch", "Blah blah blah.", MergeTarget::Delegates, - pr.base, - pr.oid, + branch.base, + branch.oid, &[], - signer, + &alice.signer, ) .unwrap(); let patch_id = patch.id; - let update = ctx.branch_with(test::setup::update_blobs()); + let update = repo + .checkout() + .branch_with([("README.md", b"Hello, Radicle!")]); let revision_id = patch - .update("I've made changes.", pr.base, update.oid, signer) + .update("I've made changes.", branch.base, update.oid, &alice.signer) .unwrap(); assert_eq!(patch.revisions().count(), 2); - patch.redact(revision_id, signer).unwrap(); + patch.redact(revision_id, &alice.signer).unwrap(); assert_eq!(patch.latest().0, &RevisionId::from(patch_id)); assert_eq!(patch.revisions().count(), 1); // The patch's root must always exist. - assert!(patch.redact(*patch.latest().0, signer).is_err()); + assert!(patch.redact(*patch.latest().0, &alice.signer).is_err()); } } diff --git a/radicle/src/cob/store.rs b/radicle/src/cob/store.rs index db4e76fe..d343115a 100644 --- a/radicle/src/cob/store.rs +++ b/radicle/src/cob/store.rs @@ -6,10 +6,10 @@ use std::ops::ControlFlow; use std::sync::Arc; use nonempty::NonEmpty; -use radicle_crdt::Lamport; use serde::{Deserialize, Serialize}; -use crate::cob::op::{Op, Ops}; +use crate::cob::common::Timestamp; +use crate::cob::op::Op; use crate::cob::{ActorId, Create, EntryId, History, ObjectId, TypeName, Update, Updated}; use crate::git; use crate::prelude::*; @@ -20,7 +20,7 @@ use crate::{cob, identity}; /// History type for standard radicle COBs. pub const HISTORY_TYPE: &str = "radicle"; -pub trait HistoryAction { +pub trait HistoryAction: std::fmt::Debug { /// Parent objects this action depends on. For example, patch revisions /// have the commit objects as their parent. fn parents(&self) -> Vec { @@ -42,7 +42,7 @@ pub trait FromHistory: Sized + Default + PartialEq { /// Apply a list of operations to the state. fn apply( &mut self, - ops: impl IntoIterator>, + op: Op, repo: &R, ) -> Result<(), Self::Error>; @@ -50,14 +50,11 @@ pub trait FromHistory: Sized + Default + PartialEq { fn validate(&self) -> Result<(), Self::Error>; /// Create an object from a history. - fn from_history( - history: &History, - repo: &R, - ) -> Result<(Self, Lamport), Self::Error> { + fn from_history(history: &History, repo: &R) -> Result { let obj = history.traverse(Self::default(), |mut acc, _, entry| { - match Ops::try_from(entry) { - Ok(Ops(ops)) => { - if let Err(err) = acc.apply(ops, repo) { + match Op::try_from(entry) { + Ok(op) => { + if let Err(err) = acc.apply(op, repo) { log::warn!("Error applying op to `{}` state: {err}", Self::type_name()); return ControlFlow::Break(acc); } @@ -75,7 +72,7 @@ pub trait FromHistory: Sized + Default + PartialEq { obj.validate()?; - Ok((obj, history.clock().into())) + Ok(obj) } /// Create an object from individual operations. @@ -85,8 +82,9 @@ pub trait FromHistory: Sized + Default + PartialEq { repo: &R, ) -> Result { let mut state = Self::default(); - state.apply(ops, repo)?; - + for op in ops { + state.apply(op, repo)?; + } Ok(state) } } @@ -197,7 +195,7 @@ where message: &str, actions: impl Into>, signer: &G, - ) -> Result<(ObjectId, T, Lamport), Error> { + ) -> Result<(ObjectId, T), Error> { let actions = actions.into(); let parents = actions.iter().flat_map(T::Action::parents).collect(); let contents = actions.try_map(encoding::encode)?; @@ -214,11 +212,11 @@ where contents, }, )?; - let (object, clock) = T::from_history(cob.history(), self.repo).map_err(Error::apply)?; + let object = T::from_history(cob.history(), self.repo).map_err(Error::apply)?; self.repo.sign_refs(signer).map_err(Error::SignRefs)?; - Ok((*cob.id(), object, clock)) + Ok((*cob.id(), object)) } /// Remove an object. @@ -250,30 +248,28 @@ where T::Action: Serialize, { /// Get an object. - pub fn get(&self, id: &ObjectId) -> Result, Error> { + pub fn get(&self, id: &ObjectId) -> Result, Error> { let cob = cob::get(self.repo, T::type_name(), id)?; if let Some(cob) = cob { if cob.manifest().history_type != HISTORY_TYPE { return Err(Error::HistoryType(cob.manifest().history_type.clone())); } - let (obj, clock) = T::from_history(cob.history(), self.repo).map_err(Error::apply)?; + let obj = T::from_history(cob.history(), self.repo).map_err(Error::apply)?; - Ok(Some((obj, clock))) + Ok(Some(obj)) } else { Ok(None) } } /// Return all objects. - pub fn all( - &self, - ) -> Result> + 'a, Error> { + pub fn all(&self) -> Result> + 'a, Error> { let raw = cob::list(self.repo, T::type_name())?; Ok(raw.into_iter().map(|o| { - let (obj, clock) = T::from_history(o.history(), self.repo).map_err(Error::apply)?; - Ok((*o.id(), obj, clock)) + let obj = T::from_history(o.history(), self.repo).map_err(Error::apply)?; + Ok((*o.id(), obj)) })) } @@ -294,16 +290,14 @@ where #[derive(Debug)] pub struct Transaction { actor: ActorId, - clock: Lamport, actions: Vec, } impl Transaction { /// Create a new transaction. - pub fn new(actor: ActorId, clock: Lamport) -> Self { + pub fn new(actor: ActorId) -> Self { Self { actor, - clock, actions: Vec::new(), } } @@ -314,7 +308,7 @@ impl Transaction { store: &mut Store, signer: &G, operations: F, - ) -> Result<(ObjectId, T, Lamport), Error> + ) -> Result<(ObjectId, T), Error> where G: Signer, F: FnOnce(&mut Self) -> Result<(), Error>, @@ -324,20 +318,15 @@ impl Transaction { let actor = *signer.public_key(); let mut tx = Transaction { actor, - // Nb. The clock is never zero. - clock: Lamport::initial().tick(), actions: Vec::new(), }; operations(&mut tx)?; let actions = NonEmpty::from_vec(tx.actions) .expect("Transaction::initial: transaction must contain at least one operation"); - let (id, cob, clock) = store.create(message, actions, signer)?; + let (id, cob) = store.create(message, actions, signer)?; - // The history clock should be in sync with the tx clock. - assert_eq!(clock, tx.clock); - - Ok((id, cob, clock)) + Ok((id, cob)) } /// Add an operation to this transaction. @@ -351,12 +340,12 @@ impl Transaction { /// /// Returns a list of operations that can be applied onto an in-memory CRDT. pub fn commit( - mut self, + self, msg: &str, id: ObjectId, store: &mut Store, signer: &G, - ) -> Result<(Vec>, Lamport, EntryId), Error> + ) -> Result<(cob::Op, EntryId), Error> where R: ReadRepository + SignRepository + cob::Store, T::Action: Serialize + Clone, @@ -366,27 +355,17 @@ impl Transaction { let Updated { head, object } = store.update(id, msg, actions.clone(), signer)?; let id = EntryId::from(head); let author = self.actor; - let timestamp = object.history().timestamp().into(); - let clock = self.clock.tick(); + let timestamp = Timestamp::from_secs(object.history().timestamp()); let identity = store.identity; + let op = cob::Op { + id, + actions, + author, + timestamp, + identity, + }; - // The history clock should be in sync with the tx clock. - assert_eq!(object.history().clock(), self.clock.get()); - - // Start the clock from where the transcation clock started. - let ops = actions - .into_iter() - .map(|action| cob::Op { - id, - action, - author, - clock, - timestamp, - identity, - }) - .collect(); - - Ok((ops, clock, id)) + Ok((op, id)) } } diff --git a/radicle/src/cob/test.rs b/radicle/src/cob/test.rs index 47a923aa..8e9e6eca 100644 --- a/radicle/src/cob/test.rs +++ b/radicle/src/cob/test.rs @@ -1,16 +1,14 @@ -use std::collections::BTreeSet; use std::marker::PhantomData; use std::ops::Deref; use nonempty::NonEmpty; -use serde::Serialize; +use serde::{Deserialize, Serialize}; -use crate::cob::common::clock; -use crate::cob::op::{Op, Ops}; +use crate::cob::op::Op; use crate::cob::patch; use crate::cob::patch::Patch; use crate::cob::store::encoding; -use crate::cob::{EntryId, History}; +use crate::cob::{EntryId, History, Timestamp}; use crate::crypto::Signer; use crate::git; use crate::git::ext::author::Author; @@ -29,6 +27,7 @@ use super::thread; pub struct HistoryBuilder { history: History, resource: Oid, + time: Timestamp, witness: PhantomData, } @@ -55,12 +54,11 @@ impl HistoryBuilder { impl HistoryBuilder where - T::Action: Serialize + Eq + 'static, + T::Action: for<'de> Deserialize<'de> + Serialize + Eq + 'static, { - pub fn new(action: &T::Action, signer: &G) -> HistoryBuilder { + pub fn new(action: &T::Action, time: Timestamp, signer: &G) -> HistoryBuilder { let resource = arbitrary::oid(); - let timestamp = clock::Physical::now().as_secs(); - let (data, root) = encoded::(action, timestamp as i64, [], signer); + let (data, root) = encoded::(action, time, [], signer); Self { history: History::new_from_root( @@ -68,8 +66,9 @@ where *signer.public_key(), resource, NonEmpty::new(data), - timestamp, + time.as_secs(), ), + time, resource, witness: PhantomData, } @@ -84,42 +83,19 @@ where } pub fn commit(&mut self, action: &T::Action, signer: &G) -> git::ext::Oid { - let timestamp = clock::Physical::now().as_secs(); + let timestamp = self.time; let tips = self.tips(); - let (data, oid) = encoded::(action, timestamp as i64, tips, signer); + let (data, oid) = encoded::(action, timestamp, tips, signer); self.history.extend( oid, *signer.public_key(), self.resource, NonEmpty::new(data), - timestamp, + timestamp.as_secs(), ); oid } - - /// Return a sorted list of operations by traversing the history in topological order. - /// In the case of partial orderings, a random order will be returned, using the provided RNG. - pub fn sorted(&self, rng: &mut fastrand::Rng) -> Vec> { - self.history - .sorted(|a, b| if rng.bool() { a.cmp(b) } else { b.cmp(a) }) - .flat_map(|entry| { - Ops::try_from(entry).expect("HistoryBuilder::sorted: operations must be valid") - }) - .collect() - } - - /// Return `n` permutations of the topological ordering of operations. - /// *This function will never return if less than `n` permutations exist.* - pub fn permutations(&self, n: usize) -> impl IntoIterator>> { - let mut permutations = BTreeSet::new(); - let mut rng = fastrand::Rng::new(); - - while permutations.len() < n { - permutations.insert(self.sorted(&mut rng)); - } - permutations.into_iter() - } } impl Deref for HistoryBuilder { @@ -131,17 +107,20 @@ impl Deref for HistoryBuilder { } /// Create a new test history. -pub fn history(action: &T::Action, signer: &G) -> HistoryBuilder +pub fn history( + action: &T::Action, + time: Timestamp, + signer: &G, +) -> HistoryBuilder where T::Action: Serialize + Eq + 'static, { - HistoryBuilder::new(action, signer) + HistoryBuilder::new(action, time, signer) } /// An object that can be used to create and sign operations. pub struct Actor { pub signer: G, - pub clock: clock::Lamport, } impl Default for Actor { @@ -152,10 +131,7 @@ impl Default for Actor { impl Actor { pub fn new(signer: G) -> Self { - Self { - signer, - clock: clock::Lamport::default(), - } + Self { signer } } } @@ -164,8 +140,8 @@ impl Actor { pub fn op_with( &mut self, action: A, - clock: clock::Lamport, identity: Oid, + timestamp: Timestamp, ) -> Op { let data = encoding::encode(serde_json::json!({ "action": action, @@ -175,13 +151,12 @@ impl Actor { let oid = git::raw::Oid::hash_object(git::raw::ObjectType::Blob, &data).unwrap(); let id = oid.into(); let author = *self.signer.public_key(); - let timestamp = clock::Physical::now(); + let actions = NonEmpty::new(action); Op { id, - action, + actions, author, - clock, timestamp, identity, } @@ -189,10 +164,10 @@ impl Actor { /// Create a new operation. pub fn op(&mut self, action: A) -> Op { - let clock = self.clock.tick(); let identity = arbitrary::oid(); + let timestamp = Timestamp::now(); - self.op_with(action, clock, identity) + self.op_with(action, identity, timestamp) } /// Get the actor's DID. @@ -234,7 +209,7 @@ impl Actor { /// that feeds into the hash entropy, so that changing any input will change the resulting oid. pub fn encoded( action: &T::Action, - timestamp: i64, + timestamp: Timestamp, parents: impl IntoIterator, signer: &G, ) -> (Vec, git::ext::Oid) { @@ -244,7 +219,7 @@ pub fn encoded( let author = Author { name: "radicle".to_owned(), email: signer.public_key().to_human(), - time: git_ext::author::Time::new(timestamp, 0), + time: git_ext::author::Time::new(timestamp.as_secs() as i64, 0), }; let commit = Commit::new::<_, _, OwnedTrailer>( oid, diff --git a/radicle/src/cob/thread.rs b/radicle/src/cob/thread.rs index 81bcf494..e1df7b4d 100644 --- a/radicle/src/cob/thread.rs +++ b/radicle/src/cob/thread.rs @@ -1,8 +1,8 @@ use std::cmp::Ordering; +use std::collections::{BTreeMap, BTreeSet}; use std::str::FromStr; use once_cell::sync::Lazy; -use radicle_crdt as crdt; use serde::{Deserialize, Serialize}; use thiserror::Error; @@ -11,9 +11,6 @@ use crate::cob::common::{Reaction, Timestamp}; use crate::cob::{ActorId, EntryId, Op}; use crate::prelude::ReadRepository; -use crdt::clock::Lamport; -use crdt::{GMap, GSet, LWWSet, Max, Redactable, Semilattice}; - /// Type name of a thread, as well as the domain for all thread operations. /// Note that threads are not usually used standalone. They are embeded into other COBs. pub static TYPENAME: Lazy = @@ -60,9 +57,9 @@ pub struct Comment { /// Comment author. author: ActorId, /// The comment body. - edits: GMap>, + edits: Vec, /// Reactions to this comment. - reactions: LWWSet<(ActorId, Reaction)>, + reactions: BTreeSet<(ActorId, Reaction)>, /// Comment this is a reply to. /// Should always be set, except for the root comment. reply_to: Option, @@ -80,8 +77,8 @@ impl Comment { Self { author, - reactions: LWWSet::default(), - edits: GMap::singleton(Lamport::initial(), Max::from(edit)), + reactions: BTreeSet::default(), + edits: vec![edit], reply_to, } } @@ -91,7 +88,7 @@ impl Comment { // SAFETY: There is always at least one edit. This is guaranteed by the [`Comment`] // constructor. #[allow(clippy::unwrap_used)] - self.edits.values().last().unwrap().get().body.as_str() + self.edits.last().unwrap().body.as_str() } /// Get the comment timestamp, which is the time of the *original* edit. To get the timestamp @@ -100,12 +97,7 @@ impl Comment { // SAFETY: There is always at least one edit. This is guaranteed by the [`Comment`] // constructor. #[allow(clippy::unwrap_used)] - self.edits - .first_key_value() - .map(|(_, v)| v) - .unwrap() - .get() - .timestamp + self.edits.first().unwrap().timestamp } /// Return the comment author. @@ -120,12 +112,12 @@ impl Comment { /// Return the ordered list of edits for this comment, including the original version. pub fn edits(&self) -> impl Iterator { - self.edits.values().map(Max::get) + self.edits.iter() } /// Add an edit. - pub fn edit(&mut self, clock: Lamport, body: String, timestamp: Timestamp) { - self.edits.insert(clock, Edit { body, timestamp }.into()) + pub fn edit(&mut self, body: String, timestamp: Timestamp) { + self.edits.push(Edit { body, timestamp }); } /// Comment reactions. @@ -182,23 +174,16 @@ impl From for nonempty::NonEmpty { #[derive(Debug, Default, Clone, PartialEq, Eq)] pub struct Thread { /// The comments under the thread. - comments: GMap>, + comments: BTreeMap>, /// Comment timeline. - timeline: GSet<(Lamport, CommentId)>, -} - -impl Semilattice for Thread { - fn merge(&mut self, other: Self) { - self.comments.merge(other.comments); - self.timeline.merge(other.timeline); - } + timeline: Vec, } impl Thread { pub fn new(id: CommentId, comment: Comment) -> Self { Self { - comments: GMap::singleton(id, Redactable::Present(comment)), - timeline: GSet::default(), + comments: BTreeMap::from_iter([(id, Some(comment))]), + timeline: Vec::default(), } } @@ -215,11 +200,7 @@ impl Thread { } pub fn comment(&self, id: &CommentId) -> Option<&Comment> { - if let Some(Redactable::Present(comment)) = self.comments.get(id) { - Some(comment) - } else { - None - } + self.comments.get(id).and_then(|o| o.as_ref()) } pub fn root(&self) -> (&CommentId, &Comment) { @@ -249,10 +230,10 @@ impl Thread { } pub fn comments(&self) -> impl DoubleEndedIterator + '_ { - self.timeline.iter().filter_map(|(_, id)| { + self.timeline.iter().filter_map(|id| { self.comments .get(id) - .and_then(Redactable::get) + .and_then(|o| o.as_ref()) .map(|comment| (id, comment)) }) } @@ -273,51 +254,46 @@ impl cob::store::FromHistory for Thread { Ok(()) } - fn apply( - &mut self, - ops: impl IntoIterator>, - _repo: &R, - ) -> Result<(), Error> { - for op in ops.into_iter() { - let id = op.id; - let author = op.author; - let timestamp = op.timestamp; + fn apply(&mut self, op: Op, _repo: &R) -> Result<(), Error> { + let id = op.id; + let author = op.author; + let timestamp = op.timestamp; - self.timeline.insert((op.clock, op.id)); + debug_assert!(!self.timeline.contains(&op.id)); - match op.action { + self.timeline.push(op.id); + + for action in op.into_iter() { + match action { Action::Comment { body, reply_to } => { if body.is_empty() { - return Err(Error::Comment(op.id)); + return Err(Error::Comment(id)); } // Nb. If a comment is already present, it must be redacted, because the // underlying store guarantees exactly-once delivery of ops. - self.comments.insert( - id, - Redactable::Present(Comment::new(author, body, reply_to, timestamp)), - ); + self.comments + .insert(id, Some(Comment::new(author, body, reply_to, timestamp))); } Action::Edit { id, body } => { if body.is_empty() { - return Err(Error::Edit(op.id)); + return Err(Error::Edit(id)); } // It's possible for a comment to be redacted before we're able to edit it, in // case of a concurrent update. // // However, it's *not* possible for the comment to be absent. Therefore we treat // that as an error. - if let Some(redactable) = self.comments.get_mut(&id) { - if let Redactable::Present(comment) = redactable { - comment.edit(op.clock, body, timestamp); + if let Some(comment) = self.comments.get_mut(&id) { + if let Some(comment) = comment { + comment.edit(body, timestamp); } } else { return Err(Error::Missing(id)); } } Action::Redact { id } => { - // Redactions must have observed a comment to be valid. if let Some(comment) = self.comments.get_mut(&id) { - comment.merge(Redactable::Redacted); + *comment = None; } else { return Err(Error::Missing(id)); } @@ -327,13 +303,13 @@ impl cob::store::FromHistory for Thread { reaction, active, } => { - let key = (op.author, reaction); - if let Some(redactable) = self.comments.get_mut(&to) { - if let Redactable::Present(comment) = redactable { + let key = (author, reaction); + if let Some(comment) = self.comments.get_mut(&to) { + if let Some(comment) = comment { if active { - comment.reactions.insert(key, op.clock); + comment.reactions.insert(key); } else { - comment.reactions.remove(key, op.clock); + comment.reactions.remove(&key); } } } else { @@ -348,14 +324,10 @@ impl cob::store::FromHistory for Thread { #[cfg(test)] mod tests { - use std::collections::BTreeSet; use std::ops::{Deref, DerefMut}; - use std::{array, iter}; use pretty_assertions::assert_eq; - use qcheck::{Arbitrary, TestResult}; - - use crdt::test::{assert_laws, WeightedGenerator}; + use qcheck_macros::quickcheck; use super::*; use crate as radicle; @@ -407,15 +379,6 @@ mod tests { body: body.to_owned(), }) } - - /// React to a comment. - pub fn react(&mut self, to: CommentId, reaction: Reaction, active: bool) -> Op { - self.op(Action::React { - to, - reaction, - active, - }) - } } impl Deref for Actor { @@ -432,111 +395,22 @@ mod tests { } } - #[derive(Clone)] - struct Changes { - permutations: [Vec>; N], - } - - impl std::fmt::Debug for Changes { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - for (i, p) in self.permutations.iter().enumerate() { - writeln!( - f, - "{i}: {:#?}", - p.iter().map(|c| &c.action).collect::>() - )?; - } - Ok(()) - } - } - - impl Arbitrary for Changes { - fn arbitrary(g: &mut qcheck::Gen) -> Self { - let rng = fastrand::Rng::with_seed(u64::arbitrary(g)); - let gen = WeightedGenerator::< - (Lamport, Op), - (Actor, Lamport, BTreeSet), - >::new(rng.clone()) - .variant(3, |(actor, clock, comments), rng| { - let comment = actor.comment( - iter::repeat_with(|| rng.alphabetic()) - .take(4) - .collect::() - .as_str(), - None, - ); - comments.insert(comment.id); - - Some((clock.tick(), comment)) - }) - .variant(2, |(actor, clock, comments), rng| { - if comments.is_empty() { - return None; - } - let id = *comments.iter().nth(rng.usize(..comments.len())).unwrap(); - let edit = actor.edit( - id, - iter::repeat_with(|| rng.alphabetic()) - .take(4) - .collect::() - .as_str(), - ); - Some((clock.tick(), edit)) - }) - .variant(2, |(actor, clock, comments), rng| { - if comments.is_empty() { - return None; - } - let to = *comments.iter().nth(rng.usize(..comments.len())).unwrap(); - let react = actor.react(to, Reaction::new('✨').unwrap(), rng.bool()); - - Some((clock.tick(), react)) - }) - .variant(2, |(actor, clock, comments), rng| { - if comments.is_empty() { - return None; - } - let id = *comments.iter().nth(rng.usize(..comments.len())).unwrap(); - comments.remove(&id); - let redact = actor.redact(id); - - Some((clock.tick(), redact)) - }); - - let mut ops = vec![Actor::::default().comment("Root", None)]; - let mut permutations: [Vec>; N] = array::from_fn(|_| Vec::new()); - - for (_, op) in gen.take(g.size()) { - ops.push(op); - } - - for p in &mut permutations { - *p = ops.clone(); - rng.shuffle(&mut ops); - } - - Changes { permutations } - } - } - #[test] fn test_redact_comment() { - let tmp = tempfile::tempdir().unwrap(); - let radicle::test::setup::Context { signer, .. } = radicle::test::setup::Context::new(&tmp); + let radicle::test::setup::Node { signer, .. } = radicle::test::setup::Node::default(); let repo = gen::(1); let mut alice = Actor::new(signer); - let mut thread = Thread::default(); let a0 = alice.comment("First comment", None); let a1 = alice.comment("Second comment", Some(a0.id())); let a2 = alice.comment("Third comment", Some(a0.id())); - thread.apply([a0, a1.clone(), a2], &repo).unwrap(); + let mut thread = Thread::from_ops([a0, a1.clone(), a2], &repo).unwrap(); assert_eq!(thread.comments().count(), 3); // Redact the second comment. let a3 = alice.redact(a1.id()); - thread.apply([a3], &repo).unwrap(); + thread.apply(a3, &repo).unwrap(); let (_, comment0) = thread.comments().nth(0).unwrap(); let (_, comment1) = thread.comments().nth(1).unwrap(); @@ -555,9 +429,7 @@ mod tests { let c1 = alice.edit(c0.id(), "Goodbye world."); let c2 = alice.edit(c0.id(), "Goodbye world!"); - let mut t1 = Thread::default(); - t1.apply([c0.clone(), c1.clone(), c2.clone()], &repo) - .unwrap(); + let t1 = Thread::from_ops([c0.clone(), c1, c2], &repo).unwrap(); let comment = t1.comment(&c0.id()); let edits = comment.unwrap().edits().collect::>(); @@ -566,11 +438,6 @@ mod tests { assert_eq!(edits[1].body.as_str(), "Goodbye world."); assert_eq!(edits[2].body.as_str(), "Goodbye world!"); assert_eq!(t1.comment(&c0.id()).unwrap().body(), "Goodbye world!"); - - let mut t2 = Thread::default(); - t2.apply([c0, c2, c1], &repo).unwrap(); // Apply in different order. - - assert_eq!(t1, t2); } #[test] @@ -579,12 +446,14 @@ mod tests { let bob = MockSigner::default(); let eve = MockSigner::default(); let repo = gen::(1); + let time = Timestamp::now(); let mut a = test::history::( &Action::Comment { body: "Thread root".to_owned(), reply_to: None, }, + time, &alice, ); a.comment("Alice comment", Some(a.root()), &alice); @@ -613,9 +482,9 @@ mod tests { assert_eq!(a, b); assert_eq!(b, e); - let (t1, _) = Thread::from_history(&a, &repo).unwrap(); - let (t2, _) = Thread::from_history(&b, &repo).unwrap(); - let (t3, _) = Thread::from_history(&e, &repo).unwrap(); + let t1 = Thread::from_history(&a, &repo).unwrap(); + let t2 = Thread::from_history(&b, &repo).unwrap(); + let t3 = Thread::from_history(&e, &repo).unwrap(); assert_eq!(t1, t2); assert_eq!(t2, t3); @@ -637,11 +506,6 @@ mod tests { vec!["Thread root", "Alice comment", "Eve comment", "Bob comment"] } ); - - for ops in a.permutations(2) { - let t = Thread::from_ops(ops, &repo).unwrap(); - assert_eq!(t, t1); - } } #[test] @@ -649,12 +513,14 @@ mod tests { let repo = gen::(1); let alice = MockSigner::default(); let bob = MockSigner::default(); + let time = Timestamp::now(); let mut a = test::history::( &Action::Comment { body: "Thread root".to_owned(), reply_to: None, }, + time, &alice, ); let mut b = a.clone(); @@ -664,7 +530,7 @@ mod tests { a.merge(b); - let (thread, _) = Thread::from_history(&a, &repo).unwrap(); + let thread = Thread::from_history(&a, &repo).unwrap(); assert_eq!(thread.comments().count(), 3); @@ -675,6 +541,65 @@ mod tests { assert_eq!(first.edits, second.edits); // despite the content being the same. } + #[quickcheck] + fn prop_ordering(timestamp: u64) { + let repo = gen::(1); + let alice = MockSigner::default(); + let bob = MockSigner::default(); + let timestamp = Timestamp::from_secs(timestamp); + + let h0 = test::history::( + &Action::Comment { + body: "Thread root".to_owned(), + reply_to: None, + }, + timestamp, + &alice, + ); + let mut h1 = h0.clone(); + let mut h2 = h0.clone(); + + let e1 = h1.commit( + &Action::Edit { + id: h0.root(), + body: String::from("Bye World."), + }, + &alice, + ); + let e2 = h2.commit( + &Action::Edit { + id: h0.root(), + body: String::from("Hi World."), + }, + &bob, + ); + + h1.merge(h2); + + let thread = Thread::from_history(&h1, &repo).unwrap(); + let (_, comment) = thread.comments().next().unwrap(); + + // E1 and E2 are concurrent, so the final edit will depend on which is the greater hash. + if e2 > e1 { + assert_eq!(comment.body(), "Hi World."); + } else { + assert_eq!(comment.body(), "Bye World."); + } + + let _e3 = h1.commit( + &Action::Edit { + id: h0.root(), + body: String::from("Hoho World!"), + }, + &alice, + ); + let thread = Thread::from_history(&h1, &repo).unwrap(); + let (_, comment) = thread.comments().next().unwrap(); + + // E3 is causally dependent on E1 and E2, so it always wins. + assert_eq!(comment.body(), "Hoho World!"); + } + #[test] fn test_comment_redact_missing() { let repo = gen::(1); @@ -682,7 +607,7 @@ mod tests { let mut t = Thread::default(); let id = arbitrary::entry_id(); - t.apply([alice.redact(id)], &repo).unwrap_err(); + t.apply(alice.redact(id), &repo).unwrap_err(); } #[test] @@ -692,54 +617,19 @@ mod tests { let mut t = Thread::default(); let id = arbitrary::entry_id(); - t.apply([alice.edit(id, "Edited")], &repo).unwrap_err(); + t.apply(alice.edit(id, "Edited"), &repo).unwrap_err(); } #[test] fn test_comment_edit_redacted() { let repo = gen::(1); let mut alice = Actor::::default(); - let mut t = Thread::default(); let a1 = alice.comment("Hi", None); let a2 = alice.redact(a1.id); let a3 = alice.edit(a1.id, "Edited"); - t.apply([a1, a2, a3], &repo).unwrap(); + let t = Thread::from_ops([a1, a2, a3], &repo).unwrap(); assert_eq!(t.comments().count(), 0); } - - #[test] - fn prop_invariants() { - fn property(repo: MockRepository, log: Changes<3>) -> TestResult { - let t = Thread::default(); - let [p1, p2, p3] = log.permutations; - - let mut t1 = t.clone(); - if t1.apply(p1, &repo).is_err() { - return TestResult::discard(); - } - - let mut t2 = t.clone(); - if t2.apply(p2, &repo).is_err() { - return TestResult::discard(); - } - - let mut t3 = t; - if t3.apply(p3, &repo).is_err() { - return TestResult::discard(); - } - - assert_eq!(t1, t2); - assert_eq!(t2, t3); - assert_laws(&t1, &t2, &t3); - - TestResult::passed() - } - qcheck::QuickCheck::new() - .min_tests_passed(100) - .max_tests(10000) - .gen(qcheck::Gen::new(7)) - .quickcheck(property as fn(MockRepository, Changes<3>) -> TestResult); - } } diff --git a/radicle/src/storage/git/transport/remote/mock.rs b/radicle/src/storage/git/transport/remote/mock.rs index b8b4d306..a7bbb12e 100644 --- a/radicle/src/storage/git/transport/remote/mock.rs +++ b/radicle/src/storage/git/transport/remote/mock.rs @@ -42,7 +42,19 @@ impl git2::transport::SmartSubtransport for MockTransport { url.node ))); }; + assert!( + storage.exists(), + "The storage path {} must exist", + storage.display() + ); + let git_dir = storage.join(url.repo.canonical()); + assert!( + git_dir.exists(), + "The repository {} must exist", + git_dir.display() + ); + let mut cmd = process::Command::new("git"); let mut child = cmd .arg("upload-pack") diff --git a/radicle/src/test.rs b/radicle/src/test.rs index 8144117b..50129a26 100644 --- a/radicle/src/test.rs +++ b/radicle/src/test.rs @@ -4,11 +4,72 @@ pub mod assert; pub mod fixtures; pub mod storage; -pub mod setup { - use tempfile::TempDir; +use super::storage::{Namespaces, RefUpdate}; +use crate::prelude::NodeId; +use crate::storage::WriteRepository; + +/// Perform a fetch between two local repositories. +/// This has the same outcome as doing a "real" fetch, but suffices for the simulation, and +/// doesn't require running nodes. +pub fn fetch( + repo: &W, + node: &NodeId, + namespaces: impl Into, +) -> Result, crate::storage::FetchError> { + let namespace = match namespaces.into() { + Namespaces::All => None, + Namespaces::Trusted(trusted) => trusted.into_iter().next(), + }; + let mut updates = Vec::new(); + let mut callbacks = git2::RemoteCallbacks::new(); + let mut opts = git2::FetchOptions::default(); + let refspec = if let Some(namespace) = namespace { + opts.prune(git2::FetchPrune::On); + format!("refs/namespaces/{namespace}/refs/*:refs/namespaces/{namespace}/refs/*") + } else { + opts.prune(git2::FetchPrune::Off); + "refs/namespaces/*:refs/namespaces/*".to_owned() + }; + + callbacks.update_tips(|name, old, new| { + if let Ok(name) = crate::git::RefString::try_from(name) { + if name.to_namespaced().is_some() { + updates.push(RefUpdate::from(name, old, new)); + // Returning `true` ensures the process is not aborted. + return true; + } + } + false + }); + opts.remote_callbacks(callbacks); + + let mut remote = repo.raw().remote_anonymous( + crate::storage::git::transport::remote::Url { + node: *node, + repo: repo.id(), + namespace, + } + .to_string() + .as_str(), + )?; + remote.fetch(&[refspec], Some(&mut opts), None)?; + + drop(opts); + + repo.set_identity_head()?; + repo.set_head()?; + repo.validate()?; + + Ok(updates) +} + +pub mod setup { + use std::path::{Path, PathBuf}; + + use super::storage::{Namespaces, RefUpdate}; use crate::crypto::test::signer::MockSigner; - use crate::prelude::*; + use crate::storage::git::transport::remote; use crate::{ git, profile::Home, @@ -16,49 +77,100 @@ pub mod setup { test::{fixtures, storage::git::Repository}, Storage, }; + use crate::{prelude::*, rad}; - #[derive(Debug)] - pub struct BranchWith { - pub base: git::Oid, - pub oid: git::Oid, - } - - pub struct Context { + /// A node. + /// + /// Note that this isn't a real node; only a profile with storage and a signing key. + pub struct Node { + pub root: PathBuf, pub storage: Storage, pub signer: MockSigner, - pub project: Repository, - pub working: git2::Repository, } - impl Context { - pub fn new(tmp: &TempDir) -> Self { + impl Default for Node { + fn default() -> Self { + let root = tempfile::tempdir().unwrap(); + + Self::new(root) + } + } + + impl Node { + pub fn new(root: impl AsRef) -> Self { + let root = root.as_ref().to_path_buf(); let mut rng = fastrand::Rng::new(); let signer = MockSigner::new(&mut rng); - let home = tmp.path().join("home"); + let home = root.join("home"); let paths = Home::new(home.as_path()).unwrap(); let storage = Storage::open(paths.storage()).unwrap(); - let (id, _, working, _) = - fixtures::project(tmp.path().join("copy"), &storage, &signer).unwrap(); - let project = storage.repository(id).unwrap(); + + remote::mock::register(signer.public_key(), storage.path()); Self { + root, storage, signer, - project, - working, } } - pub fn branch_with( + pub fn clone(&mut self, rid: Id, other: &Self) { + let repo = self.storage.create(rid).unwrap(); + super::fetch(&repo, other.signer.public_key(), Namespaces::All).unwrap(); + + rad::fork(rid, &self.signer, &self.storage).unwrap(); + } + + pub fn project(&self) -> NodeRepo { + let (id, _, checkout, _) = + fixtures::project(self.root.join("working"), &self.storage, &self.signer).unwrap(); + let repo = self.storage.repository(id).unwrap(); + let checkout = Some(NodeRepoCheckout { checkout }); + + NodeRepo { repo, checkout } + } + } + + /// A node repository with an optional checkout. + pub struct NodeRepo { + pub repo: Repository, + pub checkout: Option, + } + + impl NodeRepo { + pub fn fetch(&self, from: &Node) -> Vec { + super::fetch(&self.repo, from.signer.public_key(), Namespaces::All).unwrap() + } + + pub fn checkout(&self) -> &NodeRepoCheckout { + self.checkout.as_ref().unwrap() + } + } + + impl std::ops::Deref for NodeRepo { + type Target = Repository; + + fn deref(&self) -> &Self::Target { + &self.repo + } + } + + /// A repository checkout. + pub struct NodeRepoCheckout { + checkout: git::raw::Repository, + } + + impl NodeRepoCheckout { + pub fn branch_with, T: AsRef<[u8]>>( &self, - blobs: impl IntoIterator)>, + blobs: impl IntoIterator, ) -> BranchWith { let refname = git::Qualified::from(git::lit::refs_heads(git::refname!("master"))); - let base = self.working.refname_to_id(refname.as_str()).unwrap(); - let parent = self.working.find_commit(base).unwrap(); - let oid = commit(&self.working, &refname, blobs, &[&parent]); + let base = self.checkout.refname_to_id(refname.as_str()).unwrap(); + let parent = self.checkout.find_commit(base).unwrap(); + let oid = commit(&self.checkout, &refname, blobs, &[&parent]); - git::push(&self.working, &REMOTE_NAME, [(&refname, &refname)]).unwrap(); + git::push(&self.checkout, &REMOTE_NAME, [(&refname, &refname)]).unwrap(); BranchWith { base: base.into(), @@ -67,37 +179,116 @@ pub mod setup { } } - pub fn initial_blobs() -> Vec<(String, Vec)> { - vec![ - ("README.md".to_string(), b"Hello, World!".to_vec()), - ( - "CONTRIBUTING".to_string(), - b"Please follow the rules".to_vec(), - ), - ] + impl std::ops::Deref for NodeRepoCheckout { + type Target = git::raw::Repository; + + fn deref(&self) -> &Self::Target { + &self.checkout + } } - pub fn update_blobs() -> Vec<(String, Vec)> { - vec![ - ("README.md".to_string(), b"Hello, Radicle!".to_vec()), - ( - "CONTRIBUTING".to_string(), - b"Please follow the rules".to_vec(), - ), - ] + /// A node with a repository. + pub struct NodeWithRepo { + pub node: Node, + pub repo: NodeRepo, } - pub fn commit( + impl std::ops::Deref for NodeWithRepo { + type Target = Node; + + fn deref(&self) -> &Self::Target { + &self.node + } + } + + impl std::ops::DerefMut for NodeWithRepo { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.node + } + } + + impl Default for NodeWithRepo { + fn default() -> Self { + let node = Node::default(); + let repo = node.project(); + + Self { node, repo } + } + } + + /// A network of three nodes. + /// + /// Note that these are not actually running nodes in the sense of `radicle-node`. + /// These are simply profiles with their own storage, and the ability to fetch between + /// them. + pub struct Network { + pub alice: NodeWithRepo, + pub bob: NodeWithRepo, + pub eve: NodeWithRepo, + pub rid: Id, + + #[allow(dead_code)] + tmp: tempfile::TempDir, + } + + impl Default for Network { + fn default() -> Self { + let tmp = tempfile::tempdir().unwrap(); + let alice = Node::new(tmp.path().join("alice")); + let mut bob = Node::new(tmp.path().join("bob")); + let mut eve = Node::new(tmp.path().join("eve")); + let repo = alice.project(); + let rid = repo.id; + + bob.clone(repo.id, &alice); + eve.clone(repo.id, &alice); + + let alice = NodeWithRepo { node: alice, repo }; + let repo = bob.storage.repository(rid).unwrap(); + let bob = NodeWithRepo { + node: bob, + repo: NodeRepo { + repo, + checkout: None, + }, + }; + let repo = eve.storage.repository(rid).unwrap(); + let eve = NodeWithRepo { + node: eve, + repo: NodeRepo { + repo, + checkout: None, + }, + }; + + Self { + alice, + bob, + eve, + rid, + tmp, + } + } + } + + #[derive(Debug)] + pub struct BranchWith { + pub base: git::Oid, + pub oid: git::Oid, + } + + pub fn commit, T: AsRef<[u8]>>( repo: &git2::Repository, refname: &git::Qualified, - blobs: impl IntoIterator)>, + blobs: impl IntoIterator, parents: &[&git2::Commit<'_>], ) -> git::Oid { let tree = { let mut tb = repo.treebuilder(None).unwrap(); for (name, blob) in blobs.into_iter() { - let oid = repo.blob(&blob).unwrap(); - tb.insert(name, oid, git2::FileMode::Blob.into()).unwrap(); + let oid = repo.blob(blob.as_ref()).unwrap(); + tb.insert(name.as_ref(), oid, git2::FileMode::Blob.into()) + .unwrap(); } tb.write().unwrap() }; @@ -108,7 +299,7 @@ pub mod setup { Some(refname.as_str()), &author, &author, - "test commit", + "Making changes", &tree, parents, )