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_bytes = request.len(), "Received IPC 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 codec =
489        StreamCodec::stream_incoming().with_max_length(server_cfg.max_request_body_size as usize);
490    let ipc = IpcConn(tokio_util::codec::Decoder::framed(codec, local_socket_stream));
491
492    let (tx, rx) = mpsc::channel::<Box<JsonRawValue>>(server_cfg.message_buffer_capacity as usize);
493    let method_sink = MethodSink::new_with_limit(tx, server_cfg.max_response_body_size);
494    let tower_service = TowerServiceNoHttp {
495        inner: ServiceData {
496            methods,
497            id_provider,
498            stop_handle: stop_handle.clone(),
499            server_cfg: server_cfg.clone(),
500            conn_id,
501            conn_permit,
502            bounded_subscriptions: BoundedSubscriptions::new(
503                server_cfg.max_subscriptions_per_connection,
504            ),
505            method_sink,
506        },
507        rpc_middleware,
508    };
509
510    let service = http_middleware.service(tower_service);
511    tokio::spawn(async {
512        to_ipc_service(ipc, service, stop_handle, rx).in_current_span().await;
513        drop(drop_on_completion)
514    });
515}
516
517async fn to_ipc_service<S, T>(
518    ipc: IpcConn<JsonRpcStream<T>>,
519    service: S,
520    stop_handle: StopHandle,
521    rx: mpsc::Receiver<Box<JsonRawValue>>,
522) where
523    S: Service<String, Response = Option<String>> + Send + 'static,
524    S::Error: Into<Box<dyn core::error::Error + Send + Sync>>,
525    S::Future: Send + Unpin,
526    T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
527{
528    let rx_item = ReceiverStream::new(rx);
529    let conn = IpcConnDriver {
530        conn: ipc,
531        service,
532        pending_calls: Default::default(),
533        items: Default::default(),
534    };
535    let stopped = stop_handle.shutdown();
536
537    let mut conn = pin!(conn);
538    let mut rx_item = pin!(rx_item);
539    let mut stopped = pin!(stopped);
540
541    loop {
542        tokio::select! {
543            _ = &mut conn => {
544               break
545            }
546            item = rx_item.next() => {
547                let Some(item) = item else { break };
548                conn.push_back(String::from(Box::<str>::from(item)));
549            }
550            _ = &mut stopped => {
551                // shutdown
552                break
553            }
554        }
555    }
556}
557
558/// JSON-RPC IPC server settings.
559#[derive(Debug, Clone)]
560pub struct Settings {
561    /// Maximum size in bytes of a request.
562    max_request_body_size: u32,
563    /// Maximum size in bytes of a response.
564    max_response_body_size: u32,
565    /// Max length for logging for requests and responses
566    ///
567    /// Logs bigger than this limit will be truncated.
568    max_log_length: u32,
569    /// Maximum number of incoming connections allowed.
570    max_connections: u32,
571    /// Maximum number of subscriptions per connection.
572    max_subscriptions_per_connection: u32,
573    /// Number of messages that server is allowed `buffer` until backpressure kicks in.
574    message_buffer_capacity: u32,
575    /// Custom tokio runtime to run the server on.
576    tokio_runtime: Option<tokio::runtime::Handle>,
577    /// The permissions to create the IPC socket with.
578    ipc_socket_permissions: Option<String>,
579}
580
581impl Default for Settings {
582    fn default() -> Self {
583        Self {
584            max_request_body_size: TEN_MB_SIZE_BYTES,
585            max_response_body_size: TEN_MB_SIZE_BYTES,
586            max_log_length: 4096,
587            max_connections: 100,
588            max_subscriptions_per_connection: 1024,
589            message_buffer_capacity: 1024,
590            tokio_runtime: None,
591            ipc_socket_permissions: None,
592        }
593    }
594}
595
596/// Builder to configure and create a JSON-RPC server
597#[derive(Debug)]
598pub struct Builder<HttpMiddleware, RpcMiddleware> {
599    settings: Settings,
600    /// Subscription ID provider.
601    id_provider: Arc<dyn IdProvider>,
602    rpc_middleware: RpcServiceBuilder<RpcMiddleware>,
603    http_middleware: tower::ServiceBuilder<HttpMiddleware>,
604}
605
606impl Default for Builder<Identity, Identity> {
607    fn default() -> Self {
608        Self {
609            settings: Settings::default(),
610            id_provider: Arc::new(RandomIntegerIdProvider),
611            rpc_middleware: RpcServiceBuilder::new(),
612            http_middleware: tower::ServiceBuilder::new(),
613        }
614    }
615}
616
617impl<HttpMiddleware, RpcMiddleware> Builder<HttpMiddleware, RpcMiddleware> {
618    /// Set the maximum size of a request body in bytes. Default is 10 MiB.
619    pub const fn max_request_body_size(mut self, size: u32) -> Self {
620        self.settings.max_request_body_size = size;
621        self
622    }
623
624    /// Set the maximum size of a response body in bytes. Default is 10 MiB.
625    pub const fn max_response_body_size(mut self, size: u32) -> Self {
626        self.settings.max_response_body_size = size;
627        self
628    }
629
630    /// Set the maximum size of a log
631    pub const fn max_log_length(mut self, size: u32) -> Self {
632        self.settings.max_log_length = size;
633        self
634    }
635
636    /// Set the maximum number of connections allowed. Default is 100.
637    pub const fn max_connections(mut self, max: u32) -> Self {
638        self.settings.max_connections = max;
639        self
640    }
641
642    /// Set the maximum number of subscriptions per connection. Default is 1024.
643    pub const fn max_subscriptions_per_connection(mut self, max: u32) -> Self {
644        self.settings.max_subscriptions_per_connection = max;
645        self
646    }
647
648    /// The server enforces backpressure which means that
649    /// `n` messages can be buffered and if the client
650    /// can't keep up with the server.
651    ///
652    /// This `capacity` is applied per connection and
653    /// applies globally on the connection which implies
654    /// all JSON-RPC messages.
655    ///
656    /// For example if a subscription produces plenty of new items
657    /// and the client can't keep up then no new messages are handled.
658    ///
659    /// If this limit is exceeded then the server will "back-off"
660    /// and only accept new messages once the client reads pending messages.
661    ///
662    /// # Panics
663    ///
664    /// Panics if the buffer capacity is 0.
665    pub const fn set_message_buffer_capacity(mut self, c: u32) -> Self {
666        self.settings.message_buffer_capacity = c;
667        self
668    }
669
670    /// Configure a custom [`tokio::runtime::Handle`] to run the server on.
671    ///
672    /// Default: [`tokio::spawn`]
673    pub fn custom_tokio_runtime(mut self, rt: tokio::runtime::Handle) -> Self {
674        self.settings.tokio_runtime = Some(rt);
675        self
676    }
677
678    /// Sets the permissions for the IPC socket file.
679    pub fn set_ipc_socket_permissions(mut self, permissions: Option<String>) -> Self {
680        self.settings.ipc_socket_permissions = permissions;
681        self
682    }
683
684    /// Configure custom `subscription ID` provider for the server to use
685    /// to when getting new subscription calls.
686    ///
687    /// You may choose static dispatch or dynamic dispatch because
688    /// `IdProvider` is implemented for `Box<T>`.
689    ///
690    /// Default: [`RandomIntegerIdProvider`].
691    ///
692    /// # Examples
693    ///
694    /// ```rust
695    /// use jsonrpsee::server::RandomStringIdProvider;
696    /// use reth_ipc::server::Builder;
697    ///
698    /// // static dispatch
699    /// let builder1 = Builder::default().set_id_provider(RandomStringIdProvider::new(16));
700    ///
701    /// // or dynamic dispatch
702    /// let builder2 = Builder::default().set_id_provider(Box::new(RandomStringIdProvider::new(16)));
703    /// ```
704    pub fn set_id_provider<I: IdProvider + 'static>(mut self, id_provider: I) -> Self {
705        self.id_provider = Arc::new(id_provider);
706        self
707    }
708
709    /// Configure a custom [`tower::ServiceBuilder`] middleware for composing layers to be applied
710    /// to the RPC service.
711    ///
712    /// Default: No tower layers are applied to the RPC service.
713    ///
714    /// # Examples
715    ///
716    /// ```rust
717    /// #[tokio::main]
718    /// async fn main() {
719    ///     let builder = tower::ServiceBuilder::new();
720    ///     let server = reth_ipc::server::Builder::default()
721    ///         .set_http_middleware(builder)
722    ///         .build("/tmp/my-uds".into());
723    /// }
724    /// ```
725    pub fn set_http_middleware<T>(
726        self,
727        service_builder: tower::ServiceBuilder<T>,
728    ) -> Builder<T, RpcMiddleware> {
729        Builder {
730            settings: self.settings,
731            id_provider: self.id_provider,
732            http_middleware: service_builder,
733            rpc_middleware: self.rpc_middleware,
734        }
735    }
736
737    /// Enable middleware that is invoked on every JSON-RPC call.
738    ///
739    /// The middleware itself is very similar to the `tower middleware` but
740    /// it has a different service trait which takes &self instead &mut self
741    /// which means that you can't use built-in middleware from tower.
742    ///
743    /// Another consequence of `&self` is that you must wrap any of the middleware state in
744    /// a type which is Send and provides interior mutability such `Arc<Mutex>`.
745    ///
746    /// The builder itself exposes a similar API as the [`tower::ServiceBuilder`]
747    /// where it is possible to compose layers to the middleware.
748    pub fn set_rpc_middleware<T>(
749        self,
750        rpc_middleware: RpcServiceBuilder<T>,
751    ) -> Builder<HttpMiddleware, T> {
752        Builder {
753            settings: self.settings,
754            id_provider: self.id_provider,
755            rpc_middleware,
756            http_middleware: self.http_middleware,
757        }
758    }
759
760    /// Finalize the configuration of the server. Consumes the [`Builder`].
761    pub fn build(self, endpoint: String) -> IpcServer<HttpMiddleware, RpcMiddleware> {
762        IpcServer {
763            endpoint,
764            cfg: self.settings,
765            id_provider: self.id_provider,
766            http_middleware: self.http_middleware,
767            rpc_middleware: self.rpc_middleware,
768        }
769    }
770}
771
772#[cfg(test)]
773#[expect(missing_docs)]
774pub fn dummy_name() -> String {
775    use rand::Rng;
776    let num: u64 = rand::rng().random();
777    format!(r"/tmp/my-uds-{num}")
778}
779
780#[cfg(test)]
781mod tests {
782    use super::*;
783    use crate::client::IpcClientBuilder;
784    use futures::future::select;
785    use jsonrpsee::{
786        core::{
787            client::{self, ClientT, Error, Subscription, SubscriptionClientT},
788            middleware::{Batch, BatchEntry, Notification},
789            params::BatchRequestBuilder,
790        },
791        rpc_params,
792        types::{error::TOO_MANY_SUBSCRIPTIONS_CODE, ErrorCode, Request},
793        PendingSubscriptionSink, RpcModule, SubscriptionMessage,
794    };
795    use reth_tracing::init_test_tracing;
796    use std::pin::pin;
797    use tokio::sync::broadcast;
798    use tokio_stream::wrappers::BroadcastStream;
799
800    #[tokio::test]
801    #[cfg(unix)]
802    async fn test_ipc_socket_permissions() {
803        use std::os::unix::fs::PermissionsExt;
804        let endpoint = &dummy_name();
805        let perms = "0777";
806        let server = Builder::default()
807            .set_ipc_socket_permissions(Some(perms.to_string()))
808            .build(endpoint.clone());
809        let module = RpcModule::new(());
810        let handle = server.start(module).await.unwrap();
811        tokio::spawn(handle.stopped());
812
813        let meta = std::fs::metadata(endpoint).unwrap();
814        let perms = meta.permissions();
815        assert_eq!(perms.mode() & 0o777, 0o777);
816    }
817
818    async fn pipe_from_stream_with_bounded_buffer(
819        pending: PendingSubscriptionSink,
820        stream: BroadcastStream<usize>,
821    ) -> Result<(), Box<dyn core::error::Error + Send + Sync>> {
822        let sink = pending.accept().await.unwrap();
823        let closed = sink.closed();
824
825        let mut closed = pin!(closed);
826        let mut stream = pin!(stream);
827
828        loop {
829            match select(closed, stream.next()).await {
830                // subscription closed or stream is closed.
831                Either::Left((_, _)) | Either::Right((None, _)) => break Ok(()),
832
833                // received new item from the stream.
834                Either::Right((Some(Ok(item)), c)) => {
835                    let raw_value = serde_json::value::to_raw_value(&item)?;
836                    let notif = SubscriptionMessage::from(raw_value);
837
838                    // NOTE: this will block until there a spot in the queue
839                    // and you might want to do something smarter if it's
840                    // critical that "the most recent item" must be sent when it is produced.
841                    if sink.send(notif).await.is_err() {
842                        break Ok(());
843                    }
844
845                    closed = c;
846                }
847
848                // Send back the error.
849                Either::Right((Some(Err(e)), _)) => break Err(e.into()),
850            }
851        }
852    }
853
854    // Naive example that broadcasts the produced values to all active subscribers.
855    fn produce_items(tx: broadcast::Sender<usize>) {
856        for c in 1..=100 {
857            std::thread::sleep(std::time::Duration::from_millis(1));
858            let _ = tx.send(c);
859        }
860    }
861
862    #[tokio::test]
863    async fn can_set_the_max_response_body_size() {
864        // init_test_tracing();
865        let endpoint = &dummy_name();
866        let server = Builder::default().max_response_body_size(100).build(endpoint.clone());
867        let mut module = RpcModule::new(());
868        module.register_method("anything", |_, _, _| "a".repeat(101)).unwrap();
869        let handle = server.start(module).await.unwrap();
870        tokio::spawn(handle.stopped());
871
872        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
873        let response: Result<String, Error> = client.request("anything", rpc_params![]).await;
874        assert!(response.unwrap_err().to_string().contains("Exceeded max limit of"));
875    }
876
877    #[tokio::test]
878    async fn can_set_the_max_request_body_size() {
879        init_test_tracing();
880        let endpoint = &dummy_name();
881        let server = Builder::default().max_request_body_size(100).build(endpoint.clone());
882        let mut module = RpcModule::new(());
883        module.register_method("anything", |_, _, _| "succeed").unwrap();
884        let handle = server.start(module).await.unwrap();
885        tokio::spawn(handle.stopped());
886
887        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
888        let response: Result<String, Error> =
889            client.request("anything", rpc_params!["a".repeat(101)]).await;
890        assert!(response.is_err());
891        let mut batch_request_builder = BatchRequestBuilder::new();
892        let _ = batch_request_builder.insert("anything", rpc_params![]);
893        let _ = batch_request_builder.insert("anything", rpc_params![]);
894        let _ = batch_request_builder.insert("anything", rpc_params![]);
895        // the raw request string is:
896        //  [{"jsonrpc":"2.0","id":0,"method":"anything"},{"jsonrpc":"2.0","id":1, \
897        //    "method":"anything"},{"jsonrpc":"2.0","id":2,"method":"anything"}]"
898        // which is 136 bytes, more than 100 bytes.
899        let response: Result<client::BatchResponse<'_, String>, Error> =
900            client.batch_request(batch_request_builder).await;
901        assert!(response.is_err());
902    }
903
904    #[tokio::test]
905    async fn can_set_max_connections() {
906        init_test_tracing();
907
908        let endpoint = &dummy_name();
909        let server = Builder::default().max_connections(2).build(endpoint.clone());
910        let mut module = RpcModule::new(());
911        module.register_method("anything", |_, _, _| "succeed").unwrap();
912        let handle = server.start(module).await.unwrap();
913        tokio::spawn(handle.stopped());
914
915        let client1 = IpcClientBuilder::default().build(endpoint).await.unwrap();
916        let client2 = IpcClientBuilder::default().build(endpoint).await.unwrap();
917        let client3 = IpcClientBuilder::default().build(endpoint).await.unwrap();
918
919        let response1: Result<String, Error> = client1.request("anything", rpc_params![]).await;
920        let response2: Result<String, Error> = client2.request("anything", rpc_params![]).await;
921        let response3: Result<String, Error> = client3.request("anything", rpc_params![]).await;
922
923        assert!(response1.is_ok());
924        assert!(response2.is_ok());
925        // Third connection is rejected
926        assert!(response3.is_err());
927
928        // Decrement connection count
929        drop(client2);
930        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
931
932        // Can connect again
933        let client4 = IpcClientBuilder::default().build(endpoint).await.unwrap();
934        let response4: Result<String, Error> = client4.request("anything", rpc_params![]).await;
935        assert!(response4.is_ok());
936    }
937
938    #[tokio::test(flavor = "multi_thread")]
939    async fn test_pending_call_aborted_on_disconnect() {
940        init_test_tracing();
941
942        let endpoint = &dummy_name();
943        let server = Builder::default().build(endpoint.clone());
944        let (started_tx, started_rx) = oneshot::channel::<()>();
945        let (dropped_tx, dropped_rx) = oneshot::channel::<()>();
946        let mut module = RpcModule::new(std::sync::Mutex::new(Some((started_tx, dropped_tx))));
947        module
948            .register_async_method("hang", |_, ctx, _| async move {
949                // `dropped_tx` is dropped together with the call
950                let (started_tx, _dropped_tx) = ctx.lock().unwrap().take().unwrap();
951                let _ = started_tx.send(());
952                std::future::pending::<()>().await;
953                "unreachable"
954            })
955            .unwrap();
956        let handle = server.start(module).await.unwrap();
957        tokio::spawn(handle.stopped());
958
959        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
960        tokio::select! {
961            _ = client.request::<String, _>("hang", rpc_params![]) => panic!("call completed"),
962            _ = started_rx => {}
963        }
964        drop(client);
965
966        let dropped = tokio::time::timeout(std::time::Duration::from_secs(5), dropped_rx).await;
967        assert!(dropped.is_ok(), "call kept running after the connection closed");
968    }
969
970    #[tokio::test]
971    async fn test_panicking_call_returns_internal_error() {
972        init_test_tracing();
973
974        let endpoint = &dummy_name();
975        let server = Builder::default().build(endpoint.clone());
976        let mut module = RpcModule::new(());
977        module
978            .register_async_method("maybe_panic", |params, _, _| async move {
979                assert!(!params.one::<bool>().unwrap(), "requested panic");
980                "ok"
981            })
982            .unwrap();
983        let handle = server.start(module).await.unwrap();
984        tokio::spawn(handle.stopped());
985
986        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
987        let err = client.request::<String, _>("maybe_panic", rpc_params![true]).await.unwrap_err();
988        assert!(
989            matches!(&err, Error::Call(err) if err.code() == ErrorCode::InternalError.code()),
990            "{err:?}"
991        );
992
993        let mut batch_request_builder = BatchRequestBuilder::new();
994        let _ = batch_request_builder.insert("maybe_panic", rpc_params![true]);
995        let _ = batch_request_builder.insert("maybe_panic", rpc_params![false]);
996        let responses = client
997            .batch_request::<String>(batch_request_builder)
998            .await
999            .unwrap()
1000            .into_iter()
1001            .collect::<Vec<_>>();
1002        assert!(matches!(&responses[0], Err(err) if err.code() == ErrorCode::InternalError.code()));
1003        assert_eq!(responses[1].as_deref(), Ok("ok"));
1004    }
1005
1006    #[tokio::test]
1007    async fn test_rpc_request() {
1008        init_test_tracing();
1009        let endpoint = &dummy_name();
1010        let server = Builder::default().build(endpoint.clone());
1011        let mut module = RpcModule::new(());
1012        let msg = r#"{"jsonrpc":"2.0","id":83,"result":"0x7a69"}"#;
1013        module.register_method("eth_chainId", move |_, _, _| msg).unwrap();
1014        let handle = server.start(module).await.unwrap();
1015        tokio::spawn(handle.stopped());
1016
1017        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
1018        let response: String = client.request("eth_chainId", rpc_params![]).await.unwrap();
1019        assert_eq!(response, msg);
1020    }
1021
1022    #[tokio::test]
1023    async fn test_batch_request() {
1024        let endpoint = &dummy_name();
1025        let server = Builder::default().build(endpoint.clone());
1026        let mut module = RpcModule::new(());
1027        module.register_method("anything", |_, _, _| "ok").unwrap();
1028        let handle = server.start(module).await.unwrap();
1029        tokio::spawn(handle.stopped());
1030
1031        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
1032        let mut batch_request_builder = BatchRequestBuilder::new();
1033        let _ = batch_request_builder.insert("anything", rpc_params![]);
1034        let _ = batch_request_builder.insert("anything", rpc_params![]);
1035        let _ = batch_request_builder.insert("anything", rpc_params![]);
1036        let result = client
1037            .batch_request(batch_request_builder)
1038            .await
1039            .unwrap()
1040            .into_ok()
1041            .unwrap()
1042            .collect::<Vec<String>>();
1043        assert_eq!(result, vec!["ok", "ok", "ok"]);
1044    }
1045
1046    #[tokio::test]
1047    async fn test_ipc_modules() {
1048        reth_tracing::init_test_tracing();
1049        let endpoint = &dummy_name();
1050        let server = Builder::default().build(endpoint.clone());
1051        let mut module = RpcModule::new(());
1052        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"}"#;
1053        module.register_method("rpc_modules", move |_, _, _| msg).unwrap();
1054        let handle = server.start(module).await.unwrap();
1055        tokio::spawn(handle.stopped());
1056
1057        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
1058        let response: String = client.request("rpc_modules", rpc_params![]).await.unwrap();
1059        assert_eq!(response, msg);
1060    }
1061
1062    #[tokio::test(flavor = "multi_thread")]
1063    async fn test_rpc_subscription() {
1064        let endpoint = &dummy_name();
1065        let server = Builder::default().build(endpoint.clone());
1066        let (tx, _rx) = broadcast::channel::<usize>(16);
1067
1068        let mut module = RpcModule::new(tx.clone());
1069        std::thread::spawn(move || produce_items(tx));
1070
1071        module
1072            .register_subscription(
1073                "subscribe_hello",
1074                "s_hello",
1075                "unsubscribe_hello",
1076                |_, pending, tx, _| async move {
1077                    let rx = tx.subscribe();
1078                    let stream = BroadcastStream::new(rx);
1079                    pipe_from_stream_with_bounded_buffer(pending, stream).await?;
1080                    Ok(())
1081                },
1082            )
1083            .unwrap();
1084
1085        let handle = server.start(module).await.unwrap();
1086        tokio::spawn(handle.stopped());
1087
1088        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
1089        let sub: Subscription<usize> =
1090            client.subscribe("subscribe_hello", rpc_params![], "unsubscribe_hello").await.unwrap();
1091
1092        let items = sub.take(16).collect::<Vec<_>>().await;
1093        assert_eq!(items.len(), 16);
1094    }
1095
1096    #[tokio::test(flavor = "multi_thread")]
1097    async fn test_max_subscriptions_per_connection() {
1098        let endpoint = &dummy_name();
1099        let server = Builder::default().max_subscriptions_per_connection(1).build(endpoint.clone());
1100        let mut module = RpcModule::new(());
1101        module
1102            .register_subscription(
1103                "subscribe_hello",
1104                "s_hello",
1105                "unsubscribe_hello",
1106                |_, pending, _, _| async move {
1107                    let Ok(sink) = pending.accept().await else { return };
1108                    sink.closed().await;
1109                },
1110            )
1111            .unwrap();
1112        let handle = server.start(module).await.unwrap();
1113        tokio::spawn(handle.stopped());
1114
1115        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
1116        let _sub: Subscription<usize> =
1117            client.subscribe("subscribe_hello", rpc_params![], "unsubscribe_hello").await.unwrap();
1118        let err = client
1119            .subscribe::<usize, _>("subscribe_hello", rpc_params![], "unsubscribe_hello")
1120            .await
1121            .unwrap_err();
1122        assert!(
1123            matches!(&err, Error::Call(err) if err.code() == TOO_MANY_SUBSCRIPTIONS_CODE),
1124            "{err}"
1125        );
1126    }
1127
1128    #[tokio::test]
1129    async fn test_rpc_middleware() {
1130        #[derive(Clone)]
1131        struct ModifyRequestIf<S>(S);
1132
1133        impl<S> RpcServiceT for ModifyRequestIf<S>
1134        where
1135            S: Send + Sync + RpcServiceT,
1136        {
1137            type MethodResponse = S::MethodResponse;
1138            type NotificationResponse = S::NotificationResponse;
1139            type BatchResponse = S::BatchResponse;
1140
1141            fn call<'a>(
1142                &self,
1143                mut req: Request<'a>,
1144            ) -> impl Future<Output = Self::MethodResponse> + Send + 'a {
1145                // Re-direct all calls that isn't `say_hello` to `say_goodbye`
1146                if req.method == "say_hello" {
1147                    req.method = "say_goodbye".into();
1148                } else if req.method == "say_goodbye" {
1149                    req.method = "say_hello".into();
1150                }
1151
1152                self.0.call(req)
1153            }
1154
1155            fn batch<'a>(
1156                &self,
1157                mut batch: Batch<'a>,
1158            ) -> impl Future<Output = Self::BatchResponse> + Send + 'a {
1159                for call in batch.iter_mut() {
1160                    match call {
1161                        Ok(BatchEntry::Call(req)) => {
1162                            if req.method == "say_hello" {
1163                                req.method = "say_goodbye".into();
1164                            } else if req.method == "say_goodbye" {
1165                                req.method = "say_hello".into();
1166                            }
1167                        }
1168                        Ok(BatchEntry::Notification(n)) => {
1169                            if n.method == "say_hello" {
1170                                n.method = "say_goodbye".into();
1171                            } else if n.method == "say_goodbye" {
1172                                n.method = "say_hello".into();
1173                            }
1174                        }
1175                        // Invalid request, we don't care about it.
1176                        Err(_err) => {}
1177                    }
1178                }
1179
1180                self.0.batch(batch)
1181            }
1182
1183            fn notification<'a>(
1184                &self,
1185                mut n: Notification<'a>,
1186            ) -> impl Future<Output = Self::NotificationResponse> + Send + 'a {
1187                if n.method == "say_hello" {
1188                    n.method = "say_goodbye".into();
1189                } else if n.method == "say_goodbye" {
1190                    n.method = "say_hello".into();
1191                }
1192                self.0.notification(n)
1193            }
1194        }
1195
1196        reth_tracing::init_test_tracing();
1197        let endpoint = &dummy_name();
1198
1199        let rpc_middleware = RpcServiceBuilder::new().layer_fn(ModifyRequestIf);
1200        let server = Builder::default().set_rpc_middleware(rpc_middleware).build(endpoint.clone());
1201
1202        let mut module = RpcModule::new(());
1203        let goodbye_msg = r#"{"jsonrpc":"2.0","id":1,"result":"goodbye"}"#;
1204        let hello_msg = r#"{"jsonrpc":"2.0","id":2,"result":"hello"}"#;
1205        module.register_method("say_hello", move |_, _, _| hello_msg).unwrap();
1206        module.register_method("say_goodbye", move |_, _, _| goodbye_msg).unwrap();
1207        let handle = server.start(module).await.unwrap();
1208        tokio::spawn(handle.stopped());
1209
1210        let client = IpcClientBuilder::default().build(endpoint).await.unwrap();
1211        let say_hello_response: String = client.request("say_hello", rpc_params![]).await.unwrap();
1212        let say_goodbye_response: String =
1213            client.request("say_goodbye", rpc_params![]).await.unwrap();
1214
1215        assert_eq!(say_hello_response, goodbye_msg);
1216        assert_eq!(say_goodbye_response, hello_msg);
1217    }
1218}
reth_ipc/server/mod.rs

reth_ipc/server/
mod.rs
