add sync lanes

4 files changed, 75 insertions, 3 deletions

diff --git a/src/node.rs b/src/node.rs
index 9200d16..5b55c8d 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -136,13 +136,17 @@ impl Node {
         let Self { state, sync_rx } = self;
 
         let consumer_state = state.clone();
+        let lanes = Arc::new(peer::SyncLanes::start(&state));
         tokio::spawn(async move {
             let inner_state = consumer_state.clone();
             let _ = consumer_state
                 .transport
                 .start_consumer(Box::new(move |sender, msg, _payload_len| {
                     let state_clone = inner_state.clone();
-                    Box::pin(async move { peer::handle_message(&state_clone, sender, msg).await })
+                    let lanes = Arc::clone(&lanes);
+                    Box::pin(async move {
+                        peer::handle_message(&state_clone, &lanes, sender, msg).await
+                    })
                 }))
                 .await;
         });
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index 684a358..73cc629 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -12,12 +12,17 @@ use super::handshake::{
     discover_peers, ensure_handshake, establish_peer, peer_swarm, remember_peer, send_hello_reply,
 };
 use super::keepalive::{heal_if_diverged, local_keepalive, note_keepalive};
+use super::lanes::SyncLanes;
 use crate::state::AppState;
-use crate::swarm::sync::{SyncEvent, handle_sync};
+use crate::swarm::sync::SyncEvent;
 use crate::transport::{self, ParsedMessage};
 
+/// Handles `message` from `peer` in arrival order, except that a sync message is only
+/// queued here, on `lanes`: everything before it (admission, liveness, the peer's epoch)
+/// is done in order, applying it isn't.
 pub(crate) async fn handle_message(
     state: &AppState,
+    lanes: &SyncLanes,
     peer: String,
     message: transport::ParsedMessage,
 ) -> Result<()> {
@@ -130,7 +135,7 @@ pub(crate) async fn handle_message(
                 return Ok(());
             }
             crate::swarm::source::synced_from(state, &peer);
-            handle_sync(state, &peer, envelope).await?;
+            lanes.enqueue(&peer, envelope);
         }
         transport::ParsedMessage::Blob(blob) => {
             crate::swarm::blob::handle(state, &peer, blob).await;
diff --git a/src/swarm/peer/lanes.rs b/src/swarm/peer/lanes.rs
new file mode 100644
index 0000000..88639ea
--- /dev/null
+++ b/src/swarm/peer/lanes.rs
@@ -0,0 +1,60 @@
+//! Sync lanes: inbound sync messages are applied off the consumer, so one slow apply (a
+//! large delta takes seconds) doesn't hold up keepalives, HELLOs and blob chunks behind
+//! it. Each sender's messages go to the same lane, in arrival order; a few lanes are
+//! enough, since applying takes the document's write lock anyway.
+
+use std::collections::hash_map::DefaultHasher;
+use std::hash::{Hash, Hasher};
+
+use tokio::sync::mpsc::{self, error::TrySendError};
+use tracing::{debug, warn};
+
+use crate::state::AppState;
+use crate::swarm::sync::handle_sync;
+use crate::transport::SyncEnvelope;
+
+const LANES: usize = 4;
+/// Sync messages waiting per lane. A full lane drops what arrives: its sender gets no
+/// answer and resends after its reply timeout, by which time the backlog is gone.
+const QUEUED_PER_LANE: usize = 256;
+
+pub(crate) struct SyncLanes {
+    lanes: Vec<mpsc::Sender<(String, SyncEnvelope)>>,
+}
+
+impl SyncLanes {
+    /// Starts a task per lane.
+    pub(crate) fn start(state: &AppState) -> Self {
+        let lanes = (0..LANES)
+            .map(|_| {
+                let (tx, mut rx) = mpsc::channel::<(String, SyncEnvelope)>(QUEUED_PER_LANE);
+                let state = state.clone();
+                tokio::spawn(async move {
+                    while let Some((peer, envelope)) = rx.recv().await {
+                        if let Err(error) = handle_sync(&state, &peer, envelope).await {
+                            warn!(peer = %peer, error = %format!("{error:#}"), "Failed to process sync message");
+                        }
+                    }
+                });
+                tx
+            })
+            .collect();
+        Self { lanes }
+    }
+
+    /// Queues `envelope` from `peer` on its lane.
+    pub(crate) fn enqueue(&self, peer: &str, envelope: SyncEnvelope) {
+        let mut hasher = DefaultHasher::new();
+        peer.hash(&mut hasher);
+        let lane = &self.lanes[hasher.finish() as usize % LANES];
+        match lane.try_send((peer.to_owned(), envelope)) {
+            Ok(()) => {}
+            Err(TrySendError::Full(_)) => {
+                debug!(peer = %peer, "Sync lane full; dropping the message (its sender resends)")
+            }
+            Err(TrySendError::Closed(_)) => {
+                warn!(peer = %peer, "Sync lane stopped; dropping the message")
+            }
+        }
+    }
+}
diff --git a/src/swarm/peer/mod.rs b/src/swarm/peer/mod.rs
index f0222c4..2e5c1dc 100644
--- a/src/swarm/peer/mod.rs
+++ b/src/swarm/peer/mod.rs
@@ -8,6 +8,7 @@
 //! * `forget`: swarm-wide forget records.
 //! * `worker`: the periodic peer worker.
 //! * `inbound`: dispatch of inbound messages.
+//! * `lanes`: sync messages applied off the consumer.
 
 mod admit;
 mod epoch;
@@ -15,6 +16,7 @@ mod forget;
 mod handshake;
 mod inbound;
 mod keepalive;
+mod lanes;
 mod worker;
 
 pub(crate) use forget::{forget_unless_alive, forgotten_path};
@@ -23,4 +25,5 @@ pub(crate) use handshake::{
 };
 pub(crate) use inbound::handle_message;
 pub(crate) use keepalive::send_ping;
+pub(crate) use lanes::SyncLanes;
 pub(crate) use worker::run_peer_worker;