Fix WebRTC light client sync stalls

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::<Vec<_>>() {
                     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,

lib/src/sync/all.rs

@@ -413,6 +413,7 @@ impl<TRq, TSrc, TBl> AllSync<TRq, TSrc, TBl> {
                     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<TRq, TSrc, TBl> AllSync<TRq, TSrc, TBl> {
                     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<TRq, TSrc, TBl> AllSync<TRq, TSrc, TBl> {
                     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<TRq, TSrc, TBl> AllSync<TRq, TSrc, TBl> {
                     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<TBl>, AllForksRequestExtra, AllForksSourceExtra>,
     warp_sync: &'a mut Option<warp_sync::WarpSync<WarpSyncSourceExtra, WarpSyncRequestExtra>>,
+    best_block_number: u64,
     marker: PhantomData<TRq>,
 }
 
@@ -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<TBl>, AllForksRequestExtra, AllForksSourceExtra>,
     warp_sync: &'a mut Option<warp_sync::WarpSync<WarpSyncSourceExtra, WarpSyncRequestExtra>>,
+    best_block_number: u64,
     marker: PhantomData<TRq>,
 }
 
@@ -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<TBl>, AllForksRequestExtra, AllForksSourceExtra>,
     warp_sync: &'a mut Option<warp_sync::WarpSync<WarpSyncSourceExtra, WarpSyncRequestExtra>>,
+    best_block_number: u64,
     marker: PhantomData<TRq>,
 }
 
@@ -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
         };

lib/src/sync/warp_sync.rs

@@ -640,15 +640,17 @@ impl<TSrc, TRq> WarpSync<TSrc, TRq> {
 
     /// 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());
 

light-base/src/network_service.rs

@@ -957,6 +957,7 @@ async fn background_task<TPlat: PlatformRef>(mut task: BackgroundTask<TPlat>) {
             NetworkEvent(service::Event<async_channel::Sender<service::CoordinatorToConnection>>),
             CanAssignSlot(PeerId, ChainId),
             NextRecentConnectionRestore,
+            NotificationRetryReady,
             CanStartConnect(PeerId),
             CanOpenGossip(PeerId, ChainId),
             MessageFromConnection {
@@ -971,6 +972,8 @@ async fn background_task<TPlat: PlatformRef>(mut task: BackgroundTask<TPlat>) {
             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<TPlat: PlatformRef>(mut task: BackgroundTask<TPlat>) {
             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<TPlat: PlatformRef>(mut task: BackgroundTask<TPlat>) {
                 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<TPlat: PlatformRef>(mut task: BackgroundTask<TPlat>) {
                 .or(next_recent_connection_restore)
                 .or(finished_sending_event)
                 .or(start_discovery)
+                .or(notification_retry)
                 .await
         };
 
@@ -2542,6 +2554,21 @@ async fn background_task<TPlat: PlatformRef>(mut task: BackgroundTask<TPlat>) {
                 // 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<TPlat: PlatformRef>(mut task: BackgroundTask<TPlat>) {
                 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

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<TPlat: PlatformRef>(
     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<TPlat: PlatformRef>(
         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!(
@@ -381,6 +392,7 @@ pub(super) async fn webrtc_multi_stream_connection_task<TPlat: PlatformRef>(
                                 "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| {
@@ -453,8 +465,16 @@ pub(super) async fn webrtc_multi_stream_connection_task<TPlat: PlatformRef>(
                 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)
                         })
                     });

light-base/src/sync_service/standalone.rs

@@ -577,9 +577,15 @@ pub(super) async fn start_standalone_chain<TPlat: PlatformRef>(
                         // 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,

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();