feat: experimental apis for f05 state dump

Author: LesnyRumcajs

Cargo.lock

@@ -22,6 +22,19 @@ multihash-codetable = { version = "0.1", features = ["blake2b", "blake2s", "blak
 rust2go = "0.4"
 tokio = "1"
 
+# TODO: Remove this once the `fvm_ipld_amt` crate is published to crates.io.
+[patch.crates-io]
+# fvm_ipld_amt = { path = "../ref-fvm/ipld/amt" }
+# fvm_ipld_bitfield = { path = "../ref-fvm/ipld/bitfield" }
+# fvm_ipld_blockstore = { path = "../ref-fvm/ipld/blockstore" }
+# fvm_ipld_encoding = { path = "../ref-fvm/ipld/encoding" }
+# fvm_ipld_hamt = { path = "../ref-fvm/ipld/hamt" }
+fvm_ipld_amt = { git = "https://github.com/filecoin-project/ref-fvm", rev = "a633547ae414a333b2d076beef87d4d30cdb7fb4" }
+fvm_ipld_bitfield = { git = "https://github.com/filecoin-project/ref-fvm", rev = "a633547ae414a333b2d076beef87d4d30cdb7fb4" }
+fvm_ipld_blockstore = { git = "https://github.com/filecoin-project/ref-fvm", rev = "a633547ae414a333b2d076beef87d4d30cdb7fb4" }
+fvm_ipld_encoding = { git = "https://github.com/filecoin-project/ref-fvm", rev = "a633547ae414a333b2d076beef87d4d30cdb7fb4" }
+fvm_ipld_hamt = { git = "https://github.com/filecoin-project/ref-fvm", rev = "a633547ae414a333b2d076beef87d4d30cdb7fb4" }
+
 [dependencies]
 ahash = "0.8"
 anes = "0.2"

Cargo.toml

@@ -37,6 +37,7 @@ pub struct Client {
     v0: tokio::sync::OnceCell<UrlClient>,
     v1: tokio::sync::OnceCell<UrlClient>,
     v2: tokio::sync::OnceCell<UrlClient>,
+    experimental: tokio::sync::OnceCell<UrlClient>,
 }
 
 impl Client {
@@ -84,6 +85,7 @@ impl Client {
             v0: Default::default(),
             v1: Default::default(),
             v2: Default::default(),
+            experimental: Default::default(),
         }
     }
     pub fn base_url(&self) -> &Url {
@@ -162,6 +164,7 @@ impl Client {
             ApiPaths::V0 => &self.v0,
             ApiPaths::V1 => &self.v1,
             ApiPaths::V2 => &self.v2,
+            ApiPaths::Experimental => &self.experimental,
         }
         .get_or_try_init(|| async {
             let url = self
@@ -170,6 +173,7 @@ impl Client {
                     ApiPaths::V0 => "rpc/v0",
                     ApiPaths::V1 => "rpc/v1",
                     ApiPaths::V2 => "rpc/v2",
+                    ApiPaths::Experimental => "rpc/experimental",
                 })
                 .map_err(|it| {
                     ClientError::Custom(format!("creating url for endpoint failed: {it}"))

src/rpc/client.rs

@@ -10,7 +10,7 @@ use crate::cid_collections::CidHashSet;
 use crate::eth::EthChainId;
 use crate::interpreter::{MessageCallbackCtx, VMTrace};
 use crate::libp2p::NetworkMessage;
-use crate::lotus_json::lotus_json_with_self;
+use crate::lotus_json::{LotusJson, lotus_json_with_self};
 use crate::networks::ChainConfig;
 use crate::rpc::registry::actors_reg::load_and_serialize_actor_state;
 use crate::shim::actors::init;
@@ -68,6 +68,7 @@ use nunny::vec as nonempty;
 use parking_lot::Mutex;
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
+use std::io::Write;
 use std::ops::Mul;
 use std::path::PathBuf;
 use std::{sync::Arc, time::Duration};
@@ -497,7 +498,7 @@ impl RpcMethod<1> for StateMarketDeals {
         let sa = market_state.states(ctx.store())?;
 
         let mut out = HashMap::new();
-        da.for_each(|deal_id, d| {
+        da.for_each_cacheless(|deal_id, d| {
             let s = sa.get(deal_id)?.unwrap_or(market::DealState {
                 sector_start_epoch: -1,
                 last_updated_epoch: -1,
@@ -519,6 +520,125 @@ impl RpcMethod<1> for StateMarketDeals {
     }
 }
 
+pub enum StateMarketDealsDump {}
+
+impl RpcMethod<2> for StateMarketDealsDump {
+    const NAME: &'static str = "Forest.StateMarketDealsDump";
+    const PARAM_NAMES: [&'static str; 2] = ["tipsetKey", "outputFile"];
+    const API_PATHS: BitFlags<ApiPaths> = make_bitflags!(ApiPaths::Experimental);
+    const PERMISSION: Permission = Permission::Read;
+    const DESCRIPTION: Option<&'static str> =
+        Some("Dumps information about every deal in the Storage Market to a NDJSON file.");
+
+    type Params = (ApiTipsetKey, String);
+    type Ok = ();
+
+    async fn handle(
+        ctx: Ctx<impl Blockstore + Send + Sync + 'static>,
+        (ApiTipsetKey(tsk), output_file): Self::Params,
+    ) -> Result<Self::Ok, ServerError> {
+        let ts = ctx.chain_store().load_required_tipset_or_heaviest(&tsk)?;
+        let market_state: market::State = ctx.state_manager.get_actor_state(&ts)?;
+
+        let da = market_state.proposals(ctx.store())?;
+        let sa = market_state.states(ctx.store())?;
+
+        let output_path = PathBuf::from(&output_file);
+        if let Some(parent) = output_path.parent() {
+            std::fs::create_dir_all(parent)
+                .context("Failed to create output directory for market deals")?;
+        }
+        let file = std::fs::File::create(&output_path)
+            .context("Failed to create market deals output file")?;
+        let mut writer = std::io::BufWriter::new(file);
+
+        da.for_each_cacheless(|deal_id, d| {
+            let s = sa.get(deal_id)?.unwrap_or_else(DealState::empty);
+            let market_deal = ApiMarketDeal {
+                proposal: d?.into(),
+                state: s.into(),
+            };
+            write!(
+                writer,
+                "{}\n",
+                crate::lotus_json::HasLotusJson::into_lotus_json_string(market_deal)?
+            )?;
+            Ok(())
+        })?;
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, JsonSchema)]
+#[serde(rename_all = "PascalCase")]
+pub struct StateMarketDealsFilter {
+    #[schemars(with = "LotusJson<Option<Address>>")]
+    #[serde(with = "crate::lotus_json")]
+    pub allowed_clients: Option<Vec<Address>>,
+    #[schemars(with = "LotusJson<Option<Address>>")]
+    #[serde(with = "crate::lotus_json")]
+    pub allowed_providers: Option<Vec<Address>>,
+}
+
+lotus_json_with_self!(StateMarketDealsFilter);
+
+pub enum StateMarketDealsFiltered {}
+
+impl RpcMethod<2> for StateMarketDealsFiltered {
+    const NAME: &'static str = "Forest.StateMarketDealsFiltered";
+    const PARAM_NAMES: [&'static str; 2] = ["tipsetKey", "filter"];
+    const API_PATHS: BitFlags<ApiPaths> = make_bitflags!(ApiPaths::Experimental);
+    const PERMISSION: Permission = Permission::Read;
+    const DESCRIPTION: Option<&'static str> = Some(
+        "Returns information about every deal in the Storage Market, optionally filtered by client and provider addresses.",
+    );
+
+    type Params = (ApiTipsetKey, StateMarketDealsFilter);
+    type Ok = HashMap<u64, ApiMarketDeal>;
+
+    async fn handle(
+        ctx: Ctx<impl Blockstore + Send + Sync + 'static>,
+        (
+            ApiTipsetKey(tsk),
+            StateMarketDealsFilter {
+                allowed_clients,
+                allowed_providers,
+            },
+        ): Self::Params,
+    ) -> Result<Self::Ok, ServerError> {
+        let ts = ctx.chain_store().load_required_tipset_or_heaviest(&tsk)?;
+        let market_state: market::State = ctx.state_manager.get_actor_state(&ts)?;
+
+        let da = market_state.proposals(ctx.store())?;
+        let sa = market_state.states(ctx.store())?;
+
+        let mut out = HashMap::default();
+
+        let allowed_clients = allowed_clients
+            .into_iter()
+            .flatten()
+            .collect::<HashSet<_>>();
+        let allowed_providers = allowed_providers
+            .into_iter()
+            .flatten()
+            .collect::<HashSet<_>>();
+        da.for_each_cacheless(|deal_id, d| {
+            let state = sa.get(deal_id)?.unwrap_or_else(DealState::empty);
+
+            let proposal = d?;
+            if !allowed_clients.contains(&proposal.client.into())
+                && !allowed_providers.contains(&proposal.provider.into())
+            {
+                return Ok(());
+            }
+
+            out.insert(deal_id, MarketDeal { proposal, state }.into());
+            Ok(())
+        })?;
+        Ok(out)
+    }
+}
+
 /// looks up the miner info of the given address.
 pub enum StateMinerInfo {}
 

src/rpc/methods/state.rs

@@ -207,6 +207,8 @@ macro_rules! for_each_rpc_method {
         $callback!($crate::rpc::state::StateLookupRobustAddress);
         $callback!($crate::rpc::state::StateMarketBalance);
         $callback!($crate::rpc::state::StateMarketDeals);
+        $callback!($crate::rpc::state::StateMarketDealsFiltered);
+        $callback!($crate::rpc::state::StateMarketDealsDump);
         $callback!($crate::rpc::state::StateMarketParticipants);
         $callback!($crate::rpc::state::StateMarketStorageDeal);
         $callback!($crate::rpc::state::StateMinerActiveSectors);

src/rpc/mod.rs

@@ -124,6 +124,9 @@ pub enum ApiPaths {
     /// Only expose this method on `/rpc/v2`
     #[strum(ascii_case_insensitive)]
     V2 = 0b00000100,
+    /// Experimental methods, subject to change
+    #[strum(ascii_case_insensitive)]
+    Experimental = 0b10000000,
 }
 
 impl ApiPaths {
@@ -436,6 +439,9 @@ mod tests {
         assert_eq!(v1, ApiPaths::V1);
         let v2 = ApiPaths::from_uri(&"http://127.0.0.1:2345/rpc/v2".parse().unwrap()).unwrap();
         assert_eq!(v2, ApiPaths::V2);
+        let experimental =
+            ApiPaths::from_uri(&"http://127.0.0.1:2345/rpc/experimental".parse().unwrap()).unwrap();
+        assert_eq!(experimental, ApiPaths::Experimental);
 
         ApiPaths::from_uri(&"http://127.0.0.1:2345/rpc/v3".parse().unwrap()).unwrap_err();
     }

src/rpc/reflect/mod.rs

@@ -16,7 +16,12 @@ use tower::Layer;
 static VERSION_METHODS_MAPPINGS: LazyLock<HashMap<ApiPaths, HashSet<&'static str>>> =
     LazyLock::new(|| {
         let mut map = HashMap::default();
-        for version in [ApiPaths::V0, ApiPaths::V1, ApiPaths::V2] {
+        for version in [
+            ApiPaths::V0,
+            ApiPaths::V1,
+            ApiPaths::V2,
+            ApiPaths::Experimental,
+        ] {
             let mut supported = HashSet::default();
 
             macro_rules! insert {

src/rpc/segregation_layer.rs

@@ -386,6 +386,53 @@ where
                 .for_each(|key, deal_proposal| f(key, DealProposal::try_from(deal_proposal)))?),
         }
     }
+    pub fn for_each_cacheless(
+        &self,
+        mut f: impl FnMut(u64, Result<DealProposal, anyhow::Error>) -> anyhow::Result<(), anyhow::Error>,
+    ) -> anyhow::Result<()> {
+        match self {
+            DealProposals::V9(deal_array) => {
+                Ok(deal_array.for_each_cacheless(|key, deal_proposal| {
+                    f(key, DealProposal::try_from(deal_proposal))
+                })?)
+            }
+            DealProposals::V10(deal_array) => {
+                Ok(deal_array.for_each_cacheless(|key, deal_proposal| {
+                    f(key, DealProposal::try_from(deal_proposal))
+                })?)
+            }
+            DealProposals::V11(deal_array) => {
+                Ok(deal_array.for_each_cacheless(|key, deal_proposal| {
+                    f(key, DealProposal::try_from(deal_proposal))
+                })?)
+            }
+            DealProposals::V12(deal_array) => {
+                Ok(deal_array.for_each_cacheless(|key, deal_proposal| {
+                    f(key, DealProposal::try_from(deal_proposal))
+                })?)
+            }
+            DealProposals::V13(deal_array) => {
+                Ok(deal_array.for_each_cacheless(|key, deal_proposal| {
+                    f(key, DealProposal::try_from(deal_proposal))
+                })?)
+            }
+            DealProposals::V14(deal_array) => {
+                Ok(deal_array.for_each_cacheless(|key, deal_proposal| {
+                    f(key, DealProposal::try_from(deal_proposal))
+                })?)
+            }
+            DealProposals::V15(deal_array) => {
+                Ok(deal_array.for_each_cacheless(|key, deal_proposal| {
+                    f(key, DealProposal::try_from(deal_proposal))
+                })?)
+            }
+            DealProposals::V16(deal_array) => {
+                Ok(deal_array.for_each_cacheless(|key, deal_proposal| {
+                    f(key, DealProposal::try_from(deal_proposal))
+                })?)
+            }
+        }
+    }
 
     pub fn get(&self, key: u64) -> anyhow::Result<Option<DealProposal>> {
         match self {
@@ -400,6 +447,19 @@ where
         }
         .transpose()
     }
+
+    pub fn count(&self) -> u64 {
+        match self {
+            DealProposals::V9(deal_array) => deal_array.count(),
+            DealProposals::V10(deal_array) => deal_array.count(),
+            DealProposals::V11(deal_array) => deal_array.count(),
+            DealProposals::V12(deal_array) => deal_array.count(),
+            DealProposals::V13(deal_array) => deal_array.count(),
+            DealProposals::V14(deal_array) => deal_array.count(),
+            DealProposals::V15(deal_array) => deal_array.count(),
+            DealProposals::V16(deal_array) => deal_array.count(),
+        }
+    }
 }
 
 #[derive(Clone, Serialize, Deserialize, PartialEq)]