forest/rpc/
compression_layer.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4//! HTTP response compression middleware that normalizes the response body
5//! back to `jsonrpsee`'s [`HttpBody`] so the layer can be conditionally
6//! installed via [`tower::util::option_layer`].
7//!
8//! Ported from `reth_rpc_layer::compression_layer`.
9
10use std::{
11    env,
12    future::Future,
13    pin::Pin,
14    sync::LazyLock,
15    task::{Context, Poll},
16};
17
18use jsonrpsee::server::{HttpBody, HttpRequest, HttpResponse};
19use tower::{Layer, Service};
20use tower_http::compression::predicate::SizeAbove;
21use tower_http::compression::{Compression, CompressionLayer as TowerCompressionLayer};
22
23const COMPRESS_MIN_BODY_SIZE_VAR: &str = "FOREST_RPC_COMPRESS_MIN_BODY_SIZE";
24
25/// RPC response compression policy, read from [`COMPRESS_MIN_BODY_SIZE_VAR`].
26///
27/// `None` means no [`CompressionLayer`] is installed at all; `Some(bytes)`
28/// means install a layer that compresses responses whose body is at least
29/// `bytes`.
30///
31/// - Any value in `0..=u64::MAX` sets the minimum response size that will
32///   be gzip-encoded; smaller responses are sent as-is. Values above
33///   `u64::MAX` are clamped because `SizeAbove` is backed by a `u64`.
34/// - Any negative integer (e.g. `-1`) disables compression entirely.
35/// - Unset defaults to disabled.
36pub(crate) static COMPRESS_MIN_BODY_SIZE: LazyLock<Option<u64>> = LazyLock::new(|| {
37    parse_compress_min_body_size(env::var(COMPRESS_MIN_BODY_SIZE_VAR).ok().as_deref())
38});
39
40/// Interpret a [`COMPRESS_MIN_BODY_SIZE_VAR`] value.
41///
42/// Returns `None` to signal "compression disabled", `Some(bytes)` for the
43/// minimum response size above which compression should be applied.
44/// Unset and unparsable values fall back to disabled.
45/// Values above `u64::MAX` are clamped to `u64::MAX`.
46fn parse_compress_min_body_size(raw: Option<&str>) -> Option<u64> {
47    // Parse as i128 so any realistically-typable integer lands in one of the
48    // defined branches (negative → disabled, too-large → clamp) rather than
49    // silently disabling just because it didn't fit in i32.
50    let Ok(parsed) = raw?.parse::<i128>() else {
51        tracing::warn!(
52            "{COMPRESS_MIN_BODY_SIZE_VAR}={raw:?} is not a valid integer; \
53             disabling compression"
54        );
55        return None;
56    };
57    if parsed < 0 {
58        return None;
59    }
60    let max = i128::from(u64::MAX);
61    if parsed > max {
62        tracing::warn!(
63            "{COMPRESS_MIN_BODY_SIZE_VAR}={parsed} exceeds the maximum of {max}; \
64             clamping to {max} bytes"
65        );
66    }
67    // The prior branches bound `parsed.min(max)` to `[0, u64::MAX]`.
68    Some(u64::try_from(parsed.min(max)).expect("bounded above to u64::MAX"))
69}
70
71/// Compresses responses with a body above `min_body_size` bytes.
72#[derive(Clone)]
73pub(crate) struct CompressionLayer {
74    inner: TowerCompressionLayer<SizeAbove>,
75}
76
77impl CompressionLayer {
78    /// Compress responses whose body is at least `min_body_size` bytes.
79    pub(crate) fn new(min_body_size: u64) -> Self {
80        Self {
81            inner: TowerCompressionLayer::new().compress_when(SizeAbove::new(min_body_size)),
82        }
83    }
84}
85
86impl<S> Layer<S> for CompressionLayer {
87    type Service = CompressionService<S>;
88
89    fn layer(&self, inner: S) -> Self::Service {
90        CompressionService {
91            inner: self.inner.layer(inner),
92        }
93    }
94}
95
96#[derive(Clone)]
97pub(crate) struct CompressionService<S> {
98    inner: Compression<S, SizeAbove>,
99}
100
101impl<S, ReqBody> Service<HttpRequest<ReqBody>> for CompressionService<S>
102where
103    S: Service<HttpRequest<ReqBody>, Response = HttpResponse>,
104    S::Future: Send + 'static,
105{
106    type Response = HttpResponse;
107    type Error = S::Error;
108    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
109
110    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
111        self.inner.poll_ready(cx)
112    }
113
114    fn call(&mut self, req: HttpRequest<ReqBody>) -> Self::Future {
115        let fut = self.inner.call(req);
116        Box::pin(async move {
117            // Re-box to match `Identity`'s response body type (see module doc).
118            let resp = fut.await?;
119            let (parts, compressed_body) = resp.into_parts();
120            Ok(Self::Response::from_parts(
121                parts,
122                HttpBody::new(compressed_body),
123            ))
124        })
125    }
126}
127
128#[cfg(test)]
129mod tests {
130    use super::*;
131    use http::header::{ACCEPT_ENCODING, CONTENT_ENCODING};
132    use std::{convert::Infallible, future::ready};
133
134    const TEST_DATA: &str = "cthulhu fhtagn ";
135    const REPEAT_COUNT: usize = 1000;
136
137    #[derive(Clone)]
138    struct MockService;
139
140    impl Service<HttpRequest> for MockService {
141        type Response = HttpResponse;
142        type Error = Infallible;
143        type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
144
145        fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
146            Poll::Ready(Ok(()))
147        }
148
149        fn call(&mut self, _: HttpRequest) -> Self::Future {
150            let body = HttpBody::from(TEST_DATA.repeat(REPEAT_COUNT));
151            ready(Ok(HttpResponse::builder().body(body).unwrap()))
152        }
153    }
154
155    async fn body_size(resp: HttpResponse) -> usize {
156        let body = axum::body::Body::new(resp.into_body());
157        axum::body::to_bytes(body, usize::MAX).await.unwrap().len()
158    }
159
160    fn uncompressed_size() -> usize {
161        TEST_DATA.repeat(REPEAT_COUNT).len()
162    }
163
164    #[tokio::test]
165    async fn gzip_compresses_when_requested() {
166        let mut svc = CompressionLayer::new(0).layer(MockService);
167        let req = HttpRequest::builder()
168            .header(ACCEPT_ENCODING, "gzip")
169            .body(HttpBody::empty())
170            .unwrap();
171        let resp = svc.call(req).await.unwrap();
172        assert_eq!(resp.headers().get(CONTENT_ENCODING).unwrap(), "gzip");
173        assert!(body_size(resp).await < uncompressed_size());
174    }
175
176    #[tokio::test]
177    async fn passthrough_when_encoding_not_requested() {
178        let mut svc = CompressionLayer::new(0).layer(MockService);
179        let req = HttpRequest::builder().body(HttpBody::empty()).unwrap();
180        let resp = svc.call(req).await.unwrap();
181        assert!(resp.headers().get(CONTENT_ENCODING).is_none());
182        assert_eq!(body_size(resp).await, uncompressed_size());
183    }
184
185    #[tokio::test]
186    async fn below_threshold_is_not_compressed() {
187        let mut svc = CompressionLayer::new(u64::MAX).layer(MockService);
188        let req = HttpRequest::builder()
189            .header(ACCEPT_ENCODING, "gzip")
190            .body(HttpBody::empty())
191            .unwrap();
192        let resp = svc.call(req).await.unwrap();
193        assert!(resp.headers().get(CONTENT_ENCODING).is_none());
194        assert_eq!(body_size(resp).await, uncompressed_size());
195    }
196
197    #[test]
198    fn parse_unset_disables() {
199        assert_eq!(parse_compress_min_body_size(None), None);
200    }
201
202    #[test]
203    fn parse_negative_disables() {
204        assert_eq!(parse_compress_min_body_size(Some("-1")), None);
205        assert_eq!(parse_compress_min_body_size(Some("-999999")), None);
206        assert_eq!(parse_compress_min_body_size(Some("-2147483648")), None); // i32::MIN
207        // Values below i32::MIN must still disable rather than fall back.
208        assert_eq!(
209            parse_compress_min_body_size(Some("-9223372036854775808")),
210            None
211        ); // i64::MIN
212    }
213
214    #[test]
215    fn parse_accepts_in_range_values() {
216        assert_eq!(parse_compress_min_body_size(Some("0")), Some(0));
217        assert_eq!(parse_compress_min_body_size(Some("512")), Some(512));
218        assert_eq!(parse_compress_min_body_size(Some("1024")), Some(1024));
219        assert_eq!(
220            parse_compress_min_body_size(Some("18446744073709551615")),
221            Some(u64::MAX)
222        );
223    }
224
225    #[test]
226    fn parse_clamps_above_u64_max() {
227        assert_eq!(
228            parse_compress_min_body_size(Some("18446744073709551616")), // u64::MAX + 1
229            Some(u64::MAX)
230        );
231        assert_eq!(
232            parse_compress_min_body_size(Some("118446744073709551616")),
233            Some(u64::MAX)
234        );
235        assert_eq!(
236            parse_compress_min_body_size(Some("170141183460469231731687303715884105727")), // i128::MAX
237            Some(u64::MAX)
238        );
239    }
240
241    #[test]
242    fn parse_invalid_disables() {
243        assert_eq!(parse_compress_min_body_size(Some("")), None);
244        assert_eq!(parse_compress_min_body_size(Some("lots")), None);
245        assert_eq!(parse_compress_min_body_size(Some("1.5")), None);
246    }
247}
forest/rpc/compression_layer.rs

forest/rpc/
compression_layer.rs
