🥮 mooncakes.io

    moontask

    Task queues and schedules for MoonBit: workers, retries, periodic beats, and the chain, group and chord canvas. Brokers and result stores are separate modules.

    task-queue
    job-queue
    queue
    worker
    celery
    scheduler
    cron
    crontab
    schedule
    retry
    backoff
    prefetch
    workflow
    canvas
    chain
    group
    chord
    dag
    beat
    periodic
    background-jobs
    distributed
    moonbit
    Download zip
    Version
    0.1.0
    License
    Apache-2.0
    Repository
    Last updated
    10 hours ago
    Downloads
    7

    #moontask

    Work that outlives the request that asked for it.

    Task queues, retries, periodic beats, and the canvas that chains, groups and joins runs — for MoonBit.

    Home · Guide · API · mooncakes

    An app owns a broker, a result store and a clock. A task is a name, a policy and a body; sending one writes a message and hands back a handle that decodes the stored result back into the body's return type. A worker takes messages off the broker, runs them, and settles each one the way its policy says.

    let now = fn() { @env.now() }
    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=4)
    @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()
    })

    #What is in the module

    PackageWhat it is
    moontask/specThe contracts — Broker, Backend, Counter, Inspector — and the message, outcome and state they carry
    moontaskApp, Task, Handle, Worker, Beat, the canvas, and an in-memory broker that is also a store and a counter
    moontask/scheduleA job on a rhythm inside one process, with no broker behind it
    moontask/waitThe one place a wait is written, so the WebAssembly targets are handled once

    Two further modules live beside it, each with its own moon.mod so a queue that does not need them does not download them:

    ModuleWhat it is
    moontask-redisA broker and result store on Redis, with per-consumer leases and recovery of a worker that stopped — or on Celery's own key layout, so one queue can be shared with a Celery worker
    moontask-adminA management interface as an ASGI application: a page, reports, sending, revoking, and a live stream of counts

    #A task's policy

    FieldMeaningDefault
    queueWhere the message goes"default"
    retriesHow many times a failed run is tried again0
    backoffThe wait between attemptsexponential, 1s base, 10min cap, jittered
    timeoutHow long one run may takeno limit
    acks_lateConfirm after the run instead of on receiptfalse
    ignore_resultKeep no resultfalse
    rateRuns per window, per workerno limit

    The jitter inside an exponential backoff is derived from the task id and the attempt rather than drawn at random, so the same task waits the same on every machine and a test can assert the delay.

    #The canvas

    let chained = @moontask.chain([step.sig(1), step.sig(2), step.sig(3)]).send(app)  // answers the last
    let grouped = @moontask.group([step.sig(4), step.sig(5)]).send(app)               // answers a list
    let joined  = @moontask.chord([step.sig(1), step.sig(2), step.sig(3)], total.sig([])).send(app)

    A chain is a message carrying its own tail. A chord gives every part the same counter key and the callback behind it, so whichever worker finishes last sends the callback; a part that fails for good records the callback as failed rather than leaving it pending.

    #On a rhythm

    let beat = app.beat()
    beat.add("tick", tick, (), @moontask.Schedule::Every(interval=1000L))
    beat.add("nightly", report, (), @moontask.Schedule::Cron(expr~))

    #Across processes

    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 queue is a list per name, messages waiting for their eta sit in a sorted set, and what a consumer holds is a list of its own kept alive by a lease. broker.recover() hands back what a worker that stopped refreshing its lease was holding.

    The same module also speaks Celery's layout, so a queue can be worked from both sides:

    let broker = @redis.Broker::celery(conn, clock=now, consumer="worker-7")

    the queuethe list named after it, split across Celery's priority steps
    the payloadCelery's envelope: task and id in headers, arguments in the body
    what a consumer holdsCelery's unacked hash and its index, so either side can hand back a claim the other left
    a chaineach message carries the rest of its chain; a link that is not immutable is called with the previous link's result prepended
    a chordthe parts' results land in Celery's join keys as they settle, so the last part — on either side — closes it
    an etathe message sits on the queue and the consumer holds it, which is what Celery does

    interop/ holds the harness that proves it: bash interop/run_links.sh all runs eight scenarios against a real Celery 5.5 — a chain and a chord in each direction, a reservation handed back in each direction, and an eta in each direction — and prints what each side ran.

    #Writing a broker

    Four calls. The receipt a claim hands back is opaque to the library:

    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
    }

    #Examples

    Two runnable examples ship in src/examples: 01-queue takes a task from send to answer with one retry in between, and 02-canvas runs a chain, a group and a chord. Both pin their clock, so their output is the same on every run.

    moon run src/examples/01-queue
moon run src/examples/02-canvas

    #License

    Apache-2.0.

    App

    source
    pub struct App {
      broker : &
    Broker
      backend : &
    Backend    ?
      counter : &
    Counter    ?
      clock : () -> Int64
      tasks : Map[String, Registered]
      running : Map[String, 
    Task    [Result[
    Json    , String]]]
      seq : Int
    }

    The object every other type hangs off.

    It owns the three seams (broker, result store, chord counter), the clock the library reads instead of the system clock, and the registry that turns a message name back into a body. Nothing here is global: two apps in one process do not see each other's tasks.

    App::backend_of

    source

    The result store the app was given, or None when it keeps nothing.

    App::beat

    source
    fn App::beat(self : App) -> Beat

    A beat on app with nothing scheduled.

    App::has_task

    source
    fn App::has_task(self : App, name : String) -> Bool

    Whether a body is registered under name.

    App::new

    source

    An app on broker, with backend keeping results, counter counting chord parts, and clock as the only source of time.

    The store and the counter are optional: an app without a store runs tasks nobody waits on, and an app without a counter refuses to build a chord. A broker that also implements Backend and Counter — the in-memory one and the Redis one both do — is passed three times.

    App::now

    source
    fn App::now(self : App) -> Int64

    The app's time, in milliseconds since the epoch.

    App::policy_of

    source
    fn App::policy_of(self : App, name : String) -> Policy?

    The policy registered under name, or None when nobody registered it.

    App::send

    source
    async fn App::send(self : App, name : String, args : 
    Json    , eta? : Int64, countdown? : Int64, expires? : Int64, queue? : String, priority? : Int, id? : String) -> Handle[
    Json    ] raise 
    Refused

    Publish a run of the task called name with a JSON argument, without a Task value in hand.

    This is the entry for a dispatcher that learns the name at run time — an HTTP handler, a management page, a message from another system — and the only thing it needs to keep is the name. An unknown name is refused rather than published, since a queue that accepts work nobody can run only moves the failure to a worker.

    App::task

    source
    fn[A : 
    FromJson    , R : 
    ToJson    ] App::task(self : App, name : String, run : async (A, Context) -> R, policy? : Policy) -> Task[A, R]

    Register run under name and return the task a caller sends.

    Registering twice under the same name replaces the first, which is what makes it possible to run the same task under two policies in one test.

    App::task_names

    source
    fn App::task_names(self : App) -> Array[String]

    Every registered name, sorted so a page listing them is stable.

    App::worker

    source
    fn App::worker(self : App, queues? : Array[String], concurrency? : Int, prefetch? : Int) -> Worker

    A worker over queues that runs up to concurrency tasks at once.

    prefetch is how many messages one worker may hold unsettled: a worker that holds more than it can run looks busy while the queue behind it starves.

    Backoff

    source
    pub(all) enum Backoff {
      Fixed(delay~ : Int64)
      Exponential(base~ : Int64, cap~ : Int64, jitter~ : Bool)
    } derive(Eq, 
    Debug    )

    How long to wait before the next attempt.

    Fixed waits the same time every time. Exponential doubles from base until it reaches cap, and with jitter on it spreads the wait over the whole interval instead of always waiting the full one — that is what keeps a thousand workers from retrying in the same millisecond after an outage.

    The spread is derived from the task id and the attempt number rather than drawn at random, so the same task retried twice on two machines waits the same both times, and a test can assert the delay.

    Backoff::delay

    source
    fn Backoff::delay(self : Backoff, attempt : Int, seed : String) -> Int64

    The wait before attempt attempt, counting the first attempt as one.

    seed is what the spread is derived from: the task id in practice. Attempt zero never waits, since nothing has failed yet.

    Backoff::equal

    source
    fn Backoff::equal(Backoff, Backoff) -> Bool

    Backoff::not_equal

    source
    fn Backoff::not_equal(x : Backoff, y : Backoff) -> Bool

    Backoff::to_repr

    source

    Beat

    source
    pub struct Beat {
      app : App
      entries : Array[BeatEntry]
      stopping : Bool
    }

    The object that sends tasks on a schedule.

    A beat holds a list of entries and, every tick, sends the ones whose time has come. It keeps the phase: a schedule of every hour fires at the same minute every hour even if one tick ran late, because the next time is computed from the time that was due, not from the time the tick happened.

    Beat::add

    source
    fn[A : 
    ToJson    , R] Beat::add(self : Beat, name : String, task : Task[A, R], args : A, schedule : Schedule) -> Unit

    Schedule task with args under name.

    The name identifies the entry, not the task: the same task may be scheduled twice under two names and two schedules.

    Beat::pending

    source
    fn Beat::pending(self : Beat) -> Array[(String, Int64)]

    The entry names and when each fires next, soonest first.

    Beat::run

    source
    async fn Beat::run(self : Beat) -> Unit raise 
    Refused

    Send due entries, then wait for the next one. Runs until stop.

    Beat::stop

    source
    fn Beat::stop(self : Beat) -> Unit

    Stop after the tick in progress.

    BeatEntry

    source
    type BeatEntry

    One thing the beat sends.

    Context

    source
    pub(all) struct Context {
      id : String
      name : String
      queue : String
      retries : Int
      attempt : Int
      last : Bool
      now : Int64
    } derive(Eq, 
    Debug    )

    What a task body is told about the run it is in.

    retries is how many attempts before this one failed, attempt is this one counting from one, and last says whether another attempt follows a failure — a body that must not repeat a side effect checks it before doing the work again.

    Context::equal

    source
    fn Context::equal(Context, Context) -> Bool

    Context::not_equal

    source
    fn Context::not_equal(x : Context, y : Context) -> Bool

    Context::to_repr

    source

    Handle

    source
    pub struct Handle[R] {
      app : App
      ids : Array[String]
      all : Bool
      codec : (
    Json    ) -> R raise 
    Refused
    }

    A run, or a set of runs, a caller can wait on and look at.

    The result type is the one the task returns: get decodes the stored JSON back into it. state and revoke work on anything, since a caller that wants to cancel a run does not care what it would have returned.

    Handle::get

    source
    async fn[R] Handle::get(self : Handle[R], timeout? : Int64) -> R raise 
    Refused

    Wait for the run and answer its value, or refuse with the reason it will not arrive: Refused::Failed when a body raised on its last attempt, Refused::Timeout when timeout milliseconds passed first, and Refused::Missing when the app keeps no results.

    Handle::ids

    source
    fn[R] Handle::ids(self : Handle[R]) -> Array[String]

    The ids the handle waits on.

    Handle::of

    source
    fn[R : 
    FromJson    ] Handle::of(app : App, ids : Array[String], all : Bool) -> Handle[R]

    A handle over ids whose value decodes into R; all makes get answer the list of results in the order of ids.

    Handle::of_json

    source
    fn Handle::of_json(app : App, ids : Array[String], all : Bool) -> Handle[
    Json    ]

    A handle over ids of runs that already answer in JSON.

    Handle::revoke

    source
    async fn[R] Handle::revoke(self : Handle[R], terminate? : Bool) -> Unit raise 
    Refused

    Cancel the run or the set: it will not be sent again, and with terminate a worker in this process that is running it right now stops.

    Terminating reaches only this process's workers; a run already going in another process is left to finish, since nothing else is reachable.

    Handle::state

    source

    What the run is doing now, without waiting for it.

    A set reports Failed once any part failed, Revoked once any part was revoked, Succeeded once every part settled, and the state of the first unsettled run otherwise.

    Limiter

    source
    type Limiter

    How a task is allowed to go, per worker: tokens runs left in the window that ends at reset.

    MemoryBroker

    source

    A broker, result store, chord counter and inspector over maps.

    clock is the same function the app is given: the broker needs time to know when a delayed message is due, and the library never reads the system clock itself.

    MemoryBroker::ack

    source

    MemoryBroker::add

    source
    async fn MemoryBroker::add(self : MemoryBroker, key : String, delta : Int) -> Int raise 
    Refused

    MemoryBroker::consume

    source

    MemoryBroker::drop

    source
    async fn MemoryBroker::drop(self : MemoryBroker, key : String) -> Unit raise 
    Refused

    MemoryBroker::fetch

    source

    MemoryBroker::new

    source
    fn MemoryBroker::new(clock~ : () -> Int64) -> MemoryBroker

    An empty broker whose time comes from clock.

    MemoryBroker::peek

    source

    MemoryBroker::publish

    source

    MemoryBroker::queues

    source

    MemoryBroker::reject

    source

    MemoryBroker::settle

    source

    MemoryBroker::store

    source

    Policy

    source
    pub(all) struct Policy {
      queue : String
      retries : Int
      backoff : Backoff
      timeout : Int64?
      acks_late : Bool
      ignore_result : Bool
      rate : Rate?
    } derive(Eq, 
    Debug    )

    Everything about how a task is run, retried and confirmed.

    The defaults are the ones a queue is usually run with, and each of them is the safe side of the choice: no retries (a retry repeats the side effects of the first attempt), no timeout (a hung task is visible in the queue rather than silently cut), confirm on receipt (the message is not redelivered if the worker dies, which is the cheaper mistake for idempotent work).

    Policy::backoff_delay

    source
    fn Policy::backoff_delay(self : Policy, attempt : Int, seed : String) -> Int64

    The wait before attempt attempt of a task whose id is seed.

    Policy::equal

    source
    fn Policy::equal(Policy, Policy) -> Bool

    Policy::not_equal

    source
    fn Policy::not_equal(x : Policy, y : Policy) -> Bool

    Policy::to_repr

    source

    Rate

    source
    pub(all) struct Rate {
      count : Int
      per : Int64
    } derive(Eq, 
    Debug    )

    How fast one task may go: count runs per per milliseconds.

    Rate::equal

    source
    fn Rate::equal(Rate, Rate) -> Bool

    Rate::not_equal

    source
    fn Rate::not_equal(x : Rate, y : Rate) -> Bool

    Rate::to_repr

    source

    Registered

    source
    type Registered

    A registered task, as the registry holds it: the policy and a body that takes the wire argument and answers in strings, so one map holds tasks of every type.

    Schedule

    source
    pub(all) enum Schedule {
      Every(interval~ : Int64)
      Cron(expr~ : 
    Cron    )
      At(at~ : Int64)
    } derive(Eq, 
    Debug    )

    When a beat sends a task.

    Schedule::after

    source
    fn Schedule::after(self : Schedule, at : Int64) -> Int64?

    The first time this schedule fires strictly after at, or None when it never fires again.

    Schedule::equal

    source
    fn Schedule::equal(Schedule, Schedule) -> Bool

    Schedule::first

    source
    fn Schedule::first(self : Schedule, now : Int64) -> Int64

    The first time this schedule fires at or after now.

    Schedule::not_equal

    source
    fn Schedule::not_equal(x : Schedule, y : Schedule) -> Bool

    Schedule::to_repr

    source

    Shape

    source
    pub(all) enum Shape {
      One
      All
    } derive(Eq, 
    Debug    )

    Whether a sig waits on one run or on all of them.

    Shape::equal

    source
    fn Shape::equal(Shape, Shape) -> Bool

    Shape::not_equal

    source
    fn Shape::not_equal(x : Shape, y : Shape) -> Bool

    Shape::to_repr

    source

    Sig

    source
    pub struct Sig {
      publish : Array[
    Message    ]
      waits : Array[String]
      shape : Shape
    }

    A piece of work that has not been published yet.

    The messages are ready to go; waits is what a caller waits on afterwards, which for a chain is its last link and for a chord is the callback. Building a sig publishes nothing, so a canvas can be built once and sent twice.

    Every message this sig stands for, in the order they run: the links of a chain, the parts of a group.

    Sig::send

    source

    Publish every message of the sig on app and return a handle on what it waits for.

    A chord's counter is opened here, before any part can finish: a part that arrived first would otherwise find no counter and take its own arrival for the end of the chord.

    Task

    source
    pub struct Task[A, R] {
      app : App
      name : String
      policy : Policy
      run : async (A, Context) -> R
    }

    A registered body. Sending it writes a message and hands back a handle.

    The argument type is what the body takes; the result type is what it returns, and the handle decodes the stored value back into it. Both go over the wire as JSON, so a task may be sent by one process and run by another.

    Task::name

    source
    fn[A, R] Task::name(self : Task[A, R]) -> String

    The name the message carries.

    Task::policy

    source
    fn[A, R] Task::policy(self : Task[A, R]) -> Policy

    The policy the registration was given.

    Task::send

    source
    async fn[A : 
    ToJson    , R : 
    FromJson    ] Task::send(self : Task[A, R], args : A, eta? : Int64, countdown? : Int64, expires? : Int64, queue? : String, priority? : Int, id? : String) -> Handle[R] raise 
    Refused

    Publish args and return a handle on the run.

    eta is when the message may first run and countdown is how long from now that is; eta wins when both are given, since it is the more specific of the two. expires is the time after which running it is pointless: a worker that finds it expired revokes it instead of running it late.

    Task::sig

    source
    fn[A : 
    ToJson    , R] Task::sig(self : Task[A, R], args : A) -> Sig

    The run of args as a node of a canvas, without publishing it.

    Worker

    source
    pub struct Worker {
      app : App
      queues : Array[String]
      concurrency : Int
      prefetch : Int
      limiters : Map[String, Limiter]
      consumers : Array[
    Task    [Unit]]
      inflight : Int
      stopping : Bool
      done : 
    Cond
      failure : Error?
    }

    The object that runs tasks.

    A worker consumes from queues, runs up to concurrency bodies at once, and settles each message according to the task's policy: a failure within the retry budget goes back on the queue with an eta computed from the backoff, a success advances the chain or the chord the message carries.

    stop is graceful: it stops taking new messages, waits up to grace milliseconds for the ones in flight, then cancels whatever is left.

    Worker::is_stopping

    source
    fn Worker::is_stopping(self : Worker) -> Bool

    Whether stop has been called.

    Worker::prefetch

    source
    fn Worker::prefetch(self : Worker) -> Int

    The most messages this worker holds unsettled at once.

    Worker::queues

    source
    fn Worker::queues(self : Worker) -> Array[String]

    The queues this worker takes from.

    Worker::run

    source
    async fn Worker::run(self : Worker) -> Unit

    Run until stop is called. The call returns once every consumer has ended, which is why a caller runs it in its own task.

    Worker::stop

    source
    async fn Worker::stop(self : Worker, grace? : Int64) -> Unit

    Stop taking new work, give what is running up to grace milliseconds to finish, then cancel the rest.

    chain

    source
    fn chain(sigs : Array[Sig]) -> Sig

    The runs in sigs, in order, each one started when the one before it succeeds.

    A chain of chains is flattened, so chain([a, chain([b, c])]) is a → b → c. Building a chain publishes nothing.

    chord

    source
    fn chord(parts : Array[Sig], callback : Sig) -> Sig

    The runs in parts, then callback once every part has succeeded.

    The callback is sent with the list of part results as its argument, in the order the parts were given. Every part carries the same key and the same list of ids, so the part that finishes last can collect the set by itself.

    A part that fails for good does not hold the chord open: the last part to finish records the callback as failed, so nobody waits on a callback that will never be sent.

    group

    source
    fn group(sigs : Array[Sig]) -> Sig

    Every run in sigs, all at once, waited on together.

    A chain given to a group is not broken up: its head is published and its tail travels with it, so the group waits on the chain's last link. The handle's value is the list of results in the order the sigs were given.

    milliseconds_of

    source
    fn milliseconds_of(moment : 
    Moment    ) -> Int64?

    Milliseconds since the epoch of an instant.

    moment_of

    source
    fn moment_of(at : Int64) -> 
    Moment    ?

    An instant from milliseconds since the epoch.

    policy

    source
    let policy : Policy

    The defaults, documented field by field on [Policy].

    Source Files

    Powered by MoonBit

    Site sourceReport issueModulesSkillsCLIBuild queueStatistics

    © 2026 mooncakes.io