forest/rpc/methods/
sync.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4mod types;
5
6use crate::blocks::{Block, FullTipset, GossipBlock};
7use crate::chain;
8use crate::chain_sync::{NodeSyncStatus, SyncStatusReport, TipsetValidator};
9use crate::libp2p::{IdentTopic, NetworkMessage, PUBSUB_BLOCK_STR};
10use crate::prelude::*;
11use crate::rpc::{ApiPaths, Ctx, Permission, RpcMethod, ServerError};
12use anyhow::anyhow;
13use enumflags2::BitFlags;
14use fvm_ipld_encoding::to_vec;
15pub use types::*;
16
17pub enum SyncCheckBad {}
18impl RpcMethod<1> for SyncCheckBad {
19    const NAME: &'static str = "Filecoin.SyncCheckBad";
20    const PARAM_NAMES: [&'static str; 1] = ["cid"];
21    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
22    const PERMISSION: Permission = Permission::Read;
23    const DESCRIPTION: &'static str =
24        "Returns the reason the given block is marked bad, or an empty string if it is not.";
25
26    type Params = (Cid,);
27    type Ok = String;
28
29    async fn handle(
30        ctx: Ctx,
31        (cid,): Self::Params,
32        _: &http::Extensions,
33    ) -> Result<Self::Ok, ServerError> {
34        Ok(ctx
35            .bad_blocks
36            .as_ref()
37            .context("bad block cache is disabled")?
38            .get(&cid)
39            .map(|_| "bad".to_string())
40            .unwrap_or_default())
41    }
42}
43
44pub enum SyncMarkBad {}
45impl RpcMethod<1> for SyncMarkBad {
46    const NAME: &'static str = "Filecoin.SyncMarkBad";
47    const PARAM_NAMES: [&'static str; 1] = ["cid"];
48    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
49    const PERMISSION: Permission = Permission::Admin;
50    const DESCRIPTION: &'static str = "Marks the block with the given CID as bad.";
51
52    type Params = (Cid,);
53    type Ok = ();
54
55    async fn handle(
56        ctx: Ctx,
57        (cid,): Self::Params,
58        _: &http::Extensions,
59    ) -> Result<Self::Ok, ServerError> {
60        ctx.bad_blocks
61            .as_ref()
62            .context("bad block cache is disabled")?
63            .push(cid);
64        Ok(())
65    }
66}
67
68pub enum SyncSnapshotProgress {}
69impl RpcMethod<0> for SyncSnapshotProgress {
70    const NAME: &'static str = "Forest.SyncSnapshotProgress";
71    const PARAM_NAMES: [&'static str; 0] = [];
72    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
73    const PERMISSION: Permission = Permission::Read;
74    const DESCRIPTION: &'static str =
75        "Returns the snapshot download progress. Return Null if the tracking isn't started";
76
77    type Params = ();
78    type Ok = SnapshotProgressState;
79
80    async fn handle(
81        ctx: Ctx,
82        (): Self::Params,
83        _: &http::Extensions,
84    ) -> Result<Self::Ok, ServerError> {
85        Ok(ctx.get_snapshot_progress_tracker())
86    }
87}
88
89pub enum SyncStatus {}
90impl RpcMethod<0> for SyncStatus {
91    const NAME: &'static str = "Forest.SyncStatus";
92    const PARAM_NAMES: [&'static str; 0] = [];
93    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
94    const PERMISSION: Permission = Permission::Read;
95    const DESCRIPTION: &'static str = "Returns the current sync status of the node.";
96
97    type Params = ();
98    type Ok = Arc<SyncStatusReport>;
99
100    async fn handle(
101        ctx: Ctx,
102        (): Self::Params,
103        _: &http::Extensions,
104    ) -> Result<Self::Ok, ServerError> {
105        let sync_status = ctx.sync_status.load().shallow_clone();
106        Ok(sync_status)
107    }
108}
109
110pub enum SyncSubmitBlock {}
111impl RpcMethod<1> for SyncSubmitBlock {
112    const NAME: &'static str = "Filecoin.SyncSubmitBlock";
113    const PARAM_NAMES: [&'static str; 1] = ["block"];
114    const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
115    const PERMISSION: Permission = Permission::Write;
116    const DESCRIPTION: &'static str = "Submits a newly created block to the network.";
117
118    type Params = (GossipBlock,);
119    type Ok = ();
120
121    // NOTE: This currently skips all the sanity-checks and directly passes the message onto the
122    // swarm.
123    async fn handle(
124        ctx: Ctx,
125        (block_msg,): Self::Params,
126        _: &http::Extensions,
127    ) -> Result<Self::Ok, ServerError> {
128        if !matches!(ctx.sync_status.load().status, NodeSyncStatus::Synced) {
129            Err(anyhow!("the node isn't in 'follow' mode"))?
130        }
131        let genesis_network_name = ctx.chain_config().network.genesis_name();
132        let encoded_message = to_vec(&block_msg)?;
133        let pubsub_block_str = format!("{PUBSUB_BLOCK_STR}/{genesis_network_name}");
134        let (bls_messages, secp_messages) =
135            chain::store::block_messages(ctx.db(), &block_msg.header)?;
136        let block = Block {
137            header: block_msg.header.clone(),
138            bls_messages,
139            secp_messages,
140        };
141        let ts = FullTipset::from(block);
142        let genesis_ts = ctx.chain_store().genesis_tipset();
143
144        TipsetValidator(&ts)
145            .validate(
146                ctx.chain_store(),
147                ctx.bad_blocks.as_ref(),
148                &genesis_ts,
149                ctx.chain_config().block_delay_secs,
150            )
151            .context("failed to validate the tipset")?;
152
153        ctx.tipset_send
154            .try_send(ts)
155            .context("tipset queue is full")?;
156
157        ctx.network_send().send(NetworkMessage::PubsubMessage {
158            topic: IdentTopic::new(pubsub_block_str),
159            message: encoded_message,
160        })?;
161        Ok(())
162    }
163}
164
165#[cfg(test)]
166mod tests {
167    use std::sync::Arc;
168
169    use super::*;
170    use crate::blocks::RawBlockHeader;
171    use crate::blocks::{CachingBlockHeader, Tipset};
172    use crate::chain::ChainStore;
173    use crate::chain_sync::network_context::SyncNetworkContext;
174    use crate::db::{Blockstore as _, MemoryDB};
175    use crate::key_management::{KeyStore, KeyStoreConfig};
176    use crate::libp2p::{NetworkMessage, PeerManager};
177    use crate::message_pool::{MessagePool, MpoolLocker, NonceTracker};
178    use crate::networks::ChainConfig;
179    use crate::rpc::RPCState;
180    use crate::rpc::eth::filter::EthEventHandler;
181    use crate::shim::address::Address;
182    use crate::state_manager::StateManager;
183    use crate::utils::encoding::from_slice_with_fallback;
184    use parking_lot::RwLock;
185    use tokio::sync::mpsc;
186    use tokio::task::JoinSet;
187
188    fn ctx() -> (Arc<RPCState>, flume::Receiver<NetworkMessage>) {
189        let (network_send, network_rx) = flume::bounded(5);
190        let (tipset_send, _) = flume::bounded(5);
191        let mut services = JoinSet::new();
192        let db = Arc::new(MemoryDB::default());
193        let chain_config = Arc::new(ChainConfig::default());
194
195        let genesis_header = CachingBlockHeader::new(RawBlockHeader {
196            miner_address: Address::new_id(0),
197            timestamp: 7777,
198            ..Default::default()
199        });
200
201        let cs = ChainStore::new(db, chain_config, genesis_header).unwrap();
202        let state_manager = StateManager::new(cs.shallow_clone()).unwrap();
203        let cs_for_test = &cs;
204        let mpool_network_send = network_send.clone();
205        let mpool = {
206            let bz = hex::decode("904300e80781586082cb7477a801f55c1f2ea5e5d1167661feea60a39f697e1099af132682b81cc5047beacf5b6e80d5f52b9fd90323fb8510a5396416dd076c13c85619e176558582744053a3faef6764829aa02132a1571a76aabdc498a638ea0054d3bb57f41d82015860812d2396cc4592cdf7f829374b01ffd03c5469a4b0a9acc5ccc642797aa0a5498b97b28d90820fedc6f79ff0a6005f5c15dbaca3b8a45720af7ed53000555667207a0ccb50073cd24510995abd4c4e45c1e9e114905018b2da9454190499941e818201582012dd0a6a7d0e222a97926da03adb5a7768d31cc7c5c2bd6828e14a7d25fa3a608182004b76616c69642070726f6f6681d82a5827000171a0e4022030f89a8b0373ad69079dbcbc5addfe9b34dce932189786e50d3eb432ede3ba9c43000f0001d82a5827000171a0e4022052238c7d15c100c1b9ebf849541810c9e3c2d86e826512c6c416d2318fcd496dd82a5827000171a0e40220e5658b3d18cd06e1db9015b4b0ec55c123a24d5be1ea24d83938c5b8397b4f2fd82a5827000171a0e4022018d351341c302a21786b585708c9873565a0d07c42521d4aaf52da3ff6f2e461586102c000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001a5f2c5439586102b5cd48724dce0fec8799d77fd6c5113276e7f470c8391faa0b5a6033a3eaf357d635705c36abe10309d73592727289680515afd9d424793ba4796b052682d21b03c5c8a37d94827fecc59cdc5750e198fdf20dee012f4d627c6665132298ab95004500053724e0").unwrap();
207            let header = from_slice_with_fallback::<CachingBlockHeader>(&bz).unwrap();
208            let ts = Tipset::from(header);
209            let db = cs_for_test.db();
210            let tsk = ts.key();
211            cs_for_test.set_heaviest_tipset(ts.clone()).unwrap();
212
213            for i in tsk.to_cids() {
214                let bz2 = bz.clone();
215                db.put_keyed(&i, &bz2).unwrap();
216            }
217
218            MessagePool::new(
219                cs,
220                mpool_network_send,
221                Default::default(),
222                state_manager.chain_config().clone(),
223                &mut services,
224            )
225            .unwrap()
226        };
227        let start_time = chrono::Utc::now();
228
229        let peer_manager = Arc::new(PeerManager::default());
230        let sync_network_context =
231            SyncNetworkContext::new(network_send, peer_manager, state_manager.db_owned());
232        let nonce_tracker = NonceTracker::new();
233        let state = Arc::new(RPCState {
234            state_manager,
235            keystore: Arc::new(RwLock::new(KeyStore::new(KeyStoreConfig::Memory).unwrap())),
236            mpool,
237            bad_blocks: Some(Default::default()),
238            sync_status: Default::default(),
239            eth_event_handler: Arc::new(EthEventHandler::new()),
240            eth_logs_feed: Default::default(),
241            sync_network_context,
242            start_time,
243            shutdown: mpsc::channel(1).0, // dummy for tests
244            tipset_send,
245            snapshot_progress_tracker: Default::default(),
246            mpool_locker: MpoolLocker::new(),
247            nonce_tracker,
248            temp_dir: Arc::new(std::env::temp_dir()),
249        });
250        (state, network_rx)
251    }
252
253    #[tokio::test]
254    async fn set_check_bad() {
255        let (ctx, _) = ctx();
256
257        let cid = "bafy2bzacea3wsdh6y3a36tb3skempjoxqpuyompjbmfeyf34fi3uy6uue42v4"
258            .parse::<Cid>()
259            .unwrap();
260
261        let reason = SyncCheckBad::handle(ctx.clone(), (cid,), &Default::default())
262            .await
263            .unwrap();
264        assert_eq!(reason, "");
265
266        // Mark that block as bad manually and check again to verify
267        SyncMarkBad::handle(ctx.clone(), (cid,), &Default::default())
268            .await
269            .unwrap();
270
271        let reason = SyncCheckBad::handle(ctx.clone(), (cid,), &Default::default())
272            .await
273            .unwrap();
274        assert_eq!(reason, "bad");
275    }
276
277    #[tokio::test]
278    async fn sync_status_test() {
279        let (ctx, _) = ctx();
280
281        let st_copy = ctx.sync_status.clone();
282
283        let sync_status = SyncStatus::handle(ctx.clone(), (), &Default::default())
284            .await
285            .unwrap();
286        assert_eq!(sync_status, st_copy.load().clone());
287
288        // update cloned state
289        st_copy.store(
290            st_copy
291                .load()
292                .as_ref()
293                .clone()
294                .with_status(NodeSyncStatus::Syncing)
295                .with_current_head_epoch(4)
296                .into(),
297        );
298
299        let sync_status = SyncStatus::handle(ctx.clone(), (), &Default::default())
300            .await
301            .unwrap();
302
303        assert_eq!(sync_status, st_copy.load().clone());
304    }
305}
forest/rpc/methods/sync.rs

forest/rpc/methods/
sync.rs
