feat: improve legacy top down, have a simpler storage node runner

Author: karlem

fendermint/vm/interpreter/src/fvm/legacy_topdown.rs

@@ -47,6 +47,21 @@ impl LegacyTopDownHandler {
             height: finality.height as u64,
             block_hash: finality.block_hash,
         };
+        let quorum_threshold = atomically(|| self.votes.quorum_threshold()).await;
+
+        // In a single-validator subnet, self-attestation should not depend on local
+        // cache catch-up. Otherwise aggressive proposals can be rejected by the same
+        // node that created them, stalling consensus at one height.
+        if quorum_threshold == 1 {
+            let committed_height = atomically(|| {
+                self.provider
+                    .last_committed_finality()
+                    .map(|f| f.map(|f| f.height).unwrap_or_default())
+            })
+            .await;
+            return prop.height > committed_height;
+        }
+
         atomically(|| self.provider.check_proposal(&prop)).await
     }
 
@@ -103,37 +118,101 @@ impl LegacyTopDownHandler {
     async fn chain_message_from_finality_or_quorum(&self) -> Option<ChainMessage> {
         atomically(|| self.votes.pause_votes_until_find_quorum()).await;
 
-        let (parent, quorum) = atomically(|| {
+        let (parent, quorum, quorum_threshold) = atomically(|| {
             let parent = self.provider.next_proposal()?;
 
             let quorum = self
                 .votes
                 .find_quorum()?
                 .map(|(height, block_hash)| IPCParentFinality { height, block_hash });
 
-            Ok((parent, quorum))
+            let quorum_threshold = self.votes.quorum_threshold()?;
+
+            Ok((parent, quorum, quorum_threshold))
         })
         .await;
 
         let parent = parent?;
 
-        let quorum = if let Some(quorum) = quorum {
-            quorum
-        } else {
-            emit!(
-                DEBUG,
-                ParentFinalityMissingQuorum {
-                    block_height: parent.height,
-                    block_hash: &hex::encode(&parent.block_hash),
-                }
-            );
-            return None;
-        };
-
-        let finality = if parent.height <= quorum.height {
-            parent
+        // Fast-path for single-validator subnets: quorum threshold is 1, so requiring
+        // a separate vote-derived quorum can unnecessarily throttle catch-up.
+        //
+        // In addition, bypass `next_proposal()` bounds (`max_proposal_range`, `proposal_delay`)
+        // and favor the freshest finalized parent view queried directly from parent RPC
+        // (`chain_head - chain_head_delay`). This avoids being throttled by local cache
+        // catch-up speed when the node is far behind.
+        let finality = if quorum_threshold == 1 {
+            let committed_height = atomically(|| {
+                self.provider
+                    .last_committed_finality()
+                    .map(|f| f.map(|f| f.height).unwrap_or_default())
+            })
+            .await;
+
+            let remote_finalized = self
+                .provider
+                .latest_finalized_parent_view()
+                .await
+                .ok()
+                .flatten();
+            let candidate = if let Some(remote_finalized) =
+                remote_finalized.filter(|f| f.height > committed_height)
+            {
+                remote_finalized
+            } else {
+                let latest_non_null = atomically(|| {
+                    let latest = self.provider.latest_height()?;
+                    let committed = self.provider.last_committed_finality()?;
+
+                    let (latest, committed_height) = match (latest, committed) {
+                        (Some(latest), Some(committed)) => (latest, committed.height),
+                        _ => return Ok(None),
+                    };
+
+                    if latest <= committed_height {
+                        return Ok(None);
+                    }
+
+                    let latest_non_null = self
+                        .provider
+                        .first_non_null_block(latest)?
+                        .filter(|h| *h > committed_height);
+
+                    let Some(height) = latest_non_null else {
+                        return Ok(None);
+                    };
+
+                    let Some(block_hash) = self.provider.block_hash(height)? else {
+                        return Ok(None);
+                    };
+
+                    Ok(Some(IPCParentFinality { height, block_hash }))
+                })
+                .await;
+
+                latest_non_null.unwrap_or(parent)
+            };
+
+            candidate
         } else {
-            quorum
+            let quorum = if let Some(quorum) = quorum {
+                quorum
+            } else {
+                emit!(
+                    DEBUG,
+                    ParentFinalityMissingQuorum {
+                        block_height: parent.height,
+                        block_hash: &hex::encode(&parent.block_hash),
+                    }
+                );
+                return None;
+            };
+
+            if parent.height <= quorum.height {
+                parent
+            } else {
+                quorum
+            }
         };
 
         Some(ChainMessage::Ipc(IpcMessage::TopDownExec(ParentFinality {

fendermint/vm/interpreter/src/fvm/topdown.rs

@@ -20,6 +20,9 @@ use crate::types::AppliedMessage;
 use ipc_api::cross::IpcEnvelope;
 use ipc_observability::emit;
 
+/// Maximum inclusive parent-height span processed per legacy topdown chunk.
+const LEGACY_TOPDOWN_CHUNK_SIZE: BlockHeight = 200;
+
 #[derive(Clone, Debug)]
 pub struct F3ExecutionCacheRetryConfig {
     pub backoff_initial: std::time::Duration,
@@ -399,41 +402,75 @@ where
         // be _at least_ 1 height behind.
         let (execution_fr, execution_to) = (prev_height + 1, finality.height);
 
-        // error happens if we cannot get the validator set from ipc agent after retries
-        let validator_changes = legacy
-            .validator_changes_from(execution_fr, execution_to)
-            .await
-            .context("failed to fetch validator changes")?;
-
-        tracing::debug!(
-            from = execution_fr,
-            to = execution_to,
-            msgs = validator_changes.len(),
-            "chain interpreter received total validator changes"
-        );
+        let mut total_validator_changes = 0usize;
+        let mut total_topdown_msgs = 0usize;
+        let mut ret: Option<AppliedMessage> = None;
+        let mut chunk_start = execution_fr;
+        while chunk_start <= execution_to {
+            let chunk_end = chunk_start
+                .saturating_add(LEGACY_TOPDOWN_CHUNK_SIZE.saturating_sub(1))
+                .min(execution_to);
+
+            // error happens if we cannot get the validator set from ipc agent after retries
+            let validator_changes = legacy
+                .validator_changes_from(chunk_start, chunk_end)
+                .await
+                .context("failed to fetch validator changes")?;
+            total_validator_changes += validator_changes.len();
+
+            tracing::debug!(
+                chunk_start,
+                chunk_end,
+                changes = validator_changes.len(),
+                "chain interpreter received validator changes chunk"
+            );
 
-        self.inner
-            .gateway_caller
-            .store_validator_changes(state, validator_changes)
-            .context("failed to store validator changes")?;
+            self.inner
+                .gateway_caller
+                .store_validator_changes(state, validator_changes)
+                .context("failed to store validator changes")?;
+
+            // error happens if we cannot get the cross messages from ipc agent after retries
+            let msgs = legacy
+                .top_down_msgs_from(chunk_start, chunk_end)
+                .await
+                .context("failed to fetch top down messages")?;
+            total_topdown_msgs += msgs.len();
+
+            tracing::debug!(
+                chunk_start,
+                chunk_end,
+                number_of_messages = msgs.len(),
+                "chain interpreter received topdown messages chunk",
+            );
+
+            if !msgs.is_empty() {
+                ret = Some(
+                    self.execute_topdown_msgs(state, msgs)
+                        .await
+                        .context("failed to execute top down messages")?,
+                );
+            }
 
-        // error happens if we cannot get the cross messages from ipc agent after retries
-        let msgs = legacy
-            .top_down_msgs_from(execution_fr, execution_to)
-            .await
-            .context("failed to fetch top down messages")?;
+            chunk_start = chunk_end.saturating_add(1);
+        }
 
         tracing::debug!(
-            number_of_messages = msgs.len(),
             start = execution_fr,
             end = execution_to,
-            "chain interpreter received topdown msgs",
+            total_validator_changes,
+            total_topdown_msgs,
+            "chain interpreter processed topdown effects in chunks",
         );
 
-        let ret = self
-            .execute_topdown_msgs(state, msgs)
-            .await
-            .context("failed to execute top down messages")?;
+        // Preserve previous behavior: return an AppliedMessage even when there are no topdown msgs.
+        let ret = if let Some(ret) = ret {
+            ret
+        } else {
+            self.execute_topdown_msgs(state, Vec::<IpcEnvelope>::new())
+                .await
+                .context("failed to execute empty top down messages batch")?
+        };
 
         tracing::debug!("chain interpreter applied topdown msgs");