connectors: source restart leaks the plugin instance when producer setup fails

[mlevkov]

What happens

SourceManager::start_connector, which restart_connector calls after stopping the old instance, calls source::init_source and then source::setup_source_producer:

// core/connectors/runtime/src/manager/source.rs
source::init_source(&container, &config.plugin_config.clone().unwrap_or_default(), plugin_id, state)?;   // :232
info!("Source connector with ID: {plugin_id} for plugin: {key} initialized successfully.");

let (producer, encoder, transforms) =
    source::setup_source_producer(key, config, iggy_client).await?;                                       // :241

When setup_source_producer fails, the ? returns with init_source already succeeded, so the plugin holds an initialized instance under plugin_id that nothing will ever close. details.info.id = plugin_id is not reached until :261, so the stored id still names the previous instance and a later stop closes that one instead. The orphan lives for the rest of the process.

The boot path handles the identical failure correctly, and is the model for the fix:

// core/connectors/runtime/src/source.rs, the setup_source_producer error arm
Err(error) => {                                                                                          // :245
    let message = format!("Failed to set up source producer: {error}");
    error!("Source: {name} ({key}) - {message}");
    let connector = source_connectors.get_mut(&path).expect("source connector was inserted above");
    let close_result = (connector.container.iggy_source_close)(plugin_id);                                // :251
    ...
}

Why it is worth fixing rather than tolerating

For a plugin whose init_source only allocates, an orphan is wasted memory. For one that acquires a process-global resource in open(), it is a live fault. The HTTP source connector in #3798 is the worked example: instances share one listener keyed by address, so the orphan stays joined and keeps answering 200 into a bridge that no poll() drains, holds /health at 503 because a joined instance is not polling, and makes every subsequent restart attempt fail on the duplicate instance_name it never released. A transient broker blip at restart time turns into an instance that is unreachable until the whole runtime is bounced.

Fix

Mirror the boot path: close the plugin instance on the setup_source_producer failure branch before returning. Setting details.info.id before the fallible call, or carrying the id out of the error arm, is what keeps a later stop from closing the wrong instance.

One thing that does not work, in case it looks tempting: evicting on a poll_is_live-style signal from the plugin side. That flag is false for the whole producer-setup window, so a plugin cannot distinguish "setup still in progress" from "setup failed and I have been orphaned".

Provenance

Raised by @hubcio in review on #3798, twice, as out of scope for that PR, with the fix above. Filing it so it is tracked once #3798 merges and its review thread stops being where this lives.

Contribution

• I'm willing to submit a pull request to fix this bug