Tasks
The handler
interface TaskHandler<I : Any, O : Any> {
val type: TaskType<I, O>
suspend fun handle(input: I, ctx: TaskContext): O
}
ctx.attempt is 1-based. ctx.runId is the run's id. ctx.heartbeat(details) keeps the
attempt alive and is where cancellation is delivered.
Retries
Retrying belongs to the orchestrator. A handler that throws is simply retried per the run's
RetryPolicy, with the interval doubling between attempts. Watch it in the UI: every attempt is
listed, with the error and the interval before the next one.
override suspend fun handle(input: GreetingInput, ctx: TaskContext): Greeting {
if (ctx.attempt < input.failAttemptsBelow) error("simulated failure on attempt ${ctx.attempt}")
return Greeting("${input.greeting}, ${input.name}!")
}
Stopping the retries
throw NonRetryableTaskError("BVN format is invalid; retrying cannot help")
The run ends Failed on that attempt. RetryPolicy.nonRetryableErrors does the same by
exception class name, for exceptions you do not own.
Long tasks
Anything that runs for more than a few seconds should heartbeat and set heartbeatTimeout in
its options:
override suspend fun handle(input: SyncInput, ctx: TaskContext): SyncOutput {
for (page in pages(input)) {
process(page)
ctx.heartbeat("page=${page.number}")
}
return SyncOutput(done = true)
}
With heartbeatTimeout = 30.seconds, a worker that dies mid-sync is noticed within 30 seconds
and the attempt is retried elsewhere, instead of waiting out the full attemptTimeout.
Standalone or inside a flow
The same handler serves both. From application code it is started with RunOptions, which
include the id and the queue. From inside a flow it is started with TaskOptions, and the
orchestrator assigns the id and the flow's queue.