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, humanize_rpc_error};
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::IndexBackfill);
102        $callback!($crate::rpc::chain::IndexBackfillStatus);
103        $callback!($crate::rpc::chain::IndexBackfillCancel);
104        $callback!($crate::rpc::chain::ChainGetTipsetByParentState);
105
106        // common vertical
107        $callback!($crate::rpc::common::Session);
108        $callback!($crate::rpc::common::Shutdown);
109        $callback!($crate::rpc::common::StartTime);
110        $callback!($crate::rpc::common::Version);
111
112        // eth vertical
113        $callback!($crate::rpc::eth::EthAccounts);
114        $callback!($crate::rpc::eth::EthAddressToFilecoinAddress);
115        $callback!($crate::rpc::eth::FilecoinAddressToEthAddress);
116        $callback!($crate::rpc::eth::EthBaseFee);
117        $callback!($crate::rpc::eth::BaseFeeByHeight);
118        $callback!($crate::rpc::eth::EthBlockNumber);
119        $callback!($crate::rpc::eth::EthCall);
120        $callback!($crate::rpc::eth::EthChainId);
121        $callback!($crate::rpc::eth::EthEstimateGas);
122        $callback!($crate::rpc::eth::EthFeeHistory);
123        $callback!($crate::rpc::eth::EthGasPrice);
124        $callback!($crate::rpc::eth::EthGetBalance);
125        $callback!($crate::rpc::eth::EthGetBlockByHash);
126        $callback!($crate::rpc::eth::EthGetBlockByNumber);
127        $callback!($crate::rpc::eth::EthGetBlockReceipts);
128        $callback!($crate::rpc::eth::EthGetBlockReceiptsLimited);
129        $callback!($crate::rpc::eth::EthGetBlockTransactionCountByHash);
130        $callback!($crate::rpc::eth::EthGetBlockTransactionCountByNumber);
131        $callback!($crate::rpc::eth::EthGetCode);
132        $callback!($crate::rpc::eth::EthGetLogs);
133        $callback!($crate::rpc::eth::EthGetFilterLogs);
134        $callback!($crate::rpc::eth::EthGetFilterChanges);
135        $callback!($crate::rpc::eth::EthGetMessageCidByTransactionHash);
136        $callback!($crate::rpc::eth::EthGetStorageAt);
137        $callback!($crate::rpc::eth::EthGetTransactionByHash);
138        $callback!($crate::rpc::eth::EthGetTransactionByHashLimited);
139        $callback!($crate::rpc::eth::EthGetTransactionCount);
140        $callback!($crate::rpc::eth::EthGetTransactionHashByCid);
141        $callback!($crate::rpc::eth::EthGetTransactionByBlockNumberAndIndex);
142        $callback!($crate::rpc::eth::EthGetTransactionByBlockHashAndIndex);
143        $callback!($crate::rpc::eth::EthMaxPriorityFeePerGas);
144        $callback!($crate::rpc::eth::EthProtocolVersion);
145        $callback!($crate::rpc::eth::EthGetTransactionReceipt);
146        $callback!($crate::rpc::eth::EthGetTransactionReceiptLimited);
147        $callback!($crate::rpc::eth::EthNewFilter);
148        $callback!($crate::rpc::eth::EthNewPendingTransactionFilter);
149        $callback!($crate::rpc::eth::EthNewBlockFilter);
150        $callback!($crate::rpc::eth::EthUninstallFilter);
151        $callback!($crate::rpc::eth::EthUnsubscribe);
152        $callback!($crate::rpc::eth::EthSubscribe);
153        $callback!($crate::rpc::eth::EthSyncing);
154        $callback!($crate::rpc::eth::EthTraceBlock);
155        $callback!($crate::rpc::eth::EthTraceCall);
156        $callback!($crate::rpc::eth::EthTraceFilter);
157        $callback!($crate::rpc::eth::EthTraceTransaction);
158        $callback!($crate::rpc::eth::EthDebugTraceTransaction);
159        $callback!($crate::rpc::eth::EthTraceReplayBlockTransactions);
160        $callback!($crate::rpc::eth::Web3ClientVersion);
161        $callback!($crate::rpc::eth::EthSendRawTransaction);
162        $callback!($crate::rpc::eth::EthSendRawTransactionUntrusted);
163
164        // gas vertical
165        $callback!($crate::rpc::gas::GasEstimateFeeCap);
166        $callback!($crate::rpc::gas::GasEstimateGasLimit);
167        $callback!($crate::rpc::gas::GasEstimateGasPremium);
168        $callback!($crate::rpc::gas::GasEstimateMessageGas);
169
170        // market vertical
171        $callback!($crate::rpc::market::MarketAddBalance);
172
173        // miner vertical
174        $callback!($crate::rpc::miner::MinerCreateBlock);
175        $callback!($crate::rpc::miner::MinerGetBaseInfo);
176
177        // mpool vertical
178        $callback!($crate::rpc::mpool::MpoolBatchPush);
179        $callback!($crate::rpc::mpool::MpoolBatchPushUntrusted);
180        $callback!($crate::rpc::mpool::MpoolGetConfig);
181        $callback!($crate::rpc::mpool::MpoolGetNonce);
182        $callback!($crate::rpc::mpool::MpoolPending);
183        $callback!($crate::rpc::mpool::MpoolPush);
184        $callback!($crate::rpc::mpool::MpoolPushMessage);
185        $callback!($crate::rpc::mpool::MpoolPushUntrusted);
186        $callback!($crate::rpc::mpool::MpoolSelect);
187
188        // msig vertical
189        $callback!($crate::rpc::msig::MsigGetAvailableBalance);
190        $callback!($crate::rpc::msig::MsigGetPending);
191        $callback!($crate::rpc::msig::MsigGetVested);
192        $callback!($crate::rpc::msig::MsigGetVestingSchedule);
193
194        // net vertical
195        $callback!($crate::rpc::net::NetAddrsListen);
196        $callback!($crate::rpc::net::NetAgentVersion);
197        $callback!($crate::rpc::net::NetAutoNatStatus);
198        $callback!($crate::rpc::net::NetBandwidthStats);
199        $callback!($crate::rpc::net::NetConnect);
200        $callback!($crate::rpc::net::NetDisconnect);
201        $callback!($crate::rpc::net::NetFindPeer);
202        $callback!($crate::rpc::net::NetInfo);
203        $callback!($crate::rpc::net::NetListening);
204        $callback!($crate::rpc::net::NetPeers);
205        $callback!($crate::rpc::net::NetProtectAdd);
206        $callback!($crate::rpc::net::NetProtectList);
207        $callback!($crate::rpc::net::NetProtectRemove);
208        $callback!($crate::rpc::net::NetVersion);
209        $callback!($crate::rpc::net::NetChainExchange);
210
211        // node vertical
212        $callback!($crate::rpc::node::NodeStatus);
213
214        // state vertical
215        $callback!($crate::rpc::state::StateAccountKey);
216        $callback!($crate::rpc::state::StateCall);
217        $callback!($crate::rpc::state::StateCirculatingSupply);
218        $callback!($crate::rpc::state::ForestStateCompute);
219        $callback!($crate::rpc::state::StateCompute);
220        $callback!($crate::rpc::state::StateDealProviderCollateralBounds);
221        $callback!($crate::rpc::state::StateFetchRoot);
222        $callback!($crate::rpc::state::StateGetActor);
223        $callback!($crate::rpc::state::StateGetActorV2);
224        $callback!($crate::rpc::state::StateGetID);
225        $callback!($crate::rpc::state::StateGetAllAllocations);
226        $callback!($crate::rpc::state::StateGetAllClaims);
227        $callback!($crate::rpc::state::StateGetAllocation);
228        $callback!($crate::rpc::state::StateGetAllocationForPendingDeal);
229        $callback!($crate::rpc::state::StateGetAllocationIdForPendingDeal);
230        $callback!($crate::rpc::state::StateGetAllocations);
231        $callback!($crate::rpc::state::StateGetBeaconEntry);
232        $callback!($crate::rpc::state::StateGetClaim);
233        $callback!($crate::rpc::state::StateGetClaims);
234        $callback!($crate::rpc::state::StateGetNetworkParams);
235        $callback!($crate::rpc::state::StateGetRandomnessDigestFromBeacon);
236        $callback!($crate::rpc::state::StateGetRandomnessDigestFromTickets);
237        $callback!($crate::rpc::state::StateGetRandomnessFromBeacon);
238        $callback!($crate::rpc::state::StateGetRandomnessFromTickets);
239        $callback!($crate::rpc::state::StateGetReceipt);
240        $callback!($crate::rpc::state::StateListActors);
241        $callback!($crate::rpc::state::StateListMessages);
242        $callback!($crate::rpc::state::StateListMiners);
243        $callback!($crate::rpc::state::StateLookupID);
244        $callback!($crate::rpc::state::StateLookupRobustAddress);
245        $callback!($crate::rpc::state::StateMarketBalance);
246        $callback!($crate::rpc::state::StateMarketDeals);
247        $callback!($crate::rpc::state::StateMarketParticipants);
248        $callback!($crate::rpc::state::StateMarketStorageDeal);
249        $callback!($crate::rpc::state::StateMinerActiveSectors);
250        $callback!($crate::rpc::state::StateMinerAllocated);
251        $callback!($crate::rpc::state::StateMinerAvailableBalance);
252        $callback!($crate::rpc::state::StateMinerDeadlines);
253        $callback!($crate::rpc::state::StateMinerFaults);
254        $callback!($crate::rpc::state::StateMinerInfo);
255        $callback!($crate::rpc::state::StateMinerInitialPledgeCollateral);
256        $callback!($crate::rpc::state::StateMinerPartitions);
257        $callback!($crate::rpc::state::StateMinerPower);
258        $callback!($crate::rpc::state::StateMinerPreCommitDepositForPower);
259        $callback!($crate::rpc::state::StateMinerProvingDeadline);
260        $callback!($crate::rpc::state::StateMinerRecoveries);
261        $callback!($crate::rpc::state::StateMinerSectorAllocated);
262        $callback!($crate::rpc::state::StateMinerSectorCount);
263        $callback!($crate::rpc::state::StateMinerSectors);
264        $callback!($crate::rpc::state::StateNetworkName);
265        $callback!($crate::rpc::state::StateNetworkVersion);
266        $callback!($crate::rpc::state::StateActorInfo);
267        $callback!($crate::rpc::state::StateReadState);
268        $callback!($crate::rpc::state::StateDecodeParams);
269        $callback!($crate::rpc::state::StateReplay);
270        $callback!($crate::rpc::state::StateSearchMsg);
271        $callback!($crate::rpc::state::StateSearchMsgLimited);
272        $callback!($crate::rpc::state::StateSectorExpiration);
273        $callback!($crate::rpc::state::StateSectorGetInfo);
274        $callback!($crate::rpc::state::StateSectorPartition);
275        $callback!($crate::rpc::state::StateSectorPreCommitInfo);
276        $callback!($crate::rpc::state::StateSectorPreCommitInfoV0);
277        $callback!($crate::rpc::state::StateVerifiedClientStatus);
278        $callback!($crate::rpc::state::StateVerifiedRegistryRootKey);
279        $callback!($crate::rpc::state::StateVerifierStatus);
280        $callback!($crate::rpc::state::StateVMCirculatingSupplyInternal);
281        $callback!($crate::rpc::state::StateWaitMsg);
282        $callback!($crate::rpc::state::StateWaitMsgV0);
283        $callback!($crate::rpc::state::StateMinerInitialPledgeForSector);
284        $callback!($crate::rpc::state::StateMinerCreationDeposit);
285
286        // sync vertical
287        $callback!($crate::rpc::sync::SyncCheckBad);
288        $callback!($crate::rpc::sync::SyncMarkBad);
289        $callback!($crate::rpc::sync::SyncSnapshotProgress);
290        $callback!($crate::rpc::sync::SyncStatus);
291        $callback!($crate::rpc::sync::SyncSubmitBlock);
292
293        // wallet vertical
294        $callback!($crate::rpc::wallet::WalletBalance);
295        $callback!($crate::rpc::wallet::WalletDefaultAddress);
296        $callback!($crate::rpc::wallet::WalletDelete);
297        $callback!($crate::rpc::wallet::WalletExport);
298        $callback!($crate::rpc::wallet::WalletHas);
299        $callback!($crate::rpc::wallet::WalletImport);
300        $callback!($crate::rpc::wallet::WalletList);
301        $callback!($crate::rpc::wallet::WalletNew);
302        $callback!($crate::rpc::wallet::WalletSetDefault);
303        $callback!($crate::rpc::wallet::WalletSign);
304        $callback!($crate::rpc::wallet::WalletSignMessage);
305        $callback!($crate::rpc::wallet::WalletValidateAddress);
306        $callback!($crate::rpc::wallet::WalletVerify);
307
308        // f3
309        $callback!($crate::rpc::f3::GetRawNetworkName);
310        $callback!($crate::rpc::f3::F3GetCertificate);
311        $callback!($crate::rpc::f3::F3GetECPowerTable);
312        $callback!($crate::rpc::f3::F3GetF3PowerTable);
313        $callback!($crate::rpc::f3::F3GetF3PowerTableByInstance);
314        $callback!($crate::rpc::f3::F3IsRunning);
315        $callback!($crate::rpc::f3::F3GetProgress);
316        $callback!($crate::rpc::f3::F3GetManifest);
317        $callback!($crate::rpc::f3::F3ListParticipants);
318        $callback!($crate::rpc::f3::F3GetLatestCertificate);
319        $callback!($crate::rpc::f3::F3GetOrRenewParticipationTicket);
320        $callback!($crate::rpc::f3::F3Participate);
321        $callback!($crate::rpc::f3::F3ExportLatestSnapshot);
322        $callback!($crate::rpc::f3::GetHead);
323        $callback!($crate::rpc::f3::GetParent);
324        $callback!($crate::rpc::f3::GetParticipatingMinerIDs);
325        $callback!($crate::rpc::f3::GetPowerTable);
326        $callback!($crate::rpc::f3::GetTipset);
327        $callback!($crate::rpc::f3::GetTipsetByEpoch);
328        $callback!($crate::rpc::f3::Finalize);
329        $callback!($crate::rpc::f3::ProtectPeer);
330        $callback!($crate::rpc::f3::SignMessage);
331
332        // misc
333        $callback!($crate::rpc::misc::GetActorEventsRaw);
334    };
335}
336use compression_layer::{COMPRESS_MIN_BODY_SIZE, CompressionLayer};
337pub(crate) use for_each_rpc_method;
338use sync::SnapshotProgressTracker;
339use tower_http::sensitive_headers::SetSensitiveRequestHeadersLayer;
340
341#[allow(unused)]
342/// All handler definitions.
343///
344/// Usage guide:
345/// ```ignore
346/// use crate::rpc::{self, prelude::*};
347///
348/// let client = rpc::Client::from(..);
349/// ChainHead::call(&client, ()).await?;
350/// fn foo() -> rpc::ClientError {..}
351/// fn bar() -> rpc::ServerError {..}
352/// ```
353pub mod prelude {
354    use super::*;
355
356    pub use reflect::RpcMethodExt as _;
357
358    macro_rules! export {
359        ($ty:ty) => {
360            pub use $ty;
361        };
362    }
363
364    for_each_rpc_method!(export);
365}
366
367/// Collects all the RPC method names and permission available in the Forest
368pub fn collect_rpc_method_info() -> Vec<(&'static str, Permission)> {
369    use crate::rpc::RpcMethod;
370
371    let mut methods = Vec::new();
372
373    macro_rules! add_method {
374        ($ty:ty) => {
375            methods.push((<$ty>::NAME, <$ty>::PERMISSION));
376        };
377    }
378
379    for_each_rpc_method!(add_method);
380
381    methods
382}
383
384/// All the methods live in their own folder
385///
386/// # Handling types
387/// - If a `struct` or `enum` is only used in the RPC API, it should live in `src/rpc`.
388///   - If it is used in only one API vertical (i.e `auth` or `chain`), then it should live
389///     in either:
390///     - `src/rpc/methods/auth.rs` (if there are only a few).
391///     - `src/rpc/methods/auth/types.rs` (if there are so many that they would cause clutter).
392///   - If it is used _across_ API verticals, it should live in `src/rpc/types.rs`
393///
394/// # Interactions with the [`lotus_json`] APIs
395/// - Types may have fields which must go through [`LotusJson`],
396///   and MUST reflect that in their [`JsonSchema`].
397///   You have two options for this:
398///   - Use `#[attributes]` to control serialization and schema generation:
399///     ```ignore
400///     #[derive(Deserialize, Serialize, JsonSchema)]
401///     struct Foo {
402///         #[serde(with = "crate::lotus_json")] // perform the conversion
403///         #[schemars(with = "LotusJson<Cid>")] // advertise the schema to be converted
404///         cid: Cid, // use the native type in application logic
405///     }
406///     ```
407///   - Use [`LotusJson`] directly. This means that serialization and the [`JsonSchema`]
408///     will never go out of sync.
409///     ```ignore
410///     #[derive(Deserialize, Serialize, JsonSchema)]
411///     struct Foo {
412///         cid: LotusJson<Cid>, // use the shim type in application logic, manually performing conversions
413///     }
414///     ```
415///
416/// [`lotus_json`]: crate::lotus_json
417/// [`HasLotusJson`]: crate::lotus_json::HasLotusJson
418/// [`LotusJson`]: crate::lotus_json::LotusJson
419/// [`JsonSchema`]: schemars::JsonSchema
420mod methods {
421    pub mod auth;
422    pub mod beacon;
423    pub mod chain;
424    pub mod common;
425    pub mod eth;
426    pub mod f3;
427    pub mod gas;
428    pub mod market;
429    pub mod miner;
430    pub mod misc;
431    pub mod mpool;
432    pub mod msig;
433    pub mod net;
434    pub mod node;
435    pub mod state;
436    pub mod sync;
437    pub mod wallet;
438}
439
440use crate::rpc::auth_layer::{AuthLayer, resolve_claims};
441pub use crate::rpc::channel::CANCEL_METHOD_NAME;
442use crate::rpc::channel::RpcModule as FilRpcModule;
443use crate::rpc::eth::pubsub::EthPubSub;
444use crate::rpc::metrics_layer::MetricsLayer;
445use crate::{chain_sync::network_context::SyncNetworkContext, key_management::KeyStore};
446
447use crate::blocks::FullTipset;
448use crate::utils::misc::env::env_or_default;
449use jsonrpsee::{
450    Methods,
451    core::middleware::RpcServiceBuilder,
452    server::{RpcModule, Server, StopHandle, TowerServiceBuilder},
453};
454use parking_lot::RwLock;
455use std::env;
456use std::sync::{Arc, LazyLock};
457use std::time::Duration;
458use tokio::sync::mpsc;
459use tower::Service;
460
461use crate::rpc::sync::SnapshotProgressState;
462use openrpc_types::{self, ParamStructure};
463
464pub const DEFAULT_PORT: u16 = 2345;
465
466/// Request timeout read from environment variables
467static DEFAULT_REQUEST_TIMEOUT: LazyLock<Duration> = LazyLock::new(|| {
468    env::var("FOREST_RPC_DEFAULT_TIMEOUT")
469        .ok()
470        .and_then(|it| Duration::from_secs(it.parse().ok()?).into())
471        .unwrap_or(Duration::from_secs(60))
472});
473
474/// Maximum concurrent connections accepted by the RPC server.
475///
476/// Configurable via `FOREST_RPC_MAX_CONNECTIONS`. The value also bounds the
477/// TCP listen backlog so that bursts of connection attempts do not get
478/// silently dropped by the kernel.
479pub fn default_max_connections() -> u32 {
480    static VALUE: LazyLock<u32> = LazyLock::new(|| {
481        env::var("FOREST_RPC_MAX_CONNECTIONS")
482            .ok()
483            .and_then(|it| it.parse().ok())
484            .unwrap_or(1000)
485    });
486    *VALUE
487}
488
489const MAX_REQUEST_BODY_SIZE: u32 = 64 * 1024 * 1024;
490
491/// Maximum JSON-RPC response body size in bytes. Defaults to 64 MiB.
492///
493/// `eth_getTransactionReceipt` and `eth_getBlockReceipts` can return very
494/// large responses for log-heavy transactions (a single tx emitting hundreds
495/// of thousands of events can exceed 64 MiB). Operators serving such queries
496/// can raise this with `FOREST_RPC_MAX_RESPONSE_BODY_SIZE` (in bytes).
497static MAX_RESPONSE_BODY_SIZE: LazyLock<u32> =
498    LazyLock::new(|| env_or_default("FOREST_RPC_MAX_RESPONSE_BODY_SIZE", MAX_REQUEST_BODY_SIZE));
499
500/// This is where you store persistent data, or at least access to stateful
501/// data.
502pub struct RPCState {
503    pub keystore: Arc<RwLock<KeyStore>>,
504    pub state_manager: crate::state_manager::StateManager,
505    pub mpool: crate::message_pool::MessagePool<crate::chain::ChainStore>,
506    pub bad_blocks: Option<crate::chain_sync::BadBlockCache>,
507    pub sync_status: crate::chain_sync::SyncStatus,
508    pub eth_event_handler: Arc<EthEventHandler>,
509    pub eth_logs_feed: std::sync::OnceLock<eth::pubsub::LogsFeed>,
510    pub sync_network_context: SyncNetworkContext,
511    pub tipset_send: flume::Sender<FullTipset>,
512    pub start_time: chrono::DateTime<chrono::Utc>,
513    pub snapshot_progress_tracker: SnapshotProgressTracker,
514    pub shutdown: mpsc::Sender<()>,
515    pub mpool_locker: crate::message_pool::MpoolLocker,
516    pub nonce_tracker: crate::message_pool::NonceTracker,
517    pub temp_dir: Arc<std::path::PathBuf>,
518}
519
520impl RPCState {
521    pub fn beacon(&self) -> &Arc<crate::beacon::BeaconSchedule> {
522        self.state_manager.beacon_schedule()
523    }
524
525    pub fn chain_store(&self) -> &crate::chain::ChainStore {
526        self.state_manager.chain_store()
527    }
528
529    pub fn chain_index(&self) -> &crate::chain::index::ChainIndex {
530        self.chain_store().chain_index()
531    }
532
533    pub fn chain_config(&self) -> &Arc<crate::networks::ChainConfig> {
534        self.state_manager.chain_config()
535    }
536
537    pub fn genesis_info(&self) -> &Arc<crate::state_manager::circulating_supply::GenesisInfo> {
538        self.state_manager.genesis_info()
539    }
540
541    pub fn db(&self) -> &DbImpl {
542        self.state_manager.db()
543    }
544
545    pub fn db_owned(&self) -> DbImpl {
546        self.state_manager.db_owned()
547    }
548
549    pub fn network_send(&self) -> &flume::Sender<crate::libp2p::NetworkMessage> {
550        self.sync_network_context.network_send()
551    }
552
553    pub fn get_snapshot_progress_tracker(&self) -> SnapshotProgressState {
554        self.snapshot_progress_tracker.state()
555    }
556}
557
558#[derive(Clone)]
559struct PerConnection<RpcMiddleware, HttpMiddleware> {
560    stop_handle: StopHandle,
561    svc_builder: Arc<TowerServiceBuilder<RpcMiddleware, HttpMiddleware>>,
562    keystore: Arc<RwLock<KeyStore>>,
563}
564
565/// A bare HTTP response carrying just `status` and an empty body.
566fn bare_http_response<B: Default>(status: http::StatusCode) -> http::Response<B> {
567    http::Response::builder()
568        .status(status)
569        .body(B::default())
570        .unwrap_or_else(|_| http::Response::new(B::default()))
571}
572
573pub async fn start_rpc(
574    state: RPCState,
575    rpc_listener: tokio::net::TcpListener,
576    stop_handle: StopHandle,
577    filter_list: Option<Arc<FilterList>>,
578    metrics_mode: MetricsMode,
579) -> anyhow::Result<()> {
580    let filter_list = filter_list.unwrap_or_default();
581    // `Arc` is needed because we will share the state between two modules
582    let state = Arc::new(state);
583    let keystore = state.keystore.shallow_clone();
584    let mut modules = create_modules(state.shallow_clone());
585
586    let mut pubsub_module = FilRpcModule::default();
587    pubsub_module.register_channel("Filecoin.ChainNotify", {
588        let state_clone = state.shallow_clone();
589        move |params| chain::chain_notify(params, &state_clone)
590    })?;
591
592    for module in modules.values_mut() {
593        // register eth subscription APIs
594        module.merge(EthPubSub::new(state.shallow_clone()).into_rpc())?;
595        module.merge(pubsub_module.clone())?;
596    }
597
598    let methods: Arc<HashMap<ApiPaths, Methods>> =
599        Arc::new(modules.into_iter().map(|(k, v)| (k, v.into())).collect());
600
601    let server_config = ServerConfig::builder()
602        .max_request_body_size(MAX_REQUEST_BODY_SIZE)
603        // Default size (10 MiB) is not enough for methods like `Filecoin.StateMinerActiveSectors`
604        .max_response_body_size(*MAX_RESPONSE_BODY_SIZE)
605        .max_connections(default_max_connections())
606        .set_id_provider(RandomHexStringIdProvider::new())
607        .build();
608    let max_response_body_size = *MAX_RESPONSE_BODY_SIZE as usize;
609    let per_conn = PerConnection {
610        stop_handle: stop_handle.clone(),
611        svc_builder: Server::builder()
612            .set_config(server_config)
613            .set_http_middleware(
614                tower::ServiceBuilder::new()
615                    .option_layer(COMPRESS_MIN_BODY_SIZE.map(CompressionLayer::new))
616                    // Mark the `Authorization` request header as sensitive so it doesn't show in logs
617                    .layer(SetSensitiveRequestHeadersLayer::new(std::iter::once(
618                        http::header::AUTHORIZATION,
619                    ))),
620            )
621            .to_service_builder()
622            .into(),
623        keystore,
624    };
625    tracing::info!("Ready for RPC connections");
626    loop {
627        let sock = tokio::select! {
628        res = rpc_listener.accept() => {
629            match res {
630              Ok((stream, _remote_addr)) => {
631                let _ = stream.set_nodelay(true); // Disable Nagle's algorithm
632                stream
633              }
634              Err(e) => {
635                tracing::error!("failed to accept v4 connection: {:?}", e);
636                continue;
637              }
638            }
639          }
640          _ = per_conn.stop_handle.clone().shutdown() => break,
641        };
642
643        let svc = tower::service_fn({
644            let methods = methods.shallow_clone();
645            let per_conn = per_conn.clone();
646            let filter_list = filter_list.shallow_clone();
647            move |req: http::Request<_>| {
648                let svc_or_result = if let Ok(path) = ApiPaths::from_uri(req.uri()) {
649                    let methods = methods.get(&path).cloned().unwrap_or_default();
650                    let PerConnection {
651                        stop_handle,
652                        svc_builder,
653                        keystore,
654                    } = per_conn.clone();
655                    // Authenticate the connection once, here at the HTTP layer (for a
656                    // WebSocket this is the upgrade request), before any JSON-RPC
657                    // dispatch.
658                    match resolve_claims(&keystore, req.headers().get(http::header::AUTHORIZATION))
659                    {
660                        Ok(claims) => {
661                            // NOTE, the rpc middleware must be initialized here to be able to be created once per connection
662                            // with data from the connection such as the headers in this example
663                            let rpc_middleware = RpcServiceBuilder::new()
664                                .layer(SetExtensionLayer { path })
665                                .layer(SegregationLayer)
666                                .layer(FilterLayer::new(filter_list.shallow_clone()))
667                                .layer(validation_layer::JsonValidationLayer)
668                                .layer(AuthLayer::new(claims))
669                                .layer(LogLayer::default())
670                                // `ParallelBatchLayer` fans a batch out into per-entry `call`s, so it must be
671                                // outer to `MetricsLayer` for batched methods to be measured. Both must stay
672                                // inner to the batch-transforming layers above.
673                                .layer(ParallelBatchLayer::new(max_response_body_size))
674                                .layer(MetricsLayer::new(metrics_mode));
675                            Either::Left(
676                                Arc::unwrap_or_clone(svc_builder)
677                                    .set_rpc_middleware(rpc_middleware)
678                                    .build(methods, stop_handle),
679                            )
680                        }
681                        Err(reason) => {
682                            tracing::debug!("rejecting unauthorized request: {reason}");
683                            Either::Right(Ok(bare_http_response(http::StatusCode::UNAUTHORIZED)))
684                        }
685                    }
686                } else {
687                    Either::Right(Ok(bare_http_response(http::StatusCode::NOT_FOUND)))
688                };
689                async move {
690                    match svc_or_result {
691                        Either::Left(mut svc) => {
692                            // https://github.com/rust-lang/rust/issues/102211 the error type can't be inferred
693                            // to be `Box<dyn std::error::Error + Send + Sync>` so we need to convert it to a concrete type
694                            // as workaround.
695                            svc.call(req).await.map_err(|e| anyhow::anyhow!("{:?}", e))
696                        }
697                        Either::Right(result) => result,
698                    }
699                }
700            }
701        });
702
703        tokio::spawn(jsonrpsee::server::serve_with_graceful_shutdown(
704            sock,
705            svc,
706            stop_handle.clone().shutdown(),
707        ));
708    }
709
710    Ok(())
711}
712
713fn create_modules(state: Arc<RPCState>) -> HashMap<ApiPaths, RpcModule<RPCState>> {
714    let mut modules = HashMap::default();
715    for api_version in ApiPaths::value_variants() {
716        modules.insert(*api_version, RpcModule::from_arc(state.shallow_clone()));
717    }
718    macro_rules! register {
719        ($ty:ty) => {
720            // Register only non-subscription RPC methods.
721            // Subscription methods are registered separately in the RPC module.
722            if !<$ty>::SUBSCRIPTION {
723                <$ty>::register(&mut modules, ParamStructure::ByPosition).unwrap();
724            }
725        };
726    }
727    for_each_rpc_method!(register);
728    modules
729}
730
731/// If `include` is not [`None`], only methods that are listed will be returned
732pub fn openrpc(path: ApiPaths, include: Option<&[&str]>) -> openrpc_types::OpenRPC {
733    use schemars::generate::{SchemaGenerator, SchemaSettings};
734
735    let mut methods = vec![];
736    // spec says draft07
737    let mut settings = SchemaSettings::draft07();
738    // ..but uses `components`
739    settings.definitions_path = "#/components/schemas/".into();
740    let mut generator = SchemaGenerator::new(settings);
741    macro_rules! callback {
742        ($ty:ty) => {
743            if <$ty>::API_PATHS.contains(path) {
744                match include {
745                    Some(include) => match include.contains(&<$ty>::NAME) {
746                        true => {
747                            methods.push(openrpc_types::ReferenceOr::Item(<$ty>::openrpc(
748                                &mut generator,
749                                ParamStructure::ByPosition,
750                                &<$ty>::NAME,
751                            )));
752                            if let Some(alias) = &<$ty>::NAME_ALIAS {
753                                methods.push(openrpc_types::ReferenceOr::Item(<$ty>::openrpc(
754                                    &mut generator,
755                                    ParamStructure::ByPosition,
756                                    &alias,
757                                )));
758                            }
759                        }
760                        false => {}
761                    },
762                    None => {
763                        methods.push(openrpc_types::ReferenceOr::Item(<$ty>::openrpc(
764                            &mut generator,
765                            ParamStructure::ByPosition,
766                            &<$ty>::NAME,
767                        )));
768                        if let Some(alias) = &<$ty>::NAME_ALIAS {
769                            methods.push(openrpc_types::ReferenceOr::Item(<$ty>::openrpc(
770                                &mut generator,
771                                ParamStructure::ByPosition,
772                                &alias,
773                            )));
774                        }
775                    }
776                }
777            }
778        };
779    }
780    for_each_rpc_method!(callback);
781    openrpc_types::OpenRPC {
782        methods,
783        components: Some(openrpc_types::Components {
784            schemas: Some(
785                generator
786                    .take_definitions(false)
787                    .into_iter()
788                    .filter_map(|(k, v)| {
789                        if let Ok(v) = Schema::try_from(v) {
790                            Some((k, v))
791                        } else {
792                            None
793                        }
794                    })
795                    .collect(),
796            ),
797            ..Default::default()
798        }),
799        openrpc: openrpc_types::OPEN_RPC_SPECIFICATION_VERSION,
800        info: openrpc_types::Info {
801            title: String::from("forest"),
802            version: env!("CARGO_PKG_VERSION").into(),
803            ..Default::default()
804        },
805        ..Default::default()
806    }
807}
808
809#[cfg(test)]
810mod tests {
811    use super::*;
812    use crate::{
813        db::MemoryDB,
814        networks::NetworkChain,
815        rpc::{client::UrlClient, common::ShiftingVersion},
816        tool::offline_server::server::offline_rpc_state,
817    };
818    use jsonrpsee::{
819        core::{
820            client::{BatchResponse, ClientT},
821            params::BatchRequestBuilder,
822        },
823        server::stop_channel,
824    };
825    use std::net::{Ipv4Addr, SocketAddr};
826    use tokio::task::JoinSet;
827
828    // To update RPC specs:
829    // `cargo test --lib -- rpc::tests::openrpc`
830    // `cargo insta review`
831
832    #[test]
833    fn openrpc_v0() {
834        openrpc(ApiPaths::V0);
835    }
836
837    #[test]
838    fn openrpc_v1() {
839        openrpc(ApiPaths::V1);
840    }
841
842    #[test]
843    fn openrpc_v2() {
844        openrpc(ApiPaths::V2);
845    }
846
847    fn openrpc(path: ApiPaths) {
848        let spec = super::openrpc(path, None);
849        insta::assert_yaml_snapshot!(path.path(), spec);
850    }
851
852    #[test]
853    fn openrpc_casing() {
854        let violations = ApiPaths::value_variants()
855            .iter()
856            .flat_map(|path| check_openrpc_casing(*path))
857            .collect_vec();
858        assert!(
859            violations.is_empty(),
860            "OpenRPC casing violations ({}):\n{}",
861            violations.len(),
862            violations.join("\n")
863        );
864    }
865
866    fn check_openrpc_casing(path: ApiPaths) -> Vec<String> {
867        casing_violations(
868            path.path(),
869            &serde_json::to_value(super::openrpc(path, None)).unwrap(),
870        )
871    }
872
873    /// Each rule of [`casing_violations`] must flag a planted offender, and
874    /// each exemption (shared schema, `oneOf` variant tag, CID `/` key) must
875    /// hold — guarding against the checker regressing into a silent pass.
876    #[test]
877    fn openrpc_casing_detects_violations() {
878        let doc = serde_json::json!({
879            "methods": [
880                {"name": "Filecoin.Good", "params": [{"name": "goodParam"}],
881                 "result": {"schema": {"$ref": "#/components/schemas/Shared"}}},
882                {"name": "eth_getThing", "params": []},
883                {"name": "BadNoNamespace", "params": []},
884                {"name": "Forest.Bad_Method", "params": []},
885                {"name": "Forest.Thing",
886                 "params": [{"name": "bad_param",
887                             "schema": {"$ref": "#/components/schemas/Shared"}}],
888                 "result": {"schema": {"$ref": "#/components/schemas/ForestOnly"}}},
889            ],
890            "components": {"schemas": {
891                // reachable from Forest.Thing AND Filecoin.Good: the non-Forest
892                // reference must win, exempting PascalOk from the camelCase rule
893                "Shared": {"type": "object",
894                           "properties": {"PascalOk": true, "snake_bad": true}},
895                "ForestOnly": {"type": "object",
896                               "properties": {"camelOk": true, "PascalBad": true,
897                                              "nested": {"$ref": "#/components/schemas/ForestNested"}}},
898                // reachable only transitively, via ForestOnly
899                "ForestNested": {"oneOf": [
900                    {"type": "object", "properties": {"VariantTag": true}},
901                    {"type": "object",
902                     "properties": {"AlsoPascal": true, "ok": true, "/": true}},
903                ]},
904            }},
905        });
906        let violations = casing_violations("test", &doc)
907            .into_iter()
908            .sorted()
909            .collect_vec();
910        assert_eq!(
911            violations,
912            [
913                "test: BadNoNamespace: bad method name",
914                "test: Forest.Bad_Method: bad method name",
915                "test: Forest.Thing: param `bad_param`",
916                "test: schema `ForestNested`: property `AlsoPascal` is not lowerCamelCase",
917                "test: schema `ForestOnly`: property `PascalBad` is not lowerCamelCase",
918                "test: schema `Shared`: property `snake_bad` contains `_`",
919            ]
920        );
921    }
922
923    /// Enforces the JSON-RPC casing conventions on an OpenRPC document
924    fn casing_violations(label: &str, doc: &serde_json::Value) -> Vec<String> {
925        use serde_json::Value;
926        use std::collections::{BTreeMap, BTreeSet};
927
928        let mut violations = vec![];
929
930        fn is_lower_camel(s: &str) -> bool {
931            s.starts_with(|c: char| c.is_ascii_lowercase())
932                && s.chars().all(|c| c.is_ascii_alphanumeric())
933        }
934
935        fn is_pascal(s: &str) -> bool {
936            s.starts_with(|c: char| c.is_ascii_uppercase())
937                && s.chars().all(|c| c.is_ascii_alphanumeric())
938        }
939
940        fn method_name_ok(name: &str) -> bool {
941            if let Some((ns, rest)) = name.split_once('.') {
942                matches!(ns, "Filecoin" | "Forest" | "F3") && is_pascal(rest)
943            } else if let Some((ns, rest)) = name.split_once('_') {
944                matches!(ns, "eth" | "net" | "web3" | "trace" | "debug") && is_lower_camel(rest)
945            } else {
946                false
947            }
948        }
949
950        fn collect_refs(node: &Value, out: &mut BTreeSet<String>) {
951            match node {
952                Value::Object(map) => {
953                    if let Some(name) = map
954                        .get("$ref")
955                        .and_then(Value::as_str)
956                        .and_then(|r| r.strip_prefix("#/components/schemas/"))
957                    {
958                        out.insert(name.to_string());
959                    }
960                    map.values().for_each(|v| collect_refs(v, out));
961                }
962                Value::Array(it) => it.iter().for_each(|v| collect_refs(v, out)),
963                _ => {}
964            }
965        }
966
967        fn check_props(
968            node: &Value,
969            ctx: &str,
970            in_one_of: bool,
971            require_camel: bool,
972            violations: &mut Vec<String>,
973        ) {
974            match node {
975                Value::Object(map) => {
976                    if let Some(props) = map.get("properties").and_then(Value::as_object) {
977                        let variant_tag = in_one_of && props.len() == 1;
978                        for key in props.keys() {
979                            if key == "/" {
980                                continue;
981                            }
982                            if key.contains('_') {
983                                violations.push(format!("{ctx}: property `{key}` contains `_`"));
984                            } else if require_camel && !variant_tag && !is_lower_camel(key) {
985                                violations
986                                    .push(format!("{ctx}: property `{key}` is not lowerCamelCase"));
987                            }
988                        }
989                    }
990                    for (key, value) in map {
991                        check_props(value, ctx, key == "oneOf", require_camel, violations);
992                    }
993                }
994                Value::Array(it) => it
995                    .iter()
996                    .for_each(|v| check_props(v, ctx, in_one_of, require_camel, violations)),
997                _ => {}
998            }
999        }
1000
1001        let methods = doc
1002            .get("methods")
1003            .and_then(Value::as_array)
1004            .expect("openrpc doc has methods");
1005        let schemas = doc
1006            .pointer("/components/schemas")
1007            .and_then(Value::as_object)
1008            .expect("openrpc doc has component schemas");
1009
1010        let direct_refs: BTreeMap<String, BTreeSet<String>> = schemas
1011            .iter()
1012            .map(|(name, schema)| {
1013                let mut refs = BTreeSet::new();
1014                collect_refs(schema, &mut refs);
1015                (name.clone(), refs)
1016            })
1017            .collect();
1018        let transitive = |seed: BTreeSet<String>| {
1019            let mut seen = BTreeSet::new();
1020            let mut stack = Vec::from_iter(seed);
1021            while let Some(name) = stack.pop() {
1022                if let Some(refs) = direct_refs.get(&name)
1023                    && seen.insert(name)
1024                {
1025                    stack.extend(refs.iter().cloned());
1026                }
1027            }
1028            seen
1029        };
1030
1031        let mut forest_seed = BTreeSet::new();
1032        let mut other_seed = BTreeSet::new();
1033        for method in methods {
1034            let name = method
1035                .get("name")
1036                .and_then(Value::as_str)
1037                .expect("method has a name");
1038            let ctx = format!("{label}: {name}");
1039            if !method_name_ok(name) {
1040                violations.push(format!("{ctx}: bad method name"));
1041            }
1042            if let Some(params) = method.get("params").and_then(Value::as_array) {
1043                for param in params {
1044                    let pname = param
1045                        .get("name")
1046                        .and_then(Value::as_str)
1047                        .expect("param has a name");
1048                    if !is_lower_camel(pname) {
1049                        violations.push(format!("{ctx}: param `{pname}`"));
1050                    }
1051                }
1052            }
1053            let forest_owned = name.starts_with("Forest.");
1054            check_props(method, &ctx, false, forest_owned, &mut violations);
1055
1056            collect_refs(
1057                method,
1058                if forest_owned {
1059                    &mut forest_seed
1060                } else {
1061                    &mut other_seed
1062                },
1063            );
1064        }
1065        let forest_reachable = transitive(forest_seed);
1066        let other_reachable = transitive(other_seed);
1067
1068        for (name, schema) in schemas {
1069            let ctx = format!("{label}: schema `{name}`");
1070            let forest_only = forest_reachable.contains(name) && !other_reachable.contains(name);
1071            check_props(schema, &ctx, false, forest_only, &mut violations);
1072        }
1073        violations
1074    }
1075
1076    #[test]
1077    fn test_rpc_server() {
1078        const TIMEOUT: Duration = Duration::from_secs(5);
1079        let (done_tx, done_rx) = flume::bounded(1);
1080        let rt = tokio::runtime::Builder::new_multi_thread()
1081            .enable_all()
1082            .build()
1083            .unwrap();
1084        rt.block_on(async move { test_rpc_server_inner(done_tx).await });
1085        done_rx.recv().unwrap();
1086        // To mitigate the transient timeout issue
1087        rt.shutdown_timeout(TIMEOUT);
1088    }
1089
1090    async fn test_rpc_server_inner(done_tx: flume::Sender<()>) {
1091        let chain = NetworkChain::Calibnet;
1092        let db = Arc::new(MemoryDB::default());
1093        let mut services = JoinSet::new();
1094        let (state, mut shutdown_recv) = offline_rpc_state(chain, db, None, None, &mut services)
1095            .await
1096            .unwrap();
1097        let block_delay_secs = state.chain_config().block_delay_secs;
1098        let shutdown_send = state.shutdown.clone();
1099        let jwt_read_permissions = vec!["read".to_owned()];
1100        let jwt_read = super::methods::auth::AuthNew::create_token(
1101            &state.keystore.read(),
1102            chrono::Duration::hours(1),
1103            jwt_read_permissions.clone(),
1104        )
1105        .unwrap();
1106        let rpc_listener =
1107            tokio::net::TcpListener::bind(SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 0))
1108                .await
1109                .unwrap();
1110        let rpc_address = rpc_listener.local_addr().unwrap();
1111        let (stop_handle, server_handle) = stop_channel();
1112
1113        // Start an RPC server
1114
1115        let handle = tokio::spawn(start_rpc(
1116            state,
1117            rpc_listener,
1118            stop_handle,
1119            None,
1120            MetricsMode::Enabled,
1121        ));
1122
1123        println!("sending a few http requests");
1124
1125        let client = Client::from_url(
1126            format!("http://{}:{}/", rpc_address.ip(), rpc_address.port())
1127                .parse()
1128                .unwrap(),
1129        );
1130
1131        let response = super::methods::common::Version::call(&client, ())
1132            .await
1133            .unwrap();
1134        assert_eq!(
1135            &response.version,
1136            &*crate::utils::version::FOREST_VERSION_STRING
1137        );
1138        assert_eq!(response.block_delay, block_delay_secs);
1139        assert_eq!(response.api_version, ShiftingVersion::new(2, 3, 0));
1140
1141        let response = super::methods::auth::AuthVerify::call(&client, (jwt_read.clone(),))
1142            .await
1143            .unwrap();
1144        assert_eq!(response, jwt_read_permissions);
1145
1146        // `AuthVerify` verifies a raw JWT; a `Bearer `-prefixed argument must fail.
1147        super::methods::auth::AuthVerify::call(&client, (format!("Bearer {jwt_read}"),))
1148            .await
1149            .unwrap_err();
1150
1151        drop(client);
1152
1153        // A bad token is rejected with a bare HTTP 401, before JSON-RPC dispatch.
1154        let http = reqwest::Client::new();
1155        let rpc_url = format!("http://{}:{}/rpc/v1", rpc_address.ip(), rpc_address.port());
1156        let jsonrpc_body = serde_json::json!({
1157            "jsonrpc": "2.0",
1158            "method": "Filecoin.Version",
1159            "params": [],
1160            "id": 0,
1161        });
1162
1163        let bad_token = http
1164            .post(&rpc_url)
1165            .header(reqwest::header::AUTHORIZATION, "Bearer not-a-real-token")
1166            .json(&jsonrpc_body)
1167            .send()
1168            .await
1169            .unwrap();
1170        assert_eq!(bad_token.status(), reqwest::StatusCode::UNAUTHORIZED);
1171
1172        // A valid token is not rejected at the HTTP layer.
1173        let good_token = http
1174            .post(&rpc_url)
1175            .header(reqwest::header::AUTHORIZATION, format!("Bearer {jwt_read}"))
1176            .json(&jsonrpc_body)
1177            .send()
1178            .await
1179            .unwrap();
1180        assert_eq!(good_token.status(), reqwest::StatusCode::OK);
1181
1182        println!("sending a few websocket requests");
1183
1184        let client = Client::from_url(
1185            format!("ws://{}:{}/", rpc_address.ip(), rpc_address.port())
1186                .parse()
1187                .unwrap(),
1188        );
1189
1190        let response = super::methods::auth::AuthVerify::call(&client, (jwt_read,))
1191            .await
1192            .unwrap();
1193        assert_eq!(response, jwt_read_permissions);
1194
1195        drop(client);
1196
1197        // Sending a batch request
1198        let client = UrlClient::new(
1199            format!("http://{}:{}/rpc/v1", rpc_address.ip(), rpc_address.port())
1200                .parse()
1201                .unwrap(),
1202            None,
1203        )
1204        .await
1205        .unwrap();
1206        let mut batch_request_builder = BatchRequestBuilder::new();
1207        let empty_payload: [(); 0] = [];
1208        batch_request_builder
1209            .insert("Filecoin.Version", empty_payload)
1210            .unwrap();
1211        batch_request_builder
1212            .insert("eth_chainId", empty_payload)
1213            .unwrap();
1214        let batch_response: BatchResponse<serde_json::Value> =
1215            client.batch_request(batch_request_builder).await.unwrap();
1216        assert_eq!(batch_response.len(), 2);
1217        assert_eq!(batch_response.num_successful_calls(), 2);
1218        assert_eq!(batch_response.num_failed_calls(), 0);
1219
1220        // `eth_chainId` is only ever requested inside the batch above, so its presence in the RPC
1221        // timing metric proves batched methods flow through `MetricsLayer`. Guards against batch
1222        // entries bypassing metrics (which happens if `MetricsLayer` is outer to `ParallelBatchLayer`).
1223        let mut encoded = String::new();
1224        prometheus_client::encoding::text::encode_registry(
1225            &mut encoded,
1226            &crate::metrics::default_registry(),
1227        )
1228        .unwrap();
1229        let recorded = encoded.lines().any(|line| {
1230            line.starts_with("rpc_processing_time_count{")
1231                && line.contains(r#"method="eth_chainId""#)
1232                && line
1233                    .rsplit(' ')
1234                    .next()
1235                    .and_then(|v| v.parse::<u64>().ok())
1236                    .is_some_and(|count| count == 1)
1237        });
1238        assert!(
1239            recorded,
1240            "batched method `eth_chainId` was not recorded in rpc_processing_time:\n{encoded}"
1241        );
1242
1243        // Gracefully shutdown the RPC server
1244        println!("sending shutdown signal");
1245        shutdown_send.send(()).await.unwrap();
1246        println!("waiting on shutdown receiver");
1247        shutdown_recv.recv().await;
1248        println!("sending server stop signal");
1249        server_handle.stop().unwrap();
1250        println!("waiting on graceful shutdown");
1251        handle.await.unwrap().unwrap();
1252        println!("done");
1253        done_tx.send(()).unwrap();
1254    }
1255}
forest/rpc/mod.rs

forest/rpc/
mod.rs
