mab122
/
radicle-heartwood-lfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

cob: Test deterministic traversal order 

We test to make sure that in the event of concurrent operations,
traversal order is deterministic.

This required some changes to the testing infrastructure. We're also
able to simplify the `Actor` type that now has redundant functionality.
This commit is contained in:
Alexis Sellier 2023-05-26 18:05:22 +02:00
parent 0e14dacb99
commit b60569d3cb
No known key found for this signature in database
6 changed files with 207 additions and 211 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
radicle-cob/src/backend/git/change.rs
View File

@ -126,6 +126,7 @@ impl change::Storage for git2::Repository {
signature.clone(),
tree,
)?;
Ok(Change {
id,
revision: revision.into(),

18
radicle-cob/src/history.rs
View File

@ -102,6 +102,7 @@ impl History {
.fold(&self.root, init, |acc, k, v, _| f(acc, k, v))
}
/// Return a topologically-sorted list of history entries.
pub fn sorted<F>(&self, compare: F) -> impl Iterator<Item = &Entry>
where
F: FnMut(&EntryId, &EntryId) -> Ordering,
@ -113,6 +114,7 @@ impl History {
.map(|node| &node.value)
}
/// Extend this history with a new entry.
pub fn extend<Id>(
&mut self,
new_id: Id,
@ -140,7 +142,23 @@ impl History {
}
}
/// Merge two histories.
pub fn merge(&mut self, other: Self) {
self.graph.merge(other.graph);
}
/// Get the number of history entries.
pub fn len(&self) -> usize {
self.graph.len()
}
/// Check if the graph is empty.
pub fn is_empty(&self) -> bool {
self.graph.is_empty()
}
/// Get the root entry.
pub fn root(&self) -> EntryId {
self.root
}
}

17
radicle/src/cob/patch.rs
View File

@ -1349,12 +1349,7 @@ mod test {
impl<const N: usize> Arbitrary for Changes<N> {
fn arbitrary(g: &mut qcheck::Gen) -> Self {
type State = (
Actor<MockSigner, Action>,
clock::Lamport,
Vec<EntryId>,
Vec<Tag>,
);
type State = (Actor<MockSigner>, clock::Lamport, Vec<EntryId>, Vec<Tag>);
let rng = fastrand::Rng::with_seed(u64::arbitrary(g));
let oids = iter::repeat_with(|| {
@ -1607,8 +1602,8 @@ mod test {
let base = git::Oid::from_str("d8711a8d43dc919fe39ae4b7c2f7b24667f5d470").unwrap();
let commit = git::Oid::from_str("cb18e95ada2bb38aadd8e6cef0963ce37a87add3").unwrap();
let mut alice = Actor::<MockSigner, _>::default();
let mut bob = Actor::<MockSigner, _>::default();
let mut alice = Actor::<MockSigner>::default();
let mut bob = Actor::<MockSigner>::default();
let proj = gen::<Project>(1);
let doc = Doc::new(proj, nonempty![alice.did(), bob.did()], 1)
@ -1696,7 +1691,7 @@ mod test {
fn test_revision_redacted() {
let base = git::Oid::from_str("cb18e95ada2bb38aadd8e6cef0963ce37a87add3").unwrap();
let oid = git::Oid::from_str("518d5069f94c03427f694bb494ac1cd7d1339380").unwrap();
let mut alice = Actor::<_, Action>::new(MockSigner::default());
let mut alice = Actor::new(MockSigner::default());
let mut patch = Patch::default();
let repo = gen::<MockRepository>(1);
@ -1732,7 +1727,7 @@ mod test {
let base = git::Oid::from_str("cb18e95ada2bb38aadd8e6cef0963ce37a87add3").unwrap();
let oid = git::Oid::from_str("518d5069f94c03427f694bb494ac1cd7d1339380").unwrap();
let repo = gen::<MockRepository>(1);
let mut alice = Actor::<_, Action>::new(MockSigner::default());
let mut alice = Actor::new(MockSigner::default());
let mut p1 = Patch::default();
let mut p2 = Patch::default();
@ -1755,7 +1750,7 @@ mod test {
let base = git::Oid::from_str("cb18e95ada2bb38aadd8e6cef0963ce37a87add3").unwrap();
let oid = git::Oid::from_str("518d5069f94c03427f694bb494ac1cd7d1339380").unwrap();
let id = gen::<Id>(1);
let mut alice = Actor::<_, Action>::new(MockSigner::default());
let mut alice = Actor::new(MockSigner::default());
let mut doc = gen::<Doc<Verified>>(1);
doc.delegates.push(alice.signer.public_key().into());
let repo = MockRepository::new(id, doc);

1
radicle/src/cob/store.rs
View File

@ -53,6 +53,7 @@ pub trait FromHistory: Sized + Default + PartialEq {
history: &History,
repo: &R,
) -> Result<(Self, Lamport), Self::Error> {
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) {

163
radicle/src/cob/test.rs
View File

@ -1,4 +1,4 @@
use std::collections::{BTreeMap, BTreeSet};
use std::collections::BTreeSet;
use std::marker::PhantomData;
use std::ops::Deref;
@ -10,64 +10,96 @@ use crate::cob::op::{Op, Ops};
use crate::cob::patch;
use crate::cob::patch::Patch;
use crate::cob::store::encoding;
use crate::cob::History;
use crate::crypto::{PublicKey, Signer};
use crate::cob::{EntryId, History};
use crate::crypto::Signer;
use crate::git;
use crate::git::ext::author::Author;
use crate::git::ext::commit::headers::Headers;
use crate::git::ext::commit::{trailers::OwnedTrailer, Commit};
use crate::git::Oid;
use crate::prelude::Did;
use crate::storage::ReadRepository;
use crate::test::arbitrary;
use super::store::FromHistory;
use super::thread;
/// Convenience type for building histories.
#[derive(Debug, Clone)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HistoryBuilder<T> {
history: History,
resource: Oid,
witness: PhantomData<T>,
}
impl<T> AsRef<History> for HistoryBuilder<T> {
fn as_ref(&self) -> &History {
&self.history
}
}
impl HistoryBuilder<thread::Thread> {
pub fn comment<G: Signer>(
&mut self,
body: impl ToString,
reply_to: Option<thread::CommentId>,
signer: &G,
) -> Oid {
let action = thread::Action::Comment {
body: body.to_string(),
reply_to,
};
self.commit(&action, signer)
}
}
impl<T: FromHistory> HistoryBuilder<T>
where
T::Action: Serialize + Eq + 'static,
{
pub fn new(op: &Op<T::Action>) -> HistoryBuilder<T> {
let entry = arbitrary::oid();
pub fn new<G: Signer>(action: &T::Action, signer: &G) -> HistoryBuilder<T> {
let resource = arbitrary::oid();
let data = encoding::encode(&op.action).unwrap();
let timestamp = clock::Physical::now().as_secs();
let (data, root) = encoded::<T, _>(action, timestamp as i64, [], signer);
Self {
history: History::new_from_root(
entry,
op.author,
root,
*signer.public_key(),
resource,
NonEmpty::new(data),
op.timestamp.as_secs(),
timestamp,
),
resource,
witness: PhantomData,
}
}
pub fn append(&mut self, op: &Op<T::Action>) -> &mut Self {
let data = encoding::encode(&op.action).unwrap();
self.history.extend(
arbitrary::oid(),
op.author,
self.resource,
NonEmpty::new(data),
op.timestamp.as_secs(),
);
self
pub fn root(&self) -> EntryId {
self.history.root()
}
pub fn merge(&mut self, other: Self) {
self.history.merge(other.history);
}
pub fn commit<G: Signer>(&mut self, action: &T::Action, signer: &G) -> git::ext::Oid {
let timestamp = clock::Physical::now().as_secs();
let tips = self.tips();
let (data, oid) = encoded::<T, _>(action, timestamp as i64, tips, signer);
self.history.extend(
oid,
*signer.public_key(),
self.resource,
NonEmpty::new(data),
timestamp,
);
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<Op<T::Action>> {
self.history
.sorted(|a, b| if rng.bool() { a.cmp(b) } else { b.cmp(a) })
@ -99,61 +131,45 @@ impl<A> Deref for HistoryBuilder<A> {
}
/// Create a new test history.
pub fn history<T: FromHistory>(op: &Op<T::Action>) -> HistoryBuilder<T>
pub fn history<T: FromHistory, G: Signer>(action: &T::Action, signer: &G) -> HistoryBuilder<T>
where
T::Action: Serialize + Eq + 'static,
{
HistoryBuilder::new(op)
HistoryBuilder::new(action, signer)
}
/// An object that can be used to create and sign operations.
pub struct Actor<G, A> {
pub struct Actor<G> {
pub signer: G,
pub clock: clock::Lamport,
pub ops: BTreeMap<(clock::Lamport, PublicKey), Op<A>>,
}
impl<G: Default, A> Default for Actor<G, A> {
impl<G: Default> Default for Actor<G> {
fn default() -> Self {
Self::new(G::default())
}
}
impl<G, A> Actor<G, A> {
impl<G> Actor<G> {
pub fn new(signer: G) -> Self {
Self {
signer,
clock: clock::Lamport::default(),
ops: BTreeMap::default(),
}
}
}
impl<G: Signer, A: Clone + Serialize> Actor<G, A> {
pub fn receive(&mut self, ops: impl IntoIterator<Item = Op<A>>) -> clock::Lamport {
for op in ops {
let clock = op.clock;
self.ops.insert((clock, op.author), op);
self.clock.merge(clock);
}
self.clock
}
/// Reset actor state to initial state.
pub fn reset(&mut self) {
self.ops.clear();
self.clock = clock::Lamport::default();
}
/// Returned an ordered list of events.
pub fn timeline(&self) -> impl Iterator<Item = &Op<A>> {
self.ops.values()
}
impl<G: Signer> Actor<G> {
/// Create a new operation.
pub fn op_with(&mut self, action: A, clock: clock::Lamport, identity: Oid) -> Op<A> {
let id = arbitrary::oid().into();
pub fn op_with<A: Clone + Serialize>(
&mut self,
action: A,
clock: clock::Lamport,
identity: Oid,
) -> Op<A> {
let data = encoding::encode(&action).unwrap();
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();
@ -168,14 +184,11 @@ impl<G: Signer, A: Clone + Serialize> Actor<G, A> {
}
/// Create a new operation.
pub fn op(&mut self, action: A) -> Op<A> {
pub fn op<A: Clone + Serialize>(&mut self, action: A) -> Op<A> {
let clock = self.clock.tick();
let identity = arbitrary::oid();
let op = self.op_with(action, clock, identity);
self.ops.insert((self.clock, op.author), op.clone());
op
self.op_with(action, clock, identity)
}
/// Get the actor's DID.
@ -184,7 +197,7 @@ impl<G: Signer, A: Clone + Serialize> Actor<G, A> {
}
}
impl<G: Signer> Actor<G, super::patch::Action> {
impl<G: Signer> Actor<G> {
/// Create a patch.
pub fn patch<R: ReadRepository>(
&mut self,
@ -210,3 +223,37 @@ impl<G: Signer> Actor<G, super::patch::Action> {
)
}
}
/// Encode an action and return its hash.
///
/// Doesn't encode in the same way as we do in production, but attempts to include the same data
/// that feeds into the hash entropy, so that changing any input will change the resulting oid.
pub fn encoded<T: FromHistory, G: Signer>(
action: &T::Action,
timestamp: i64,
parents: impl IntoIterator<Item = Oid>,
signer: &G,
) -> (Vec<u8>, git::ext::Oid) {
let data = encoding::encode(action).unwrap();
let oid = git::raw::Oid::hash_object(git::raw::ObjectType::Blob, &data).unwrap();
let parents = parents.into_iter().map(|o| *o);
let author = Author {
name: "radicle".to_owned(),
email: signer.public_key().to_human(),
time: git_ext::author::Time::new(timestamp, 0),
};
let commit = Commit::new::<_, _, OwnedTrailer>(
oid,
parents,
author.clone(),
author,
Headers::new(),
String::default(),
[],
)
.to_string();
let hash = git::raw::Oid::hash_object(git::raw::ObjectType::Commit, commit.as_bytes()).unwrap();
(data, hash.into())
}

218
radicle/src/cob/thread.rs
View File

@ -353,7 +353,7 @@ mod tests {
/// An object that can be used to create and sign changes.
pub struct Actor<G> {
inner: cob::test::Actor<G, Action>,
inner: cob::test::Actor<G>,
}
impl<G: Default + Signer> Default for Actor<G> {
@ -403,7 +403,7 @@ mod tests {
}
impl<G> Deref for Actor<G> {
type Target = cob::test::Actor<G, Action>;
type Target = cob::test::Actor<G>;
fn deref(&self) -> &Self::Target {
&self.inner
@ -559,173 +559,107 @@ mod tests {
}
#[test]
fn test_timelines_basic() {
let mut alice = Actor::<MockSigner>::default();
let mut bob = Actor::<MockSigner>::default();
let a0 = alice.comment("Thread root", None);
let a1 = alice.comment("First comment", Some(a0.id()));
let a2 = alice.comment("Second comment", Some(a0.id()));
bob.receive([a0.clone(), a1.clone(), a2.clone()]);
assert_eq!(
bob.timeline().collect::<Vec<_>>(),
alice.timeline().collect::<Vec<_>>()
);
assert_eq!(alice.timeline().collect::<Vec<_>>(), vec![&a0, &a1, &a2]);
bob.reset();
bob.receive([a0, a2, a1]);
assert_eq!(
bob.timeline().collect::<Vec<_>>(),
alice.timeline().collect::<Vec<_>>()
);
}
#[test]
fn test_timelines_concurrent() {
let mut alice = Actor::<MockSigner>::default();
let mut bob = Actor::<MockSigner>::default();
let mut eve = Actor::<MockSigner>::default();
let a0 = alice.comment("Thread root", None);
let a1 = alice.comment("First comment", Some(a0.id()));
bob.receive([a0.clone(), a1.clone()]);
let b0 = bob.comment("Bob's first reply to Alice", Some(a0.id()));
let b1 = bob.comment("Bob's second reply to Alice", Some(a0.id()));
eve.receive([a0.clone(), b1.clone(), b0.clone()]);
let e0 = eve.comment("Eve's first reply to Alice", Some(a0.id()));
bob.receive([e0.clone()]);
let b2 = bob.comment("Bob's third reply to Alice", Some(a0.id()));
eve.receive([b2.clone(), a1.clone()]);
let e1 = eve.comment("Eve's second reply to Alice", Some(a0.id()));
alice.receive([b0.clone(), b1.clone(), b2.clone(), e0.clone(), e1.clone()]);
bob.receive([e1.clone()]);
let a2 = alice.comment("Second comment", Some(a0.id()));
eve.receive([a2.clone()]);
bob.receive([a2.clone()]);
assert_eq!(alice.ops.len(), 8);
assert_eq!(bob.ops.len(), 8);
assert_eq!(eve.ops.len(), 8);
assert_eq!(
bob.timeline().collect::<Vec<_>>(),
alice.timeline().collect::<Vec<_>>()
);
assert_eq!(
eve.timeline().collect::<Vec<_>>(),
alice.timeline().collect::<Vec<_>>()
);
assert_eq!(
vec![&a0, &a1, &b0, &b1, &e0, &b2, &e1, &a2],
alice.timeline().collect::<Vec<_>>(),
);
}
#[test]
fn test_histories() {
fn test_timeline() {
let alice = MockSigner::default();
let bob = MockSigner::default();
let eve = MockSigner::default();
let repo = gen::<MockRepository>(1);
let mut alice = Actor::<MockSigner>::default();
let mut bob = Actor::<MockSigner>::default();
let mut eve = Actor::<MockSigner>::default();
let mut a = test::history::<Thread, _>(
&Action::Comment {
body: "Thread root".to_owned(),
reply_to: None,
},
&alice,
);
a.comment("Alice comment", Some(a.root()), &alice);
let a0 = alice.comment("Alice's comment", None);
let b0 = bob.comment("Bob's reply", Some(a0.id())); // Bob and Eve's replies are concurrent.
let e0 = eve.comment("Eve's reply", Some(a0.id()));
let mut a = test::history::<Thread>(&a0);
let mut b = a.clone();
let b1 = b.comment("Bob comment", Some(a.root()), &bob);
let mut e = a.clone();
let e1 = e.comment("Eve comment", Some(a.root()), &eve);
b.append(&b0);
e.append(&e0);
assert_eq!(a.as_ref().len(), 2);
assert_eq!(b.as_ref().len(), 3);
assert_eq!(e.as_ref().len(), 3);
a.merge(b);
a.merge(e);
a.merge(b.clone());
a.merge(e.clone());
let (expected, _) = Thread::from_history(&a, &repo).unwrap();
for permutation in a.permutations(2) {
let actual = Thread::from_ops(permutation, &repo).unwrap();
assert_eq!(actual, expected);
assert_eq!(a.as_ref().len(), 4);
b.merge(a.clone());
b.merge(e.clone());
e.merge(a.clone());
e.merge(b.clone());
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();
assert_eq!(t1, t2);
assert_eq!(t2, t3);
let timeline1 = t1.comments().collect::<Vec<_>>();
let timeline2 = t2.comments().collect::<Vec<_>>();
let timeline3 = t3.comments().collect::<Vec<_>>();
assert_eq!(timeline1, timeline2);
assert_eq!(timeline2, timeline3);
assert_eq!(timeline1.len(), 4);
assert_eq!(
timeline1.iter().map(|(_, c)| c.body()).collect::<Vec<_>>(),
// Since the operations are concurrent, the ordering depends on the ordering between
// the operation ids.
if e1 > b1 {
vec!["Thread root", "Alice comment", "Bob comment", "Eve comment"]
} else {
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]
fn test_duplicate_comments() {
let repo = gen::<MockRepository>(1);
let alice = MockSigner::default();
let bob = MockSigner::default();
let mut alice = Actor::<MockSigner>::default();
let mut bob = Actor::<MockSigner>::default();
let a0 = alice.comment("Hello World!", None);
let b0 = bob.comment("Hello World!", None);
let mut a = test::history::<Thread>(&a0);
let mut a = test::history::<Thread, _>(
&Action::Comment {
body: "Thread root".to_owned(),
reply_to: None,
},
&alice,
);
let mut b = a.clone();
b.append(&b0);
a.comment("Hello World!", None, &alice);
b.comment("Hello World!", None, &bob);
a.merge(b);
let (thread, _) = Thread::from_history(&a, &repo).unwrap();
assert_eq!(thread.comments().count(), 2);
assert_eq!(thread.comments().count(), 3);
let (first_id, first) = thread.comments().nth(0).unwrap();
let (second_id, second) = thread.comments().nth(1).unwrap();
let (first_id, first) = thread.comments().nth(1).unwrap();
let (second_id, second) = thread.comments().nth(2).unwrap();
assert!(first_id != second_id); // The ids are not the same,
assert_eq!(first.edits, second.edits); // despite the content being the same.
}
#[test]
fn test_duplicate_comments_same_author() {
let repo = gen::<MockRepository>(1);
let mut alice = Actor::<MockSigner>::default();
let a0 = alice.comment("Hello World!", None);
let a1 = alice.comment("Hello World!", None);
let a2 = alice.comment("Hello World!", None);
// These simulate two devices sharing the same key.
let mut h1 = test::history::<Thread>(&a0);
let mut h2 = h1.clone();
let mut h3 = h1.clone();
// Alice writes the same comment on both devices, not realizing what she has done.
h1.append(&a1);
h2.append(&a2);
// Eventually the histories are merged by a third party.
h3.merge(h1);
h3.merge(h2);
let (thread, _) = Thread::from_history(&h3, &repo).unwrap();
// The three comments, distinct yet identical in terms of content, are preserved.
assert_eq!(thread.comments().count(), 3);
let (first_id, first) = thread.comments().nth(0).unwrap();
let (second_id, second) = thread.comments().nth(1).unwrap();
let (third_id, third) = thread.comments().nth(2).unwrap();
// Their IDs are not the same.
assert!(first_id != second_id);
assert!(second_id != third_id);
// Their content are the same.
assert_eq!(first, second);
assert_eq!(second, third);
}
#[test]
fn test_comment_edit_reinsert() {
let repo = gen::<MockRepository>(1);