🥮 mooncakes.io

    moonbitstack/moontask/spec does not have a README file

    Backend

    source
    pub(open) trait Backend {
      async fn store(Self, id : String, outcome : Outcome) -> Unit raise Refused
      async fn fetch(Self, id : String) -> Outcome? raise Refused
    }

    Where results wait for a caller.

    fetch returns None both for a run nobody has recorded yet and for a run whose store keeps nothing, which is why a caller that needs the difference asks for a store before it starts.

    Broker

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

    Where messages wait and what settles them.

    consume blocks until a message is available or the call is cancelled, and hands back a Delivery whose receipt the same broker settles later. The four calls are all a worker needs: publish to enqueue, consume to claim, ack when the run settled and the message must not come back, reject when it must — with requeue false it goes to whatever dead letter the broker keeps.

    prefetch bounds how many messages one worker holds unsettled at a time, so a fast producer cannot bury a slow one.

    Counter

    source
    pub(open) trait Counter {
      async fn add(Self, key : String, delta : Int) -> Int raise Refused
      async fn drop(Self, key : String) -> Unit raise Refused
      async fn settle(Self, chord : Chord, id : String, outcome : Outcome) -> ChordState raise Refused
    }

    The counter a chord waits on.

    It is separate from Backend because counting parts and keeping results are different guarantees: a store may keep results for an hour and a counter for a minute, and a chord only needs the counter. add returns the value after the add, so the part that brings the count to zero is the one that fires the callback; it must be atomic, since two parts finishing at once would otherwise both see a non-zero count or both see zero.

    add with a positive delta opens a chord of that many parts before any of them is published; a negative delta is a plain decrement. settle is the call a worker makes once a part will not run again, and it is the one that answers with the parts' results when that part was the last.

    Inspector

    source
    pub(open) trait Inspector {
      async fn queues(Self) -> Array[QueueInfo] raise Refused
      async fn peek(Self, queue : String, limit~ : Int) -> Array[Message] raise Refused
    }

    A broker that can report its queues and show what waits in them.

    A worker only needs Broker, so a broker is not obliged to implement this. A management page is the reason it optionally does: without it the page can show results but not what is still queued.

    Refused

    source
    pub(all) suberror Refused {
      Unknown(name~ : String)
      Malformed(got~ : String)
      Transport(got~ : String)
      Missing(kind~ : String)
      Timeout(limit~ : Int64)
      Failed(got~ : String, retries~ : Int)
      Empty(kind~ : String)
    } derive(Eq, 
    Debug    )

    The one error type of this library.

    A broker, a result store and a chord counter all refuse in these terms, so a caller catches one type no matter which seam failed. got/want/limit are the organisation's words for "what arrived", "what was required" and "the ceiling that was hit".
    impl Show for Refused

    Refused::equal

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

    Refused::not_equal

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

    Refused::output

    source
    fn Refused::output(self : Refused, logger : &Logger) -> Unit

    Refused::to_repr

    source

    Refused::to_string

    source
    fn Refused::to_string(self : Refused) -> String

    Chord

    source
    pub(all) struct Chord {
      key : String
      index : Int
      total : Int
      ids : Array[String]
      callback : Message
    } derive(Eq, 
    Debug    )

    One part's view of the chord it belongs to: which key counts the parts, which part this is, every part's id so the callback can collect their results, and the message to send once the last part is done.

    Every part carries the same ids, so the worker that happens to finish last can collect the whole set without asking a coordinator.
    impl FromJson for Chord
    impl ToJson for Chord

    Chord::equal

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

    Chord::from_json

    source

    Chord::not_equal

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

    Chord::to_json

    source

    Chord::to_repr

    source

    ChordState

    source
    pub(all) enum ChordState {
      Waiting(remaining~ : Int)
      Settled(values~ : Array[
    Json    ], failure~ : String?)
    }

    What a chord's counter answers when one of its parts settles.

    The counter owns the join, and not only the count, because where the parts' results are kept is a property of the store the counter lives in: a counter in memory has the store beside it, and Celery's redis counter stashes each result in the same sorted set it counts, keyed by the part's index.

    Delivery

    source
    pub(all) struct Delivery {
      message : Message
      queue : String
      receipt : String
    } derive(Eq, 
    Debug    )

    What a broker hands a worker: one message, the queue it came from, and the receipt the broker needs back to settle it.

    receipt is opaque to this library. An in-memory broker puts a sequence there, an ordered-log broker the id of the entry; whatever it is, the worker only ever hands it back to the same broker.

    Delivery::equal

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

    Delivery::not_equal

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

    Delivery::to_repr

    source

    Message

    source
    pub(all) struct Message {
      id : String
      name : String
      args : 
    Json
      queue : String
      priority : Int
      eta : Int64?
      expires : Int64?
      retries : Int
      chain : Array[Message]
      chord : Chord?
      single_arg : Bool
      immutable : Bool
    } derive(Eq, 
    Debug    )

    One unit of work, as it travels through a broker.

    The field set is the published protocol's, so a broker can be inspected with the tools of that ecosystem: id names the run, name the registered task, args the argument value, retries how many times it has already failed, eta when it may first run and expires when it is too late to run it. queue and priority are the routing a broker uses to pick the next one.

    The two fields below retries are how a chain and a chord travel: a message carries the rest of its own chain with it, so no coordinator is needed to advance it.
    impl FromJson for Message
    impl ToJson for Message

    Message::equal

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

    Message::from_celery

    source
    fn Message::from_celery(json : 
    Json    , queue~ : String) -> Message raise Refused

    Read a Celery envelope back into a message.

    queue is what the broker knows and the envelope names, so a message read off a list is told which list it came from.

    Message::from_json

    source

    Message::is_due

    source
    fn Message::is_due(self : Message, now : Int64) -> Bool

    Whether the message may run at now: its eta, when set, has come.

    Message::is_expired

    source
    fn Message::is_expired(self : Message, now : Int64) -> Bool

    Whether the message is past the time it may still run at now.

    Message::new

    source
    fn Message::new(id : String, name : String, args : 
    Json    ) -> Message

    A message for name with argument args, queued on default with no delay and no retries behind it.

    Message::not_equal

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

    Message::parse

    source
    fn Message::parse(text : StringView) -> Message raise Refused

    A message read from its JSON text. The FromJson impl is the same reading; this is the entry a broker uses, so a bad payload is refused in this library's terms rather than in the JSON module's.

    Message::text

    source
    fn Message::text(self : Message) -> String

    The JSON text of a message, which is what a broker stores.

    Message::to_celery

    source

    The message as Celery's envelope, ready to be pushed on a queue.

    Message::to_json

    source

    Message::to_repr

    source

    Outcome

    source
    pub(all) struct Outcome {
      state : State
      result : 
    Json    ?
      error : String?
      retries : Int
      at : Int64
    } derive(Eq, 
    Debug    )

    What a worker records when a run ends: the state it ended in, the value or the error, and when.

    result is the decoded return value of the body and error is the message of the raised error; exactly one of them is set for a final state.
    impl FromJson for Outcome
    impl ToJson for Outcome

    Outcome::equal

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

    Outcome::from_celery

    source
    fn Outcome::from_celery(json : 
    Json    ) -> Outcome raise Refused

    An outcome read back from Celery's result backend.

    Outcome::from_json

    source

    Outcome::not_equal

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

    Outcome::to_celery

    source
    fn Outcome::to_celery(self : Outcome, id : String) -> 
    Json

    An outcome as Celery's result backend writes it, so a Celery client waiting on AsyncResult reads what a worker here finished.

    Outcome::to_json

    source

    Outcome::to_repr

    source

    QueueInfo

    source
    pub(all) struct QueueInfo {
      name : String
      ready : Int64
      delayed : Int64
      unacked : Int64
    } derive(Eq, 
    Debug    )

    What a broker reports about one of its queues.

    unacked counts messages a worker holds and has not settled, which is the number a dashboard shows next to the ready count so a stuck consumer is visible.

    QueueInfo::equal

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

    QueueInfo::not_equal

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

    QueueInfo::to_repr

    source

    State

    source
    pub(all) enum State {
      Pending
      Received
      Started
      Retrying
      Succeeded
      Failed
      Revoked
    } derive(Eq, 
    Debug    )

    Where a task is in its life. The set is the one a caller has to be able to tell apart, and it is the one published result stores agree on:

    • Pending — accepted by a broker, not yet picked up.
    • Received — a worker took it off the queue.
    • Started — the body is running.
    • Retrying — the body failed and the task is queued again.
    • Succeeded — the body returned; the result is stored.
    • Failed — the body raised on its last attempt; the error is stored.
    • Revoked — cancelled before or during the run; it will not be sent again.
    impl Show for State

    State::code

    source
    fn State::code(self : State) -> String

    The label a result store writes for a state. These are the labels the published protocol uses, so a result written by another producer reads back as the same state.

    State::equal

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

    State::is_final

    source
    fn State::is_final(self : State) -> Bool

    Whether no further state will follow. A caller waiting for a result stops here.

    State::not_equal

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

    State::of

    source
    fn State::of(code : StringView) -> State raise Refused

    The state a store's label names. An unknown label is refused rather than read as Pending: a store that grew a state this library does not know is a store this library must not guess about.

    State::output

    source
    fn State::output(self : State, logger : &Logger) -> Unit

    State::to_repr

    source

    State::to_string

    source
    fn State::to_string(self : State) -> String

    celery_group_key

    source
    fn celery_group_key(group : String, part : String) -> String

    The keys a chord's join lives under in Celery's redis backend: the results as they arrive (.j, a sorted set scored by the part's index), the adjustment to the size (.t), and the size itself (.s).

    celery_result_key

    source
    fn celery_result_key(id : String) -> String

    The key Celery's result backend stores a run under.

    outcome_failed

    source
    fn outcome_failed(error : String, retries : Int, at : Int64) -> Outcome

    The outcome of a run that raised error on attempt retries.

    outcome_succeeded

    source
    fn outcome_succeeded(value : 
    Json    , retries : Int, at : Int64) -> Outcome

    The outcome of a run that returned value.

    Source Files

    Powered by MoonBit

    Site sourceReport issueModulesSkillsCLIBuild queueStatistics

    © 2026 mooncakes.io