Repository navigation
Recovery should not be interrupted by a DLQ workflow #461
Description
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:
dbos-transact-java/transact/src/main/java/dev/dbos/transact/database/dao/WorkflowDAO.java
Lines 144 to 162 in 221ceb1
| 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:
dbos-transact-java/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java
Lines 267 to 279 in 221ceb1
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:
dbos-transact-java/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java
Lines 1117 to 1128 in 221ceb1
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.
Activity
Don't let a dead-lettered workflow abort recovery
Recover by re-enqueueing, never by executing in-process
Recover by re-enqueueing, never by executing in-process (dbos-inc#526)