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

forest/rpc/
mod.rs
