node, chain: Add extensible compression support for RPC requests

fubhy · lutter · commit c78190d6470a · 2026-02-05T12:24:07.000-08:00
- Replace boolean compression_enabled with Compression enum (None, Gzip) - Support per-provider compression configuration via "compression" field - Add placeholders for future compression methods (Brotli, Deflate) - Update transport layer to handle compression enum with match statement - Add comprehensive unit tests for compression configuration parsing - Update example configuration and documentation Configuration examples: compression = "gzip" # Enable gzip compression compression = "none" # Disable compression (default) Addresses issue #5671 with future-extensible design.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/chain/ethereum/src/network.rs b/chain/ethereum/src/network.rs
@@ -314,6 +314,7 @@ mod tests {
     use graph::components::network_provider::ProviderManager;
     use graph::components::network_provider::ProviderName;
     use graph::data::value::Word;
+    use graph::endpoint::Compression;
     use graph::http::HeaderMap;
     use graph::{
         endpoint::EndpointMetrics,
@@ -395,6 +396,7 @@ mod tests {
             metrics.clone(),
             "",
             false,
+            Compression::None,
         );
         let provider_metrics = Arc::new(ProviderEthRpcMetrics::new(mock_registry.clone()));
 
@@ -499,6 +501,7 @@ mod tests {
             metrics.clone(),
             "",
             false,
+            Compression::None,
         );
         let provider_metrics = Arc::new(ProviderEthRpcMetrics::new(mock_registry.clone()));
 
@@ -571,6 +574,7 @@ mod tests {
             metrics.clone(),
             "",
             false,
+            Compression::None,
         );
         let provider_metrics = Arc::new(ProviderEthRpcMetrics::new(mock_registry.clone()));
 
@@ -636,6 +640,7 @@ mod tests {
             metrics.clone(),
             "",
             false,
+            Compression::None,
         );
         let provider_metrics = Arc::new(ProviderEthRpcMetrics::new(mock_registry.clone()));
 
@@ -924,6 +929,7 @@ mod tests {
             endpoint_metrics.clone(),
             "",
             false,
+            Compression::None,
         );
 
         Arc::new(
diff --git a/chain/ethereum/src/transport.rs b/chain/ethereum/src/transport.rs
@@ -1,6 +1,6 @@
 use alloy::transports::{TransportError, TransportErrorKind, TransportFut};
 use graph::components::network_provider::ProviderName;
-use graph::endpoint::{ConnectionType, EndpointMetrics, RequestLabels};
+use graph::endpoint::{Compression, ConnectionType, EndpointMetrics, RequestLabels};
 use graph::prelude::alloy::rpc::json_rpc::{RequestPacket, ResponsePacket};
 use graph::prelude::alloy::transports::{ipc::IpcConnect, ws::WsConnect};
 use graph::prelude::*;
@@ -49,11 +49,15 @@ impl Transport {
         metrics: Arc<EndpointMetrics>,
         provider: impl AsRef<str>,
         no_eip2718: bool,
+        compression: Compression,
     ) -> Self {
-        let client = reqwest::Client::builder()
-            .default_headers(headers)
-            .build()
-            .expect("Failed to build HTTP client");
+        let mut client_builder = reqwest::Client::builder().default_headers(headers);
+
+        if matches!(compression, Compression::Gzip) {
+            client_builder = client_builder.gzip(true);
+        }
+
+        let client = client_builder.build().expect("Failed to build HTTP client");
 
         let patching_transport = PatchingHttp::new(client, rpc, no_eip2718);
         let metrics_transport =
diff --git a/graph/Cargo.toml b/graph/Cargo.toml
@@ -26,7 +26,7 @@ diesel_derives = { workspace = true }
 chrono = "0.4.43"
 envconfig = { workspace = true }
 Inflector = "0.11.3"
-reqwest = { version = "0.12.23", features = ["json", "stream", "multipart"] }
+reqwest = { version = "0.12.23", features = ["json", "stream", "multipart", "gzip"] }
 ethabi = "17.2"
 hex = "0.4.3"
 http0 = { version = "0", package = "http" }
diff --git a/graph/src/endpoint.rs b/graph/src/endpoint.rs
@@ -7,6 +7,7 @@ use std::{
 };
 
 use prometheus::IntCounterVec;
+use serde::{Deserialize, Serialize};
 use slog::{warn, Logger};
 
 use crate::components::network_provider::ProviderName;
@@ -17,6 +18,16 @@ use crate::{components::metrics::MetricsRegistry, data::value::Word};
 /// avoid locking since we don't need to modify the entire struture.
 type ProviderCount = Arc<HashMap<ProviderName, AtomicU64>>;
 
+/// Compression methods for RPC transports
+#[derive(Copy, Clone, Debug, Default, Deserialize, Serialize, PartialEq)]
+pub enum Compression {
+    #[default]
+    #[serde(rename = "none")]
+    None,
+    #[serde(rename = "gzip")]
+    Gzip,
+}
+
 /// This struct represents all the current labels except for the result
 /// which is added separately. If any new labels are necessary they should
 /// remain in the same order as added in [`EndpointMetrics::new`]
diff --git a/node/resources/tests/full_config.toml b/node/resources/tests/full_config.toml
@@ -48,6 +48,7 @@ shard = "primary"
 provider = [
   { label = "mainnet-0", url = "http://rpc.mainnet.io", features = ["archive", "traces"] },
   { label = "mainnet-1", details = { type = "web3call", url = "http://rpc.mainnet.io", features = ["archive", "traces"] }},
+  { label = "mainnet-2", details = { type = "web3", url = "http://rpc.mainnet.io", features = ["archive"], compression = "gzip" }},
   { label = "firehose", details = { type = "firehose", url = "http://localhost:9000", features = [] }},
 ]
 
diff --git a/node/src/chain.rs b/node/src/chain.rs
@@ -210,7 +210,8 @@ pub async fn create_ethereum_networks_for_chain(
             logger,
             "Creating transport";
             "url" => &web3.url,
-            "capabilities" => capabilities
+            "capabilities" => capabilities,
+            "compression" => ?web3.compression
         );
 
         use crate::config::Transport::*;
@@ -223,6 +224,7 @@ pub async fn create_ethereum_networks_for_chain(
                 endpoint_metrics.cheap_clone(),
                 &provider.label,
                 no_eip2718,
+                web3.compression,
             ),
             Ipc => Transport::new_ipc(&web3.url).await,
             Ws => Transport::new_ws(&web3.url).await,
diff --git a/node/src/config.rs b/node/src/config.rs
@@ -2,6 +2,7 @@ use graph::{
     anyhow::Error,
     blockchain::BlockchainKind,
     components::network_provider::ChainName,
+    endpoint::Compression,
     env::ENV_VARS,
     firehose::{SubgraphLimit, SUBGRAPHS_PER_CONN},
     itertools::Itertools,
@@ -516,6 +517,7 @@ impl ChainSection {
                         features,
                         headers: Default::default(),
                         rules: vec![],
+                        compression: Compression::None,
                     }),
                 };
                 let entry = chains.entry(name.to_string()).or_insert_with(|| Chain {
@@ -694,6 +696,10 @@ pub struct Web3Provider {
 
     #[serde(default, rename = "match")]
     rules: Vec<Web3Rule>,
+
+    /// Compression method for RPC requests and responses
+    #[serde(default)]
+    pub compression: Compression,
 }
 
 impl Web3Provider {
@@ -891,6 +897,7 @@ impl<'de> Deserialize<'de> for Provider {
                             .ok_or_else(|| serde::de::Error::missing_field("features"))?,
                         headers: headers.unwrap_or_else(HeaderMap::new),
                         rules: nodes,
+                        compression: Compression::None,
                     }),
                 };
 
@@ -1203,6 +1210,7 @@ mod tests {
         Chain, Config, FirehoseProvider, Provider, ProviderDetails, Shard, Transport, Web3Provider,
     };
     use graph::blockchain::BlockchainKind;
+    use graph::endpoint::Compression;
     use graph::firehose::SubgraphLimit;
     use graph::http::{HeaderMap, HeaderValue};
     use graph::prelude::regex::Regex;
@@ -1291,6 +1299,7 @@ mod tests {
                     features: BTreeSet::new(),
                     headers: HeaderMap::new(),
                     rules: Vec::new(),
+                    compression: Compression::None,
                 }),
             },
             actual
@@ -1317,6 +1326,7 @@ mod tests {
                     features: BTreeSet::new(),
                     headers: HeaderMap::new(),
                     rules: Vec::new(),
+                    compression: Compression::None,
                 }),
             },
             actual
@@ -1378,6 +1388,7 @@ mod tests {
                     features,
                     headers,
                     rules: Vec::new(),
+                    compression: Compression::None,
                 }),
             },
             actual
@@ -1403,6 +1414,7 @@ mod tests {
                     features: BTreeSet::new(),
                     headers: HeaderMap::new(),
                     rules: Vec::new(),
+                    compression: Compression::None,
                 }),
             },
             actual
@@ -1613,6 +1625,7 @@ mod tests {
                     features: BTreeSet::new(),
                     headers: HeaderMap::new(),
                     rules: Vec::new(),
+                    compression: Compression::None,
                 }),
             },
             actual
@@ -1625,6 +1638,66 @@ mod tests {
         assert!(SubgraphLimit::Limit(10) > SubgraphLimit::Disabled);
     }
 
+    #[test]
+    fn it_parses_web3_provider_with_compression() {
+        let actual = toml::from_str(
+            r#"
+            label = "compressed"
+            details = { type = "web3", url = "http://localhost:8545", features = ["archive"], compression = "gzip" }
+        "#,
+        )
+        .unwrap();
+
+        assert_eq!(
+            Provider {
+                label: "compressed".to_owned(),
+                details: ProviderDetails::Web3(Web3Provider {
+                    transport: Transport::Rpc,
+                    url: "http://localhost:8545".to_owned(),
+                    features: {
+                        let mut features = BTreeSet::new();
+                        features.insert("archive".to_string());
+                        features
+                    },
+                    headers: HeaderMap::new(),
+                    rules: Vec::new(),
+                    compression: Compression::Gzip,
+                }),
+            },
+            actual
+        );
+    }
+
+    #[test]
+    fn it_parses_web3_provider_with_no_compression() {
+        let actual = toml::from_str(
+            r#"
+            label = "uncompressed"
+            details = { type = "web3", url = "http://localhost:8545", features = ["archive"], compression = "none" }
+        "#,
+        )
+        .unwrap();
+
+        assert_eq!(
+            Provider {
+                label: "uncompressed".to_owned(),
+                details: ProviderDetails::Web3(Web3Provider {
+                    transport: Transport::Rpc,
+                    url: "http://localhost:8545".to_owned(),
+                    features: {
+                        let mut features = BTreeSet::new();
+                        features.insert("archive".to_string());
+                        features
+                    },
+                    headers: HeaderMap::new(),
+                    rules: Vec::new(),
+                    compression: Compression::None,
+                }),
+            },
+            actual
+        );
+    }
+
     #[test]
     fn duplicated_labels_are_not_allowed_within_chain() {
         let mut actual = toml::from_str::<ChainSection>(

Original file line number	Diff line number	Diff line change
`@@ -48,6 +48,7 @@ shard = "primary"`
`48`	`48`	`provider = [`
`49`	`49`	`{ label = "mainnet-0", url = "http://rpc.mainnet.io", features = ["archive", "traces"] },`
`50`	`50`	`{ label = "mainnet-1", details = { type = "web3call", url = "http://rpc.mainnet.io", features = ["archive", "traces"] }},`
	`51`	`+ { label = "mainnet-2", details = { type = "web3", url = "http://rpc.mainnet.io", features = ["archive"], compression = "gzip" }},`
`51`	`52`	`{ label = "firehose", details = { type = "firehose", url = "http://localhost:9000", features = [] }},`
`52`	`53`	`]`
`53`	`54`