Integrate replication manager with networking stack

aquadoggo/Cargo.toml

@@ -70,6 +70,7 @@ tokio = { version = "1.25.0", features = [
     "sync",
     "time",
 ] }
+tokio-stream = { version = "0.1.14", features = ["sync"] }
 tower-http = { version = "0.3.4", default-features = false, features = [
     "cors",
 ] }

aquadoggo/src/bus.rs

@@ -1,8 +1,10 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+use libp2p::PeerId;
 use p2panda_rs::operation::OperationId;
 
 use crate::manager::Sender;
+use crate::replication::SyncMessage;
 
 /// Sender for cross-service communication bus.
 pub type ServiceSender = Sender<ServiceMessage>;
@@ -12,4 +14,16 @@ pub type ServiceSender = Sender<ServiceMessage>;
 pub enum ServiceMessage {
     /// A new operation arrived at the node.
     NewOperation(OperationId),
+
+    /// Node established a bi-directional connection to another node.
+    ConnectionEstablished(PeerId),
+
+    /// Node closed a connection to another node.
+    ConnectionClosed(PeerId),
+
+    /// Node sent a message to remote node for replication.
+    SentReplicationMessage(PeerId, SyncMessage),
+
+    /// Node received a message from remote node for replication.
+    ReceivedReplicationMessage(PeerId, SyncMessage),
 }

aquadoggo/src/config.rs

@@ -98,6 +98,10 @@ impl Configuration {
             }
         };
 
+        // Derive peer id from key pair
+        let key_pair = NetworkConfiguration::load_or_generate_key_pair(config.base_path.clone())?;
+        config.network.set_peer_id(&key_pair.public());
+
         Ok(config)
     }
 }

aquadoggo/src/materializer/service.rs

@@ -88,28 +88,30 @@ pub async fn materializer_service(
 
     // Listen to incoming new entries and operations and move them into task queue
     let handle = task::spawn(async move {
-        while let Ok(ServiceMessage::NewOperation(operation_id)) = rx.recv().await {
-            // Resolve document id of regarding operation
-            match context
-                .store
-                .get_document_id_by_operation_id(&operation_id)
-                .await
-                .unwrap_or_else(|_| {
-                    panic!(
-                        "Failed database query when retreiving document for operation_id {}",
-                        operation_id
-                    )
-                }) {
-                Some(document_id) => {
-                    // Dispatch "reduce" task which will materialize the regarding document
-                    factory.queue(Task::new("reduce", TaskInput::new(Some(document_id), None)));
-                }
-                None => {
-                    // Panic when we couldn't find the regarding document in the database. We can
-                    // safely assure that this is due to a critical bug affecting the database
-                    // integrity. Panicking here will close `handle` and by that signal a node
-                    // shutdown.
-                    panic!("Could not find document for operation_id {}", operation_id);
+        loop {
+            if let Ok(ServiceMessage::NewOperation(operation_id)) = rx.recv().await {
+                // Resolve document id of regarding operation
+                match context
+                    .store
+                    .get_document_id_by_operation_id(&operation_id)
+                    .await
+                    .unwrap_or_else(|_| {
+                        panic!(
+                            "Failed database query when retreiving document for operation_id {}",
+                            operation_id
+                        )
+                    }) {
+                    Some(document_id) => {
+                        // Dispatch "reduce" task which will materialize the regarding document
+                        factory.queue(Task::new("reduce", TaskInput::new(Some(document_id), None)));
+                    }
+                    None => {
+                        // Panic when we couldn't find the regarding document in the database. We can
+                        // safely assure that this is due to a critical bug affecting the database
+                        // integrity. Panicking here will close `handle` and by that signal a node
+                        // shutdown.
+                        panic!("Could not find document for operation_id {}", operation_id);
+                    }
                 }
             }
         }

aquadoggo/src/network/behaviour.rs

@@ -8,6 +8,7 @@ use libp2p::{autonat, connection_limits, identify, mdns, ping, relay, rendezvous
 use log::debug;
 
 use crate::network::config::NODE_NAMESPACE;
+use crate::network::replication;
 use crate::network::NetworkConfiguration;
 
 /// Network behaviour for the aquadoggo node.
@@ -42,9 +43,11 @@ pub struct Behaviour {
     /// Register with a rendezvous server and query remote peer addresses.
     pub rendezvous_client: Toggle<rendezvous::client::Behaviour>,
 
-    /// Serve as a rendezvous point for remote peers to register their external addresses
-    /// and query the addresses of other peers.
+    /// Serve as a rendezvous point for remote peers to register their external addresses and query
+    /// the addresses of other peers.
     pub rendezvous_server: Toggle<rendezvous::server::Behaviour>,
+
+    pub replication: replication::Behaviour,
 }
 
 impl Behaviour {
@@ -103,6 +106,7 @@ impl Behaviour {
         // address has been provided
         let rendezvous_client = if network_config.rendezvous_address.is_some() {
             debug!("Rendezvous client network behaviour enabled");
+            // @TODO: Why does this need the whole key pair?!
             Some(rendezvous::client::Behaviour::new(key_pair))
         } else {
             None
@@ -132,6 +136,8 @@ impl Behaviour {
             None
         };
 
+        let replication = replication::Behaviour::new();
+
         Ok(Self {
             autonat: autonat.into(),
             identify: identify.into(),
@@ -142,6 +148,7 @@ impl Behaviour {
             rendezvous_server: rendezvous_server.into(),
             relay_client: relay_client.into(),
             relay_server: relay_server.into(),
+            replication,
         })
     }
 }

aquadoggo/src/network/config.rs

@@ -4,7 +4,7 @@ use std::path::PathBuf;
 
 use anyhow::Result;
 use libp2p::connection_limits::ConnectionLimits;
-use libp2p::identity::Keypair;
+use libp2p::identity::{Keypair, PublicKey};
 use libp2p::{Multiaddr, PeerId};
 use log::info;
 use serde::{Deserialize, Serialize};
@@ -72,8 +72,8 @@ pub struct NetworkConfiguration {
 
     /// Ping behaviour enabled.
     ///
-    /// Send outbound pings to connected peers every 15 seconds and respond to inbound pings.
-    /// Every sent ping must yield a response within 20 seconds in order to be successful.
+    /// Send outbound pings to connected peers every 15 seconds and respond to inbound pings. Every
+    /// sent ping must yield a response within 20 seconds in order to be successful.
     pub ping: bool,
 
     /// QUIC transport port.
@@ -103,6 +103,9 @@ pub struct NetworkConfiguration {
     ///
     /// Serve as a rendezvous point for peer discovery, allowing peer registration and queries.
     pub rendezvous_server_enabled: bool,
+
+    /// Our local peer id.
+    pub peer_id: Option<PeerId>,
 }
 
 impl Default for NetworkConfiguration {
@@ -127,11 +130,17 @@ impl Default for NetworkConfiguration {
             rendezvous_address: None,
             rendezvous_peer_id: None,
             rendezvous_server_enabled: false,
+            peer_id: None,
         }
     }
 }
 
 impl NetworkConfiguration {
+    /// Derive peer id from a given public key.
+    pub fn set_peer_id(&mut self, public_key: &PublicKey) {
+        self.peer_id = Some(PeerId::from_public_key(public_key));
+    }
+
     /// Define the connection limits of the swarm.
     pub fn connection_limits(&self) -> ConnectionLimits {
         ConnectionLimits::default()
@@ -144,8 +153,8 @@ impl NetworkConfiguration {
 
     /// Load the key pair from the file at the specified path.
     ///
-    /// If the file does not exist, a random key pair is generated and saved.
-    /// If no path is specified, a random key pair is generated.
+    /// If the file does not exist, a random key pair is generated and saved. If no path is
+    /// specified, a random key pair is generated.
     pub fn load_or_generate_key_pair(path: Option<PathBuf>) -> Result<Keypair> {
         let key_pair = match path {
             Some(mut path) => {

aquadoggo/src/network/mod.rs

@@ -3,12 +3,10 @@
 mod behaviour;
 mod config;
 mod identity;
+mod replication;
 mod service;
 mod swarm;
 mod transport;
-// @TODO: Remove this as soon as we integrated it into the libp2p swarm
-#[allow(dead_code)]
-mod replication;
 
 pub use config::NetworkConfiguration;
 pub use service::network_service;

aquadoggo/src/network/replication/behaviour.rs

@@ -11,17 +11,16 @@ use libp2p::swarm::{
 use libp2p::{Multiaddr, PeerId};
 
 use crate::network::replication::handler::{Handler, HandlerInEvent, HandlerOutEvent};
-use crate::network::replication::protocol::Message;
+use crate::replication::SyncMessage;
 
 #[derive(Debug)]
-pub enum BehaviourOutEvent {
-    MessageReceived(PeerId, Message),
-    Error,
+pub enum Event {
+    MessageReceived(PeerId, SyncMessage),
 }
 
 #[derive(Debug)]
 pub struct Behaviour {
-    events: VecDeque<ToSwarm<BehaviourOutEvent, HandlerInEvent>>,
+    events: VecDeque<ToSwarm<Event, HandlerInEvent>>,
 }
 
 impl Behaviour {
@@ -33,18 +32,17 @@ impl Behaviour {
 }
 
 impl Behaviour {
-    fn send_message(&mut self, peer_id: PeerId, message: Message) {
+    pub fn send_message(&mut self, peer_id: PeerId, message: SyncMessage) {
         self.events.push_back(ToSwarm::NotifyHandler {
             peer_id,
             event: HandlerInEvent::Message(message),
             handler: NotifyHandler::Any,
         });
     }
 
-    fn handle_received_message(&mut self, peer_id: &PeerId, message: Message) {
-        // @TODO: Handle incoming messages
+    fn handle_received_message(&mut self, peer_id: &PeerId, message: SyncMessage) {
         self.events
-            .push_back(ToSwarm::GenerateEvent(BehaviourOutEvent::MessageReceived(
+            .push_back(ToSwarm::GenerateEvent(Event::MessageReceived(
                 *peer_id, message,
             )));
     }
@@ -53,7 +51,7 @@ impl Behaviour {
 impl NetworkBehaviour for Behaviour {
     type ConnectionHandler = Handler;
 
-    type OutEvent = BehaviourOutEvent;
+    type OutEvent = Event;
 
     fn handle_established_inbound_connection(
         &mut self,
@@ -124,17 +122,18 @@ mod tests {
     use libp2p::swarm::{keep_alive, Swarm};
     use libp2p_swarm_test::SwarmExt;
 
-    use crate::network::replication::Message;
+    use crate::replication::{Message, SyncMessage, TargetSet};
 
-    use super::{Behaviour as ReplicationBehaviour, BehaviourOutEvent};
+    use super::{Behaviour as ReplicationBehaviour, Event};
 
     #[tokio::test]
     async fn peers_connect() {
         // Create two swarms
         let mut swarm1 = Swarm::new_ephemeral(|_| ReplicationBehaviour::new());
         let mut swarm2 = Swarm::new_ephemeral(|_| ReplicationBehaviour::new());
 
-        // Listen on swarm1 and connect from swarm2, this should establish a bi-directional connection.
+        // Listen on swarm1 and connect from swarm2, this should establish a bi-directional
+        // connection.
         swarm1.listen().await;
         swarm2.connect(&mut swarm1).await;
 
@@ -189,9 +188,10 @@ mod tests {
         assert_eq!(info2.connection_counters().num_established(), 1);
 
         // Send a message from to swarm1 local peer from swarm2 local peer.
-        swarm1
-            .behaviour_mut()
-            .send_message(swarm2_peer_id, Message::Dummy(0));
+        swarm1.behaviour_mut().send_message(
+            swarm2_peer_id,
+            SyncMessage::new(0, Message::SyncRequest(0.into(), TargetSet::new(&vec![]))),
+        );
 
         // Await a swarm event on swarm2.
         //
@@ -220,20 +220,22 @@ mod tests {
         let swarm2_peer_id = *swarm2.local_peer_id();
 
         // Send a message from to swarm1 local peer from swarm2 local peer.
-        swarm1
-            .behaviour_mut()
-            .send_message(swarm2_peer_id, Message::Dummy(0));
+        swarm1.behaviour_mut().send_message(
+            swarm2_peer_id,
+            SyncMessage::new(0, Message::SyncRequest(0.into(), TargetSet::new(&vec![]))),
+        );
 
         // Send a message from to swarm2 local peer from swarm1 local peer.
-        swarm2
-            .behaviour_mut()
-            .send_message(swarm1_peer_id, Message::Dummy(1));
+        swarm2.behaviour_mut().send_message(
+            swarm1_peer_id,
+            SyncMessage::new(1, Message::SyncRequest(0.into(), TargetSet::new(&vec![]))),
+        );
 
         // Collect the next 2 behaviour events which occur in either swarms.
         for _ in 0..2 {
             tokio::select! {
-                BehaviourOutEvent::MessageReceived(peer_id, message) = swarm1.next_behaviour_event() => res1.push((peer_id, message)),
-                BehaviourOutEvent::MessageReceived(peer_id, message) = swarm2.next_behaviour_event() => res2.push((peer_id, message)),
+                Event::MessageReceived(peer_id, message) = swarm1.next_behaviour_event() => res1.push((peer_id, message)),
+                Event::MessageReceived(peer_id, message) = swarm2.next_behaviour_event() => res2.push((peer_id, message)),
             }
         }
 
@@ -244,11 +246,17 @@ mod tests {
         // swarm1 should have received the message from swarm2 peer.
         let (peer_id, message) = &res1[0];
         assert_eq!(peer_id, &swarm2_peer_id);
-        assert_eq!(message, &Message::Dummy(1));
+        assert_eq!(
+            message,
+            &SyncMessage::new(1, Message::SyncRequest(0.into(), TargetSet::new(&vec![])))
+        );
 
         // swarm2 should have received the message from swarm1 peer.
         let (peer_id, message) = &res2[0];
         assert_eq!(peer_id, &swarm1_peer_id);
-        assert_eq!(message, &Message::Dummy(0));
+        assert_eq!(
+            message,
+            &SyncMessage::new(0, Message::SyncRequest(0.into(), TargetSet::new(&vec![])))
+        );
     }
 }