From one task to a queue across processes.

This walks the library in the order it is used: a task that runs, a worker that runs it, what happens when it fails, what happens after it succeeds, and how to move the whole thing onto a server.

Install and import

Add the package and import the two names you need first. Time is always a function you pass in: the library never reads the system clock, which is what makes the examples below reproducible.
```
moon add moonbitstack/moontask
```
```
import {
  "moonbitstack/moontask",
  "moonbitstack/moontask/spec",
}
```

An app, a task, a worker

The app owns three things: where messages go (Broker), where results go (Backend), and what counts the parts of a chord (Counter). The in-memory broker is all three at once, which is enough to run the whole library in one process.

let clock : Clock = { at: 1_700_000_000_000L }
let now = fn() { clock.at }
let broker = @moontask.MemoryBroker::new(clock=now)
let app = @moontask.App::new(broker, backend=broker, counter=broker, clock=now)

let greet = app.task("greet", fn(name : String, ctx : @moontask.Context) {
  "hello \{name}, attempt \{ctx.attempt}"
})

let worker = app.worker(concurrency=2)
@async.with_task_group(async fn(group) {
  group.spawn_bg(async fn() { worker.run() }, no_wait=true)
  let handle = greet.send("ada")
  println(handle.get(timeout=5000L))
  worker.stop()
})

It prints:

hello ada, attempt 1

When a run fails

A task body that raises is recorded, not lost. Inside the retry budget the message is published again with an eta computed from the policy's backoff; past the budget the failure is what the store keeps. The spread inside an exponential backoff is derived from the task id, so the same task waits the same on every machine and a test can assert the delay.
```
let flaky = app.task("flaky", fn(_ : Unit, _) -> Unit raise {
  attempts.push(1)
  if attempts.length() < 2 {
    fail("not yet")
  }
}, policy={ ..@moontask.policy, retries: 2, backoff: @moontask.Backoff::Fixed(delay=1000L) })
```
Between the two attempts the handle reports RETRY; once the second attempt returns it reports SUCCESS:
```
state   RETRY
state   SUCCESS
attempts 2
```
The policy also says how long one run may take (timeout), whether the message is confirmed before or after the run (acks_late), whether the result is kept (ignore_result) and how fast the task may go (rate).
What happens after a run succeeds

A chain is one message carrying its own tail; a group is several messages published together; a chord counts its parts and sends one callback. The worker that finishes last is the one that advances the canvas, so there is no coordinator to run.
```
let chained = @moontask.chain([step.sig(1), step.sig(2), step.sig(3)]).send(app)
println(chained.get(timeout=5000L).stringify())          // 30
let grouped = @moontask.group([step.sig(4), step.sig(5)]).send(app)
println(grouped.get(timeout=5000L).stringify())          // [40,50]
let joined = @moontask.chord([step.sig(1), step.sig(2), step.sig(3)], total.sig([])).send(app)
println(joined.get(timeout=5000L).stringify())           // 60
```
The chord's callback is sent with the list of part results, in the order the parts were given. A part that fails for good records the callback as failed instead of leaving it pending forever.

On a rhythm

A beat sends tasks on a schedule: every interval, on a cron boundary, or once at an instant.

let beat = app.beat()
beat.add("tick", tick, (), @moontask.Schedule::Every(interval=1000L))
beat.add("nightly", report, (), @moontask.Schedule::Cron(@cron.parse("0 0 2 * * *")))

group.spawn_bg(async fn() { beat.run() }, no_wait=true)

Inside a single process, without a broker, a scheduler runs a job on the same kind of rhythm:

let scheduler = @schedule.Scheduler::new(clock=now)
scheduler.add(@schedule.every(10).minutes().at(":30").run(fn() { sweep() }))
scheduler.run()

The next run is computed from the run that was due, not from the moment the loop woke, so a loop that was busy for a while does not drag the rhythm with it.

Across processes

The Redis module implements the same three contracts, so the app above changes in one line. A queue is a list per name; messages waiting for their eta sit in a sorted set; each consumer keeps what it holds in a list of its own, kept alive by a lease.
```
let conn = @socket.Connection::connect(@moonredis.ClientOptions::at("127.0.0.1", 6379))
let broker = @redis.Broker::new(conn, clock=now, consumer="worker-7")
let app = @moontask.App::new(broker, backend=broker.store(), counter=broker.store(), clock=now)
```
A worker that dies stops refreshing its lease. recover hands the work it was holding back to the queue, which is what makes a killed worker's tasks run again:
```
println(broker.recover())   // messages moved back to their queues
```
The same module can speak Celery's layout instead, so one queue can be worked from both sides. The ready list, the envelope, the reservation table and a chord's join keys then belong to Celery, and either side can run what the other published — and hand back what a worker on the other side left behind:
```
let broker = @redis.Broker::celery(conn, clock=now, consumer="worker-7")
```

A page over the queue

The management module answers the same questions over HTTP, and is an ASGI application rather than a server of its own, so a server runs it and a framework mounts it.

let admin = @admin.Admin::new(app, inspector=broker)
let asgi = admin.to_asgi()
// GET  /                 the page
// GET  /api/tasks        what is registered, and how each runs
// GET  /api/queues       ready, delayed and unconfirmed counts
// GET  /api/results/{id} the state of one run
// POST /api/tasks/{name} send a run
// GET  /api/events       a stream of counts

Writing a broker

A transport implements Broker and the library runs on it unchanged. Four calls: publish, claim, confirm, reject. The receipt the claim hands back is opaque to the library — a sequence number, a log entry, whatever the transport needs to settle the message later.

pub(open) trait Broker {
  async fn publish(Self, queue : String, message : Message) -> Unit raise Refused
  async fn consume(Self, queues : ArrayView[String], prefetch~ : Int32) -> Delivery raise Refused
  async fn ack(Self, delivery : Delivery) -> Unit raise Refused
  async fn reject(Self, delivery : Delivery, requeue~ : Bool) -> Unit raise Refused
}

From one task to a queue across processes.

Install and import

An app, a task, a worker

When a run fails

What happens after a run succeeds

On a rhythm

Across processes

A page over the queue

Writing a broker