Introduction

When tasks take too long to execute, running them synchronously with flow.executeTask can lead to stability issues in your workflows.

To mitigate this, you can defer the execution of a workflow and resume it after the nested task completes. This approach enhances reliability by allowing long-running tasks to process asynchronously.

@Executor("example-task")
export class ExampleExecutor implements TaskExecutor {

    async processTask(exampleCommand: ExampleCommand, meta: TaskMetadata, flow: Flow): Promise<TaskResult> {
        preprocess();
        return flow
            .defer() // Defer the workflow 
            .withTask('example-nested-task', {})  // Define the nested task to execute
            .continueOn("myContinuationHandler") // Specify the continuation handler
    }
}

@Continuation("myContinuationHandler")
export class MyContinuationHandler extends ContinuationHandler {

    async onResponse(
        name: string,
        result: TaskResult, // nested task result
        flow: Flow
    ): Promise<ExampleResponse> {
        return postProcess(); // Process the result after the nested task completes
    }
}

The onResponse method is triggered in the following scenarios:

• When each nested task completes or fails.
• When each nested task publishes in-progress results.

The default implementation of onResponse checks result.status and automatically delegates to the appropriate method on your continuation handler:

• onSuccess for successful completion
• onFailure for errors
• onProgress for in-progress updates

You can override any of these methods in your handler to customize how the parent task should respond.

By default:

• onFailure throws an error containing details from the failed nested task.
• onProgress forwards the nested task’s progress as the parent task’s progress.

If you need to transform the in-progress result before publishing it, you can override onProgress like this:

@Continuation("myContinuationHandler")
export class MyContinuationHandler extends ContinuationHandler {

    async onProgress(
        name: string,
        result: TaskResult,
        flow: Flow
    ): Promise<ExampleResponse> {
        return {
            status: TaskStatus.IN_PROGRESS,
            result: {
                'nestedTask': result.result
            }
        };
    }
}

Handling multiple deferred tasks

When executing long-running tasks, you can defer multiple tasks simultaneously within your workflow. This allows them to run in parallel, enhancing efficiency and reducing overall execution time.

Here's an example of how to define multiple nested tasks to run in parallel:

@Executor("example-task")
export class ExampleExecutor implements TaskExecutor {

    async processTask(exampleCommand: ExampleCommand, meta: TaskMetadata, flow: Flow): Promise<TaskResult> {
        preprocess();
        return flow
            .defer() // Defer the workflow
            .withTask('example-nested-task', params1, { name: "first-task" })  // Define a nested task
            .withTask('example-nested-task', params2, { name: "second-task" }) // Define a nested task
            .continueOn("myContinuationHandler"); // Specify the continuation handler
    }
}

How the handler is invoked

onResponse (and by extension onSuccess / onFailure / onProgress) is called once per nested-task update. With two deferred tasks, the handler runs at least twice, and whatever it returns becomes the parent task's published result for that call. The default behaviors are designed for the single-task case and need to be overridden for parallel tasks:

• onSuccess marks the parent task as completed as soon as the first successful response is received — so the remaining tasks' results would be discarded.
• onFailure marks the parent task as failed as soon as any failure is reported.
• onProgress forwards the in-progress result of whichever child triggered the call, without merging.

Use flow.deferredTasks() to gate completion on allCompleted() and to merge sibling results:

@Continuation("myContinuationHandler")
export class MyContinuationHandler extends ContinuationHandler {

    async onSuccess(name: string, taskResult: TaskResult, flow: Flow): Promise<TaskResult> {
        const deferred = flow.deferredTasks();

        // Still waiting on sibling tasks — publish progress and exit.
        if (!deferred.allCompleted()) {
            const completed = await deferred.getAllCompletedResults();
            return {
                status: TaskStatus.IN_PROGRESS,
                result: { completed: completed.map((t) => ({ name: t.name, result: t.result })) },
            };
        }

        // All tasks are done — merge and finish.
        const all = await deferred.getAllResults();
        return {
            status: TaskStatus.COMPLETED,
            result: { merged: all.map((t) => ({ name: t.name, result: t.result })) },
        };
    }

    async onFailure(name: string, taskResult: TaskResult, flow: Flow): Promise<TaskResult> {
        // Decide whether one child failing should fail the whole parent, or whether
        // you want to wait for siblings to finish first. Default = fail immediately.
        return super.onFailure(name, taskResult, flow);
    }
}

tip

allCompleted() is synchronous — it reflects state captured when the handler was invoked. The other flow.deferredTasks() accessors load task data lazily and must be awaited.

Querying deferred tasks

flow.deferredTasks() exposes helpers for inspecting the workflow's deferred tasks from inside a continuation handler:

Method	Returns	Description
getCurrentTask()	Promise<DeferredTaskResult>	The task whose update triggered the current continuation call. Use this to access the child command ((await ...).command) since the handler signature passes only the task name.
getResult(name)	Promise<DeferredTaskResult>	The first deferred task matching name.
getResults(name)	Promise<DeferredTaskResult[]>	All deferred tasks matching name. Useful when multiple withTask calls share the same name.
getAllResults()	Promise<DeferredTaskResult[]>	Every deferred task in this workflow, regardless of status.
getAllCompletedResults()	Promise<DeferredTaskResult[]>	Only the deferred tasks that have reached COMPLETED. Pairs well with allCompleted() === false to publish partial progress.
allCompleted()	boolean	true once every deferred task has reported a final status. Synchronous — no await.

Each DeferredTaskResult exposes { name, command, result, id, status, progress }, so handlers can inspect the original child command, its progress metrics, and its final result side-by-side.

Naming deferred tasks

.withTask(command, params, options?) accepts an optional { name: '<key>' }. When set, that value becomes:

• the lookup key for getResult(name) / getResults(name), and
• the name argument passed as the first parameter to your continuation handler.

When options.name is omitted, the command name is used in its place. Naming is required to disambiguate when you call the same command more than once (as in the multi-defer example above).

Cross-worker topic access

When a worker defers a task to another worker, access is gated by Kafka ACLs. New workers under the backend.pluggable.* topic prefix get access to each other automatically; if a new worker defers to a legacy topic, add an explicit topic_acls entry — see Kafka Topic Configuration.

Legacy workers deferring to other topics are managed outside the pluggable-workers/ YAML setup and need to be handled in the legacy worker's own Terraform configuration.