feat(plugin-sdk): FFI panic guards, RAII call guard, timeout, provenance, worker thread, v9 wire bump

apps/skit/src/config.rs

@@ -1,7 +1,6 @@
 // SPDX-FileCopyrightText: © 2025 StreamKit Contributors
 //
 // SPDX-License-Identifier: MPL-2.0
-
 use figment::{
     providers::{Env, Format, Toml},
     Figment,
@@ -17,6 +16,15 @@ const fn default_engine_batch_size() -> usize {
     32
 }
 
+/// Deserialize `Option<u64>` with a minimum clamp of 1 for timeout values.
+fn deserialize_clamp_timeout<'de, D>(deserializer: D) -> Result<Option<u64>, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    let val: Option<u64> = Option::deserialize(deserializer)?;
+    Ok(val.map(|v| v.max(1)))
+}
+
 /// Preset tuning profiles for the engine.
 #[derive(Deserialize, Serialize, Debug, Clone, Copy, JsonSchema)]
 #[serde(rename_all = "kebab-case")]
@@ -187,6 +195,10 @@ const fn default_max_body_size() -> usize {
     100 * 1024 * 1024
 }
 
+const fn default_native_call_timeout_value() -> u64 {
+    300
+}
+
 fn default_cors_allowed_origins() -> Vec<String> {
     vec![
         // Portless localhost (e.g., reverse proxy on 80/443)
@@ -354,6 +366,17 @@ impl Default for ServerConfig {
 #[derive(Deserialize, Serialize, Debug, Clone, JsonSchema)]
 pub struct PluginConfig {
     pub directory: String,
+    /// Native plugin FFI call timeout in seconds (default: 300, minimum: 1).
+    ///
+    /// Set to `null` to use only the default backstop timeout on the reply
+    /// side; the send-side backpressure guard remains bounded regardless.
+    ///
+    /// Values below 1 are clamped to 1 to prevent instant timeouts.
+    #[serde(
+        default = "PluginConfig::default_native_call_timeout_secs",
+        deserialize_with = "deserialize_clamp_timeout"
+    )]
+    pub native_call_timeout_secs: Option<u64>,
     #[serde(flatten, default)]
     pub http_management: PluginHttpConfig,
     #[serde(flatten, default)]
@@ -450,6 +473,7 @@ impl Default for PluginConfig {
     fn default() -> Self {
         Self {
             directory: ".plugins".to_string(),
+            native_call_timeout_secs: Some(default_native_call_timeout_value()),
             http_management: PluginHttpConfig::default(),
             marketplace: PluginMarketplaceConfig::default(),
             trusted_pubkeys: Vec::new(),
@@ -460,6 +484,15 @@ impl Default for PluginConfig {
     }
 }
 
+impl PluginConfig {
+    // Serde default hooks must return the exact field type; the wrapped value
+    // distinguishes missing config from explicit null.
+    #[allow(clippy::unnecessary_wraps)]
+    const fn default_native_call_timeout_secs() -> Option<u64> {
+        Some(default_native_call_timeout_value())
+    }
+}
+
 const fn default_require_registry_origin() -> bool {
     false
 }

apps/skit/src/marketplace_installer.rs

@@ -2108,6 +2108,7 @@ mod tests {
             plugin_dir.to_path_buf(),
             wasm_dir,
             native_dir,
+            Some(std::time::Duration::from_secs(300)),
         )?;
         Ok(Arc::new(tokio::sync::Mutex::new(manager)))
     }
@@ -2201,6 +2202,7 @@ mod tests {
 
         let config = PluginConfig {
             directory: plugin_dir.to_string_lossy().to_string(),
+            native_call_timeout_secs: Some(300),
             http_management: crate::config::PluginHttpConfig { allow_http_management: false },
             marketplace: crate::config::PluginMarketplaceConfig {
                 marketplace_enabled: true,
@@ -2302,6 +2304,7 @@ mod tests {
 
         let config = PluginConfig {
             directory: plugin_dir.to_string_lossy().to_string(),
+            native_call_timeout_secs: Some(300),
             http_management: crate::config::PluginHttpConfig { allow_http_management: false },
             marketplace: crate::config::PluginMarketplaceConfig {
                 marketplace_enabled: true,
@@ -2369,6 +2372,7 @@ mod tests {
 
         let config = PluginConfig {
             directory: plugin_dir.to_string_lossy().to_string(),
+            native_call_timeout_secs: Some(300),
             http_management: crate::config::PluginHttpConfig { allow_http_management: false },
             marketplace: crate::config::PluginMarketplaceConfig {
                 marketplace_enabled: true,

apps/skit/src/plugins.rs

@@ -143,6 +143,7 @@ pub struct UnifiedPluginManager {
     plugin_base_dir: PathBuf,
     wasm_directory: PathBuf,
     native_directory: PathBuf,
+    native_call_timeout: Option<std::time::Duration>,
     engine: Arc<Engine>,
     #[allow(dead_code)] // Will be used when plugins are migrated to new resource system
     resource_manager: Arc<streamkit_core::ResourceManager>,
@@ -165,6 +166,7 @@ impl UnifiedPluginManager {
         plugin_base_dir: PathBuf,
         wasm_directory: PathBuf,
         native_directory: PathBuf,
+        native_call_timeout: Option<std::time::Duration>,
     ) -> Result<Self> {
         if !wasm_directory.exists() {
             std::fs::create_dir_all(&wasm_directory).with_context(|| {
@@ -188,6 +190,7 @@ impl UnifiedPluginManager {
             plugin_base_dir,
             wasm_directory,
             native_directory,
+            native_call_timeout,
             engine,
             resource_manager,
             plugins_loaded_gauge: meter
@@ -776,12 +779,13 @@ impl UnifiedPluginManager {
             return Err(anyhow!("Native plugin file {} does not exist", path.to_string_lossy()));
         }
 
-        let plugin = LoadedNativePlugin::load(path)
+        let mut plugin = LoadedNativePlugin::load(path)
             .map_err(|e| {
                 tracing::error!(error = %e, path = ?path, "Detailed native plugin load error");
                 e
             })
             .with_context(|| format!("failed to load native plugin {}", path.to_string_lossy()))?;
+        plugin.set_call_timeout(self.native_call_timeout);
 
         let metadata = plugin.metadata();
         let original_kind = metadata.kind.clone();

apps/skit/src/server/mod.rs

@@ -4061,6 +4061,7 @@ pub fn create_app(
         plugin_base_dir,
         wasm_plugin_dir,
         native_plugin_dir,
+        config.plugins.native_call_timeout_secs.map(std::time::Duration::from_secs),
     )
     .expect("Failed to initialize unified plugin manager");
     let plugin_manager = Arc::new(tokio::sync::Mutex::new(plugin_manager));

crates/plugin-native/Cargo.toml

@@ -18,6 +18,7 @@ streamkit-plugin-sdk-native = { workspace = true }
 
 libloading = "0.9"
 anyhow = "1.0"
+opentelemetry = { workspace = true }
 tracing = { workspace = true }
 tokio = { workspace = true, features = ["sync", "macros", "rt", "time"] }
 async-trait = { workspace = true }

crates/plugin-native/src/lib.rs

@@ -7,14 +7,18 @@
 //! This crate provides the host-side runtime for loading and executing native plugins
 //! that use the C ABI interface.
 
+pub mod metrics;
 pub mod wrapper;
 
 use anyhow::{anyhow, Context, Result};
 use libloading::{Library, Symbol};
 use std::path::Path;
 use std::sync::Arc;
 use streamkit_core::{NodeRegistry, PinCardinality};
-use streamkit_plugin_sdk_native::types::{CNativePluginAPI, NATIVE_PLUGIN_API_VERSION};
+use streamkit_plugin_sdk_native::types::PLUGIN_SET_LOG_ENABLED_SYMBOL;
+use streamkit_plugin_sdk_native::types::{
+    CNativePluginAPI, CSetLogEnabledCallback, NATIVE_PLUGIN_API_VERSION,
+};
 use streamkit_plugin_sdk_native::{conversions, types::PLUGIN_API_SYMBOL};
 use tracing::{info, warn};
 
@@ -35,6 +39,8 @@ pub struct LoadedNativePlugin {
     library: Arc<Library>,
     api: &'static CNativePluginAPI,
     metadata: PluginMetadata,
+    call_timeout: Option<std::time::Duration>,
+    set_log_enabled_callback: Option<CSetLogEnabledCallback>,
 }
 
 /// Metadata extracted from a plugin
@@ -106,12 +112,16 @@ impl LoadedNativePlugin {
         // the lifetime of the loaded library, which we keep alive via Arc<Library>.
         let api = unsafe { &*api_ptr };
 
-        // Check API version compatibility — accept v6 (pre-BinaryWithMeta),
-        // v7 (added BinaryWithMeta), and v8 (added EncodedAudio metadata).
-        // All are wire-compatible; v6 plugins never emit BinaryWithMeta
-        // packets, v7 plugins don't declare EncodedAudio pin types (they
-        // fall back to Binary), but runtime packet transport is unaffected
-        // since EncodedAudio is a metadata-only discriminant.
+        // Check API version compatibility — accept v6 through v9.
+        // v6: pre-BinaryWithMeta.
+        // v7: added BinaryWithMeta.
+        // v8: added EncodedAudio metadata.
+        // v9: zero-copy binary packets (buffer_handle), logger overhaul
+        //     (set_log_enabled_callback, target = plugin kind).
+        // v6–v8 are wire-compatible; v9 adds a Binary→BinaryWithMeta wire
+        // upgrade (see v9 notes in types.rs).  Version-gated features use
+        // runtime api_version checks (e.g. buffer_handle for v9, downgrade
+        // for v6).
         if api.version < MIN_SUPPORTED_API_VERSION || api.version > NATIVE_PLUGIN_API_VERSION {
             let plugin_version = api.version;
             return Err(anyhow!(
@@ -123,6 +133,25 @@ impl LoadedNativePlugin {
         // Extract metadata
         let mut metadata = Self::extract_metadata(api)?;
 
+        let set_log_enabled_callback = if api.version >= 9 {
+            // SAFETY: Optional extension symbol. When present, the SDK macro
+            // exports it with the exact `CSetLogEnabledCallback` signature.
+            match unsafe { library.get::<CSetLogEnabledCallback>(PLUGIN_SET_LOG_ENABLED_SYMBOL) } {
+                Ok(symbol) => Some(*symbol),
+                Err(e) => {
+                    warn!(
+                        kind = %metadata.kind,
+                        api_version = api.version,
+                        error = %e,
+                        "v9 plugin did not export log-enabled callback symbol"
+                    );
+                    None
+                },
+            }
+        } else {
+            None
+        };
+
         // Detect source plugin capability from the v3 API fields.
         // If the plugin provides `get_source_config`, we probe it with a temporary
         // instance to read tick parameters.  If instance creation fails we fall back
@@ -160,7 +189,13 @@ impl LoadedNativePlugin {
 
         info!(kind = %metadata.kind, "Successfully loaded native plugin");
 
-        Ok(Self { library: Arc::new(library), api, metadata })
+        Ok(Self {
+            library: Arc::new(library),
+            api,
+            metadata,
+            call_timeout: Some(wrapper::DEFAULT_CALL_TIMEOUT),
+            set_log_enabled_callback,
+        })
     }
 
     /// Extract metadata from the plugin
@@ -300,6 +335,21 @@ impl LoadedNativePlugin {
         &self.library
     }
 
+    /// Override the reply-side timeout for FFI calls (process_packet, flush,
+    /// tick).  This controls how long the async side waits for the worker
+    /// thread's oneshot reply.
+    ///
+    /// Pass `None` to fall back to the default backstop timeout
+    /// ([`DEFAULT_CALL_TIMEOUT`](crate::wrapper::DEFAULT_CALL_TIMEOUT),
+    /// 300 s) instead of a caller-chosen duration.  The reply side is
+    /// **never** truly unbounded — the backstop always applies.
+    ///
+    /// The channel-send timeout (backpressure guard) also uses
+    /// `DEFAULT_CALL_TIMEOUT` when this is `None`.
+    pub const fn set_call_timeout(&mut self, timeout: Option<std::time::Duration>) {
+        self.call_timeout = timeout;
+    }
+
     /// Create a new node instance from this plugin
     ///
     /// # Errors
@@ -316,6 +366,8 @@ impl LoadedNativePlugin {
             self.api,
             self.metadata.clone(),
             params,
+            self.call_timeout,
+            self.set_log_enabled_callback,
         )?;
 
         Ok(Box::new(wrapper))