Request justifications of blocks on epoch change (#441)

* Track blocks finalized numbers in sync state machine

* Store the finalized height during the warp sync

* Finish the change

* PR link

full-node/src/run/network_service.rs

@@ -962,6 +962,7 @@ async fn update_round(inner: &Arc<Inner>, event_senders: &mut [mpsc::Sender<Even
                         state.set_id,
                         state.commit_finalized_height,
                     );
+                    // TODO: report to the sync state machine
                 }
                 service::Event::GrandpaCommitMessage {
                     chain_index,

lib/src/sync/all.rs

@@ -433,6 +433,7 @@ impl<TRq, TSrc, TBl> AllSync<TRq, TSrc, TBl> {
                     user_data,
                     best_block_number,
                     best_block_hash,
+                    finalized_block_height: None,
                 };
 
                 let inner_source_id = sync.add_source(source_extra);
@@ -1340,6 +1341,42 @@ impl<TRq, TSrc, TBl> AllSync<TRq, TSrc, TBl> {
         }
     }
 
+    /// Update the finalized block height of the given source.
+    ///
+    /// # Panic
+    ///
+    /// Panics if `source_id` is invalid.
+    ///
+    pub fn update_source_finality_state(
+        &mut self,
+        source_id: SourceId,
+        finalized_block_height: u64,
+    ) {
+        let source_id = self.shared.sources.get(source_id.0).unwrap();
+
+        match (&mut self.inner, source_id) {
+            (AllSyncInner::AllForks(sync), SourceMapping::AllForks(source_id)) => {
+                sync.update_source_finality_state(*source_id, finalized_block_height)
+            }
+            (AllSyncInner::Optimistic { .. }, _) => {} // TODO: the optimistic sync could get some help from the finalized block
+            (
+                AllSyncInner::GrandpaWarpSync { inner },
+                SourceMapping::GrandpaWarpSync(source_id),
+            ) => {
+                // TODO: the warp syncing algorithm could maybe be interested in the finalized block height
+                let n = &mut inner[*source_id].finalized_block_height;
+                *n = Some(n.map_or(finalized_block_height, |b| {
+                    cmp::max(b, finalized_block_height)
+                }));
+            }
+
+            // Invalid internal states.
+            (AllSyncInner::AllForks(_), _) => unreachable!(),
+            (AllSyncInner::GrandpaWarpSync { .. }, _) => unreachable!(),
+            (AllSyncInner::Poisoned, _) => unreachable!(),
+        }
+    }
+
     /// Update the state machine with a Grandpa commit message received from the network.
     ///
     /// This function only inserts the commit message into the state machine, and does not
@@ -2734,6 +2771,7 @@ struct OptimisticRequestExtra<TRq> {
 struct GrandpaWarpSyncSourceExtra<TSrc> {
     outer_source_id: SourceId,
     user_data: TSrc,
+    finalized_block_height: Option<u64>,
     best_block_number: u64,
     best_block_hash: [u8; 32],
 }
@@ -2860,6 +2898,10 @@ impl<TRq> Shared<TRq> {
                 }
             };
 
+            if let Some(finalized_block_height) = source.finalized_block_height {
+                all_forks.update_source_finality_state(updated_source_id, finalized_block_height);
+            }
+
             self.sources[source.outer_source_id.0] = SourceMapping::AllForks(updated_source_id);
         }
 

lib/src/sync/all_forks.rs

@@ -89,7 +89,12 @@ use crate::{
 };
 
 use alloc::{borrow::ToOwned as _, vec::Vec};
-use core::{mem, num::NonZeroU32, ops, time::Duration};
+use core::{
+    cmp, mem,
+    num::{NonZeroU32, NonZeroU64},
+    ops,
+    time::Duration,
+};
 
 mod disjoint;
 mod pending_blocks;
@@ -197,6 +202,9 @@ struct Source<TSrc> {
     /// source isn't malicious, we will able to make *some* progress in the finality.
     unverified_finality_proofs: SourcePendingJustificationProofs,
 
+    /// Height of the highest finalized block according to that source. `None` if unknown.
+    finalized_block_number: Option<u64>,
+
     /// Similar to [`Source::unverified_finality_proofs`]. Contains proofs that have been checked
     /// and have been determined to not be verifiable right now.
     pending_finality_proofs: SourcePendingJustificationProofs,
@@ -702,9 +710,40 @@ impl<TBl, TRq, TSrc> AllForksSync<TBl, TRq, TSrc> {
     pub fn desired_requests(
         &'_ self,
     ) -> impl Iterator<Item = (SourceId, &'_ TSrc, RequestParams)> + '_ {
-        // TODO: need to periodically query for justifications of non-finalized blocks that change GrandPa authorities
-
-        self.inner
+        // Query justifications of blocks that are necessary in order for finality to progress
+        // against sources that have reported these blocks as finalized.
+        // TODO: make it clear in the API docs that justifications should be requested as part of a request
+        // TODO: this is O(n)
+        let justification_requests =
+            self.chain
+                .finality_checkpoints()
+                .flat_map(move |(block_height, block_hash)| {
+                    self.inner
+                        .blocks
+                        .sources()
+                        .filter(move |s| {
+                            // We assume that all sources have the same finalized blocks and thus
+                            // don't check hashes.
+                            self.inner.blocks[*s].unverified_finality_proofs.is_none()
+                                && self.inner.blocks[*s]
+                                    .finalized_block_number
+                                    .map_or(false, |n| n >= block_height)
+                        })
+                        .map(move |source_id| {
+                            (
+                                source_id,
+                                &self.inner.blocks[source_id].user_data,
+                                RequestParams {
+                                    first_block_hash: *block_hash,
+                                    first_block_height: block_height,
+                                    num_blocks: NonZeroU64::new(1).unwrap(),
+                                },
+                            )
+                        })
+                });
+
+        let block_requests = self
+            .inner
             .blocks
             .desired_requests()
             .filter(move |rq| {
@@ -718,7 +757,9 @@ impl<TBl, TRq, TSrc> AllForksSync<TBl, TRq, TSrc> {
                     &self.inner.blocks[rq.source_id].user_data,
                     rq.request_params,
                 )
-            })
+            });
+
+        justification_requests.chain(block_requests)
     }
 
     /// Inserts a new request in the data structure.
@@ -748,6 +789,7 @@ impl<TBl, TRq, TSrc> AllForksSync<TBl, TRq, TSrc> {
     /// > **Note**: It is in no way mandatory to actually call this function and cancel the
     /// >           requests that are returned.
     pub fn obsolete_requests(&'_ self) -> impl Iterator<Item = (RequestId, &'_ TRq)> + '_ {
+        // TODO: requests meant to query justifications only are considered obsolete by the underlying state machine, which right now is okay because the underlying state machine is pretty loose in its definition of obsolete
         self.inner.blocks.obsolete_requests()
     }
 
@@ -908,6 +950,27 @@ impl<TBl, TRq, TSrc> AllForksSync<TBl, TRq, TSrc> {
         }
     }
 
+    /// Update the finalized block height of the given source.
+    ///
+    /// # Panic
+    ///
+    /// Panics if `source_id` is invalid.
+    ///
+    pub fn update_source_finality_state(
+        &mut self,
+        source_id: SourceId,
+        finalized_block_height: u64,
+    ) {
+        let source = &mut self.inner.blocks[source_id];
+        source.finalized_block_number = Some(
+            source
+                .finalized_block_number
+                .map_or(finalized_block_height, |b| {
+                    cmp::max(b, finalized_block_height)
+                }),
+        );
+    }
+
     /// Update the state machine with a Grandpa commit message received from the network.
     ///
     /// This function only inserts the commit message into the state machine, and does not
@@ -932,6 +995,14 @@ impl<TBl, TRq, TSrc> AllForksSync<TBl, TRq, TSrc> {
             Err(_) => return GrandpaCommitMessageOutcome::ParseError,
         };
 
+        // The finalized block number of the source is increased even if the commit message
+        // isn't known to be valid yet.
+        source.finalized_block_number = Some(
+            source
+                .finalized_block_number
+                .map_or(block_number, |b| cmp::max(b, block_number)),
+        );
+
         source.unverified_finality_proofs.insert(
             block_number,
             FinalityProofs::GrandpaCommit(scale_encoded_commit),
@@ -1732,6 +1803,7 @@ impl<'a, TBl, TRq, TSrc> AddSourceOldBlock<'a, TBl, TRq, TSrc> {
             Source {
                 user_data: source_user_data,
                 unverified_finality_proofs: SourcePendingJustificationProofs::None,
+                finalized_block_number: None,
                 pending_finality_proofs: SourcePendingJustificationProofs::None,
             },
             self.best_block_number,
@@ -1778,6 +1850,7 @@ impl<'a, TBl, TRq, TSrc> AddSourceKnown<'a, TBl, TRq, TSrc> {
             Source {
                 user_data: source_user_data,
                 unverified_finality_proofs: SourcePendingJustificationProofs::None,
+                finalized_block_number: None,
                 pending_finality_proofs: SourcePendingJustificationProofs::None,
             },
             self.best_block_number,
@@ -1813,6 +1886,7 @@ impl<'a, TBl, TRq, TSrc> AddSourceUnknown<'a, TBl, TRq, TSrc> {
             Source {
                 user_data: source_user_data,
                 unverified_finality_proofs: SourcePendingJustificationProofs::None,
+                finalized_block_number: None,
                 pending_finality_proofs: SourcePendingJustificationProofs::None,
             },
             self.best_block_number,

light-base/src/network_service.rs

@@ -871,6 +871,11 @@ pub enum Event {
         chain_index: usize,
         announce: service::EncodedBlockAnnounce,
     },
+    GrandpaNeighborPacket {
+        peer_id: PeerId,
+        chain_index: usize,
+        finalized_block_height: u64,
+    },
     /// Received a GrandPa commit message from the network.
     GrandpaCommitMessage {
         peer_id: PeerId,
@@ -1252,6 +1257,11 @@ async fn update_round<TPlat: PlatformRef>(
                         state.set_id,
                         state.commit_finalized_height,
                     );
+                    break Event::GrandpaNeighborPacket {
+                        chain_index,
+                        peer_id,
+                        finalized_block_height: state.commit_finalized_height,
+                    };
                 }
                 service::Event::GrandpaCommitMessage {
                     chain_index,

light-base/src/sync_service/standalone.rs

@@ -1173,6 +1173,16 @@ impl<TPlat: PlatformRef> Task<TPlat> {
                 }
             }
 
+            network_service::Event::GrandpaNeighborPacket {
+                peer_id,
+                chain_index,
+                finalized_block_height,
+            } if chain_index == self.network_chain_index => {
+                let sync_source_id = *self.peers_source_id_map.get(&peer_id).unwrap();
+                self.sync
+                    .update_source_finality_state(sync_source_id, finalized_block_height);
+            }
+
             network_service::Event::GrandpaCommitMessage {
                 chain_index,
                 peer_id,

wasm-node/CHANGELOG.md

@@ -9,6 +9,7 @@
 
 ### Fixed
 
+- Fix finality stalling on epoch change by explicitly requesting justifications of blocks that a peer has reported as finalized but that isn't finalized locally. ([#441](https://github.com/smol-dot/smoldot/pull/441))
 - Fix `AlreadyDestroyedError` not being properly thrown if a function is called after `terminate()`. ([#438](https://github.com/smol-dot/smoldot/pull/438))
 
 ## 1.0.2 - 2023-04-12