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