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 crate::utils::encoding::hex;
168    use std::sync::Arc;
169
170    use super::*;
171    use crate::blocks::RawBlockHeader;
172    use crate::blocks::{CachingBlockHeader, Tipset};
173    use crate::chain::ChainStore;
174    use crate::chain_sync::network_context::SyncNetworkContext;
175    use crate::db::{Blockstore as _, MemoryDB};
176    use crate::key_management::{KeyStore, KeyStoreConfig};
177    use crate::libp2p::{NetworkMessage, PeerManager};
178    use crate::message_pool::{MessagePool, MpoolLocker, NonceTracker};
179    use crate::networks::ChainConfig;
180    use crate::rpc::RPCState;
181    use crate::rpc::eth::filter::EthEventHandler;
182    use crate::shim::address::Address;
183    use crate::state_manager::StateManager;
184    use crate::utils::encoding::from_slice_with_fallback;
185    use parking_lot::RwLock;
186    use tokio::sync::mpsc;
187    use tokio::task::JoinSet;
188
189    fn ctx() -> (Arc<RPCState>, flume::Receiver<NetworkMessage>) {
190        let (network_send, network_rx) = flume::bounded(5);
191        let (tipset_send, _) = flume::bounded(5);
192        let mut services = JoinSet::new();
193        let db = Arc::new(MemoryDB::default());
194        let chain_config = Arc::new(ChainConfig::default());
195
196        let genesis_header = CachingBlockHeader::new(RawBlockHeader {
197            miner_address: Address::new_id(0),
198            timestamp: 7777,
199            ..Default::default()
200        });
201
202        let cs = ChainStore::new(db, chain_config, genesis_header).unwrap();
203        let state_manager = StateManager::new(cs.shallow_clone()).unwrap();
204        let cs_for_test = &cs;
205        let mpool_network_send = network_send.clone();
206        let mpool = {
207            let bz = hex::decode("904300e80781586082cb7477a801f55c1f2ea5e5d1167661feea60a39f697e1099af132682b81cc5047beacf5b6e80d5f52b9fd90323fb8510a5396416dd076c13c85619e176558582744053a3faef6764829aa02132a1571a76aabdc498a638ea0054d3bb57f41d82015860812d2396cc4592cdf7f829374b01ffd03c5469a4b0a9acc5ccc642797aa0a5498b97b28d90820fedc6f79ff0a6005f5c15dbaca3b8a45720af7ed53000555667207a0ccb50073cd24510995abd4c4e45c1e9e114905018b2da9454190499941e818201582012dd0a6a7d0e222a97926da03adb5a7768d31cc7c5c2bd6828e14a7d25fa3a608182004b76616c69642070726f6f6681d82a5827000171a0e4022030f89a8b0373ad69079dbcbc5addfe9b34dce932189786e50d3eb432ede3ba9c43000f0001d82a5827000171a0e4022052238c7d15c100c1b9ebf849541810c9e3c2d86e826512c6c416d2318fcd496dd82a5827000171a0e40220e5658b3d18cd06e1db9015b4b0ec55c123a24d5be1ea24d83938c5b8397b4f2fd82a5827000171a0e4022018d351341c302a21786b585708c9873565a0d07c42521d4aaf52da3ff6f2e461586102c000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001a5f2c5439586102b5cd48724dce0fec8799d77fd6c5113276e7f470c8391faa0b5a6033a3eaf357d635705c36abe10309d73592727289680515afd9d424793ba4796b052682d21b03c5c8a37d94827fecc59cdc5750e198fdf20dee012f4d627c6665132298ab95004500053724e0").unwrap();
208            let header = from_slice_with_fallback::<CachingBlockHeader>(&bz).unwrap();
209            let ts = Tipset::from(header);
210            let db = cs_for_test.db();
211            let tsk = ts.key();
212            cs_for_test.set_heaviest_tipset(ts.clone()).unwrap();
213
214            for i in tsk.to_cids() {
215                let bz2 = bz.clone();
216                db.put_keyed(&i, &bz2).unwrap();
217            }
218
219            MessagePool::new(
220                cs,
221                mpool_network_send,
222                Default::default(),
223                state_manager.chain_config().clone(),
224                &mut services,
225            )
226            .unwrap()
227        };
228        let start_time = chrono::Utc::now();
229
230        let peer_manager = Arc::new(PeerManager::default());
231        let sync_network_context =
232            SyncNetworkContext::new(network_send, peer_manager, state_manager.db_owned());
233        let nonce_tracker = NonceTracker::new();
234        let state = Arc::new(RPCState {
235            state_manager,
236            keystore: Arc::new(RwLock::new(KeyStore::new(KeyStoreConfig::Memory).unwrap())),
237            mpool,
238            bad_blocks: Some(Default::default()),
239            sync_status: Default::default(),
240            eth_event_handler: Arc::new(EthEventHandler::new()),
241            eth_logs_feed: Default::default(),
242            sync_network_context,
243            start_time,
244            shutdown: mpsc::channel(1).0, // dummy for tests
245            tipset_send,
246            snapshot_progress_tracker: Default::default(),
247            mpool_locker: MpoolLocker::new(),
248            nonce_tracker,
249            temp_dir: Arc::new(std::env::temp_dir()),
250        });
251        (state, network_rx)
252    }
253
254    #[tokio::test]
255    async fn set_check_bad() {
256        let (ctx, _) = ctx();
257
258        let cid = "bafy2bzacea3wsdh6y3a36tb3skempjoxqpuyompjbmfeyf34fi3uy6uue42v4"
259            .parse::<Cid>()
260            .unwrap();
261
262        let reason = SyncCheckBad::handle(ctx.clone(), (cid,), &Default::default())
263            .await
264            .unwrap();
265        assert_eq!(reason, "");
266
267        // Mark that block as bad manually and check again to verify
268        SyncMarkBad::handle(ctx.clone(), (cid,), &Default::default())
269            .await
270            .unwrap();
271
272        let reason = SyncCheckBad::handle(ctx.clone(), (cid,), &Default::default())
273            .await
274            .unwrap();
275        assert_eq!(reason, "bad");
276    }
277
278    #[tokio::test]
279    async fn sync_status_test() {
280        let (ctx, _) = ctx();
281
282        let st_copy = ctx.sync_status.clone();
283
284        let sync_status = SyncStatus::handle(ctx.clone(), (), &Default::default())
285            .await
286            .unwrap();
287        assert_eq!(sync_status, st_copy.load().clone());
288
289        // update cloned state
290        st_copy.store(
291            st_copy
292                .load()
293                .as_ref()
294                .clone()
295                .with_status(NodeSyncStatus::Syncing)
296                .with_current_head_epoch(4)
297                .into(),
298        );
299
300        let sync_status = SyncStatus::handle(ctx.clone(), (), &Default::default())
301            .await
302            .unwrap();
303
304        assert_eq!(sync_status, st_copy.load().clone());
305    }
306}
forest/rpc/methods/sync.rs

forest/rpc/methods/
sync.rs
