feat: add geyser plugin safe-start checks

geyser-plugin/README.md

@@ -123,6 +123,65 @@ admin = "127.0.0.1:3000"
 
 `kafka.bootstrap_servers`, `kafka.topic`, `plugin.local_rpc_url`, and `plugin.admin` are required. The `admin` bind address serves `POST /filters/accounts` and, when `plugin.metrics` is `true`, also `GET /metrics`. If `ksql.url` is set, it must be a valid absolute `http` or `https` base URL and startup will fail if the restore query cannot complete. Legacy filter arrays and legacy transaction, slot-status, block, and wrapping options are rejected during config parsing.
 
+## Startup checks
+
+Startup checks run from `KafkaPlugin::on_load` during normal validator/plugin startup. They run whether the validator is launched through `make geyser-plugin-launch` or directly with `solana-test-validator --geyser-plugin-config ...`; no separate preflight binary or Makefile target is required.
+
+Startup validates:
+
+- validator JSON wrapper
+- runtime TOML config
+- plugin library path
+- admin bind address
+- ksqlDB startup restore when `ksql.url` is configured
+- Kafka bootstrap/topic readiness
+- local RPC URL syntax only, not local RPC liveness
+
+The local RPC endpoint is not required to be reachable during startup checks because it belongs to the validator being launched.
+
+### Safe-start manual test matrix
+
+| Scenario | Temporary change | Expected prefix |
+| --- | --- | --- |
+| malformed validator JSON | invalid JSON in wrapper copy | Agave rejects the wrapper before plugin load with `FailedToLoadPlugin`; no segfault |
+| missing runtime TOML | JSON `config_file` points to missing TOML | `ERROR config startup check failed` |
+| malformed runtime TOML | invalid TOML in runtime copy | `ERROR config startup check failed` |
+| missing Kafka bootstrap | empty `kafka.bootstrap_servers` | `ERROR config startup check failed` |
+| Kafka down | no Kafka on configured bootstrap | `ERROR kafka startup check failed` |
+| ksqlDB down | `ksql.url = "http://127.0.0.1:1"` | `ERROR ksql startup check failed` |
+| invalid ksql table | `table = "bad-name"` | `ERROR config startup check failed` |
+| admin port in use | keep listener on configured port | `ERROR admin startup check failed` |
+| malformed local RPC URL | `local_rpc_url = "127.0.0.1:8899"` | `ERROR config startup check failed` |
+
+Proof command for the original issue:
+
+```shell
+make geyser-plugin-launch
+```
+
+Expected with no dependencies running:
+
+- exits non-zero
+- prints a Kafka startup check error with action `ensure Kafka is reachable at kafka.bootstrap_servers or update kafka.bootstrap_servers`
+- validator/plugin startup exits gracefully
+- does not print `Segmentation fault`
+
+Direct validator path that must receive the same checks:
+
+```shell
+cd geyser-plugin
+solana-test-validator --log --reset --geyser-plugin-config plugin-config.json
+```
+
+Expected failures and messages must match the Makefile path because the checks live in `KafkaPlugin::on_load`.
+
+Success path:
+
+```shell
+make kafka-ready
+make geyser-plugin-launch
+```
+
 ## Whitelist Management
 
 Account inclusion is managed through the HTTP API:

geyser-plugin/src/config.rs

@@ -77,10 +77,10 @@ pub struct PluginConfig {
 
 #[derive(Debug, Deserialize)]
 #[serde(deny_unknown_fields)]
-struct ValidatorConfig {
+pub(crate) struct ValidatorConfig {
     #[allow(dead_code)]
-    libpath: String,
-    config_file: PathBuf,
+    pub(crate) libpath: PathBuf,
+    pub(crate) config_file: PathBuf,
 }
 
 fn default_shutdown_timeout_ms() -> u64 {
@@ -125,6 +125,7 @@ impl Default for KsqlConfig {
 
 impl Config {
     /// Read plugin config from either a validator JSON wrapper or a TOML runtime config.
+    #[allow(dead_code)]
     pub fn read_from<P: AsRef<Path>>(config_path: P) -> PluginResult<Self> {
         let config_path = config_path.as_ref();
         let contents = read_to_string(config_path)?;
@@ -171,14 +172,14 @@ impl Config {
         }
     }
 
-    fn fill_defaults(&mut self) {
+    pub(crate) fn fill_defaults(&mut self) {
         self.set_default("request.required.acks", "1");
         self.set_default("message.timeout.ms", "30000");
         self.set_default("compression.type", "lz4");
         self.set_default("partitioner", "murmur2_random");
     }
 
-    fn validate(&self) -> PluginResult<()> {
+    pub(crate) fn validate(&self) -> PluginResult<()> {
         if self.kafka.bootstrap_servers.trim().is_empty() {
             return Err(GeyserPluginError::ConfigFileReadError {
                 msg: "missing required config field `kafka.bootstrap_servers`"
@@ -192,10 +193,40 @@ impl Config {
             });
         }
 
-        if self.plugin.local_rpc_url.trim().is_empty() {
+        let trimmed_local_rpc_url = self.plugin.local_rpc_url.trim();
+        if trimmed_local_rpc_url.is_empty() {
             return Err(GeyserPluginError::ConfigFileReadError {
-                msg: "missing required config field `plugin.local_rpc_url`"
-                    .to_owned(),
+                msg:
+                    "invalid config field `plugin.local_rpc_url`: URL must not be empty"
+                        .to_owned(),
+            });
+        }
+
+        let parsed_local_rpc_url =
+            Url::parse(trimmed_local_rpc_url).map_err(|error| {
+                GeyserPluginError::ConfigFileReadError {
+                    msg: format!(
+                        "invalid config field `plugin.local_rpc_url`: {error}"
+                    ),
+                }
+            })?;
+
+        match parsed_local_rpc_url.scheme() {
+            "http" | "https" => {}
+            scheme => {
+                return Err(GeyserPluginError::ConfigFileReadError {
+                    msg: format!(
+                        "invalid config field `plugin.local_rpc_url`: unsupported scheme `{scheme}`"
+                    ),
+                });
+            }
+        }
+
+        if !parsed_local_rpc_url.has_host() {
+            return Err(GeyserPluginError::ConfigFileReadError {
+                msg:
+                    "invalid config field `plugin.local_rpc_url`: host is required"
+                        .to_owned(),
             });
         }
 
@@ -238,26 +269,53 @@ impl Config {
                         .to_owned(),
                 });
             }
+        }
 
-            if self.ksql.table.trim().is_empty() {
-                return Err(GeyserPluginError::ConfigFileReadError {
-                    msg: "invalid config field `ksql.table`: table must not be empty".to_owned(),
-                });
+        validate_ksql_identifier(&self.ksql.table).map_err(|error| {
+            GeyserPluginError::ConfigFileReadError {
+                msg: format!("invalid config field `ksql.table`: {error}"),
             }
-        }
+        })?;
 
         Ok(())
     }
 }
 
+/// Validates that `identifier` is a safe ksqlDB identifier suitable for
+/// direct interpolation into a SQL statement. The identifier must start with
+/// an ASCII letter or `_` and may otherwise contain only ASCII alphanumeric
+/// characters or `_`.
+pub(crate) fn validate_ksql_identifier(
+    identifier: &str,
+) -> std::io::Result<&str> {
+    let mut chars = identifier.chars();
+    let first = chars.next().ok_or_else(|| {
+        std::io::Error::other("ksql identifier must not be empty")
+    })?;
+    if !(first.is_ascii_alphabetic() || first == '_') {
+        return Err(std::io::Error::other(format!(
+            "invalid ksql identifier `{identifier}`: must start with an ASCII letter or `_`"
+        )));
+    }
+    for c in chars {
+        if !(c.is_ascii_alphanumeric() || c == '_') {
+            return Err(std::io::Error::other(format!(
+                "invalid ksql identifier `{identifier}`: only ASCII alphanumeric characters and `_` are allowed"
+            )));
+        }
+    }
+    Ok(identifier)
+}
+
+#[allow(dead_code)]
 fn read_to_string(path: &Path) -> PluginResult<String> {
     let mut file = File::open(path)?;
     let mut contents = String::new();
     file.read_to_string(&mut contents)?;
     Ok(contents)
 }
 
-fn resolve_runtime_config_path(
+pub(crate) fn resolve_runtime_config_path(
     wrapper_path: &Path,
     runtime_path: &Path,
 ) -> PathBuf {
@@ -273,12 +331,45 @@ fn resolve_runtime_config_path(
 
 #[cfg(test)]
 mod tests {
-    use super::Config;
+    use super::{Config, validate_ksql_identifier};
     use std::{
         fs,
         time::{SystemTime, UNIX_EPOCH},
     };
 
+    #[test]
+    fn test_validates_simple_identifier() {
+        assert_eq!(validate_ksql_identifier("accounts").unwrap(), "accounts");
+        assert_eq!(validate_ksql_identifier("_x").unwrap(), "_x");
+        assert_eq!(validate_ksql_identifier("A1_b2").unwrap(), "A1_b2");
+    }
+
+    #[test]
+    fn test_rejects_empty_identifier() {
+        let error = validate_ksql_identifier("").unwrap_err().to_string();
+        assert!(error.contains("must not be empty"));
+    }
+
+    #[test]
+    fn test_rejects_identifier_starting_with_digit() {
+        let error = validate_ksql_identifier("1bad").unwrap_err().to_string();
+        assert!(error.contains("must start with an ASCII letter"));
+    }
+
+    #[test]
+    fn test_rejects_identifier_with_invalid_characters() {
+        let error = validate_ksql_identifier("accounts; DROP TABLE x")
+            .unwrap_err()
+            .to_string();
+        assert!(error.contains("only ASCII alphanumeric"));
+    }
+
+    #[test]
+    fn test_rejects_identifier_with_quote() {
+        let error = validate_ksql_identifier("a\"b").unwrap_err().to_string();
+        assert!(error.contains("only ASCII alphanumeric"));
+    }
+
     fn parse_config(toml: &str) -> Result<Config, String> {
         let mut config: Config =
             toml::from_str(toml).map_err(|error| error.to_string())?;
@@ -606,6 +697,141 @@ admin = "127.0.0.1:8080"
         );
     }
 
+    #[test]
+    fn test_rejects_invalid_ksql_table_identifier() {
+        let error = parse_config(
+            r#"
+libpath = "target/release/libsolana_accountsdb_plugin_kafka.so"
+
+[kafka]
+bootstrap_servers = "localhost:9092"
+topic = "solana.testnet.account_updates"
+
+[ksql]
+url = "http://127.0.0.1:8088"
+table = "bad-name"
+
+[plugin]
+local_rpc_url = "http://127.0.0.1:8899"
+admin = "127.0.0.1:8080"
+"#,
+        )
+        .unwrap_err();
+
+        assert!(error.contains("invalid config field `ksql.table`"));
+        assert!(error.contains("only ASCII alphanumeric"));
+    }
+
+    #[test]
+    fn test_rejects_ksql_table_starting_with_digit() {
+        let error = parse_config(
+            r#"
+libpath = "target/release/libsolana_accountsdb_plugin_kafka.so"
+
+[kafka]
+bootstrap_servers = "localhost:9092"
+topic = "solana.testnet.account_updates"
+
+[ksql]
+table = "1bad"
+
+[plugin]
+local_rpc_url = "http://127.0.0.1:8899"
+admin = "127.0.0.1:8080"
+"#,
+        )
+        .unwrap_err();
+
+        assert!(error.contains("invalid config field `ksql.table`"));
+        assert!(error.contains("must start with an ASCII letter"));
+    }
+
+    #[test]
+    fn test_rejects_empty_local_rpc_url() {
+        let error = parse_config(
+            r#"
+libpath = "target/release/libsolana_accountsdb_plugin_kafka.so"
+
+[kafka]
+bootstrap_servers = "localhost:9092"
+topic = "solana.testnet.account_updates"
+
+[plugin]
+local_rpc_url = "   "
+admin = "127.0.0.1:8080"
+"#,
+        )
+        .unwrap_err();
+
+        assert!(error.contains("invalid config field `plugin.local_rpc_url`"));
+        assert!(error.contains("URL must not be empty"));
+    }
+
+    #[test]
+    fn test_rejects_local_rpc_url_without_scheme() {
+        let error = parse_config(
+            r#"
+libpath = "target/release/libsolana_accountsdb_plugin_kafka.so"
+
+[kafka]
+bootstrap_servers = "localhost:9092"
+topic = "solana.testnet.account_updates"
+
+[plugin]
+local_rpc_url = "127.0.0.1:8899"
+admin = "127.0.0.1:8080"
+"#,
+        )
+        .unwrap_err();
+
+        assert!(error.contains("invalid config field `plugin.local_rpc_url`"));
+        assert!(error.contains("relative URL without a base"));
+    }
+
+    #[test]
+    fn test_rejects_local_rpc_url_with_unsupported_scheme() {
+        let error = parse_config(
+            r#"
+libpath = "target/release/libsolana_accountsdb_plugin_kafka.so"
+
+[kafka]
+bootstrap_servers = "localhost:9092"
+topic = "solana.testnet.account_updates"
+
+[plugin]
+local_rpc_url = "ftp://127.0.0.1:8899"
+admin = "127.0.0.1:8080"
+"#,
+        )
+        .unwrap_err();
+
+        assert!(error.contains("invalid config field `plugin.local_rpc_url`"));
+        assert!(error.contains("unsupported scheme `ftp`"));
+    }
+
+    #[test]
+    fn test_rejects_local_rpc_url_without_host() {
+        let error = parse_config(
+            r#"
+libpath = "target/release/libsolana_accountsdb_plugin_kafka.so"
+
+[kafka]
+bootstrap_servers = "localhost:9092"
+topic = "solana.testnet.account_updates"
+
+[plugin]
+local_rpc_url = "http://:8899"
+admin = "127.0.0.1:8080"
+"#,
+        )
+        .unwrap_err();
+
+        assert!(error.contains("invalid config field `plugin.local_rpc_url`"));
+        assert!(
+            error.contains("empty host") || error.contains("host is required")
+        );
+    }
+
     #[test]
     fn test_passes_through_kafka_client_overrides() {
         let config = parse_config(