conductor-oss / conductor-oss/conductor

[BUG] TaskStatusListener.onTaskInProgress not fired for sync system tasks (HUMAN) — engine-driven transitions to IN_PROGRESS bypass the listener

Open
#1,176 0 comments 0 reactions 0 assignees View on GitHub
bug
Dominant language
Java
Stars
32.2k
Forks
1k
Avg merge
2d 5h
Merged PRs (30d)
41

Description

### Describe the bug

When a workflow contains a sync system task (`WorkflowSystemTask.isAsync() == false`) that transitions itself to `IN_PROGRESS` from `WorkflowExecutor.scheduleTask(...)` — `HUMAN` is the canonical example — the registered `TaskStatusListener` never receives an `onTaskInProgress` event. The terminal `COMPLETED` event still fires because human completion arrives via `POST /api/tasks` → `WorkflowExecutor.updateTask(TaskResult)`, which is the only place that calls `notifyTaskStatusListener(task)` today.

Where the gap is, in `core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutorOps.java` (sync branch of `scheduleTask`):

```java
for (TaskModel task : systemTasks) {
WorkflowSystemTask workflowSystemTask = systemTaskRegistry.get(task.getTaskType());
...
if (!workflowSystemTask.isAsync()) {
try {
workflowSystemTask.start(workflow, task, this); // Human.start() -> setStatus(IN_PROGRESS)
} catch (Exception e) { ... }
startedSystemTasks = true;
executionDAOFacade.updateTask(task); // persisted, but listener never notified
} else {
tasksToBeQueued.add(task); // async path - SCHEDULED fires via addTaskToQueue
}
}
```

The same shape exists for async system tasks in `AsyncSystemTaskExecutor.executeSystemTask(...)` -`systemTask.start(...)` flips the task to `IN_PROGRESS`, the `finally` block calls `executionDAOFacade.updateTask(task)`, and the listener is again never invoked. We have not independently reproduced this on `WAIT` yet, but the code path is structurally identical so the same fix applies.

**Reproduction (verified on v3.30.2)**

Wire up `task-status-listener` with the HTTP publisher pointing at any receiver:

```properties
conductor.task-status-listener.type=task_publisher
conductor.status-notifier.notification.url=http://my-listener:8080
conductor.status-notifier.notification.endpointTask=task_status
conductor.status-notifier.notification.subscribedTaskStatuses=SCHEDULED,IN_PROGRESS,COMPLETED,FAILED,CANCELED,TIMED_OUT
```

Workflow definition with a single HUMAN task:

```json
{
"name": "human_demo",
"version": 1,
"tasks": [
{ "name": "review", "taskReferenceName": "review", "type": "HUMAN" }
]
}
```

Start the workflow, wait for the human task, then complete it via:

```bash
curl -X POST http://localhost:8080/api/tasks \
-H 'Content-Type: application/json' \
-d '{"taskId":"","status":"COMPLETED","workflowInstanceId":""}'
```

Notifications observed at the receiver:

```
POST /task_status -> { "taskId": "...", "status": "COMPLETED" }
# no IN_PROGRESS event
```

Expected: `IN_PROGRESS` is also emitted at the moment the engine flips the human task from `SCHEDULED` to `IN_PROGRESS` (i.e. when the workflow scheduler runs `Human.start(...)`).

**Suggested fix**

Notify the listener after the engine persists a sync system task transition:

```diff
--- a/core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutorOps.java
+++ b/core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutorOps.java
@@ scheduleTask(...)
if (!workflowSystemTask.isAsync()) {
try {
workflowSystemTask.start(workflow, task, this);
} catch (Exception e) { ... }
startedSystemTasks = true;
executionDAOFacade.updateTask(task);
+ try {
+ notifyTaskStatusListener(task);
+ } catch (Exception e) {
+ LOGGER.error("Error notifying TaskStatusListener for sync system task {}",
+ task.getTaskId(), e);
+ }
} else {
tasksToBeQueued.add(task);
}
```

And the symmetric call in the async path:

```diff
--- a/core/src/main/java/com/netflix/conductor/core/execution/AsyncSystemTaskExecutor.java
+++ b/core/src/main/java/com/netflix/conductor/core/execution/AsyncSystemTaskExecutor.java
@@ executeSystemTask(...) finally
} finally {
executionDAOFacade.updateTask(task);
+ try {
+ // notify listener for engine-driven system-task transitions
+ // (mirrors notifyTaskStatusListener in WorkflowExecutor.updateTask)
+ } catch (Exception e) { ... }
if (shouldRemoveTaskFromQueue) { ... }
```

(Or, cleaner: fold listener notification into a single helper that wraps every `executionDAOFacade.updateTask(task)` site so engine-driven and worker-driven transitions emit consistently.)

Happy to PR — would also like guidance on whether the preferred shape is per-call-site notification or a centralized notify-on-persist helper.

Contributor guide

Open the contributing guide

Research direction

Start in core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutorOps.java at the sync branch of scheduleTask, then compare it with AsyncSystemTaskExecutor.executeSystemTask(...) and WorkflowExecutor.updateTask(TaskResult). Reproduce with the HUMAN workflow and task-status-listener configuration described in the issue. Done means engine-driven IN_PROGRESS transitions notify TaskStatusListener consistently for sync and async system tasks without breaking terminal notifications.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
backend, distributed-systems
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Quiet
Clarity
Clearly specified
Newbie friendliness
74/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.