Tasks and the registry

A task is a named async function. You register it once with app.task<T>() in code that both producers and workers load; workers look the name up in the registry when a message arrives.

import 'package:gisila_queue/gisila_queue.dart';

final app = GisilaQueue(broker: InMemoryBroker(), backend: InMemoryResultBackend());

final add = app.task<int>(
  'math.add',
  (ctx) async => (ctx.arg(0) as int) + (ctx.arg(1) as int),
  maxRetries: 3,
);

GisilaQueue.task()

Registers handler under name and returns a TaskRef<T>.

TaskRef<T> task<T>(
  String name,
  TaskHandler handler, {
  String? queue,
  int maxRetries = 0,
  Duration retryBackoff = const Duration(seconds: 1),
  Duration retryBackoffMax = const Duration(minutes: 10),
  bool retryJitter = true,
  bool autoRetry = false,
  Duration? timeLimit,
  Duration? softTimeLimit,
  RateLimit? rateLimit,
})

When to use: for every unit of background work. Names are the contract between producers and workers, so use stable dotted names (emails.send_welcome) and register them in a shared function that both entrypoints call.

Parameter	Type	Default	Description
name	String	required	Unique name. Registering it twice throws ArgumentError.
handler	TaskHandler	required	The work.
queue	String?	null	Default queue for this task. Used only when the call doesn’t pass queue: and the Router doesn’t match.
maxRetries	int	0	Retries allowed after the first run. Copied into each message by TaskRef.delay/applyAsync.
retryBackoff	Duration	1 s	Base for exponential backoff.
retryBackoffMax	Duration	10 min	Backoff cap.
retryJitter	bool	true	Randomise each delay into [d/2, d].
autoRetry	bool	false	Retry on any uncaught exception while retries remain. Without it only ctx.retry() retries.
timeLimit	Duration?	null	Hard limit: the worker stops waiting and records a failure (or retry). The handler is not killed.
softTimeLimit	Duration?	null	Cooperative limit exposed as ctx.softTimeLimitExceeded. Not enforced.
rateLimit	RateLimit?	null	Max runs per window, per worker process.

Caution

maxRetries only reaches the worker through the message. TaskRef.delay/applyAsync and TaskRef.sig copy it in. A call to app.send() uses its own maxRetries argument (default 0), and tasks fired by Beat or by GisilaRose’s fire action are sent with maxRetries: 0, so they never retry.

See also: Retries and acks, Retries concept.

TaskHandler

typedef TaskHandler = Future<Object?> Function(TaskContext ctx);

The return value becomes the task’s result. It must be JSON-encodable (num, String, bool, null, List, Map), because the result backend stores it as JSON.

TaskContext

Per-run context passed to the handler.

class TaskContext {
  final String id;
  final String task;
  final List<Object?> args;
  final Map<String, Object?> kwargs;
  final int retries;
  final int maxRetries;
  final String queue;
  final QueueLogger logger;
  final DateTime startedAt;
  final Duration? softTimeLimit;
  ...
}

When to use: read arguments, decide whether to retry, and poll the cooperative cancellation flags in long handlers.

Member	Type	Description
id	String	Task id, same as the producer’s AsyncResult.id.
task	String	Registered name.
args / kwargs	List<Object?> / Map<String, Object?>	Call arguments, decoded from JSON.
arg(int i)	Object?	args[i], or null if out of range.
retries	int	Retries already done (0 on the first run).
maxRetries	int	Retry ceiling carried by the message.
canRetry	bool	retries < maxRetries.
queue	String	Queue the message came from.
logger	QueueLogger	The app’s logger.
startedAt	DateTime	When this run started.
softTimeLimit	Duration?	The task’s soft limit.
softTimeLimitExceeded	bool	true once softTimeLimit has passed since startedAt.
isAbandoned	bool	true once the hard timeLimit passed and the worker stopped waiting. Anything returned afterwards is discarded.
isRevoked	bool	true if a revoke for this id arrived on the control channel while running. Cooperative only.
retry({Duration? countdown, Object? cause})	Never	Throws RetryTask. The worker re-enqueues if canRetry, otherwise records a failure. countdown overrides the backoff.
markAbandoned()	void	Called by the worker; not for handlers.

Example

app.task<int>('reports.render', (ctx) async {
  var pages = 0;
  for (final id in (ctx.arg(0) as List).cast<int>()) {
    if (ctx.softTimeLimitExceeded || ctx.isRevoked || ctx.isAbandoned) break;
    await renderPage(id);
    pages++;
  }
  return pages;
}, timeLimit: const Duration(minutes: 5), softTimeLimit: const Duration(minutes: 4));

Task

The immutable definition stored in the registry. You normally use the TaskRef instead.

class Task {
  Task({
    required String name,
    required TaskHandler handler,
    String? queue,
    int maxRetries = 0,
    Duration retryBackoff = const Duration(seconds: 1),
    Duration retryBackoffMax = const Duration(minutes: 10),
    bool retryJitter = true,
    bool autoRetry = false,
    Duration? timeLimit,
    Duration? softTimeLimit,
    RateLimit? rateLimit,
  });

  Duration backoffFor(int retryNumber, {math.Random? rng});
}

Fields match the task() parameters above. backoffFor(n) returns the delay before the nth retry (1-based): retryBackoff * 2^(n-1), capped at retryBackoffMax, then, with retryJitter, a uniform pick in [d/2, d]. With defaults that is about 1 s, 2 s, 4 s, … up to 10 minutes.

When to use: read it through app.registry or app.lookup(name), or call backoffFor in tests.

RateLimit

Caps how often a task runs.

class RateLimit {
  const RateLimit(int count, Duration per);
  factory RateLimit.parse(String spec); // '10/s', '100/m', '1000/h'
  final int count;
  final Duration per;
}

When to use: to protect a downstream API with a known quota. The limit is enforced per worker process, not across the cluster: three workers with '10/s' can run 30 per second together. For a cluster-wide limit, run one worker process for that queue.

When a run would exceed the limit, the worker re-enqueues the message with an ETA for when a slot frees up, sets its state to scheduled, emits taskRateLimited, and acks the original. It does not count as a retry and does not hold a concurrency slot.

RateLimit.parse throws FormatException for anything other than <n>/s, <n>/m or <n>/h.

app.task<void>('sms.send', sendSms, rateLimit: RateLimit.parse('5/s'));

Other exports

Symbol	Description
SlidingWindowLimiter	The limiter the worker uses for RateLimit. tryAcquire([DateTime? now]) returns null when a slot was taken, or the Duration to wait. Non-blocking.

See also: Producer API, Workers.