Reliability Guide

This guide summarizes reliability practices for task systems using Stem.

Recovery workflow

1. Identify the failing task or queue.
2. Inspect recent errors and DLQ entries.
3. Fix the root cause before replaying.
4. Replay only the affected tasks.

Broker fetch notes

• Redis Streams uses consumer groups plus XAUTOCLAIM to reclaim idle deliveries; long-running tasks should rely on automatic renewal or explicitly extend their leases.
• Postgres uses polling with locked_until leases; tasks become visible again after the lease expires.

Worker lease renewal is isolated from the task loop. A failed automatic renewal is contained, recorded as the stem.lease.renewal_failed metric, and logged with the task, queue, and broker receipt; a shorter retry cadence gives transient broker failures another chance before expiry. If the lease is ultimately lost, the broker may redeliver the task. Stem persists the terminal result before acknowledging, so a redelivery can observe terminal state instead of executing the handler a second time.

The same lease remains active through terminal result persistence, group/chord bookkeeping, retry or dead-letter publication, linked-task dispatch, and acknowledgement. A slow backend or broker round trip during that final handling window therefore gets the same renewal protection as the handler itself.

Lease protection starts as soon as the delivery enters the worker, so slow consume middleware, signature validation, backend lookups, and rate-limit decisions are covered before handler execution begins.

If the same envelope is redelivered while its first delivery is still active in the same worker, the duplicate is acknowledged without invoking the handler again. This is a worker-local guard; recovery after a process crash still follows Stem's at-least-once delivery model and requires idempotent external side effects.

Hard shutdown requeues active deliveries, including deliveries already held in the broker prefetch window. A replacement worker can therefore drain the full batch; applications should still expect at-least-once execution for any handler that was already running when the process stopped.

When two different workers complete the same task concurrently, built-in result backends use AtomicTerminalResultBackend to arbitrate the terminal state. Only the worker that wins that atomic write performs terminal group, chord, linked-task, signal, and unique-lock side effects. This prevents a late success or failure record from replacing the result that already won. Custom result backends remain compatible, but must implement that optional capability if they need the same cross-process first-writer-wins guarantee.

The worker suite also exercises the failure path where lease renewal fails while a handler is still running. After the visibility timeout, a replacement worker receives the redelivery and may finish first; the original handler's late completion is then ignored by terminal arbitration. This is still at-least-once execution: external side effects must be idempotent, and a successful handler is not proof that its broker acknowledgement was durable.

Renewal is scheduled from roughly half of the remaining lease. The configured minimum interval is used for ordinary leases, but it does not delay renewal of short leases past their deadline. If the lease duration is shorter than the default one-second floor, Stem uses the shorter safe interval instead.

Workflow lease notes

• Workflow runs are lease-based. Workers must renew leases while executing, and other workers can take over after the lease expires.
• SQLite integration coverage reopens the durable workflow store after a worker/runtime restart and resumes a persisted checkpoint without re-running the completed step. The same checkpoint/replay contract applies to the other durable workflow stores, subject to their adapter guarantees.
• Keep runLeaseDuration >= broker visibility timeout to prevent redelivered workflow tasks from being dropped before takeover is possible.
• Keep leaseExtension renewals ahead of both the workflow lease expiry and the broker visibility timeout.

Poison-pill handling

• Payload-decoding failures are terminal: Stem acknowledges the delivery, records a failed result, and dead-letters it with reason invalid-payload. They do not consume the task's normal handler retry budget because retrying the same malformed bytes cannot repair the message.
• If a task fails repeatedly for the same reason, treat it as a poison pill.
• Move it to the DLQ and add guardrails or validation to prevent repeats.
• Record the failure pattern for future detection.

Scheduler reliability

• Run multiple Beat instances only when backed by a shared lock store.
• Beat revalidates lock ownership immediately before publishing; a lost lease records a failed dispatch instead of publishing after another scheduler may have acquired the entry.
• Fenced lock stores attach stem-lock-fencing-token to scheduled envelopes. Downstream state stores that support fencing should reject writes carrying a token older than the last accepted token. The token prevents stale owners from overwriting state after a lease expires; it does not make external side effects exactly once; this remains at-least-once execution.
• Monitor schedule drift and failures to detect store latency.
• Re-apply schedules after deploys to ensure definitions stay current.

Retries and backoff

• Use bounded retries with jittered backoff to avoid thundering herds.
• Separate transient failures from permanent failures.
• For permanent errors, fail fast and alert.

Configure a jittered retry strategy

retry_task/bin/worker.dart

  final worker = Worker(
    broker: broker,
    tasks: tasks,
    backend: backend,
    queue: 'retry-demo',
    consumerName: workerName,
    retryStrategy: ExponentialJitterRetryStrategy(
      base: const Duration(milliseconds: 200),
      max: const Duration(seconds: 1),
    ),
  );

Log retry signals and outcomes

retry_task/lib/shared.dart

List<SignalSubscription> attachLogging(String label) {
  String prefix(String event) => '[retry][$label][$event]';

  void log(String event, Map<String, Object?> data) {
    // ignore: avoid_print
    print('${prefix(event)} ${jsonEncode(data)}');
  }

  return <SignalSubscription>[
    StemSignals.onBeforeTaskPublish((payload, context) {
      log('before_task_publish', {
        'task': payload.envelope.name,
        'id': payload.envelope.id,
        'attempt': payload.attempt,
        'sender': context.sender,
      });
    }),
    StemSignals.onTaskPrerun((payload, _) {
      log('task_prerun', {
        'task': payload.envelope.name,
        'id': payload.envelope.id,
        'attempt': payload.context.attempt,
        'worker': payload.worker.id,
      });
    }),
    StemSignals.onTaskRetry((payload, _) {
      log('task_retry', {
        'task': payload.envelope.name,
        'id': payload.envelope.id,
        'attempt': payload.attempt,
        'reason': payload.reason.toString(),
        'nextRunAt': payload.nextRetryAt.toIso8601String(),
        'worker': payload.worker.id,
      });
    }),
    StemSignals.onTaskFailure((payload, _) {
      log('task_failed', {
        'task': payload.envelope.name,
        'id': payload.envelope.id,
        'attempt': payload.attempt,
        'worker': payload.worker.id,
        'error': payload.error.toString(),
      });
    }),
    StemSignals.onTaskPostrun((payload, _) {
      log('task_postrun', {
        'task': payload.envelope.name,
        'id': payload.envelope.id,
        'state': payload.state.name,
        'worker': payload.worker.id,
      });
    }),
  ];
}

Create a task that exercises retries

retry_task/lib/shared.dart

FutureOr<void> _alwaysFailEntrypoint(
  TaskInvocationContext context,
  Map<String, Object?> args,
) {
  final attempt = context.attempt;
  final max = context.meta['maxRetries'] ?? 3;
  // ignore: avoid_print
  print('[retry][task][attempt] {"attempt":$attempt,"max":$max}');
  throw StateError('Simulated failure on attempt $attempt');
}

Heartbeats and progress

Use heartbeats and progress updates to make long-running tasks observable. Automatic worker lease renewal protects normal executions; call context.extendLease(...) when a task needs an explicit additional lease.

Configure worker heartbeat intervals

progress_heartbeat/bin/worker.dart

  final worker = Worker(
    broker: broker,
    tasks: tasks,
    backend: backend,
    queue: progressQueue,
    subscription: RoutingSubscription.singleQueue(progressQueue),
    consumerName: workerName,
    heartbeatInterval: const Duration(seconds: 2),
    workerHeartbeatInterval: const Duration(seconds: 5),
    prefetchMultiplier: 1,
  );

Emit progress and heartbeats inside a task

progress_heartbeat/lib/shared.dart

class ProgressTask extends TaskHandler<String> {
  @override
  String get name => 'progress.demo';

  @override
  TaskOptions get options => const TaskOptions(queue: progressQueue);

  @override
  TaskMetadata get metadata => const TaskMetadata(
    description: 'Task that reports progress and heartbeats on a loop.',
    tags: ['progress', 'heartbeat'],
  );

  @override
  Future<String> call(TaskContext context, Map<String, Object?> args) async {
    final steps = (args['steps'] as num?)?.toInt() ?? 10;
    final delayMs = (args['delayMs'] as num?)?.toInt() ?? 800;

    stdout.writeln('[task][start] id=${context.id} steps=$steps');

    for (var i = 0; i < steps; i += 1) {
      context.heartbeat();
      await context.progress(
        (i + 1) / steps,
        data: {'step': i + 1, 'total': steps},
      );
      await Future<void>.delayed(Duration(milliseconds: delayMs));
    }

    stdout.writeln('[task][done] id=${context.id}');
    return 'ok';
  }
}

Stream worker events for observability

progress_heartbeat/lib/shared.dart

void attachWorkerEventLogging(Worker worker) {
  worker.events.listen((event) {
    switch (event.type) {
      case WorkerEventType.progress:
        stdout.writeln(
          '[event][progress] id=${event.envelope?.id} progress=${event.progress} data=${event.data}',
        );
        break;
      case WorkerEventType.heartbeat:
        stdout.writeln('[event][heartbeat] id=${event.envelopeId}');
        break;
      case WorkerEventType.completed:
        stdout.writeln('[event][completed] id=${event.envelope?.id}');
        break;
      case WorkerEventType.failed:
        stdout.writeln('[event][failed] id=${event.envelope?.id}');
        break;
      case WorkerEventType.revoked:
        stdout.writeln('[event][revoked] id=${event.envelope?.id}');
        break;
      case WorkerEventType.retried:
      case WorkerEventType.timeout:
      case WorkerEventType.error:
        stdout.writeln('[event][${event.type.name}] id=${event.envelope?.id}');
        break;
    }
  });
}

Observability signals

• Track retry rates and DLQ volume as reliability signals.
• Monitor queue backlog and worker heartbeats to detect stalls.
• Tie task IDs to business logs for fast root-cause analysis.
• Use StemSignals.taskRetry / taskFailed to drive notifications when error rates spike.

Operational checks

The optional stem_cli package can inspect the configured adapter context:

stem observe queues
stem observe workers
stem dlq list --queue <queue>

Run stem <command> --help for the installed version's required options.

Next steps

• Troubleshooting
• Observability & Ops
• Worker Control CLI