spetz commented on code in PR #4064:
URL: https://github.com/apache/iggy/pull/4064#discussion_r3965888670
##########
core/connectors/runtime/src/manager/source.rs:
##########
@@ -236,35 +236,65 @@ impl SourceManager {
state,
)?;
info!("Source connector with ID: {plugin_id} for plugin: {key}
initialized successfully.");
+ // Armed from here until the id is recorded below.
`SourceInstanceGuard`
+ // carries why that window strands the instance.
+ let instance_guard =
+ source::SourceInstanceGuard::for_container(container.clone(),
plugin_id, key);
let (producer, encoder, transforms) =
- source::setup_source_producer(key, config, iggy_client).await?;
+ match source::setup_source_producer(key, config,
iggy_client).await {
+ Ok(parts) => parts,
+ Err(error) => {
+ // Closed here rather than left to `drop` so this error
+ // reaches the caller after teardown, not alongside it.
+ // `drop` stays the net for a cancellation, and for any `?`
+ // added inside this window later.
+ instance_guard.close().await;
+ return Err(error);
+ }
+ };
let handle_callback = container.iggy_source_handle_v2;
let batch_result_callback = container.iggy_source_batch_result;
- let handler_tasks = source::spawn_source_handler(
- plugin_id,
- key,
- config.verbose,
- config.benchmark,
- producer,
- encoder,
- transforms,
- state_storage,
- handle_callback,
- batch_result_callback,
- context.clone(),
- );
+ // The lock is taken before the spawn so nothing can await between
+ // registering the tasks and recording the id that reaches them. A
+ // cancellation in that gap left the `SOURCE_SENDERS` entry and both
+ // spawned tasks behind with no id naming them, and the forwarding loop
+ // then ran for the life of the process. The guard closes the plugin
+ // instance on that path but cannot reach either of those.
+ //
+ // The forwarding loop's own first act is to take this lock, so it
waits
+ // for this block to end rather than racing it.
{
let mut details = details.lock().await;
- details.info.id = plugin_id;
- details.info.status = ConnectorStatus::Running;
- details.info.last_error = None;
- details.config = config.clone();
- details.handler_tasks = handler_tasks;
- metrics.increment_sources_running();
+ details.record_started(plugin_id, config, || {
+ source::spawn_source_handler(
+ plugin_id,
+ key,
+ config.verbose,
+ config.benchmark,
+ producer,
+ encoder,
+ transforms,
+ state_storage,
+ handle_callback,
+ batch_result_callback,
+ context.clone(),
+ )
+ });
}
+ // `details.info.id` now names this instance, so a later stop reaches
it.
+ instance_guard.disarm();
+
+ // Through `update_status`, which is the only thing that moves the
gauge
+ // and moves it only on a real transition. Writing the status here and
+ // incrementing by hand as well meant two mechanisms counting one
start:
+ // 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))
Review 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
--
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]
