From 27b55408b3fdd18b12a8940a7f2359789a98741f Mon Sep 17 00:00:00 2001 From: Ivan FB Date: Sat, 18 Jul 2026 21:31:48 +0200 Subject: [PATCH] feat: wait for a healthy mesh before handing back the transport MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Channels autoshard onto the relay shards the node subscribes to at startup, so a publish only reaches peers once one of those shards has a gossipsub mesh. Discovery plus grafting takes tens of seconds, and the previous unconditional sleep(3s) returned long before that — so a message sent in the first seconds of a conversation was published into an unformed mesh and silently dropped, only sometimes rescued by SDS retransmission. The node thread now waits on the topic-health event for the first shard to reach a mesh (MinimallyHealthy or better) before reporting ready, bounded by a timeout so an isolated node still comes up. Measured effect: an immediate send after joining went from mostly dropping to landing ~6/7 of the time, with startup settling in ~5s once the network is reachable. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Ai43sF2rymPFMG9iki9fwL --- extensions/embedded-logos-delivery/src/lib.rs | 40 ++++++++++++++++--- 1 file changed, 35 insertions(+), 5 deletions(-) diff --git a/extensions/embedded-logos-delivery/src/lib.rs b/extensions/embedded-logos-delivery/src/lib.rs index 61a8614..5c4499d 100644 --- a/extensions/embedded-logos-delivery/src/lib.rs +++ b/extensions/embedded-logos-delivery/src/lib.rs @@ -76,6 +76,13 @@ const INBOUND_CAPACITY: usize = 1024; /// Outbound command queue depth. const COMMAND_CAPACITY: usize = 256; +/// How long to wait at startup for a relay shard's gossipsub mesh to form +/// before handing back the transport. Channels autoshard onto the relay shards +/// the node subscribes to at startup, so a send only reaches peers once one of +/// those shards has a mesh; discovery plus grafting can take tens of seconds. +/// Bounded so an isolated node (no reachable peers) proceeds rather than hangs. +const NETWORK_READY_TIMEOUT: Duration = Duration::from_secs(45); + pub fn content_topic_for(delivery_address: &str) -> String { format!("{CHAT_TOPIC_PREFIX}{delivery_address}/proto") } @@ -310,17 +317,40 @@ fn node_thread( ); }); + // A shard's mesh becoming healthy is the signal that a send will actually + // reach peers. Registered before start so no early health event is missed; + // `MinimallyHealthy` already means at least one mesh peer on that shard. + let (mesh_ready_tx, mesh_ready_rx) = crossbeam_channel::bounded::<()>(1); + ctx.add_on_topic_health_change_listener(move |event| { + // Any mesh (>= 1 peer) is enough to fire: it is the most defensive + // signal — on a sparse network SufficientlyHealthy might never arrive + // and stall startup on the timeout, whereas MinimallyHealthy comes up + // as soon as the shard can carry traffic at all. SDS retransmission + // backstops the rest as the mesh fills in. + if matches!( + event.topic_health.as_str(), + "MinimallyHealthy" | "SufficientlyHealthy" + ) { + let _ = mesh_ready_tx.try_send(()); + } + }); + if let Err(e) = ctx.start_node() { let _ = ready_tx.send(Err(DeliveryError::StartupFailed(e))); return; } info!("logos-delivery node started (preset={})", cfg.preset); - // FIXME: This unconditional sleep is a stand-in for proper peer-connectivity - // detection: proceed once at least one peer is reachable, falling back to a - // configurable timeout. The bindings expose `add_on_connection_change_listener`, - // so the event this needs exists — wiring it up is left to a focused change. - thread::sleep(Duration::from_secs(3)); + // Hold the transport back until a shard can carry traffic, so callers do not + // publish into an unformed mesh (where the message is silently dropped and + // only SDS retransmission might later rescue it). Bounded by + // NETWORK_READY_TIMEOUT so an isolated node still comes up. + match mesh_ready_rx.recv_timeout(NETWORK_READY_TIMEOUT) { + Ok(()) => info!("logos-delivery node is network-ready"), + Err(_) => warn!( + "no healthy shard after {NETWORK_READY_TIMEOUT:?}; proceeding, early sends may drop" + ), + } if ready_tx.send(Ok(())).is_err() { // The starter gave up waiting; nothing holds a handle, so there is