From 9c79474963a779bf35950d22b6564f6c8259c135 Mon Sep 17 00:00:00 2001 From: mlevkov Date: Fri, 4 Sep 2026 23:28:42 -0700 Subject: [PATCH 01/15] fix(connectors): close the source instance a failed start leaves behind Closes #4062. `start_connector` allocates a fresh plugin id, calls `init_source`, and only records that id on `SourceDetails` once the handler tasks are spawned. In between, 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. `setup_source_producer` returning early through `?` therefore stranded the new one for the life of the process, while the boot path in `source::init` cleaned up on the identical failure. For a plugin whose open only allocates, the orphan is wasted memory. For one that takes a process-global resource, it is a live fault: a shared listener stays bound and answering into a queue nothing drains, and every retried restart then fails on the identity the orphan never released. A guard rather than a cleanup branch at the one call that can fail today, because the window is defined by the two statements that open and record the instance, not by which call between them happens to be fallible. Adding a `?` inside it stays correct. The close-and-report itself is shared with the boot path so the two cannot drift. --- core/connectors/runtime/src/manager/source.rs | 8 + core/connectors/runtime/src/source.rs | 146 +++++++++++++++++- 2 files changed, 148 insertions(+), 6 deletions(-) diff --git a/core/connectors/runtime/src/manager/source.rs b/core/connectors/runtime/src/manager/source.rs index 8ef68e8b19..b30d1f2697 100644 --- a/core/connectors/runtime/src/manager/source.rs +++ b/core/connectors/runtime/src/manager/source.rs @@ -236,6 +236,12 @@ impl SourceManager { state, )?; info!("Source connector with ID: {plugin_id} for plugin: {key} initialized successfully."); + // Armed from here until the id is recorded below. Until then nothing + // outside the plugin knows this instance exists, so any early return + // would strand it: `stop_connector` closes `details.info.id`, which + // still names the previous one. + let instance = + source::SourceInstanceGuard::new(container.iggy_source_close, plugin_id, key); let (producer, encoder, transforms) = source::setup_source_producer(key, config, iggy_client).await?; @@ -265,6 +271,8 @@ impl SourceManager { details.handler_tasks = handler_tasks; metrics.increment_sources_running(); } + // `details.info.id` now names this instance, so a later stop reaches it. + instance.disarm(); Ok(()) } diff --git a/core/connectors/runtime/src/source.rs b/core/connectors/runtime/src/source.rs index c58d6352c1..8037c56c0e 100644 --- a/core/connectors/runtime/src/source.rs +++ b/core/connectors/runtime/src/source.rs @@ -248,12 +248,7 @@ pub async fn init( 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})" - ); - } + close_failed_source(connector.container.iggy_source_close, plugin_id, &key); if let Some(plugin) = connector .plugins .iter_mut() @@ -305,6 +300,63 @@ pub(crate) fn init_source( } } +/// Closes a source instance that `iggy_source_open` created and nothing else +/// will ever reach. +/// +/// Between `init_source` succeeding and the plugin id being recorded on +/// `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 in that window +/// stranded the new one for the life of the process. +/// +/// A guard rather than a cleanup branch at the one call site that can fail +/// today, because the window is defined by the two statements that open and +/// record the instance, not by which call between them happens to be fallible. +/// Adding a `?` inside it stays correct. +pub(crate) struct SourceInstanceGuard<'a> { + close: extern "C" fn(u32) -> i32, + plugin_id: u32, + key: &'a str, + armed: bool, +} + +impl<'a> SourceInstanceGuard<'a> { + pub(crate) fn new(close: extern "C" fn(u32) -> i32, plugin_id: u32, key: &'a str) -> Self { + Self { + close, + plugin_id, + key, + armed: true, + } + } + + /// Hands ownership of the instance to the caller, once something else can + /// close it. Call only after the plugin id is durably recorded. + pub(crate) fn disarm(mut self) { + self.armed = false; + } +} + +impl Drop for SourceInstanceGuard<'_> { + fn drop(&mut self) { + if self.armed { + close_failed_source(self.close, self.plugin_id, self.key); + } + } +} + +/// Closes an instance whose setup did not finish, reporting a refusal rather +/// than returning it: both callers are already on a failure path and have an +/// error of their own to surface. +pub(crate) fn close_failed_source(close: extern "C" fn(u32) -> i32, plugin_id: u32, key: &str) { + let close_result = 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})" + ); + } +} + pub(crate) async fn setup_source_producer( key: &str, config: &SourceConfig, @@ -961,6 +1013,88 @@ mod tests { } } + /// Records what `SourceInstanceGuard` passed to the FFI. A stub has to be a + /// plain `extern "C" fn`, so recording goes through statics rather than a + /// captured closure. Each test therefore gets its **own** stub and statics: + /// sharing one pair would make two tests that both reset and read it race, + /// since the suite runs them in the same process at the same time. + static ARMED_CLOSED_ID: AtomicU32 = AtomicU32::new(0); + static ARMED_CALLS: AtomicU32 = AtomicU32::new(0); + + extern "C" fn armed_close(id: u32) -> i32 { + ARMED_CLOSED_ID.store(id, Ordering::SeqCst); + ARMED_CALLS.fetch_add(1, Ordering::SeqCst); + 0 + } + + static DISARMED_CALLS: AtomicU32 = AtomicU32::new(0); + + extern "C" fn disarmed_close(_id: u32) -> i32 { + DISARMED_CALLS.fetch_add(1, Ordering::SeqCst); + 0 + } + + static REFUSED_CALLS: AtomicU32 = AtomicU32::new(0); + + extern "C" fn refusing_close(_id: u32) -> i32 { + REFUSED_CALLS.fetch_add(1, Ordering::SeqCst); + -1 + } + + #[test] + fn given_an_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(); + + drop(SourceInstanceGuard::new(armed_close, plugin_id, "random")); + + assert_eq!( + ARMED_CALLS.load(Ordering::SeqCst), + 1, + "a guard still armed owns the instance and must close it" + ); + assert_eq!( + ARMED_CLOSED_ID.load(Ordering::SeqCst), + plugin_id, + "closing any other id would leave this instance open and kill a live one" + ); + } + + #[test] + fn given_a_disarmed_guard_when_dropped_should_leave_the_instance_open() { + // Disarmed means `details.info.id` names the instance, so `stop_connector` + // will close it. Closing here too would tear down a source that just + // started successfully. + let plugin_id = next_plugin_id(); + + SourceInstanceGuard::new(disarmed_close, plugin_id, "random").disarm(); + + assert_eq!( + DISARMED_CALLS.load(Ordering::SeqCst), + 0, + "the instance is the manager's once its id is recorded" + ); + } + + #[test] + fn given_a_refused_close_when_guard_drops_should_not_panic() { + // 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. + let plugin_id = next_plugin_id(); + + drop(SourceInstanceGuard::new( + refusing_close, + plugin_id, + "random", + )); + + assert_eq!(REFUSED_CALLS.load(Ordering::SeqCst), 1); + } + #[test] fn given_serialized_batch_when_callback_runs_should_forward_batch_id() { let plugin_id = next_plugin_id(); From 1f2c2a490f30ebf23f72a7f4482fb5c21b449e4c Mon Sep 17 00:00:00 2001 From: mlevkov Date: Tue, 8 Sep 2026 13:25:28 -0700 Subject: [PATCH 02/15] fix(connectors): tie the source close pointer to its library The guard held a bare `extern "C" fn` read out of a dlopened `.so`. Nothing tied it to the `Container` that owns the mapping. It worked only because `container` is declared before the guard and therefore drops after it. A declaration order dependency is thin support for an FFI call into a loaded library, and it stops being enough at all once the call is deferred off the calling thread. `for_container` now captures the `Arc>` inside the closure the guard calls, so the mapping is owned for as long as the close is reachable. The field is `Arc i32 + Send + Sync>` rather than an `Arc>`. A `Container` only exists via dlopen, so a guard carrying that field cannot be constructed in a unit test at all, and this guard needs more tests rather than fewer. The single production constructor takes the container, so the ownership stays enforced by the API and not by convention. `close_failed_source` now takes one callable shape. An `extern "C" fn` does not implement `Fn`, so the boot path wraps its pointer at the call site. The doc comment claimed the plugin id was "durably recorded". `SourceDetails` is memory only and the id never reaches the state store, so it now says recorded on `SourceDetails`. --- core/connectors/runtime/src/manager/source.rs | 2 +- core/connectors/runtime/src/source.rs | 56 ++++++++++++++----- 2 files changed, 43 insertions(+), 15 deletions(-) diff --git a/core/connectors/runtime/src/manager/source.rs b/core/connectors/runtime/src/manager/source.rs index b30d1f2697..8f2f90a001 100644 --- a/core/connectors/runtime/src/manager/source.rs +++ b/core/connectors/runtime/src/manager/source.rs @@ -241,7 +241,7 @@ impl SourceManager { // would strand it: `stop_connector` closes `details.info.id`, which // still names the previous one. let instance = - source::SourceInstanceGuard::new(container.iggy_source_close, plugin_id, key); + source::SourceInstanceGuard::for_container(container.clone(), plugin_id, key); let (producer, encoder, transforms) = source::setup_source_producer(key, config, iggy_client).await?; diff --git a/core/connectors/runtime/src/source.rs b/core/connectors/runtime/src/source.rs index 8037c56c0e..355fc30284 100644 --- a/core/connectors/runtime/src/source.rs +++ b/core/connectors/runtime/src/source.rs @@ -248,7 +248,8 @@ pub async fn init( let connector = source_connectors .get_mut(&path) .expect("source connector was inserted above"); - close_failed_source(connector.container.iggy_source_close, plugin_id, &key); + let close = connector.container.iggy_source_close; + close_failed_source(&|id| close(id), plugin_id, &key); if let Some(plugin) = connector .plugins .iter_mut() @@ -313,34 +314,56 @@ pub(crate) fn init_source( /// today, because the window is defined by the two statements that open and /// record the instance, not by which call between them happens to be fallible. /// Adding a `?` inside it stays correct. -pub(crate) struct SourceInstanceGuard<'a> { - close: extern "C" fn(u32) -> i32, +pub(crate) struct SourceInstanceGuard { + close: Arc i32 + Send + Sync>, plugin_id: u32, - key: &'a str, + key: String, armed: bool, } -impl<'a> SourceInstanceGuard<'a> { - pub(crate) fn new(close: extern "C" fn(u32) -> i32, plugin_id: u32, key: &'a str) -> Self { +impl SourceInstanceGuard { + /// Arms a guard over an instance the caller has just opened through + /// `container`. + /// + /// The captured `Arc` is the point. `iggy_source_close` is a pointer read + /// out of a `dlopen`ed library and stays callable only while something + /// keeps that library mapped, so the guard owns the container rather than + /// relying on it being declared before the guard and therefore dropped + /// after it. + 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: Arc i32 + Send + Sync>, plugin_id: u32, key: &str) -> Self { Self { close, plugin_id, - key, + key: key.to_owned(), armed: true, } } /// Hands ownership of the instance to the caller, once something else can - /// close it. Call only after the plugin id is durably recorded. + /// close it. Call only after the plugin id is recorded on `SourceDetails`. pub(crate) fn disarm(mut self) { self.armed = false; } } -impl Drop for SourceInstanceGuard<'_> { +impl Drop for SourceInstanceGuard { fn drop(&mut self) { if self.armed { - close_failed_source(self.close, self.plugin_id, self.key); + close_failed_source(self.close.as_ref(), self.plugin_id, &self.key); } } } @@ -348,7 +371,7 @@ impl Drop for SourceInstanceGuard<'_> { /// Closes an instance whose setup did not finish, reporting a refusal rather /// than returning it: both callers are already on a failure path and have an /// error of their own to surface. -pub(crate) fn close_failed_source(close: extern "C" fn(u32) -> i32, plugin_id: u32, key: &str) { +pub(crate) fn close_failed_source(close: &dyn Fn(u32) -> i32, plugin_id: u32, key: &str) { let close_result = close(plugin_id); if close_result != 0 { warn!( @@ -1048,7 +1071,11 @@ mod tests { // would strand it for the life of the process. let plugin_id = next_plugin_id(); - drop(SourceInstanceGuard::new(armed_close, plugin_id, "random")); + drop(SourceInstanceGuard::new( + Arc::new(move |id| armed_close(id)), + plugin_id, + "random", + )); assert_eq!( ARMED_CALLS.load(Ordering::SeqCst), @@ -1069,7 +1096,8 @@ mod tests { // started successfully. let plugin_id = next_plugin_id(); - SourceInstanceGuard::new(disarmed_close, plugin_id, "random").disarm(); + SourceInstanceGuard::new(Arc::new(move |id| disarmed_close(id)), plugin_id, "random") + .disarm(); assert_eq!( DISARMED_CALLS.load(Ordering::SeqCst), @@ -1087,7 +1115,7 @@ mod tests { let plugin_id = next_plugin_id(); drop(SourceInstanceGuard::new( - refusing_close, + Arc::new(move |id| refusing_close(id)), plugin_id, "random", )); From 20d03bc7d35bb6b5bd847503079945fd27a0cf4e Mon Sep 17 00:00:00 2001 From: mlevkov Date: Tue, 8 Sep 2026 13:28:11 -0700 Subject: [PATCH 03/15] fix(connectors): keep plugin teardown off the worker in drop glue `SourceContainer::close` drives the plugin's own `close()` under `block_on`, and `block_on(handle)` before it. Calling that from the guard's `drop` put an unbounded plugin teardown on a tokio worker, in the one place no timeout can ever be added, because drop glue cannot await. `drop` now hands the close to the blocking pool. The closure it carries owns the container, so the library stays mapped until the call returns. That is what makes deferring safe. Deferring costs the ordering between teardown and the returned error, so the known failure arm no longer relies on `drop`. `setup_source_producer`'s error arm awaits `close()`: off the worker and ordered, so by the time the error reaches an operator the instance is gone and a restart retried straight away cannot collide with it. `drop` stays the net for a cancellation and for any `?` added in the window later. `Handle::try_current` picks between them. A guard dropped outside a runtime has no worker to protect and nothing to hand the work to, so it closes inline. That is also the path the plain `#[test]` cases take. The deferred test asserts the close lands on a different thread from the one that dropped the guard, which is the only observable difference between handing it over and running it inline. The ordered test asserts exactly one call after `close()` returns, so the teardown was awaited and the following drop did not repeat it. `instance` became `instance_guard` while these lines were being rewritten. It holds a guard, and `instance.disarm()` read as disarming the instance. --- core/connectors/runtime/src/manager/source.rs | 16 ++- core/connectors/runtime/src/source.rs | 105 +++++++++++++++++- 2 files changed, 116 insertions(+), 5 deletions(-) diff --git a/core/connectors/runtime/src/manager/source.rs b/core/connectors/runtime/src/manager/source.rs index 8f2f90a001..96224850f6 100644 --- a/core/connectors/runtime/src/manager/source.rs +++ b/core/connectors/runtime/src/manager/source.rs @@ -240,11 +240,21 @@ impl SourceManager { // outside the plugin knows this instance exists, so any early return // would strand it: `stop_connector` closes `details.info.id`, which // still names the previous one. - let 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) => { + // Closed here rather than left to `drop` so this error + // reaches the caller after teardown, not alongside it. + // `drop` stays the net for a cancellation, and for any `?` + // added inside this window 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; @@ -272,7 +282,7 @@ impl SourceManager { metrics.increment_sources_running(); } // `details.info.id` now names this instance, so a later stop reaches it. - instance.disarm(); + instance_guard.disarm(); Ok(()) } diff --git a/core/connectors/runtime/src/source.rs b/core/connectors/runtime/src/source.rs index 355fc30284..ebd45845d4 100644 --- a/core/connectors/runtime/src/source.rs +++ b/core/connectors/runtime/src/source.rs @@ -51,6 +51,7 @@ use crate::{ }; 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; @@ -358,12 +359,50 @@ impl SourceInstanceGuard { pub(crate) fn disarm(mut self) { self.armed = false; } + + /// Closes the instance and waits for the plugin to finish, so an error the + /// caller returns afterwards means the instance is already gone. A restart + /// retried straight away then has nothing left to collide with. + /// + /// `drop` cannot offer that ordering, which is why the error arms call this + /// instead of relying on it. + pub(crate) async fn close(mut self) { + self.armed = false; + let close = self.close.clone(); + let plugin_id = self.plugin_id; + let key = std::mem::take(&mut self.key); + if tokio::task::spawn_blocking(move || close_failed_source(close.as_ref(), 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) { - if self.armed { - close_failed_source(self.close.as_ref(), self.plugin_id, &self.key); + if !self.armed { + return; + } + + let close = self.close.clone(); + let plugin_id = self.plugin_id; + let key = std::mem::take(&mut self.key); + // Nothing can await here, so the plugin's teardown cannot be bounded + // here either: `SourceContainer::close` drives the plugin's own + // `close()` under `block_on`, and that runs for as long as the plugin + // takes. Hand it to the blocking pool, where blocking is what the + // thread is for. The closure carries the container, so the library + // stays mapped until the call returns. + match Handle::try_current() { + Ok(handle) => { + handle.spawn_blocking(move || close_failed_source(close.as_ref(), plugin_id, &key)); + } + // No runtime to hand it to, and no worker to protect either. + Err(_) => close_failed_source(close.as_ref(), plugin_id, &key), } } } @@ -1012,6 +1051,7 @@ mod tests { use std::collections::VecDeque; use std::future::ready; use std::sync::atomic::{AtomicU32, Ordering}; + use std::time::Duration; static TEST_PLUGIN_ID: AtomicU32 = AtomicU32::new(u32::MAX / 2); @@ -1106,6 +1146,67 @@ mod tests { ); } + #[tokio::test] + async fn given_an_armed_guard_when_dropped_inside_a_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_an_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 calls = Arc::new(AtomicU32::new(0)); + let recorded = calls.clone(); + + let guard = SourceInstanceGuard::new( + Arc::new(move |_id| { + recorded.fetch_add(1, Ordering::SeqCst); + 0 + }), + plugin_id, + "random", + ); + guard.close().await; + + assert_eq!( + calls.load(Ordering::SeqCst), + 1, + "close() must await the teardown, and the drop that follows must not repeat it" + ); + } + #[test] fn given_a_refused_close_when_guard_drops_should_not_panic() { // The plugin answers -1 for an id it does not know. Both callers are From a1722c5bf4d0900f0422d300fc38c81b5abae427 Mon Sep 17 00:00:00 2001 From: mlevkov Date: Tue, 8 Sep 2026 13:30:55 -0700 Subject: [PATCH 04/15] fix(connectors): stop a cancelled start from leaking both tasks The guard closed the plugin instance but could not reach the runtime half of the same leak. `spawn_source_handler` registers the `SOURCE_SENDERS` entry and spawns both tasks, and the id that reaches them was recorded only after the next `details.lock().await`. A cancellation in that gap, which is what a client disconnect dropping the axum handler future does, left the entry and both tasks behind with nothing naming them, so the forwarding loop ran for the life of the process. The lock is now taken before the spawn, so the spawn and the id record sit in one block with no await between them. `spawn_source_handler` is synchronous, so holding the lock across it costs a spawn and nothing else. The forwarding loop's own first act is to take the same lock, so it waits for the block to end instead of racing it. No test covers this. Reaching it means cancelling the future at one specific await with a dlopened container and a live broker in place, and both integration routes were already rejected for this PR for reasons that still hold. What enforces it is that the block contains no await, and an await added inside it would reopen the window silently. --- core/connectors/runtime/src/manager/source.rs | 37 ++++++++++++------- 1 file changed, 23 insertions(+), 14 deletions(-) diff --git a/core/connectors/runtime/src/manager/source.rs b/core/connectors/runtime/src/manager/source.rs index 96224850f6..5436256d82 100644 --- a/core/connectors/runtime/src/manager/source.rs +++ b/core/connectors/runtime/src/manager/source.rs @@ -258,27 +258,36 @@ impl SourceManager { 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 that 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. + // + // `spawn_source_handler` is synchronous, so holding the lock across it + // costs a spawn. The forwarding loop's own first act is to take this + // lock, so it simply waits for this block to end. { let mut details = details.lock().await; + 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(); } // `details.info.id` now names this instance, so a later stop reaches it. From abacd38d27d4ad50000a9c1462feb36bcad23c51 Mon Sep 17 00:00:00 2001 From: mlevkov Date: Tue, 8 Sep 2026 13:33:23 -0700 Subject: [PATCH 05/15] fix(connectors): stop sources_running climbing on every restart A start wrote `Running` onto `SourceDetails` and called `increment_sources_running` by hand. The forwarding loop reports `Running` through `update_status` as well, because the boot path spawns it without coming through `start_connector`, and a stop decrements once. Two reports against one decrement, so the gauge climbed with every restart. Whether it climbed depended on which of the two reached the lock first. Reported by hand first, the loop's `update_status` saw `Running` already set and did nothing, so the ratchet needed the loop to win the race. `start_connector` now reports through `update_status` like everything else. That moves the gauge only on a real transition and clears `last_error` on the way. The loop's own report becomes a no-op here and stays the only report on the boot path. Reported rather than deleted outright, so a connector that just started does not answer `Stopped` to `GET /sources` until the loop takes the lock. The new test pins what single ownership rests on: a second report of a status the connector already holds does not move the gauge. The deletion itself is not covered, because reaching `start_connector` needs a dlopened container. --- core/connectors/runtime/src/manager/source.rs | 35 +++++++++++++++++-- 1 file changed, 32 insertions(+), 3 deletions(-) diff --git a/core/connectors/runtime/src/manager/source.rs b/core/connectors/runtime/src/manager/source.rs index 5436256d82..4139cd9407 100644 --- a/core/connectors/runtime/src/manager/source.rs +++ b/core/connectors/runtime/src/manager/source.rs @@ -285,14 +285,20 @@ impl SourceManager { context.clone(), ); details.info.id = plugin_id; - details.info.status = ConnectorStatus::Running; - details.info.last_error = None; details.config = config.clone(); - metrics.increment_sources_running(); } // `details.info.id` now names this instance, so a later stop reaches it. instance_guard.disarm(); + // Through `update_status`, which is the only thing that moves the gauge + // and moves it only on a real transition. Writing the status here and + // incrementing by hand as well meant two mechanisms counting one start: + // the forwarding loop reports `Running` too, for the boot path that + // never comes through here, and a stop decrements once. Reported twice + // and taken back once, the gauge climbed with every restart. + self.update_status(key, ConnectorStatus::Running, Some(metrics)) + .await; + Ok(()) } @@ -490,6 +496,29 @@ mod tests { assert_eq!(metrics.get_sources_running(), 1); } + #[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()); From cbc4a31e0317767128ad30c21c28c1456aa50965 Mon Sep 17 00:00:00 2001 From: mlevkov Date: Tue, 8 Sep 2026 13:37:01 -0700 Subject: [PATCH 06/15] test(connectors): make the guard tests test the guard The disarmed test asserted a counter that started at 0 and that nothing else touched, so it held whether or not a guard had ever been built. It now closes an armed guard through the same recorder first, so the recorder is proven able to move before it is required not to. None of the three covered the shape the guard exists for. Dropping or disarming inline is something the compiler already guarantees; what the guard is for is a `?` returning with it still armed. `start_with_fallible_step` mirrors that control flow and takes an injected failure, so both halves are covered: the early return closes the instance, and reaching `disarm` does not. The error arm in `start_connector` calls `close()` directly now, so this is the net under a `?` added inside the window later. The three `extern "C" fn` stubs and their four statics are gone. A guard takes its close as a closure, so each test owns its recorder and there was never anything to share. That also removes the reason the stubs were split up, which was a race `next_plugin_id` had already made unreachable. Names lost their articles to match `given_serialized_batch_...` in the same module, and the refused-close test now says what it proves rather than that it does not panic, which the harness gives for free. `SourceClose` names the close type because clippy's `type_complexity` asked for a name once the test helper returned one. --- core/connectors/runtime/src/source.rs | 203 ++++++++++++++++---------- 1 file changed, 124 insertions(+), 79 deletions(-) diff --git a/core/connectors/runtime/src/source.rs b/core/connectors/runtime/src/source.rs index ebd45845d4..b90aead166 100644 --- a/core/connectors/runtime/src/source.rs +++ b/core/connectors/runtime/src/source.rs @@ -302,6 +302,11 @@ pub(crate) fn init_source( } } +/// A plugin's `iggy_source_close` together with whatever keeps the library that +/// exports it mapped. Held instead of the bare `extern "C" fn` 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. /// @@ -316,7 +321,7 @@ pub(crate) fn init_source( /// record the instance, not by which call between them happens to be fallible. /// Adding a `?` inside it stays correct. pub(crate) struct SourceInstanceGuard { - close: Arc i32 + Send + Sync>, + close: SourceClose, plugin_id: u32, key: String, armed: bool, @@ -345,7 +350,7 @@ impl SourceInstanceGuard { /// 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: Arc i32 + Send + Sync>, plugin_id: u32, key: &str) -> Self { + fn new(close: SourceClose, plugin_id: u32, key: &str) -> Self { Self { close, plugin_id, @@ -1050,6 +1055,7 @@ 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; @@ -1076,83 +1082,127 @@ mod tests { } } - /// Records what `SourceInstanceGuard` passed to the FFI. A stub has to be a - /// plain `extern "C" fn`, so recording goes through statics rather than a - /// captured closure. Each test therefore gets its **own** stub and statics: - /// sharing one pair would make two tests that both reset and read it race, - /// since the suite runs them in the same process at the same time. - static ARMED_CLOSED_ID: AtomicU32 = AtomicU32::new(0); - static ARMED_CALLS: AtomicU32 = AtomicU32::new(0); - - extern "C" fn armed_close(id: u32) -> i32 { - ARMED_CLOSED_ID.store(id, Ordering::SeqCst); - ARMED_CALLS.fetch_add(1, Ordering::SeqCst); - 0 - } - - static DISARMED_CALLS: AtomicU32 = AtomicU32::new(0); - - extern "C" fn disarmed_close(_id: u32) -> i32 { - DISARMED_CALLS.fetch_add(1, Ordering::SeqCst); - 0 + /// 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, + ) } - static REFUSED_CALLS: AtomicU32 = AtomicU32::new(0); - - extern "C" fn refusing_close(_id: u32) -> i32 { - REFUSED_CALLS.fetch_add(1, Ordering::SeqCst); - -1 + /// 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_an_armed_guard_when_dropped_should_close_the_instance() { + 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( - Arc::new(move |id| armed_close(id)), - plugin_id, - "random", - )); + 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_disarmed_guard_when_dropped_should_leave_the_instance_open() { + // Disarmed means `details.info.id` names the instance, so + // `stop_connector` will close it. Closing here too would tear down a + // source that just started successfully. + // + // The armed guard goes first so the recorder is proven to move before + // it is required not to. Asserting an empty recorder on its own holds + // whether or not a guard was ever built. + let armed_id = next_plugin_id(); + let (close, closed) = recording_close(0); + drop(SourceInstanceGuard::new(close.clone(), armed_id, "random")); assert_eq!( - ARMED_CALLS.load(Ordering::SeqCst), - 1, - "a guard still armed owns the instance and must close it" + *closed.lock().expect("close recorder"), + vec![armed_id], + "this recorder has to be able to move, or the assertion below is vacuous" ); + + SourceInstanceGuard::new(close, next_plugin_id(), "random").disarm(); + assert_eq!( - ARMED_CLOSED_ID.load(Ordering::SeqCst), - plugin_id, - "closing any other id would leave this instance open and kill a live one" + *closed.lock().expect("close recorder"), + vec![armed_id], + "the instance is the manager's once its id is recorded" ); } #[test] - fn given_a_disarmed_guard_when_dropped_should_leave_the_instance_open() { - // Disarmed means `details.info.id` names the instance, so `stop_connector` - // will close it. Closing here too would tear down a source that just - // started successfully. + 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); - SourceInstanceGuard::new(Arc::new(move |id| disarmed_close(id)), plugin_id, "random") - .disarm(); + 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!( - DISARMED_CALLS.load(Ordering::SeqCst), - 0, - "the instance is the manager's once its id is recorded" + *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_an_armed_guard_when_dropped_inside_a_runtime_should_close_off_the_worker() { + 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. + // 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(); @@ -1182,46 +1232,41 @@ mod tests { } #[tokio::test] - async fn given_an_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. + 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 calls = Arc::new(AtomicU32::new(0)); - let recorded = calls.clone(); + let (close, closed) = recording_close(0); - let guard = SourceInstanceGuard::new( - Arc::new(move |_id| { - recorded.fetch_add(1, Ordering::SeqCst); - 0 - }), - plugin_id, - "random", - ); - guard.close().await; + SourceInstanceGuard::new(close, plugin_id, "random") + .close() + .await; assert_eq!( - calls.load(Ordering::SeqCst), - 1, - "close() must await the teardown, and the drop that follows must not repeat it" + *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_a_refused_close_when_guard_drops_should_not_panic() { + 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. + // 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( - Arc::new(move |id| refusing_close(id)), - plugin_id, - "random", - )); + drop(SourceInstanceGuard::new(close, plugin_id, "random")); - assert_eq!(REFUSED_CALLS.load(Ordering::SeqCst), 1); + assert_eq!( + *closed.lock().expect("close recorder"), + vec![plugin_id], + "a refusal must not become a retry or a second close" + ); } #[test] From 0dcf36957199b687371e01dc7aea1d6c1ce9fbbc Mon Sep 17 00:00:00 2001 From: mlevkov Date: Tue, 8 Sep 2026 13:41:18 -0700 Subject: [PATCH 07/15] refactor(connectors): one close for a plugin instance that never started The guard's doc comment argued against a cleanup branch that `init` still had, so one of the two had to go. `init` now uses the guard as well, which retires the duplicated close this PR deliberately left behind. That needed `SourceConnector.container` to be an `Arc`, which costs nothing: `main.rs` wrapped the container in one the moment `init` returned, so the allocation moved earlier rather than being added. `Arc>` still coerces for `get_plugin_version` and `init_source`. In `init` the instance is not unreachable in the same way it is on the restart path, since its id is in the map. What makes it unreachable is that `handle` skips a plugin with `error` set, so nothing ever reaches the instance `init_source` created. The error arm awaits `close()` for the same ordering the restart path takes. `close_failed_source` closed a plugin instance rather than a connector, and `sink.rs` had the same body inline one word apart. Both now call `close_plugin_instance`, which takes "source" or "sink". The two log lines are unchanged, character for character. Not taken from that finding: it also suggests the helper for the stop path at `manager/source.rs:167`. That path drops the close result and logs "Closed" regardless, which is a separate bug and out of this diff. --- core/connectors/runtime/src/main.rs | 24 +++++++++++++-- core/connectors/runtime/src/sink.rs | 11 +++---- core/connectors/runtime/src/source.rs | 43 ++++++++++++++------------- 3 files changed, 48 insertions(+), 30 deletions(-) diff --git a/core/connectors/runtime/src/main.rs b/core/connectors/runtime/src/main.rs index 12d627c84a..0591af86eb 100644 --- a/core/connectors/runtime/src/main.rs +++ b/core/connectors/runtime/src/main.rs @@ -186,7 +186,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 +456,28 @@ 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. +/// +/// `kind` is "source" or "sink". The two sides had this body inline, one word +/// apart. +pub(crate) fn close_plugin_instance( + close: &dyn Fn(u32) -> i32, + kind: &str, + plugin_id: u32, + key: &str, +) { + let close_result = close(plugin_id); + if close_result != 0 { + 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/sink.rs b/core/connectors/runtime/src/sink.rs index 7a17724510..5798d0912e 100644 --- a/core/connectors/runtime/src/sink.rs +++ b/core/connectors/runtime/src/sink.rs @@ -22,7 +22,8 @@ use crate::log::LOG_CALLBACK; use crate::metrics::{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), "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 b90aead166..6822274d1b 100644 --- a/core/connectors/runtime/src/source.rs +++ b/core/connectors/runtime/src/source.rs @@ -45,7 +45,7 @@ use crate::log::LOG_CALLBACK; 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, }; @@ -186,7 +186,7 @@ pub async fn init( source_connectors.insert( path.clone(), SourceConnector { - container, + container: Arc::new(container), plugins: Vec::new(), }, ); @@ -227,6 +227,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 @@ -239,6 +248,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}." ); @@ -246,11 +256,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 = connector.container.iggy_source_close; - close_failed_source(&|id| close(id), plugin_id, &key); if let Some(plugin) = connector .plugins .iter_mut() @@ -376,9 +385,11 @@ impl SourceInstanceGuard { let close = self.close.clone(); let plugin_id = self.plugin_id; let key = std::mem::take(&mut self.key); - if tokio::task::spawn_blocking(move || close_failed_source(close.as_ref(), plugin_id, &key)) - .await - .is_err() + if tokio::task::spawn_blocking(move || { + close_plugin_instance(close.as_ref(), "source", plugin_id, &key) + }) + .await + .is_err() { warn!( "Teardown of failed source connector with ID: {plugin_id} did not run to completion." @@ -404,26 +415,16 @@ impl Drop for SourceInstanceGuard { // stays mapped until the call returns. match Handle::try_current() { Ok(handle) => { - handle.spawn_blocking(move || close_failed_source(close.as_ref(), plugin_id, &key)); + handle.spawn_blocking(move || { + close_plugin_instance(close.as_ref(), "source", plugin_id, &key) + }); } // No runtime to hand it to, and no worker to protect either. - Err(_) => close_failed_source(close.as_ref(), plugin_id, &key), + Err(_) => close_plugin_instance(close.as_ref(), "source", plugin_id, &key), } } } -/// Closes an instance whose setup did not finish, reporting a refusal rather -/// than returning it: both callers are already on a failure path and have an -/// error of their own to surface. -pub(crate) fn close_failed_source(close: &dyn Fn(u32) -> i32, plugin_id: u32, key: &str) { - let close_result = 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})" - ); - } -} - pub(crate) async fn setup_source_producer( key: &str, config: &SourceConfig, From 9c7bb0d2171b86cdb5c0169b63c004410bf1a146 Mon Sep 17 00:00:00 2001 From: mlevkov Date: Tue, 8 Sep 2026 13:42:38 -0700 Subject: [PATCH 08/15] refactor(connectors): make a discarded guard a build error `SourceInstanceGuard::new(..);` as a bare statement builds an armed guard and drops it on the spot, closing the instance it was meant to protect. `#[must_use]` plus `warnings = "deny"` turns that into a build error rather than a puzzle. Checked by writing one: `error: unused source::SourceInstanceGuard that must be used`. The call site in `start_connector` repeated the type's own second paragraph almost word for word. It now says only what it adds, which is where arming starts and where it stops. The type's third paragraph argued against a cleanup branch that `init` no longer has, so it says why a guard rather than why not a branch. --- core/connectors/runtime/src/manager/source.rs | 6 ++---- core/connectors/runtime/src/source.rs | 9 +++++---- 2 files changed, 7 insertions(+), 8 deletions(-) diff --git a/core/connectors/runtime/src/manager/source.rs b/core/connectors/runtime/src/manager/source.rs index 4139cd9407..df3f4e5309 100644 --- a/core/connectors/runtime/src/manager/source.rs +++ b/core/connectors/runtime/src/manager/source.rs @@ -236,10 +236,8 @@ impl SourceManager { state, )?; info!("Source connector with ID: {plugin_id} for plugin: {key} initialized successfully."); - // Armed from here until the id is recorded below. Until then nothing - // outside the plugin knows this instance exists, so any early return - // would strand it: `stop_connector` closes `details.info.id`, which - // still names the previous one. + // 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); diff --git a/core/connectors/runtime/src/source.rs b/core/connectors/runtime/src/source.rs index 6822274d1b..2a6c1d6a37 100644 --- a/core/connectors/runtime/src/source.rs +++ b/core/connectors/runtime/src/source.rs @@ -325,10 +325,11 @@ pub(crate) type SourceClose = Arc i32 + Send + Sync>; /// which is still the previous instance. An early return in that window /// stranded the new one for the life of the process. /// -/// A guard rather than a cleanup branch at the one call site that can fail -/// today, because the window is defined by the two statements that open and -/// record the instance, not by which call between them happens to be fallible. -/// Adding a `?` inside it stays correct. +/// A guard rather than a cleanup branch on each fallible call, because the +/// window is defined by the two statements that open and record the instance, +/// not by which call between them happens to be fallible today. Adding a `?` +/// inside it stays correct. Both call sites use it. +#[must_use = "dropping an armed guard closes the source instance"] pub(crate) struct SourceInstanceGuard { close: SourceClose, plugin_id: u32, From d90b72fa56836bff6e34295a1e6a44b2f11c5e98 Mon Sep 17 00:00:00 2001 From: mlevkov Date: Tue, 8 Sep 2026 14:20:17 -0700 Subject: [PATCH 09/15] test(connectors): assert one source stays counted once across restarts Nothing asserted `sources_running` with a source actually configured. `stats_endpoint_returns_runtime_stats` reads it with none, so it checks 0. This restarts the random source three times through the API and requires the gauge back at 1 each time, then reads it once more after a settle window so a late report is not missed. What it does not do is isolate the fix it was written for. Reverting the report through `update_status`, putting the direct write and the hand increment back, leaves it passing 3 of 3, because taking the lock before the spawn already makes `start_connector` win the race to that lock. Reverting that too, so the code is what it was before either change, still leaves it passing 6 of 6: `start_connector` acquires an uncontended mutex immediately, while the spawned loop has to be scheduled and then walk a DashMap before it reaches the same lock, so in practice it does not get there first. So the double count was real in the code and latent in behaviour. This guards the invariant rather than proving the defect, which is worth having anyway: it is the only assertion on the gauge with a source running, and it covers the source restart path that only the state and CDC tests touched. --- .../tests/connectors/random/random_source.rs | 73 +++++++++++++++++++ 1 file changed, 73 insertions(+) diff --git a/core/integration/tests/connectors/random/random_source.rs b/core/integration/tests/connectors/random/random_source.rs index 772a2a2250..cd0db025db 100644 --- a/core/integration/tests/connectors/random/random_source.rs +++ b/core/integration/tests/connectors/random/random_source.rs @@ -97,6 +97,79 @@ 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::new(); + + 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(STATE_STABILITY_WINDOW).await; + assert_eq!( + sources_running(&http, &api_url).await, + 1, + "one running source must stay counted once, whatever it took to restart it" + ); +} + +async fn sources_running(http: &Client, api_url: &str) -> u32 { + http.get(format!("{api_url}/stats")) + .header("api-key", API_KEY) + .send() + .await + .expect("runtime stats should be available") + .json::() + .await + .expect("runtime stats should be valid") + .sources_running +} + +async fn wait_for_sources_running(http: &Client, api_url: &str, expected: u32) { + let observed = timeout(WAIT_TIMEOUT, async { + loop { + let running = sources_running(http, api_url).await; + if running == expected { + return running; + } + sleep(RETRY_INTERVAL).await; + } + }) + .await; + assert!( + observed.is_ok(), + "sources_running never reached {expected}; last read {}", + sources_running(http, api_url).await + ); +} + async fn wait_for_state_file(state_path: &Path) { timeout(Duration::from_secs(5), async { while !state_path.exists() { From 0c4b859f86d243522a33b4becfc02e533eaf7aaa Mon Sep 17 00:00:00 2001 From: mlevkov Date: Tue, 8 Sep 2026 14:52:15 -0700 Subject: [PATCH 10/15] fix(connectors): make an await in the start window a compile error Holding the spawn and the id record in one lock block was correct but rested on a comment. Nothing stopped an await being added between them later, and that reopens the cancellation window silently: the `SOURCE_SENDERS` entry and both spawned tasks are left with no id naming them, and no guard can reach either. `SourceDetails::record_started` takes the spawn as a closure and is deliberately not `async`, so the two happen together and the compiler refuses an await between them. Checked by adding one: `error[E0728]: await is only allowed inside async functions and blocks`. It also makes the step testable, which nothing in that region was. The test asserts both halves land, because a stop reaches the instance through the id and drains it through the tasks, so losing either leaves something behind. Dropping the tasks in the recorder fails it. The cancellation itself still has no test and cannot have one: it needs the future dropped at one await, with a dlopened container live. The difference is that the invariant no longer depends on anyone reading the comment. --- core/connectors/runtime/src/manager/source.rs | 80 ++++++++++++++----- 1 file changed, 61 insertions(+), 19 deletions(-) diff --git a/core/connectors/runtime/src/manager/source.rs b/core/connectors/runtime/src/manager/source.rs index df3f4e5309..c33049e00b 100644 --- a/core/connectors/runtime/src/manager/source.rs +++ b/core/connectors/runtime/src/manager/source.rs @@ -257,33 +257,32 @@ impl SourceManager { let handle_callback = container.iggy_source_handle_v2; let batch_result_callback = container.iggy_source_batch_result; - // The lock is taken before the spawn so that nothing can await between + // 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. // - // `spawn_source_handler` is synchronous, so holding the lock across it - // costs a spawn. The forwarding loop's own first act is to take this - // lock, so it simply waits for this block to end. + // 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; - 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.config = config.clone(); + details.record_started(plugin_id, config, || { + 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` now names this instance, so a later stop reaches it. instance_guard.disarm(); @@ -359,6 +358,27 @@ pub struct SourceDetails { pub restart_guard: Arc>, } +impl SourceDetails { + /// Records an instance that has just started, spawning its handlers in the + /// same breath. + /// + /// Deliberately not `async`, and that is the point. The id has to be + /// recorded under the same lock hold as the spawn: a cancellation between + /// the two strands the `SOURCE_SENDERS` entry and both tasks with nothing + /// naming them, which no guard can reach. Taking `spawn` as a closure is + /// what lets the compiler refuse an await added between them. + fn record_started( + &mut self, + plugin_id: u32, + config: &SourceConfig, + spawn: impl FnOnce() -> Vec>, + ) { + self.handler_tasks = spawn(); + self.info.id = plugin_id; + self.config = config.clone(); + } +} + impl fmt::Debug for SourceDetails { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("SourceDetails") @@ -494,6 +514,28 @@ mod tests { assert_eq!(metrics.get_sources_running(), 1); } + #[tokio::test] + async fn record_started_should_store_the_id_and_the_spawned_tasks() { + // Both have to land under one lock hold, so they are recorded together + // and there is nowhere to await between them. A stop reaches the + // instance through the id and drains it through the tasks, so losing + // either leaves something behind. + let mut details = create_test_source_details("pg", 1); + let config = details.config.clone(); + + details.record_started(7, &config, || vec![tokio::spawn(async {})]); + + assert_eq!( + details.info.id, 7, + "a later stop closes whatever id this recorded" + ); + assert_eq!( + details.handler_tasks.len(), + 1, + "a stop drains the tasks recorded here, so they cannot be dropped" + ); + } + #[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 From 898c6a9a1feca4d8ea544da135dcdc624201cb4d Mon Sep 17 00:00:00 2001 From: mlevkov Date: Thu, 10 Sep 2026 12:19:46 -0700 Subject: [PATCH 11/15] fix(connectors): move the status transition and the gauge together The initial `Running` was reported after the startup lock was released, so it raced the forwarding loop. The loop reports `Running` itself and can fail its first batch immediately, and the delayed report then overwrote `Error`, cleared `last_error`, and crossed into `Running` a second time. spetz measured the result: `Running`, no `last_error`, and `sources_running = 2` for one instance. Two halves, and the second is why the count reached 2. `set_error` moved the status out of `Running` without moving the gauge, so the increment was never given back, and the loop's later `Stopped` could not correct it either: by then the old status was `Error` and neither branch of the gauge rule fires. `SourceDetails::apply_status` now owns the transition and the gauge move together. `update_status` takes the lock and delegates, `set_error` delegates and then sets its message, and `start_connector` calls it in the same hold as the id record, so nothing can land between the two. The trailing unconditional report is gone. The regression test puts the error between the two `Running` reports, which is what the consecutive-`Running` test cannot reach: there the second report finds the status already `Running`, so no crossing happens and no arithmetic runs. Removing the gauge move from `set_error` leaves the gauge at 1 where it should read 0, and the test fails on it. The second test pins `set_error`'s observable contract, status `Error` with the message intact, and its comment says what it does not pin: the message is assigned after the transition, so widening the clear to include `Error` leaves it green. Checked, and that mutant survives. The ordering is the mechanism, and no unit test can observe an ordering inside a single lock hold. Not changed here: `SinkManager::set_error` has the same gauge asymmetry, but the sink path is outside this PR's two files and its consume loop never reports status at all, so it needs its own change. --- core/connectors/runtime/src/manager/source.rs | 149 ++++++++++++++---- core/connectors/runtime/src/source.rs | 15 +- 2 files changed, 131 insertions(+), 33 deletions(-) diff --git a/core/connectors/runtime/src/manager/source.rs b/core/connectors/runtime/src/manager/source.rs index c33049e00b..70f97c4b23 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)); } } @@ -283,19 +278,14 @@ impl SourceManager { context.clone(), ) }); + // 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(); - // Through `update_status`, which is the only thing that moves the gauge - // and moves it only on a real transition. Writing the status here and - // incrementing by hand as well meant two mechanisms counting one start: - // the forwarding loop reports `Running` too, for the boot path that - // never comes through here, and a stop decrements once. Reported twice - // and taken back once, the gauge climbed with every restart. - self.update_status(key, ConnectorStatus::Running, Some(metrics)) - .await; - Ok(()) } @@ -359,6 +349,34 @@ pub struct SourceDetails { } 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(); + } + } + /// Records an instance that has just started, spawning its handlers in the /// same breath. /// @@ -536,6 +554,77 @@ mod tests { ); } + #[tokio::test] + async fn should_not_double_count_when_an_error_falls_between_two_running_reports() { + // The interleaving spetz measured on #4064: 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 @@ -575,7 +664,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) @@ -590,7 +679,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; @@ -653,7 +742,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) @@ -705,6 +794,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/source.rs b/core/connectors/runtime/src/source.rs index 2a6c1d6a37..0120dea1e0 100644 --- a/core/connectors/runtime/src/source.rs +++ b/core/connectors/runtime/src/source.rs @@ -653,7 +653,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 @@ -694,7 +697,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 { @@ -723,7 +729,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; } } From cddcc5d78fdc5cf7b9341c6b39de9342fbdc51b0 Mon Sep 17 00:00:00 2001 From: mlevkov Date: Sat, 12 Sep 2026 11:41:24 -0700 Subject: [PATCH 12/15] fix(connectors): do not register a source or sink whose open failed `iggy_source_open` stored the container in the instance map whatever the open returned, and `SourceContainer::open` assigns the source before it looks at the result, so a failed open left a fully constructed instance behind. The runtime gets its error back before it has recorded the plugin id, so nothing outside the plugin can name that instance to close it, and it stays for the life of the process holding whatever the plugin took before it failed. The rollback is not registering it. Dropping the container releases the instance the same way any other failed construction is released, and the duplicate id guard above is untouched, so reopening the same id still refuses rather than silently replacing a live instance. Sinks had the same shape and get the same fix. Not covered by a test: the FFI entry points are `cfg(not(test))`, so a unit test cannot call them, and the map lives inside the plugin where the runtime cannot observe it. The change is small enough to read, which is the argument for it rather than around it. --- core/connectors/sdk/src/sink.rs | 7 +++++++ core/connectors/sdk/src/source.rs | 10 ++++++++++ 2 files changed, 17 insertions(+) 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 } From 3ec33fd5de550a1b4b416554e74ed4ec8190471a Mon Sep 17 00:00:00 2001 From: mlevkov Date: Sat, 12 Sep 2026 11:41:24 -0700 Subject: [PATCH 13/15] test(connectors): stop the runtime stats waits from hanging on a stalled runtime The wait for a gauge value reported its last read by issuing another request, and that request was the only one with no budget over it. It runs only after the wait has already timed out, which is precisely when the runtime is stalled, and the client had no timeout either, so the assertion that should have failed hung instead and reported nothing. The loop now carries the value it saw, and every request the file makes is built with the wait timeout on it. The three stats readers were also the same request and the same decode written out three times. They share one helper that hands back the `Result`, which is what keeps the retry loop treating a failed read as "not yet" while the two direct readers keep failing on it. The settle window was named for state storage and used for a gauge. Both waits are waiting on the same thing, a report that may land just after the poll before it, so the constant says that instead. --- .../tests/connectors/random/random_source.rs | 67 ++++++++++++------- 1 file changed, 41 insertions(+), 26 deletions(-) diff --git a/core/integration/tests/connectors/random/random_source.rs b/core/integration/tests/connectors/random/random_source.rs index cd0db025db..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 @@ -110,7 +113,7 @@ async fn sources_running_does_not_climb_across_restarts(harness: &TestHarness) { .connectors_runtime() .expect("connectors runtime") .http_url(); - let http = Client::new(); + let http = client(); wait_for_sources_running(&http, &api_url, 1).await; @@ -132,7 +135,7 @@ async fn sources_running_does_not_climb_across_restarts(harness: &TestHarness) { // Read it once more after the gauge has settled. A report that lands after // the poll above would otherwise go unseen. - sleep(STATE_STABILITY_WINDOW).await; + sleep(SETTLE_WINDOW).await; assert_eq!( sources_running(&http, &api_url).await, 1, @@ -140,33 +143,56 @@ async fn sources_running_does_not_climb_across_restarts(harness: &TestHarness) { ); } -async fn sources_running(http: &Client, api_url: &str) -> u32 { +/// 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 - .expect("runtime stats should be available") + .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) { - let observed = timeout(WAIT_TIMEOUT, async { + // 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 running; + return; } sleep(RETRY_INTERVAL).await; } }) .await; assert!( - observed.is_ok(), - "sources_running never reached {expected}; last read {}", - sources_running(http, api_url).await + reached.is_ok(), + "sources_running never reached {expected}; last read {last:?}" ); } @@ -181,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 @@ -201,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() From 7ba585312480f5f50bd5e3a0cd278441bc3d20f8 Mon Sep 17 00:00:00 2001 From: mlevkov Date: Sat, 12 Sep 2026 11:44:29 -0700 Subject: [PATCH 14/15] refactor(connectors): give the instance guard one representation of ownership The guard carried a close callback and an `armed` flag that had to agree with it, so both teardown paths cloned the callback to leave the flag behind them. `Option` is the same state said once: `disarm` clears it, the awaited close and `Drop` each take it, and neither clones. Awaited close still finishes before its caller returns, and `Drop` still hands the work to the blocking pool with the container captured so the library stays mapped. `record_started` is inlined at its one call site. It existed so the spawn was passed as a closure and the compiler would refuse an await between the spawn and the id record. That enforcement goes with it, so the requirement is written where the statements are: an await between them strands the `SOURCE_SENDERS` entry and both tasks with nothing naming them. Registration, the status transition and the disarm stay inside the one lock hold, in that order. Two tests go too. The callback's test tested the callback. The disarmed-guard drop is already covered by the success half of the fallible-step helper, which runs a guard through `disarm` and asserts nothing was closed. --- core/connectors/runtime/src/manager/source.rs | 76 +++++-------------- core/connectors/runtime/src/source.rs | 49 +++--------- 2 files changed, 32 insertions(+), 93 deletions(-) diff --git a/core/connectors/runtime/src/manager/source.rs b/core/connectors/runtime/src/manager/source.rs index 70f97c4b23..85cec50a34 100644 --- a/core/connectors/runtime/src/manager/source.rs +++ b/core/connectors/runtime/src/manager/source.rs @@ -263,21 +263,26 @@ impl SourceManager { // for this block to end rather than racing it. { let mut details = details.lock().await; - details.record_started(plugin_id, config, || { - source::spawn_source_handler( - plugin_id, - key, - config.verbose, - config.benchmark, - producer, - encoder, - transforms, - state_storage, - handle_callback, - batch_result_callback, - context.clone(), - ) - }); + // 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.config = config.clone(); // 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. @@ -376,25 +381,6 @@ impl SourceDetails { metrics.decrement_sources_running(); } } - - /// Records an instance that has just started, spawning its handlers in the - /// same breath. - /// - /// Deliberately not `async`, and that is the point. The id has to be - /// recorded under the same lock hold as the spawn: a cancellation between - /// the two strands the `SOURCE_SENDERS` entry and both tasks with nothing - /// naming them, which no guard can reach. Taking `spawn` as a closure is - /// what lets the compiler refuse an await added between them. - fn record_started( - &mut self, - plugin_id: u32, - config: &SourceConfig, - spawn: impl FnOnce() -> Vec>, - ) { - self.handler_tasks = spawn(); - self.info.id = plugin_id; - self.config = config.clone(); - } } impl fmt::Debug for SourceDetails { @@ -532,28 +518,6 @@ mod tests { assert_eq!(metrics.get_sources_running(), 1); } - #[tokio::test] - async fn record_started_should_store_the_id_and_the_spawned_tasks() { - // Both have to land under one lock hold, so they are recorded together - // and there is nowhere to await between them. A stop reaches the - // instance through the id and drains it through the tasks, so losing - // either leaves something behind. - let mut details = create_test_source_details("pg", 1); - let config = details.config.clone(); - - details.record_started(7, &config, || vec![tokio::spawn(async {})]); - - assert_eq!( - details.info.id, 7, - "a later stop closes whatever id this recorded" - ); - assert_eq!( - details.handler_tasks.len(), - 1, - "a stop drains the tasks recorded here, so they cannot be dropped" - ); - } - #[tokio::test] async fn should_not_double_count_when_an_error_falls_between_two_running_reports() { // The interleaving spetz measured on #4064: the forwarding loop reports diff --git a/core/connectors/runtime/src/source.rs b/core/connectors/runtime/src/source.rs index 0120dea1e0..c9cedd4353 100644 --- a/core/connectors/runtime/src/source.rs +++ b/core/connectors/runtime/src/source.rs @@ -331,10 +331,13 @@ pub(crate) type SourceClose = Arc i32 + Send + Sync>; /// inside it stays correct. Both call sites use it. #[must_use = "dropping an armed guard closes the source instance"] pub(crate) struct SourceInstanceGuard { - close: SourceClose, + /// `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, - armed: bool, } impl SourceInstanceGuard { @@ -362,17 +365,16 @@ impl SourceInstanceGuard { /// that holds a close pointer without its library. Tests pass a closure. fn new(close: SourceClose, plugin_id: u32, key: &str) -> Self { Self { - close, + close: Some(close), plugin_id, key: key.to_owned(), - armed: true, } } /// 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.armed = false; + self.close = None; } /// Closes the instance and waits for the plugin to finish, so an error the @@ -382,8 +384,9 @@ impl SourceInstanceGuard { /// `drop` cannot offer that ordering, which is why the error arms call this /// instead of relying on it. pub(crate) async fn close(mut self) { - self.armed = false; - let close = self.close.clone(); + 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 || { @@ -401,11 +404,10 @@ impl SourceInstanceGuard { impl Drop for SourceInstanceGuard { fn drop(&mut self) { - if !self.armed { + let Some(close) = self.close.take() else { return; - } + }; - let close = self.close.clone(); let plugin_id = self.plugin_id; let key = std::mem::take(&mut self.key); // Nothing can await here, so the plugin's teardown cannot be bounded @@ -1141,33 +1143,6 @@ mod tests { ); } - #[test] - fn given_disarmed_guard_when_dropped_should_leave_the_instance_open() { - // Disarmed means `details.info.id` names the instance, so - // `stop_connector` will close it. Closing here too would tear down a - // source that just started successfully. - // - // The armed guard goes first so the recorder is proven to move before - // it is required not to. Asserting an empty recorder on its own holds - // whether or not a guard was ever built. - let armed_id = next_plugin_id(); - let (close, closed) = recording_close(0); - drop(SourceInstanceGuard::new(close.clone(), armed_id, "random")); - assert_eq!( - *closed.lock().expect("close recorder"), - vec![armed_id], - "this recorder has to be able to move, or the assertion below is vacuous" - ); - - SourceInstanceGuard::new(close, next_plugin_id(), "random").disarm(); - - assert_eq!( - *closed.lock().expect("close recorder"), - vec![armed_id], - "the instance is the manager's once its id is recorded" - ); - } - #[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 From 1d5753fcb9b1317dbb37a544ddda2e8f9badb3ee Mon Sep 17 00:00:00 2001 From: mlevkov Date: Sat, 12 Sep 2026 11:47:33 -0700 Subject: [PATCH 15/15] refactor(connectors): type the cleanup label and say the argument once `close_plugin_instance` took the word "source" or "sink" as a string its signature did not constrain, while `ConnectorType` already defines exactly those two labels. It takes the enum now, and `as_label` is visible in the crate rather than only to the encoder. The helper also sat between two connector struct declarations; it moves above them. The cleanup argument was written out across the close type, the guard type, its constructor, the awaited close, the `Drop` body and both call sites, mostly repeating itself. It is on the guard type now, once, and still says all of it: the window and why it is a guard rather than a branch per fallible call, that startup and restart both hand off through it, that the library must stay mapped for a deferred call, that `Drop` offloads to the blocking pool, and that awaited close and `Drop` are not interchangeable because only one gives the caller an ordering. The other sites point at it. A test comment recorded which person measured an interleaving and on which PR. It states the interleaving instead. --- core/connectors/runtime/src/main.rs | 9 ++- core/connectors/runtime/src/manager/source.rs | 9 ++- core/connectors/runtime/src/metrics.rs | 2 +- core/connectors/runtime/src/sink.rs | 4 +- core/connectors/runtime/src/source.rs | 59 ++++++++----------- 5 files changed, 39 insertions(+), 44 deletions(-) diff --git a/core/connectors/runtime/src/main.rs b/core/connectors/runtime/src/main.rs index 0591af86eb..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; @@ -460,16 +461,18 @@ struct SinkConnectorWrapper { /// rather than returning it: every caller is already on a failure path with an /// error of its own to surface. /// -/// `kind` is "source" or "sink". The two sides had this body inline, one word -/// apart. +/// 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: &str, + 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})" ); diff --git a/core/connectors/runtime/src/manager/source.rs b/core/connectors/runtime/src/manager/source.rs index 85cec50a34..dcdd85b937 100644 --- a/core/connectors/runtime/src/manager/source.rs +++ b/core/connectors/runtime/src/manager/source.rs @@ -240,10 +240,9 @@ impl SourceManager { match source::setup_source_producer(key, config, iggy_client).await { Ok(parts) => parts, Err(error) => { - // Closed here rather than left to `drop` so this error - // reaches the caller after teardown, not alongside it. - // `drop` stays the net for a cancellation, and for any `?` - // added inside this window later. + // 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); } @@ -520,7 +519,7 @@ mod tests { #[tokio::test] async fn should_not_double_count_when_an_error_falls_between_two_running_reports() { - // The interleaving spetz measured on #4064: the forwarding loop 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 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 5798d0912e..88bc43a659 100644 --- a/core/connectors/runtime/src/sink.rs +++ b/core/connectors/runtime/src/sink.rs @@ -19,7 +19,7 @@ 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, close_plugin_instance, resolve_plugin_path, @@ -184,7 +184,7 @@ pub async fn init( .get_mut(&path) .expect("sink connector was inserted above"); let close = connector.container.iggy_sink_close; - close_plugin_instance(&|id| close(id), "sink", plugin_id, &key); + 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 c9cedd4353..7c54108d8f 100644 --- a/core/connectors/runtime/src/source.rs +++ b/core/connectors/runtime/src/source.rs @@ -42,6 +42,7 @@ 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, @@ -312,23 +313,27 @@ pub(crate) fn init_source( } /// A plugin's `iggy_source_close` together with whatever keeps the library that -/// exports it mapped. Held instead of the bare `extern "C" fn` so the call stays -/// valid once it is deferred off the calling thread. +/// 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 being recorded on -/// `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 in that window -/// stranded the new one for the life of the process. +/// 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. /// -/// A guard rather than a cleanup branch on each fallible call, because the -/// window is defined by the two statements that open and record the instance, -/// not by which call between them happens to be fallible today. Adding a `?` -/// inside it stays correct. Both call sites use 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 @@ -342,13 +347,8 @@ pub(crate) struct SourceInstanceGuard { impl SourceInstanceGuard { /// Arms a guard over an instance the caller has just opened through - /// `container`. - /// - /// The captured `Arc` is the point. `iggy_source_close` is a pointer read - /// out of a `dlopen`ed library and stays callable only while something - /// keeps that library mapped, so the guard owns the container rather than - /// relying on it being declared before the guard and therefore dropped - /// after it. + /// `container`, which it captures rather than borrows for the reason the + /// type documents. pub(crate) fn for_container( container: Arc>, plugin_id: u32, @@ -377,12 +377,8 @@ impl SourceInstanceGuard { self.close = None; } - /// Closes the instance and waits for the plugin to finish, so an error the - /// caller returns afterwards means the instance is already gone. A restart - /// retried straight away then has nothing left to collide with. - /// - /// `drop` cannot offer that ordering, which is why the error arms call this - /// instead of relying on it. + /// 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; @@ -390,7 +386,7 @@ impl SourceInstanceGuard { 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(), "source", plugin_id, &key) + close_plugin_instance(close.as_ref(), ConnectorType::Source, plugin_id, &key) }) .await .is_err() @@ -410,20 +406,17 @@ impl Drop for SourceInstanceGuard { let plugin_id = self.plugin_id; let key = std::mem::take(&mut self.key); - // Nothing can await here, so the plugin's teardown cannot be bounded - // here either: `SourceContainer::close` drives the plugin's own - // `close()` under `block_on`, and that runs for as long as the plugin - // takes. Hand it to the blocking pool, where blocking is what the - // thread is for. The closure carries the container, so the library - // stays mapped until the call returns. + // `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(), "source", plugin_id, &key) + 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(), "source", plugin_id, &key), + Err(_) => close_plugin_instance(close.as_ref(), ConnectorType::Source, plugin_id, &key), } } }