conductor-oss / conductor-oss/conductor
TaskStatusListener not invoked for system task lifecycle transitions (both sync and async)
- Dominant language
- Java
- Stars
- 32.2k
- Forks
- 1k
- Avg merge
- 2d 5h
- Merged PRs (30d)
- 41
Description
## Description
`TaskStatusListener` notifications are only fully wired for worker tasks (SIMPLE/custom). Both synchronous and asynchronous system tasks bypass most `TaskStatusListener` callbacks because their execution paths persist task state directly via `executionDAOFacade.updateTask()` / `executionDAOFacade.updateTasks()` instead of going through `WorkflowExecutorOps.updateTask(TaskResult)` where `notifyTaskStatusListener()` is called.
Any custom `TaskStatusListener` implementation will receive an incomplete view of workflow execution — missing lifecycle events for system tasks like HTTP, KAFKA_PUBLISH, WAIT, JOIN, SWITCH, FORK, INLINE, SET_VARIABLE, and others.
## Root Cause
`TaskStatusListener` callbacks are wired into four specific code paths:
| Site | Location | Callback | Triggered by |
|------|----------|----------|-------------|
| 1 | `WorkflowExecutorOps.addTaskToQueue()` (line 1695) | `onTaskScheduled()` | Task added to queue |
| 2 | `ExecutionService.poll()` (line 196) | `onTaskInProgress()` | Worker polls a task |
| 3 | `WorkflowExecutorOps.updateTask(TaskResult)` (line 911) | `notifyTaskStatusListener()` | External `TaskResult` submission via REST/gRPC/event |
| 4 | `WorkflowExecutorOps.cancelNonTerminalTasks()` (line 1296) | `notifyTaskStatusListener()` → `onTaskCanceled()` | Workflow termination |
However, system tasks are executed through internal paths that bypass all of these:
- **Sync system tasks** are started inline in `WorkflowExecutorOps.scheduleTask()` (line 1643) and executed in `WorkflowExecutorOps.decide()` (line 1163). Both persist via `executionDAOFacade.updateTask(task)` directly. They are never added to a queue, never polled, and never updated via `TaskResult`.
- **Async system tasks** are started and executed in `AsyncSystemTaskExecutor.execute()` (lines 152-154). This method persists via `executionDAOFacade.updateTask(task)` in its `finally` block (line 190). It does not go through `WorkflowExecutorOps.updateTask(TaskResult)`.
## Notification Coverage by Task Type
| Lifecycle Event | Worker Task | Async System Task | Sync System Task |
|-----------------|-------------|-------------------|------------------|
| SCHEDULED | `onTaskScheduled` via `addTaskToQueue` | `onTaskScheduled` via `addTaskToQueue` | **Not notified** |
| IN_PROGRESS | `onTaskInProgress` via `ExecutionService.poll` | **Not notified** | **Not notified** |
| COMPLETED | `onTaskCompleted` via `updateTask(TaskResult)` | **Not notified** | **Not notified** |
| FAILED | `onTaskFailed` via `updateTask(TaskResult)` | **Not notified** | **Not notified** |
| TIMED_OUT | `onTaskTimedOut` via `updateTask(TaskResult)` | **Not notified** | **Not notified** |
| CANCELED | `onTaskCanceled` via `cancelNonTerminalTasks` | `onTaskCanceled` via `cancelNonTerminalTasks` | `onTaskCanceled` via `cancelNonTerminalTasks` (rare — sync tasks are usually already terminal) |
## Affected System Tasks
**Sync** (`isAsync()` returns `false`, the default):
`SWITCH`, `FORK`, `SET_VARIABLE`, `INLINE`, `LAMBDA`, `TERMINATE`, `DO_WHILE`, `DECISION`, `EXCLUSIVE_JOIN`, `NOOP`, `HUMAN`, `EVENT`, `SUB_WORKFLOW`
**Async** (`isAsync()` returns `true`):
`HTTP`, `KAFKA_PUBLISH`, `WAIT`, `JOIN`, `START_WORKFLOW`, annotated worker tasks via `AnnotatedWorkflowSystemTask`
## Affected Code Locations
1. **`WorkflowExecutorOps.scheduleTask()`** — after `workflowSystemTask.start()` for sync tasks (line 1643-1654)
2. **`WorkflowExecutorOps.decide()`** — after `workflowSystemTask.execute()` for sync tasks (line 1156-1166)
3. **`AsyncSystemTaskExecutor.execute()`** — after `systemTask.start()` and `systemTask.execute()` for async tasks (line 149-190)
## Impact
- Any `TaskStatusListener` implementation relying on callbacks for all task types will have an incomplete picture of workflow execution.
- Observability, auditing, and custom logic tied to task lifecycle events will silently miss all system task transitions except SCHEDULED (async only) and CANCELED (during workflow termination).
## Expected Behavior
`TaskStatusListener` should be notified for all task status transitions regardless of whether the task is a worker task, async system task, or sync system task.
## Suggested Fix
Add `notifyTaskStatusListener(task)` calls in the three affected locations.
### 1. `WorkflowExecutorOps.scheduleTask()` — after sync system task `start()`
```java
if (!workflowSystemTask.isAsync()) {
try {
workflowSystemTask.start(workflow, task, this);
} catch (Exception e) {
// ... existing error handling ...
}
startedSystemTasks = true;
executionDAOFacade.updateTask(task);
try {
notifyTaskStatusListener(task);
} catch (Exception e) {
LOGGER.error("Error notifying TaskStatusListener for task: {} in workflow: {}",
task.getTaskId(), workflow.getWorkflowId(), e);
}
} else {
tasksToBeQueued.add(task);
}
```
### 2. `WorkflowExecutorOps.decide()` — after sync system task `execute()`
```java
if (!workflowSystemTask.isAsync()
&& workflowSystemTask.execute(workflow, task, this)) {
tasksToBeUpdated.add(task);
stateChanged = true;
try {
notifyTaskStatusListener(task);
} catch (Exception e) {
LOGGER.error("Error notifying TaskStatusListener for task: {} in workflow: {}",
task.getTaskId(), task.getWorkflowInstanceId(), e);
}
}
```
### 3. `AsyncSystemTaskExecutor.execute()` — after async system task `start()` / `execute()`
This requires access to `notifyTaskStatusListener` which is private to `WorkflowExecutorOps`. Options:
- Extract `notifyTaskStatusListener` to a shared component (e.g., a `TaskStatusListenerNotifier` service)
- Expose it through the `WorkflowExecutor` interface
- Inject `TaskStatusListener` directly into `AsyncSystemTaskExecutor` and replicate the dispatch logic
## Steps to Reproduce
1. Implement a custom `TaskStatusListener` that logs all callbacks.
2. Create a workflow containing both worker tasks and system tasks (e.g., `SWITCH`, `HTTP`, `INLINE`).
3. Execute the workflow.
4. Observe that `TaskStatusListener` callbacks fire for worker task transitions but are missing for system task transitions.
Contributor guide
Assessment
This issue has not been assessed yet.