STR-509: Shared memory plugin for Geyser

Cargo.toml

@@ -97,8 +97,8 @@ vergen = "9.0.6"
 wincode = "0.4.4"
 
 # Agave Monorepo
-agave-geyser-plugin-interface = "4.0.0"
-solana-account-decoder = "4.0.0"
+agave-geyser-plugin-interface = "~4.0.0-beta.2"
+solana-account-decoder = "~4.0.0-rc.0"
 solana-logger = "3.0.0"
 solana-storage-proto = { version = "4.0.0", features = ["agave-unstable-api"] }
 solana-transaction-context = "4.0.0"

rust-toolchain.toml

@@ -1,5 +1,5 @@
 [toolchain]
-channel = "1.93.1"
+channel = "1.95.0"
 components = ["clippy", "rustfmt"]
 targets = []
 profile = "minimal"

yellowstone-grpc-geyser/Cargo.toml

@@ -11,7 +11,11 @@ keywords = { workspace = true }
 publish = { workspace = true }
 
 [lib]
-crate-type = ["cdylib", "rlib"]
+crate-type = ["rlib"]
+
+[[bin]]
+name = "yellowstone-grpc-geyser"
+path = "src/main.rs"
 
 [[bin]]
 name = "config-check"
@@ -48,7 +52,7 @@ rayon = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
-tokio = { workspace = true, features = ["rt-multi-thread", "macros", "fs"] }
+tokio = { workspace = true, features = ["rt-multi-thread", "macros", "fs", "signal"] }
 tokio-util = { workspace = true, features = ["rt"]}
 tokio-rustls = { workspace = true, features = ["aws-lc-rs"] }
 tokio-stream = { workspace = true }
@@ -87,6 +91,8 @@ prost_011 = { workspace = true, optional = true }
 yellowstone-grpc-proto = { workspace = true, features = ["tonic", "account-data-as-bytes"] }
 yellowstone-block-machine = { workspace = true, default-features = false }
 yellowstone-grpc-tools = { workspace = true }
+yellowstone-shmem-client = { git = "https://github.com/rpcpool/yellowstone-shmem.git", branch = "main" }
+yellowstone-shmem-common = { git = "https://github.com/rpcpool/yellowstone-shmem.git", branch = "main" }
 
 [target.'cfg(any(target_os = "linux", target_os = "windows"))'.dependencies]
 # wrapped in mod cpu_core_affinity
@@ -97,7 +103,6 @@ name = "encode"
 harness = false
 required-features = ["bench"]
 
-
 [build-dependencies]
 anyhow = { workspace = true }
 cargo-lock = { workspace = true }
@@ -116,6 +121,7 @@ solana-signature = { workspace = true }
 solana-signer = { workspace = true }
 solana-storage-proto = { workspace = true }
 solana-transaction = { workspace = true, features = ["bincode", "blake3"] }
+yellowstone-conduit = { git = "https://github.com/rpcpool/yellowstone-conduit.git", branch = "main" }
 
 [lints]
 workspace = true

yellowstone-grpc-geyser/config.json

@@ -9,10 +9,8 @@
   },
   "grpc": {
     "address": "0.0.0.0:10000",
-    "tls_config": {
-      "cert_path": "",
-      "key_path": ""
-    },
+    "shmem_path": "/dev/shm/yellowstone-shmem",
+    "tls_config": null,
     "compression": {
       "accept": ["gzip", "zstd"],
       "send": ["gzip", "zstd"]
@@ -44,7 +42,7 @@
         "data_slice_max": 2
       },
       "slots": {
-        "max": 1
+          "max": 10
       },
       "transactions": {
         "max": 1,

yellowstone-grpc-geyser/src/config.rs

@@ -1,8 +1,5 @@
 use {
     crate::plugin::filter::limits::FilterLimits,
-    agave_geyser_plugin_interface::geyser_plugin_interface::{
-        GeyserPluginError, Result as PluginResult,
-    },
     bytesize::ByteSize,
     serde::{de, Deserialize, Deserializer},
     std::{
@@ -28,21 +25,21 @@ pub struct Config {
     pub tokio: ConfigTokio,
     pub grpc: ConfigGrpc,
     #[serde(default)]
+    pub shmem: Option<ConfigShmem>,
+    #[serde(default)]
     pub prometheus: Option<ConfigPrometheus>,
     /// Collect client filters, processed slot and make it available on prometheus port `/debug_clients`
     #[serde(default)]
     pub debug_clients_http: bool,
 }
 
 impl Config {
-    fn load_from_str(config: &str) -> PluginResult<Self> {
-        serde_json::from_str(config).map_err(|error| GeyserPluginError::ConfigFileReadError {
-            msg: error.to_string(),
-        })
+    fn load_from_str(config: &str) -> Result<Self, ConfigError> {
+        Ok(serde_json::from_str(config)?)
     }
 
-    pub fn load_from_file<P: AsRef<Path>>(file: P) -> PluginResult<Self> {
-        let config = read_to_string(file).map_err(GeyserPluginError::ConfigFileOpenError)?;
+    pub fn load_from_file<P: AsRef<Path>>(file: P) -> Result<Self, ConfigError> {
+        let config = read_to_string(file)?;
         Self::load_from_str(&config)
     }
 }
@@ -264,6 +261,14 @@ impl<'de> Deserialize<'de> for GrpcAddresses {
     }
 }
 
+#[derive(Debug, thiserror::Error)]
+pub enum ConfigError {
+    #[error(transparent)]
+    Io(#[from] std::io::Error),
+    #[error(transparent)]
+    JsonParse(#[from] serde_json::Error),
+}
+
 #[derive(Debug, Clone, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct ConfigGrpc {
@@ -509,6 +514,42 @@ pub struct ConfigPrometheus {
     pub address: SocketAddr,
 }
 
+#[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct ConfigShmem {
+    /// Path to the yellowstone-shmem shared memory file.
+    /// When set, the grpc service reads geyser events from the shmem ring
+    /// instead of the in-process plugin channel.
+    /// Example: `/dev/shm/yellowstone-shmem`
+    pub path: PathBuf,
+
+    /// Path to the shmem geyser plugin snapshot mmap file.
+    /// When set, the grpc service reads snapshot from the from the mmap file
+    /// instead of the in-process plugin channel.
+    /// Example: `/dev/shm/yellowstone-shmem-snapshot`
+    #[serde(default)]
+    pub snapshot_path: Option<PathBuf>,
+
+    /// Interval in seconds between shmem health reports.
+    /// Set to 0 to disable diagnostics entirely.
+    #[serde(default = "ConfigShmem::default_health_interval_secs")]
+    pub health_interval_secs: u64,
+
+    /// Interval in seconds between ring restart checks.
+    #[serde(default = "ConfigShmem::default_restart_check_interval_secs")]
+    pub restart_check_interval_secs: u64,
+}
+
+impl ConfigShmem {
+    fn default_health_interval_secs() -> u64 {
+        10
+    }
+
+    pub fn default_restart_check_interval_secs() -> u64 {
+        5
+    }
+}
+
 #[derive(Deserialize)]
 #[serde(untagged)]
 enum ValueIntStr<'a, T> {