conductor-oss / conductor-oss/conductor

TaskStatusListener not invoked for system task lifecycle transitions (both sync and async)

Open
#847 2 comments 0 reactions 1 assignee Claimed by @NicholasDCole View on GitHub
bug
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.