hubcio commented on code in PR #4064:
URL: https://github.com/apache/iggy/pull/4064#discussion_r3942182612
##########
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 =
Review Comment:
nit: `instance` holds a guard, so `instance.disarm()` below reads as
disarming the instance. `instance_guard` matches `shutdown_guard` and
`tmp_guard` elsewhere in the repo.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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);
Review Comment:
warning: this can fire after `spawn_source_handler` ran, so
`iggy_source_close` hits `block_on(handle)` and `block_on(source.close())` -
unbounded plugin teardown on a tokio worker inside drop glue, where no timeout
fits. either document that contract on the type or move cleanup to an explicit
`finish()` on the error arms.
##########
core/connectors/runtime/src/manager/source.rs:
##########
@@ -265,6 +271,8 @@ impl SourceManager {
details.handler_tasks = handler_tasks;
metrics.increment_sources_running();
Review Comment:
warning: the forwarding loop's first `update_status(Running)` already bumped
this gauge, and stop only decrements once, so `sources_running` ratchets up per
restart. drop the direct status write and this increment, let `update_status`
own it.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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> {
Review Comment:
nit: add `#[must_use = "dropping an armed guard closes the source
instance"]`. a bare `SourceInstanceGuard::new(..);` statement closes the
instance on the spot, and with `warnings = "deny"` the attribute turns that
into a build error.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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,
Review Comment:
nit: the fn pointer has no lifetime tie to the `Container` that owns the
`.so` - it works only because `container` is declared before the guard and so
drops after it. hold an `Arc<Container<SourceApi>>` and read
`iggy_source_close` inside `drop`, rather than leaning on declaration order.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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) {
Review Comment:
nit: this closes a plugin instance, not a source connector -
`close_source_instance` reads right, and would also suit the stop path at
`manager/source.rs:167`. it can be a plain `fn`, both callers are in this file.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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() {
Review Comment:
nit: the articles are new here - every other test in the module reads
`given_serialized_batch_...`, `given_missing_sender_...`.
`given_armed_guard_when_dropped_should_close_instance` and siblings.
also at lines 1066, 1082.
##########
core/connectors/runtime/src/manager/source.rs:
##########
@@ -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();
Review Comment:
warning: cancel at the `details.lock().await` above (client disconnect drops
the axum handler future) and the guard closes the instance but leaves the
`SOURCE_SENDERS` entry and both spawned tasks behind, so the forwarding loop
runs forever. take the lock before `spawn_source_handler` and set `info.id`
there.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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() {
Review Comment:
nit: the harness gives no-panic for free. what line 1095 actually proves is
that close ran once and the -1 was swallowed -
`given_refused_close_when_guard_drops_should_close_once_and_swallow_refusal`.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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.
Review Comment:
nit: "durably recorded" promises a disk guarantee that doesn't exist -
`SourceDetails` is memory only and `plugin_id` never reaches the state store.
the struct doc above already says "recorded on `SourceDetails`", reuse that.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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() {
Review Comment:
nit: this passes even if the guard is never built - `DISARMED_CALLS` starts
at 0 and nothing else touches it. close an armed guard through the same stub
first, then disarm one and assert the count didn't move.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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]
Review Comment:
nit: all three tests drop or disarm inline, which the compiler already
guarantees. none covers the shape the guard exists for - a `?` returning early
with the guard still armed.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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
Review Comment:
simplification: this paragraph is repeated almost word for word at
`manager/source.rs:239-242`. keep it here and cut the call-site copy to the one
fact it adds.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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})"
+ );
+ }
+}
Review Comment:
simplification: `sink.rs:185-190` still has this body inline, one word
apart. both close pointers are `extern "C" fn(u32) -> i32`, so one helper
taking a "source"/"sink" label covers both.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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
Review Comment:
nit: `init` at line 251 is exactly the cleanup branch this paragraph argues
against, and it's still there. use the guard there too, or cut the paragraph.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -961,6 +1013,88 @@ mod tests {
}
}
+ /// Records what `SourceInstanceGuard` passed to the FFI. A stub has to be
a
Review Comment:
simplification: `next_plugin_id()` already hands each test a unique id, so a
shared recorder can't race - the caveat guards against a design nobody's using.
one id-keyed map plus `ok_close` and `refusing_close` replaces three stubs and
four statics.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -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,
Review Comment:
simplification: `key` exists to label one `warn!`, and it drags in the
lifetime param and the `impl<'a>`. both call sites already log the key on the
same failure, and `plugin_id` identifies the instance.
--
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]
