reth_ipc/server/
mod.rs

1//! JSON-RPC IPC server implementation
2
3use crate::server::connection::{IpcConn, JsonRpcStream};
4use futures::StreamExt;
5use futures_util::future::Either;
6use interprocess::local_socket::{
7    tokio::prelude::{LocalSocketListener, LocalSocketStream},
8    traits::tokio::{Listener, Stream},
9    GenericFilePath, ListenerOptions, ToFsName,
10};
11use jsonrpsee::{
12    core::{
13        middleware::layer::{Either as RpcEither, RpcLoggerLayer},
14        JsonRawValue, TEN_MB_SIZE_BYTES,
15    },
16    server::{
17        middleware::rpc::RpcServiceT, stop_channel, ConnectionGuard, ConnectionPermit, IdProvider,
18        RandomIntegerIdProvider, ServerHandle, StopHandle,
19    },
20    BoundedSubscriptions, MethodResponse, MethodSink, Methods,
21};
22use std::{
23    future::Future,
24    io,
25    pin::{pin, Pin},
26    sync::Arc,
27    task::{Context, Poll},
28};
29use tokio::{
30    io::{AsyncRead, AsyncWrite, AsyncWriteExt},
31    sync::oneshot,
32};
33use tower::{layer::util::Identity, Layer, Service};
34use tracing::{debug, instrument, trace, warn, Instrument};
35// re-export so can be used during builder setup
36use crate::{
37    server::{connection::IpcConnDriver, rpc_service::RpcServiceCfg},
38    stream_codec::StreamCodec,
39};
40use tokio::sync::mpsc;
41use tokio_stream::wrappers::ReceiverStream;
42use tokio_util::task::AbortOnDropHandle;
43use tower::layer::{util::Stack, LayerFn};
44
45mod connection;
46mod ipc;
47mod rpc_service;
48
49pub use rpc_service::RpcService;
50
51/// Ipc Server implementation
52///
53/// This is an adapted `jsonrpsee` Server, but for `Ipc` connections.
54pub struct IpcServer<HttpMiddleware = Identity, RpcMiddleware = Identity> {
55    /// The endpoint we listen for incoming transactions
56    endpoint: String,
57    id_provider: Arc<dyn IdProvider>,
58    cfg: Settings,
59    rpc_middleware: RpcServiceBuilder<RpcMiddleware>,
60    http_middleware: tower::ServiceBuilder<HttpMiddleware>,
61}
62
63impl<HttpMiddleware, RpcMiddleware> IpcServer<HttpMiddleware, RpcMiddleware> {
64    /// Returns the configured endpoint
65    pub fn endpoint(&self) -> String {
66        self.endpoint.clone()
67    }
68}
69
70impl<HttpMiddleware, RpcMiddleware> IpcServer<HttpMiddleware, RpcMiddleware>
71where
72    RpcMiddleware: Layer<RpcService, Service: RpcServiceT> + Clone + Send + 'static,
73    HttpMiddleware: Layer<
74            TowerServiceNoHttp<RpcMiddleware>,
75            Service: Service<
76                String,
77                Response = Option<String>,
78                Error = Box<dyn core::error::Error + Send + Sync + 'static>,
79                Future: Send + Unpin,
80            > + Send,
81        > + Send
82        + 'static,
83{
84    /// Start responding to connections requests.
85    ///
86    /// This will run on the tokio runtime until the server is stopped or the `ServerHandle` is
87    /// dropped.
88    ///
89    /// ```
90    /// use jsonrpsee::RpcModule;
91    /// use reth_ipc::server::Builder;
92    /// async fn run_server() -> Result<(), Box<dyn core::error::Error + Send + Sync>> {
93    ///     let server = Builder::default().build("/tmp/my-uds".into());
94    ///     let mut module = RpcModule::new(());
95    ///     module.register_method("say_hello", |_, _, _| "lo")?;
96    ///     let handle = server.start(module).await?;
97    ///
98    ///     // In this example we don't care about doing shutdown so let's it run forever.
99    ///     // You may use the `ServerHandle` to shut it down or manage it yourself.
100    ///     let server = tokio::spawn(handle.stopped());
101    ///     server.await.unwrap();
102    ///     Ok(())
103    /// }
104    /// ```
105    pub async fn start(
106        mut self,
107        methods: impl Into<Methods>,
108    ) -> Result<ServerHandle, IpcServerStartError> {
109        let methods = methods.into();
110
111        let (stop_handle, server_handle) = stop_channel();
112
113        // use a signal channel to wait until we're ready to accept connections
114        let (tx, rx) = oneshot::channel();
115
116        match self.cfg.tokio_runtime.take() {
117            Some(rt) => rt.spawn(self.start_inner(methods, stop_handle, tx)),
118            None => tokio::spawn(self.start_inner(methods, stop_handle, tx)),
119        };
120        rx.await.expect("channel is open")?;
121
122        Ok(server_handle)
123    }
124
125    async fn start_inner(
126        self,
127        methods: Methods,
128        stop_handle: StopHandle,
129        on_ready: oneshot::Sender<Result<(), IpcServerStartError>>,
130    ) {
131        trace!(endpoint = ?self.endpoint, "starting ipc server");
132
133        if cfg!(unix) {
134            // ensure the file does not exist
135            if std::fs::remove_file(&self.endpoint).is_ok() {
136                debug!(endpoint = ?self.endpoint, "removed existing IPC endpoint file");
137            }
138        }
139
140        let listener = match self
141            .endpoint
142            .as_str()
143            .to_fs_name::<GenericFilePath>()
144            .and_then(|name| ListenerOptions::new().name(name).create_tokio())
145        {
146            Ok(listener) => {
147                #[cfg(unix)]
148                {
149                    // set permissions only on unix
150                    use std::os::unix::fs::PermissionsExt;
151                    if let Some(perms_str) = &self.cfg.ipc_socket_permissions &&
152                        let Ok(mode) = u32::from_str_radix(&perms_str.replace("0o", ""), 8)
153                    {
154                        let perms = std::fs::Permissions::from_mode(mode);
155                        let _ = std::fs::set_permissions(&self.endpoint, perms);
156                    }
157                }
158                listener
159            }
160            Err(err) => {
161                on_ready
162                    .send(Err(IpcServerStartError { endpoint: self.endpoint.clone(), source: err }))
163                    .ok();
164                return;
165            }
166        };
167
168        // signal that we're ready to accept connections
169        on_ready.send(Ok(())).ok();
170
171        let mut id: u32 = 0;
172        let connection_guard = ConnectionGuard::new(self.cfg.max_connections as usize);
173
174        let stopped = stop_handle.clone().shutdown();
175        let mut stopped = pin!(stopped);
176
177        let (drop_on_completion, mut process_connection_awaiter) = mpsc::channel::<()>(1);
178
179        trace!("accepting ipc connections");
180        loop {
181            match try_accept_conn(&listener, stopped).await {
182                AcceptConnection::Established { local_socket_stream, stop } => {
183                    let Some(conn_permit) = connection_guard.try_acquire() else {
184                        let (_reader, mut writer) = local_socket_stream.split();
185                        let _ = writer
186                            .write_all(b"Too many connections. Please try again later.")
187                            .await;
188                        stopped = stop;
189                        continue;
190                    };
191
192                    let max_conns = connection_guard.max_connections();
193                    let curr_conns = max_conns - connection_guard.available_connections();
194                    trace!("Accepting new connection {}/{}", curr_conns, max_conns);
195
196                    let conn_permit = Arc::new(conn_permit);
197
198                    process_connection(ProcessConnection {
199                        http_middleware: &self.http_middleware,
200                        rpc_middleware: self.rpc_middleware.clone(),
201                        conn_permit,
202                        conn_id: id,
203                        server_cfg: self.cfg.clone(),
204                        stop_handle: stop_handle.clone(),
205                        drop_on_completion: drop_on_completion.clone(),
206                        methods: methods.clone(),
207                        id_provider: self.id_provider.clone(),
208                        local_socket_stream,
209                    });
210
211                    id = id.wrapping_add(1);
212                    stopped = stop;
213                }
214                AcceptConnection::Shutdown => {
215                    break;
216                }
217                AcceptConnection::Err((err, stop)) => {
218                    tracing::error!(%err, "Failed accepting a new IPC connection");
219                    stopped = stop;
220                }
221            }
222        }
223
224        // Drop the last Sender
225        drop(drop_on_completion);
226
227        // Once this channel is closed it is safe to assume that all connections have been
228        // gracefully shutdown
229        while process_connection_awaiter.recv().await.is_some() {
230            // Generally, messages should not be sent across this channel,
231            // but we'll loop here to wait for `None` just to be on the safe side
232        }
233    }
234}
235
236enum AcceptConnection<S> {
237    Shutdown,
238    Established { local_socket_stream: LocalSocketStream, stop: S },
239    Err((io::Error, S)),
240}
241
242async fn try_accept_conn<S>(listener: &LocalSocketListener, stopped: S) -> AcceptConnection<S>
243where
244    S: Future + Unpin,
245{
246    match futures_util::future::select(pin!(listener.accept()), stopped).await {
247        Either::Left((res, stop)) => match res {
248            Ok(local_socket_stream) => AcceptConnection::Established { local_socket_stream, stop },
249            Err(e) => AcceptConnection::Err((e, stop)),
250        },
251        Either::Right(_) => AcceptConnection::Shutdown,
252    }
253}
254
255impl<HttpMiddleware, RpcMiddleware> std::fmt::Debug for IpcServer<HttpMiddleware, RpcMiddleware> {
256    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
257        f.debug_struct("IpcServer")
258            .field("endpoint", &self.endpoint)
259            .field("cfg", &self.cfg)
260            .field("id_provider", &self.id_provider)
261            .finish()
262    }
263}
264
265/// Error thrown when server couldn't be started.
266#[derive(Debug, thiserror::Error)]
267#[error("failed to listen on ipc endpoint `{endpoint}`: {source}")]
268pub struct IpcServerStartError {
269    endpoint: String,
270    #[source]
271    source: io::Error,
272}
273
274/// Data required by the server to handle requests received via an IPC connection
275#[derive(Debug, Clone)]
276#[allow(dead_code)]
277pub(crate) struct ServiceData {
278    /// Registered server methods.
279    pub(crate) methods: Methods,
280    /// Subscription ID provider.
281    pub(crate) id_provider: Arc<dyn IdProvider>,
282    /// Stop handle.
283    pub(crate) stop_handle: StopHandle,
284    /// Connection ID
285    pub(crate) conn_id: u32,
286    /// Connection Permit.
287    pub(crate) conn_permit: Arc<ConnectionPermit>,
288    /// Limits the number of subscriptions for this connection
289    pub(crate) bounded_subscriptions: BoundedSubscriptions,
290    /// Sink that is used to send back responses to the connection.
291    ///
292    /// This is used for subscriptions.
293    pub(crate) method_sink: MethodSink,
294    /// `ServerConfig`
295    pub(crate) server_cfg: Settings,
296}
297
298/// Similar to [`tower::ServiceBuilder`] but doesn't
299/// support any tower middleware implementations.
300#[derive(Debug, Clone)]
301pub struct RpcServiceBuilder<L>(tower::ServiceBuilder<L>);
302
303impl Default for RpcServiceBuilder<Identity> {
304    fn default() -> Self {
305        Self(tower::ServiceBuilder::new())
306    }
307}
308
309impl RpcServiceBuilder<Identity> {
310    /// Create a new [`RpcServiceBuilder`].
311    pub const fn new() -> Self {
312        Self(tower::ServiceBuilder::new())
313    }
314}
315
316impl<L> RpcServiceBuilder<L> {
317    /// Optionally add a new layer `T` to the [`RpcServiceBuilder`].
318    ///
319    /// See the documentation for [`tower::ServiceBuilder::option_layer`] for more details.
320    pub fn option_layer<T>(
321        self,
322        layer: Option<T>,
323    ) -> RpcServiceBuilder<Stack<RpcEither<T, Identity>, L>> {
324        let layer = if let Some(layer) = layer {
325            RpcEither::Left(layer)
326        } else {
327            RpcEither::Right(Identity::new())
328        };
329        self.layer(layer)
330    }
331
332    /// Add a new layer `T` to the [`RpcServiceBuilder`].
333    ///
334    /// See the documentation for [`tower::ServiceBuilder::layer`] for more details.
335    pub fn layer<T>(self, layer: T) -> RpcServiceBuilder<Stack<T, L>> {
336        RpcServiceBuilder(self.0.layer(layer))
337    }
338
339    /// Add a [`tower::Layer`] built from a function that accepts a service and returns another
340    /// service.
341    ///
342    /// See the documentation for [`tower::ServiceBuilder::layer_fn`] for more details.
343    pub fn layer_fn<F>(self, f: F) -> RpcServiceBuilder<Stack<LayerFn<F>, L>> {
344        RpcServiceBuilder(self.0.layer_fn(f))
345    }
346
347    /// Add a logging layer to [`RpcServiceBuilder`]
348    ///
349    /// This logs each request and response for every call.
350    pub fn rpc_logger(self, max_log_len: u32) -> RpcServiceBuilder<Stack<RpcLoggerLayer, L>> {
351        RpcServiceBuilder(self.0.layer(RpcLoggerLayer::new(max_log_len)))
352    }
353
354    /// Wrap the service `S` with the middleware.
355    pub(crate) fn service<S>(&self, service: S) -> L::Service
356    where
357        L: tower::Layer<S>,
358    {
359        self.0.service(service)
360    }
361}
362
363/// `JsonRPSee` service compatible with `tower`.
364///
365/// # Note
366/// This is similar to [`hyper::service::service_fn`](https://docs.rs/hyper/latest/hyper/service/fn.service_fn.html).
367#[derive(Debug, Clone)]
368pub struct TowerServiceNoHttp<L> {
369    inner: ServiceData,
370    rpc_middleware: RpcServiceBuilder<L>,
371}
372
373impl<RpcMiddleware> Service<String> for TowerServiceNoHttp<RpcMiddleware>
374where
375    RpcMiddleware: Layer<RpcService>,
376    <RpcMiddleware as Layer<RpcService>>::Service:
377        Send + Sync + 'static + RpcServiceT<MethodResponse = MethodResponse>,
378{
379    /// The response of a handled RPC call
380    ///
381    /// This is an `Option` because subscriptions and call responses are handled differently.
382    /// This will be `Some` for calls, and `None` for subscriptions, because the subscription
383    /// response will be emitted via the `method_sink`.
384    type Response = Option<String>;
385
386    type Error = Box<dyn core::error::Error + Send + Sync + 'static>;
387
388    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
389
390    /// Opens door for back pressure implementation.
391    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
392        Poll::Ready(Ok(()))
393    }
394
395    fn call(&mut self, request: String) -> Self::Future {
396        trace!("{:?}", request);
397
398        let cfg = RpcServiceCfg {
399            bounded_subscriptions: self.inner.bounded_subscriptions.clone(),
400            id_provider: self.inner.id_provider.clone(),
401            sink: self.inner.method_sink.clone(),
402        };
403
404        let max_response_body_size = self.inner.server_cfg.max_response_body_size as usize;
405        let max_request_body_size = self.inner.server_cfg.max_request_body_size as usize;
406        let conn = self.inner.conn_permit.clone();
407        let rpc_service = self.rpc_middleware.service(RpcService::new(
408            self.inner.methods.clone(),
409            max_response_body_size,
410            self.inner.conn_id.into(),
411            cfg,
412        ));
413        // an ipc connection needs to handle read+write concurrently
414        // even if the underlying rpc handler spawns the actual work or is does a lot of async any
415        // additional overhead performed by `handle_request` can result in I/O latencies, for
416        // example tracing calls are relatively CPU expensive on serde::serialize alone, moving this
417        // work to a separate task takes the pressure off the connection so all concurrent responses
418        // are also serialized concurrently and the connection can focus on read+write
419        //
420        // The connection drops its pending calls when it closes, so the call must not outlive the
421        // returned future, otherwise it keeps running without anyone waiting for the response.
422        let f = AbortOnDropHandle::new(tokio::task::spawn(async move {
423            ipc::call_with_service(
424                request,
425                rpc_service,
426                max_response_body_size,
427                max_request_body_size,
428                conn,
429            )
430            .await
431        }));
432
433        Box::pin(async move {
434            // Call panics are answered by the call itself. Anything left here has no request id to
435            // respond to, and the connection writes errors verbatim, which would corrupt the
436            // stream.
437            Ok(f.await.unwrap_or_else(|err| {
438                warn!(%err, "IPC call task failed");
439                None
440            }))
441        })
442    }
443}
444
445struct ProcessConnection<'a, HttpMiddleware, RpcMiddleware> {
446    http_middleware: &'a tower::ServiceBuilder<HttpMiddleware>,
447    rpc_middleware: RpcServiceBuilder<RpcMiddleware>,
448    conn_permit: Arc<ConnectionPermit>,
449    conn_id: u32,
450    server_cfg: Settings,
451    stop_handle: StopHandle,
452    drop_on_completion: mpsc::Sender<()>,
453    methods: Methods,
454    id_provider: Arc<dyn IdProvider>,
455    local_socket_stream: LocalSocketStream,
456}
457
458/// Spawns the IPC connection onto a new task
459#[instrument(name = "connection", skip_all, fields(conn_id = %params.conn_id))]
460fn process_connection<RpcMiddleware, HttpMiddleware>(
461    params: ProcessConnection<'_, HttpMiddleware, RpcMiddleware>,
462) where
463    RpcMiddleware: Layer<RpcService> + Clone + Send + 'static,
464    for<'a> <RpcMiddleware as Layer<RpcService>>::Service: RpcServiceT,
465    HttpMiddleware: Layer<TowerServiceNoHttp<RpcMiddleware>> + Send + 'static,
466    <HttpMiddleware as Layer<TowerServiceNoHttp<RpcMiddleware>>>::Service: Send
467    + Service<
468        String,
469        Response = Option<String>,
470        Error = Box<dyn core::error::Error + Send + Sync + 'static>,
471    >,
472    <<HttpMiddleware as Layer<TowerServiceNoHttp<RpcMiddleware>>>::Service as Service<String>>::Future:
473    Send + Unpin,
474{
475    let ProcessConnection {
476        http_middleware,
477        rpc_middleware,
478        conn_permit,
479        conn_id,
480        server_cfg,
481        stop_handle,
482        drop_on_completion,
483        id_provider,
484        methods,
485        local_socket_stream,
486    } = params;
487
488    let ipc = IpcConn(tokio_util::codec::Decoder::framed(
489        StreamCodec::stream_incoming(),
490        local_socket_stream,
491    ));
492
493    let (tx, rx) = mpsc::channel::<Box<JsonRawValue>>(server_cfg.message_buffer_capacity as usize);
494    let method_sink = MethodSink::new_with_limit(tx, server_cfg.max_response_body_size);
495    let tower_service = TowerServiceNoHttp {
496        inner: ServiceData {
497            methods,
498            id_provider,
499            stop_handle: stop_handle.clone(),
500            server_cfg: server_cfg.clone(),
501            conn_id,
502            conn_permit,
503            bounded_subscriptions: BoundedSubscriptions::new(
504                server_cfg.max_subscriptions_per_connection,
505            ),
506            method_sink,
507        },
508        rpc_middleware,
509    };
510
511    let service = http_middleware.service(tower_service);
512    tokio::spawn(async {
513        to_ipc_service(ipc, service, stop_handle, rx).in_current_span().await;
514        drop(drop_on_completion)
515    });
516}
517
518async fn to_ipc_service<S, T>(
519    ipc: IpcConn<JsonRpcStream<T>>,
520    service: S,
521    stop_handle: StopHandle,
522    rx: mpsc::Receiver<Box<JsonRawValue>>,
523) where
524    S: Service<String, Response = Option<String>> + Send + 'static,
525    S::Error: Into<Box<dyn core::error::Error + Send + Sync>>,
526    S::Future: Send + Unpin,
527    T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
528{
529    let rx_item = ReceiverStream::new(rx);
530    let conn = IpcConnDriver {
531        conn: ipc,
532        service,
533        pending_calls: Default::default(),
534        items: Default::default(),
535    };
536    let stopped = stop_handle.shutdown();
537
538    let mut conn = pin!(conn);
539    let mut rx_item = pin!(rx_item);
540    let mut stopped = pin!(stopped);
541
542    loop {
543        tokio::select! {
544            _ = &mut conn => {
545               break
546            }
547            item = rx_item.next() => {
548                let Some(item) = item else { break };
549                conn.push_back(String::from(Box::<str>::from(item)));
550            }
551            _ = &mut stopped => {
552                // shutdown
553                break
554            }
555        }
556    }
557}
558
559/// JSON-RPC IPC server settings.
560#[derive(Debug, Clone)]
561pub struct Settings {
562    /// Maximum size in bytes of a request.
563    max_request_body_size: u32,
564    /// Maximum size in bytes of a response.
565    max_response_body_size: u32,
566    /// Max length for logging for requests and responses
567    ///
568    /// Logs bigger than this limit will be truncated.
569    max_log_length: u32,
570    /// Maximum number of incoming connections allowed.
571    max_connections: u32,
572    /// Maximum number of subscriptions per connection.
573    max_subscriptions_per_connection: u32,
574    /// Number of messages that server is allowed `buffer` until backpressure kicks in.
575    message_buffer_capacity: u32,
576    /// Custom tokio runtime to run the server on.
577    tokio_runtime: Option<tokio::runtime::Handle>,
578    /// The permissions to create the IPC socket with.
579    ipc_socket_permissions: Option<String>,
580}
581
582impl Default for Settings {
583    fn default() -> Self {
584        Self {
585            max_request_body_size: TEN_MB_SIZE_BYTES,
586            max_response_body_size: TEN_MB_SIZE_BYTES,
587            max_log_length: 4096,
588            max_connections: 100,
589            max_subscriptions_per_connection: 1024,
590            message_buffer_capacity: 1024,
591            tokio_runtime: None,
592            ipc_socket_permissions: None,
593        }
594    }
595}
596
597/// Builder to configure and create a JSON-RPC server
598#[derive(Debug)]
599pub struct Builder<HttpMiddleware, RpcMiddleware> {
600    settings: Settings,
601    /// Subscription ID provider.
602    id_provider: Arc<dyn IdProvider>,
603    rpc_middleware: RpcServiceBuilder<RpcMiddleware>,
604    http_middleware: tower::ServiceBuilder<HttpMiddleware>,
605}
606
607impl Default for Builder<Identity, Identity> {
608    fn default() -> Self {
609        Self {
610            settings: Settings::default(),
611            id_provider: Arc::new(RandomIntegerIdProvider),
612            rpc_middleware: RpcServiceBuilder::new(),
613            http_middleware: tower::ServiceBuilder::new(),
614        }
615    }
616}
617
618impl<HttpMiddleware, RpcMiddleware> Builder<HttpMiddleware, RpcMiddleware> {
619    /// Set the maximum size of a request body in bytes. Default is 10 MiB.
620    pub const fn max_request_body_size(mut self, size: u32) -> Self {
621        self.settings.max_request_body_size = size;
622        self
623    }
624
625    /// Set the maximum size of a response body in bytes. Default is 10 MiB.
626    pub const fn max_response_body_size(mut self, size: u32) -> Self {
627        self.settings.max_response_body_size = size;
628        self
629    }
630
631    /// Set the maximum size of a log
632    pub const fn max_log_length(mut self, size: u32) -> Self {
633        self.settings.max_log_length = size;
634        self
635    }
636
637    /// Set the maximum number of connections allowed. Default is 100.
638    pub const fn max_connections(mut self, max: u32) -> Self {
639        self.settings.max_connections = max;
640        self
641    }
642
643    /// Set the maximum number of subscriptions per connection. Default is 1024.
644    pub const fn max_subscriptions_per_connection(mut self, max: u32) -> Self {
645        self.settings.max_subscriptions_per_connection = max;
646        self
647    }
648
649    /// The server enforces backpressure which means that
650    /// `n` messages can be buffered and if the client
651    /// can't keep up with the server.
652    ///
653    /// This `capacity` is applied per connection and
654    /// applies globally on the connection which implies
655    /// all JSON-RPC messages.
656    ///
657    /// For example if a subscription produces plenty of new items
658    /// and the client can't keep up then no new messages are handled.
659    ///
660    /// If this limit is exceeded then the server will "back-off"
661    /// and only accept new messages once the client reads pending messages.
662    ///
663    /// # Panics
664    ///
665    /// Panics if the buffer capacity is 0.
666    pub const fn set_message_buffer_capacity(mut self, c: u32) -> Self {
667        self.settings.message_buffer_capacity = c;
668        self
669    }
670
671    /// Configure a custom [`tokio::runtime::Handle`] to run the server on.
672    ///
673    /// Default: [`tokio::spawn`]
674    pub fn custom_tokio_runtime(mut self, rt: tokio::runtime::Handle) -> Self {
675        self.settings.tokio_runtime = Some(rt);
676        self
677    }
678
679    /// Sets the permissions for the IPC socket file.
680    pub fn set_ipc_socket_permissions(mut self, permissions: Option<String>) -> Self {
681        self.settings.ipc_socket_permissions = permissions;
682        self
683    }
684
685    /// Configure custom `subscription ID` provider for the server to use
686    /// to when getting new subscription calls.
687    ///
688    /// You may choose static dispatch or dynamic dispatch because
689    /// `IdProvider` is implemented for `Box<T>`.
690    ///
691    /// Default: [`RandomIntegerIdProvider`].
692    ///
693    /// # Examples
694    ///
695    /// ```rust
696    /// use jsonrpsee::server::RandomStringIdProvider;
697    /// use reth_ipc::server::Builder;
698    ///
699    /// // static dispatch
700    /// let builder1 = Builder::default().set_id_provider(RandomStringIdProvider::new(16));
701    ///
702    /// // or dynamic dispatch
703    /// let builder2 = Builder::default().set_id_provider(Box::new(RandomStringIdProvider::new(16)));
704    /// ```
705    pub fn set_id_provider<I: IdProvider + 'static>(mut self, id_provider: I) -> Self {
706        self.id_provider = Arc::new(id_provider);
707        self
708    }
709
710    /// Configure a custom [`tower::ServiceBuilder`] middleware for composing layers to be applied
711    /// to the RPC service.
712    ///
713    /// Default: No tower layers are applied to the RPC service.
714    ///
715    /// # Examples
716    ///
717    /// ```rust
718    /// #[tokio::main]
719    /// async fn main() {
720    ///     let builder = tower::ServiceBuilder::new();
721    ///     let server = reth_ipc::server::Builder::default()
722    ///         .set_http_middleware(builder)
723    ///         .build("/tmp/my-uds".into());
724    /// }
725    /// ```
726    pub fn set_http_middleware<T>(
727        self,
728        service_builder: tower::ServiceBuilder<T>,
729    ) -> Builder<T, RpcMiddleware> {
730        Builder {
731            settings: self.settings,
732            id_provider: self.id_provider,
733            http_middleware: service_builder,
734            rpc_middleware: self.rpc_middleware,
735        }
736    }
737
738    /// Enable middleware that is invoked on every JSON-RPC call.
739    ///
740    /// The middleware itself is very similar to the `tower middleware` but
741    /// it has a different service trait which takes &self instead &mut self
742    /// which means that you can't use built-in middleware from tower.
743    ///
744    /// Another consequence of `&self` is that you must wrap any of the middleware state in
745    /// a type which is Send and provides interior mutability such `Arc<Mutex>`.
746    ///
747    /// The builder itself exposes a similar API as the [`tower::ServiceBuilder`]
748    /// where it is possible to compose layers to the middleware.
749    pub fn set_rpc_middleware<T>(
750        self,
751        rpc_middleware: RpcServiceBuilder<T>,
752    ) -> Builder<HttpMiddleware, T> {
753        Builder {
754            settings: self.settings,
755            id_provider: self.id_provider,
756            rpc_middleware,
757            http_middleware: self.http_middleware,
758        }
759    }
760
761    /// Finalize the configuration of the server. Consumes the [`Builder`].
762    pub fn build(self, endpoint: String) -> IpcServer<HttpMiddleware, RpcMiddleware> {
763        IpcServer {
764            endpoint,
765            cfg: self.settings,
766            id_provider: self.id_provider,
767            http_middleware: self.http_middleware,
768            rpc_middleware: self.rpc_middleware,
769        }
770    }
771}
772
773#[cfg(test)]
774#[expect(missing_docs)]
775pub fn dummy_name() -> String {
776    use rand::Rng;
777    let num: u64 = rand::rng().random();
778    format!(r"/tmp/my-uds-{num}")
779}
780
781#[cfg(test)]
782mod tests {
783    use super::*;
784    use crate::client::IpcClientBuilder;
785    use futures::future::select;
786    use jsonrpsee::{
787        core::{
788            client::{self, ClientT, Error, Subscription, SubscriptionClientT},
789            middleware::{Batch, BatchEntry, Notification},
790            params::BatchRequestBuilder,
791        },
792        rpc_params,
793        types::{error::TOO_MANY_SUBSCRIPTIONS_CODE, ErrorCode, Request},
794        PendingSubscriptionSink, RpcModule, SubscriptionMessage,
795    };
796    use reth_tracing::init_test_tracing;
797    use std::pin::pin;
798    use tokio::sync::broadcast;
799    use tokio_stream::wrappers::BroadcastStream;
800
801    #[tokio::test]
802    #[cfg(unix)]
803    async fn test_ipc_socket_permissions() {
804        use std::os::unix::fs::PermissionsExt;
805        let endpoint = &dummy_name();
806        let perms = "0777";
807        let server = Builder::default()
808            .set_ipc_socket_permissions(Some(perms.to_string()))
809            .build(endpoint.clone());
810        let module = RpcModule::new(());
811        let handle = server.start(module).await.unwrap();
812        tokio::spawn(handle.stopped());
813
814        let meta = std::fs::metadata(endpoint).unwrap();
815        let perms = meta.permissions();
816        assert_eq!(perms.mode() & 0o777, 0o777);
817    }
818
819    async fn pipe_from_stream_with_bounded_buffer(
820        pending: PendingSubscriptionSink,
821        stream: BroadcastStream<usize>,
822    ) -> Result<(), Box<dyn core::error::Error + Send + Sync>> {
823        let sink = pending.accept().await.unwrap();
824        let closed = sink.closed();
825
826        let mut closed = pin!(closed);
827        let mut stream = pin!(stream);
828
829        loop {
830            match select(closed, stream.next()).await {
831                // subscription closed or stream is closed.
832                Either::Left((_, _)) | Either::Right((None, _)) => break Ok(()),
833
834                // received new item from the stream.
835                Either::Right((Some(Ok(item)), c)) => {
836                    let raw_value = serde_json::value::to_raw_value(&item)?;
837                    let notif = SubscriptionMessage::from(raw_value);
838
839                    // NOTE: this will block until there a spot in the queue
840                    // and you might want to do something smarter if it's
841                    // critical that "the most recent item" must be sent when it is produced.
842                    if sink.send(notif).await.is_err() {
843                        break Ok(());
844                    }
845
846                    closed = c;
847                }
848
849                // Send back the error.
850                Either::Right((Some(Err(e)), _)) => break Err(e.into()),
851            }
852        }
853    }
854
855    // Naive example that broadcasts the produced values to all active subscribers.
856    fn produce_items(tx: broadcast::Sender<usize>) {
857        for c in 1..=100 {
858            std::thread::sleep(std::time::Duration::from_millis(1));
859            let _ = tx.send(c);
860        }
861    }
862
863    #[tokio::test]
864    async fn can_set_the_max_response_body_size() {
865        // init_test_tracing();
866        let endpoint = &dummy_name();
867        let server = Builder::default().max_response_body_size(100).build(endpoint.clone());
868        let mut module = RpcModule::new(());
869        module.register_method("anything", |_, _, _| "a".repeat(101)).unwrap();
870        let handle = server.start(module).await.unwrap();
871        tokio::spawn(handle.stopped());
872
873        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
874        let response: Result<String, Error> = client.request("anything", rpc_params![]).await;
875        assert!(response.unwrap_err().to_string().contains("Exceeded max limit of"));
876    }
877
878    #[tokio::test]
879    async fn can_set_the_max_request_body_size() {
880        init_test_tracing();
881        let endpoint = &dummy_name();
882        let server = Builder::default().max_request_body_size(100).build(endpoint.clone());
883        let mut module = RpcModule::new(());
884        module.register_method("anything", |_, _, _| "succeed").unwrap();
885        let handle = server.start(module).await.unwrap();
886        tokio::spawn(handle.stopped());
887
888        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
889        let response: Result<String, Error> =
890            client.request("anything", rpc_params!["a".repeat(101)]).await;
891        assert!(response.is_err());
892        let mut batch_request_builder = BatchRequestBuilder::new();
893        let _ = batch_request_builder.insert("anything", rpc_params![]);
894        let _ = batch_request_builder.insert("anything", rpc_params![]);
895        let _ = batch_request_builder.insert("anything", rpc_params![]);
896        // the raw request string is:
897        //  [{"jsonrpc":"2.0","id":0,"method":"anything"},{"jsonrpc":"2.0","id":1, \
898        //    "method":"anything"},{"jsonrpc":"2.0","id":2,"method":"anything"}]"
899        // which is 136 bytes, more than 100 bytes.
900        let response: Result<client::BatchResponse<'_, String>, Error> =
901            client.batch_request(batch_request_builder).await;
902        assert!(response.is_err());
903    }
904
905    #[tokio::test]
906    async fn can_set_max_connections() {
907        init_test_tracing();
908
909        let endpoint = &dummy_name();
910        let server = Builder::default().max_connections(2).build(endpoint.clone());
911        let mut module = RpcModule::new(());
912        module.register_method("anything", |_, _, _| "succeed").unwrap();
913        let handle = server.start(module).await.unwrap();
914        tokio::spawn(handle.stopped());
915
916        let client1 = IpcClientBuilder::default().build(endpoint).await.unwrap();
917        let client2 = IpcClientBuilder::default().build(endpoint).await.unwrap();
918        let client3 = IpcClientBuilder::default().build(endpoint).await.unwrap();
919
920        let response1: Result<String, Error> = client1.request("anything", rpc_params![]).await;
921        let response2: Result<String, Error> = client2.request("anything", rpc_params![]).await;
922        let response3: Result<String, Error> = client3.request("anything", rpc_params![]).await;
923
924        assert!(response1.is_ok());
925        assert!(response2.is_ok());
926        // Third connection is rejected
927        assert!(response3.is_err());
928
929        // Decrement connection count
930        drop(client2);
931        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
932
933        // Can connect again
934        let client4 = IpcClientBuilder::default().build(endpoint).await.unwrap();
935        let response4: Result<String, Error> = client4.request("anything", rpc_params![]).await;
936        assert!(response4.is_ok());
937    }
938
939    #[tokio::test(flavor = "multi_thread")]
940    async fn test_pending_call_aborted_on_disconnect() {
941        init_test_tracing();
942
943        let endpoint = &dummy_name();
944        let server = Builder::default().build(endpoint.clone());
945        let (started_tx, started_rx) = oneshot::channel::<()>();
946        let (dropped_tx, dropped_rx) = oneshot::channel::<()>();
947        let mut module = RpcModule::new(std::sync::Mutex::new(Some((started_tx, dropped_tx))));
948        module
949            .register_async_method("hang", |_, ctx, _| async move {
950                // `dropped_tx` is dropped together with the call
951                let (started_tx, _dropped_tx) = ctx.lock().unwrap().take().unwrap();
952                let _ = started_tx.send(());
953                std::future::pending::<()>().await;
954                "unreachable"
955            })
956            .unwrap();
957        let handle = server.start(module).await.unwrap();
958        tokio::spawn(handle.stopped());
959
960        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
961        tokio::select! {
962            _ = client.request::<String, _>("hang", rpc_params![]) => panic!("call completed"),
963            _ = started_rx => {}
964        }
965        drop(client);
966
967        let dropped = tokio::time::timeout(std::time::Duration::from_secs(5), dropped_rx).await;
968        assert!(dropped.is_ok(), "call kept running after the connection closed");
969    }
970
971    #[tokio::test]
972    async fn test_panicking_call_returns_internal_error() {
973        init_test_tracing();
974
975        let endpoint = &dummy_name();
976        let server = Builder::default().build(endpoint.clone());
977        let mut module = RpcModule::new(());
978        module
979            .register_async_method("maybe_panic", |params, _, _| async move {
980                assert!(!params.one::<bool>().unwrap(), "requested panic");
981                "ok"
982            })
983            .unwrap();
984        let handle = server.start(module).await.unwrap();
985        tokio::spawn(handle.stopped());
986
987        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
988        let err = client.request::<String, _>("maybe_panic", rpc_params![true]).await.unwrap_err();
989        assert!(
990            matches!(&err, Error::Call(err) if err.code() == ErrorCode::InternalError.code()),
991            "{err:?}"
992        );
993
994        let mut batch_request_builder = BatchRequestBuilder::new();
995        let _ = batch_request_builder.insert("maybe_panic", rpc_params![true]);
996        let _ = batch_request_builder.insert("maybe_panic", rpc_params![false]);
997        let responses = client
998            .batch_request::<String>(batch_request_builder)
999            .await
1000            .unwrap()
1001            .into_iter()
1002            .collect::<Vec<_>>();
1003        assert!(matches!(&responses[0], Err(err) if err.code() == ErrorCode::InternalError.code()));
1004        assert_eq!(responses[1].as_deref(), Ok("ok"));
1005    }
1006
1007    #[tokio::test]
1008    async fn test_rpc_request() {
1009        init_test_tracing();
1010        let endpoint = &dummy_name();
1011        let server = Builder::default().build(endpoint.clone());
1012        let mut module = RpcModule::new(());
1013        let msg = r#"{"jsonrpc":"2.0","id":83,"result":"0x7a69"}"#;
1014        module.register_method("eth_chainId", move |_, _, _| msg).unwrap();
1015        let handle = server.start(module).await.unwrap();
1016        tokio::spawn(handle.stopped());
1017
1018        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
1019        let response: String = client.request("eth_chainId", rpc_params![]).await.unwrap();
1020        assert_eq!(response, msg);
1021    }
1022
1023    #[tokio::test]
1024    async fn test_batch_request() {
1025        let endpoint = &dummy_name();
1026        let server = Builder::default().build(endpoint.clone());
1027        let mut module = RpcModule::new(());
1028        module.register_method("anything", |_, _, _| "ok").unwrap();
1029        let handle = server.start(module).await.unwrap();
1030        tokio::spawn(handle.stopped());
1031
1032        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
1033        let mut batch_request_builder = BatchRequestBuilder::new();
1034        let _ = batch_request_builder.insert("anything", rpc_params![]);
1035        let _ = batch_request_builder.insert("anything", rpc_params![]);
1036        let _ = batch_request_builder.insert("anything", rpc_params![]);
1037        let result = client
1038            .batch_request(batch_request_builder)
1039            .await
1040            .unwrap()
1041            .into_ok()
1042            .unwrap()
1043            .collect::<Vec<String>>();
1044        assert_eq!(result, vec!["ok", "ok", "ok"]);
1045    }
1046
1047    #[tokio::test]
1048    async fn test_ipc_modules() {
1049        reth_tracing::init_test_tracing();
1050        let endpoint = &dummy_name();
1051        let server = Builder::default().build(endpoint.clone());
1052        let mut module = RpcModule::new(());
1053        let msg = r#"{"admin":"1.0","debug":"1.0","engine":"1.0","eth":"1.0","ethash":"1.0","miner":"1.0","net":"1.0","rpc":"1.0","txpool":"1.0","web3":"1.0"}"#;
1054        module.register_method("rpc_modules", move |_, _, _| msg).unwrap();
1055        let handle = server.start(module).await.unwrap();
1056        tokio::spawn(handle.stopped());
1057
1058        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
1059        let response: String = client.request("rpc_modules", rpc_params![]).await.unwrap();
1060        assert_eq!(response, msg);
1061    }
1062
1063    #[tokio::test(flavor = "multi_thread")]
1064    async fn test_rpc_subscription() {
1065        let endpoint = &dummy_name();
1066        let server = Builder::default().build(endpoint.clone());
1067        let (tx, _rx) = broadcast::channel::<usize>(16);
1068
1069        let mut module = RpcModule::new(tx.clone());
1070        std::thread::spawn(move || produce_items(tx));
1071
1072        module
1073            .register_subscription(
1074                "subscribe_hello",
1075                "s_hello",
1076                "unsubscribe_hello",
1077                |_, pending, tx, _| async move {
1078                    let rx = tx.subscribe();
1079                    let stream = BroadcastStream::new(rx);
1080                    pipe_from_stream_with_bounded_buffer(pending, stream).await?;
1081                    Ok(())
1082                },
1083            )
1084            .unwrap();
1085
1086        let handle = server.start(module).await.unwrap();
1087        tokio::spawn(handle.stopped());
1088
1089        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
1090        let sub: Subscription<usize> =
1091            client.subscribe("subscribe_hello", rpc_params![], "unsubscribe_hello").await.unwrap();
1092
1093        let items = sub.take(16).collect::<Vec<_>>().await;
1094        assert_eq!(items.len(), 16);
1095    }
1096
1097    #[tokio::test(flavor = "multi_thread")]
1098    async fn test_max_subscriptions_per_connection() {
1099        let endpoint = &dummy_name();
1100        let server = Builder::default().max_subscriptions_per_connection(1).build(endpoint.clone());
1101        let mut module = RpcModule::new(());
1102        module
1103            .register_subscription(
1104                "subscribe_hello",
1105                "s_hello",
1106                "unsubscribe_hello",
1107                |_, pending, _, _| async move {
1108                    let Ok(sink) = pending.accept().await else { return };
1109                    sink.closed().await;
1110                },
1111            )
1112            .unwrap();
1113        let handle = server.start(module).await.unwrap();
1114        tokio::spawn(handle.stopped());
1115
1116        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
1117        let _sub: Subscription<usize> =
1118            client.subscribe("subscribe_hello", rpc_params![], "unsubscribe_hello").await.unwrap();
1119        let err = client
1120            .subscribe::<usize, _>("subscribe_hello", rpc_params![], "unsubscribe_hello")
1121            .await
1122            .unwrap_err();
1123        assert!(
1124            matches!(&err, Error::Call(err) if err.code() == TOO_MANY_SUBSCRIPTIONS_CODE),
1125            "{err}"
1126        );
1127    }
1128
1129    #[tokio::test]
1130    async fn test_rpc_middleware() {
1131        #[derive(Clone)]
1132        struct ModifyRequestIf<S>(S);
1133
1134        impl<S> RpcServiceT for ModifyRequestIf<S>
1135        where
1136            S: Send + Sync + RpcServiceT,
1137        {
1138            type MethodResponse = S::MethodResponse;
1139            type NotificationResponse = S::NotificationResponse;
1140            type BatchResponse = S::BatchResponse;
1141
1142            fn call<'a>(
1143                &self,
1144                mut req: Request<'a>,
1145            ) -> impl Future<Output = Self::MethodResponse> + Send + 'a {
1146                // Re-direct all calls that isn't `say_hello` to `say_goodbye`
1147                if req.method == "say_hello" {
1148                    req.method = "say_goodbye".into();
1149                } else if req.method == "say_goodbye" {
1150                    req.method = "say_hello".into();
1151                }
1152
1153                self.0.call(req)
1154            }
1155
1156            fn batch<'a>(
1157                &self,
1158                mut batch: Batch<'a>,
1159            ) -> impl Future<Output = Self::BatchResponse> + Send + 'a {
1160                for call in batch.iter_mut() {
1161                    match call {
1162                        Ok(BatchEntry::Call(req)) => {
1163                            if req.method == "say_hello" {
1164                                req.method = "say_goodbye".into();
1165                            } else if req.method == "say_goodbye" {
1166                                req.method = "say_hello".into();
1167                            }
1168                        }
1169                        Ok(BatchEntry::Notification(n)) => {
1170                            if n.method == "say_hello" {
1171                                n.method = "say_goodbye".into();
1172                            } else if n.method == "say_goodbye" {
1173                                n.method = "say_hello".into();
1174                            }
1175                        }
1176                        // Invalid request, we don't care about it.
1177                        Err(_err) => {}
1178                    }
1179                }
1180
1181                self.0.batch(batch)
1182            }
1183
1184            fn notification<'a>(
1185                &self,
1186                mut n: Notification<'a>,
1187            ) -> impl Future<Output = Self::NotificationResponse> + Send + 'a {
1188                if n.method == "say_hello" {
1189                    n.method = "say_goodbye".into();
1190                } else if n.method == "say_goodbye" {
1191                    n.method = "say_hello".into();
1192                }
1193                self.0.notification(n)
1194            }
1195        }
1196
1197        reth_tracing::init_test_tracing();
1198        let endpoint = &dummy_name();
1199
1200        let rpc_middleware = RpcServiceBuilder::new().layer_fn(ModifyRequestIf);
1201        let server = Builder::default().set_rpc_middleware(rpc_middleware).build(endpoint.clone());
1202
1203        let mut module = RpcModule::new(());
1204        let goodbye_msg = r#"{"jsonrpc":"2.0","id":1,"result":"goodbye"}"#;
1205        let hello_msg = r#"{"jsonrpc":"2.0","id":2,"result":"hello"}"#;
1206        module.register_method("say_hello", move |_, _, _| hello_msg).unwrap();
1207        module.register_method("say_goodbye", move |_, _, _| goodbye_msg).unwrap();
1208        let handle = server.start(module).await.unwrap();
1209        tokio::spawn(handle.stopped());
1210
1211        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
1212        let say_hello_response: String = client.request("say_hello", rpc_params![]).await.unwrap();
1213        let say_goodbye_response: String =
1214            client.request("say_goodbye", rpc_params![]).await.unwrap();
1215
1216        assert_eq!(say_hello_response, goodbye_msg);
1217        assert_eq!(say_goodbye_response, hello_msg);
1218    }
1219}
reth_ipc/server/mod.rs

reth_ipc/server/
mod.rs
