Retries, timeouts, heartbeats

ProStep options needs a JoobQ Pro key. Without one every Core feature keeps working: the Pro routes of the SDK gateway answer 402 pro_required and the console's Pro routes are not served. Get a licence · Core and Pro · Pro features index

The engine enforces these options; they are part of the blueprint.

Option	On	Meaning
retry (attempts, initial delay, backoff, maximum)	step	The engine retries a failed attempt by this policy
retry.non_retryable: [type, ...]	step	Error types that end the step at once (below)
timeout_ms	step	How long one attempt may run (default 24 h); a timed-out attempt is retried by the policy
heartbeat_ms (at least 1000)	step	How long the handler may go without calling ctx.heartbeat()
schedule_to_start_ms (at least 1000)	step	How long the step may wait for a worker to take it
blueprint timeout_ms	workflow	The whole run's deadline: the run ends timed_out. None by default

Retries

Crystal

workflow "conf-retry-#{LANG}" do
  step "flaky", Flaky, retry: retry_policy(attempts: 3, initial: 100.milliseconds)
end,

Python

wf("retry").step("flaky", flaky, retry=retry_policy(attempts=3, initial_ms=100)),

TypeScript

wf('retry').step('flaky', async (_i: any, ctx: Ctx) => {
  const calls = ctx.attempt // the engine retries the step: its attempt number counts the calls
  if (calls < 3) throw new Error(`flaky call ${calls}`)
  return { calls }
}, { retry: retryPolicy({ attempts: 3, initialMs: 100 }) }),

Ruby

wf("retry").step("flaky", lambda { |_i, ctx|
  calls = ctx.attempt # the engine retries the step: its attempt number counts the calls
  raise "flaky call #{calls}" if calls < 3

  { calls: calls }
}, retry: retry_policy(attempts: 3, initial_ms: 100)),

Go

wf("retry").Step("flaky", func(in map[string]any, ec *EC) (any, error) {
	calls := ec.Task.Attempt // the engine retries the step: its attempt number counts the calls
	if calls < 3 {
		return nil, joobq.Errorf("flaky call %d", calls)
	}
	return map[string]any{"calls": calls}, nil
}, joobq.WithRetry[*App](joobq.RetryPolicy(3, 100, 2, 1000))),

Rust

wf("retry").step_with("flaky", |_i: Value, ctx: &Ctx| {
    let calls = ctx.attempt(); // the engine retries the step: its attempt number counts the calls
    if calls < 3 {
        return Err(Error::new(format!("flaky call {}", calls)));
    }
    Ok(json!({"calls": calls}))
}, StepOptions { retry: Some(retry_policy(3, 100, 2.0, 1000)), ..Default::default() }),

Java

wf("retry").step("flaky", (i, ctx) -> {
    long calls = ctx.attempt(); // the engine retries the step: its attempt number counts the calls
    if (calls < 3) throw new IllegalStateException("flaky call " + calls);
    return map("calls", calls);
}, StepOptions.<App>none().retry(DSL.retryPolicy(3, 100, 2.0, 1000))),

PHP

wf('retry')->step('flaky', function ($i, $ctx) {
    $calls = (int)$ctx->task['attempt']; // the engine retries the step: its attempt number counts the calls
    if ($calls < 3) {
        throw new RuntimeException("flaky call {$calls}");
    }
    return ['calls' => $calls];
}, retry: DSL::retryPolicy(attempts: 3, initialMs: 100)),

.NET

Wf("retry").Step("flaky", (i, ctx) =>
{
    var calls = ctx.Attempt; // the engine retries the step: its attempt number counts the calls
    if (calls < 3) throw new InvalidOperationException($"flaky call {calls}");
    return Task.FromResult<object?>(new { calls });
}, retry: Dsl.RetryPolicy(attempts: 3, initialMs: 100)),

Tested against a real server in every language shown.

Non-retryable errors

A step whose error type is in non_retryable fails at once, whatever attempts are left. The type is matched exactly against the error_class the worker reports: the simple class name in Python, Java and .NET; error.class.name in Ruby and get_class in PHP (with their module or namespace, for example Billing::CardDeclined, App\CardDeclined); err.name in TypeScript (set this.name in the error's constructor); the type's name in Go (pointers removed, no package); the error's class in Rust (Error::with_class). Raising the SDK's own non-retryable error (NonRetryableError, ...) also ends the step (retryable: false).

Crystal

# A failure the `nonretry` workflow lists as non-retryable (top level: its class name is the error type, as the SDKs').
class CardDeclined < Exception
end

Python

class CardDeclined(Exception):
    """A failure the `nonretry` workflow lists as non-retryable: its class name is the error type the engine matches."""

TypeScript

// The `nonretry` workflow lists this as non-retryable: its `name` is the error type (`error_class`) the engine matches.
class CardDeclined extends Error {
  constructor(message: string) {
    super(message)
    this.name = 'CardDeclined'
  }
}

Ruby

# A failure the `nonretry` workflow lists as non-retryable: defined at the top level, so its class name (the error type
# the worker reports and the engine matches) is exactly "CardDeclined".
class CardDeclined < StandardError; end

Go

// CardDeclined is the failure the nonretry workflow lists as non-retryable: the SDK reports an error's type name
// (without the package) as its error_class, so this one is "CardDeclined".
type CardDeclined struct{}

func (*CardDeclined) Error() string { return "card declined" }

Rust

fn declined(_i: Value, _c: &Ctx) -> joobq::Result<Value> {
    // Its class is the error type the `nonretry` workflow lists as non-retryable (a retryable error otherwise).
    Err(Error::new("card declined").with_class("CardDeclined"))
}

Java

/** A failure the {@code nonretry} workflow lists as non-retryable: the worker reports its simple class name. */
static final class CardDeclined extends RuntimeException {
    CardDeclined(String message) {
        super(message);
    }
}

PHP

/** A failure the `nonretry` workflow lists as non-retryable: a global class, so its reported error type is `CardDeclined`. */
final class CardDeclined extends RuntimeException
{
}

.NET

/// <summary>A failure the <c>nonretry</c> workflow lists as non-retryable: its class name is the error type the engine matches.</summary>
public sealed class CardDeclined : Exception
{
    public CardDeclined(string message) : base(message) { }
}

Tested against a real server in every language shown.

Crystal

workflow "conf-nonretry-#{LANG}" do
  step "decline", Declined, retry: retry_policy(attempts: 5, initial: 10.milliseconds, non_retryable: ["CardDeclined"])
end,

Python

wf("nonretry").step("decline", declined,
                    retry=retry_policy(attempts=5, initial_ms=10, non_retryable=["CardDeclined"])),

TypeScript

wf('nonretry').step('decline', declined, { retry: retryPolicy({ attempts: 5, initialMs: 10, nonRetryable: ['CardDeclined'] }) }),

Ruby

wf("nonretry").step("decline", ->(_i, _ctx) { raise CardDeclined, "card declined" },
                    retry: retry_policy(attempts: 5, initial_ms: 10, non_retryable: ["CardDeclined"])),

Go

wf("nonretry").Step("decline", func(map[string]any, *EC) (any, error) { return nil, &CardDeclined{} },
	joobq.WithRetry[*App](joobq.RetryPolicy(5, 10, 0, 0).WithNonRetryable("CardDeclined"))),

Rust

wf("nonretry").step_with("decline", declined,
    StepOptions { retry: Some(retry_policy(5, 10, 2.0, 30_000).non_retryable(["CardDeclined"])), ..Default::default() }),

Java

wf("nonretry").step("decline", Worker::declined,
    StepOptions.<App>none().retry(DSL.retryPolicy(5, 10, 2.0, 30_000, List.of("CardDeclined")))),

PHP

wf('nonretry')->step('decline', fn () => throw new CardDeclined('card declined'),
    retry: DSL::retryPolicy(attempts: 5, initialMs: 10, nonRetryable: ['CardDeclined'])),

.NET

Wf("nonretry").Step("decline", Declined, retry: Dsl.RetryPolicy(attempts: 5, initialMs: 10, nonRetryable: new[] { "CardDeclined" })),

Tested against a real server in every language shown.

Heartbeats

With heartbeat_ms, the task carries heartbeat_ms and ctx.heartbeat() also sends POST /sdk/tasks/:token/heartbeat (at most every third of it). Only that call is a heartbeat to the engine: the worker's own lease renewals (POST /sdk/tasks/heartbeat) are not, so a handler that hangs in a live worker is timed out.

Crystal

workflow "conf-heartbeat-#{LANG}" do
  step "beat", Beating, heartbeat: 2500.milliseconds, retry: retry_policy(attempts: 2, initial: 10.milliseconds)
end,

Python

wf("heartbeat").step("beat", beating, heartbeat_ms=2500, retry=retry_policy(attempts=2, initial_ms=10)),

TypeScript

wf('heartbeat').step('beat', beating, { heartbeatMs: 2500, retry: retryPolicy({ attempts: 2, initialMs: 10 }) }),

Ruby

wf("heartbeat").step("beat", BEATING, heartbeat_ms: 2500, retry: retry_policy(attempts: 2, initial_ms: 10)),

Go

wf("heartbeat").Step("beat", func(_ map[string]any, ec *EC) (any, error) {
	// Attempt 1 never beats and is timed out by the heartbeat timeout; attempt 2 beats and finishes.
	if ec.Task.Attempt == 1 {
		time.Sleep(6 * time.Second)
		return map[string]any{"attempt": 1}, nil
	}
	for range 10 {
		_, _ = ec.Heartbeat()
		time.Sleep(200 * time.Millisecond)
	}
	return map[string]any{"attempt": ec.Task.Attempt}, nil
}, joobq.WithHeartbeat[*App](2500), joobq.WithRetry[*App](joobq.RetryPolicy(2, 10, 0, 0))),

Rust

wf("heartbeat").step_with("beat", beating,
    StepOptions { heartbeat_ms: Some(2500), retry: Some(retry_policy(2, 10, 2.0, 30_000)), ..Default::default() }),

Java

wf("heartbeat").step("beat", Worker::beating,
    StepOptions.<App>none().heartbeatMs(2500).retry(DSL.retryPolicy(2, 10, 2.0, 30_000))),

PHP

wf('heartbeat')->step('beat', 'beating', heartbeatMs: 2500, retry: DSL::retryPolicy(attempts: 2, initialMs: 10)),

.NET

Wf("heartbeat").Step("beat", Beating, heartbeatMs: 2500, retry: Dsl.RetryPolicy(attempts: 2, initialMs: 10)),

Tested against a real server in every language shown.

The handler: attempt 1 never beats and is timed out, attempt 2 beats and finishes.

Crystal

# Attempt 1 never beats and is timed out by its heartbeat; attempt 2 beats and finishes.
class Beating
  def perform(input : JSON::Any, ctx : JoobQPro::Workflows::ActivityContext) : JSON::Any
    if ctx.attempt == 1
      sleep 6.seconds
    else
      10.times do
        ctx.heartbeat
        sleep 200.milliseconds
      end
    end
    Conf.any({attempt: ctx.attempt})
  end
end

Python

def beating(inp, ctx):
    # Attempt 1 never beats and is timed out by heartbeat_ms; attempt 2 beats and finishes.
    if ctx.attempt == 1:
        time.sleep(6)
        return {"attempt": ctx.attempt}
    for _ in range(10):
        ctx.heartbeat()
        time.sleep(0.2)
    return {"attempt": ctx.attempt}

TypeScript

// Attempt 1 never beats and is timed out by heartbeatMs; later attempts beat and finish.
const beating = async (_i: any, ctx: Ctx) => {
  if (ctx.attempt === 1) { await sleep(6000); return { attempt: ctx.attempt } }
  for (let n = 0; n < 10; n++) { await ctx.heartbeat(); await sleep(200) }
  return { attempt: ctx.attempt }
}

Ruby

# Attempt 1 never beats and is timed out by heartbeat_ms; attempt 2 beats and finishes.
BEATING = lambda { |_i, ctx|
  if ctx.attempt == 1
    sleep 6
    next { attempt: ctx.attempt }
  end
  10.times do
    ctx.heartbeat
    sleep 0.2
  end
  { attempt: ctx.attempt }
}

Rust

fn beating(_i: Value, ctx: &Ctx) -> joobq::Result<Value> {
    // Attempt 1 never beats and is timed out by heartbeat_ms; attempt 2 beats and finishes.
    if ctx.attempt() == 1 {
        std::thread::sleep(std::time::Duration::from_secs(6));
        return Ok(json!({"attempt": 1}));
    }
    for _ in 0..10 {
        ctx.heartbeat()?;
        std::thread::sleep(std::time::Duration::from_millis(200));
    }
    Ok(json!({"attempt": ctx.attempt()}))
}

Java

/** Attempt 1 never beats and is timed out by heartbeat_ms; later attempts beat and finish. */
static Object beating(Object i, JoobQClient.ExecutionContext<App> ctx) throws Exception {
    if (ctx.attempt() == 1) {
        Thread.sleep(6000);
        return map("attempt", ctx.attempt());
    }
    for (int n = 0; n < 10; n++) {
        ctx.heartbeat();
        Thread.sleep(200);
    }
    return map("attempt", ctx.attempt());
}

PHP

/** Attempt 1 never beats and is timed out by heartbeat_ms; a later attempt beats and finishes. */
function beating(mixed $i, ExecutionContext $ctx): array
{
    $attempt = (int)$ctx->task['attempt'];
    if ($attempt === 1) {
        sleep_ms(6000);
        return ['attempt' => $attempt];
    }
    for ($n = 0; $n < 10; $n++) {
        $ctx->heartbeat();
        sleep_ms(200);
    }
    return ['attempt' => $attempt];
}

.NET

static async Task<object?> Beating(JsonNode? i, ExecutionContext<App> ctx)
{
    // Attempt 1 never beats and is timed out by heartbeat_ms; attempt 2 beats and finishes.
    if (ctx.Attempt == 1)
    {
        await Task.Delay(6000);
        return new { attempt = ctx.Attempt };
    }
    for (var n = 0; n < 10; n++)
    {
        await ctx.HeartbeatAsync();
        await Task.Delay(200);
    }
    return new { attempt = ctx.Attempt };
}

Not in every language

• Go: The Go worker writes this handler inline, in the blueprint shown above.

Tested against a real server in every language shown.

Every worker of a task queue whose blueprints use heartbeat_ms must run an SDK that beats: an older worker ignores the task's heartbeat_ms, never beats, and its steps are timed out.

Workflow timeout

Crystal

workflow "conf-deadline-#{LANG}", timeout: 1500.milliseconds do
  step "slow", Sleepy
end,

Python

workflow(f"conf-deadline-{LANG}", task_queue=WF_QUEUE, timeout_ms=1500).step("slow", sleepy),

TypeScript

workflow<App>(`conf-deadline-${LANG}`, { taskQueue: WF_QUEUE, timeoutMs: 1500 }).step('slow', sleepy),

Ruby

wf("deadline", timeout_ms: 1500).step("slow", lambda { |_i, _ctx|
  sleep 4
  { ok: true }
}),

Go

joobq.Workflow[*App](fmt.Sprintf("conf-deadline-%s", lang), joobq.WorkflowOptions{TaskQueue: wfQueue, TimeoutMs: 1500}).
	Step("slow", func(map[string]any, *EC) (any, error) {
		time.Sleep(4 * time.Second)
		return map[string]any{"ok": true}, nil
	}),

Rust

wf("deadline").timeout_ms(1500).step("slow", sleepy),

Java

wf("deadline").timeoutMs(1500).step("slow", Worker::sleepy),

PHP

DSL::workflow('conf-deadline-' . LANG, timeoutMs: 1500, taskQueue: WF_QUEUE)->step('slow', function () {
    sleep_ms(4000);
    return ['ok' => true];
}),

.NET

Dsl.Workflow<App>($"conf-deadline-{Lang}", timeoutMs: 1500, taskQueue: WfQueue).Step("slow", Sleepy),

Tested against a real server in every language shown.

The run above ends timed_out after 1.5 s while its only step sleeps for 4 s.