refactor(connectors): isolate per-connector init failures in runtime

core/connectors/runtime/src/context.rs

@@ -21,7 +21,7 @@ use crate::configs::runtime::ConnectorsRuntimeConfig;
 use crate::metrics::Metrics;
 use crate::stream::IggyClients;
 use crate::{
-    SinkConnectorWrapper, SourceConnectorWrapper,
+    FailedPlugin, SinkConnectorWrapper, SourceConnectorWrapper,
     manager::{
         sink::{SinkDetails, SinkInfo, SinkManager},
         source::{SourceDetails, SourceInfo, SourceManager},
@@ -54,13 +54,20 @@ pub fn init(
     sources_config: &HashMap<String, SourceConfig>,
     sink_wrappers: &[SinkConnectorWrapper],
     source_wrappers: &[SourceConnectorWrapper],
+    failed_sinks: &[FailedPlugin],
+    failed_sources: &[FailedPlugin],
     config_provider: Box<dyn ConnectorsConfigProvider>,
     iggy_clients: Arc<IggyClients>,
     state_path: String,
 ) -> RuntimeContext {
     let metrics = Arc::new(Metrics::init());
-    let sinks = SinkManager::new(map_sinks(sinks_config, sink_wrappers));
-    let sources = SourceManager::new(map_sources(sources_config, source_wrappers));
+    let mut sink_details = map_sinks(sinks_config, sink_wrappers);
+    sink_details.extend(map_failed_sinks(sinks_config, failed_sinks));
+    let mut source_details = map_sources(sources_config, source_wrappers);
+    source_details.extend(map_failed_sources(sources_config, failed_sources));
+
+    let sinks = SinkManager::new(sink_details);
+    let sources = SourceManager::new(source_details);
 
     metrics.set_sinks_total(sinks_config.len() as u32);
     metrics.set_sources_total(sources_config.len() as u32);
@@ -167,3 +174,68 @@ fn map_sources(
     }
     sources
 }
+
+const UNKNOWN_PLUGIN_VERSION: &str = "unknown";
+
+fn map_failed_sinks(
+    sinks_config: &HashMap<String, SinkConfig>,
+    failed: &[FailedPlugin],
+) -> Vec<SinkDetails> {
+    let mut sinks = Vec::with_capacity(failed.len());
+    for plugin in failed {
+        let Some(config) = sinks_config.get(&plugin.key) else {
+            error!("Missing sink config for failed plugin: {}", plugin.key);
+            continue;
+        };
+        sinks.push(SinkDetails {
+            info: SinkInfo {
+                id: plugin.id,
+                key: plugin.key.clone(),
+                name: plugin.name.clone(),
+                path: plugin.path.clone(),
+                version: UNKNOWN_PLUGIN_VERSION.to_owned(),
+                enabled: plugin.enabled,
+                status: ConnectorStatus::Error,
+                last_error: Some(ConnectorError::new(&plugin.error)),
+                plugin_config_format: plugin.config_format,
+            },
+            config: config.clone(),
+            shutdown_tx: None,
+            task_handles: vec![],
+            container: None,
+            restart_guard: Arc::new(Mutex::new(())),
+        });
+    }
+    sinks
+}
+
+fn map_failed_sources(
+    sources_config: &HashMap<String, SourceConfig>,
+    failed: &[FailedPlugin],
+) -> Vec<SourceDetails> {
+    let mut sources = Vec::with_capacity(failed.len());
+    for plugin in failed {
+        let Some(config) = sources_config.get(&plugin.key) else {
+            error!("Missing source config for failed plugin: {}", plugin.key);
+            continue;
+        };
+        sources.push(SourceDetails {
+            info: SourceInfo {
+                id: plugin.id,
+                key: plugin.key.clone(),
+                name: plugin.name.clone(),
+                path: plugin.path.clone(),
+                version: UNKNOWN_PLUGIN_VERSION.to_owned(),
+                enabled: plugin.enabled,
+                status: ConnectorStatus::Error,
+                last_error: Some(ConnectorError::new(&plugin.error)),
+                plugin_config_format: plugin.config_format,
+            },
+            config: config.clone(),
+            handler_tasks: vec![],
+            container: None,
+            restart_guard: Arc::new(Mutex::new(())),
+        });
+    }
+    sources
+}

core/connectors/runtime/src/main.rs

@@ -156,15 +156,15 @@ async fn main() -> Result<(), RuntimeError> {
         connectors_config.sinks().len()
     );
     let sources_config = connectors_config.sources();
-    let sources = source::init(
+    let (sources, failed_sources) = source::init(
         sources_config.clone(),
         &iggy_clients.producer,
         &config.state.path,
     )
     .await?;
 
     let sinks_config = connectors_config.sinks();
-    let sinks = sink::init(sinks_config.clone(), &iggy_clients.consumer).await?;
+    let (sinks, failed_sinks) = sink::init(sinks_config.clone(), &iggy_clients.consumer).await?;
 
     let mut sink_wrappers = vec![];
     let mut sink_containers_by_key: HashMap<String, Arc<Container<SinkApi>>> = HashMap::new();
@@ -200,6 +200,8 @@ async fn main() -> Result<(), RuntimeError> {
         sources_config,
         &sink_wrappers,
         &source_wrappers,
+        &failed_sinks,
+        &failed_sources,
         connectors_config_provider,
         iggy_clients.clone(),
         config.state.path.clone(),
@@ -459,6 +461,42 @@ struct SourceConnectorWrapper {
     plugins: Vec<SourceConnectorPlugin>,
 }
 
+/// Records a connector that failed before its FFI container could be loaded
+/// (path resolution, dlopen, or pre-container setup). Surfaced to the runtime
+/// API/health view as `ConnectorStatus::Error` so one bad config does not hide
+/// the connector or block healthy peers from running.
+pub(crate) struct FailedPlugin {
+    pub id: u32,
+    pub key: String,
+    pub name: String,
+    pub path: String,
+    pub config_format: Option<ConfigFormat>,
+    pub error: String,
+    pub enabled: bool,
+}
+
+impl FailedPlugin {
+    pub(crate) fn new(
+        id: u32,
+        key: &str,
+        name: &str,
+        path: &str,
+        config_format: Option<ConfigFormat>,
+        enabled: bool,
+        error: String,
+    ) -> Self {
+        Self {
+            id,
+            key: key.to_owned(),
+            name: name.to_owned(),
+            path: path.to_owned(),
+            config_format,
+            error,
+            enabled,
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;