forest/rpc/
mod.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4mod auth_layer;
5mod channel;
6mod client;
7mod compression_layer;
8mod error;
9mod filter_layer;
10mod filter_list;
11pub mod json_validator;
12mod log_layer;
13mod metrics_layer;
14mod parallel_batch_layer;
15mod reflect;
16mod registry;
17mod request;
18mod segregation_layer;
19mod set_extension_layer;
20pub mod types;
21mod validation_layer;
22
23use crate::db::DbImpl;
24use crate::prelude::*;
25use crate::rpc::eth::types::RandomHexStringIdProvider;
26use crate::rpc::methods::eth::pubsub_trait::EthPubSubApiServer;
27use crate::shim::clock::ChainEpoch;
28use ahash::HashMap;
29use clap::ValueEnum as _;
30pub use client::Client;
31pub use error::ServerError;
32use eth::filter::EthEventHandler;
33use filter_layer::FilterLayer;
34pub use filter_list::FilterList;
35use futures::future::Either;
36use jsonrpsee::server::ServerConfig;
37use log_layer::LogLayer;
38pub use metrics_layer::MetricsMode;
39use parallel_batch_layer::ParallelBatchLayer;
40use reflect::Ctx;
41pub use reflect::{ApiPaths, Permission, RpcMethod, RpcMethodExt};
42pub use request::Request;
43use schemars::Schema;
44use segregation_layer::SegregationLayer;
45use set_extension_layer::SetExtensionLayer;
46
47pub use methods::*;
48
49/// Protocol or transport-specific error
50pub use jsonrpsee::core::ClientError;
51
52/// Sentinel value, indicating no limit on how far back to search in the chain (all the way to genesis epoch).
53pub const LOOKBACK_NO_LIMIT: ChainEpoch = -1;
54
55/// The macro `callback` will be passed in each type that implements
56/// [`RpcMethod`].
57///
58/// This is a macro because there is no way to abstract the `ARITY` on that
59/// trait.
60///
61/// All methods should be entered here.
62#[macro_export]
63macro_rules! for_each_rpc_method {
64    ($callback:path) => {
65        // auth vertical
66        $callback!($crate::rpc::auth::AuthNew);
67        $callback!($crate::rpc::auth::AuthVerify);
68
69        // beacon vertical
70        $callback!($crate::rpc::beacon::BeaconGetEntry);
71
72        // chain vertical
73        $callback!($crate::rpc::chain::ChainPruneSnapshot);
74        $callback!($crate::rpc::chain::ChainExport);
75        $callback!($crate::rpc::chain::ChainGetBlock);
76        $callback!($crate::rpc::chain::ChainGetBlockMessages);
77        $callback!($crate::rpc::chain::ChainGetEvents);
78        $callback!($crate::rpc::chain::ChainGetGenesis);
79        $callback!($crate::rpc::chain::ChainGetFinalizedTipset);
80        $callback!($crate::rpc::chain::ChainGetMessage);
81        $callback!($crate::rpc::chain::ChainGetMessagesInTipset);
82        $callback!($crate::rpc::chain::ChainGetMinBaseFee);
83        $callback!($crate::rpc::chain::ChainGetParentMessages);
84        $callback!($crate::rpc::chain::ChainGetParentReceipts);
85        $callback!($crate::rpc::chain::ChainGetPath);
86        $callback!($crate::rpc::chain::ChainGetTipSet);
87        $callback!($crate::rpc::chain::ChainGetTipSetV2);
88        $callback!($crate::rpc::chain::ChainGetTipSetFinalityStatus);
89        $callback!($crate::rpc::chain::ChainGetTipSetAfterHeight);
90        $callback!($crate::rpc::chain::ChainGetTipSetByHeight);
91        $callback!($crate::rpc::chain::ChainHasObj);
92        $callback!($crate::rpc::chain::ChainHead);
93        $callback!($crate::rpc::chain::ChainReadObj);
94        $callback!($crate::rpc::chain::ChainSetHead);
95        $callback!($crate::rpc::chain::ChainStatObj);
96        $callback!($crate::rpc::chain::ChainTipSetWeight);
97        $callback!($crate::rpc::chain::ForestChainExport);
98        $callback!($crate::rpc::chain::ForestChainExportDiff);
99        $callback!($crate::rpc::chain::ForestChainExportStatus);
100        $callback!($crate::rpc::chain::ForestChainExportCancel);
101        $callback!($crate::rpc::chain::ChainGetTipsetByParentState);
102
103        // common vertical
104        $callback!($crate::rpc::common::Session);
105        $callback!($crate::rpc::common::Shutdown);
106        $callback!($crate::rpc::common::StartTime);
107        $callback!($crate::rpc::common::Version);
108
109        // eth vertical
110        $callback!($crate::rpc::eth::EthAccounts);
111        $callback!($crate::rpc::eth::EthAddressToFilecoinAddress);
112        $callback!($crate::rpc::eth::FilecoinAddressToEthAddress);
113        $callback!($crate::rpc::eth::EthBaseFee);
114        $callback!($crate::rpc::eth::BaseFeeByHeight);
115        $callback!($crate::rpc::eth::EthBlockNumber);
116        $callback!($crate::rpc::eth::EthCall);
117        $callback!($crate::rpc::eth::EthChainId);
118        $callback!($crate::rpc::eth::EthEstimateGas);
119        $callback!($crate::rpc::eth::EthFeeHistory);
120        $callback!($crate::rpc::eth::EthGasPrice);
121        $callback!($crate::rpc::eth::EthGetBalance);
122        $callback!($crate::rpc::eth::EthGetBlockByHash);
123        $callback!($crate::rpc::eth::EthGetBlockByNumber);
124        $callback!($crate::rpc::eth::EthGetBlockReceipts);
125        $callback!($crate::rpc::eth::EthGetBlockReceiptsLimited);
126        $callback!($crate::rpc::eth::EthGetBlockTransactionCountByHash);
127        $callback!($crate::rpc::eth::EthGetBlockTransactionCountByNumber);
128        $callback!($crate::rpc::eth::EthGetCode);
129        $callback!($crate::rpc::eth::EthGetLogs);
130        $callback!($crate::rpc::eth::EthGetFilterLogs);
131        $callback!($crate::rpc::eth::EthGetFilterChanges);
132        $callback!($crate::rpc::eth::EthGetMessageCidByTransactionHash);
133        $callback!($crate::rpc::eth::EthGetStorageAt);
134        $callback!($crate::rpc::eth::EthGetTransactionByHash);
135        $callback!($crate::rpc::eth::EthGetTransactionByHashLimited);
136        $callback!($crate::rpc::eth::EthGetTransactionCount);
137        $callback!($crate::rpc::eth::EthGetTransactionHashByCid);
138        $callback!($crate::rpc::eth::EthGetTransactionByBlockNumberAndIndex);
139        $callback!($crate::rpc::eth::EthGetTransactionByBlockHashAndIndex);
140        $callback!($crate::rpc::eth::EthMaxPriorityFeePerGas);
141        $callback!($crate::rpc::eth::EthProtocolVersion);
142        $callback!($crate::rpc::eth::EthGetTransactionReceipt);
143        $callback!($crate::rpc::eth::EthGetTransactionReceiptLimited);
144        $callback!($crate::rpc::eth::EthNewFilter);
145        $callback!($crate::rpc::eth::EthNewPendingTransactionFilter);
146        $callback!($crate::rpc::eth::EthNewBlockFilter);
147        $callback!($crate::rpc::eth::EthUninstallFilter);
148        $callback!($crate::rpc::eth::EthUnsubscribe);
149        $callback!($crate::rpc::eth::EthSubscribe);
150        $callback!($crate::rpc::eth::EthSyncing);
151        $callback!($crate::rpc::eth::EthTraceBlock);
152        $callback!($crate::rpc::eth::EthTraceCall);
153        $callback!($crate::rpc::eth::EthTraceFilter);
154        $callback!($crate::rpc::eth::EthTraceTransaction);
155        $callback!($crate::rpc::eth::EthDebugTraceTransaction);
156        $callback!($crate::rpc::eth::EthTraceReplayBlockTransactions);
157        $callback!($crate::rpc::eth::Web3ClientVersion);
158        $callback!($crate::rpc::eth::EthSendRawTransaction);
159        $callback!($crate::rpc::eth::EthSendRawTransactionUntrusted);
160
161        // gas vertical
162        $callback!($crate::rpc::gas::GasEstimateFeeCap);
163        $callback!($crate::rpc::gas::GasEstimateGasLimit);
164        $callback!($crate::rpc::gas::GasEstimateGasPremium);
165        $callback!($crate::rpc::gas::GasEstimateMessageGas);
166
167        // market vertical
168        $callback!($crate::rpc::market::MarketAddBalance);
169
170        // miner vertical
171        $callback!($crate::rpc::miner::MinerCreateBlock);
172        $callback!($crate::rpc::miner::MinerGetBaseInfo);
173
174        // mpool vertical
175        $callback!($crate::rpc::mpool::MpoolBatchPush);
176        $callback!($crate::rpc::mpool::MpoolBatchPushUntrusted);
177        $callback!($crate::rpc::mpool::MpoolGetConfig);
178        $callback!($crate::rpc::mpool::MpoolGetNonce);
179        $callback!($crate::rpc::mpool::MpoolPending);
180        $callback!($crate::rpc::mpool::MpoolPush);
181        $callback!($crate::rpc::mpool::MpoolPushMessage);
182        $callback!($crate::rpc::mpool::MpoolPushUntrusted);
183        $callback!($crate::rpc::mpool::MpoolSelect);
184
185        // msig vertical
186        $callback!($crate::rpc::msig::MsigGetAvailableBalance);
187        $callback!($crate::rpc::msig::MsigGetPending);
188        $callback!($crate::rpc::msig::MsigGetVested);
189        $callback!($crate::rpc::msig::MsigGetVestingSchedule);
190
191        // net vertical
192        $callback!($crate::rpc::net::NetAddrsListen);
193        $callback!($crate::rpc::net::NetAgentVersion);
194        $callback!($crate::rpc::net::NetAutoNatStatus);
195        $callback!($crate::rpc::net::NetConnect);
196        $callback!($crate::rpc::net::NetDisconnect);
197        $callback!($crate::rpc::net::NetFindPeer);
198        $callback!($crate::rpc::net::NetInfo);
199        $callback!($crate::rpc::net::NetListening);
200        $callback!($crate::rpc::net::NetPeers);
201        $callback!($crate::rpc::net::NetProtectAdd);
202        $callback!($crate::rpc::net::NetProtectList);
203        $callback!($crate::rpc::net::NetProtectRemove);
204        $callback!($crate::rpc::net::NetVersion);
205        $callback!($crate::rpc::net::NetChainExchange);
206
207        // node vertical
208        $callback!($crate::rpc::node::NodeStatus);
209
210        // state vertical
211        $callback!($crate::rpc::state::StateAccountKey);
212        $callback!($crate::rpc::state::StateCall);
213        $callback!($crate::rpc::state::StateCirculatingSupply);
214        $callback!($crate::rpc::state::ForestStateCompute);
215        $callback!($crate::rpc::state::StateCompute);
216        $callback!($crate::rpc::state::StateDealProviderCollateralBounds);
217        $callback!($crate::rpc::state::StateFetchRoot);
218        $callback!($crate::rpc::state::StateGetActor);
219        $callback!($crate::rpc::state::StateGetActorV2);
220        $callback!($crate::rpc::state::StateGetID);
221        $callback!($crate::rpc::state::StateGetAllAllocations);
222        $callback!($crate::rpc::state::StateGetAllClaims);
223        $callback!($crate::rpc::state::StateGetAllocation);
224        $callback!($crate::rpc::state::StateGetAllocationForPendingDeal);
225        $callback!($crate::rpc::state::StateGetAllocationIdForPendingDeal);
226        $callback!($crate::rpc::state::StateGetAllocations);
227        $callback!($crate::rpc::state::StateGetBeaconEntry);
228        $callback!($crate::rpc::state::StateGetClaim);
229        $callback!($crate::rpc::state::StateGetClaims);
230        $callback!($crate::rpc::state::StateGetNetworkParams);
231        $callback!($crate::rpc::state::StateGetRandomnessDigestFromBeacon);
232        $callback!($crate::rpc::state::StateGetRandomnessDigestFromTickets);
233        $callback!($crate::rpc::state::StateGetRandomnessFromBeacon);
234        $callback!($crate::rpc::state::StateGetRandomnessFromTickets);
235        $callback!($crate::rpc::state::StateGetReceipt);
236        $callback!($crate::rpc::state::StateListActors);
237        $callback!($crate::rpc::state::StateListMessages);
238        $callback!($crate::rpc::state::StateListMiners);
239        $callback!($crate::rpc::state::StateLookupID);
240        $callback!($crate::rpc::state::StateLookupRobustAddress);
241        $callback!($crate::rpc::state::StateMarketBalance);
242        $callback!($crate::rpc::state::StateMarketDeals);
243        $callback!($crate::rpc::state::StateMarketParticipants);
244        $callback!($crate::rpc::state::StateMarketStorageDeal);
245        $callback!($crate::rpc::state::StateMinerActiveSectors);
246        $callback!($crate::rpc::state::StateMinerAllocated);
247        $callback!($crate::rpc::state::StateMinerAvailableBalance);
248        $callback!($crate::rpc::state::StateMinerDeadlines);
249        $callback!($crate::rpc::state::StateMinerFaults);
250        $callback!($crate::rpc::state::StateMinerInfo);
251        $callback!($crate::rpc::state::StateMinerInitialPledgeCollateral);
252        $callback!($crate::rpc::state::StateMinerPartitions);
253        $callback!($crate::rpc::state::StateMinerPower);
254        $callback!($crate::rpc::state::StateMinerPreCommitDepositForPower);
255        $callback!($crate::rpc::state::StateMinerProvingDeadline);
256        $callback!($crate::rpc::state::StateMinerRecoveries);
257        $callback!($crate::rpc::state::StateMinerSectorAllocated);
258        $callback!($crate::rpc::state::StateMinerSectorCount);
259        $callback!($crate::rpc::state::StateMinerSectors);
260        $callback!($crate::rpc::state::StateNetworkName);
261        $callback!($crate::rpc::state::StateNetworkVersion);
262        $callback!($crate::rpc::state::StateActorInfo);
263        $callback!($crate::rpc::state::StateReadState);
264        $callback!($crate::rpc::state::StateDecodeParams);
265        $callback!($crate::rpc::state::StateReplay);
266        $callback!($crate::rpc::state::StateSearchMsg);
267        $callback!($crate::rpc::state::StateSearchMsgLimited);
268        $callback!($crate::rpc::state::StateSectorExpiration);
269        $callback!($crate::rpc::state::StateSectorGetInfo);
270        $callback!($crate::rpc::state::StateSectorPartition);
271        $callback!($crate::rpc::state::StateSectorPreCommitInfo);
272        $callback!($crate::rpc::state::StateSectorPreCommitInfoV0);
273        $callback!($crate::rpc::state::StateVerifiedClientStatus);
274        $callback!($crate::rpc::state::StateVerifiedRegistryRootKey);
275        $callback!($crate::rpc::state::StateVerifierStatus);
276        $callback!($crate::rpc::state::StateVMCirculatingSupplyInternal);
277        $callback!($crate::rpc::state::StateWaitMsg);
278        $callback!($crate::rpc::state::StateWaitMsgV0);
279        $callback!($crate::rpc::state::StateMinerInitialPledgeForSector);
280
281        // sync vertical
282        $callback!($crate::rpc::sync::SyncCheckBad);
283        $callback!($crate::rpc::sync::SyncMarkBad);
284        $callback!($crate::rpc::sync::SyncSnapshotProgress);
285        $callback!($crate::rpc::sync::SyncStatus);
286        $callback!($crate::rpc::sync::SyncSubmitBlock);
287
288        // wallet vertical
289        $callback!($crate::rpc::wallet::WalletBalance);
290        $callback!($crate::rpc::wallet::WalletDefaultAddress);
291        $callback!($crate::rpc::wallet::WalletDelete);
292        $callback!($crate::rpc::wallet::WalletExport);
293        $callback!($crate::rpc::wallet::WalletHas);
294        $callback!($crate::rpc::wallet::WalletImport);
295        $callback!($crate::rpc::wallet::WalletList);
296        $callback!($crate::rpc::wallet::WalletNew);
297        $callback!($crate::rpc::wallet::WalletSetDefault);
298        $callback!($crate::rpc::wallet::WalletSign);
299        $callback!($crate::rpc::wallet::WalletSignMessage);
300        $callback!($crate::rpc::wallet::WalletValidateAddress);
301        $callback!($crate::rpc::wallet::WalletVerify);
302
303        // f3
304        $callback!($crate::rpc::f3::GetRawNetworkName);
305        $callback!($crate::rpc::f3::F3GetCertificate);
306        $callback!($crate::rpc::f3::F3GetECPowerTable);
307        $callback!($crate::rpc::f3::F3GetF3PowerTable);
308        $callback!($crate::rpc::f3::F3GetF3PowerTableByInstance);
309        $callback!($crate::rpc::f3::F3IsRunning);
310        $callback!($crate::rpc::f3::F3GetProgress);
311        $callback!($crate::rpc::f3::F3GetManifest);
312        $callback!($crate::rpc::f3::F3ListParticipants);
313        $callback!($crate::rpc::f3::F3GetLatestCertificate);
314        $callback!($crate::rpc::f3::F3GetOrRenewParticipationTicket);
315        $callback!($crate::rpc::f3::F3Participate);
316        $callback!($crate::rpc::f3::F3ExportLatestSnapshot);
317        $callback!($crate::rpc::f3::GetHead);
318        $callback!($crate::rpc::f3::GetParent);
319        $callback!($crate::rpc::f3::GetParticipatingMinerIDs);
320        $callback!($crate::rpc::f3::GetPowerTable);
321        $callback!($crate::rpc::f3::GetTipset);
322        $callback!($crate::rpc::f3::GetTipsetByEpoch);
323        $callback!($crate::rpc::f3::Finalize);
324        $callback!($crate::rpc::f3::ProtectPeer);
325        $callback!($crate::rpc::f3::SignMessage);
326
327        // misc
328        $callback!($crate::rpc::misc::GetActorEventsRaw);
329    };
330}
331use compression_layer::{COMPRESS_MIN_BODY_SIZE, CompressionLayer};
332pub(crate) use for_each_rpc_method;
333use sync::SnapshotProgressTracker;
334use tower_http::sensitive_headers::SetSensitiveRequestHeadersLayer;
335
336#[allow(unused)]
337/// All handler definitions.
338///
339/// Usage guide:
340/// ```ignore
341/// use crate::rpc::{self, prelude::*};
342///
343/// let client = rpc::Client::from(..);
344/// ChainHead::call(&client, ()).await?;
345/// fn foo() -> rpc::ClientError {..}
346/// fn bar() -> rpc::ServerError {..}
347/// ```
348pub mod prelude {
349    use super::*;
350
351    pub use reflect::RpcMethodExt as _;
352
353    macro_rules! export {
354        ($ty:ty) => {
355            pub use $ty;
356        };
357    }
358
359    for_each_rpc_method!(export);
360}
361
362/// Collects all the RPC method names and permission available in the Forest
363pub fn collect_rpc_method_info() -> Vec<(&'static str, Permission)> {
364    use crate::rpc::RpcMethod;
365
366    let mut methods = Vec::new();
367
368    macro_rules! add_method {
369        ($ty:ty) => {
370            methods.push((<$ty>::NAME, <$ty>::PERMISSION));
371        };
372    }
373
374    for_each_rpc_method!(add_method);
375
376    methods
377}
378
379/// All the methods live in their own folder
380///
381/// # Handling types
382/// - If a `struct` or `enum` is only used in the RPC API, it should live in `src/rpc`.
383///   - If it is used in only one API vertical (i.e `auth` or `chain`), then it should live
384///     in either:
385///     - `src/rpc/methods/auth.rs` (if there are only a few).
386///     - `src/rpc/methods/auth/types.rs` (if there are so many that they would cause clutter).
387///   - If it is used _across_ API verticals, it should live in `src/rpc/types.rs`
388///
389/// # Interactions with the [`lotus_json`] APIs
390/// - Types may have fields which must go through [`LotusJson`],
391///   and MUST reflect that in their [`JsonSchema`].
392///   You have two options for this:
393///   - Use `#[attributes]` to control serialization and schema generation:
394///     ```ignore
395///     #[derive(Deserialize, Serialize, JsonSchema)]
396///     struct Foo {
397///         #[serde(with = "crate::lotus_json")] // perform the conversion
398///         #[schemars(with = "LotusJson<Cid>")] // advertise the schema to be converted
399///         cid: Cid, // use the native type in application logic
400///     }
401///     ```
402///   - Use [`LotusJson`] directly. This means that serialization and the [`JsonSchema`]
403///     will never go out of sync.
404///     ```ignore
405///     #[derive(Deserialize, Serialize, JsonSchema)]
406///     struct Foo {
407///         cid: LotusJson<Cid>, // use the shim type in application logic, manually performing conversions
408///     }
409///     ```
410///
411/// [`lotus_json`]: crate::lotus_json
412/// [`HasLotusJson`]: crate::lotus_json::HasLotusJson
413/// [`LotusJson`]: crate::lotus_json::LotusJson
414/// [`JsonSchema`]: schemars::JsonSchema
415mod methods {
416    pub mod auth;
417    pub mod beacon;
418    pub mod chain;
419    pub mod common;
420    pub mod eth;
421    pub mod f3;
422    pub mod gas;
423    pub mod market;
424    pub mod miner;
425    pub mod misc;
426    pub mod mpool;
427    pub mod msig;
428    pub mod net;
429    pub mod node;
430    pub mod state;
431    pub mod sync;
432    pub mod wallet;
433}
434
435use crate::rpc::auth_layer::{AuthLayer, resolve_claims};
436pub use crate::rpc::channel::CANCEL_METHOD_NAME;
437use crate::rpc::channel::RpcModule as FilRpcModule;
438use crate::rpc::eth::pubsub::EthPubSub;
439use crate::rpc::metrics_layer::MetricsLayer;
440use crate::{chain_sync::network_context::SyncNetworkContext, key_management::KeyStore};
441
442use crate::blocks::FullTipset;
443use crate::utils::misc::env::env_or_default;
444use jsonrpsee::{
445    Methods,
446    core::middleware::RpcServiceBuilder,
447    server::{RpcModule, Server, StopHandle, TowerServiceBuilder},
448};
449use parking_lot::RwLock;
450use std::env;
451use std::sync::{Arc, LazyLock};
452use std::time::Duration;
453use tokio::sync::mpsc;
454use tower::Service;
455
456use crate::rpc::sync::SnapshotProgressState;
457use openrpc_types::{self, ParamStructure};
458
459pub const DEFAULT_PORT: u16 = 2345;
460
461/// Request timeout read from environment variables
462static DEFAULT_REQUEST_TIMEOUT: LazyLock<Duration> = LazyLock::new(|| {
463    env::var("FOREST_RPC_DEFAULT_TIMEOUT")
464        .ok()
465        .and_then(|it| Duration::from_secs(it.parse().ok()?).into())
466        .unwrap_or(Duration::from_secs(60))
467});
468
469/// Maximum concurrent connections accepted by the RPC server.
470///
471/// Configurable via `FOREST_RPC_MAX_CONNECTIONS`. The value also bounds the
472/// TCP listen backlog so that bursts of connection attempts do not get
473/// silently dropped by the kernel.
474pub fn default_max_connections() -> u32 {
475    static VALUE: LazyLock<u32> = LazyLock::new(|| {
476        env::var("FOREST_RPC_MAX_CONNECTIONS")
477            .ok()
478            .and_then(|it| it.parse().ok())
479            .unwrap_or(1000)
480    });
481    *VALUE
482}
483
484const MAX_REQUEST_BODY_SIZE: u32 = 64 * 1024 * 1024;
485
486/// Maximum JSON-RPC response body size in bytes. Defaults to 64 MiB.
487///
488/// `eth_getTransactionReceipt` and `eth_getBlockReceipts` can return very
489/// large responses for log-heavy transactions (a single tx emitting hundreds
490/// of thousands of events can exceed 64 MiB). Operators serving such queries
491/// can raise this with `FOREST_RPC_MAX_RESPONSE_BODY_SIZE` (in bytes).
492static MAX_RESPONSE_BODY_SIZE: LazyLock<u32> =
493    LazyLock::new(|| env_or_default("FOREST_RPC_MAX_RESPONSE_BODY_SIZE", MAX_REQUEST_BODY_SIZE));
494
495/// This is where you store persistent data, or at least access to stateful
496/// data.
497pub struct RPCState {
498    pub keystore: Arc<RwLock<KeyStore>>,
499    pub state_manager: crate::state_manager::StateManager,
500    pub mpool: crate::message_pool::MessagePool<crate::chain::ChainStore>,
501    pub bad_blocks: Option<crate::chain_sync::BadBlockCache>,
502    pub sync_status: crate::chain_sync::SyncStatus,
503    pub eth_event_handler: Arc<EthEventHandler>,
504    pub eth_logs_feed: std::sync::OnceLock<eth::pubsub::LogsFeed>,
505    pub sync_network_context: SyncNetworkContext,
506    pub tipset_send: flume::Sender<FullTipset>,
507    pub start_time: chrono::DateTime<chrono::Utc>,
508    pub snapshot_progress_tracker: SnapshotProgressTracker,
509    pub shutdown: mpsc::Sender<()>,
510    pub mpool_locker: crate::message_pool::MpoolLocker,
511    pub nonce_tracker: crate::message_pool::NonceTracker,
512    pub temp_dir: Arc<std::path::PathBuf>,
513}
514
515impl RPCState {
516    pub fn beacon(&self) -> &Arc<crate::beacon::BeaconSchedule> {
517        self.state_manager.beacon_schedule()
518    }
519
520    pub fn chain_store(&self) -> &crate::chain::ChainStore {
521        self.state_manager.chain_store()
522    }
523
524    pub fn chain_index(&self) -> &crate::chain::index::ChainIndex {
525        self.chain_store().chain_index()
526    }
527
528    pub fn chain_config(&self) -> &Arc<crate::networks::ChainConfig> {
529        self.state_manager.chain_config()
530    }
531
532    pub fn db(&self) -> &DbImpl {
533        self.state_manager.db()
534    }
535
536    pub fn db_owned(&self) -> DbImpl {
537        self.state_manager.db_owned()
538    }
539
540    pub fn network_send(&self) -> &flume::Sender<crate::libp2p::NetworkMessage> {
541        self.sync_network_context.network_send()
542    }
543
544    pub fn get_snapshot_progress_tracker(&self) -> SnapshotProgressState {
545        self.snapshot_progress_tracker.state()
546    }
547}
548
549#[derive(Clone)]
550struct PerConnection<RpcMiddleware, HttpMiddleware> {
551    stop_handle: StopHandle,
552    svc_builder: Arc<TowerServiceBuilder<RpcMiddleware, HttpMiddleware>>,
553    keystore: Arc<RwLock<KeyStore>>,
554}
555
556/// A bare HTTP response carrying just `status` and an empty body.
557fn bare_http_response<B: Default>(status: http::StatusCode) -> http::Response<B> {
558    http::Response::builder()
559        .status(status)
560        .body(B::default())
561        .unwrap_or_else(|_| http::Response::new(B::default()))
562}
563
564pub async fn start_rpc(
565    state: RPCState,
566    rpc_listener: tokio::net::TcpListener,
567    stop_handle: StopHandle,
568    filter_list: Option<Arc<FilterList>>,
569    metrics_mode: MetricsMode,
570) -> anyhow::Result<()> {
571    let filter_list = filter_list.unwrap_or_default();
572    // `Arc` is needed because we will share the state between two modules
573    let state = Arc::new(state);
574    let keystore = state.keystore.shallow_clone();
575    let mut modules = create_modules(state.shallow_clone());
576
577    let mut pubsub_module = FilRpcModule::default();
578    pubsub_module.register_channel("Filecoin.ChainNotify", {
579        let state_clone = state.shallow_clone();
580        move |params| chain::chain_notify(params, &state_clone)
581    })?;
582
583    for module in modules.values_mut() {
584        // register eth subscription APIs
585        module.merge(EthPubSub::new(state.shallow_clone()).into_rpc())?;
586        module.merge(pubsub_module.clone())?;
587    }
588
589    let methods: Arc<HashMap<ApiPaths, Methods>> =
590        Arc::new(modules.into_iter().map(|(k, v)| (k, v.into())).collect());
591
592    let server_config = ServerConfig::builder()
593        .max_request_body_size(MAX_REQUEST_BODY_SIZE)
594        // Default size (10 MiB) is not enough for methods like `Filecoin.StateMinerActiveSectors`
595        .max_response_body_size(*MAX_RESPONSE_BODY_SIZE)
596        .max_connections(default_max_connections())
597        .set_id_provider(RandomHexStringIdProvider::new())
598        .build();
599    let max_response_body_size = *MAX_RESPONSE_BODY_SIZE as usize;
600    let per_conn = PerConnection {
601        stop_handle: stop_handle.clone(),
602        svc_builder: Server::builder()
603            .set_config(server_config)
604            .set_http_middleware(
605                tower::ServiceBuilder::new()
606                    .option_layer(COMPRESS_MIN_BODY_SIZE.map(CompressionLayer::new))
607                    // Mark the `Authorization` request header as sensitive so it doesn't show in logs
608                    .layer(SetSensitiveRequestHeadersLayer::new(std::iter::once(
609                        http::header::AUTHORIZATION,
610                    ))),
611            )
612            .to_service_builder()
613            .into(),
614        keystore,
615    };
616    tracing::info!("Ready for RPC connections");
617    loop {
618        let sock = tokio::select! {
619        res = rpc_listener.accept() => {
620            match res {
621              Ok((stream, _remote_addr)) => {
622                let _ = stream.set_nodelay(true); // Disable Nagle's algorithm
623                stream
624              }
625              Err(e) => {
626                tracing::error!("failed to accept v4 connection: {:?}", e);
627                continue;
628              }
629            }
630          }
631          _ = per_conn.stop_handle.clone().shutdown() => break,
632        };
633
634        let svc = tower::service_fn({
635            let methods = methods.shallow_clone();
636            let per_conn = per_conn.clone();
637            let filter_list = filter_list.shallow_clone();
638            move |req: http::Request<_>| {
639                let svc_or_result = if let Ok(path) = ApiPaths::from_uri(req.uri()) {
640                    let methods = methods.get(&path).cloned().unwrap_or_default();
641                    let PerConnection {
642                        stop_handle,
643                        svc_builder,
644                        keystore,
645                    } = per_conn.clone();
646                    // Authenticate the connection once, here at the HTTP layer (for a
647                    // WebSocket this is the upgrade request), before any JSON-RPC
648                    // dispatch.
649                    match resolve_claims(&keystore, req.headers().get(http::header::AUTHORIZATION))
650                    {
651                        Ok(claims) => {
652                            // NOTE, the rpc middleware must be initialized here to be able to be created once per connection
653                            // with data from the connection such as the headers in this example
654                            let rpc_middleware = RpcServiceBuilder::new()
655                                .layer(SetExtensionLayer { path })
656                                .layer(SegregationLayer)
657                                .layer(FilterLayer::new(filter_list.shallow_clone()))
658                                .layer(validation_layer::JsonValidationLayer)
659                                .layer(AuthLayer::new(claims))
660                                .layer(LogLayer::default())
661                                // `ParallelBatchLayer` fans a batch out into per-entry `call`s, so it must be
662                                // outer to `MetricsLayer` for batched methods to be measured. Both must stay
663                                // inner to the batch-transforming layers above.
664                                .layer(ParallelBatchLayer::new(max_response_body_size))
665                                .layer(MetricsLayer::new(metrics_mode));
666                            Either::Left(
667                                Arc::unwrap_or_clone(svc_builder)
668                                    .set_rpc_middleware(rpc_middleware)
669                                    .build(methods, stop_handle),
670                            )
671                        }
672                        Err(reason) => {
673                            tracing::debug!("rejecting unauthorized request: {reason}");
674                            Either::Right(Ok(bare_http_response(http::StatusCode::UNAUTHORIZED)))
675                        }
676                    }
677                } else {
678                    Either::Right(Ok(bare_http_response(http::StatusCode::NOT_FOUND)))
679                };
680                async move {
681                    match svc_or_result {
682                        Either::Left(mut svc) => {
683                            // https://github.com/rust-lang/rust/issues/102211 the error type can't be inferred
684                            // to be `Box<dyn std::error::Error + Send + Sync>` so we need to convert it to a concrete type
685                            // as workaround.
686                            svc.call(req).await.map_err(|e| anyhow::anyhow!("{:?}", e))
687                        }
688                        Either::Right(result) => result,
689                    }
690                }
691            }
692        });
693
694        tokio::spawn(jsonrpsee::server::serve_with_graceful_shutdown(
695            sock,
696            svc,
697            stop_handle.clone().shutdown(),
698        ));
699    }
700
701    Ok(())
702}
703
704fn create_modules(state: Arc<RPCState>) -> HashMap<ApiPaths, RpcModule<RPCState>> {
705    let mut modules = HashMap::default();
706    for api_version in ApiPaths::value_variants() {
707        modules.insert(*api_version, RpcModule::from_arc(state.shallow_clone()));
708    }
709    macro_rules! register {
710        ($ty:ty) => {
711            // Register only non-subscription RPC methods.
712            // Subscription methods are registered separately in the RPC module.
713            if !<$ty>::SUBSCRIPTION {
714                <$ty>::register(&mut modules, ParamStructure::ByPosition).unwrap();
715            }
716        };
717    }
718    for_each_rpc_method!(register);
719    modules
720}
721
722/// If `include` is not [`None`], only methods that are listed will be returned
723pub fn openrpc(path: ApiPaths, include: Option<&[&str]>) -> openrpc_types::OpenRPC {
724    use schemars::generate::{SchemaGenerator, SchemaSettings};
725
726    let mut methods = vec![];
727    // spec says draft07
728    let mut settings = SchemaSettings::draft07();
729    // ..but uses `components`
730    settings.definitions_path = "#/components/schemas/".into();
731    let mut generator = SchemaGenerator::new(settings);
732    macro_rules! callback {
733        ($ty:ty) => {
734            if <$ty>::API_PATHS.contains(path) {
735                match include {
736                    Some(include) => match include.contains(&<$ty>::NAME) {
737                        true => {
738                            methods.push(openrpc_types::ReferenceOr::Item(<$ty>::openrpc(
739                                &mut generator,
740                                ParamStructure::ByPosition,
741                                &<$ty>::NAME,
742                            )));
743                            if let Some(alias) = &<$ty>::NAME_ALIAS {
744                                methods.push(openrpc_types::ReferenceOr::Item(<$ty>::openrpc(
745                                    &mut generator,
746                                    ParamStructure::ByPosition,
747                                    &alias,
748                                )));
749                            }
750                        }
751                        false => {}
752                    },
753                    None => {
754                        methods.push(openrpc_types::ReferenceOr::Item(<$ty>::openrpc(
755                            &mut generator,
756                            ParamStructure::ByPosition,
757                            &<$ty>::NAME,
758                        )));
759                        if let Some(alias) = &<$ty>::NAME_ALIAS {
760                            methods.push(openrpc_types::ReferenceOr::Item(<$ty>::openrpc(
761                                &mut generator,
762                                ParamStructure::ByPosition,
763                                &alias,
764                            )));
765                        }
766                    }
767                }
768            }
769        };
770    }
771    for_each_rpc_method!(callback);
772    openrpc_types::OpenRPC {
773        methods,
774        components: Some(openrpc_types::Components {
775            schemas: Some(
776                generator
777                    .take_definitions(false)
778                    .into_iter()
779                    .filter_map(|(k, v)| {
780                        if let Ok(v) = Schema::try_from(v) {
781                            Some((k, v))
782                        } else {
783                            None
784                        }
785                    })
786                    .collect(),
787            ),
788            ..Default::default()
789        }),
790        openrpc: openrpc_types::OPEN_RPC_SPECIFICATION_VERSION,
791        info: openrpc_types::Info {
792            title: String::from("forest"),
793            version: env!("CARGO_PKG_VERSION").into(),
794            ..Default::default()
795        },
796        ..Default::default()
797    }
798}
799
800#[cfg(test)]
801mod tests {
802    use super::*;
803    use crate::{
804        db::MemoryDB,
805        networks::NetworkChain,
806        rpc::{client::UrlClient, common::ShiftingVersion},
807        tool::offline_server::server::offline_rpc_state,
808    };
809    use jsonrpsee::{
810        core::{
811            client::{BatchResponse, ClientT},
812            params::BatchRequestBuilder,
813        },
814        server::stop_channel,
815    };
816    use std::net::{Ipv4Addr, SocketAddr};
817    use tokio::task::JoinSet;
818
819    // To update RPC specs:
820    // `cargo test --lib -- rpc::tests::openrpc`
821    // `cargo insta review`
822
823    #[test]
824    fn openrpc_v0() {
825        openrpc(ApiPaths::V0);
826    }
827
828    #[test]
829    fn openrpc_v1() {
830        openrpc(ApiPaths::V1);
831    }
832
833    #[test]
834    fn openrpc_v2() {
835        openrpc(ApiPaths::V2);
836    }
837
838    fn openrpc(path: ApiPaths) {
839        let spec = super::openrpc(path, None);
840        insta::assert_yaml_snapshot!(path.path(), spec);
841    }
842
843    #[test]
844    fn test_rpc_server() {
845        const TIMEOUT: Duration = Duration::from_secs(5);
846        let (done_tx, done_rx) = flume::bounded(1);
847        let rt = tokio::runtime::Builder::new_multi_thread()
848            .enable_all()
849            .build()
850            .unwrap();
851        rt.block_on(async move { test_rpc_server_inner(done_tx).await });
852        done_rx.recv().unwrap();
853        // To mitigate the transient timeout issue
854        rt.shutdown_timeout(TIMEOUT);
855    }
856
857    async fn test_rpc_server_inner(done_tx: flume::Sender<()>) {
858        let chain = NetworkChain::Calibnet;
859        let db = Arc::new(MemoryDB::default());
860        let mut services = JoinSet::new();
861        let (state, mut shutdown_recv) = offline_rpc_state(chain, db, None, None, &mut services)
862            .await
863            .unwrap();
864        let block_delay_secs = state.chain_config().block_delay_secs;
865        let shutdown_send = state.shutdown.clone();
866        let jwt_read_permissions = vec!["read".to_owned()];
867        let jwt_read = super::methods::auth::AuthNew::create_token(
868            &state.keystore.read(),
869            chrono::Duration::hours(1),
870            jwt_read_permissions.clone(),
871        )
872        .unwrap();
873        let rpc_listener =
874            tokio::net::TcpListener::bind(SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 0))
875                .await
876                .unwrap();
877        let rpc_address = rpc_listener.local_addr().unwrap();
878        let (stop_handle, server_handle) = stop_channel();
879
880        // Start an RPC server
881
882        let handle = tokio::spawn(start_rpc(
883            state,
884            rpc_listener,
885            stop_handle,
886            None,
887            MetricsMode::Enabled,
888        ));
889
890        println!("sending a few http requests");
891
892        let client = Client::from_url(
893            format!("http://{}:{}/", rpc_address.ip(), rpc_address.port())
894                .parse()
895                .unwrap(),
896        );
897
898        let response = super::methods::common::Version::call(&client, ())
899            .await
900            .unwrap();
901        assert_eq!(
902            &response.version,
903            &*crate::utils::version::FOREST_VERSION_STRING
904        );
905        assert_eq!(response.block_delay, block_delay_secs);
906        assert_eq!(response.api_version, ShiftingVersion::new(2, 3, 0));
907
908        let response = super::methods::auth::AuthVerify::call(&client, (jwt_read.clone(),))
909            .await
910            .unwrap();
911        assert_eq!(response, jwt_read_permissions);
912
913        // `AuthVerify` verifies a raw JWT; a `Bearer `-prefixed argument must fail.
914        super::methods::auth::AuthVerify::call(&client, (format!("Bearer {jwt_read}"),))
915            .await
916            .unwrap_err();
917
918        drop(client);
919
920        // A bad token is rejected with a bare HTTP 401, before JSON-RPC dispatch.
921        let http = reqwest::Client::new();
922        let rpc_url = format!("http://{}:{}/rpc/v1", rpc_address.ip(), rpc_address.port());
923        let jsonrpc_body = serde_json::json!({
924            "jsonrpc": "2.0",
925            "method": "Filecoin.Version",
926            "params": [],
927            "id": 0,
928        });
929
930        let bad_token = http
931            .post(&rpc_url)
932            .header(reqwest::header::AUTHORIZATION, "Bearer not-a-real-token")
933            .json(&jsonrpc_body)
934            .send()
935            .await
936            .unwrap();
937        assert_eq!(bad_token.status(), reqwest::StatusCode::UNAUTHORIZED);
938
939        // A valid token is not rejected at the HTTP layer.
940        let good_token = http
941            .post(&rpc_url)
942            .header(reqwest::header::AUTHORIZATION, format!("Bearer {jwt_read}"))
943            .json(&jsonrpc_body)
944            .send()
945            .await
946            .unwrap();
947        assert_eq!(good_token.status(), reqwest::StatusCode::OK);
948
949        println!("sending a few websocket requests");
950
951        let client = Client::from_url(
952            format!("ws://{}:{}/", rpc_address.ip(), rpc_address.port())
953                .parse()
954                .unwrap(),
955        );
956
957        let response = super::methods::auth::AuthVerify::call(&client, (jwt_read,))
958            .await
959            .unwrap();
960        assert_eq!(response, jwt_read_permissions);
961
962        drop(client);
963
964        // Sending a batch request
965        let client = UrlClient::new(
966            format!("http://{}:{}/rpc/v1", rpc_address.ip(), rpc_address.port())
967                .parse()
968                .unwrap(),
969            None,
970        )
971        .await
972        .unwrap();
973        let mut batch_request_builder = BatchRequestBuilder::new();
974        let empty_payload: [(); 0] = [];
975        batch_request_builder
976            .insert("Filecoin.Version", empty_payload)
977            .unwrap();
978        batch_request_builder
979            .insert("eth_chainId", empty_payload)
980            .unwrap();
981        let batch_response: BatchResponse<serde_json::Value> =
982            client.batch_request(batch_request_builder).await.unwrap();
983        assert_eq!(batch_response.len(), 2);
984        assert_eq!(batch_response.num_successful_calls(), 2);
985        assert_eq!(batch_response.num_failed_calls(), 0);
986
987        // `eth_chainId` is only ever requested inside the batch above, so its presence in the RPC
988        // timing metric proves batched methods flow through `MetricsLayer`. Guards against batch
989        // entries bypassing metrics (which happens if `MetricsLayer` is outer to `ParallelBatchLayer`).
990        let mut encoded = String::new();
991        prometheus_client::encoding::text::encode_registry(
992            &mut encoded,
993            &crate::metrics::default_registry(),
994        )
995        .unwrap();
996        let recorded = encoded.lines().any(|line| {
997            line.starts_with("rpc_processing_time_count{")
998                && line.contains(r#"method="eth_chainId""#)
999                && line
1000                    .rsplit(' ')
1001                    .next()
1002                    .and_then(|v| v.parse::<u64>().ok())
1003                    .is_some_and(|count| count == 1)
1004        });
1005        assert!(
1006            recorded,
1007            "batched method `eth_chainId` was not recorded in rpc_processing_time:\n{encoded}"
1008        );
1009
1010        // Gracefully shutdown the RPC server
1011        println!("sending shutdown signal");
1012        shutdown_send.send(()).await.unwrap();
1013        println!("waiting on shutdown receiver");
1014        shutdown_recv.recv().await;
1015        println!("sending server stop signal");
1016        server_handle.stop().unwrap();
1017        println!("waiting on graceful shutdown");
1018        handle.await.unwrap().unwrap();
1019        println!("done");
1020        done_tx.send(()).unwrap();
1021    }
1022}
forest/rpc/mod.rs

forest/rpc/
mod.rs
