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"

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 = { path = "../../yellowstone-shmem/crates/yellowstone-shmem-client" }
+yellowstone-shmem-common = { path = "../../yellowstone-shmem/crates/yellowstone-shmem-common" }
 
 [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/plugin-config.json

@@ -0,0 +1,6 @@
+{
+    "libpath": "/workspaces/devpool/yellowstone-grpc/target/release/libyellowstone_grpc_geyser.so",
+    "grpc": {
+        "shmem_path": "/dev/shm/yellowstone-shmem"
+    }
+}

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::{
@@ -35,14 +32,12 @@ pub struct Config {
 }
 
 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 +259,35 @@ impl<'de> Deserialize<'de> for GrpcAddresses {
     }
 }
 
+#[derive(Debug)]
+pub enum ConfigError {
+    Io(std::io::Error),
+    Parse(serde_json::Error),
+}
+
+impl std::fmt::Display for ConfigError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Io(e) => write!(f, "config file error: {e}"),
+            Self::Parse(e) => write!(f, "config parse error: {e}"),
+        }
+    }
+}
+
+impl std::error::Error for ConfigError {}
+
+impl From<std::io::Error> for ConfigError {
+    fn from(e: std::io::Error) -> Self {
+        Self::Io(e)
+    }
+}
+
+impl From<serde_json::Error> for ConfigError {
+    fn from(e: serde_json::Error) -> Self {
+        Self::Parse(e)
+    }
+}
+
 #[derive(Debug, Clone, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct ConfigGrpc {
@@ -383,6 +407,25 @@ pub struct ConfigGrpc {
 
     #[serde(default)]
     pub listen: Option<Vec<ListenConfig>>,
+
+    /// 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`
+    #[serde(default)]
+    pub shmem_path: Option<String>,
+
+    /// 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 shmem_snapshot_path: Option<String>,
+
+    /// Interval in seconds between shmem health reports.
+    /// Set to 0 to disable diagnostics entirely.
+    #[serde(default = "ConfigGrpc::default_shmem_health_interval_secs")]
+    pub shmem_health_interval_secs: u64,
 }
 
 #[derive(Debug, Clone, Deserialize)]
@@ -445,6 +488,10 @@ impl ConfigGrpc {
     const fn default_replay_stored_slots() -> u64 {
         150
     }
+
+    fn default_shmem_health_interval_secs() -> u64 {
+        10
+    }
 }
 
 #[derive(Debug, Clone, Deserialize)]