1 files changed, 60 insertions, 0 deletions
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")
+            }
+        }
+    }
+}