mlevkov commented on code in PR #4064:
URL: https://github.com/apache/iggy/pull/4064#discussion_r3997339705
##########
core/connectors/runtime/src/manager/source.rs:
##########
@@ -236,35 +231,60 @@ impl SourceManager {
state,
)?;
Review Comment:
Confirmed and fixed in `fix(connectors): do not register a source or sink
whose open failed`.
It is a bit worse than the registry entry. `SourceContainer::open` assigns
`self.source` before it looks at the result, so the stranded instance is fully
constructed, not a half-built one. The macro then inserts it whatever the open
returned, and the runtime has its error back before it has recorded the plugin
id, so nothing outside the plugin can name it to close it.
The rollback is not inserting: the container drops on the failure path and
releases the instance the way any other failed construction does. The duplicate
id guard above is untouched, so reopening the same id still refuses rather than
replacing a live instance.
**`iggy_sink_open` had the same shape and I fixed it too.** Same
unconditional insert after the same kind of open. Say if you would rather that
rode separately.
No test, and I would rather say so than imply otherwise: 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 see it. The change is three lines,
which is the argument for reading it rather than around it.
##########
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:
Fixed in `test(connectors): stop the runtime stats waits from hanging on a
stalled runtime`. Both halves, and the second one is nastier than "can hang".
The unbudgeted request is the one in the assertion message. It is only
evaluated once the wait has already timed out, which is exactly the moment the
runtime is stalled, and with no timeout on the client it never returns. So the
test that should fail with "never reached 1" hangs instead and reports nothing
at all. The loop now carries the value it saw and the message prints that.
The client is built with `WAIT_TIMEOUT` on it, so no request in the file can
outlive the wait it belongs to.
##########
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:
Inlined in `refactor(connectors): give the instance guard one representation
of ownership`. Registration, the status transition and the disarm are all still
inside the one lock hold, in that order.
One thing worth putting in front of you rather than just doing. The closure
was not incidental indirection: it arrived as `0c4b859f8 fix(connectors): make
an await in the start window a compile error`, and taking the spawn as a
closure into a non-async fn is what made the compiler refuse an await between
the spawn and the id record. Your note says to keep that property knowing the
compiler will stop enforcing it, so I have written the requirement where the
three statements now sit. Flagging it in case the commit changes your mind;
happy either way.
Its test went with it.
##########
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:
Done in `refactor(connectors): give the instance guard one representation of
ownership`.
`Option<SourceClose>` throughout: `disarm` clears it, the awaited close and
`Drop` each `take()` it, and neither clones the callback any more. The flag is
gone, so there is no second thing that has to agree with the first.
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.
Checked it rather than assumed: making `disarm` a no-op turns
`given_fallible_step_when_it_succeeds_should_leave_the_instance_open` red, so
the ownership release is still the thing under test.
##########
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:
Consolidated in `refactor(connectors): type the cleanup label and say the
argument once`.
The argument now lives on `SourceInstanceGuard` once and keeps everything
you listed: the window and why it is a guard rather than a cleanup branch per
fallible call, that startup and restart both hand off through it, that the
library has to 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. `SourceClose`, `for_container`,
`close`, the `Drop` body and the call sites point at it instead of restating it.
**One question.** I could not place the second location.
`manager/source.rs:92` in the current head is inside `set_error`, which is the
gauge transition comment rather than cleanup prose, and the branch has taken
master three times since you looked, so I suspect the anchor drifted. I trimmed
the two call-site comments in that file that do repeat the cleanup argument. If
you meant somewhere else, point me at it and I will do that one too.
##########
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:
Removed in `refactor(connectors): give the instance guard one representation
of ownership`. You are right that it is covered:
`given_fallible_step_when_it_succeeds_should_leave_the_instance_open` runs a
guard through `disarm` and asserts nothing was closed, and I verified it by
making `disarm` a no-op, which turns that test red.
The armed-drop, runtime-drop, awaited-close and both helper cases are all
still there.
One thing to hand back, since it is the reason the deleted test was written
that way. The surviving success case asserts an **empty** recorder with nothing
in it proving the recorder can move, which is the vacuity the removed test
guarded against by dropping an armed guard first. It is saved only by its
error-case sibling using the same type in a different test. That holds today;
it stops holding if that sibling ever goes.
--
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]
