diff --git a/core/connectors/runtime/src/main.rs b/core/connectors/runtime/src/main.rs index 12d627c84a..8890c1d430 100644 --- a/core/connectors/runtime/src/main.rs +++ b/core/connectors/runtime/src/main.rs @@ -18,6 +18,7 @@ use crate::configs::connectors::{ ConnectorKey, ConnectorsConfig, ConnectorsConfigProvider, create_connectors_config_provider, }; +use crate::metrics::ConnectorType; use ::configs::ConfigProvider; use clap::Parser; use configs::connectors::ConfigFormat; @@ -186,7 +187,7 @@ async fn main() -> Result<(), RuntimeError> { let mut source_wrappers = vec![]; let mut source_containers_by_key: HashMap>> = HashMap::new(); for (_path, source) in sources { - let container = Arc::new(source.container); + let container = source.container; let handle_callback = container.iggy_source_handle_v2; let batch_result_callback = container.iggy_source_batch_result; for plugin in &source.plugins { @@ -456,8 +457,30 @@ struct SinkConnectorWrapper { plugins: Vec, } +/// Closes a plugin instance whose setup did not finish, reporting a refusal +/// rather than returning it: every caller is already on a failure path with an +/// error of its own to surface. +/// +/// The two sides had this body inline, one word apart. That word is the label +/// [`ConnectorType`] already defines, so it is taken as the enum rather than a +/// string nothing constrains. +pub(crate) fn close_plugin_instance( + close: &dyn Fn(u32) -> i32, + kind: ConnectorType, + plugin_id: u32, + key: &str, +) { + let close_result = close(plugin_id); + if close_result != 0 { + let kind = kind.as_label(); + warn!( + "iggy_{kind}_close returned {close_result} while cleaning up failed {kind} connector with ID: {plugin_id} ({key})" + ); + } +} + struct SourceConnector { - container: Container, + container: Arc>, plugins: Vec, } diff --git a/core/connectors/runtime/src/manager/source.rs b/core/connectors/runtime/src/manager/source.rs index 8ef68e8b19..dcdd85b937 100644 --- a/core/connectors/runtime/src/manager/source.rs +++ b/core/connectors/runtime/src/manager/source.rs @@ -82,28 +82,23 @@ impl SourceManager { metrics: Option<&Arc>, ) { if let Some(source) = self.sources.get(key) { - let mut source = source.lock().await; - let old_status = source.info.status; - source.info.status = status; - if matches!(status, ConnectorStatus::Running | ConnectorStatus::Stopped) { - source.info.last_error = None; - } - if let Some(metrics) = metrics { - if old_status != ConnectorStatus::Running && status == ConnectorStatus::Running { - metrics.increment_sources_running(); - } else if old_status == ConnectorStatus::Running - && status != ConnectorStatus::Running - { - metrics.decrement_sources_running(); - } - } + source.lock().await.apply_status(status, metrics); } } - pub async fn set_error(&self, key: &str, error_message: &str) { + pub async fn set_error(&self, key: &str, error_message: &str, metrics: Option<&Arc>) { if let Some(source) = self.sources.get(key) { let mut source = source.lock().await; - source.info.status = ConnectorStatus::Error; + // Through the shared transition, so leaving `Running` moves the + // gauge. Skipping it left an errored instance counted as running, + // and the loop's later `Stopped` could not correct that either, + // because by then the old status was `Error` and neither branch + // fires. + // + // The message is assigned after the transition, and that ordering is + // what preserves it. `Error` being outside the set that clears + // `last_error` is belt and braces here, not the mechanism. + source.apply_status(ConnectorStatus::Error, metrics); source.info.last_error = Some(ConnectorError::new(error_message)); } } @@ -236,35 +231,64 @@ impl SourceManager { state, )?; info!("Source connector with ID: {plugin_id} for plugin: {key} initialized successfully."); + // Armed from here until the id is recorded below. `SourceInstanceGuard` + // carries why that window strands the instance. + let instance_guard = + source::SourceInstanceGuard::for_container(container.clone(), plugin_id, key); let (producer, encoder, transforms) = - source::setup_source_producer(key, config, iggy_client).await?; + match source::setup_source_producer(key, config, iggy_client).await { + Ok(parts) => parts, + Err(error) => { + // Awaited rather than left to `drop`, so this error reaches + // the caller after teardown. `drop` stays the net for a + // cancellation and for a `?` added here later. + instance_guard.close().await; + return Err(error); + } + }; let handle_callback = container.iggy_source_handle_v2; let batch_result_callback = container.iggy_source_batch_result; - let handler_tasks = source::spawn_source_handler( - plugin_id, - key, - config.verbose, - config.benchmark, - producer, - encoder, - transforms, - state_storage, - handle_callback, - batch_result_callback, - context.clone(), - ); + // The lock is taken before the spawn so nothing can await between + // registering the tasks and recording the id that reaches them. A + // cancellation in that gap left the `SOURCE_SENDERS` entry and both + // spawned tasks behind with no id naming them, and the forwarding loop + // then ran for the life of the process. The guard closes the plugin + // instance on that path but cannot reach either of those. + // + // The forwarding loop's own first act is to take this lock, so it waits + // for this block to end rather than racing it. { let mut details = details.lock().await; + // Nothing between these three statements may await. The spawn used + // to be passed in as a closure so the compiler refused one; inlined + // here that is a rule rather than a check, so keep it: an await + // between the spawn and the id strands the `SOURCE_SENDERS` entry + // and both tasks with nothing naming them. + details.handler_tasks = source::spawn_source_handler( + plugin_id, + key, + config.verbose, + config.benchmark, + producer, + encoder, + transforms, + state_storage, + handle_callback, + batch_result_callback, + context.clone(), + ); details.info.id = plugin_id; - details.info.status = ConnectorStatus::Running; - details.info.last_error = None; details.config = config.clone(); - details.handler_tasks = handler_tasks; - metrics.increment_sources_running(); + // In the same hold as the id record, not after it. Released first, + // this transition raced the forwarding loop's own report and could + // overwrite an `Error` the loop had already set. + details.apply_status(ConnectorStatus::Running, Some(metrics)); } + // `details.info.id` now names this instance, so a later stop reaches it. + instance_guard.disarm(); Ok(()) } @@ -328,6 +352,36 @@ pub struct SourceDetails { pub restart_guard: Arc>, } +impl SourceDetails { + /// Applies a status transition and the gauge move that belongs with it. + /// + /// On `&mut self` rather than behind a key, so it can run inside a lock the + /// caller already holds. `update_status` takes the lock and delegates; + /// `start_connector` calls it in the same hold as the id record. + /// + /// That matters: applied after releasing that lock, the initial `Running` + /// could land after the forwarding loop had already reported `Running` and + /// then failed its first batch, overwriting `Error`, clearing `last_error`, + /// and counting the instance a second time. + fn apply_status(&mut self, status: ConnectorStatus, metrics: Option<&Arc>) { + let old_status = self.info.status; + self.info.status = status; + if matches!(status, ConnectorStatus::Running | ConnectorStatus::Stopped) { + self.info.last_error = None; + } + let Some(metrics) = metrics else { + return; + }; + // Only a real crossing of `Running` moves the gauge, so repeated + // reports of a status the connector already holds cost nothing. + if old_status != ConnectorStatus::Running && status == ConnectorStatus::Running { + metrics.increment_sources_running(); + } else if old_status == ConnectorStatus::Running && status != ConnectorStatus::Running { + metrics.decrement_sources_running(); + } + } +} + impl fmt::Debug for SourceDetails { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("SourceDetails") @@ -463,6 +517,100 @@ mod tests { assert_eq!(metrics.get_sources_running(), 1); } + #[tokio::test] + async fn should_not_double_count_when_an_error_falls_between_two_running_reports() { + // The interleaving that double counts: the forwarding loop reports + // `Running`, fails its first batch, and a second `Running` report lands + // afterwards. That second report crosses into `Running` again, so it + // increments a gauge the error never gave back, and the instance is + // counted twice. + // + // What the consecutive-`Running` test cannot see: there the second + // report finds the status already `Running`, so no crossing happens and + // no arithmetic is exercised. The error in the middle is the whole point. + let metrics = Arc::new(Metrics::init()); + let mut details = create_test_source_details("pg", 1); + details.info.status = ConnectorStatus::Stopped; + let manager = SourceManager::new(vec![details]); + + manager + .update_status("pg", ConnectorStatus::Running, Some(&metrics)) + .await; + manager + .set_error("pg", "first batch failed", Some(&metrics)) + .await; + assert_eq!( + metrics.get_sources_running(), + 0, + "an instance that has failed is not running, and the gauge has to say so \ + or nothing later can correct it" + ); + + manager + .update_status("pg", ConnectorStatus::Running, Some(&metrics)) + .await; + + assert_eq!( + metrics.get_sources_running(), + 1, + "one instance, however many times its status crossed Running" + ); + } + + #[tokio::test] + async fn should_keep_the_error_message_when_the_status_becomes_error() { + // `set_error` routes through a transition that clears `last_error` for + // some statuses, so its observable contract is worth pinning: the + // status ends `Error` and the message survives. + // + // Worth knowing what this does NOT pin. The message is assigned after + // the transition, so widening the clear to include `Error` leaves this + // green; checked, and the mutant survives. The ordering is the + // mechanism, and no unit test can see an ordering inside one lock hold. + let metrics = Arc::new(Metrics::init()); + let manager = SourceManager::new(vec![create_test_source_details("pg", 1)]); + + manager + .set_error("pg", "producer setup failed", Some(&metrics)) + .await; + + let source = manager.get("pg").await.expect("source must exist"); + let source = source.lock().await; + assert_eq!(source.info.status, ConnectorStatus::Error); + assert_eq!( + source + .info + .last_error + .as_ref() + .map(|error| error.message.as_str()), + Some("producer setup failed"), + "the transition must not clear the message set right after it" + ); + } + + #[tokio::test] + async fn should_increment_metrics_once_when_running_is_reported_twice() { + // Both a start and the forwarding loop report `Running` for the same + // instance, so the gauge has to count instances rather than reports. + let metrics = Arc::new(Metrics::init()); + let mut details = create_test_source_details("pg", 1); + details.info.status = ConnectorStatus::Stopped; + let manager = SourceManager::new(vec![details]); + + manager + .update_status("pg", ConnectorStatus::Running, Some(&metrics)) + .await; + manager + .update_status("pg", ConnectorStatus::Running, Some(&metrics)) + .await; + + assert_eq!( + metrics.get_sources_running(), + 1, + "a second report of a status the connector already has must not move the gauge" + ); + } + #[tokio::test] async fn should_decrement_metrics_when_leaving_running() { let metrics = Arc::new(Metrics::init()); @@ -479,7 +627,7 @@ mod tests { #[tokio::test] async fn should_clear_error_when_status_becomes_running() { let manager = SourceManager::new(vec![create_test_source_details("pg", 1)]); - manager.set_error("pg", "some error").await; + manager.set_error("pg", "some error", None).await; manager .update_status("pg", ConnectorStatus::Running, None) @@ -494,7 +642,7 @@ mod tests { async fn should_set_error_status_and_message() { let manager = SourceManager::new(vec![create_test_source_details("pg", 1)]); - manager.set_error("pg", "connection failed").await; + manager.set_error("pg", "connection failed", None).await; let source = manager.get("pg").await.unwrap(); let details = source.lock().await; @@ -557,7 +705,7 @@ mod tests { #[tokio::test] async fn should_clear_error_when_status_becomes_stopped() { let manager = SourceManager::new(vec![create_test_source_details("pg", 1)]); - manager.set_error("pg", "some error").await; + manager.set_error("pg", "some error", None).await; manager .update_status("pg", ConnectorStatus::Stopped, None) @@ -609,6 +757,6 @@ mod tests { async fn set_error_should_be_noop_for_unknown_key() { let manager = SourceManager::new(vec![]); - manager.set_error("nonexistent", "some error").await; + manager.set_error("nonexistent", "some error", None).await; } } diff --git a/core/connectors/runtime/src/metrics.rs b/core/connectors/runtime/src/metrics.rs index 71b5f93090..5cc1058591 100644 --- a/core/connectors/runtime/src/metrics.rs +++ b/core/connectors/runtime/src/metrics.rs @@ -39,7 +39,7 @@ pub enum ConnectorType { } impl ConnectorType { - fn as_label(&self) -> &'static str { + pub(crate) fn as_label(&self) -> &'static str { match self { ConnectorType::Source => "source", ConnectorType::Sink => "sink", diff --git a/core/connectors/runtime/src/sink.rs b/core/connectors/runtime/src/sink.rs index 7a17724510..88bc43a659 100644 --- a/core/connectors/runtime/src/sink.rs +++ b/core/connectors/runtime/src/sink.rs @@ -19,10 +19,11 @@ use crate::benchmark; use crate::configs::connectors::SinkConfig; use crate::context::RuntimeContext; use crate::log::LOG_CALLBACK; -use crate::metrics::{Metrics, SinkLabels}; +use crate::metrics::{ConnectorType, Metrics, SinkLabels}; use crate::{ FailedPlugin, PLUGIN_ID, RuntimeError, SinkApi, SinkConnector, SinkConnectorConsumer, - SinkConnectorPlugin, SinkConnectorWrapper, resolve_plugin_path, transform, + SinkConnectorPlugin, SinkConnectorWrapper, close_plugin_instance, resolve_plugin_path, + transform, }; use dlopen2::wrapper::Container; use futures::StreamExt; @@ -182,12 +183,8 @@ pub async fn init( let connector = sink_connectors .get_mut(&path) .expect("sink connector was inserted above"); - let close_result = (connector.container.iggy_sink_close)(plugin_id); - if close_result != 0 { - warn!( - "iggy_sink_close returned {close_result} while cleaning up failed sink connector with ID: {plugin_id} ({key})" - ); - } + let close = connector.container.iggy_sink_close; + close_plugin_instance(&|id| close(id), ConnectorType::Sink, plugin_id, &key); if let Some(plugin) = connector .plugins .iter_mut() diff --git a/core/connectors/runtime/src/source.rs b/core/connectors/runtime/src/source.rs index c58d6352c1..7c54108d8f 100644 --- a/core/connectors/runtime/src/source.rs +++ b/core/connectors/runtime/src/source.rs @@ -42,15 +42,17 @@ use crate::benchmark; use crate::configs::connectors::SourceConfig; use crate::context::RuntimeContext; use crate::log::LOG_CALLBACK; +use crate::metrics::ConnectorType; use crate::metrics::SourceLabels; use crate::{ FailedPlugin, PLUGIN_ID, RuntimeError, SourceApi, SourceConnector, SourceConnectorPlugin, - SourceConnectorProducer, SourceConnectorWrapper, resolve_plugin_path, + SourceConnectorProducer, SourceConnectorWrapper, close_plugin_instance, resolve_plugin_path, state::{StateStorage, StateStorageFactory}, transform, }; use iggy_connector_sdk::api::ConnectorStatus; use prometheus_client::metrics::counter::Counter; +use tokio::runtime::Handle; use tokio::task::JoinHandle; const MAX_FAILED_TAIL_RETRIES: u32 = 3; @@ -185,7 +187,7 @@ pub async fn init( source_connectors.insert( path.clone(), SourceConnector { - container, + container: Arc::new(container), plugins: Vec::new(), }, ); @@ -226,6 +228,15 @@ pub async fn init( continue; } + // A plugin left with `error` set is skipped by `handle`, so nothing + // would ever reach the instance `init_source` just created. + let instance_guard = { + let connector = source_connectors + .get_mut(&path) + .expect("source connector was inserted above"); + SourceInstanceGuard::for_container(connector.container.clone(), plugin_id, &key) + }; + match setup_source_producer(&key, &config, iggy_client).await { Ok((producer, encoder, transforms)) => { let connector = source_connectors @@ -238,6 +249,7 @@ pub async fn init( .expect("source plugin was pushed above"); plugin.producer = Some(SourceConnectorProducer { producer, encoder }); plugin.transforms = transforms; + instance_guard.disarm(); info!( "Source container with name: {name} ({key}) initialized successfully with ID: {plugin_id}." ); @@ -245,15 +257,10 @@ pub async fn init( Err(error) => { let message = format!("Failed to set up source producer: {error}"); error!("Source: {name} ({key}) - {message}"); + instance_guard.close().await; let connector = source_connectors .get_mut(&path) .expect("source connector was inserted above"); - let close_result = (connector.container.iggy_source_close)(plugin_id); - if close_result != 0 { - warn!( - "iggy_source_close returned {close_result} while cleaning up failed source connector with ID: {plugin_id} ({key})" - ); - } if let Some(plugin) = connector .plugins .iter_mut() @@ -305,6 +312,115 @@ pub(crate) fn init_source( } } +/// A plugin's `iggy_source_close` together with whatever keeps the library that +/// exports it mapped, so the call stays valid once it is deferred off the +/// calling thread. +pub(crate) type SourceClose = Arc i32 + Send + Sync>; + +/// Closes a source instance that `iggy_source_open` created and nothing else +/// will ever reach. +/// +/// Between `init_source` succeeding and the plugin id reaching `SourceDetails`, +/// the instance exists inside the plugin and nothing outside it knows the id: +/// `stop_connector` closes whatever `details.info.id` holds, which is still the +/// previous instance. An early return there stranded the new one for the life +/// of the process. A guard rather than a cleanup branch per fallible call, +/// because the window is those two statements rather than whichever call +/// between them is fallible today, so a `?` added inside it stays correct. +/// Startup and restart both hand off through it. +/// +/// Teardown runs two ways and they are not interchangeable. [`Self::close`] +/// awaits, so an error returned after it means the instance is already gone +/// and an immediate retry has nothing to collide with. `Drop` cannot await, so +/// it hands the work to the blocking pool; the closure carries the container, +/// which is what keeps the library mapped until the call returns. +#[must_use = "dropping an armed guard closes the source instance"] +pub(crate) struct SourceInstanceGuard { + /// `Some` while this guard owns the instance, `None` once something else + /// does. One representation rather than a close plus a flag that had to + /// agree with it, and taking it is what lets both teardown paths run + /// without cloning the callback. + close: Option, + plugin_id: u32, + key: String, +} + +impl SourceInstanceGuard { + /// Arms a guard over an instance the caller has just opened through + /// `container`, which it captures rather than borrows for the reason the + /// type documents. + pub(crate) fn for_container( + container: Arc>, + plugin_id: u32, + key: &str, + ) -> Self { + Self::new( + Arc::new(move |id| (container.iggy_source_close)(id)), + plugin_id, + key, + ) + } + + /// Kept behind `for_container` so no production caller can build a guard + /// that holds a close pointer without its library. Tests pass a closure. + fn new(close: SourceClose, plugin_id: u32, key: &str) -> Self { + Self { + close: Some(close), + plugin_id, + key: key.to_owned(), + } + } + + /// Hands ownership of the instance to the caller, once something else can + /// close it. Call only after the plugin id is recorded on `SourceDetails`. + pub(crate) fn disarm(mut self) { + self.close = None; + } + + /// The awaited half of the teardown the type documents. Error arms call it + /// rather than leaving the work to `Drop`, which cannot offer the ordering. + pub(crate) async fn close(mut self) { + let Some(close) = self.close.take() else { + return; + }; + let plugin_id = self.plugin_id; + let key = std::mem::take(&mut self.key); + if tokio::task::spawn_blocking(move || { + close_plugin_instance(close.as_ref(), ConnectorType::Source, plugin_id, &key) + }) + .await + .is_err() + { + warn!( + "Teardown of failed source connector with ID: {plugin_id} did not run to completion." + ); + } + } +} + +impl Drop for SourceInstanceGuard { + fn drop(&mut self) { + let Some(close) = self.close.take() else { + return; + }; + + let plugin_id = self.plugin_id; + let key = std::mem::take(&mut self.key); + // `SourceContainer::close` drives the plugin's own `close()` under + // `block_on` and runs for as long as the plugin takes, so it goes to + // the blocking pool where blocking is what the thread is for. + match Handle::try_current() { + Ok(handle) => { + handle.spawn_blocking(move || { + close_plugin_instance(close.as_ref(), ConnectorType::Source, plugin_id, &key) + }); + } + // No runtime to hand it to, and no worker to protect either. + Err(_) => close_plugin_instance(close.as_ref(), ConnectorType::Source, plugin_id, &key), + } + } +} + pub(crate) async fn setup_source_producer( key: &str, config: &SourceConfig, @@ -532,7 +648,10 @@ pub(crate) async fn source_forwarding_loop( matches!(pending_state_error.as_ref(), Some(SdkError::StateLatched)) || (pending_state_error.is_none() && state_latched); if !preserve_original_error { - context.sources.set_error(&plugin_key, &error_msg).await; + context + .sources + .set_error(&plugin_key, &error_msg, Some(&context.metrics)) + .await; } } else { context @@ -573,7 +692,10 @@ pub(crate) async fn source_forwarding_loop( ); error!("{error_msg}"); context.metrics.inc_errors_with_labels(&labels.counter); - context.sources.set_error(&plugin_key, &error_msg).await; + context + .sources + .set_error(&plugin_key, &error_msg, Some(&context.metrics)) + .await; } } } else { @@ -602,7 +724,10 @@ pub(crate) async fn source_forwarding_loop( ); error!("{error_msg}"); context.metrics.inc_errors_with_labels(&labels.counter); - context.sources.set_error(&plugin_key, &error_msg).await; + context + .sources + .set_error(&plugin_key, &error_msg, Some(&context.metrics)) + .await; } } @@ -936,7 +1061,9 @@ mod tests { use super::*; use std::collections::VecDeque; use std::future::ready; + use std::sync::Mutex; use std::sync::atomic::{AtomicU32, Ordering}; + use std::time::Duration; static TEST_PLUGIN_ID: AtomicU32 = AtomicU32::new(u32::MAX / 2); @@ -961,6 +1088,166 @@ mod tests { } } + /// A close that records the ids it was handed and answers `result`. + /// + /// The guard takes its close as a closure, so each test owns its recorder + /// and nothing is shared between tests. An `extern "C" fn` cannot capture, + /// which is what used to force this through statics. + fn recording_close(result: i32) -> (SourceClose, Arc>>) { + let closed = Arc::new(Mutex::new(Vec::new())); + let recorded = closed.clone(); + ( + Arc::new(move |id| { + recorded.lock().expect("close recorder").push(id); + result + }), + closed, + ) + } + + /// The shape `start_connector` has: a guard armed over an instance nothing + /// else knows about, then a fallible step whose `?` returns before anything + /// records the id. + fn start_with_fallible_step( + close: SourceClose, + plugin_id: u32, + step: Result<(), RuntimeError>, + ) -> Result<(), RuntimeError> { + let instance_guard = SourceInstanceGuard::new(close, plugin_id, "random"); + step?; + instance_guard.disarm(); + Ok(()) + } + + #[test] + fn given_armed_guard_when_dropped_should_close_the_instance() { + // The leak this exists for: `init_source` has created the instance and + // nothing outside the plugin knows its id yet, so an early return here + // would strand it for the life of the process. + let plugin_id = next_plugin_id(); + let (close, closed) = recording_close(0); + + drop(SourceInstanceGuard::new(close, plugin_id, "random")); + + assert_eq!( + *closed.lock().expect("close recorder"), + vec![plugin_id], + "a guard still armed owns the instance and must close exactly it" + ); + } + + #[test] + fn given_fallible_step_when_it_returns_early_should_close_the_instance() { + // The shape dropping or disarming inline cannot show, and the one the + // guard is there for: the `?` leaves with the guard still armed and + // never reaches `disarm`. The error arms call `close()` directly now, + // so this is the net under a `?` added inside the window later. + let plugin_id = next_plugin_id(); + let (close, closed) = recording_close(0); + + let result = start_with_fallible_step( + close, + plugin_id, + Err(RuntimeError::InvalidConfiguration("injected".to_string())), + ); + + assert!(result.is_err(), "the injected failure has to propagate"); + assert_eq!( + *closed.lock().expect("close recorder"), + vec![plugin_id], + "a `?` must not strand the instance it left behind" + ); + } + + #[test] + fn given_fallible_step_when_it_succeeds_should_leave_the_instance_open() { + // The other half of the same helper: reaching `disarm` hands the + // instance on rather than closing it. + let plugin_id = next_plugin_id(); + let (close, closed) = recording_close(0); + + let result = start_with_fallible_step(close, plugin_id, Ok(())); + + assert!(result.is_ok()); + assert!( + closed.lock().expect("close recorder").is_empty(), + "a step that succeeded leaves the instance for the manager to close" + ); + } + + #[tokio::test] + async fn given_armed_guard_when_dropped_in_runtime_should_close_off_the_worker() { + // `drop` cannot await, and `SourceContainer::close` drives the plugin's + // own teardown under `block_on`, so closing here would hold a worker for + // however long the plugin takes. It goes to the blocking pool instead, + // which is what the differing thread asserts. The close still has to + // happen. + let plugin_id = next_plugin_id(); + let dropping_thread = std::thread::current().id(); + let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel(); + + drop(SourceInstanceGuard::new( + Arc::new(move |id| { + let _ = sender.send((id, std::thread::current().id())); + 0 + }), + plugin_id, + "random", + )); + + let (closed_id, closing_thread) = + tokio::time::timeout(Duration::from_secs(5), receiver.recv()) + .await + .expect("the deferred close should run") + .expect("the deferred close should report the instance"); + assert_eq!( + closed_id, plugin_id, + "the deferred close must reach the instance the guard was armed over" + ); + assert_ne!( + closing_thread, dropping_thread, + "closing on the dropping thread holds it for the plugin's teardown" + ); + } + + #[tokio::test] + async fn given_armed_guard_when_closed_should_finish_before_returning() { + // What the error arms rely on: once `close()` has returned the instance + // is gone, so the error they return cannot reach an operator who then + // retries into a collision with it. + let plugin_id = next_plugin_id(); + let (close, closed) = recording_close(0); + + SourceInstanceGuard::new(close, plugin_id, "random") + .close() + .await; + + assert_eq!( + *closed.lock().expect("close recorder"), + vec![plugin_id], + "close() has to await the teardown, and the drop after it must not repeat it" + ); + } + + #[test] + fn given_refused_close_when_guard_drops_should_close_once_and_swallow_refusal() { + // The plugin answers -1 for an id it does not know. Both callers are + // already returning an error of their own, so the refusal is reported + // and not propagated: unwinding out of `drop` would be worse than the + // leak it is cleaning up after. The harness gives no-panic for free, so + // what this asserts is the single call. + let plugin_id = next_plugin_id(); + let (close, closed) = recording_close(-1); + + drop(SourceInstanceGuard::new(close, plugin_id, "random")); + + assert_eq!( + *closed.lock().expect("close recorder"), + vec![plugin_id], + "a refusal must not become a retry or a second close" + ); + } + #[test] fn given_serialized_batch_when_callback_runs_should_forward_batch_id() { let plugin_id = next_plugin_id(); diff --git a/core/connectors/sdk/src/sink.rs b/core/connectors/sdk/src/sink.rs index 332f73c4e0..cbf8974f83 100644 --- a/core/connectors/sdk/src/sink.rs +++ b/core/connectors/sdk/src/sink.rs @@ -267,6 +267,13 @@ macro_rules! sink_connector { let mut container = SinkContainer::new(id); let result = container.open(id, config_ptr, config_len, log_callback, <$type>::new); + if result != 0 { + // Rolled back rather than registered, for the reason the + // source macro gives: a failed open is still stored on the + // container, and registering it strands an instance nothing + // outside can name to close. + return result; + } INSTANCES.insert(id, container); result } diff --git a/core/connectors/sdk/src/source.rs b/core/connectors/sdk/src/source.rs index 532d2643f9..a264a56df4 100644 --- a/core/connectors/sdk/src/source.rs +++ b/core/connectors/sdk/src/source.rs @@ -594,6 +594,16 @@ macro_rules! source_connector { log_callback, <$type>::new, ); + if result != 0 { + // Rolled back rather than registered. `open` stores the + // instance on the container whatever it returns, so a failed + // one would sit here for the life of the process: the runtime + // gets an error back before it has recorded the id, so nothing + // outside can name it to close it. Dropping the container is + // the rollback, and it releases whatever the plugin took + // before it failed. + return result; + } INSTANCES.insert(id, container); result } diff --git a/core/integration/tests/connectors/random/random_source.rs b/core/integration/tests/connectors/random/random_source.rs index 772a2a2250..42a3f6c51b 100644 --- a/core/integration/tests/connectors/random/random_source.rs +++ b/core/integration/tests/connectors/random/random_source.rs @@ -27,7 +27,10 @@ use tokio::time::{sleep, timeout}; const API_KEY: &str = "test-api-key"; const SOURCE_KEY: &str = "random"; const RETRY_INTERVAL: Duration = Duration::from_millis(100); -const STATE_STABILITY_WINDOW: Duration = Duration::from_secs(1); +/// How long a counter is given to settle after the change that moves it. +/// Shared by the state-file and gauge waits: both are waiting on the same +/// thing, a report that may land just after the poll that preceded it. +const SETTLE_WINDOW: Duration = Duration::from_secs(1); const WAIT_TIMEOUT: Duration = Duration::from_secs(10); #[iggy_harness( @@ -54,7 +57,7 @@ async fn state_save_failure_preserves_state_and_source_recovers(harness: &TestHa .connectors_runtime() .expect("connectors runtime") .http_url(); - let http = Client::new(); + let http = client(); let errors_before_failure = source_errors(&http, &api_url).await; let state_dir = state_path.parent().expect("source state directory"); let unavailable_state_dir = state_dir.with_extension("unavailable"); @@ -68,7 +71,7 @@ async fn state_save_failure_preserves_state_and_source_recovers(harness: &TestHa .expect("source state should remain readable"); wait_for_source_error_after(&http, &api_url, errors_before_failure).await; - sleep(STATE_STABILITY_WINDOW).await; + sleep(SETTLE_WINDOW).await; assert_eq!( tokio::fs::read(&unavailable_state_path) .await @@ -97,6 +100,102 @@ async fn state_save_failure_preserves_state_and_source_recovers(harness: &TestHa random_source_liveness::assert_produces_messages(harness).await; } +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/random/source.toml")), + seed = seeds::connector_stream +)] +async fn sources_running_does_not_climb_across_restarts(harness: &TestHarness) { + // The gauge counts running instances, and a restart takes one down before + // bringing one up, so one configured source stays at one however often it + // is restarted. It used to be reported by two mechanisms and taken back by + // one. + let api_url = harness + .connectors_runtime() + .expect("connectors runtime") + .http_url(); + let http = client(); + + wait_for_sources_running(&http, &api_url, 1).await; + + for round in 1..=3 { + let response = http + .post(format!("{api_url}/sources/{SOURCE_KEY}/restart")) + .header("api-key", API_KEY) + .send() + .await + .expect("restart request should be sent"); + assert_eq!( + response.status().as_u16(), + 204, + "restart {round} should be accepted" + ); + + wait_for_sources_running(&http, &api_url, 1).await; + } + + // Read it once more after the gauge has settled. A report that lands after + // the poll above would otherwise go unseen. + sleep(SETTLE_WINDOW).await; + assert_eq!( + sources_running(&http, &api_url).await, + 1, + "one running source must stay counted once, whatever it took to restart it" + ); +} + +/// Every request carries [`WAIT_TIMEOUT`], so none of them can outlive the +/// wait they belong to. Without it a stalled runtime hangs the test rather +/// than failing it, and a hung test reports nothing at all. +fn client() -> Client { + Client::builder() + .timeout(WAIT_TIMEOUT) + .build() + .expect("the test client must build") +} + +/// The one request the stats helpers share. Handing back the `Result` rather +/// than unwrapping it is what lets the retry loops keep treating a failed read +/// as "not yet" while the direct readers keep failing on it. +async fn fetch_stats(http: &Client, api_url: &str) -> reqwest::Result { + http.get(format!("{api_url}/stats")) + .header("api-key", API_KEY) + .send() + .await? + .json::() + .await +} + +async fn sources_running(http: &Client, api_url: &str) -> u32 { + fetch_stats(http, api_url) + .await + .expect("runtime stats should be valid") + .sources_running +} + +async fn wait_for_sources_running(http: &Client, api_url: &str, expected: u32) { + // The last value the loop actually saw, rather than a fresh read in the + // failure message. That read was the one request with no budget over it: + // it only runs once the wait has already timed out, which is exactly when + // the runtime is stalled, so the test hung instead of failing and reported + // nothing at all. + let mut last = None; + let reached = timeout(WAIT_TIMEOUT, async { + loop { + let running = sources_running(http, api_url).await; + last = Some(running); + if running == expected { + return; + } + sleep(RETRY_INTERVAL).await; + } + }) + .await; + assert!( + reached.is_ok(), + "sources_running never reached {expected}; last read {last:?}" + ); +} + async fn wait_for_state_file(state_path: &Path) { timeout(Duration::from_secs(5), async { while !state_path.exists() { @@ -108,13 +207,7 @@ async fn wait_for_state_file(state_path: &Path) { } async fn source_errors(http: &Client, api_url: &str) -> u64 { - let stats = http - .get(format!("{api_url}/stats")) - .header("api-key", API_KEY) - .send() - .await - .expect("runtime stats should be available") - .json::() + let stats = fetch_stats(http, api_url) .await .expect("runtime stats should be valid"); stats @@ -128,12 +221,7 @@ async fn source_errors(http: &Client, api_url: &str) -> u64 { async fn wait_for_source_error_after(http: &Client, api_url: &str, previous_errors: u64) { timeout(WAIT_TIMEOUT, async { loop { - if let Ok(response) = http - .get(format!("{api_url}/stats")) - .header("api-key", API_KEY) - .send() - .await - && let Ok(stats) = response.json::().await + if let Ok(stats) = fetch_stats(http, api_url).await && let Some(source) = stats .connectors .iter()