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 1 commit into
Conversation
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.
|
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 85.51% 21.89% -63.63%
Complexity 1402 1402
=============================================
Files 1240 1239 -1
Lines 186294 154271 -32023
Branches 152598 120575 -32023
=============================================
- Hits 159316 33774 -125542
- Misses 22923 119717 +96794
+ Partials 4055 780 -3275
🚀 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
2 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.