Skip to content

Recovery should not be interrupted by a DLQ workflow #461

Description

@maxdml

When recovering workflows, the entire recovery run is aborted if one of the recovered workflows gets a DLQ error.

The DLQ transition itself throws DBOSMaxRecoveryAttemptsExceededException after committing MAX_RECOVERY_ATTEMPTS_EXCEEDED:

if (maxRetries != null && attempts > maxRetries + 1) {
var sql =
"""
UPDATE "%s".workflow_status
SET status = ?, deduplication_id = NULL, started_at_epoch_ms = NULL, queue_name = NULL
WHERE workflow_uuid = ? AND status = ?
"""
.formatted(ctx.schema());
try (PreparedStatement stmt = conn.prepareStatement(sql)) {
stmt.setString(1, WorkflowState.MAX_RECOVERY_ATTEMPTS_EXCEEDED.name());
stmt.setString(2, initStatus.workflowId());
stmt.setString(3, WorkflowState.PENDING.name());
stmt.executeUpdate();
}
throw new DBOSMaxRecoveryAttemptsExceededException(initStatus.workflowId(), maxRetries);

Neither recovery path catches it per-workflow:

  • Startup recovery wraps the entire loop in a single catch, so one throw abandons all remaining pending workflows:

    Runnable recoveryTask =
    () -> {
    try {
    var workflows = systemDatabase.listWorkflows(recoveryQuery);
    for (var wf : workflows) {
    recoverWorkflow(wf.workflowId(), wf.queueName());
    }
    } catch (Throwable t) {
    logger.error("Recovery task failed", t);
    }
    };
    executorService.submit(recoveryTask);

  • The public recoverPendingWorkflows (used by Conductor) maps over the stream with no per-element catch, so the exception aborts the whole run and surfaces to the caller:

    public List<WorkflowHandle<?, ?>> recoverPendingWorkflows(List<String> executorIds) {
    Objects.requireNonNull(executorIds);
    var input =
    new ListWorkflowsInput()
    .withStatus(WorkflowState.PENDING)
    .withExecutorIds(executorIds)
    .withApplicationVersion(appVersion);
    var workflows = systemDatabase.listWorkflows(input);
    return workflows.stream()
    .map(wf -> recoverWorkflow(wf.workflowId(), wf.queueName()))
    .collect(Collectors.toList());

For reference, Python (_recovery.py recover_pending_workflows) and TypeScript (dbos-executor.ts recoverPendingWorkflows) catch per-workflow, log, and continue — a dead-lettered workflow never aborts recovery of the rest.

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions