Integrate replication manager with networking stack

CHANGELOG.md

@@ -23,6 +23,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Replication protocol session manager [#363](https://github.com/p2panda/aquadoggo/pull/363)
 - Replication message de- / serialization [#375](https://github.com/p2panda/aquadoggo/pull/375)
 - Naive protocol replication [#380](https://github.com/p2panda/aquadoggo/pull/380)
+- Integrate replication manager with networking stack [#387](https://github.com/p2panda/aquadoggo/pull/387) 🥞
 
 ### Changed
 

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

@@ -3,6 +3,8 @@
 use p2panda_rs::operation::OperationId;
 
 use crate::manager::Sender;
+use crate::network::Peer;
+use crate::replication::SyncMessage;
 
 /// Sender for cross-service communication bus.
 pub type ServiceSender = Sender<ServiceMessage>;
@@ -12,4 +14,19 @@ 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.
+    PeerConnected(Peer),
+
+    /// Node closed a connection to another node.
+    PeerDisconnected(Peer),
+
+    /// Node sent a message to remote node for replication.
+    SentReplicationMessage(Peer, SyncMessage),
+
+    /// Node received a message from remote node for replication.
+    ReceivedReplicationMessage(Peer, SyncMessage),
+
+    /// Replication protocol failed with an critical error.
+    ReplicationFailed(Peer),
 }

aquadoggo/src/config.rs

@@ -98,6 +98,29 @@ impl Configuration {
             }
         };
 
+        // Derive peer id from key pair
+        // @TODO: This needs refactoring: https://github.com/p2panda/aquadoggo/issues/388
+        let key_pair = NetworkConfiguration::load_or_generate_key_pair(config.base_path.clone())?;
+        config.network.set_peer_id(&key_pair.public());
+
         Ok(config)
     }
 }
+
+#[cfg(test)]
+impl Configuration {
+    /// Returns a new configuration object for a node which stores all data temporarily in memory.
+    pub fn new_ephemeral() -> Self {
+        let mut config = Configuration {
+            database_url: Some("sqlite::memory:".to_string()),
+            ..Default::default()
+        };
+
+        // Generate a random key pair and just keep it in memory
+        // @TODO: This needs refactoring: https://github.com/p2panda/aquadoggo/issues/388
+        let key_pair: libp2p::identity::Keypair = crate::network::identity::Identity::new();
+        config.network.set_peer_id(&key_pair.public());
+
+        config
+    }
+}

aquadoggo/src/db/stores/document.rs

@@ -459,6 +459,7 @@ async fn insert_document_view(
             )
         VALUES
             ($1, $2, $3)
+        ON CONFLICT(document_view_id) DO NOTHING -- @TODO: temp fix for double document view insertions: https://github.com/p2panda/aquadoggo/issues/398
         ",
     )
     .bind(document_view.id().to_string())

aquadoggo/src/db/stores/entry.rs

@@ -277,14 +277,14 @@ impl EntryStore for SqlStore {
         log_id: &LogId,
         initial_seq_num: &SeqNum,
     ) -> Result<Vec<StorageEntry>, EntryStorageError> {
+        // Formatting query string in this way as `sqlx` currently doesn't support binding list
+        // arguments for IN queries.
         let cert_pool_seq_nums = get_lipmaa_links_back_to(initial_seq_num.as_u64(), 1)
             .iter()
-            .map(|seq_num| seq_num.to_string())
+            .map(|seq_num| format!("'{seq_num}'"))
             .collect::<Vec<String>>()
             .join(",");
 
-        // Formatting query string in this way as `sqlx` currently
-        // doesn't support binding list arguments for IN queries.
         let sql_str = format!(
             "SELECT
                 public_key,
@@ -299,7 +299,7 @@ impl EntryStore for SqlStore {
             WHERE
                 public_key = $1
                 AND log_id = $2
-                AND CAST(seq_num AS NUMERIC) IN ({})
+                AND seq_num IN ({})
             ORDER BY
                 CAST(seq_num AS NUMERIC) DESC
             ",
@@ -337,6 +337,8 @@ impl SqlStore {
                 logs.schema = $1
             GROUP BY
                 entries.public_key, entries.log_id
+            ORDER BY
+                entries.public_key, CAST(entries.log_id AS NUMERIC)
             ",
         )
         .bind(schema_id.to_string())

aquadoggo/src/graphql/scalars/document_view_id_scalar.rs

@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
-use std::{fmt::Display, str::FromStr};
+use std::fmt::Display;
+use std::str::FromStr;
 
 use dynamic_graphql::{Error, Result, Scalar, ScalarValue, Value};
 use p2panda_rs::document::DocumentViewId;

aquadoggo/src/graphql/schema.rs

@@ -194,7 +194,7 @@ impl GraphQLSchemaManager {
         let shared = self.shared.clone();
         let schemas = self.schemas.clone();
 
-        info!("Subscribing GraphQL manager to schema provider");
+        debug!("Subscribing GraphQL manager to schema provider");
         let mut on_schema_added = shared.schema_provider.on_schema_added();
 
         // Create the new GraphQL based on the current state of known p2panda application schemas

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

@@ -1,5 +1,7 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+use std::time::Duration;
+
 use anyhow::Result;
 use libp2p::identity::Keypair;
 use libp2p::swarm::behaviour::toggle::Toggle;
@@ -8,9 +10,31 @@ use libp2p::{autonat, connection_limits, identify, mdns, ping, relay, rendezvous
 use log::debug;
 
 use crate::network::config::NODE_NAMESPACE;
+use crate::network::peers;
 use crate::network::NetworkConfiguration;
 
+/// How often do we broadcast mDNS queries into the network.
+const MDNS_QUERY_INTERVAL: Duration = Duration::from_secs(5);
+
+/// How often do we ping other peers to check for a healthy connection.
+const PING_INTERVAL: Duration = Duration::from_secs(5);
+
+/// How long do we wait for an answer from the other peer before we consider the connection as
+/// stale.
+const PING_TIMEOUT: Duration = Duration::from_secs(3);
+
 /// Network behaviour for the aquadoggo node.
+///
+/// In libp2p all different behaviours are "merged" into one "main behaviour" with help of the
+/// `NetworkBehaviour` derive macro.
+///
+/// All behaviours share the same connections with each other. Together they form something we
+/// could call our "custom" networking behaviour.
+///
+/// It is possible for a peer to not support all behaviours, internally libp2p negotiates the
+/// capabilities of each peer for us and upgrades the protocol accordingly. For example two peers
+/// can handle p2panda messages with each others (using the `peers` behaviour) but do not
+/// necessarily need to be able to support the `relay` behaviour.
 #[derive(NetworkBehaviour)]
 pub struct Behaviour {
     /// Determine NAT status by requesting remote peers to dial the public address of the
@@ -42,9 +66,12 @@ 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>,
+
+    /// Register peer connections and handle p2panda messaging with them.
+    pub peers: peers::Behaviour,
 }
 
 impl Behaviour {
@@ -86,15 +113,25 @@ impl Behaviour {
         // Create an mDNS behaviour with default configuration if the mDNS flag is set
         let mdns = if network_config.mdns {
             debug!("mDNS network behaviour enabled");
-            Some(mdns::Behaviour::new(Default::default(), peer_id)?)
+            Some(mdns::Behaviour::new(
+                mdns::Config {
+                    query_interval: MDNS_QUERY_INTERVAL,
+                    ..mdns::Config::default()
+                },
+                peer_id,
+            )?)
         } else {
             None
         };
 
         // Create a ping behaviour with default configuration if the ping flag is set
         let ping = if network_config.ping {
             debug!("Ping network behaviour enabled");
-            Some(ping::Behaviour::default())
+            Some(ping::Behaviour::new(
+                ping::Config::new()
+                    .with_interval(PING_INTERVAL)
+                    .with_timeout(PING_TIMEOUT),
+            ))
         } else {
             None
         };
@@ -132,6 +169,9 @@ impl Behaviour {
             None
         };
 
+        // Create behaviour to manage peer connections and handle p2panda messaging
+        let peers = peers::Behaviour::new();
+
         Ok(Self {
             autonat: autonat.into(),
             identify: identify.into(),
@@ -142,6 +182,7 @@ impl Behaviour {
             rendezvous_server: rendezvous_server.into(),
             relay_client: relay_client.into(),
             relay_server: relay_server.into(),
+            peers,
         })
     }
 }

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) => {