Add zstd decoding

Cargo.lock

@@ -5290,6 +5290,7 @@ dependencies = [
  "serde_json",
  "solana-account-decoder",
  "solana-client",
+ "solana-geyser-connector-plugin-grpc",
  "solana-rpc",
  "solana-sdk",
  "tokio",

lib/Cargo.toml

@@ -35,7 +35,7 @@ serde = "1.0.130"
 serde_derive = "1.0.130"
 serde_json = "1.0.68"
 
-tonic = { version = "0.6", features = ["tls"] }
+tonic = { version = "0.6", features = ["tls", "compression"] }
 prost = "0.9"
 
 bs58 = "*"
@@ -53,6 +53,8 @@ async-stream = "0.2"
 async-channel = "1.6"
 async-trait = "0.1"
 
-[build-dependencies]
-tonic-build = "0.6"
+solana-geyser-connector-plugin-grpc = { path = "../geyser-plugin-grpc" }
+
+[build-dependencies]
+tonic-build = { version = "0.6", features = ["compression"] }
 

lib/src/grpc_plugin_source.rs

@@ -23,6 +23,8 @@ use crate::{
     SnapshotSourceConfig, SourceConfig, TlsConfig,
 };
 
+use solana_geyser_connector_plugin_grpc::compression::zstd_decompress;
+
 type SnapshotData = Response<Vec<RpcKeyedAccount>>;
 
 enum Message {
@@ -336,7 +338,8 @@ pub async fn process_events(
                         }
                         *writes = update.write_version;
                         latest_write.retain(|&k, _| k >= update.slot - latest_write_retention);
-
+                        let mut uncompressed: Vec<u8> = Vec::new();
+                        zstd_decompress(&update.data, &mut uncompressed).unwrap();
                         account_write_queue_sender
                             .send(AccountWrite {
                                 pubkey: Pubkey::new(&update.pubkey),
@@ -346,7 +349,7 @@ pub async fn process_events(
                                 owner: Pubkey::new(&update.owner),
                                 executable: update.executable,
                                 rent_epoch: update.rent_epoch,
-                                data: update.data,
+                                data: uncompressed,
                                 is_selected: update.is_selected,
                             })
                             .await