thruput_json_rpc_server/
batch.rs

1//! Ordered, bounded dispatch for JSON-RPC arrays.
2
3use std::{collections::BTreeMap, num::NonZeroUsize, str::FromStr};
4
5use futures_util::{StreamExt, stream::FuturesUnordered};
6use jsonrpsee::{
7    core::{
8        middleware::{Batch, BatchEntry, Notification, RpcServiceT},
9        server::{BatchResponseBuilder, MethodResponse},
10    },
11    types::{Id, Request, error::reject_too_big_batch_response},
12};
13
14/// Maximum JSON-RPC array entries accepted by the gateway and embedded node.
15pub const MAX_RPC_BATCH_REQUESTS: u32 = 1_000;
16
17/// Per-array limit on active and buffered entries (1..=1024, default 128).
18/// Independent of backend submit batching and array-length limits.
19#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
20#[serde(try_from = "usize")]
21pub struct RpcBatchConcurrency(NonZeroUsize);
22
23impl RpcBatchConcurrency {
24    /// Return the validated entry limit.
25    pub const fn get(self) -> usize {
26        self.0.get()
27    }
28}
29
30impl Default for RpcBatchConcurrency {
31    fn default() -> Self {
32        Self(NonZeroUsize::new(128).expect("bug: default concurrency is positive"))
33    }
34}
35
36impl TryFrom<usize> for RpcBatchConcurrency {
37    type Error = InvalidRpcBatchConcurrency;
38
39    fn try_from(value: usize) -> Result<Self, Self::Error> {
40        match NonZeroUsize::new(value) {
41            Some(value) if value.get() <= 1024 => Ok(Self(value)),
42            _ => Err(InvalidRpcBatchConcurrency),
43        }
44    }
45}
46
47impl FromStr for RpcBatchConcurrency {
48    type Err = InvalidRpcBatchConcurrency;
49
50    fn from_str(value: &str) -> Result<Self, Self::Err> {
51        value
52            .parse::<usize>()
53            .map_err(|_| InvalidRpcBatchConcurrency)?
54            .try_into()
55    }
56}
57
58/// An RPC array concurrency setting outside the supported resource bound.
59#[derive(Debug, thiserror::Error)]
60#[error("RPC batch concurrency must be in 1..=1024")]
61pub struct InvalidRpcBatchConcurrency;
62
63/// Bounded concurrent dispatch for every nonempty JSON-RPC array.
64///
65/// Execution order is unspecified; responses and their extensions retain request order.
66/// Notifications delegate to the inner notification service and produce no array entry.
67/// Empty arrays and single calls retain the inner service's behavior.
68/// Completed responses share the array's byte budget, even while awaiting earlier entries.
69/// Overflow cancels outstanding dispatch without waiting for the head of the array.
70/// Cancellation does not roll back side effects already accepted by a backend.
71#[derive(Clone, Debug)]
72pub struct ConcurrentBatchService<S> {
73    inner: S,
74    max_response_body_size: u32,
75    concurrency: RpcBatchConcurrency,
76}
77
78impl<S> ConcurrentBatchService<S> {
79    /// Wrap a service using the same response limit as its jsonrpsee `ServerConfig`.
80    pub fn new(inner: S, max_response_body_size: u32, concurrency: RpcBatchConcurrency) -> Self {
81        Self {
82            inner,
83            max_response_body_size,
84            concurrency,
85        }
86    }
87}
88
89impl<S> RpcServiceT for ConcurrentBatchService<S>
90where
91    S: RpcServiceT<
92            MethodResponse = MethodResponse,
93            BatchResponse = MethodResponse,
94            NotificationResponse = MethodResponse,
95        > + Clone
96        + Send
97        + Sync
98        + 'static,
99{
100    type MethodResponse = MethodResponse;
101    type BatchResponse = MethodResponse;
102    type NotificationResponse = MethodResponse;
103
104    fn call<'a>(&self, req: Request<'a>) -> impl Future<Output = MethodResponse> + Send + 'a {
105        self.inner.call(req)
106    }
107
108    fn notification<'a>(
109        &self,
110        notification: Notification<'a>,
111    ) -> impl Future<Output = MethodResponse> + Send + 'a {
112        self.inner.notification(notification)
113    }
114
115    fn batch<'a>(&self, batch: Batch<'a>) -> impl Future<Output = MethodResponse> + Send + 'a {
116        let inner = self.inner.clone();
117        let max_response_body_size = self.max_response_body_size as usize;
118        let concurrency = self.concurrency.get();
119        async move {
120            if batch.is_empty() {
121                return inner.batch(batch).await;
122            }
123
124            // Keep futures lazy and bounded, including completed responses blocked by
125            // an earlier entry. No detached work survives dropping this batch future.
126            let mut entries = batch.into_iter().enumerate();
127            let mut responses = FuturesUnordered::new();
128            let mut completed = BTreeMap::new();
129            let mut next_output = 0;
130            // Match BatchResponseBuilder: '[' plus each JSON response and ',' (or final ']').
131            // Never release bytes on ordered emission: the builder still retains them.
132            let mut response_bytes = 1usize;
133            let mut builder = BatchResponseBuilder::new_with_limit(max_response_body_size);
134            loop {
135                for (index, entry) in entries
136                    .by_ref()
137                    .take(concurrency.saturating_sub(responses.len() + completed.len()))
138                {
139                    let inner = &inner;
140                    responses.push(async move {
141                        let response = match entry {
142                            Ok(BatchEntry::Call(req)) => Some(inner.call(req).await),
143                            Ok(BatchEntry::Notification(notification)) => {
144                                inner.notification(notification).await;
145                                None
146                            }
147                            Err(err) => {
148                                let (err, id) = err.into_parts();
149                                Some(MethodResponse::error(id, err))
150                            }
151                        };
152                        (index, response)
153                    });
154                }
155                let Some((index, response)) = responses.next().await else {
156                    break;
157                };
158                if let Some(response) = &response {
159                    response_bytes = response_bytes
160                        .saturating_add(response.as_json().get().len())
161                        .saturating_add(1);
162                    if response_bytes > max_response_body_size {
163                        return MethodResponse::error(
164                            Id::Null,
165                            reject_too_big_batch_response(max_response_body_size),
166                        );
167                    }
168                }
169                completed.insert(index, response);
170                while let Some(response) = completed.remove(&next_output) {
171                    next_output += 1;
172                    if let Some(response) = response {
173                        // Append in request order to preserve JSON and extension precedence.
174                        if let Err(err) = builder.append(response) {
175                            return err;
176                        }
177                    }
178                }
179            }
180            if builder.is_empty() {
181                MethodResponse::notification()
182            } else {
183                MethodResponse::from_batch(builder.finish())
184            }
185        }
186    }
187}
188
189#[cfg(test)]
190pub(crate) mod tests {
191    use std::sync::{
192        Arc, Mutex,
193        atomic::{AtomicUsize, Ordering},
194    };
195
196    use jsonrpsee::{
197        core::{middleware::BatchEntryErr, server::ResponsePayload},
198        types::{ErrorObject, Id, error::ErrorCode},
199    };
200    use serde_json::{Value, json};
201    use tokio::sync::Semaphore;
202
203    use super::*;
204
205    const SEND: &str = "sendTransaction";
206    const RECEIPT: &str = "getTransactionReceipt";
207    const LIMIT: u32 = 4096;
208    const BATCH_CONCURRENCY: usize = 4;
209
210    #[derive(Default)]
211    struct State {
212        started: AtomicUsize,
213        active: AtomicUsize,
214        peak: AtomicUsize,
215        completed: Mutex<Vec<u64>>,
216        batches: AtomicUsize,
217        notifications: AtomicUsize,
218    }
219
220    #[derive(Clone)]
221    struct Probe {
222        state: Arc<State>,
223        gates: Arc<Vec<Semaphore>>,
224        limit: u32,
225        payload_bytes: usize,
226    }
227
228    impl Probe {
229        fn new(count: usize, limit: u32) -> Self {
230            Self {
231                state: Arc::default(),
232                gates: Arc::new((0..count).map(|_| Semaphore::new(0)).collect()),
233                limit,
234                payload_bytes: 0,
235            }
236        }
237    }
238
239    struct Active(Arc<State>);
240
241    impl Drop for Active {
242        fn drop(&mut self) {
243            self.0.active.fetch_sub(1, Ordering::SeqCst);
244        }
245    }
246
247    impl RpcServiceT for Probe {
248        type MethodResponse = MethodResponse;
249        type BatchResponse = MethodResponse;
250        type NotificationResponse = MethodResponse;
251
252        fn call<'a>(&self, req: Request<'a>) -> impl Future<Output = MethodResponse> + Send + 'a {
253            let this = self.clone();
254            async move {
255                let index = *req.id().as_number().unwrap();
256                this.state.started.fetch_add(1, Ordering::SeqCst);
257                let active = this.state.active.fetch_add(1, Ordering::SeqCst) + 1;
258                this.state.peak.fetch_max(active, Ordering::SeqCst);
259                let _active = Active(this.state.clone());
260                this.gates[usize::try_from(index).unwrap()]
261                    .acquire()
262                    .await
263                    .unwrap()
264                    .forget();
265                this.state.completed.lock().unwrap().push(index);
266                let mut response = if this.payload_bytes == 0 {
267                    MethodResponse::response(
268                        req.id(),
269                        ResponsePayload::success(index),
270                        this.limit as usize,
271                    )
272                } else {
273                    MethodResponse::response(
274                        req.id(),
275                        ResponsePayload::success("x".repeat(this.payload_bytes)),
276                        this.limit as usize,
277                    )
278                };
279                response.extensions_mut().insert(index);
280                response
281            }
282        }
283
284        fn notification<'a>(
285            &self,
286            _: Notification<'a>,
287        ) -> impl Future<Output = MethodResponse> + Send + 'a {
288            self.state.notifications.fetch_add(1, Ordering::SeqCst);
289            std::future::ready(MethodResponse::notification())
290        }
291
292        fn batch<'a>(&self, batch: Batch<'a>) -> impl Future<Output = MethodResponse> + Send + 'a {
293            let this = self.clone();
294            this.state.batches.fetch_add(1, Ordering::SeqCst);
295            async move {
296                let mut builder = BatchResponseBuilder::new_with_limit(this.limit as usize);
297                let mut notification = false;
298                for entry in batch {
299                    let response = match entry {
300                        Ok(BatchEntry::Call(req)) => this.call(req).await,
301                        Ok(BatchEntry::Notification(n)) => {
302                            notification = true;
303                            this.notification(n).await;
304                            continue;
305                        }
306                        Err(err) => {
307                            let (err, id) = err.into_parts();
308                            MethodResponse::error(id, err)
309                        }
310                    };
311                    if let Err(err) = builder.append(response) {
312                        return err;
313                    }
314                }
315                if notification && builder.is_empty() {
316                    MethodResponse::notification()
317                } else {
318                    MethodResponse::from_batch(builder.finish())
319                }
320            }
321        }
322    }
323
324    fn calls(method: &'static str, count: usize) -> Batch<'static> {
325        Batch::from(
326            (0..count)
327                .map(|id| {
328                    Ok(BatchEntry::Call(Request::borrowed(
329                        method,
330                        None,
331                        Id::Number(u64::try_from(id).unwrap()),
332                    )))
333                })
334                .collect(),
335        )
336    }
337
338    fn value(response: &MethodResponse) -> Value {
339        serde_json::from_str(response.as_json().get()).unwrap()
340    }
341
342    #[test]
343    fn concurrency_defaults_to_128_and_validates_all_construction_paths() {
344        assert_eq!(RpcBatchConcurrency::default().get(), 128);
345        for value in [0, 1025, usize::MAX] {
346            assert!(RpcBatchConcurrency::try_from(value).is_err());
347            assert!(value.to_string().parse::<RpcBatchConcurrency>().is_err());
348            assert!(serde_json::from_value::<RpcBatchConcurrency>(json!(value)).is_err());
349        }
350        for value in [1, 128, 1024] {
351            assert_eq!(RpcBatchConcurrency::try_from(value).unwrap().get(), value);
352            assert_eq!(
353                value
354                    .to_string()
355                    .parse::<RpcBatchConcurrency>()
356                    .unwrap()
357                    .get(),
358                value
359            );
360            assert_eq!(
361                serde_json::from_value::<RpcBatchConcurrency>(json!(value))
362                    .unwrap()
363                    .get(),
364                value
365            );
366        }
367    }
368
369    #[tokio::test]
370    async fn homogeneous_calls_bound_active_and_buffered_work_and_preserve_order() {
371        for (method, concurrency) in [SEND, RECEIPT, "getChainTip", "unknown"]
372            .into_iter()
373            .flat_map(|method| [1, 4, 128, 1024].map(|window| (method, window)))
374        {
375            let count = concurrency * 2 + 3;
376            let limit = u32::try_from(count * 64).unwrap();
377            let probe = Probe::new(count, limit);
378            let service =
379                ConcurrentBatchService::new(probe.clone(), limit, concurrency.try_into().unwrap());
380            let mut response = Box::pin(service.batch(calls(method, count)));
381            // FuturesUnordered may yield before polling the full window.
382            std::future::poll_fn(|cx| {
383                assert!(response.as_mut().poll(cx).is_pending());
384                if probe.state.active.load(Ordering::SeqCst) == concurrency {
385                    std::task::Poll::Ready(())
386                } else {
387                    std::task::Poll::Pending
388                }
389            })
390            .await;
391
392            // Complete the rest of the window in reverse order, leaving its head blocked.
393            for index in (1..concurrency).rev() {
394                probe.gates[index].add_permits(1);
395                std::future::poll_fn(|cx| {
396                    assert!(response.as_mut().poll(cx).is_pending());
397                    if probe.state.active.load(Ordering::SeqCst) == index {
398                        std::task::Poll::Ready(())
399                    } else {
400                        std::task::Poll::Pending
401                    }
402                })
403                .await;
404            }
405            assert_eq!(probe.state.started.load(Ordering::SeqCst), concurrency);
406            assert_eq!(probe.state.active.load(Ordering::SeqCst), 1);
407            assert_eq!(
408                *probe.state.completed.lock().unwrap(),
409                (1..u64::try_from(concurrency).unwrap())
410                    .rev()
411                    .collect::<Vec<_>>()
412            );
413
414            for gate in probe.gates.iter() {
415                gate.add_permits(1);
416            }
417            let response = response.await;
418            let expected: Vec<_> = (0..count)
419                .map(|id| json!({"jsonrpc":"2.0", "id":id, "result":id}))
420                .collect();
421            assert_eq!(value(&response), json!(expected));
422            assert_eq!(probe.state.started.load(Ordering::SeqCst), count);
423            assert_eq!(probe.state.peak.load(Ordering::SeqCst), concurrency);
424            assert_eq!(probe.state.batches.load(Ordering::SeqCst), 0);
425        }
426    }
427
428    #[tokio::test]
429    async fn dropping_batch_cancels_inflight_calls_without_starting_queued_calls() {
430        let probe = Probe::new(BATCH_CONCURRENCY + 1, LIMIT);
431        let service = ConcurrentBatchService::new(
432            probe.clone(),
433            LIMIT,
434            BATCH_CONCURRENCY.try_into().unwrap(),
435        );
436        let mut response = Box::pin(service.batch(calls(SEND, BATCH_CONCURRENCY + 1)));
437        assert!(futures_util::poll!(&mut response).is_pending());
438        drop(response);
439        assert_eq!(probe.state.active.load(Ordering::SeqCst), 0);
440        assert_eq!(
441            probe.state.started.load(Ordering::SeqCst),
442            BATCH_CONCURRENCY
443        );
444        assert!(probe.state.completed.lock().unwrap().is_empty());
445    }
446
447    #[tokio::test]
448    async fn invalid_entries_preserve_error_ids_and_positions() {
449        let probe = Probe::new(2, LIMIT);
450        for gate in probe.gates.iter() {
451            gate.add_permits(1);
452        }
453        let mut batch = calls(RECEIPT, 2).into_iter().collect::<Vec<_>>();
454        batch.insert(
455            0,
456            Err(BatchEntryErr::new(
457                Id::Null,
458                ErrorCode::InvalidRequest.into(),
459            )),
460        );
461        batch.insert(
462            2,
463            Err(BatchEntryErr::new(
464                Id::Str("bad".into()),
465                ErrorCode::InvalidParams.into(),
466            )),
467        );
468        let response =
469            ConcurrentBatchService::new(probe.clone(), LIMIT, RpcBatchConcurrency::default())
470                .batch(Batch::from(batch))
471                .await;
472        assert_eq!(
473            value(&response),
474            json!([
475                {"jsonrpc":"2.0","id":null,"error":{"code":-32600,"message":"Invalid request"}},
476                {"jsonrpc":"2.0","id":0,"result":0},
477                {"jsonrpc":"2.0","id":"bad","error":{"code":-32602,"message":"Invalid params"}},
478                {"jsonrpc":"2.0","id":1,"result":1}
479            ])
480        );
481        assert_eq!(probe.state.batches.load(Ordering::SeqCst), 0);
482    }
483
484    #[tokio::test]
485    async fn overflow_matches_sequential_error_and_drops_remaining_window() {
486        let probe = Probe::new(BATCH_CONCURRENCY * 2, 60);
487        let service =
488            ConcurrentBatchService::new(probe.clone(), 60, BATCH_CONCURRENCY.try_into().unwrap());
489        let mut response = Box::pin(service.batch(calls(SEND, BATCH_CONCURRENCY * 2)));
490        assert!(futures_util::poll!(&mut response).is_pending());
491        probe.gates[0].add_permits(1);
492        probe.gates[1].add_permits(1);
493        let response = response.await;
494        let baseline = Probe::new(2, 60);
495        for gate in baseline.gates.iter() {
496            gate.add_permits(1);
497        }
498        let expected = baseline.batch(calls(SEND, 2)).await;
499        assert_eq!(response.as_json().get(), expected.as_json().get());
500        assert_eq!(response.as_error_code(), Some(-32011));
501        assert_eq!(probe.state.active.load(Ordering::SeqCst), 0);
502        assert!(probe.state.started.load(Ordering::SeqCst) <= BATCH_CONCURRENCY + 1);
503        assert_eq!(*probe.state.completed.lock().unwrap(), vec![0, 1]);
504    }
505
506    #[tokio::test]
507    async fn large_completed_receipts_overflow_before_head_finishes_and_cancel_outstanding_calls() {
508        let limit = jsonrpsee::core::TEN_MB_SIZE_BYTES;
509        let mut probe = Probe::new(BATCH_CONCURRENCY + 1, limit);
510        probe.payload_bytes = 9 * 1024 * 1024;
511        let service = ConcurrentBatchService::new(
512            probe.clone(),
513            limit,
514            BATCH_CONCURRENCY.try_into().unwrap(),
515        );
516        let mut response = Box::pin(service.batch(calls(RECEIPT, BATCH_CONCURRENCY + 1)));
517        assert!(futures_util::poll!(&mut response).is_pending());
518        assert_eq!(probe.state.active.load(Ordering::SeqCst), BATCH_CONCURRENCY);
519
520        probe.gates[1].add_permits(1);
521        assert!(futures_util::poll!(&mut response).is_pending());
522        probe.gates[2].add_permits(1);
523        let std::task::Poll::Ready(response) = futures_util::poll!(&mut response) else {
524            panic!("overflow must not wait for the gated first receipt");
525        };
526        let expected =
527            MethodResponse::error(Id::Null, reject_too_big_batch_response(limit as usize));
528        assert_eq!(response.as_json().get(), expected.as_json().get());
529        assert_eq!(probe.state.active.load(Ordering::SeqCst), 0);
530        assert_eq!(
531            probe.state.started.load(Ordering::SeqCst),
532            BATCH_CONCURRENCY
533        );
534        assert_eq!(*probe.state.completed.lock().unwrap(), vec![1, 2]);
535    }
536
537    #[tokio::test]
538    async fn completed_byte_budget_includes_already_emitted_prefix() {
539        let probe = Probe::new(4, 90);
540        let service =
541            ConcurrentBatchService::new(probe.clone(), 90, BATCH_CONCURRENCY.try_into().unwrap());
542        let mut response = Box::pin(service.batch(calls(RECEIPT, 4)));
543        assert!(futures_util::poll!(&mut response).is_pending());
544        for index in [0, 2] {
545            probe.gates[index].add_permits(1);
546            assert!(futures_util::poll!(&mut response).is_pending());
547        }
548        probe.gates[3].add_permits(1);
549        let std::task::Poll::Ready(response) = futures_util::poll!(&mut response) else {
550            panic!("ordered emission must not release the prefix's byte budget");
551        };
552        assert_eq!(response.as_error_code(), Some(-32011));
553        assert_eq!(probe.state.active.load(Ordering::SeqCst), 0);
554        assert_eq!(*probe.state.completed.lock().unwrap(), vec![0, 2, 3]);
555    }
556
557    #[tokio::test]
558    async fn byte_budget_matches_builder_boundary_and_preserves_ordered_extensions() {
559        let baseline = Probe::new(3, LIMIT);
560        for gate in baseline.gates.iter() {
561            gate.add_permits(3);
562        }
563        let expected = baseline.batch(calls(RECEIPT, 3)).await;
564        let exact_limit = u32::try_from(expected.as_json().get().len()).unwrap();
565        for limit in [exact_limit, exact_limit - 1] {
566            let probe = Probe::new(3, limit);
567            let service = ConcurrentBatchService::new(
568                probe.clone(),
569                limit,
570                BATCH_CONCURRENCY.try_into().unwrap(),
571            );
572            let mut response = Box::pin(service.batch(calls(RECEIPT, 3)));
573            assert!(futures_util::poll!(&mut response).is_pending());
574            for index in [2, 1] {
575                probe.gates[index].add_permits(1);
576                assert!(futures_util::poll!(&mut response).is_pending());
577            }
578            probe.gates[0].add_permits(1);
579            let response = response.await;
580            let baseline = Probe {
581                limit,
582                ..baseline.clone()
583            };
584            let expected = baseline.batch(calls(RECEIPT, 3)).await;
585            assert_eq!(response.as_json().get(), expected.as_json().get());
586            assert_eq!(
587                response.extensions().get::<u64>(),
588                expected.extensions().get::<u64>()
589            );
590            if limit == exact_limit {
591                assert_eq!(response.extensions().get::<u64>(), Some(&2));
592            } else {
593                assert_eq!(response.as_error_code(), Some(-32011));
594            }
595        }
596    }
597
598    #[tokio::test]
599    async fn mixed_methods_and_notifications_use_concurrent_dispatch() {
600        for other in [RECEIPT, "getChainTip", "unknown", SEND] {
601            let probe = Probe::new(2, LIMIT);
602            let service =
603                ConcurrentBatchService::new(probe.clone(), LIMIT, RpcBatchConcurrency::default());
604            let mut batch = calls(SEND, 1);
605            if other == SEND {
606                let mut entries = batch.into_iter().collect::<Vec<_>>();
607                entries.push(Ok(BatchEntry::Notification(
608                    serde_json::from_str(r#"{"jsonrpc":"2.0","method":"sendTransaction"}"#)
609                        .unwrap(),
610                )));
611                batch = Batch::from(entries);
612            } else {
613                batch.push(Request::borrowed(other, None, Id::Number(1)));
614            }
615            let mut response = Box::pin(service.batch(batch));
616            assert!(futures_util::poll!(&mut response).is_pending());
617            assert_eq!(probe.state.batches.load(Ordering::SeqCst), 0);
618            let calls = if other == SEND { 1 } else { 2 };
619            assert_eq!(probe.state.started.load(Ordering::SeqCst), calls);
620            assert_eq!(
621                probe.state.notifications.load(Ordering::SeqCst),
622                usize::from(other == SEND)
623            );
624            for gate in probe.gates.iter() {
625                gate.add_permits(1);
626            }
627            response.await;
628            assert_eq!(probe.state.peak.load(Ordering::SeqCst), calls);
629            assert_eq!(
630                probe.state.notifications.load(Ordering::SeqCst),
631                usize::from(other == SEND)
632            );
633        }
634    }
635
636    #[tokio::test]
637    async fn only_empty_arrays_delegate_and_preserve_invalid_request() {
638        let probe = Probe::new(0, LIMIT);
639        let service =
640            ConcurrentBatchService::new(probe.clone(), LIMIT, RpcBatchConcurrency::default());
641        let empty = service.batch(Batch::new()).await;
642        let baseline = Probe::new(0, LIMIT).batch(Batch::new()).await;
643        assert_eq!(empty.as_json().get(), baseline.as_json().get());
644        assert_eq!(value(&empty)["error"]["code"], -32600);
645        let invalid = service
646            .batch(Batch::from(vec![Err(BatchEntryErr::new(
647                Id::Null,
648                ErrorObject::from(ErrorCode::InvalidRequest),
649            ))]))
650            .await;
651        assert_eq!(value(&invalid)[0]["error"]["code"], -32600);
652        let notification = service
653            .batch(Batch::from(vec![Ok(BatchEntry::Notification(
654                serde_json::from_str(r#"{"jsonrpc":"2.0","method":"sendTransaction"}"#).unwrap(),
655            ))]))
656            .await;
657        assert!(notification.is_notification());
658        assert_eq!(probe.state.batches.load(Ordering::SeqCst), 1);
659        assert_eq!(probe.state.started.load(Ordering::SeqCst), 0);
660        assert_eq!(probe.state.notifications.load(Ordering::SeqCst), 1);
661    }
662
663    pub(crate) async fn http_request(address: std::net::SocketAddr, body: &str) -> String {
664        use tokio::io::{AsyncReadExt, AsyncWriteExt};
665
666        tokio::time::timeout(std::time::Duration::from_secs(5), async {
667            let mut socket = tokio::net::TcpStream::connect(address).await.unwrap();
668            let request = format!(
669                "POST / HTTP/1.1\r\nHost: {address}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
670                body.len()
671            );
672            socket.write_all(request.as_bytes()).await.unwrap();
673            let mut response = String::new();
674            socket.read_to_string(&mut response).await.unwrap();
675            response.split_once("\r\n\r\n").unwrap().1.to_owned()
676        }).await.unwrap()
677    }
678
679    async fn start_server(
680        concurrent: bool,
681        limit: u32,
682    ) -> (
683        std::net::SocketAddr,
684        jsonrpsee::server::ServerHandle,
685        Arc<AtomicUsize>,
686    ) {
687        let invoked = Arc::new(AtomicUsize::new(0));
688        let mut module = jsonrpsee::RpcModule::new(invoked.clone());
689        for method in [SEND, RECEIPT, "getChainTip"] {
690            module
691                .register_async_method(method, |params, invoked, _| async move {
692                    invoked.fetch_add(1, Ordering::SeqCst);
693                    let (result,): (Value,) = params.parse()?;
694                    if result == "fail" {
695                        Err(ErrorObject::owned(-32000, "backend failure", None::<()>))
696                    } else {
697                        Ok(result)
698                    }
699                })
700                .unwrap();
701        }
702        let config = jsonrpsee::server::ServerConfig::builder()
703            .max_response_body_size(limit)
704            .set_batch_request_config(jsonrpsee::server::BatchRequestConfig::Limit(40))
705            .build();
706        if concurrent {
707            let middleware =
708                jsonrpsee::core::middleware::RpcServiceBuilder::new().layer_fn(move |inner| {
709                    ConcurrentBatchService::new(inner, limit, RpcBatchConcurrency::default())
710                });
711            let server = jsonrpsee::server::Server::builder()
712                .set_config(config)
713                .set_rpc_middleware(middleware)
714                .build("127.0.0.1:0")
715                .await
716                .unwrap();
717            (server.local_addr().unwrap(), server.start(module), invoked)
718        } else {
719            let server = jsonrpsee::server::Server::builder()
720                .set_config(config)
721                .build("127.0.0.1:0")
722                .await
723                .unwrap();
724            (server.local_addr().unwrap(), server.start(module), invoked)
725        }
726    }
727
728    #[tokio::test]
729    async fn http_responses_match_jsonrpsee_for_mixed_invalid_and_overflow_inputs() {
730        let call = |method, id, params| json!({"jsonrpc":"2.0", "method":method, "id":id, "params":params});
731        let send = call(SEND, json!("same"), json!([7]));
732        let receipt = call(RECEIPT, Value::Null, json!([null]));
733        let notification = json!({"jsonrpc":"2.0", "method":SEND, "params":[7]});
734        let tip = call("getChainTip", json!(2), json!([true]));
735        let unknown = call("unknown", json!(3), json!([]));
736        let cases = vec![
737            "[".to_owned(),
738            json!([]).to_string(),
739            json!([null, 7, {}, []]).to_string(),
740            send.to_string(),
741            notification.to_string(),
742            json!([send]).to_string(),
743            json!([send, send]).to_string(),
744            json!([receipt, receipt]).to_string(),
745            json!([null, send, 7, {}, send, []]).to_string(),
746            json!([send, call(SEND, json!(2), json!("bad params")), send]).to_string(),
747            json!([send, call(SEND, json!(2), json!([])), send]).to_string(),
748            json!([send, call(SEND, json!(2), json!(["fail"])), send]).to_string(),
749            json!([send, call(SEND, json!({}), json!([7])), send]).to_string(),
750            json!([call(SEND, json!(1), json!(["x".repeat(5000)])), send]).to_string(),
751            json!([notification, notification]).to_string(),
752            json!([send, notification, send]).to_string(),
753            json!([send, receipt]).to_string(),
754            json!([send, tip, send]).to_string(),
755            json!([tip, tip]).to_string(),
756            json!([send, notification, null, unknown, receipt, tip]).to_string(),
757            json!([unknown, unknown]).to_string(),
758            json!([send, unknown]).to_string(),
759            json!([notification, null]).to_string(),
760            json!(vec![send; 41]).to_string(),
761        ];
762        for limit in [60, LIMIT] {
763            let (address, handle, invoked) = start_server(true, limit).await;
764            let (baseline, baseline_handle, baseline_invoked) = start_server(false, limit).await;
765            for body in &cases {
766                let actual = http_request(address, body).await;
767                let expected = http_request(baseline, body).await;
768                assert_eq!(actual, expected, "limit={limit}, request={body}");
769            }
770            // A notification-only request must not invoke registered handlers.
771            let before = invoked.load(Ordering::SeqCst);
772            let notifications = json!([notification]).to_string();
773            assert_eq!(
774                http_request(address, &notifications).await,
775                http_request(baseline, &notifications).await
776            );
777            assert_eq!(invoked.load(Ordering::SeqCst), before);
778            assert!(baseline_invoked.load(Ordering::SeqCst) > 0);
779            handle.stop().unwrap();
780            baseline_handle.stop().unwrap();
781            handle.stopped().await;
782            baseline_handle.stopped().await;
783        }
784    }
785}
thruput_json_rpc_server/batch.rs

thruput_json_rpc_server/
batch.rs
