hubcio commented on code in PR #4064:
URL: https://github.com/apache/iggy/pull/4064#discussion_r3996826146
##########
core/connectors/runtime/src/manager/source.rs:
##########
@@ -463,6 +532,122 @@ 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
Review Comment:
nit: this comment records PR history instead of the test invariant. remove
the PR number and person's name.
##########
core/connectors/runtime/src/main.rs:
##########
@@ -456,8 +456,28 @@ 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.
+///
+/// `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,
Review Comment:
nit: `kind` accepts arbitrary strings even though `ConnectorType` already
defines the labels. use that enum and make `as_label` visible within the crate.
##########
core/connectors/runtime/src/main.rs:
##########
@@ -456,8 +456,28 @@ 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.
+///
+/// `kind` is "source" or "sink". The two sides had this body inline, one word
+/// apart.
+pub(crate) fn close_plugin_instance(
Review Comment:
nit: this helper splits the connector struct declarations. move it above the
grouped types.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -961,6 +1093,193 @@ 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_disarmed_guard_when_dropped_should_leave_the_instance_open() {
Review Comment:
simplification: the successful helper test already covers this disarmed
guard. remove this duplicate, keeping the helper success/error cases and the
separate armed-drop, runtime-drop and awaited-close tests.
##########
core/connectors/runtime/src/manager/source.rs:
##########
@@ -328,6 +348,55 @@ 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();
+ }
+ }
+
+ /// 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(
Review Comment:
simplification: this callback adds indirection around startup assignments.
inline it under the existing lock and remove its test, but keep registration,
status and disarming free of `await` because the compiler will stop enforcing
that.
##########
core/connectors/runtime/src/manager/source.rs:
##########
@@ -236,35 +231,60 @@ impl SourceManager {
state,
)?;
Review Comment:
warning: this pre-existing failed open leaves the fresh instance in the SDK
registry before the guard exists. roll back failed opens while preserving
duplicate-id rejection.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -305,6 +311,121 @@ 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<dyn Fn(u32) -> i32 + Send + Sync>;
+
+/// Closes a source instance that `iggy_source_open` created and nothing else
Review Comment:
simplification: trim the repeated cleanup prose, but keep both startup and
restart handoffs, library lifetime, worker offloading, and the difference
between awaited close and deferred `Drop`.
also at `core/connectors/runtime/src/manager/source.rs:92`.
##########
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;
Review Comment:
nit: this gauge wait reuses a constant named for state storage. use
`GAUGE_SETTLE_WINDOW`, or a neutral shared name if both waits should use the
same policy.
##########
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();
Review Comment:
warning: requests outside the polling timeout can hang when the runtime
stalls. set `.timeout(WAIT_TIMEOUT)` on the client and report the last
completed poll without another request.
##########
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 {
Review Comment:
simplification: the stats helpers repeat the same request and decoding.
share a helper returning `reqwest::Result<ConnectorRuntimeStats>` so callers
that retry errors keep doing so, while immediate readers retain their existing
checks.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -305,6 +311,121 @@ 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<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 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 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,
+ key: String,
+ armed: bool,
Review Comment:
simplification: the flag duplicates cleanup ownership and forces callback
clones. use `Option<SourceClose>`, clearing it on disarm and taking it during
teardown, while preserving awaited close and deferred `Drop`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
