fix(connectors): close the source instance a failed start leaves behind - #4064
fix(connectors): close the source instance a failed start leaves behind#4064mlevkov wants to merge 12 commits into
Conversation
|
Thanks for the PR. It is labeled Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
|
/ready |
|
/request-review @hubcio |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #4064 +/- ##
=============================================
- Coverage 86.39% 57.27% -29.13%
+ Complexity 1455 1449 -6
=============================================
Files 1259 1255 -4
Lines 205513 166443 -39070
Branches 170714 131740 -38974
=============================================
- Hits 177550 95326 -82224
- Misses 23535 66780 +43245
+ Partials 4428 4337 -91
🚀 New features to boost your workflow:
|
hubcio
left a comment
hubcio
left a comment
There was a problem hiding this comment.
a few things outside the diff, none of them blocking this PR:
core/connectors/sdk/src/source.rs:597-iggy_source_openinserts intoINSTANCESeven whenopen()failed, andSourceContainer::openstores the source before returning 1. a plugin whoseopen()bound a listener and then errored keeps it for the life of the process. same leak class as this PR, one statement earlier. the fix is to skip the insert and drop the container, not to close from the runtime side - the SDK already stored the source, so that would runSource::close()on an instance that never opened.core/connectors/runtime/src/source.rs:71-SOURCE_SENDERSbeing a process global is why an orphaned forwarding loop can't die. the sink owns itswatch::SenderinSinkDetails, so its identical window self-heals. an RAII registration stored next tohandler_taskswould fix the source side properly.core/connectors/runtime/src/manager/sink.rs:200-223- same unrecorded-instance window on the sink path,details.info.idonly set at 223. narrower than the source case, since the consume tasks exit when thewatch::Senderdrops, so the same guard alone is enough there.core/connectors/runtime/src/manager/source.rs:186-190-stop_connectornever clearsdetails.info.id, so after a failed start every later stop re-closes a dead id and line 168 logs "Closed" for it. the shutdown sweep hits this too.core/connectors/runtime/src/manager/source.rs:167- the stop path drops theiggy_source_closeresult and logs "Closed" unconditionally. a -1 there means teardown was skipped while theINSTANCESentry is already gone, so nothing can retry.core/connectors/runtime/src/manager/source.rs:311- a failed restart leaves the connector Stopped withlast_errorcleared, soGET /sourcesshows nothing wrong. oneset_erroronstart_connector(..).await?covers all five fallible steps.core/connectors/runtime/src/source.rs:384-427-setup_source_producerbuilds andinit()s a producer per configured stream but keeps only the last, so two configured streams silently produce to one.
| metrics.increment_sources_running(); | ||
| } | ||
| // `details.info.id` now names this instance, so a later stop reaches it. | ||
| instance.disarm(); |
There was a problem hiding this 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.
| @@ -265,6 +271,8 @@ impl SourceManager { | |||
| details.handler_tasks = handler_tasks; | |||
| metrics.increment_sources_running(); | |||
There was a problem hiding this 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.
| impl Drop for SourceInstanceGuard<'_> { | ||
| fn drop(&mut self) { | ||
| if self.armed { | ||
| close_failed_source(self.close, self.plugin_id, self.key); |
There was a problem hiding this 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.
There was a problem hiding this comment.
one more option, if you take the Arc suggestion on line 317: move that Arc<Container<SourceApi>> into a spawn_blocking and read iggy_source_close there. Container is Send + Sync, so the task keeps the library mapped and the close stops parking a worker. tradeoff is the teardown becomes unordered against the Err return.
| // 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 = |
There was a problem hiding this 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.
| /// 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, |
There was a problem hiding this 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.
| -1 | ||
| } | ||
|
|
||
| #[test] |
There was a problem hiding this 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.
| /// 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 |
There was a problem hiding this 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.
| } | ||
| } | ||
|
|
||
| /// Records what `SourceInstanceGuard` passed to the FFI. A stub has to be a |
There was a problem hiding this 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.
| "iggy_source_close returned {close_result} while cleaning up failed source connector with ID: {plugin_id} ({key})" | ||
| ); | ||
| } | ||
| } |
There was a problem hiding this 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.
| pub(crate) struct SourceInstanceGuard<'a> { | ||
| close: extern "C" fn(u32) -> i32, | ||
| plugin_id: u32, | ||
| key: &'a str, |
There was a problem hiding this 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.
github-actions
Bot
added
S-waiting-on-author
1d20ef5 to
e88133e
Compare
e88133e to
eed068f
Compare
|
All 18 inline findings are in, at
You were right that the fix was incomplete both ways. A1 is a bit worse than Rebased onto master twice since this was posted, so the shas above are the Five places I did not do exactly what you askedA3, the A2, both of your options rather than one. A4, reported rather than deleted. Dropping the status write outright leaves a C7, the C1, not the stop path. You noted the helper would also suit What the integration tests do and do not reachThey do reach more than I first credited them with. Two things are still not pinned by any test:
Everything else was mutation-checked, with each mutant confirmed to compile Out of diffFiled the multi-stream producer bug as #4097. |
|
/ready |
|
/request-review @hubcio |
github-actions
Bot
added
S-waiting-on-review
b3a4000 to
efb9ab0
Compare
| // 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)) |
There was a problem hiding this comment.
This update_status(Running) runs after releasing the startup lock, so the forwarding loop can report Running and then call set_error for its first batch before this call executes. The delayed update overwrites Error, clears last_error, and increments sources_running again because set_error does not decrement it.
A controlled interleaving reproduced Running, no last_error, and sources_running = 2 for one instance. Performing the initial transition under the startup lock preserved Error, its details, and a count of 1.
Please apply the initial status transition under the existing details lock, sharing the transition logic with update_status, and remove this trailing unconditional update. Add a regression test with an error between the two Running reports. The current consecutive-Running test cannot catch this ordering
github-actions
Bot
added
S-waiting-on-author
Closes apache#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.
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<Container<SourceApi>>` inside the closure the guard calls, so the mapping is owned for as long as the close is reachable. The field is `Arc<dyn Fn(u32) -> i32 + Send + Sync>` rather than an `Arc<Container<SourceApi>>`. 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`.
`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.
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.
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.
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.
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<Container<SourceApi>>` 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.
`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.
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.
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.
efb9ab0 to
8e3e929
Compare
|
Fixed at You were right, and the second half of your sentence is the part I had missed.
The regression test puts the error between the two One thing I did not change and one I could not test.
The ordering itself has no test. The rebase moved every sha, so the ones in my earlier comment on this PR are |
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.
8e3e929 to
898c6a9
Compare
|
/ready |
github-actions
Bot
added
S-waiting-on-review
|
/request-review @spetz |
3 participants
Closes #4062.
The leak
SourceManager::start_connectortakes a freshplugin_id, callsinit_source, and records that id onSourceDetailsonly after the handler tasks are spawned. In between, the instance exists inside the plugin and nothing outside it knows the id:stop_connectorcloses whateverdetails.info.idholds, which is still the previous instance.setup_source_producerreturning early through?therefore stranded the new one for the life of the process.source::initalready cleaned up on the identical failure, which is the asymmetry @hubcio pointed at.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, not a cleanup branch
This deviates from the fix in the review, which was to mirror
source::init's error arm at the call site, so it is worth saying why rather than leaving it to be found.The window is defined by the two statements that open the instance and record its id, not by which call between them happens to be fallible today. A cleanup branch is correct only for the one
?that exists now, and silently wrong for the next one somebody adds.SourceInstanceGuardis armed atinit_sourceand disarmed once the id is recorded, so every path out of that window closes the instance, including a panic.It also made the behaviour testable.
Container<SourceApi>only comes fromdlopen, sostart_connectorcannot be exercised in a unit test at all, while a guard holding the bareextern "C" fncan be driven directly.The close-and-report itself is now one function shared with
source::init, so the two sites cannot drift.source::initkeeps its existing control flow; only the duplicated body moved.No
cleanup_senderon this path:spawn_source_handleris what registers the sender, and it has not run yet.Tests
Three, each mutation-checked, each mutant confirmed to compile first:
-1, the code the SDK returns for an unknown id) is reported and not propagated, because unwinding out ofdropwould be worse than the leak it is cleaning up afterEach test owns its stub and statics rather than sharing a pair, which would have made two of them race in the same process.
What is not covered, and why
The guard's placement in
start_connectorhas no test. I verified that rather than assuming it: disarming the guard immediately after construction restores the original leak, compiles, and the suite still passes.Reaching that path needs a real
Container, so it cannot be a unit test, and the only route intostart_connectorisPOST /sources/{key}/restart. Makingsetup_source_producerfail there means either a config the local provider will serve on restart but not at boot, which today works only because of the version selection in #3848 and would break when that is fixed, or stopping the broker mid-test. Both couple this regression test to something unrelated to it, so I left it out rather than write a test that fails for the wrong reason later. Happy to add either if you would rather have the coverage than the independence.source::init's cleanup remains covered only byerror_isolation.rsasserting the connector reportsError, which it did before this change too.Verification
cargo fmt,cargo sort --no-format, clippy at both feature sets, rustdoc under-D warnings, 198 unit tests iniggy-connectors, andstdout_sink+random_sourcestill build.