From cbfecb7a3aff64d7de4b9a08ac12f9a8a5b1bc61 Mon Sep 17 00:00:00 2001 From: Haiko Schol Date: Wed, 18 Feb 2026 19:58:22 +0700 Subject: [PATCH 1/4] Fix WebRTC notification handshake deadlock in multi-stream task loop --- light-base/src/network_service/tasks.rs | 23 +++++++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/light-base/src/network_service/tasks.rs b/light-base/src/network_service/tasks.rs index a9934721ec..e404906995 100644 --- a/light-base/src/network_service/tasks.rs +++ b/light-base/src/network_service/tasks.rs @@ -20,7 +20,7 @@ use crate::{ platform::{PlatformRef, SubstreamDirection}, }; -use alloc::{boxed::Box, string::String}; +use alloc::{boxed::Box, string::String, sync::Arc}; use core::{pin, time::Duration}; use futures_lite::FutureExt as _; use futures_util::{StreamExt as _, future, stream::FuturesUnordered}; @@ -251,6 +251,11 @@ pub(super) async fn webrtc_multi_stream_connection_task( let mut next_substream_id = 0; // We need to pin the receiver, as the type doesn't implement `Unpin`. let mut coordinator_to_connection = pin::pin!(coordinator_to_connection); + // Shared event used to wake up all substream wait futures when + // `inject_coordinator_message` queues write data (e.g. AcceptInNotifications). + // Without this, the substream would never be re-polled since the platform's + // `wait_read_write_again` only wakes on incoming data or timers. + let coordinator_write_ready = Arc::new(event_listener::Event::new()); loop { // Start opening new outbound substreams, if needed. @@ -347,6 +352,12 @@ pub(super) async fn webrtc_multi_stream_connection_task( match wake_up_reason { WakeUpReason::CoordinatorMessage(message) => { connection_task.inject_coordinator_message(&platform.now(), message); + // Wake up all substream wait futures so they get re-polled. + // This is necessary because inject_coordinator_message may have + // queued write data (e.g. AcceptInNotifications queues a handshake + // response) but the platform's wait_read_write_again only wakes on + // incoming data or timers -- not on pending writes. + coordinator_write_ready.notify(usize::MAX); } WakeUpReason::CoordinatorDead => { log!( @@ -453,8 +464,16 @@ pub(super) async fn webrtc_multi_stream_connection_task( if let SubstreamFate::Continue = substream_fate { when_substreams_rw_ready.push({ let platform = platform.clone(); + let write_ready = coordinator_write_ready.clone(); Box::pin(async move { - platform.wait_read_write_again(socket.as_mut()).await; + // Wait for either platform data or a coordinator write + // notification (e.g. after AcceptInNotifications queues a + // handshake response that needs to be flushed). + let write_listener = write_ready.listen(); + platform + .wait_read_write_again(socket.as_mut()) + .or(write_listener) + .await; (socket, substream_id) }) }); From d09ad8a4c642064256050d8a99173cd020955845 Mon Sep 17 00:00:00 2001 From: Timothy Wu Date: Tue, 24 Feb 2026 23:29:58 -0500 Subject: [PATCH 2/4] Fix warp sync initialization and UnknownTargetBlock ban during initial sync Initialize warp sync source finalized_block_height with best_block_number from gossip handshake instead of 0, so warp sync triggers immediately without waiting for a GrandPa neighbor packet. Also exclude UnknownTargetBlock justification errors from the 40s ban, as they are benign during initial catchup sync. --- lib/src/sync/all.rs | 22 +++++++++++++++++++--- lib/src/sync/warp_sync.rs | 12 +++++++----- light-base/src/sync_service/standalone.rs | 6 ++++++ 3 files changed, 32 insertions(+), 8 deletions(-) diff --git a/lib/src/sync/all.rs b/lib/src/sync/all.rs index 4d429507a7..f64765b28f 100644 --- a/lib/src/sync/all.rs +++ b/lib/src/sync/all.rs @@ -413,6 +413,7 @@ impl AllSync { all_forks, slab_insertion: self.shared.sources.vacant_entry(), warp_sync: &mut self.warp_sync, + best_block_number, marker: PhantomData, }) } @@ -421,6 +422,7 @@ impl AllSync { all_forks, slab_insertion: self.shared.sources.vacant_entry(), warp_sync: &mut self.warp_sync, + best_block_number, marker: PhantomData, }) } @@ -429,6 +431,7 @@ impl AllSync { all_forks, slab_insertion: self.shared.sources.vacant_entry(), warp_sync: &mut self.warp_sync, + best_block_number, marker: PhantomData, }) } @@ -437,6 +440,7 @@ impl AllSync { all_forks, slab_insertion: self.shared.sources.vacant_entry(), warp_sync: &mut self.warp_sync, + best_block_number, marker: PhantomData, }) } @@ -1485,6 +1489,7 @@ pub struct AddSourceOldBlock<'a, TRq, TSrc, TBl> { all_forks: all_forks::AddSourceOldBlock<'a, Option, AllForksRequestExtra, AllForksSourceExtra>, warp_sync: &'a mut Option>, + best_block_number: u64, marker: PhantomData, } @@ -1503,7 +1508,10 @@ impl<'a, TRq, TSrc, TBl> AddSourceOldBlock<'a, TRq, TSrc, TBl> { .add_source(AllForksSourceExtra { outer_source_id }); let warp_sync_source_id = if let Some(warp_sync) = self.warp_sync { - Some(warp_sync.add_source(WarpSyncSourceExtra { outer_source_id })) + Some(warp_sync.add_source( + WarpSyncSourceExtra { outer_source_id }, + self.best_block_number, + )) } else { None }; @@ -1526,6 +1534,7 @@ pub struct AddSourceKnown<'a, TRq, TSrc, TBl> { all_forks: all_forks::AddSourceKnown<'a, Option, AllForksRequestExtra, AllForksSourceExtra>, warp_sync: &'a mut Option>, + best_block_number: u64, marker: PhantomData, } @@ -1552,7 +1561,10 @@ impl<'a, TRq, TSrc, TBl> AddSourceKnown<'a, TRq, TSrc, TBl> { .add_source(AllForksSourceExtra { outer_source_id }); let warp_sync_source_id = if let Some(warp_sync) = self.warp_sync { - Some(warp_sync.add_source(WarpSyncSourceExtra { outer_source_id })) + Some(warp_sync.add_source( + WarpSyncSourceExtra { outer_source_id }, + self.best_block_number, + )) } else { None }; @@ -1575,6 +1587,7 @@ pub struct AddSourceUnknown<'a, TRq, TSrc, TBl> { all_forks: all_forks::AddSourceUnknown<'a, Option, AllForksRequestExtra, AllForksSourceExtra>, warp_sync: &'a mut Option>, + best_block_number: u64, marker: PhantomData, } @@ -1601,7 +1614,10 @@ impl<'a, TRq, TSrc, TBl> AddSourceUnknown<'a, TRq, TSrc, TBl> { ); let warp_sync_source_id = if let Some(warp_sync) = self.warp_sync { - Some(warp_sync.add_source(WarpSyncSourceExtra { outer_source_id })) + Some(warp_sync.add_source( + WarpSyncSourceExtra { outer_source_id }, + self.best_block_number, + )) } else { None }; diff --git a/lib/src/sync/warp_sync.rs b/lib/src/sync/warp_sync.rs index 2dcea3b69b..b3456786b7 100644 --- a/lib/src/sync/warp_sync.rs +++ b/lib/src/sync/warp_sync.rs @@ -640,15 +640,17 @@ impl WarpSync { /// Add a source to the list of sources. /// - /// The source has a finalized block height of 0, which should later be updated using - /// [`WarpSync::set_source_finality_state`]. - pub fn add_source(&mut self, user_data: TSrc) -> SourceId { + /// `best_block_number` is used as the initial finalized block height for this source. + /// It can later be updated using [`WarpSync::set_source_finality_state`]. + pub fn add_source(&mut self, user_data: TSrc, best_block_number: u64) -> SourceId { let source_id = SourceId(self.sources.insert(Source { user_data, - finalized_block_height: 0, + finalized_block_height: best_block_number, })); - let _inserted = self.sources_by_finalized_height.insert((0, source_id)); + let _inserted = self + .sources_by_finalized_height + .insert((best_block_number, source_id)); debug_assert!(_inserted); debug_assert!(self.sources.len() >= self.sources_by_finalized_height.len()); diff --git a/light-base/src/sync_service/standalone.rs b/light-base/src/sync_service/standalone.rs index b5d39b75a0..eccd789d69 100644 --- a/light-base/src/sync_service/standalone.rs +++ b/light-base/src/sync_service/standalone.rs @@ -577,9 +577,15 @@ pub(super) async fn start_standalone_chain( // Errors of type `JustificationEngineMismatch` indicate that the chain // uses a finality engine that smoldot doesn't recognize. This is a benign // error that shouldn't lead to a ban. + // Errors of type `UnknownTargetBlock` can happen during initial sync when + // a justification targets a block not yet chain-verified from genesis. + // This is benign and shouldn't lead to a ban. if !matches!( error, all::JustificationVerifyError::JustificationEngineMismatch + | all::JustificationVerifyError::FinalityVerify( + smoldot::chain::blocks_tree::FinalityVerifyError::UnknownTargetBlock { .. } + ) ) { log!( &task.platform, From 84fdf0eb1b3bcfbef9739bf670f92c198230c659 Mon Sep 17 00:00:00 2001 From: Timothy Wu Date: Wed, 25 Feb 2026 17:45:45 -0500 Subject: [PATCH 3/4] Add diagnostic logging and deferred notification retry mechanism Add substream_id to connection-activity log in multi-stream task loop to identify which WebRTC data channel each read/write belongs to. Add GossipInboundResult event to surface inbound notification substream outcomes (accepted, rejected duplicate, rejected cold-open) for debugging. Replace immediate Tx/Grandpa substream retry-on-failure with a 2-second deferred retry to avoid starving WebRTC connections with rapid substream open attempts. --- full-node/src/network_service.rs | 30 ++- lib/src/network/service.rs | 268 ++++++++++++++++-------- light-base/src/network_service.rs | 33 ++- light-base/src/network_service/tasks.rs | 1 + 4 files changed, 243 insertions(+), 89 deletions(-) diff --git a/full-node/src/network_service.rs b/full-node/src/network_service.rs index ff3f7db8ce..318be673a5 100644 --- a/full-node/src/network_service.rs +++ b/full-node/src/network_service.rs @@ -869,12 +869,15 @@ async fn background_task(mut inner: Inner) { CanStartConnect(PeerId), CanOpenGossip(PeerId, ChainId), StartKademliaDiscoveries, + NotificationRetryReady, MessageToConnection { connection_id: service::ConnectionId, message: service::CoordinatorToConnection, }, } + let notification_retry_time = inner.network.next_notification_retry_time().cloned(); + let wake_up_reason = async { inner .to_background_rx @@ -890,7 +893,7 @@ async fn background_task(mut inner: Inner) { let num_pending_out_attempts = &inner.num_pending_out_attempts; async move { if let Some(event) = (event_senders_ready && event_pending_send.is_none()) - .then(|| network.next_event()) + .then(|| network.next_event(&Instant::now())) .flatten() { WakeUpReason::NetworkEvent(event) @@ -986,6 +989,14 @@ async fn background_task(mut inner: Inner) { socket_addr, } }) + .or(async { + if let Some(retry_time) = notification_retry_time { + smol::Timer::at(retry_time).await; + WakeUpReason::NotificationRetryReady + } else { + future::pending().await + } + }) .await; match wake_up_reason { @@ -1066,6 +1077,10 @@ async fn background_task(mut inner: Inner) { ))); } + WakeUpReason::NotificationRetryReady => { + // The event loop will restart, calling next_event() which + // processes ready pending retries at the top. + } WakeUpReason::StartKademliaDiscoveries => { for chain_id in inner.network.chains().collect::>() { let random_peer_id = @@ -1778,6 +1793,19 @@ async fn background_task(mut inner: Inner) { // that this event can't happen. unreachable!() } + WakeUpReason::NetworkEvent(service::Event::GossipInboundResult { + peer_id, + chain_id, + outcome, + }) => { + inner.log_callback.log( + LogLevel::Debug, + format!( + "gossip-in-result; chain={}; peer_id={peer_id}; outcome={outcome:?}", + inner.network[chain_id].log_name, + ), + ); + } WakeUpReason::NetworkEvent(service::Event::RequestResult { substream_id, peer_id, diff --git a/lib/src/network/service.rs b/lib/src/network/service.rs index d5a5323c17..3bd8627c4c 100644 --- a/lib/src/network/service.rs +++ b/lib/src/network/service.rs @@ -233,11 +233,23 @@ pub struct ChainNetwork { NotificationsSubstreamState, collection::SubstreamId, )>, + + /// Pending outbound notification substream retries. When a Tx/Grandpa substream + /// is refused, a retry entry is stored here with a delay. On each `next_event()` + /// call, entries whose timer has elapsed are processed. + pending_notification_out_retries: Vec>, } #[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] struct PeerIndex(usize); +struct PendingNotificationOutRetry { + retry_after: TNow, + protocol: NotificationsProtocol, + peer_index: PeerIndex, + connection_id: collection::ConnectionId, +} + struct Chain { /// See [`ChainConfig::block_number_bytes`]. block_number_bytes: usize, @@ -379,6 +391,7 @@ where ), connections_by_peer_id: BTreeSet::new(), notification_substreams_by_peer_id: BTreeSet::new(), + pending_notification_out_retries: Vec::new(), gossip_desired_peers_by_chain: BTreeSet::new(), gossip_desired_peers: BTreeSet::new(), unconnected_desired: hashbrown::HashSet::with_capacity_and_hasher( @@ -1135,8 +1148,106 @@ where self.inner.inject_connection_message(connection_id, message) } + /// Returns the earliest time at which a pending notification substream retry + /// should be processed, or `None` if no retries are pending. + /// The caller should use this to set up an async timer to wake the event loop. + pub fn next_notification_retry_time(&self) -> Option<&TNow> { + self.pending_notification_out_retries + .iter() + .map(|r| &r.retry_after) + .min() + } + /// Returns the next event produced by the service. - pub fn next_event(&mut self) -> Option> { + pub fn next_event(&mut self, now: &TNow) -> Option> { + // Process pending notification substream retries whose timers have elapsed. + let mut retry_idx = 0; + while retry_idx < self.pending_notification_out_retries.len() { + if *now < self.pending_notification_out_retries[retry_idx].retry_after { + retry_idx += 1; + continue; + } + + let retry = self.pending_notification_out_retries.swap_remove(retry_idx); + + // Skip if connection is shutting down. + if self + .inner + .connection_state(retry.connection_id) + .shutting_down + { + continue; + } + + // Verify the block announce substream is still open for this peer. + let chain_index = match retry.protocol { + NotificationsProtocol::Transactions { chain_index } + | NotificationsProtocol::Grandpa { chain_index } => chain_index, + _ => continue, + }; + let has_block_announce = self + .notification_substreams_by_peer_id + .range( + ( + NotificationsProtocol::BlockAnnounces { chain_index }, + retry.peer_index, + SubstreamDirection::Out, + NotificationsSubstreamState::MIN, + SubstreamId::MIN, + ) + ..=( + NotificationsProtocol::BlockAnnounces { chain_index }, + retry.peer_index, + SubstreamDirection::Out, + NotificationsSubstreamState::MAX, + SubstreamId::MAX, + ), + ) + .any(|(_, _, _, state, _)| { + matches!(state, NotificationsSubstreamState::Open { .. }) + }); + if !has_block_announce { + continue; + } + + // Re-open the outbound notification substream. + let new_substream_id = self.inner.open_out_notifications( + retry.connection_id, + codec::encode_protocol_name_string(match retry.protocol { + NotificationsProtocol::Transactions { .. } => { + codec::ProtocolName::Transactions { + genesis_hash: self.chains[chain_index].genesis_hash, + fork_id: self.chains[chain_index].fork_id.as_deref(), + } + } + NotificationsProtocol::Grandpa { .. } => codec::ProtocolName::Grandpa { + genesis_hash: self.chains[chain_index].genesis_hash, + fork_id: self.chains[chain_index].fork_id.as_deref(), + }, + _ => unreachable!(), + }), + self.notifications_protocol_handshake_timeout(retry.protocol), + self.notifications_protocol_handshake(retry.protocol), + self.notifications_protocol_max_handshake_size(retry.protocol), + ); + let _was_inserted = self.notification_substreams_by_peer_id.insert(( + retry.protocol, + retry.peer_index, + SubstreamDirection::Out, + NotificationsSubstreamState::Pending, + new_substream_id, + )); + debug_assert!(_was_inserted); + let _prev_value = self.substreams.insert( + new_substream_id, + SubstreamInfo { + connection_id: retry.connection_id, + protocol: Some(Protocol::Notifications(retry.protocol)), + }, + ); + debug_assert!(_prev_value.is_none()); + } + loop { let inner_event = self.inner.next_event()?; match inner_event { @@ -2075,10 +2186,10 @@ where .is_some() ); - // If the substream failed to open, we simply try again. - // Trying again means that we might be hammering the remote with - // substream requests, however as of the writing of this text this is - // necessary in order to bypass an issue in Substrate. + // If the substream failed to open, schedule a retry after + // a delay. This avoids hammering the remote with substream + // requests, which on WebRTC can starve the connection and + // prevent other traffic from flowing. // Note that in the situation where the connection is shutting down, // we don't re-open the substream on a different connection, but // that's ok as the block announces substream should be closed soon. @@ -2087,52 +2198,14 @@ where continue; } - let new_substream_id = self.inner.open_out_notifications( - connection_id, - codec::encode_protocol_name_string(match substream_protocol { - NotificationsProtocol::Transactions { .. } => { - codec::ProtocolName::Transactions { - genesis_hash: self.chains[chain_index].genesis_hash, - fork_id: self.chains[chain_index] - .fork_id - .as_deref(), - } - } - NotificationsProtocol::Grandpa { .. } => { - codec::ProtocolName::Grandpa { - genesis_hash: self.chains[chain_index].genesis_hash, - fork_id: self.chains[chain_index] - .fork_id - .as_deref(), - } - } - _ => unreachable!(), - }), - self.notifications_protocol_handshake_timeout( - substream_protocol, - ), - self.notifications_protocol_handshake(substream_protocol), - self.notifications_protocol_max_handshake_size( - substream_protocol, - ), - ); - let _was_inserted = - self.notification_substreams_by_peer_id.insert(( - substream_protocol, + self.pending_notification_out_retries.push( + PendingNotificationOutRetry { + retry_after: now.clone() + Duration::from_secs(2), + protocol: substream_protocol, peer_index, - SubstreamDirection::Out, - NotificationsSubstreamState::Pending, - new_substream_id, - )); - debug_assert!(_was_inserted); - let _prev_value = self.substreams.insert( - new_substream_id, - SubstreamInfo { connection_id, - protocol: substream_info.protocol, }, ); - debug_assert!(_prev_value.is_none()); continue; } @@ -2147,6 +2220,12 @@ where )); debug_assert!(_was_inserted); + // Clear any pending retries on success. + self.pending_notification_out_retries.retain(|r| { + !(r.protocol == substream_protocol + && r.peer_index == peer_index) + }); + // In case of Grandpa, we immediately send a neighbor packet with // the current local state. if matches!(substream_protocol, NotificationsProtocol::Grandpa { .. }) { @@ -2283,6 +2362,17 @@ where debug_assert!(_was_inserted); } + // Clear pending retries for Tx/Grandpa substreams on gossip + // disconnect. + self.pending_notification_out_retries.retain(|r| { + !matches!( + r.protocol, + NotificationsProtocol::Transactions { chain_index: ci } + | NotificationsProtocol::Grandpa { chain_index: ci } + if ci == chain_index && r.peer_index == peer_index + ) + }); + // The transactions and Grandpa protocols are tied to the block // announces substream. As such, we also close any transactions or // grandpa substream, either pending or fully opened. @@ -2388,52 +2478,22 @@ where } // The transactions and grandpa protocols are tied to the block announces - // substream. If there is a block announce substream with the peer, we try - // to reopen these two substreams. + // substream. If there is a block announce substream with the peer, we + // schedule a retry after a delay to reopen these substreams. NotificationsProtocol::Transactions { .. } | NotificationsProtocol::Grandpa { .. } => { - // Don't actually try to reopen if the connection is shutting down. - // Note that we don't try to reopen on a different connection, as the - // block announces substream will very soon be closed too anyway. if self.inner.connection_state(connection_id).shutting_down { continue; } - let new_substream_id = self.inner.open_out_notifications( - connection_id, - codec::encode_protocol_name_string(match substream_protocol { - NotificationsProtocol::Transactions { chain_index } => { - codec::ProtocolName::Transactions { - genesis_hash: self.chains[chain_index].genesis_hash, - fork_id: self.chains[chain_index].fork_id.as_deref(), - } - } - NotificationsProtocol::Grandpa { chain_index } => { - codec::ProtocolName::Grandpa { - genesis_hash: self.chains[chain_index].genesis_hash, - fork_id: self.chains[chain_index].fork_id.as_deref(), - } - } - _ => unreachable!(), - }), - self.notifications_protocol_handshake_timeout(substream_protocol), - self.notifications_protocol_handshake(substream_protocol), - self.notifications_protocol_max_handshake_size(substream_protocol), - ); - self.substreams.insert( - new_substream_id, - SubstreamInfo { + self.pending_notification_out_retries.push( + PendingNotificationOutRetry { + retry_after: now.clone() + Duration::from_secs(2), + protocol: substream_protocol, + peer_index, connection_id, - protocol: Some(Protocol::Notifications(substream_protocol)), }, ); - self.notification_substreams_by_peer_id.insert(( - substream_protocol, - peer_index, - SubstreamDirection::Out, - NotificationsSubstreamState::Pending, - new_substream_id, - )); } } } @@ -2508,7 +2568,11 @@ where self.inner.reject_in_notifications(substream_id); let _was_removed = self.substreams.remove(&substream_id); debug_assert!(_was_removed.is_some()); - continue; + return Some(Event::GossipInboundResult { + peer_id: self.peers[peer_index.0].clone(), + chain_id: ChainId(chain_index), + outcome: GossipInboundOutcome::RejectedDuplicate, + }); } // If an outgoing block announces notifications protocol (either pending or @@ -2549,7 +2613,11 @@ where self.notifications_protocol_handshake(substream_protocol), self.notifications_protocol_max_notification_size(substream_protocol), ); - continue; + return Some(Event::GossipInboundResult { + peer_id: self.peers[peer_index.0].clone(), + chain_id: ChainId(chain_index), + outcome: GossipInboundOutcome::Accepted, + }); } // It is forbidden to cold-open a substream other than the block announces @@ -2561,7 +2629,11 @@ where self.inner.reject_in_notifications(substream_id); let _was_removed = self.substreams.remove(&substream_id); debug_assert!(_was_removed.is_some()); - continue; + return Some(Event::GossipInboundResult { + peer_id: self.peers[peer_index.0].clone(), + chain_id: ChainId(chain_index), + outcome: GossipInboundOutcome::RejectedColdOpen, + }); } // Update the local state and return the event. @@ -4424,12 +4496,34 @@ pub enum Event { /// This [`SubstreamId`] is considered dead and no longer valid. substream_id: SubstreamId, }, + + /// An inbound notification substream was received from a remote peer and has been + /// processed. This is a diagnostic event indicating the outcome. + GossipInboundResult { + /// Peer that opened the inbound substream. + peer_id: PeerId, + /// Chain of the notification substream. + chain_id: ChainId, + /// Outcome of processing the inbound substream. + outcome: GossipInboundOutcome, + }, /*Transactions { peer_id: PeerId, transactions: EncodedTransactions, }*/ } +/// See [`Event::GossipInboundResult`]. +#[derive(Debug)] +pub enum GossipInboundOutcome { + /// Auto-accepted because an outbound substream already exists. + Accepted, + /// Rejected because a duplicate inbound substream already exists. + RejectedDuplicate, + /// Rejected because non-block-announces substreams cannot be cold-opened. + RejectedColdOpen, +} + /// See [`Event::ProtocolError`]. // TODO: reexport these error types #[derive(Debug, derive_more::Display, derive_more::Error)] diff --git a/light-base/src/network_service.rs b/light-base/src/network_service.rs index 6f5cebe4ed..8757a87e88 100644 --- a/light-base/src/network_service.rs +++ b/light-base/src/network_service.rs @@ -957,6 +957,7 @@ async fn background_task(mut task: BackgroundTask) { NetworkEvent(service::Event>), CanAssignSlot(PeerId, ChainId), NextRecentConnectionRestore, + NotificationRetryReady, CanStartConnect(PeerId), CanOpenGossip(PeerId, ChainId), MessageFromConnection { @@ -971,6 +972,8 @@ async fn background_task(mut task: BackgroundTask) { StartDiscovery(ChainId), } + let notification_retry_time = task.network.next_notification_retry_time().cloned(); + let wake_up_reason = { let message_received = async { task.main_messages_rx @@ -998,7 +1001,7 @@ async fn background_task(mut task: BackgroundTask) { let service_event = async { if let Some(event) = (task.event_pending_send.is_none() && task.pending_new_subscriptions.is_empty()) - .then(|| task.network.next_event()) + .then(|| task.network.next_event(&task.platform.now())) .flatten() { WakeUpReason::NetworkEvent(event) @@ -1110,6 +1113,14 @@ async fn background_task(mut task: BackgroundTask) { let ((_, chain_id), _) = next_discovery.remove_entry(); WakeUpReason::StartDiscovery(chain_id) }; + let notification_retry = async { + if let Some(retry_time) = notification_retry_time { + task.platform.sleep_until(retry_time).await; + WakeUpReason::NotificationRetryReady + } else { + future::pending().await + } + }; message_for_chain_received .or(message_received) @@ -1118,6 +1129,7 @@ async fn background_task(mut task: BackgroundTask) { .or(next_recent_connection_restore) .or(finished_sending_event) .or(start_discovery) + .or(notification_retry) .await }; @@ -2542,6 +2554,21 @@ async fn background_task(mut task: BackgroundTask) { // Can't happen as we already instantaneously accept or reject gossip in requests. unreachable!() } + WakeUpReason::NetworkEvent(service::Event::GossipInboundResult { + peer_id, + chain_id, + outcome, + }) => { + log!( + &task.platform, + Debug, + "network", + "gossip-in-result", + chain = &task.network[chain_id].log_name, + peer_id, + ?outcome, + ); + } WakeUpReason::NetworkEvent(service::Event::IdentifyRequestIn { peer_id, substream_id, @@ -2646,6 +2673,10 @@ async fn background_task(mut task: BackgroundTask) { task.num_recent_connection_opening = task.num_recent_connection_opening.saturating_sub(1); } + WakeUpReason::NotificationRetryReady => { + // The event loop will restart, calling next_event() which + // processes ready pending retries at the top. + } WakeUpReason::CanStartConnect(expected_peer_id) => { let Some(multiaddr) = task .peering_strategy diff --git a/light-base/src/network_service/tasks.rs b/light-base/src/network_service/tasks.rs index e404906995..369a03b02e 100644 --- a/light-base/src/network_service/tasks.rs +++ b/light-base/src/network_service/tasks.rs @@ -392,6 +392,7 @@ pub(super) async fn webrtc_multi_stream_connection_task( "connections", "connection-activity", address = address_string, + substream_id, read = socket_read_write.read_bytes - read_bytes_before, written = socket_read_write.write_bytes_queued - written_bytes_before, wake_up_after = ?socket_read_write.wake_up_after.as_ref().map(|w| { From c417fbe7a58b3e669fbb2903180ce27a11146522 Mon Sep 17 00:00:00 2001 From: Timothy Wu Date: Tue, 3 Mar 2026 23:09:16 -0500 Subject: [PATCH 4/4] Add WebRTC diagnostic logging and fix inbound data channel handling Route browser-side WebRTC diagnostics through smoldot's log system via a new logCallback field on ConnectionConfig. Add a hasNegotiated guard to prevent unnecessary SDP re-negotiation in webrtc-direct mode, and handle already-open inbound data channels whose onopen event was missed. --- wasm-node/javascript/src/internals/client.ts | 8 +++ .../src/no-auto-bytecode-browser.ts | 61 ++++++++++++++++++- 2 files changed, 66 insertions(+), 3 deletions(-) diff --git a/wasm-node/javascript/src/internals/client.ts b/wasm-node/javascript/src/internals/client.ts index 626140dfb1..33e6023843 100644 --- a/wasm-node/javascript/src/internals/client.ts +++ b/wasm-node/javascript/src/internals/client.ts @@ -136,6 +136,13 @@ export interface ConnectionConfig { */ address: instance.ParsedMultiaddr, + /** + * Optional callback for platform-level diagnostic logging. + * Uses the same signature as the smoldot log callback: + * level 1=error, 2=warn, 3=info, 4=debug, 5=trace. + */ + logCallback?: (level: number, target: string, message: string) => void; + /** * Callback called when a multistream connection knows information about its handshake. Should * be called as soon as possible. @@ -342,6 +349,7 @@ export function start(options: ClientOptions, wasmModule: SmoldotBytecode | Prom const connectionId = event.connectionId; state.connections.set(connectionId, platformBindings.connect({ address: event.address, + logCallback, onConnectionReset(message) { if (state.instance.status !== "ready") throw new Error(); diff --git a/wasm-node/javascript/src/no-auto-bytecode-browser.ts b/wasm-node/javascript/src/no-auto-bytecode-browser.ts index 9a5bd9608c..4b5e87cd6d 100644 --- a/wasm-node/javascript/src/no-auto-bytecode-browser.ts +++ b/wasm-node/javascript/src/no-auto-bytecode-browser.ts @@ -194,6 +194,17 @@ function connect(config: ConnectionConfig): Connection { const { targetPort, ipVersion, targetIp, remoteTlsCertificateSha256 } = config.address; + // Log helper: routes through smoldot's log system if available, falls back to console. + const log = (level: number, message: string) => { + if (config.logCallback) { + config.logCallback(level, 'webrtc-platform', message); + } else if (level <= 2) { + console.warn('[webrtc-platform] ' + message); + } else { + console.debug('[webrtc-platform] ' + message); + } + }; + const state: { // Note that `pc` can be the connection, but also null or undefined. // `undefined` means "certificate generation in progress", while `null` means "opening must @@ -208,11 +219,19 @@ function connect(config: ConnectionConfig): Connection { // Set to `true` before any outbound substream is open. Used to detect when the first // substream is opened. isFirstOutSubstream: boolean, + // Set to `true` after the first SDP offer/answer exchange completes. + // In webrtc-direct, re-negotiation is never needed because the SDP answer + // is static (derived from the remote's multiaddress and certificate). + hasNegotiated: boolean, + // Counter for inbound data channels received (for diagnostics). + inboundChannelCount: number, } = { pc: undefined, dataChannels: new Map(), nextStreamId: 0, isFirstOutSubstream: true, + hasNegotiated: false, + inboundChannelCount: 0, }; // Kills all the JavaScript objects (the connection and all its substreams), ensuring that no @@ -250,19 +269,35 @@ function connect(config: ConnectionConfig): Connection { state.nextStreamId += 1; dataChannel.binaryType = 'arraybuffer'; + log(4, `addChannel; streamId=${streamId}, direction=${direction}, channelId=${dataChannel.id}, readyState=${dataChannel.readyState}`); + let isOpen = { value: false }; dataChannel.onopen = () => { - console.assert(!isOpen.value, "substream opened twice") + if (isOpen.value) return; // Already handled below isOpen.value = true; + log(4, `channel-open; streamId=${streamId}, direction=${direction}, channelId=${dataChannel.id}`); config.onStreamOpened(streamId, direction); config.onWritableBytes(65536, streamId); }; + // The data channel may already be in the "open" state by the time we + // register the onopen handler (e.g., for inbound channels negotiated + // while we were processing other events). In that case the onopen event + // will never fire, so we must notify smoldot immediately. + if (dataChannel.readyState === "open") { + isOpen.value = true; + log(4, `channel-already-open; streamId=${streamId}, direction=${direction}, channelId=${dataChannel.id}`); + config.onStreamOpened(streamId, direction); + config.onWritableBytes(65536, streamId); + } + dataChannel.onerror = dataChannel.onclose = (event) => { // Note that Firefox doesn't support . const message = (event instanceof RTCErrorEvent) ? event.error.toString() : "RTCDataChannel closed"; + log(isOpen.value ? 4 : 2, `channel-${isOpen.value ? 'closed' : 'failed'}; streamId=${streamId}, direction=${direction}, channelId=${dataChannel.id}, message=${message}`); + if (!isOpen.value) { // Substream wasn't opened yet and thus has failed to open. The API has no // mechanism to report substream openings failures. We could try opening it @@ -369,13 +404,28 @@ function connect(config: ConnectionConfig): Connection { // Therefore we don't care about events concerning the fact that the connection is now fully // open. state.pc.onconnectionstatechange = (_event) => { + log(4, `connection-state-change; state=${state.pc!.connectionState}`); if (state.pc!.connectionState == "closed" || state.pc!.connectionState == "disconnected" || state.pc!.connectionState == "failed") { killAllJs(); config.onConnectionReset("WebRTC state transitioned to " + state.pc!.connectionState); } }; + state.pc.oniceconnectionstatechange = (_event) => { + log(4, `ice-connection-state-change; state=${state.pc!.iceConnectionState}`); + }; + state.pc.onnegotiationneeded = async (_event) => { + // In webrtc-direct, the SDP answer is static (derived from the remote + // peer's multiaddress and certificate hash). Re-negotiation after the + // initial exchange is never needed and can disrupt the SCTP association, + // causing pending inbound DCEP OPEN messages to be silently lost. + if (state.hasNegotiated) { + log(5, 'negotiation-needed; already negotiated, ignoring'); + return; + } + log(4, 'negotiation-needed; starting initial SDP exchange'); + // Create a new offer and set it as local description. let sdpOffer = (await state.pc!.createOffer()).sdp!; // We check that the locally-generated SDP offer has a data channel with the UDP @@ -461,10 +511,13 @@ function connect(config: ConnectionConfig): Connection { "a=candidate:1 1 UDP 1 " + targetIp + " " + String(targetPort) + " typ host" + "\n"; await state.pc!.setRemoteDescription({ type: "answer", sdp: remoteSdp }); + state.hasNegotiated = true; + log(4, 'negotiation-needed; SDP exchange complete'); }; state.pc.ondatachannel = ({ channel }) => { - // TODO: is the substream maybe already open? according to the Internet it seems that no but it's unclear + state.inboundChannelCount += 1; + log(4, `ondatachannel; channelId=${channel.id}, readyState=${channel.readyState}, inboundCount=${state.inboundChannelCount}`); addChannel(channel, 'inbound') }; @@ -509,8 +562,10 @@ function connect(config: ConnectionConfig): Connection { // per the libp2p WebRTC specification. // TODO: adjusting the options based on the first substream is a bit hacky const opts = state.isFirstOutSubstream ? { negotiated: true, id: 0 } : {}; + log(4, `openOutSubstream; first=${state.isFirstOutSubstream}, opts=${JSON.stringify(opts)}`); state.isFirstOutSubstream = false; - addChannel(state.pc!.createDataChannel("", opts), 'outbound') + + addChannel(state.pc!.createDataChannel("", opts), 'outbound'); } };