Repository: p2panda/aquadoggo (language: rust)

Issue:
libp2p network behaviour and connection handler for replication protocol



Discussion / hints:

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 26d78d405..68a17749a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -19,6 +19,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - SQL query for collections [#311](https://github.com/p2panda/aquadoggo/pull/311)
 - Add custom validation to GraphQL scalar types [#318](https://github.com/p2panda/aquadoggo/pull/318)
 - Introduce property tests for GraphQL query API with `proptest` [#338](https://github.com/p2panda/aquadoggo/pull/338)
+- Introduce initial libp2p network behaviour for replication protocol [#365](https://github.com/p2panda/aquadoggo/pull/365) 
 - Replication protocol session manager [#363](https://github.com/p2panda/aquadoggo/pull/363)
 - Replication message de- / serialization [#375](https://github.com/p2panda/aquadoggo/pull/375)
 
diff --git a/Cargo.lock b/Cargo.lock
index d54e59690..9012f7ccb 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -168,6 +168,7 @@ dependencies = [
  "async-graphql",
  "async-graphql-axum",
  "async-trait",
+ "asynchronous-codec",
  "axum",
  "bamboo-rs-core-ed25519-yasmf",
  "bs58",
@@ -184,6 +185,7 @@ dependencies = [
  "hyper",
  "libp2p",
  "libp2p-quic",
+ "libp2p-swarm-test",
  "lipmaa-link 0.2.2",
  "log",
  "once_cell",
@@ -339,6 +341,18 @@ dependencies = [
  "slab",
 ]
 
+[[package]]
+name = "async-fs"
+version = "1.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "279cf904654eeebfa37ac9bb1598880884924aab82e290aa65c9e77a0e142e06"
+dependencies = [
+ "async-lock",
+ "autocfg",
+ "blocking",
+ "futures-lite",
+]
+
 [[package]]
 name = "async-global-executor"
 version = "2.3.1"
@@ -473,6 +487,36 @@ dependencies = [
  "event-listener",
 ]
 
+[[package]]
+name = "async-net"
+version = "1.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4051e67316bc7eff608fe723df5d32ed639946adcd69e07df41fd42a7b411f1f"
+dependencies = [
+ "async-io",
+ "autocfg",
+ "blocking",
+ "futures-lite",
+]
+
+[[package]]
+name = "async-process"
+version = "1.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7a9d28b1d97e08915212e2e45310d47854eafa69600756fc735fb788f75199c9"
+dependencies = [
+ "async-io",
+ "async-lock",
+ "autocfg",
+ "blocking",
+ "cfg-if",
+ "event-listener",
+ "futures-lite",
+ "rustix",
+ "signal-hook",
+ "windows-sys 0.48.0",
+]
+
 [[package]]
 name = "async-std"
 version = "1.12.0"
@@ -549,6 +593,8 @@ dependencies = [
  "futures-util",
  "memchr",
  "pin-project-lite",
+ "serde",
+ "serde_cbor",
 ]
 
 [[package]]
@@ -2141,6 +2187,7 @@ dependencies = [
  "ipnet",
  "log",
  "rtnetlink",
+ "smol",
  "system-configuration",
  "tokio",
  "windows",
@@ -2592,6 +2639,23 @@ dependencies = [
  "void",
 ]
 
+[[package]]
+name = "libp2p-plaintext"
+version = "0.39.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2ff582c0b74ffda004b716b97bc9cfe7f39960204877758b876dde86093beaa8"
+dependencies = [
+ "asynchronous-codec",
+ "bytes",
+ "futures",
+ "libp2p-core",
+ "libp2p-identity",
+ "log",
+ "quick-protobuf",
+ "unsigned-varint",
+ "void",
+]
+
 [[package]]
 name = "libp2p-quic"
 version = "0.7.0-alpha.3"
@@ -2709,12 +2773,32 @@ dependencies = [
  "syn 1.0.109",
 ]
 
+[[package]]
+name = "libp2p-swarm-test"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2f6cb31f26505d134ce7106f419de02a8d9640ea56d92f751138933fbec8184d"
+dependencies = [
+ "async-trait",
+ "futures",
+ "futures-timer",
+ "libp2p-core",
+ "libp2p-identity",
+ "libp2p-plaintext",
+ "libp2p-swarm",
+ "libp2p-tcp",
+ "libp2p-yamux",
+ "log",
+ "rand 0.8.5",
+]
+
 [[package]]
 name = "libp2p-tcp"
 version = "0.39.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "33d33698596d7722d85d3ab0c86c2c322254fce1241e91208e3679b4eb3026cf"
 dependencies = [
+ "async-io",
  "futures",
  "futures-timer",
  "if-watch",
@@ -3098,6 +3182,7 @@ version = "0.8.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6471bf08e7ac0135876a9581bf3217ef0333c191c128d34878079f42ee150411"
 dependencies = [
+ "async-io",
  "bytes",
  "futures",
  "libc",
@@ -3963,6 +4048,7 @@ version = "0.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "322c53fd76a18698f1c27381d58091de3a043d356aa5bd0d510608b565f469a0"
 dependencies = [
+ "async-global-executor",
  "futures",
  "log",
  "netlink-packet-route",
@@ -4177,6 +4263,16 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "serde_cbor"
+version = "0.11.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2bef2ebfde456fb76bbcf9f59315333decc4fda0b2b44b420243c11e0f5ec1f5"
+dependencies = [
+ "half",
+ "serde",
+]
+
 [[package]]
 name = "serde_derive"
 version = "1.0.163"
@@ -4268,6 +4364,16 @@ dependencies = [
  "digest 0.10.6",
 ]
 
+[[package]]
+name = "signal-hook"
+version = "0.3.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "732768f1176d21d09e076c23a93123d40bba92d50c4058da34d45c8de8e682b9"
+dependencies = [
+ "libc",
+ "signal-hook-registry",
+]
+
 [[package]]
 name = "signal-hook-registry"
 version = "1.4.1"
@@ -4302,6 +4408,23 @@ version = "1.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a507befe795404456341dfab10cef66ead4c041f62b8b11bbb92bffe5d0953e0"
 
+[[package]]
+name = "smol"
+version = "1.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "13f2b548cd8447f8de0fdf1c592929f70f4fc7039a05e47404b0d096ec6987a1"
+dependencies = [
+ "async-channel",
+ "async-executor",
+ "async-fs",
+ "async-io",
+ "async-lock",
+ "async-net",
+ "async-process",
+ "blocking",
+ "futures-lite",
+]
+
 [[package]]
 name = "snafu"
 version = "0.6.10"
diff --git a/aquadoggo/Cargo.toml b/aquadoggo/Cargo.toml
index d43e8d0ce..d9edf17e1 100644
--- a/aquadoggo/Cargo.toml
+++ b/aquadoggo/Cargo.toml
@@ -20,6 +20,7 @@ anyhow = "1.0.62"
 async-graphql = { version = "5.0.6", features = ["dynamic-schema"] }
 async-graphql-axum = "5.0.6"
 async-trait = "0.1.64"
+asynchronous-codec = { version = "0.6.1", features = ["cbor"] }
 axum = "0.6.10"
 bamboo-rs-core-ed25519-yasmf = "0.1.1"
 bs58 = "0.4.0"
@@ -80,6 +81,7 @@ ctor = "0.1.23"
 env_logger = "0.9.0"
 http = "0.2.9"
 hyper = "0.14.19"
+libp2p-swarm-test = "0.1.0"
 once_cell = "1.17.0"
 p2panda-rs = { git = "https://github.com/p2panda/p2panda", rev = "679107b636eea39e7713d9b21ac69df28aaad97d", features = [
     "test-utils",
diff --git a/aquadoggo/src/network/mod.rs b/aquadoggo/src/network/mod.rs
index bf2860122..1af29360f 100644
--- a/aquadoggo/src/network/mod.rs
+++ b/aquadoggo/src/network/mod.rs
@@ -6,6 +6,9 @@ mod identity;
 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;
diff --git a/aquadoggo/src/network/replication/behaviour.rs b/aquadoggo/src/network/replication/behaviour.rs
new file mode 100644
index 000000000..e6f54d039
--- /dev/null
+++ b/aquadoggo/src/network/replication/behaviour.rs
@@ -0,0 +1,254 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use std::collections::VecDeque;
+use std::task::{Context, Poll};
+
+use libp2p::core::Endpoint;
+use libp2p::swarm::{
+    ConnectionDenied, ConnectionId, FromSwarm, NetworkBehaviour, NotifyHandler, PollParameters,
+    THandler, THandlerInEvent, THandlerOutEvent, ToSwarm,
+};
+use libp2p::{Multiaddr, PeerId};
+
+use crate::network::replication::handler::{Handler, HandlerInEvent, HandlerOutEvent};
+use crate::network::replication::protocol::Message;
+
+#[derive(Debug)]
+pub enum BehaviourOutEvent {
+    MessageReceived(PeerId, Message),
+    Error,
+}
+
+#[derive(Debug)]
+pub struct Behaviour {
+    events: VecDeque<ToSwarm<BehaviourOutEvent, HandlerInEvent>>,
+}
+
+impl Behaviour {
+    pub fn new() -> Self {
+        Self {
+            events: VecDeque::new(),
+        }
+    }
+}
+
+impl Behaviour {
+    fn send_message(&mut self, peer_id: PeerId, message: Message) {
+        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
+        self.events
+            .push_back(ToSwarm::GenerateEvent(BehaviourOutEvent::MessageReceived(
+                *peer_id, message,
+            )));
+    }
+}
+
+impl NetworkBehaviour for Behaviour {
+    type ConnectionHandler = Handler;
+
+    type OutEvent = BehaviourOutEvent;
+
+    fn handle_established_inbound_connection(
+        &mut self,
+        _: ConnectionId,
+        _: PeerId,
+        _: &Multiaddr,
+        _: &Multiaddr,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        Ok(Handler::new())
+    }
+
+    fn handle_established_outbound_connection(
+        &mut self,
+        _: ConnectionId,
+        _: PeerId,
+        _: &Multiaddr,
+        _: Endpoint,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        Ok(Handler::new())
+    }
+
+    fn on_connection_handler_event(
+        &mut self,
+        peer: PeerId,
+        _connection_id: ConnectionId,
+        handler_event: THandlerOutEvent<Self>,
+    ) {
+        match handler_event {
+            HandlerOutEvent::Message(message) => {
+                self.handle_received_message(&peer, message);
+            }
+        }
+    }
+
+    fn on_swarm_event(&mut self, event: FromSwarm<Self::ConnectionHandler>) {
+        match event {
+            FromSwarm::ConnectionEstablished(_)
+            | FromSwarm::ConnectionClosed(_)
+            | FromSwarm::AddressChange(_)
+            | FromSwarm::DialFailure(_)
+            | FromSwarm::ListenFailure(_)
+            | FromSwarm::NewListener(_)
+            | FromSwarm::NewListenAddr(_)
+            | FromSwarm::ExpiredListenAddr(_)
+            | FromSwarm::ListenerError(_)
+            | FromSwarm::ListenerClosed(_)
+            | FromSwarm::NewExternalAddr(_)
+            | FromSwarm::ExpiredExternalAddr(_) => {}
+        }
+    }
+
+    fn poll(
+        &mut self,
+        _cx: &mut Context<'_>,
+        _params: &mut impl PollParameters,
+    ) -> Poll<ToSwarm<Self::OutEvent, THandlerInEvent<Self>>> {
+        if let Some(event) = self.events.pop_front() {
+            return Poll::Ready(event);
+        }
+
+        Poll::Pending
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use futures::FutureExt;
+    use libp2p::swarm::{keep_alive, Swarm};
+    use libp2p_swarm_test::SwarmExt;
+
+    use crate::network::replication::Message;
+
+    use super::{Behaviour as ReplicationBehaviour, BehaviourOutEvent};
+
+    #[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.
+        swarm1.listen().await;
+        swarm2.connect(&mut swarm1).await;
+
+        let swarm1_peer_id = *swarm1.local_peer_id();
+        let swarm2_peer_id = *swarm2.local_peer_id();
+
+        let info1 = swarm1.network_info();
+        let info2 = swarm2.network_info();
+
+        // Peers should be connected.
+        assert!(swarm2.is_connected(&swarm1_peer_id));
+        assert!(swarm1.is_connected(&swarm2_peer_id));
+
+        // Each swarm should have exactly one connected peer.
+        assert_eq!(info1.num_peers(), 1);
+        assert_eq!(info2.num_peers(), 1);
+
+        // Each swarm should have one established connection.
+        assert_eq!(info1.connection_counters().num_established(), 1);
+        assert_eq!(info2.connection_counters().num_established(), 1);
+    }
+
+    #[tokio::test]
+    async fn incompatible_network_behaviour() {
+        // Create two swarms
+        let mut swarm1 = Swarm::new_ephemeral(|_| ReplicationBehaviour::new());
+        let mut swarm2 = Swarm::new_ephemeral(|_| keep_alive::Behaviour);
+
+        // Listen on swarm1 and connect from swarm2, this should establish a bi-directional connection.
+        swarm1.listen().await;
+        swarm2.connect(&mut swarm1).await;
+
+        let swarm1_peer_id = *swarm1.local_peer_id();
+        let swarm2_peer_id = *swarm2.local_peer_id();
+
+        let info1 = swarm1.network_info();
+        let info2 = swarm2.network_info();
+
+        // Even though the network behaviours of our two peers are incompatible they still
+        // establish a connection.
+
+        // Peers should be connected.
+        assert!(swarm2.is_connected(&swarm1_peer_id));
+        assert!(swarm1.is_connected(&swarm2_peer_id));
+
+        // Each swarm should have exactly one connected peer.
+        assert_eq!(info1.num_peers(), 1);
+        assert_eq!(info2.num_peers(), 1);
+
+        // Each swarm should have one established connection.
+        assert_eq!(info1.connection_counters().num_established(), 1);
+        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));
+
+        // Await a swarm event on swarm2.
+        //
+        // We expect a timeout panic as no event will occur.
+        let result = std::panic::AssertUnwindSafe(swarm2.next_swarm_event())
+            .catch_unwind()
+            .await;
+
+        assert!(result.is_err())
+    }
+
+    #[tokio::test]
+    async fn swarm_behaviour_events() {
+        // 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.
+        swarm1.listen().await;
+        swarm2.connect(&mut swarm1).await;
+
+        let mut res1 = Vec::new();
+        let mut res2 = Vec::new();
+
+        let swarm1_peer_id = *swarm1.local_peer_id();
+        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));
+
+        // Send a message from to swarm2 local peer from swarm1 local peer.
+        swarm2
+            .behaviour_mut()
+            .send_message(swarm1_peer_id, Message::Dummy(1));
+
+        // 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)),
+            }
+        }
+
+        // Each swarm should have emitted exactly one event.
+        assert_eq!(res1.len(), 1);
+        assert_eq!(res2.len(), 1);
+
+        // 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));
+
+        // 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));
+    }
+}
diff --git a/aquadoggo/src/network/replication/handler.rs b/aquadoggo/src/network/replication/handler.rs
new file mode 100644
index 000000000..b7b29631d
--- /dev/null
+++ b/aquadoggo/src/network/replication/handler.rs
@@ -0,0 +1,340 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use std::collections::VecDeque;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use asynchronous_codec::Framed;
+use futures::{Sink, StreamExt};
+use libp2p::swarm::handler::{ConnectionEvent, FullyNegotiatedInbound, FullyNegotiatedOutbound};
+use libp2p::swarm::{
+    ConnectionHandler, ConnectionHandlerEvent, KeepAlive, NegotiatedSubstream, SubstreamProtocol,
+};
+use thiserror::Error;
+
+use crate::network::replication::{Codec, CodecError, Message, Protocol};
+
+pub struct Handler {
+    /// Upgrade configuration for the replication protocol.
+    listen_protocol: SubstreamProtocol<Protocol, ()>,
+
+    /// The single long-lived outbound substream.
+    outbound_substream: Option<OutboundSubstreamState>,
+
+    /// The single long-lived inbound substream.
+    inbound_substream: Option<InboundSubstreamState>,
+
+    /// Flag indicating that an outbound substream is being established to prevent duplicate
+    /// requests.
+    outbound_substream_establishing: bool,
+
+    /// Queue of messages that we want to send to the remote.
+    send_queue: VecDeque<Message>,
+
+    /// Flag determining whether to maintain the connection to the peer.
+    keep_alive: KeepAlive,
+}
+
+impl Handler {
+    pub fn new() -> Self {
+        Self {
+            listen_protocol: SubstreamProtocol::new(Protocol::new(), ()),
+            outbound_substream: None,
+            inbound_substream: None,
+            outbound_substream_establishing: false,
+            send_queue: VecDeque::new(),
+            keep_alive: KeepAlive::Yes,
+        }
+    }
+
+    fn on_fully_negotiated_outbound(
+        &mut self,
+        FullyNegotiatedOutbound { protocol, info: () }: FullyNegotiatedOutbound<
+            <Self as ConnectionHandler>::OutboundProtocol,
+            <Self as ConnectionHandler>::OutboundOpenInfo,
+        >,
+    ) {
+        self.outbound_substream = Some(OutboundSubstreamState::WaitingOutput(protocol));
+    }
+
+    fn on_fully_negotiated_inbound(
+        &mut self,
+        FullyNegotiatedInbound { protocol, .. }: FullyNegotiatedInbound<
+            <Self as ConnectionHandler>::InboundProtocol,
+            <Self as ConnectionHandler>::InboundOpenInfo,
+        >,
+    ) {
+        self.inbound_substream = Some(InboundSubstreamState::WaitingInput(protocol));
+    }
+}
+
+/// An event sent from the network behaviour to the connection handler.
+#[derive(Debug)]
+pub enum HandlerInEvent {
+    /// Replication message to send on outbound stream.
+    Message(Message),
+}
+
+/// The event emitted by the connection handler.
+///
+/// This informs the network behaviour of various events created by the handler.
+#[derive(Debug)]
+pub enum HandlerOutEvent {
+    /// Replication message received on the inbound stream.
+    Message(Message),
+}
+
+#[derive(Debug, Error)]
+pub enum HandlerError {
+    #[error("Failed to encode or decode CBOR")]
+    Codec(#[from] CodecError),
+}
+
+type Stream = Framed<NegotiatedSubstream, Codec>;
+
+/// State of the inbound substream, opened either by us or by the remote.
+enum InboundSubstreamState {
+    /// Waiting for a message from the remote. The idle state for an inbound substream.
+    WaitingInput(Stream),
+
+    /// The substream is being closed.
+    Closing(Stream),
+
+    /// An error occurred during processing.
+    Poisoned,
+}
+
+/// State of the outbound substream, opened either by us or by the remote.
+enum OutboundSubstreamState {
+    /// Waiting for the user to send a message. The idle state for an outbound substream.
+    WaitingOutput(Stream),
+
+    /// Waiting to send a message to the remote.
+    PendingSend(Stream, Message),
+
+    /// Waiting to flush the substream so that the data arrives to the remote.
+    PendingFlush(Stream),
+
+    /// An error occurred during processing.
+    Poisoned,
+}
+
+impl ConnectionHandler for Handler {
+    type InEvent = HandlerInEvent;
+    type OutEvent = HandlerOutEvent;
+    type Error = HandlerError;
+    type InboundProtocol = Protocol;
+    type OutboundProtocol = Protocol;
+    type InboundOpenInfo = ();
+    type OutboundOpenInfo = ();
+
+    fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
+        self.listen_protocol.clone()
+    }
+
+    fn on_connection_event(
+        &mut self,
+        event: ConnectionEvent<
+            Self::InboundProtocol,
+            Self::OutboundProtocol,
+            Self::InboundOpenInfo,
+            Self::OutboundOpenInfo,
+        >,
+    ) {
+        match event {
+            ConnectionEvent::FullyNegotiatedOutbound(fully_negotiated_outbound) => {
+                self.on_fully_negotiated_outbound(fully_negotiated_outbound);
+            }
+            ConnectionEvent::FullyNegotiatedInbound(fully_negotiated_inbound) => {
+                self.on_fully_negotiated_inbound(fully_negotiated_inbound);
+            }
+            ConnectionEvent::DialUpgradeError(_)
+            | ConnectionEvent::AddressChange(_)
+            | ConnectionEvent::ListenUpgradeError(_) => {}
+        }
+    }
+
+    fn on_behaviour_event(&mut self, event: Self::InEvent) {
+        self.keep_alive = KeepAlive::Yes;
+
+        match event {
+            HandlerInEvent::Message(message) => {
+                self.send_queue.push_back(message);
+            }
+        }
+    }
+
+    fn connection_keep_alive(&self) -> KeepAlive {
+        self.keep_alive
+    }
+
+    fn poll(
+        &mut self,
+        cx: &mut Context<'_>,
+    ) -> Poll<
+        ConnectionHandlerEvent<
+            Self::OutboundProtocol,
+            Self::OutboundOpenInfo,
+            Self::OutEvent,
+            Self::Error,
+        >,
+    > {
+        // Determine if we need to create the outbound stream
+        if !self.send_queue.is_empty()
+            && self.outbound_substream.is_none()
+            && !self.outbound_substream_establishing
+        {
+            self.outbound_substream_establishing = true;
+            return Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest {
+                protocol: self.listen_protocol(),
+            });
+        }
+
+        // Process inbound stream
+        loop {
+            match std::mem::replace(
+                &mut self.inbound_substream,
+                Some(InboundSubstreamState::Poisoned),
+            ) {
+                Some(InboundSubstreamState::WaitingInput(mut substream)) => {
+                    match substream.poll_next_unpin(cx) {
+                        Poll::Ready(Some(Ok(message))) => {
+                            // Received message from remote peer
+                            self.inbound_substream =
+                                Some(InboundSubstreamState::WaitingInput(substream));
+                            self.keep_alive = KeepAlive::Yes;
+
+                            return Poll::Ready(ConnectionHandlerEvent::Custom(
+                                HandlerOutEvent::Message(message),
+                            ));
+                        }
+                        Poll::Ready(Some(Err(_))) => {
+                            // More serious errors, close this side of the stream. If the peer is
+                            // still around, they will re-establish their connection
+                            self.inbound_substream =
+                                Some(InboundSubstreamState::Closing(substream));
+                        }
+                        Poll::Ready(None) => {
+                            // Remote peer closed the stream
+                            self.inbound_substream =
+                                Some(InboundSubstreamState::Closing(substream));
+                        }
+                        Poll::Pending => {
+                            self.inbound_substream =
+                                Some(InboundSubstreamState::WaitingInput(substream));
+                            break;
+                        }
+                    }
+                }
+                Some(InboundSubstreamState::Closing(mut substream)) => {
+                    match Sink::poll_close(Pin::new(&mut substream), cx) {
+                        Poll::Ready(res) => {
+                            if res.is_err() {
+                                // Don't close the connection but just drop the inbound substream.
+                                // In case the remote has more to send, they will open up a new
+                                // substream.
+                                // @TODO: Log error here
+                            }
+
+                            self.inbound_substream = None;
+
+                            if self.outbound_substream.is_none() {
+                                self.keep_alive = KeepAlive::No;
+                            }
+
+                            break;
+                        }
+                        Poll::Pending => {
+                            self.inbound_substream =
+                                Some(InboundSubstreamState::Closing(substream));
+                            break;
+                        }
+                    }
+                }
+                None => {
+                    self.inbound_substream = None;
+                    break;
+                }
+                Some(InboundSubstreamState::Poisoned) => {
+                    unreachable!("Error occurred during inbound stream processing")
+                }
+            }
+        }
+
+        // Process outbound stream
+        loop {
+            match std::mem::replace(
+                &mut self.outbound_substream,
+                Some(OutboundSubstreamState::Poisoned),
+            ) {
+                Some(OutboundSubstreamState::WaitingOutput(substream)) => {
+                    match self.send_queue.pop_front() {
+                        Some(message) => {
+                            self.outbound_substream =
+                                Some(OutboundSubstreamState::PendingSend(substream, message));
+                        }
+                        None => {
+                            self.outbound_substream =
+                                Some(OutboundSubstreamState::WaitingOutput(substream));
+                            break;
+                        }
+                    }
+                }
+                Some(OutboundSubstreamState::PendingSend(mut substream, message)) => {
+                    match Sink::poll_ready(Pin::new(&mut substream), cx) {
+                        Poll::Ready(Ok(())) => {
+                            match Sink::start_send(Pin::new(&mut substream), message) {
+                                Ok(()) => {
+                                    self.outbound_substream =
+                                        Some(OutboundSubstreamState::PendingFlush(substream))
+                                }
+                                Err(err) => {
+                                    return Poll::Ready(ConnectionHandlerEvent::Close(
+                                        HandlerError::Codec(err),
+                                    ));
+                                }
+                            }
+                        }
+                        Poll::Ready(Err(err)) => {
+                            return Poll::Ready(ConnectionHandlerEvent::Close(
+                                HandlerError::Codec(err),
+                            ));
+                        }
+                        Poll::Pending => {
+                            self.outbound_substream =
+                                Some(OutboundSubstreamState::PendingSend(substream, message));
+                            break;
+                        }
+                    }
+                }
+                Some(OutboundSubstreamState::PendingFlush(mut substream)) => {
+                    match Sink::poll_flush(Pin::new(&mut substream), cx) {
+                        Poll::Ready(Ok(())) => {
+                            self.outbound_substream =
+                                Some(OutboundSubstreamState::WaitingOutput(substream))
+                        }
+                        Poll::Ready(Err(err)) => {
+                            return Poll::Ready(ConnectionHandlerEvent::Close(HandlerError::Codec(
+                                err,
+                            )))
+                        }
+                        Poll::Pending => {
+                            self.outbound_substream =
+                                Some(OutboundSubstreamState::PendingFlush(substream));
+                            break;
+                        }
+                    }
+                }
+                None => {
+                    self.outbound_substream = None;
+                    break;
+                }
+                Some(OutboundSubstreamState::Poisoned) => {
+                    unreachable!("Error occurred during outbound stream processing")
+                }
+            }
+        }
+
+        Poll::Pending
+    }
+}
diff --git a/aquadoggo/src/network/replication/mod.rs b/aquadoggo/src/network/replication/mod.rs
new file mode 100644
index 000000000..05593e59d
--- /dev/null
+++ b/aquadoggo/src/network/replication/mod.rs
@@ -0,0 +1,9 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+mod behaviour;
+mod handler;
+mod protocol;
+
+pub use behaviour::Behaviour;
+pub use handler::Handler;
+pub use protocol::{Codec, CodecError, Message, Protocol, PROTOCOL_NAME};
diff --git a/aquadoggo/src/network/replication/protocol.rs b/aquadoggo/src/network/replication/protocol.rs
new file mode 100644
index 000000000..f4192f6c2
--- /dev/null
+++ b/aquadoggo/src/network/replication/protocol.rs
@@ -0,0 +1,65 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use std::pin::Pin;
+
+use asynchronous_codec::{CborCodec, CborCodecError, Framed};
+use futures::{future, AsyncRead, AsyncWrite, Future};
+use libp2p::core::UpgradeInfo;
+use libp2p::{InboundUpgrade, OutboundUpgrade};
+use serde::{Deserialize, Serialize};
+
+pub const PROTOCOL_NAME: &[u8] = b"/p2p/p2panda/1.0.0";
+
+pub type CodecError = CborCodecError;
+
+pub type Codec = CborCodec<Message, Message>;
+
+// @TODO: Get this from our other replication module
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub enum Message {
+    Dummy(u64),
+}
+
+#[derive(Clone, Debug)]
+pub struct Protocol;
+
+impl Protocol {
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+impl UpgradeInfo for Protocol {
+    type Info = Vec<u8>;
+    type InfoIter = Vec<Self::Info>;
+
+    fn protocol_info(&self) -> Self::InfoIter {
+        vec![PROTOCOL_NAME.to_vec()]
+    }
+}
+
+impl<TSocket> InboundUpgrade<TSocket> for Protocol
+where
+    TSocket: AsyncRead + AsyncWrite + Unpin + Send + 'static,
+{
+    type Output = Framed<TSocket, Codec>;
+    type Error = CodecError;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Output, Self::Error>> + Send>>;
+
+    fn upgrade_inbound(self, socket: TSocket, _protocol_id: Self::Info) -> Self::Future {
+        Box::pin(future::ok(Framed::new(socket, Codec::new())))
+    }
+}
+
+impl<TSocket> OutboundUpgrade<TSocket> for Protocol
+where
+    TSocket: AsyncWrite + AsyncRead + Unpin + Send + 'static,
+{
+    type Output = Framed<TSocket, Codec>;
+    type Error = CodecError;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Output, Self::Error>> + Send>>;
+
+    fn upgrade_outbound(self, socket: TSocket, _protocol_id: Self::Info) -> Self::Future {
+        Box::pin(future::ok(Framed::new(socket, Codec::new())))
+    }
+}


Test patch: