From 8b7bf576d2b3ceb3e5b6cdae7a00989bebb3b4c4 Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 26 Oct 2026 18:44:09 +0100 Subject: add sync lanes --- src/swarm/peer/inbound.rs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) (limited to 'src/swarm/peer/inbound.rs') 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; -- cgit v1.3.1