khepri 0.1.1

Laravel-style queues, jobs and workers for Maat applications.

Khepri

Queues, jobs and workers for Maat applications. Khepri is the scarab that rolls the sun over the horizon each morning: work that has to happen, reliably, out of sight.

Khepri is a queue kernel, not a queue platform. It does the one thing a backend genuinely cannot ship without — move slow work off the request and run it again if it fails — and stops there.

What it is not

Deliberately absent, and not planned for this release:

Akhet, or any dashboard	No UI, no metrics, no monitoring surface.
A Redis driver	One driver: the database.
Batches and chains	No fan-out, no job-to-job sequencing.
A job middleware ecosystem	No WithoutOverlapping, no rate limiting, no unique jobs.
Scheduler or cron integration	Khepri does not decide when work is created.
Supervisors and autoscaling	Run queue:work under whatever your host already uses — systemd, Forge daemons, a container restart policy.
Multiple connections	One bound QueueConnection. A connections map arrives with the second driver, not before.

Khepri also does not depend on the Application Graph, and must not start to. A graph may dispatch jobs later; a job never needs to know a graph exists.

Registering a job

Dart AOT has no reflection, so a payload cannot rebuild its own class the way Laravel's serialize() does. A job therefore states its data as JSON and is rebuilt by a factory registered against a stable name:

class SendWelcomeEmail extends Job {
  SendWelcomeEmail(this.userId);

  static SendWelcomeEmail fromPayload(Map<String, Object?> payload) =>
      SendWelcomeEmail(payload['user_id'] as int);

  final int userId;

  @override
  Map<String, Object?> toPayload() => {'user_id': userId};

  @override
  int get tries => 3;

  @override
  List<Duration> get backoff => const [Duration(seconds: 10), Duration(minutes: 1)];

  @override
  Future<void> handle() async {
    // ...
  }
}

Register it from your own provider's boot, which runs after every provider's register:

class AppServiceProvider extends ServiceProvider {
  AppServiceProvider(super.app);

  @override
  void boot() {
    app.make<JobRegistry>()
      ..register<SendWelcomeEmail>('send_welcome_email', SendWelcomeEmail.fromPayload);
  }
}

The name lives in the registry, not on the job class, so there is exactly one place the payload name and the class are paired. Renaming a job means registering the new name and keeping the old one until the queue has drained.

Dispatching and working

Add QueueServiceProvider to bootstrap/app.dart, ...khepriMigrations to database/migrations.dart, and ...khepriCommands() to your console kernel.

await dispatch(SendWelcomeEmail(user.id));
await dispatch(GenerateReport(id), queue: 'reports', delay: Duration(minutes: 5));

$ sesh migrate
$ sesh queue:work --queue=high,default

queue:work polls each queue in the order given — default is only looked at when high is empty. SIGINT and SIGTERM ask it to stop, and it exits after the job in flight finishes. --once runs a single job; --stop-when-empty drains and returns.

Recovering a failed job

$ sesh queue:failed
ID                                    Queue    Job                 Failed at            Error
0f3c...                               default  send_welcome_email  2026-09-05 14:32:07  SocketException: connection refused

$ sesh queue:retry 0f3c...          # one job, back onto the queue it failed on
$ sesh queue:retry --all            # everything in the table

A retry replays the exact payload the job failed with, starting again at attempt one. The record is dropped only after the job is back on the queue: if the process dies between the two, the job is queued and still listed, so it can be retried twice — which beats losing it.

Configuration is optional; these are the defaults:

Map<String, dynamic> get queue => {
  'table': 'jobs',
  'queue': 'default',
  'retry_after': 90,
  'failed_table': 'failed_jobs',
};

Guarantees

• At-least-once delivery. A job runs at least once; it can run more than once if a worker dies mid-job, so job handlers must be idempotent.
• No double reservation. Two workers never hold the same job at the same time. Reservation is a compare-and-swap on reserved_at — read a candidate, claim it with a conditional UPDATE, believe the affected-row count. This needs no dialect-specific locking, so it behaves the same on SQLite and Postgres.
• No lost work on a crash. A reserved job whose worker disappears becomes available again after retry_after. Set that comfortably longer than your slowest job: too short and a slow job runs twice.
• Retries are bounded by the payload, not the class. tries and backoff are captured when the job is pushed, so a deploy that changes them does not change the terms of a job already in flight.
• A failure is recorded, never dropped. A job that spends its attempts is written to failed_jobs with the exact payload it failed with — replayable with pushRaw — and only then removed from the queue.
• A poison payload does not wedge the queue. An envelope this build cannot decode, or one naming a job class that is not registered, fails immediately rather than retrying forever.
• Graceful shutdown finishes the job in hand. The stop flag is only read between jobs.

Adding a driver later

Everything above QueueConnection — dispatch, the worker, the failed-job store — is driver-agnostic. A Redis or SQS driver implements QueueConnection and ReservedJob and is rebound in the container. No job class changes.