diff --git a/node/src/protocol.rs b/node/src/protocol.rs index a0de6df5..a78efad5 100644 --- a/node/src/protocol.rs +++ b/node/src/protocol.rs @@ -51,7 +51,13 @@ pub type Timestamp = u64; /// A protocol event. #[derive(Debug, Clone)] -pub enum Event {} +pub enum Event { + RefsFetched { + from: Url, + project: Id, + updated: Vec, + }, +} /// Error returned by [`Command::Fetch`]. #[derive(thiserror::Error, Debug)] @@ -221,6 +227,11 @@ impl<'r, T: WriteStorage<'r>, S: address_book::Store, G: crypto::Signer> Protoco &self.context.storage } + /// Get the mutable storage instance. + pub fn storage_mut(&mut self) -> &mut T { + &mut self.context.storage + } + /// Get the local protocol time. pub fn local_time(&self) -> LocalTime { self.context.clock.local_time() diff --git a/node/src/protocol/peer.rs b/node/src/protocol/peer.rs index b1e24296..ea6888ef 100644 --- a/node/src/protocol/peer.rs +++ b/node/src/protocol/peer.rs @@ -201,8 +201,16 @@ impl Peer { // TODO: Check that we're tracking this user as well. if ctx.config.is_tracking(&id) { // TODO: Check refs to see if we should try to fetch or not. - let updated = ctx.fetch(&id, git); - if !updated.is_empty() { + let updated_refs = ctx.fetch(&id, git); + let is_updated = !updated_refs.is_empty(); + + ctx.io.push_back(Io::Event(Event::RefsFetched { + from: git.clone(), + project: id.clone(), + updated: updated_refs, + })); + + if is_updated { return Ok(Some(Message::RefsUpdate { id, signer, diff --git a/node/src/test/peer.rs b/node/src/test/peer.rs index 023e37c3..c95c20b2 100644 --- a/node/src/test/peer.rs +++ b/node/src/test/peer.rs @@ -204,7 +204,13 @@ where msgs.into_iter() } - /// Get a draining iterator over the peers's I/O outbox. + /// Get a draining iterator over the peer's emitted events. + pub fn events(&mut self) -> impl Iterator + '_ { + self.outbox() + .filter_map(|io| if let Io::Event(e) = io { Some(e) } else { None }) + } + + /// Get a draining iterator over the peer's I/O outbox. pub fn outbox(&mut self) -> impl Iterator> + '_ { self.protocol.outbox().drain(..) } diff --git a/node/src/test/tests.rs b/node/src/test/tests.rs index a33a0b55..4ad9ef70 100644 --- a/node/src/test/tests.rs +++ b/node/src/test/tests.rs @@ -368,6 +368,80 @@ fn test_persistent_peer_reconnect() { assert_matches!(alice.outbox().next(), None); } +#[test] +fn test_push_and_pull() { + logger::init(log::Level::Debug); + + let tempdir = tempfile::tempdir().unwrap(); + + let storage_alice = Storage::open( + tempdir.path().join("alice").join("storage"), + MockSigner::default(), + ) + .unwrap(); + let repo = fixtures::repository(tempdir.path().join("working")); + let mut alice = Peer::new("alice", [7, 7, 7, 7], storage_alice); + + let storage_bob = Storage::open( + tempdir.path().join("bob").join("storage"), + MockSigner::default(), + ) + .unwrap(); + let mut bob = Peer::new("bob", [8, 8, 8, 8], storage_bob); + + let storage_eve = Storage::open( + tempdir.path().join("eve").join("storage"), + MockSigner::default(), + ) + .unwrap(); + let mut eve = Peer::new("eve", [9, 9, 9, 9], storage_eve); + + // Alice and Bob connect to Eve. + alice.command(protocol::Command::Connect(eve.addr())); + bob.command(protocol::Command::Connect(eve.addr())); + + let mut sim = Simulation::new( + LocalTime::now(), + alice.rng.clone(), + simulator::Options::default(), + ) + .initialize([&mut alice, &mut bob, &mut eve]); + + // Here we expect Alice to connect to Eve. + sim.run_while([&mut alice, &mut bob, &mut eve], |s| !s.is_settled()); + + // Alice creates a new project. + let (proj_id, _) = rad::init( + &repo, + "alice", + "alice's repo", + storage::BranchName::from("master"), + alice.storage_mut(), + ) + .unwrap(); + + // Bob tracks Alice's project. + let (sender, _) = chan::bounded(1); + bob.command(protocol::Command::Track(proj_id.clone(), sender)); + + // Eve tracks Alice's project. + let (sender, _) = chan::bounded(1); + eve.command(protocol::Command::Track(proj_id.clone(), sender)); + + // Neither of them have it in the beginning. + assert!(eve.storage().get(&proj_id).unwrap().is_none()); + assert!(bob.storage().get(&proj_id).unwrap().is_none()); + + // Alice announces her refs. + // We now expect Eve to fetch Alice's project from Alice. + // Then we expect Bob to fetch Alice's project from Eve. + // TODO: Check that Bob is fetching from Eve and not Alice, via an event. + alice.command(protocol::Command::AnnounceRefsUpdate(proj_id.clone())); + sim.run_while([&mut alice, &mut bob, &mut eve], |s| !s.is_settled()); + assert!(eve.storage().get(&proj_id).unwrap().is_some()); + assert!(bob.storage().get(&proj_id).unwrap().is_some()); +} + #[test] fn prop_inventory_exchange_dense() { fn property(alice_inv: MockStorage, bob_inv: MockStorage, eve_inv: MockStorage) {