fix(connectors): restart loads the newest connector config, not the active one

[mlevkov]

Summary

On the default local configuration provider, a connector runs its active configuration after a process start and its highest-numbered configuration after POST /{sinks,sources}/{key}/restart. Rolling back with PUT .../configs/active therefore appears to work until the next restart, which silently reinstates the newest version.

The HTTP provider does not have this split.

Where the two paths diverge

Startup resolves the active configuration:

// core/connectors/runtime/src/main.rs:153
let connectors_config = connectors_config_provider.get_active_configs().await?;

Restart asks for None instead:

// core/connectors/runtime/src/manager/sink.rs:261   (source.rs:280 is identical)
.get_sink_config(key, None)

and the local provider resolves None to the highest version number rather than the active one:

// core/connectors/runtime/src/configs/connectors/local_provider.rs
async fn get_sink_config(&self, key: &str, version: Option<u64>) -> ... {
    if let Some(version) = version {
        ...
    } else {
        Ok(self
            .get_sink_configs(key)
            .await?
            .into_iter()
            .max_by_key(|config| config.version))
    }
}

get_source_config is the same. Meanwhile the HTTP provider resolves the identical None to the active config:

// core/connectors/runtime/src/configs/connectors/http_provider.rs:291
None => self.url_builder.build(TemplateKeys::GET_ACTIVE_SINK_CONFIG, &vars),

So version: None means "active" in one provider and "newest" in the other, and the restart path is the caller that notices.

The local provider already tracks this

It is not that the local provider lacks the concept. set_active_sink_version persists it, and get_active_configs reads it back and selects on it:

let active_config = if let Some(&version) = active_versions.sinks.get(key) {
    config_files.iter().find(|c| c.config.version == version)

Only the get_{sink,source}_config(key, None) accessor skips it, which is exactly the one restart_connector uses.

Why it matters

The intended rollback flow does not survive a restart:

1. POST /sinks/{key}/configs publishes v2, which turns out to be bad.
2. PUT /sinks/{key}/configs/active sets v1 active. GET .../configs/active confirms v1, and a process restart runs v1.
3. POST /sinks/{key}/restart runs v2 again.

An operator rolling back an incident sees the rollback take effect, then get undone by the very call they would reach for to apply it.

Suggested fix

Make None mean "active" in the local provider, matching the HTTP provider and get_active_configs. If "newest" is wanted somewhere, it deserves an explicit accessor rather than an overloaded None.

Happy to send a PR if the direction is agreed. Found while documenting the control API surface for #3804.

[mlevkov]

One more data point that narrows the fix. None has exactly two callers, and both are the restart path:

core/connectors/runtime/src/manager/sink.rs:261     .get_sink_config(key, None)
core/connectors/runtime/src/manager/source.rs:280   .get_source_config(key, None)

Every other caller passes Some(version) (api/{sink,source}.rs serving /configs/{version}). So changing what None resolves to in the local provider cannot affect anything but restart, which is the behaviour the issue is about.

That leaves one decision I would rather not guess at: what None should return when a key has configs but no active version recorded. get_active_configs already has a rule for that case; making the accessor match it would keep one definition of "active" rather than two.

Happy to send the PR once you say which way you want it.