///|
/// Single-writer concurrency over `moonbitlang/async`: private state, a
/// typed mailbox, one serial loop. A Fuwaroid is a light entity that
/// floats suspended until a message arrives, settles it in one
/// synchronous step, and drifts back to sleep — no OS threads, no locks,
/// no blocking (see README for the name).
///
/// The execution model:
/// - Handlers (`on_cmd` / `on_query`) are SYNCHRONOUS by signature, so the
/// loop never awaits between callback entry and return; callback
/// invocations are serialized and no other message can interleave them.
/// This does not make a generic `State` immutable or prevent aliases.
/// Callers must treat the state as actor-owned and must not retain or
/// mutate it from spawned work or replies.
/// - Async work (IO, subprocesses, timers) never runs inside a handler:
/// spawn it on the host group from the handler and report the outcome
/// back with `tell`. The loop is the only place where a handler return
/// value advances its state; the ownership rule is a caller contract,
/// not a compiler-enforced property of the generic API.
/// - Between complete messages (handler returned, state committed) the
/// loop yields the scheduler every `fuwaroid_yield_batch` messages, so
/// an instance that keeps its own mailbox non-empty cannot monopolize
/// the cooperative runtime: other instances, timers and cancellation
/// all get scheduled. A handler is never suspended midway.
/// - Every stop — graceful closed-and-empty drain, host cancellation
/// observed at a yield/`get` boundary, or any other terminal mailbox
/// error — goes through ONE cleanup path (`fuwaroid_cleanup`): admission
/// is closed, stranded asks are failed with the stop marker, the
/// termination reason is recorded, and only then does the loop task
/// terminate (returning normally after a graceful drain, re-raising the
/// original error for an ordinary failure, and terminating CANCELLED —
/// observable as `@async.TaskCancelled` to `Task::wait` — after a host
/// cancellation, whose signal in async 0.22.x is distinct from `Error`).
/// - `join` waits for the loop task itself to terminate and returns the
/// recorded stop reason; it never waits for or cancels background work
/// a handler spawned on the host group.
/// - The loop is spawned `no_wait` on the HOST's long-lived task group.
/// There is no actor system, no registry, no supervision tree: the host
/// owns the lifetime (structured concurrency). Handlers cannot raise
/// (their types say so); failure is an ordinary message or terminal
/// state — never a restart that resets state, which is what ledgers of
/// in-flight work require.
pub struct Fuwaroid[Cmd, Query, Reply] {
priv mailbox : @aqueue.Queue[Envelope[Cmd, Query, Reply]]
/// Every handle shares its loop's stop record and diagnostics.
priv stop : LoopStop
}
///|
/// Internal mailbox vocabulary. Public only because it appears in the
/// (private) field type of the public `Fuwaroid` handle — visibility
/// rule: consumers never construct these; the mailbox itself is private.
pub enum Envelope[Cmd, Query, Reply] {
Tell(Cmd)
Ask(Query, @aqueue.Queue[Reply])
}
///|
/// What a handler may touch: the host group (to spawn async work) and
/// this Fuwaroid's own address (to report work outcomes back). Never the
/// raw mailbox, never another Fuwaroid's state.
pub struct Ctx[Cmd, Query, Reply] {
/// Host task group — spawn async side work here; it outlives the
/// message that started it but not the host.
group : @async.TaskGroup[Unit]
/// This Fuwaroid's address. Use `tell` only. A handler is a SYNCHRONOUS
/// function, so calling the async `ask` from inside a handler is already
/// a compile error (E4149): the classic self-ask scenario — the loop busy
/// serving the current message until the ask timeout fires — is excluded
/// at the type level, not merely discouraged.
address : Fuwaroid[Cmd, Query, Reply]
}
///|
/// Why a `tell` was not accepted.
pub(all) enum SendRefusal {
MailboxClosed
MailboxFull
} derive(Eq, Debug)
///|
pub extend SendRefusal with Eq::{equal, not_equal}
///|
pub extend SendRefusal with @moonbitlang/core/debug.Debug::{to_repr}
///|
/// Why an `ask` did not produce a reply.
pub(all) enum AskFailure {
/// The request never entered the mailbox.
NotDelivered(SendRefusal)
/// No reply within the caller's timeout. The request is NOT withdrawn —
/// the Fuwaroid may still process it later; the abandoned reply is
/// dropped.
TimedOut
/// The loop stopped abnormally (host cancellation) while this request
/// was queued; its reply queue was closed by the drain.
Stopped
} derive(Eq, Debug)
///|
pub extend AskFailure with Eq::{equal, not_equal}
///|
pub extend AskFailure with @moonbitlang/core/debug.Debug::{to_repr}
///|
/// Why the loop task terminated. Recorded by the unified stop path
/// (`fuwaroid_cleanup`) before the loop task terminates. The three
/// constructors distinguish the terminal dispositions the async 0.22.x
/// runtime separates at the language level:
///
/// - `Graceful`: `close` was requested and the closed-and-empty drain
/// completed.
/// - `Cancelled`: the host (task group) cancelled the loop. In
/// `moonbitlang/async` 0.22.x cancellation is a distinct runtime signal,
/// NOT an `Error` value, so it is reported as its own constructor —
/// the loop never fabricates an error to stand in for it.
/// - `Failed(Error)`: an ordinary terminal error (e.g. a foreign mailbox
/// error); the ORIGINAL error identity is kept, never a compressed copy.
pub(all) enum StopReason {
Graceful
Cancelled
Failed(Error)
} derive(Debug)
///|
pub extend StopReason with @moonbitlang/core/debug.Debug::{to_repr}
///|
/// Internal marker closed into reply queues of asks stranded by an
/// abnormal loop stop, so their waiters fail fast instead of timing out.
pub(all) suberror StoppedError {
StoppedError
}
///|
/// Internal per-instance stop record, stored on the `Fuwaroid` handle.
/// The loop writes the termination reason and any secondary cleanup
/// failure here; `join` (and a later snapshot) read the recorded outcome
/// after — or while — waiting on the loop task.
priv struct LoopStop {
/// Why the loop terminated (written once, by `fuwaroid_cleanup`).
mut reason : StopReason?
/// Secondary failure evidence: if the cleanup itself fails (e.g. a
/// foreign drain error) while an original stop cause exists, the
/// original cause still wins propagation and the secondary failure is
/// preserved here instead of being swallowed.
mut cleanup_error : Error?
/// The loop task handle, obtained via `TaskGroup::spawn`. Termination
/// waiters wait on this task; the recorded `reason` is read from this
/// cell after termination.
mut task : @async.Task[Unit]?
/// Per-instance diagnostics and lifecycle phase: the
/// admission side (`tell`/`ask`/`close`), the serving loop
/// (`processed`) and the unified cleanup (`abandoned`, the final
/// `Stopped` phase) all write it; `Fuwaroid::snapshot` reads it.
diag : Diag
}
///|
/// Mailbox configuration for `Fuwaroid::spawn`:
/// this library owns the mailbox vocabulary, `@aqueue.Kind` is no longer
/// part of any public signature.
///
/// - `Unbounded`: admits every message; the queue grows without bound.
/// - `Bounded(n)`: at most `n` messages are buffered; a send that finds
/// the mailbox full is refused SYNCHRONOUSLY with `MailboxFull` — it
/// never waits for capacity. `n` must be a positive integer: zero
/// (rendezvous) and negative capacities are a configuration error and
/// fail-fast (`abort`) at spawn time, they are not clamped or remapped.
///
/// Measured admission-window semantics: while the loop
/// is parked in `queue.get()` — its steady state while idle — the first
/// send is handed to that parked reader point-to-point, bypassing the
/// buffer. One uninterrupted producer turn can therefore admit up to
/// `capacity + 1` messages in flight (1 handoff slot + `capacity`
/// buffered); the next send is the one refused with `MailboxFull`.
pub(all) enum Mailbox {
Unbounded
Bounded(Int)
} derive(Eq, Debug)
///|
pub extend Mailbox with Eq::{equal, not_equal}
///|
pub extend Mailbox with @moonbitlang/core/debug.Debug::{to_repr}
///|
/// Pure validation of a `Bounded` capacity: returns the failure
/// message for `spawn` to abort with, `None` if valid. Kept a pure,
/// package-internal function so the white-box tests can pin the acceptance
/// boundary directly.
fn mailbox_capacity_error(n : Int) -> String? {
if n <= 0 {
Some(
"Bounded mailbox capacity must be a positive integer (zero/negative capacities are rejected fail-fast, not clamped), got \{n}",
)
} else {
None
}
}
///|
fn LoopStop::fresh() -> LoopStop {
{ reason: None, cleanup_error: None, task: None, diag: Diag::fresh(), }
}
///|
/// Spawn a Fuwaroid on the host's long-lived task group. Commands and
/// queries are SEPARATE types: `on_cmd` folds a command into the state,
/// `on_query` folds a query and produces its reply — neither handler has
/// unreachable arms for the other flow. Both run on the single loop,
/// serially, in mailbox FIFO order, and may not raise. Use
/// `Fuwaroid::spawn_fold` when there is no query flow at all.
///
/// `mailbox` configures admission with this library's own `Mailbox`
/// vocabulary (see its documentation for bounded refusal and the measured
/// capacity+1 admission window); an invalid `Bounded` capacity aborts
/// synchronously here — spawn stays a non-raising API.
pub fn[State, Cmd, Query, Reply] Fuwaroid::spawn(
group~ : @async.TaskGroup[Unit],
init~ : State,
on_cmd~ : (Ctx[Cmd, Query, Reply], State, Cmd) -> State,
on_query~ : (Ctx[Cmd, Query, Reply], State, Query) -> (State, Reply),
mailbox? : Mailbox = Mailbox::Unbounded,
) -> Fuwaroid[Cmd, Query, Reply] {
let kind : @aqueue.Kind = match mailbox {
Unbounded => @aqueue.Unbounded
Bounded(n) =>
match mailbox_capacity_error(n) {
Some(message) =>
abort("fuwaroid: invalid mailbox configuration: \{message}")
None => @aqueue.Blocking(n)
}
}
let queue : @aqueue.Queue[Envelope[Cmd, Query, Reply]] = @aqueue.Queue(kind~)
let stop : LoopStop = LoopStop::fresh()
let fuwaroid : Fuwaroid[Cmd, Query, Reply] = { mailbox: queue, stop, }
let ctx : Ctx[Cmd, Query, Reply] = { group, address: fuwaroid, }
// TaskGroup::spawn (not spawn_bg) so the loop task handle exists for
// termination waiters (`join`); no_wait stays true: the host group does
// not wait for the loop, it cancels it when the scope ends.
// allow_failure keeps its default (false): a cleanup-path bug that lets
// a non-cancellation error escape still fails the host group.
// spawn and task installation have no suspension between them; no
// caller can observe this bootstrap-only empty task slot.
let task = group.spawn(no_wait=true, () => {
fuwaroid_loop(ctx, init, on_cmd, on_query, queue, stop~)
})
stop.task = Some(task)
fuwaroid
}
///|
/// `Fuwaroid::spawn` without a query flow: the state simply folds over
/// the command stream. The handle is `Fuwaroid[Cmd, Unit, Unit]`:
/// `ask((), timeout_ms=...)` degenerates to a served-receipt
/// (`Ok(())`) through a no-op query handler — the barrier is NOT a
/// command, it never reaches `on_cmd`; its only value is FIFO proof: the
/// reply certifies every message enqueued before it was already served.
/// (The legacy `ask(cmd)` barrier misuse is now a compile error: the
/// query type is `Unit`.)
pub fn[State, Cmd] Fuwaroid::spawn_fold(
group~ : @async.TaskGroup[Unit],
init~ : State,
on_cmd~ : (Ctx[Cmd, Unit, Unit], State, Cmd) -> State,
mailbox? : Mailbox = Mailbox::Unbounded,
) -> Fuwaroid[Cmd, Unit, Unit] {
Fuwaroid::spawn(
group~,
init~,
on_cmd~,
on_query=(_, state, _) => (state, ()),
mailbox~,
)
}
///|
/// Complete messages processed between two scheduler yields. A batch of 1
/// maximizes fairness at the cost of a scheduling round trip per message;
/// larger batches amortize the yield but delay other ready tasks and
/// cancellation observation. 32 is a starting point — retune it with the
/// benchmark harness, not by intuition.
let fuwaroid_yield_batch : Int = 32
///|
/// The serial drain, wrapped in the loop task's cancellation boundary.
/// In `moonbitlang/async` 0.22.x cancellation is a distinct runtime
/// signal: ordinary `catch` only sees `Error` values and can never
/// capture it. The structure is therefore:
///
/// - The REAL serving loop (`fuwaroid_serve`) runs inside a cancellable
/// callback under `@async.handle_cancellation`. It handles the two
/// error-shaped dispositions itself — the graceful closed-and-empty
/// drain (`QueueAlreadyClosed`) and ordinary terminal mailbox errors —
/// by running the unified cleanup (`fuwaroid_cleanup`) and re-raising
/// the original cause for a `Failed` stop.
/// - The outermost boundary distinguishes the three terminal
/// dispositions: `Some(())` = the serving loop completed (graceful,
/// reason already recorded); `Err(original)` = an ordinary error
/// escaped (the recorded `Failed` cause re-raised by cleanup —
/// propagation keeps the ORIGINAL identity); `None` = the host
/// cancelled the loop while it was serving.
/// - In the cancellation branch the task is in the sticky cancelled
/// state. The unified cleanup still runs — under
/// `@async.protect_from_cancel`, so it completes in full — and records
/// `Cancelled`. The original cancellation is then re-propagated by
/// entering one unprotected cancellation point (`@async.pause()`),
/// which raises the still-pending cancellation signal (see `pause`:
/// a cancelled, unshielded task raises immediately). The loop task
/// therefore terminates CANCELLED — never as a normal completion —
/// and an external `Task::wait` observer sees `@async.TaskCancelled`.
///
/// Cancellation outranks draining: the loop stops at the next
/// observable cancellation boundary — the batch yield after the message
/// currently being served, or the next blocking `get` — instead of
/// finishing the whole backlog. A message that `queue.get` already
/// delivered when cancellation lands is still served: `try_get` does not
/// observe cancellation, so one accepted message is never dropped
/// mid-batch. Between complete messages there is still no suspension
/// point: handler entry → handler return → state commit is atomic to the
/// cooperative runtime.
async fn[State, Cmd, Query, Reply] fuwaroid_loop(
ctx : Ctx[Cmd, Query, Reply],
init : State,
on_cmd : (Ctx[Cmd, Query, Reply], State, Cmd) -> State,
on_query : (Ctx[Cmd, Query, Reply], State, Query) -> (State, Reply),
queue : @aqueue.Queue[Envelope[Cmd, Query, Reply]],
stop~ : LoopStop,
) -> Unit {
let outcome : Result[Unit?, Error] = Ok(
@async.handle_cancellation(() => {
fuwaroid_serve(ctx, init, on_cmd, on_query, queue, stop~)
}),
) catch {
// Only ordinary errors arrive here: `catch` cannot capture the
// cancellation signal, and `handle_cancellation` reports cancellation
// as `None` instead.
error => Err(error)
}
match outcome {
// The serving loop returned after running its unified cleanup;
// a `Failed` cause was re-raised out of it, so reaching here
// normally means the graceful drain completed.
Ok(Some(_)) => ()
// Host cancellation observed inside the serving loop. Cleanup first
// (protected: cancellation is sticky), then let the original
// cancellation continue propagating so this task ends cancelled.
Ok(None) => {
fuwaroid_cleanup(queue, Cancelled, stop)
// Unprotected cancellation point: the task is still in the
// cancelled state, so this raises the pending cancellation signal
// and the coroutine terminates cancelled (never as normal Done).
@async.pause()
}
// Ordinary error escaped the serving loop: the original cause
// re-raised by its cleanup (a graceful stop's secondary cleanup
// failure arrives here the same way). The recorded reason already
// holds it; propagation keeps the ORIGINAL identity.
Err(original) => raise original
}
}
///|
/// The message loop proper: suspend on `queue.get`, serve one envelope
/// per iteration with a synchronous handler, yield the scheduler every
/// `fuwaroid_yield_batch` complete messages. The loop exits on the first
/// ordinary terminal error; `queue.get`'s ordinary queue errors are
/// caught into `cause` (notably `QueueAlreadyClosed` after a graceful
/// `close`), while cancellation is NOT catchable and simply unwinds this
/// callback to the `handle_cancellation` boundary in `fuwaroid_loop`.
/// On any ordinary exit the unified cleanup runs with the mapped reason
/// and re-raises the original cause for a `Failed` stop.
async fn[State, Cmd, Query, Reply] fuwaroid_serve(
ctx : Ctx[Cmd, Query, Reply],
init : State,
on_cmd : (Ctx[Cmd, Query, Reply], State, Cmd) -> State,
on_query : (Ctx[Cmd, Query, Reply], State, Query) -> (State, Reply),
queue : @aqueue.Queue[Envelope[Cmd, Query, Reply]],
stop~ : LoopStop,
) -> Unit {
let mut state = init
let mut served_in_batch = 0
let mut cause : Error? = None
while cause is None {
let next : Result[Envelope[Cmd, Query, Reply], Error] = Ok(queue.get()) catch {
// Ordinary queue errors only (closed-and-empty after `close`,
// foreign close errors). Cancellation is a separate signal and
// bypasses this handler entirely.
error => Err(error)
}
match next {
Ok(Tell(cmd)) => {
state = on_cmd(ctx, state, cmd)
// Counting point: the handler returned, the state is committed.
stop.diag.processed = stop.diag.processed + 1L
served_in_batch += 1
}
Ok(Ask(query, reply)) => {
let pair = on_query(ctx, state, query)
state = pair.0
let _ = reply.try_put(pair.1) catch { _ => false }
// Counting point: the handler returned, the state is committed
// (the reply handoff is best-effort and not part of the criterion).
stop.diag.processed = stop.diag.processed + 1L
served_in_batch += 1
}
Err(error) => cause = Some(error)
}
// Yield point BETWEEN complete messages only: the handler above ran
// entry-to-return with no suspension inside. `pause` re-queues this
// task at the tail of the ready queue and delivers the host's
// cancellation signal here if it was requested since the last
// boundary, so a cancelled loop cannot drain an unbounded backlog.
// `pause` cannot raise an ordinary error; cancellation unwinds to the
// `handle_cancellation` boundary on its own.
if cause is None && served_in_batch >= fuwaroid_yield_batch {
served_in_batch = 0
@async.pause()
}
}
let reason : StopReason = match cause {
// Graceful: closed-and-empty drain completed (or the loop exited
// without a cause — same disposition).
None | Some(@aqueue.QueueAlreadyClosed) => Graceful
// Ordinary failure: keep the original error identity.
Some(original) => Failed(original)
}
fuwaroid_cleanup(queue, reason, stop)
}
///|
/// THE single stop/cleanup path for the loop. Runs exactly
/// once per stop, for every disposition (`StopReason`):
///
/// 1. Close admission (`queue.close`): idempotent — safe when the mailbox
/// was already closed (graceful `close`, or a foreign close). From here
/// on the handle refuses new messages.
/// 2. Fail every accepted-but-unserved ask via `fail_leftover_asks` so
/// its waiter gets `AskFailure::Stopped` immediately. The drain is
/// shielded with `@async.protect_from_cancel`, so it also completes
/// when called from the sticky cancelled state (the cancellation
/// branch of `fuwaroid_loop`).
/// 3. Record the termination reason on `stop` and keep any secondary
/// cleanup failure as evidence on `stop.cleanup_error` instead of
/// swallowing it.
/// 4. Propagate, with the PRIMARY cause winning:
/// - `Graceful` returns normally unless the cleanup itself failed.
/// - `Failed(original)` re-raises the ORIGINAL error — a secondary
/// cleanup failure never replaces it.
/// - `Cancelled` returns normally into the still-cancelled caller,
/// which re-propagates the pending cancellation signal; a secondary
/// cleanup failure is recorded as evidence only, so the actor is
/// never disguised as an ordinary failure.
async fn[Cmd, Query, Reply] fuwaroid_cleanup(
queue : @aqueue.Queue[Envelope[Cmd, Query, Reply]],
reason : StopReason,
stop : LoopStop,
) -> Unit {
// The default close error is intentional: it only decides what NEW
// attempts to touch the mailbox see; the recorded reason is retained
// below and stranded asks get `StoppedError` in step 2.
queue.close()
stop.diag.transition(Closing)
// Counting point: every envelope the stranded-ask drain removes
// was accepted but is provably unserved — commands and queries alike
// are counted as abandoned. Admission is closed before any yield.
// Shield the whole drain so cancellation cannot strand remaining asks.
let drained : Result[Unit, Error] = Ok(
@async.protect_from_cancel(() => fail_leftover_asks(queue, diag=stop.diag)),
) catch {
error => Err(error)
}
let secondary : Error? = match drained {
Ok(_) => None
Err(error) => Some(error)
}
stop.reason = Some(reason)
stop.cleanup_error = secondary
// Counting point: the unified cleanup is complete — the phase moves to
// Stopped carrying the recorded reason (forward-only); `abandoned` was
// fixed by the drain above.
stop.diag.transition(Lifecycle::Stopped(reason))
match reason {
// Cancellation is re-propagated by the caller (the cancellation
// branch of `fuwaroid_loop`); secondary evidence stays recorded.
Cancelled => ()
Graceful =>
match secondary {
Some(error) => raise error
None => ()
}
// Ordinary failure: the original error identity wins propagation.
Failed(original) => raise original
}
}
///|
/// After an abnormal stop (host cancellation), asks may still sit in the
/// mailbox; their callers wait on reply queues that will never be filled.
/// Close each so every waiter fails fast with `StoppedError`. On a
/// graceful `close` the mailbox is already empty-and-closed here, and the
/// first `try_get` raises `QueueAlreadyClosed` — a no-op. Any other queue
/// error is propagated so cleanup cannot hide a failure.
///
/// Every envelope removed here (commands AND queries) is counted as
/// `abandoned` on the given diagnostics cell, when one is supplied: the
/// drain is the exact truncation point between "accepted and served" and
/// "accepted but unserved". Direct white-box callers that pass
/// no cell only get the draining behavior.
async fn[Cmd, Query, Reply] fail_leftover_asks(
queue : @aqueue.Queue[Envelope[Cmd, Query, Reply]],
diag? : Diag,
) -> Unit {
let mut draining = true
let mut drained_in_batch = 0
while draining {
let next : Result[Envelope[Cmd, Query, Reply]?, Error] = Ok(queue.try_get()) catch {
error => Err(error)
}
match next {
Ok(Some(Ask(_, reply))) => {
match diag {
Some(d) => d.abandoned = d.abandoned + 1L
None => ()
}
reply.close(error=StoppedError)
}
Ok(Some(Tell(_))) =>
match diag {
Some(d) => d.abandoned = d.abandoned + 1L
None => ()
}
Ok(None) => draining = false
Err(@aqueue.QueueAlreadyClosed) => draining = false
Err(error) => raise error
}
if draining {
drained_in_batch += 1
if drained_in_batch >= fuwaroid_yield_batch {
drained_in_batch = 0
@async.pause()
}
}
}
}
///|
/// Fire-and-forget send; returns immediately after the mailbox accept
/// decision. `Ok` only means the command was enqueued. Order of
/// processing is mailbox FIFO; the interleaving of concurrent `tell`s is
/// the caller's scheduling. Every outcome is counted on the instance's
/// diagnostics: an accepted admission bumps `accepted`, a full
/// refusal `rejected_full`, a closed refusal `rejected_closed`.
pub fn[Cmd, Query, Reply] Fuwaroid::tell(
self : Fuwaroid[Cmd, Query, Reply],
cmd : Cmd,
) -> Result[Unit, SendRefusal] {
let delivered : Result[Bool, Error] = Ok(
self.mailbox.try_put(Envelope::Tell(cmd)),
) catch {
error => Err(error)
}
match delivered {
Ok(true) => {
// Counting point: admission accepted.
self.stop.diag.accepted += 1L
Ok(())
}
Ok(false) => {
// Counting point: admission refused, mailbox full.
self.stop.diag.rejected_full += 1L
Err(SendRefusal::MailboxFull)
}
Err(_) => {
// Counting point: admission refused, mailbox closed.
self.stop.diag.rejected_closed += 1L
Err(SendRefusal::MailboxClosed)
}
}
}
///|
/// Pure classification of the errors `ask` can see on its reply queue:
/// mapped failures return `Some`; `None` means "not ours" — the caller
/// must propagate (notably host cancellation of the asker).
fn classify_ask_error(error : Error) -> AskFailure? {
match error {
StoppedError => Some(AskFailure::Stopped)
@async.TimeoutError => Some(AskFailure::TimedOut)
_ => None
}
}
///|
/// Send one query and wait for the handler's reply, bounded by
/// `timeout_ms`. Calling `ask` from inside a handler of the same Fuwaroid
/// is a compile error (E4149): handlers are synchronous functions and
/// `ask` is async, so the self-ask scenario — the loop busy serving the
/// current message until the timeout fires — is excluded by the type
/// system rather than merely discouraged. Cancellation of the CALLER
/// propagates (it is not converted into a failure value).
pub async fn[Cmd, Query, Reply] Fuwaroid::ask(
self : Fuwaroid[Cmd, Query, Reply],
query : Query,
timeout_ms~ : Int,
) -> Result[Reply, AskFailure] {
let reply : @aqueue.Queue[Reply] = @aqueue.Queue(kind=@aqueue.Unbounded)
let delivered : Result[Bool, Error] = Ok(
self.mailbox.try_put(Envelope::Ask(query, reply)),
) catch {
error => Err(error)
}
match delivered {
Ok(true) =>
// Counting point: admission accepted (queries count too).
self.stop.diag.accepted += 1L
Ok(false) => {
// Counting point: admission refused, mailbox full.
self.stop.diag.rejected_full += 1L
return Err(AskFailure::NotDelivered(SendRefusal::MailboxFull))
}
Err(_) => {
// Counting point: admission refused, mailbox closed.
self.stop.diag.rejected_closed += 1L
return Err(AskFailure::NotDelivered(SendRefusal::MailboxClosed))
}
}
let outcome : Result[Reply?, Error] = Ok(
@async.with_timeout_opt(timeout_ms, () => reply.get()),
) catch {
error => Err(error)
}
match outcome {
Ok(Some(answer)) => Ok(answer)
Ok(None) => Err(AskFailure::TimedOut)
Err(error) =>
match classify_ask_error(error) {
Some(failure) => Err(failure)
None => raise error
}
}
}
///|
/// Graceful mailbox stop: the mailbox accepts no new messages, everything
/// already queued (commands AND queries) is still processed in FIFO order,
/// then the loop exits. A reply to an `ask` issued just before `close` proves
/// every earlier message was already served (FIFO). `close` does not cancel
/// or await work that a handler already spawned on the host group; such work
/// may observe `MailboxClosed` when it reports back. The host group owns the
/// lifetime and waits for all of its children when it terminates.
///
/// Closing twice is idempotent (the second close is a no-op on the
/// already-closed queue). Returning does NOT mean the drain completed:
/// the backlog is drained by the loop task, which yields between batches
/// while doing so. To wait for the loop task to actually terminate and
/// learn why, use `join`. The lifecycle phase moves to `Closing` here
/// (forward-only: a late `close` cannot un-stop an already `Stopped`
/// instance) and to `Stopped(reason)` when the unified cleanup completes.
pub fn[Cmd, Query, Reply] Fuwaroid::close(
self : Fuwaroid[Cmd, Query, Reply],
) -> Unit {
// Counting point: the close REQUEST moves Running → Closing.
self.stop.diag.transition(Closing)
self.mailbox.close()
}
///|
/// Wait for this Fuwaroid's loop task to terminate and return the
/// recorded stop reason.
///
/// - Waiting uses `Task::wait`, so it is multi-waiter safe and returns
/// immediately once the loop has terminated. Only the loop task is
/// awaited: background work a handler spawned on the host group is
/// neither waited for nor cancelled by `join`.
/// - Waiter cancellation (A): the waiter's own cancellation is the
/// runtime's cancellation signal, NOT an `Error` — `catch` cannot
/// capture it, so it propagates out of `join` unchanged and is never
/// converted into a `StopReason`. The pre-check below makes even an
/// ALREADY-cancelled waiter observe itself when the loop already
/// terminated: `Task::wait` has a synchronous fast path for terminated
/// targets that performs no cancellation check, so the pre-check raises
/// the pending signal first via `@async.pause()`.
/// - Cancelled loop task (B): `Task::wait` reports a cancelled TARGET as
/// the ordinary `@async.TaskCancelled` error. That value describes the
/// waited task, not this actor's stop, so it is never stored or
/// returned as the reason: the loop's own cancellation branch already
/// recorded `Cancelled` during its protected cleanup, and `join`
/// returns exactly that.
/// - Failed loop task (C): the target's ordinary error maps to the
/// recorded `Failed(original)` — the original identity wins; a
/// recorded-but-unexplained failure falls back to the secondary
/// cleanup evidence, then to the raw task error (only reachable
/// through a cleanup-path bug).
/// - Graceful completion (D): the task ending `Done` returns the
/// recorded reason, `Graceful`.
/// - A missing internal task is an invariant violation and aborts.
///
/// Note: when the loop task fails with a NON-cancellation error, the
/// host task group's fail-fast still applies independently of `join` —
/// a joiner living in the same group is cancelled with the group before
/// it can observe `Failed`; a joiner in a different scope observes it.
pub async fn[Cmd, Query, Reply] Fuwaroid::join(
self : Fuwaroid[Cmd, Query, Reply],
) -> StopReason {
// Task::wait skips cancellation checks for already completed targets.
// Surface the waiter's own pending cancellation before taking that
// fast path: `pause` raises the cancellation signal, which propagates
// out of `join` unchanged (it is not an `Error` and cannot be caught
// into a `StopReason`).
if @async.is_being_cancelled() {
@async.pause()
}
let cell = self.stop
let task = match cell.task {
Some(task) => task
None => abort("fuwaroid: join before loop task initialization")
}
let outcome : Result[Unit, Error] = Ok(task.wait()) catch {
// Only errors of the TARGET task arrive here: `catch` cannot capture
// the waiter's own cancellation signal, so no waiter-cancellation
// branch exists anymore.
error => Err(error)
}
match outcome {
Ok(_) =>
match cell.reason {
Some(reason) => reason
None =>
abort("fuwaroid: loop completed without recording its stop reason")
}
Err(error) =>
match error {
// The loop TASK was cancelled; its cancellation branch completed
// the unified cleanup before re-propagating, so `Cancelled` is
// the recorded outcome. `TaskCancelled` itself is never used as
// the reason — it describes the waited task, not this actor.
@async.TaskCancelled => Cancelled
other =>
match cell.reason {
Some(Failed(_) as reason) => reason
Some(Graceful) | Some(Cancelled) | None =>
match cell.cleanup_error {
Some(secondary) => Failed(secondary)
None => Failed(other)
}
}
}
}
}