fix: make load_required_tipset_by_height non blocking

src/chain/store/chain_store.rs

@@ -319,16 +319,16 @@ impl ChainStore {
 
     /// Returns [`None`] when `ts` has no known child on the current heaviest chain
     /// (e.g. `ts` is the chain head). Blockstore errors are returned as [`Err`].
-    pub fn load_child_tipset(&self, ts: &Tipset) -> Result<Option<Tipset>, Error> {
+    pub async fn load_child_tipset(&self, ts: &Tipset) -> Result<Option<Tipset>, Error> {
         let head = self.heaviest_tipset();
         if head.parents() == ts.key() {
             Ok(Some(head))
         } else if head.epoch() > ts.epoch() {
-            match self.chain_index().tipset_by_height(
-                ts.epoch() + 1,
-                head,
-                ResolveNullTipset::TakeNewer,
-            )? {
+            match self
+                .chain_index()
+                .tipset_by_height_async(ts.epoch() + 1, head, ResolveNullTipset::TakeNewer)
+                .await?
+            {
                 Some(maybe_child) if maybe_child.parents() == ts.key() => Ok(Some(maybe_child)),
                 _ => Ok(None),
             }
@@ -431,7 +431,7 @@ impl ChainStore {
         }
 
         let next_ts = chain_index
-            .load_required_tipset_by_height(
+            .load_required_tipset_by_height_blocking(
                 lbr + 1,
                 heaviest_tipset.clone(),
                 ResolveNullTipset::TakeNewer,

src/chain/store/index.rs

@@ -240,20 +240,21 @@ impl ChainIndex {
         Ok(None)
     }
 
+    /// Non-blocking version of [`Self::tipset_by_height`]
     pub async fn tipset_by_height_async(
         &self,
         to: ChainEpoch,
         from: Tipset,
         resolve: ResolveNullTipset,
     ) -> Result<Option<Tipset>, Error> {
         let this = self.shallow_clone();
-        tokio::task::spawn_blocking(move || this.tipset_by_height(to, from, resolve))
-            .await
-            .map_err(|e| Error::Other(e.to_string()))?
+        tokio::task::spawn_blocking(move || this.tipset_by_height(to, from, resolve)).await?
     }
 
     /// Same as [`Self::tipset_by_height`], but errors if that would return `None`.
-    pub fn load_required_tipset_by_height(
+    /// This call can be expensive and blocking, use [`Self::load_required_tipset_by_height`]
+    /// in async contexts to avoid exhausting Tokio worker threads.
+    pub fn load_required_tipset_by_height_blocking(
         &self,
         to: ChainEpoch,
         from: Tipset,
@@ -263,6 +264,18 @@ impl ChainIndex {
             .ok_or_else(|| Error::NotFound(format!("tipset at epoch {to}").into()))
     }
 
+    /// Same as [`Self::tipset_by_height_async`], but errors if that would return `None`.
+    pub async fn load_required_tipset_by_height(
+        &self,
+        to: ChainEpoch,
+        from: Tipset,
+        resolve: ResolveNullTipset,
+    ) -> Result<Tipset, Error> {
+        self.tipset_by_height_async(to, from, resolve)
+            .await?
+            .ok_or_else(|| Error::NotFound(format!("tipset at epoch {to}").into()))
+    }
+
     /// Finds the latest beacon entry given a tipset up to 20 tipsets behind
     pub fn latest_beacon_entry(&self, tipset: Tipset) -> Result<BeaconEntry, Error> {
         for ts in tipset.chain(&self.db).take(20) {

src/chain_sync/sync_status.rs

@@ -13,7 +13,7 @@ use std::sync::Arc;
 use tracing::log;
 
 // Node considered synced if the head is within this threshold.
-const SYNCED_EPOCH_THRESHOLD: u64 = 10;
+const SYNCED_EPOCH_THRESHOLD: u64 = 2;
 
 /// Represents the overall synchronization status of the Forest node.
 #[derive(

src/daemon/mod.rs

@@ -825,7 +825,7 @@ pub(super) async fn start_services(
     }
     if !opts.stateless
         && !opts.skip_load_actors
-        && let Err(e) = ctx.state_manager.maybe_rewind_heaviest_tipset()
+        && let Err(e) = ctx.state_manager.maybe_rewind_heaviest_tipset().await
     {
         tracing::warn!("error in maybe_rewind_heaviest_tipset: {e:#}");
     }

src/dev/subcommands/export_state_tree_cmd.rs

@@ -77,11 +77,14 @@ impl ExportStateTreeCommand {
                 .await?;
         let chain_store = ChainStore::new(db.clone(), chain_config, genesis_header)?;
 
-        let start_ts = chain_store.chain_index().load_required_tipset_by_height(
-            from,
-            chain_store.heaviest_tipset(),
-            ResolveNullTipset::TakeNewer,
-        )?;
+        let start_ts = chain_store
+            .chain_index()
+            .load_required_tipset_by_height(
+                from,
+                chain_store.heaviest_tipset(),
+                ResolveNullTipset::TakeNewer,
+            )
+            .await?;
 
         let mut ipld_roots = vec![];
         for (child, ts) in start_ts

src/dev/subcommands/state_cmd.rs

@@ -80,12 +80,14 @@ impl ComputeCommand {
         let (ts, ts_next) = {
             // We don't want to track all entries that are visited by `tipset_by_height`
             db.pause_tracking();
-            let ts = chain_index.load_required_tipset_by_height(
-                epoch,
-                chain_store.heaviest_tipset(),
-                ResolveNullTipset::TakeOlder,
-            )?;
-            let ts_next = chain_store.load_child_tipset(&ts)?.with_context(|| {
+            let ts = chain_index
+                .load_required_tipset_by_height(
+                    epoch,
+                    chain_store.heaviest_tipset(),
+                    ResolveNullTipset::TakeOlder,
+                )
+                .await?;
+            let ts_next = chain_store.load_child_tipset(&ts).await?.with_context(|| {
                 format!(
                     "no child tipset for epoch {} (may be chain head)",
                     ts.epoch()
@@ -210,11 +212,13 @@ impl ValidateCommand {
         let ts = {
             // We don't want to track all entries that are visited by `tipset_by_height`
             db.pause_tracking();
-            let ts = chain_index.load_required_tipset_by_height(
-                epoch,
-                chain_store.heaviest_tipset(),
-                ResolveNullTipset::TakeOlder,
-            )?;
+            let ts = chain_index
+                .load_required_tipset_by_height(
+                    epoch,
+                    chain_store.heaviest_tipset(),
+                    ResolveNullTipset::TakeOlder,
+                )
+                .await?;
             db.resume_tracking();
             SettingsStoreExt::write_obj(&db.tracker, crate::db::setting_keys::HEAD_KEY, ts.key())?;
             // Only track the desired tipset

src/interpreter/externs.rs

@@ -261,7 +261,7 @@ impl ForestExterns {
     fn get_tipset_cid_impl(&self, epoch: ChainEpoch) -> anyhow::Result<Cid> {
         let ts = self
             .chain_index
-            .load_required_tipset_by_height(
+            .load_required_tipset_by_height_blocking(
                 epoch,
                 self.heaviest_tipset.clone(),
                 ResolveNullTipset::TakeOlder,

src/message_pool/msgpool/provider.rs

@@ -10,6 +10,7 @@ use crate::message_pool::msg_pool::{
     MAX_ACTOR_PENDING_MESSAGES, MAX_UNTRUSTED_ACTOR_PENDING_MESSAGES,
 };
 use crate::networks::Height;
+use crate::prelude::*;
 use crate::shim::{
     address::{Address, Protocol::*},
     econ::TokenAmount,
@@ -18,8 +19,6 @@ use crate::shim::{
 };
 use crate::utils::db::CborStoreExt;
 use auto_impl::auto_impl;
-use cid::Cid;
-use std::sync::Arc;
 use tokio::sync::broadcast;
 
 /// Provider Trait. This trait will be used by the message pool to interact with
@@ -120,7 +119,7 @@ impl Provider for ChainStore {
             _ => {
                 let lookback_ts = if ts.epoch() > self.chain_config().policy.chain_finality {
                     self.chain_index()
-                        .load_required_tipset_by_height(
+                        .load_required_tipset_by_height_blocking(
                             ts.epoch() - self.chain_config().policy.chain_finality,
                             ts.clone(),
                             ResolveNullTipset::TakeOlder,
@@ -144,3 +143,31 @@ impl Provider for ChainStore {
         ChainStore::messages_for_tipset(self, ts).map_err(Into::into)
     }
 }
+
+#[allow(dead_code)]
+pub trait ProviderExt {
+    /// Non-blocking version of [`Provider::resolve_to_deterministic_address_at_finality`]
+    async fn resolve_to_deterministic_address_at_finality_async(
+        &self,
+        addr: Address,
+        ts: Tipset,
+    ) -> Result<Address, Error>;
+}
+
+impl<T> ProviderExt for T
+where
+    T: Provider + ShallowClone + Send + Sync + 'static,
+{
+    async fn resolve_to_deterministic_address_at_finality_async(
+        &self,
+        addr: Address,
+        ts: Tipset,
+    ) -> Result<Address, Error> {
+        let this = self.shallow_clone();
+        tokio::task::spawn_blocking(move || {
+            this.resolve_to_deterministic_address_at_finality(&addr, &ts)
+        })
+        .await
+        .context("tokio join error")?
+    }
+}

src/rpc/methods/chain.rs

@@ -296,11 +296,10 @@ impl RpcMethod<1> for ForestChainExport {
         let chain_export_guard = ChainExportGuard::try_start_export()?;
 
         let head = ctx.chain_store().load_required_tipset_or_heaviest(&tsk)?;
-        let start_ts = ctx.chain_index().load_required_tipset_by_height(
-            epoch,
-            head,
-            ResolveNullTipset::TakeOlder,
-        )?;
+        let start_ts = ctx
+            .chain_index()
+            .load_required_tipset_by_height(epoch, head, ResolveNullTipset::TakeOlder)
+            .await?;
 
         let options = ExportOptions {
             skip_checksum,
@@ -498,11 +497,10 @@ impl RpcMethod<1> for ForestChainExportDiff {
         }
 
         let head = ctx.chain_store().heaviest_tipset();
-        let start_ts = ctx.chain_index().load_required_tipset_by_height(
-            from,
-            head,
-            ResolveNullTipset::TakeOlder,
-        )?;
+        let start_ts = ctx
+            .chain_index()
+            .load_required_tipset_by_height(from, head, ResolveNullTipset::TakeOlder)
+            .await?;
 
         crate::tool::subcommands::archive_cmd::do_export(
             ctx.chain_index().db(),
@@ -798,11 +796,10 @@ impl RpcMethod<2> for ChainGetTipSetByHeight {
         let ts = ctx
             .chain_store()
             .load_required_tipset_or_heaviest(&tipset_key)?;
-        let tss = ctx.chain_index().load_required_tipset_by_height(
-            height,
-            ts,
-            ResolveNullTipset::TakeOlder,
-        )?;
+        let tss = ctx
+            .chain_index()
+            .load_required_tipset_by_height(height, ts, ResolveNullTipset::TakeOlder)
+            .await?;
         Ok(tss)
     }
 }
@@ -828,11 +825,10 @@ impl RpcMethod<2> for ChainGetTipSetAfterHeight {
         let ts = ctx
             .chain_store()
             .load_required_tipset_or_heaviest(&tipset_key)?;
-        let tss = ctx.chain_index().load_required_tipset_by_height(
-            height,
-            ts,
-            ResolveNullTipset::TakeNewer,
-        )?;
+        let tss = ctx
+            .chain_index()
+            .load_required_tipset_by_height(height, ts, ResolveNullTipset::TakeNewer)
+            .await?;
         Ok(tss)
     }
 }
@@ -969,16 +965,16 @@ impl ChainGetTipSetV2 {
         if finalized.epoch() >= safe_height {
             Ok(finalized)
         } else {
-            Ok(ctx.chain_index().load_required_tipset_by_height(
-                safe_height,
-                head,
-                ResolveNullTipset::TakeOlder,
-            )?)
+            Ok(ctx
+                .chain_index()
+                .load_required_tipset_by_height(safe_height, head, ResolveNullTipset::TakeOlder)
+                .await?)
         }
     }
 
     pub async fn get_latest_finalized_tipset(ctx: &Ctx) -> anyhow::Result<Tipset> {
-        ChainGetTipSetFinalityStatus::get_finality_status(ctx)?
+        ChainGetTipSetFinalityStatus::get_finality_status(ctx)
+            .await?
             .finalized_tip_set
             .context("failed to resolve finalized tipset")
     }
@@ -993,11 +989,14 @@ impl ChainGetTipSetV2 {
         // Get tipset by height.
         if let Some(height) = &selector.height {
             let anchor = Self::get_tipset_by_anchor(ctx, height.anchor.as_ref()).await?;
-            let ts = ctx.chain_index().load_required_tipset_by_height(
-                height.at,
-                anchor,
-                height.resolve_null_tipset_policy(),
-            )?;
+            let ts = ctx
+                .chain_index()
+                .load_required_tipset_by_height(
+                    height.at,
+                    anchor,
+                    height.resolve_null_tipset_policy(),
+                )
+                .await?;
             return Ok(ts);
         }
         // Get tipset by tag, either latest or finalized.
@@ -1032,10 +1031,11 @@ pub enum ChainGetTipSetFinalityStatus {}
 
 const EC_CALCULATOR_FINALITY_CACHE_SIZE: usize = 4;
 impl ChainGetTipSetFinalityStatus {
-    pub fn get_finality_status(ctx: &Ctx) -> anyhow::Result<ChainFinalityStatus> {
+    pub async fn get_finality_status(ctx: &Ctx) -> anyhow::Result<ChainFinalityStatus> {
         let head = ctx.chain_store().heaviest_tipset();
         let (ec_finality_threshold_depth, ec_finalized_tip_set) =
-            Self::get_ec_finality_threshold_depth_and_tipset_with_cache(ctx, head.shallow_clone())?;
+            Self::get_ec_finality_threshold_depth_and_tipset_with_cache(ctx, head.shallow_clone())
+                .await?;
         let f3_finalized_tip_set = ctx.chain_store().f3_finalized_tipset();
         let finalized_tip_set = match (&ec_finalized_tip_set, &f3_finalized_tip_set) {
             (Some(ec), Some(f3)) => {
@@ -1058,15 +1058,18 @@ impl ChainGetTipSetFinalityStatus {
         })
     }
 
-    pub fn get_ec_finality_threshold_depth_and_tipset_with_cache(
+    pub async fn get_ec_finality_threshold_depth_and_tipset_with_cache(
         ctx: &Ctx,
         head: Tipset,
     ) -> anyhow::Result<(i64, Option<Tipset>)> {
         static CACHE: LazyLock<quick_cache::sync::Cache<TipsetKey, (i64, Option<Tipset>)>> =
             LazyLock::new(|| quick_cache::sync::Cache::new(EC_CALCULATOR_FINALITY_CACHE_SIZE));
-        CACHE.get_or_insert_with(head.shallow_clone().key(), move || {
-            Self::get_ec_finality_threshold_depth_and_tipset(ctx, head)
-        })
+        CACHE
+            .get_or_insert_async(
+                head.shallow_clone().key(),
+                Self::get_ec_finality_threshold_depth_and_tipset(ctx, head),
+            )
+            .await
     }
 
     pub fn get_ec_finality_epoch(
@@ -1166,7 +1169,7 @@ impl ChainGetTipSetFinalityStatus {
         }
     }
 
-    fn get_ec_finality_threshold_depth_and_tipset(
+    async fn get_ec_finality_threshold_depth_and_tipset(
         ctx: &Ctx,
         head: Tipset,
     ) -> anyhow::Result<(i64, Option<Tipset>)> {
@@ -1177,11 +1180,10 @@ impl ChainGetTipSetFinalityStatus {
         );
         let ec_finality_epoch =
             Self::get_ec_finality_epoch_by_depth(ctx.chain_config(), &head, depth);
-        let finalized = ctx.chain_index().tipset_by_height(
-            ec_finality_epoch,
-            head,
-            ResolveNullTipset::TakeOlder,
-        )?;
+        let finalized = ctx
+            .chain_index()
+            .tipset_by_height_async(ec_finality_epoch, head, ResolveNullTipset::TakeOlder)
+            .await?;
         Ok((depth, finalized))
     }
 }
@@ -1202,7 +1204,7 @@ impl RpcMethod<0> for ChainGetTipSetFinalityStatus {
         (): Self::Params,
         _: &http::Extensions,
     ) -> Result<Self::Ok, ServerError> {
-        Ok(Self::get_finality_status(&ctx)?)
+        Ok(Self::get_finality_status(&ctx).await?)
     }
 }