fix(scheduler): handle dropped workflow failures in consumer operations

[SparshM8]

Type of Change

• Bugfix
• New feature
• Enhancement
• Refactoring
• Dependency updates
• Documentation
• CI/CD

Description

This PR fixes a reliability issue in the scheduler consumer where failures or panics in spawned workflow tasks were silently dropped.

Problem

In consumer_operations, multiple workflows are spawned using tokio::task::spawn and then awaited via future::join_all(handler).await. However, the results of join_all were ignored. This meant:

1. If a workflow returned an Err(ProcessTrackerError), it was not logged or handled at the consumer level.
2. If a workflow task panicked or was cancelled, the JoinError was silently swallowed.

Solution

The code now iterates over the results from join_all and logs any workflow-level errors or task-level panics/cancellations. This ensures that failures are visible in the logs and can be monitored, rather than tasks silently disappearing from the processing queue.

Motivation and Context

The scheduler is a critical component for background task processing in Hyperswitch. Ensuring that workflow failures are correctly observed is essential for maintaining system reliability and debugging task execution issues.

How did you test it?

Manual code audit and verification of the tokio::task::spawn and future::join_all pattern. The fix uses standard async task result handling patterns.

Checklist

• I have read the contributing guidelines
• I have followed the code of conduct
• My code follows the style guidelines of this project
• I have performed a self-review of my own code
• I have commented my code, particularly in hard-to-understand areas
• My changes generate no new warnings
• I have added tests that prove my fix is effective or that my feature works
• New and existing unit tests pass locally with my changes
• Any dependent changes have been merged and published in downstream modules

[semanticdiff-com]

Review changes with  

Changed Files

File	Status
crates/scheduler/src/consumer.rs	11% smaller
crates/scheduler/src/metrics.rs	0% smaller

[abhisheksharma2411]

Real gap — future::join_all(handler).await; discarded every result, so a spawned workflow that panicked left no trace at all. The three-arm match is the right shape: handler is a Vec<JoinHandle<_>> from tokio::task::spawn, so Err really is a JoinError and "panicked or was cancelled" is accurate.

Two observations from reading the surrounding code, one of which I think changes what the PR is worth.

The panic arm is the valuable half; the error arm mostly duplicates an existing log

start_workflow already logs the workflow error on its way out:

let res = workflow_selector
    .trigger_workflow(&state, process.clone())
    .await
    .inspect_err(|error| {
        logger::error!(?error, "Failed to trigger workflow");
    });

That is the same error value the new Ok(Err(error)) arm receives, so after this change a single failed workflow emits "Failed to trigger workflow" and then "Workflow execution failed" — two error! lines, same cause, different wording, no extra context (no pt.id, no pt.name) on the second.

The Err(JoinError) arm is the one that adds information that exists nowhere else today: a panicking workflow currently unwinds inside the spawned task and vanishes. I would keep that arm exactly as-is, and either drop the Ok(Err) arm or make it carry something the inner log does not — the process id would be the obvious thing, since the correlating identifier is what you actually want when the two lines are interleaved across concurrent tasks.

No counter moves when a workflow fails, and TASK_PROCESSED will not tell you

The crate is metric-rich — TASKS_PICKED_COUNT, BATCHES_CONSUMED, TASK_CONSUMED, TASK_PROCESSED, TASK_FINISHED, TASK_RETRIED — and my first thought was that TASK_CONSUMED - TASK_PROCESSED already gives the failure count, making a new counter redundant. It does not:

metrics::TASK_PROCESSED.add(1, &[]);
res

is unconditional. It fires after trigger_workflow regardless of whether res is Ok or Err, so TASK_PROCESSED counts attempts, not successes. The only thing that gap measures is panics, since a panic unwinds before reaching that line.

So after this PR, workflow failure is observable only in logs, and there is no aggregate signal to alert on. Given every neighbouring state transition already has a counter, a TASK_FAILED (and arguably TASK_PANICKED) alongside these logs looks like the completion of the change rather than scope creep. Logs tell you what happened once; a counter tells you it is happening more than it was.

Neither point blocks this — swallowing panics silently is worse than double-logging, so this is an improvement as it stands.

[SparshM8]

Thanks for the feedback @abhisheksharma2411! I've updated the PR to:

1. Add a new TASK_FAILED counter metric to track both workflow failures and panics, providing the aggregate signal you mentioned.
2. Kept the Err(JoinError) arm to capture panics which were previously silent.
3. Updated the Ok(Err) arm to complement the existing logs while providing the metric increment.

This should provide full observability into failed/panicking workflows both in logs and metrics.

[abhisheksharma2411]

Apologies for the slow return — you pushed on 25 Aug and I did not come back.

TASK_FAILED closes the gap that mattered. Failure is now countable rather than only greppable, and it sits alongside TASK_CONSUMED/TASK_PROCESSED/TASK_FINISHED so the crate's metric surface stays coherent. That was the substantive half of my earlier note.

Renaming the second log to "Workflow execution failed at consumer level" also fixes the confusable-duplicate problem — two error! lines for one cause is fine once the wording tells you which layer emitted each.

One thing the single counter gives up

Both arms increment the same counter:

Ok(Err(error)) => {
    metrics::TASK_FAILED.add(1, &[]);
    ...
}
Err(error) => {
    metrics::TASK_FAILED.add(1, &[]);
    ...
}

Those are operationally different in the way this PR is about. Ok(Err(_)) is a workflow that failed and said so — handled, expected, already logged downstream. Err(JoinError) is a task that panicked or was cancelled, which is a bug and was previously invisible entirely.

Collapsing them means a panic raises TASK_FAILED by one, indistinguishable from an ordinary workflow error. An alert on that counter cannot tell "the payment provider is down" from "we are panicking in production", and the panic — the thing that had no signal at all before this PR — is back to needing a log search to notice.

A sibling counter would keep them apart in the crate's existing style:

counter_metric!(TASK_PANICKED, PT_METER); // Tasks that panicked or were cancelled

I looked at whether an attribute would be tidier — add(1, &[KeyValue::new("reason", "panic")]) — but every counter in crates/scheduler uses &[], so a label here would be the first of its kind and a second counter is the smaller change.

And the smaller point from last time

The Ok(Err) arm still carries no correlating identifier. With concurrent tasks interleaving, ?error alone does not tell you which process tracker entry failed, and pt.id is in scope at the spawn site. Same for the panic arm, where it matters more — a JoinError carries nothing about what the task was doing.

Neither blocks. Swallowing panics silently was the real problem and that is fixed.