Skip to content

fix(topology): Fix for issue causing stalling on shutdown for sinks configured w/ disk buffers #24949

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
graphcareful wants to merge 2 commits into
base: master
Choose a base branch
from
+15 −0
Open

fix(topology): Fix for issue causing stalling on shutdown for sinks configured w/ disk buffers #24949

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2ec28fa
fix(topology): Send cancel trigger to sinks during config reload
graphcareful Mar 17, 2026
eb5630a
fix(topology): Send cancel trigger within stop()
graphcareful Mar 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions src/topology/running.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,15 @@ impl RunningTopology {
pub fn stop(self) -> impl Future<Output = ()> {
// Update the API's health endpoint to signal shutdown
self.running.store(false, Ordering::Relaxed);

// Signal all sinks to flush their partial batches and stop accepting
// input. Without this, sinks rely on the natural stream EOF which only
Comment on lines +152 to +153
Copy link
Member

@bruceg bruceg Mar 17, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The bit about "stop accepting input" makes me more than a little nervous. Could this cause problems if upstream sources send more events?

Copy link
Contributor Author

@graphcareful graphcareful Mar 17, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In my manual testing before this change vector would stall on shutdown. With this change it takes a few seconds but I believe that is the time it takes to complete a round trip request of the payload to S3 (was a little less than 100MB).

I believe data will still be in the channel, orphaned. This data will be lost on shutdown, however this already occurs if the 60s timeout expires. Thoughts?

Copy link
Contributor Author

@graphcareful graphcareful Mar 17, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In theory the best option would be to stop accepting input, drain all channels, then await until disk flushes have completed, then proceed with rest of shutdown

Copy link
Contributor Author

@graphcareful graphcareful Mar 17, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm considering dropping this commit as its not really related to the original reported issue anyway

Copy link
Member

@pront pront Mar 18, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm considering dropping this commit

I agree with this. We should keep this fix as short as possible.

// arrives after upstream sources fully shut down, leaving sinks idle
// for up to `batch.timeout_secs` while holding a partial batch.
for (_, trigger) in self.detach_triggers {
trigger.into_inner().cancel();
Comment on lines +156 to +157
Copy link

@chatgpt-codex-connector chatgpt-codex-connector bot Mar 18, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Preserve graceful draining during global shutdown

Canceling every sink detach_trigger at the top of RunningTopology::stop forces all sink input streams (take_until_if(tripwire)) to end before shutdown_coordinator.shutdown_all(...) starts upstream shutdown, so sinks can terminate while events are still buffered or in flight. That changes graceful shutdown semantics into a lossy path for any pipeline with pending events, because those events are never drained by the sink.

Useful? React with 👍 / 👎.

}

// Create handy handles collections of all tasks for the subsequent
// operations.
let mut wait_handles = Vec::new();
Expand Down Expand Up @@ -651,6 +660,12 @@ impl RunningTopology {
// info about which sinks are having their buffers reused and treat them differently
// at other stages.
buffer_tx.insert((*key).clone(), self.inputs.get(key).unwrap().clone());
} else if wait_for_sinks.contains(key) {
// Cancel the trigger so the old sink shuts down immediately
// instead of draining its entire buffer, which blocks the reload.
if let Some(trigger) = self.detach_triggers.remove(key) {
trigger.into_inner().cancel();
Comment on lines +666 to +667
Copy link

@chatgpt-codex-connector chatgpt-codex-connector bot Mar 18, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Let changed sinks drain instead of canceling their input

In RunningTopology::shutdown_diff, this new branch cancels the old sink trigger for wait_for_sinks before waiting for task completion, which makes the sink stop via take_until_if(tripwire) rather than EOF from drained inputs. For changed sinks whose buffers are not being reused, queued events in the old buffer are abandoned when the old component is torn down, so reloads under resource conflict can now drop data instead of draining naturally.

Useful? React with 👍 / 👎.

}
}
self.remove_inputs(key, diff, new_config).await;
}
Expand Down
Loading