Skip to content

Fail partition processor on scheduler errors #4952

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
muhamadazmy wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Fail partition processor on scheduler errors #4952

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
13 changes: 6 additions & 7 deletions crates/worker/src/partition/leadership/leader_state.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,13 +209,8 @@ impl LeaderState {
// if we have problems with latency
let scheduler_stream =
std::pin::pin!(stream::unfold(&mut self.scheduler, |scheduler| async {
match scheduler.schedule_next(vqueue_metas).await {
Ok(decisions) => Some((ActionEffect::Scheduler(decisions), scheduler)),
Err(e) => {
error!("Fatal error when polling scheduler: {e}");
None
}
}
let result = scheduler.schedule_next(vqueue_metas).await;
Some((ActionEffect::Scheduler(result), scheduler))

@tillrohrmann tillrohrmann Jun 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Instead of forwarding the error and logging it at a different place, can we let run fail if we encounter a scheduler error? Then the pp should fail as well.

@muhamadazmy muhamadazmy Jun 17, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

The PP still fails on applying the action effects! I don't only log the error.

I was aiming first on failing run() directly, but it wasn't possible cleanly without too much changes and some allocations as well.

Instead, I emitted the scheduler decision result, then fail while applying the action. Which still fails the PP

@muhamadazmy muhamadazmy Jun 17, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

One of the reasons i didn't just poll the scheduler in the select! block is cancellation safety, since all_streams is created on the stack on each call to run() racing with the scheduler will cause loss of already polled effects.

All other branches in the select block return with error this is why it's okay now.

@tillrohrmann tillrohrmann Jun 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Sorry, I missed the part wrt to propagating the error. This should be fine.

What I don't fully understand is the cancellation safety that we get with the stream_select! vs polling the scheduler directly in the select! statement. Could you help me understand?

@muhamadazmy muhamadazmy Jun 17, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I was mistaken, the ready_chunk returns immediately if it collected some items and remaining streams returns pending. so there is no fear of loss

}));

let schema_stream = (&mut self.schema_stream).filter_map(|_| {
Expand Down Expand Up @@ -363,6 +358,10 @@ impl LeaderState {
for effect in action_effects {
match effect {
ActionEffect::Scheduler(decisions) => {
let decisions = decisions.inspect_err(|err| {
error!("Fatal error when polling scheduler: {err}");
})?;

let Decisions {
qids,
num_run,
Expand Down
2 changes: 1 addition & 1 deletion crates/worker/src/partition/leadership/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -147,7 +147,7 @@ pub(crate) enum TaskTermination {

#[derive(Debug)]
pub(crate) enum ActionEffect {
Scheduler(scheduler::Decisions),
Scheduler(Result<scheduler::Decisions, StorageError>),
Invoker(InvokerEffect),
Shuffle(shuffle::OutboxTruncation),
Timer(TimerKeyValue),
Expand Down
Loading