Skip to main content

core.async — Asynchronous execution

Backend checks: interpreter HTTP checks cover binary responses, header deadlines, cancellation before reading, and an overall deadline across slowly arriving headers. The same complete HTTP path has not yet been validated natively. Native checks cover eager handler success, typed errors and caught panics; they do not establish general cancellation or resource-cleanup correctness. See eager async evaluation and guard lifetime.

Full async toolkit: Future protocol, executors, channels, async streams, timers, structured concurrency (nursery), racing (select), circuit breakers, retry policies, parallel helpers.

Module status​

Each core.async.* module carries an explicit conformance status so you know what you can rely on today versus what is still in flight. The status is the truth-table over the module's API surface as exercised by core-tests/async/<module>/ under both verum test --interp (Tier 0 VBC interpreter) and verum test --aot (Tier 1 LLVM AOT, --test-threads 1).

StatusMeaning
completeEverything stable requires, plus the coverage bar the conformance inventory sets for its top mark: algebraic laws pinned by property tests, cross-stdlib integration verified, and the module's audit findings landed or routed. Completion also requires end-to-end checks in both the interpreter and native AOT backend; stable alone does not imply this.
stableThe covered suite passes under the interpreter. Native coverage must be stated separately; this status does not establish backend parity. See the shared status convention.
partialOnly a subset of the API has measured coverage. Consult the backend-specific limitations before using it; a passing or unignored test alone is not a production-safety guarantee.
regression-onlyModule is gated by upstream stdlib / language-level defects. Public-API tests do not pass yet — only @ignored regressions exist to lock the bug shapes. Avoid in production until promoted.
undocumentedDocumentation in this reference is authoritative, but the module has not yet been routed through the core-tests/ conformance suite. The current page is a best-effort snapshot of the source; it may drift from runtime behaviour.
unverifiedThe conformance suite has not been run against this module, so nothing on this row is a measurement. Distinct from undocumented: the module IS routed into core-tests/, but no result has been recorded since the liveness check that began demanding one.
ModuleStatusConformance suite
poll.vrpartialcore-tests/async/poll — 42 unit + 21 property + 14 integration + 7 regression-as-guardrail (79 working, 0 pinned)
waker.vrpartialcore-tests/async/waker — 9 working + 2 pinned regressions (inline-vtable redesign + record-literal Clone-corruption fix + Waker construction inlining; waker construction + clone + wake_by_ref + will_wake all now green; 2 residual pins: fn_ref-as-Int identity stability and Debug auto-derive precedence for record types)
future.vrpartialcore-tests/async/future — 17 working (construction + SelectResult variant algebra + Maybe interop) + 14 pinned regressions (the Future.poll / FutureExt.block surface is gated by the same &self auto-deref defect as waker §C)
backoff.vrpartialcore-tests/async/backoff — 14 working (BackoffStrategy variant + match-coverage) + 7 pinned regressions (Backoff. blocked by upstream CSPRNG intrinsic gap shared with reservoir)
task.vrpartialcore-tests/async/task — 16 working (JoinError variant algebra + TaskId record construction + List bookkeeping) + 2 pinned (TaskId.new atomic counter + JoinError Debug, both gated by upstream defects)
diagnostics.vrpartialcore-tests/async/diagnostics — 15 working (TaskLifecycleState 6-variant lifecycle + partition + terminal-state classification + List-histogram). Pure data-type module — no runtime dependency.
cancellation.vrpartialcore-tests/async/cancellation — 11 working (CancelReason 4-variant + Aborted(Text) payload + List bookkeeping). Timeout{deadline} arm deferred to integration once Instant works under interp.
channel.vrpartialcore-tests/async/channel — 12 working (TrySendError + TryRecvError variant algebra + payload recovery + retry-signal classification).
broadcast.vrpartialcore-tests/async/broadcast — 18 working (BroadcastRecvError + TryRecvResult 4-variant + LagPolicy 3-variant + Maybe interop).
select.vrpartialcore-tests/async/select — 12 working (Either<A,B> + SelectError + race-outcome bookkeeping).
nursery.vrpartialcore-tests/async/nursery — 8 working (NurseryErrorBehavior 3-policy + priority/severity ordering).
spawn_config.vrpartialcore-tests/async/spawn_config — 21 working (RestartPolicy + IsolationLevel + Priority 4-rank ordering + will-restart classification).
spawn_with.vrpartialcore-tests/async/spawn_with — 10 working (CircuitState 3-variant breaker lifecycle Closed → Open → HalfOpen → Closed + can-attempt classification).
executor.vrpartial (configuration surface)core-tests/async/executor — the whole configuration surface is exercised and green under the interpreter: RuntimeBuilder construction, its setters and full builder chains, the AsyncRuntimeConfig presets (default / cpu_bound / io_bound) and setter surface, bulk containers of builders, and the algebraic laws the presets obey. A live block_on(future) round-trip waits on ahead-of-time compilation.
stream.vrpartialcore-tests/async/stream — unit, property, integration and regression suites all green under the interpreter. Pinned behaviour includes the size_hint laws at the empty / once / repeat boundaries, poll-loop exhaustion stickiness, generator-driven termination through stream_from_fn, draining a List through a manual poll driver, and clone of a generic element type. Coverage of chained adapters (map / filter / take over one another) is deferred behind dispatch through constrained implement blocks.
generator.vrunverified— runtime-bound; deferred pending the executor test-bed
timer.vrpartialcore-tests/async/timer — the construction surface for Sleep, SleepUntil and Delay, the immediate-versus-first-tick partition of TimerInterval, debounce and throttle state-machine round-trips, refusal of a non-monotonic instant, reset-then-acquire across representative intervals, and TimeoutError equality. Regression pins cover the Duration.from_millis misroute described under Known limitations.
parallel.vrcomplete (interp)core-tests/async/parallel — parallel_map, parallel_filter_map, parallel_for_each, parallel_reduce and the Blelloch parallel_scan_exclusive. The pinned properties are the interesting part: results are invariant under worker count, Blelloch agrees with a reference exclusive prefix scan for + and max, parallel_reduce agrees with a left fold, and filter_map yields an index subset of map. Validation of the native-compiled path waits on ahead-of-time compilation.
panic_fence.vrpartialcore-tests/async/panic_fence — the panic_safe factory, record-literal inners, Ready(Ok) round-trips over several element types, classification of a fence outcome into its tag, and sequential consumption of a list of fenced futures. One pin remains on in-place mutation of a generic record field through &mut self, which is what the fence's documented "inner is empty after Ready" invariant rests on. Coverage of the panicking arm waits on a panicking-future test bed.
semaphore.vrregression-only outside variant algebracore-tests/async/semaphore — SemaphoreError.Closed matching and Result/Maybe wrapping have coverage. Construction has a focused control below; lifecycle coverage remains incomplete and must be validated separately.
async_iterator.vrcomplete (protocol surface)core-tests/async/async_iterator — 4 unit + 3 property + 2 integration + 1 regression GREEN under interpreter. Pins: both protocols (AsyncIterator, IntoAsyncIterator) mount cleanly without archive-load panic; protocol-bound generic functions compile (A: AsyncIterator, B: IntoAsyncIterator + Clone); IntoAsyncIterator self-conversion blanket compiles via @inline-identity body; associated-type projection B.IntoAsyncIter resolves at function boundary; List<A: AsyncIterator> round-trip signature compiles. Stream→AsyncIterator blanket impl deferred behind upstream protocol-resolver projection-reduction work (each Stream-shaped type carries its own direct AsyncIterator impl in its owning module until the resolver lands).
intrinsics.vrpartialcore-tests/async/intrinsics — 19 working (Executor.current/in_async_context coherence + future_poll_sync ReadyFuture round-trip across Int/Text/Bool payloads + IntrinsicsYieldNow two-state lifecycle Pending→Ready with exactly-one-Pending tightness). Spawn family + sleep family @intrinsics deferred pending the live-executor test-bed.

Backend coverage​

A basic eager async program compiles to a native executable:

async fn work() -> Int { 21 }

async fn main() {
    let v = work().await;
    print(f"async={v + v}");
}

verum build produces a native binary whose recorded output is async=42. This control establishes basic eager async execution, not full executor or resource-cleanup correctness.

The construction control AsyncSemaphore.new(2) produces a semaphore. Duration.from_millis(250).as_nanos() returns 250_000_000, matching Duration.from_nanos(250_000_000).as_nanos(). These focused checks do not establish semaphore lifecycle or general timer scheduling behaviour.

Poll<T> — the two-state algebra​

Status: complete. Full public-API conformance under Tier 0 (interpreter) and Tier 2 (LLVM AOT, --test-threads 1). 42 unit + 21 property + 14 integration + 7 regression-as-guardrail tests; every algebraic law on the Functor-shape (identity, composition, Pending-as-absorbing- element), the Maybe-isomorphism round-trip, and the Eq / Clone / Default protocols are exhaustively pinned over a representative Int payload domain. Conformance suite: core-tests/async/poll.

The foundational poll-state algebra. Every async surface in the stdlib (Future.poll, Stream.poll_next, AsyncIterator.poll_next, cancellation checks, …) returns Poll<T> as its hot-path completion signal.

public type Poll<T> is
    | Ready(T)      // computation completed with value
    | Pending;      // not yet complete — the caller must re-poll

Predicates​

p.is_ready() -> Bool         // O(1), inlined
p.is_pending() -> Bool       // O(1), inlined; is_ready ⊕ is_pending = true ∀ p

Functorial map​

p.map<U, F: fn(T) -> U>(self, f: F) -> Poll<U>

Pending is the absorbing element: Pending.map(f) == Pending for any f; the closure is never invoked. Ready(t).map(f) materialises Ready(f(t)).

let p: Poll<Int>  = Poll.Ready(3);
let q: Poll<Int>  = p.map(|x| x * 2);              // Poll.Ready(6)
let r: Poll<Text> = p.map(|x| f"{x}");             // Poll.Ready("3")

Extraction​

p.unwrap(self)            -> T                  // panics if Pending
p.unwrap_or(self, default: T) -> T              // Pending → default
p.ready(self)             -> Maybe<T>           // Pending → None

p.ready() is the canonical bridge to Maybe<T>: Ready(t) ↔ Some(t) and Pending ↔ None. The reverse direction is the From<Maybe<T>> impl below; the two together form a structural isomorphism between Poll<T> and Maybe<T> on the observable surface.

Conversions​

From<Maybe<T>> for Poll<T>           // Some(t) -> Ready(t), None -> Pending

Removed blanket impl. An earlier From<T> for Poll<T> blanket impl was deleted: under any substitution T = Maybe<U> it overlapped From<Maybe<T>> and the impl-coherence selector silently routed Poll.from(non_maybe_value) through the Maybe arm — collapsing the result to Poll.Pending regardless of input. Write Poll.Ready(value) directly; it is the explicit, unambiguous form and produces byte-identical code.

Poll<Result<T, E>> — async-result composition​

map_ok / map_err live on the Poll<Result<T, E>> shape so async functions whose Output is Result<T, E> compose without intermediate destructuring.

Poll.ready_ok(t: T)   -> Poll<Result<T, E>>     // shorthand for Ready(Ok(t))
Poll.ready_err(e: E)  -> Poll<Result<T, E>>     // shorthand for Ready(Err(e))

p.map_ok<U, F>(self, f: F) -> Poll<Result<U, E>>      // F: fn(T) -> U
p.map_err<U, F>(self, f: F) -> Poll<Result<T, U>>     // F: fn(E) -> U

Pending remains the absorbing element on both arms: Pending.map_ok(f) == Pending and Pending.map_err(f) == Pending. Ready(Err(e)).map_ok(f) == Ready(Err(e)) (no re-wrapping); symmetric for map_err.

let p: Poll<Result<Int, Text>> = Poll.ready_ok(10);
let mid: Poll<Result<Int, Text>> = p.map_ok(|x| x + 1);
let after: Poll<Result<Int, Text>> = mid.map_ok(|x| x * 2);  // Ready(Ok(22))

Protocol implementations​

implement<T: Eq>      Eq      for Poll<T>;     // structural on Ready/Pending
implement<T: Clone>   Clone   for Poll<T>;
implement<T>          Default for Poll<T>;     // = Pending
implement<T: Debug>   Debug   for Poll<T>;     // "Ready(<t>)" | "Pending"

Algebraic laws​

All pinned by core-tests/async/poll/property_test.vr:

LawStatement
Functor identityp.map(|x| x) == p
Functor compositionp.map(g).map(f) == p.map(|x| f(g(x)))
Pending absorbs mapPending.map(f) == Pending
Maybe round-tripPoll.from(p.ready()) == p
Result arm preservationReady(Err(e)).map_ok(f) == Ready(Err(e)); symmetric for map_err
Eq reflexive / symmetric / transitivestandard equality laws on the two-variant carrier
Default invariancePoll.default() = Pending for every payload type

Pattern matching​

Poll<T> is exhaustive on two arms; nest with Maybe or Result for the common composite patterns:

fn classify(p: Poll<Result<Int, Text>>) -> Text {
    match p {
        Poll.Ready(Ok(_))  => "ready-ok",
        Poll.Ready(Err(_)) => "ready-err",
        Poll.Pending       => "pending",
    }
}

Future protocol​

type Future is protocol {
    type Output;
    fn poll(&mut self, cx: &mut Context) -> Poll<Self.Output>;
}

type IntoFuture is protocol {
    type Future: Future;
    fn into_future(self) -> Self.Future;
}

type FutureExt is protocol extends Future {
    fn map<U, F>(self, f: F) -> MapFuture<Self, F>
        where F: fn(Self.Output) -> U;
    fn and_then<U, F, Fut2>(self, f: F) -> AndThenFuture<Self, F, Fut2>
        where F: fn(Self.Output) -> Fut2, Fut2: Future<Output = U>;
    fn block(mut self) -> Self.Output;              // block current thread
}

Factories​

ready(value) -> ReadyFuture<T>             // immediately completes
pending<T>() -> PendingFuture<T>         // never completes
lazy(|| compute()) -> Lazy<F, T>           // deferred closure

Combinators (also available on FutureExt)​

join(fut1, fut2)               -> Join2<Fut1, Fut2>      // (O1, O2)
join3(fut1, fut2, fut3)        -> Join3                   // (O1, O2, O3)
join_all(futures)              -> List<Output>            // for List<F>
try_join(fut1, fut2)           -> Result<(O1, O2), E>     // fail-fast on Err
select(fut1, fut2)             -> Select2                 // first to complete wins
select_either(fut1, fut2)      -> Either<A, B>
race(fut1, fut2)               -> T                       // winner; loser cancelled
select_all(futures)            -> SelectAllResult<T>      // first + index
try_first(futures)             -> SelectAllResult<Output> // first Ok
timeout(fut, duration)         -> Result<T, TimeoutError>

Tasks​

type TaskId is { id: Int };
type Task<T>  is { ... };
type JoinHandle<T> is { ... };
type JoinError is Cancelled | Panicked;

spawn(future) -> JoinHandle<T>                   // shorthand
spawn_blocking(f) -> JoinHandle<T>               // on thread pool
spawn_detached(future) -> ()                     // fire-and-forget
yield_now() -> YieldNow                          // cooperate

h.cancel()                                        // request cancellation
h.is_finished() -> Bool
h.id() -> TaskId
h.await -> Result<T, JoinError>                   // via Future

These are the declarations of the library types in core.async.task. Its Future implementation adds the join result: for a task whose T is Result<Value, WorkError>, its Output is Result<Result<Value, WorkError>, JoinError>. An application error and a join error occupy different layers; Cancelled and Panicked have no payload.

Known limitation: the spawn keyword also has a compiler task-handle path that yields the task body's result directly. The library record's methods and Future.Output must not be assumed for every keyword handle, or obtained by adding an annotation. See task completion contracts before storing handles in a collection or a generic wrapper.

JoinSet<T> — dynamic task collection​

let mut set: JoinSet<Int> = JoinSet.new();
set.spawn(task_a());
set.spawn(task_b());

// `JoinSet<T>` IS the future — `implement<T> Future for JoinSet<T>`
// with `Output = Result<T, JoinError>`. Awaiting it yields the FIRST
// task to finish and removes that task from the set, so the loop
// condition is emptiness. There is no `join_next`.
while !set.is_empty() {
    let res = set.await;
    match res {
        Result.Ok(value) => ...,
        Result.Err(JoinError.Cancelled) => ...,
        Result.Err(JoinError.Panicked)  => ...,   // no payload
    }
}

Channels​

MPSC — Sender<T> / Receiver<T>​

channel<T>() -> (Sender<T>, Receiver<T>)                // unbounded
bounded<T>(capacity) -> (Sender<T>, Receiver<T>)
unbounded_channel<T>()                                  // alias for channel()
bounded_channel<T>(cap)                                 // alias for bounded()

// Sync API
tx.send(value)  -> Result<(), SendError<T>>               // blocks (futex) if bounded full
tx.try_send(value) -> Result<(), TrySendError<T>>         // Full | Disconnected
tx.send_timeout(value, d) -> Result<(), SendError<T>>     // with deadline

// Async API — awaitable with waker-based backpressure
tx.send_async(value) -> SendFut<T>                        // Future<Result<(), SendError<T>>>
tx.send_cancellable(value, &token)                        // Future; Err(Cancelled) on token fire
tx.closed() -> ChannelClosed<T>                           // Future<()>; completes on close

// Inspection
tx.is_closed() -> Bool                                    // == is_disconnected
tx.is_disconnected() -> Bool
tx.capacity() -> Maybe<Int>
tx.len() -> Int

// Receiver
rx.recv() -> Maybe<T>                                     // sync blocking
rx.recv_fut() -> RecvFut<T>                               // explicit async Future
rx.recv_cancellable(&token)                               // Future; Err(Cancelled) on token fire
rx.recv_many(&mut buf, max) -> Int                        // batch-drain in one lock
rx.try_recv() -> Result<T, TryRecvError>                  // Empty | Disconnected

// Receiver implements Stream<Item = T> — works with `for await msg in rx { ... }`
// and all StreamExt combinators (map, filter, take, etc.).

Async backpressure​

For bounded channels, send_async(value).await is the idiomatic way to apply backpressure:

  • If the channel has slack — push and resolve Ok(()) immediately.
  • If full — register the caller's waker in sender_wakers, return Poll.Pending. When the receiver pops, the sender's waker fires and the future re-polls, now with space.

The blocking send() method uses the same notification path but via futex, for non-async callers. Both paths share state; a bounded channel is safe to use from a mix of async and blocking senders.

Cancellation integration​

Both async variants accept a &CancellationToken parameter via *_cancellable — on token fire the future resolves immediately with Err(CancellationError) or Err(CancellableSendError.Cancelled(..)), deregisters its waker, and returns control. Pattern:

select {
    msg = rx.recv_cancellable(&shutdown).await => match msg {
        Ok(Some(v)) => handle(v),
        Ok(None)    => return,   // channel closed
        Err(_)      => return,   // shutdown fired
    },
    _ = idle_timeout.await => return,
}

One-shot — oneshot<T>()​

let (tx, rx) = oneshot<Result<Data, Error>>();
spawn async move { tx.send(compute()); };
let result = rx.await;

Broadcast (MPMC) — broadcast_channel​

broadcast_channel<T>(capacity) -> (BroadcastSender<T>, BroadcastReceiver<T>)
broadcast_channel_with<T>(capacity, policy) -> (Sender, Receiver)

// Sender
tx.send(value) -> Result<Int, SendError<T>>               // returns listener count
tx.clone() -> BroadcastSender<T>                           // multi-producer
tx.subscribe() -> BroadcastReceiver<T>                     // new receiver starting now
tx.receiver_count() -> Int
tx.sender_count() -> Int
tx.is_closed() -> Bool
tx.close()

// Receiver — implements Future AND Stream
rx.recv() -> BroadcastRecv<T>                              // awaitable future
rx.recv_cancellable(&token) -> Result<Result<T, RecvError>, CancellationError>
rx.try_recv() -> TryRecvResult<T>                          // Value/Empty/Closed/Lagged
rx.len() -> Int
rx.is_empty() -> Bool

Lag policies — LagPolicy​

Controls behavior when a receiver falls behind the ring capacity:

VariantSemantics
LagTolerant (default)Return RecvError.Lagged(n); advance to oldest available message.
DropOldestAdvance silently; never return Lagged. Senders never block.
DropSlowReceiverUnsubscribe the slow receiver entirely. For strict keep-up SLAs.

Broadcast receivers observe every value sent after subscription; they do not see historic values.

BroadcastReceiver<T> implements both Future<Output = Result<T, RecvError>> (direct .await) and Stream<Item = Result<T, RecvError>> (for-await loops and combinators). Sender and receiver counts are maintained atomically; last-sender-drop closes the channel and wakes all receivers.

Streams​

type Stream is protocol {
    type Item;
    fn poll_next(&mut self, cx: &mut Context) -> Poll<Maybe<Self.Item>>;
}
type IntoStream is protocol { ... };
type StreamExt is protocol extends Stream { ... };

Factories​

iter(iterable) -> Iter<I>                    // any IntoIterator
once(item) -> StreamOnce<T>
once_future(fut) -> StreamOnce<Output>
empty<T>() -> StreamEmpty
repeat(item) -> StreamRepeat<T>              // infinite (T: Clone)
repeat_n(item, n) -> StreamRepeatN<T>
from_fn(|| produce_next()) -> StreamFromFn
poll_fn(|cx| ...) -> StreamFromFn
unfold(state, |s| (item, new_state)) -> StreamUnfold
interval(duration) -> Interval

Adapters (return new streams)​

s.map(|x| f(x))            s.filter(|x| pred(x))     s.filter_map(|x| ...)
s.take(n)                  s.skip(n)
s.take_while(|x| pred)     s.skip_while(|x| pred)
s.chain(other)             s.zip(other)              s.enumerate()
s.peekable()               s.flatten()               s.fuse()
s.throttle(rate)           s.debounce(duration)      s.chunks(n)
Three combinators on that list do not exist

buffer_unordered(n), buffered(n) and timeout_each(duration) are not declared on any stream type in core/. (buffered exists in core/database/postgres and core/net/tls13 — different types, not reachable from a Stream.) Everything else above is real and lives in core/async/stream.vr.

For bounded concurrency, drive the work through a JoinSet or a semaphore yourself; for a per-item deadline, wrap each future in the timeout combinator before it enters the stream.

Consumers (terminal)​

s.next() -> Poll<Maybe<Item>>
s.try_next() -> Poll<Maybe<Result<T, E>>>
s.for_each(|x| side_effect(x))
s.fold(init, |acc, x| ...) -> B
s.reduce(|a, b| ...) -> Maybe<Item>
s.collect<C>() -> C
s.find(|x| pred) -> Maybe<Item>
s.any(|x| pred) / s.all(|x| pred)
s.count() / s.last() / s.nth(n)
s.position(|x| pred) -> Maybe<Int>

Example​

async fn monitor(sensor: &Sensor) using [Logger] {
    let mut s = interval(1.secs())
        .map(|_| sensor.read())
        .filter(|r| r.is_ok())
        .map(|r| r.unwrap())
        .throttle_filter(|v| v.delta() > 0.01);

    while let Maybe.Some(reading) = s.next().await {
        Logger.info(&f"reading: {reading}");
    }
}

Generators​

fn* fibonacci() -> Int {
    let (mut a, mut b) = (0, 1);
    loop {
        yield a;
        (a, b) = (b, a + b);
    }
}

for n in fibonacci().take(10) {
    print(f"{n}");
}
  • fn* name(...) -> T — synchronous generator; returns Iterator<Item=T>.
  • async fn* name(...) -> T — async generator; returns AsyncIterator<Item=T> (a Stream<T>).

Inside a generator:

yield value;                // emit
// `return` (no value) ends the generator

Async generators support .await:

// `Http.get_streaming` and `body.next_chunk()` do not exist —
// `Response` holds its body as a `List<Byte>` field and a response is
// fully received by the time you hold it. The `.await` a generator can
// use is the request itself.
async fn* stream_events(url: &Text) -> Event using [Http] {
    let resp = Http.get(url).await?;
    for e in parse_events(resp.body_bytes()) {
        yield e;
    }
}

for await event in stream_events("wss://…") using [Http] {
    handle(event);
}

Nursery — structured concurrency​

public type NurseryOptions is {
    timeout:   Maybe<Duration>,
    max_tasks: Maybe<Int>,
    on_error:  NurseryErrorBehavior,         // field name matches the surface syntax
};

public type NurseryErrorBehavior is CancelAll | WaitAll | FailFast;

public type NurseryError is
    | Single(Heap<Error>)                    // single-task failure
    | Multiple(List<Heap<Error>>)            // WaitAll collected multiple failures
    | Timeout { pending_count: Int, completed_count: Int }
    | Cancelled
    | Panic(Text)
    | TaskLimitExceeded { limit: Int, attempted: Int };

// Builder
implement NurseryOptions {
    public fn new() -> Self;
    public fn default() -> Self;                                   // alias for new
    public fn with_timeout(self, timeout: Duration) -> Self;
    public fn with_timeout_ms(self, ms: Int) -> Self;
    public fn with_max_tasks(self, max: Int) -> Self;
    public fn with_error_behavior(self, behavior: NurseryErrorBehavior) -> Self;
}

// Underlying async functions the nursery { ... } block lowers to:
public async fn with_nursery<T>(
    options: NurseryOptions,
    f: fn(&mut Nursery) -> T,
) -> Result<T, NurseryError>;

public async fn with_nursery_timeout<T>(
    duration: Duration,
    f: fn(&mut Nursery) -> T,
) -> Result<T, NurseryError>;

Usage​

async fn fetch_batch(urls: &List<Text>) -> List<Bytes> using [Http] {
    nursery(
        timeout: 10.secs(),
        on_error: cancel_all,
        max_tasks: 100,
    ) {
        let handles: List<JoinHandle<Bytes>> = urls.iter()
            .map(|u| spawn Http.get(u.clone()))
            .collect();
        try_join_all(handles).await?
    } on_cancel {
        metrics.increment("fetch_batch.cancelled");
    } recover |e| {
        log_error(&e);
        List.new()
    }
}

Guarantees​

  • Every spawned task completes, fails, or is cancelled before the nursery scope exits.
  • on_error policies:
    • cancel_all — one failure cancels all siblings.
    • wait_all — collect all results including errors.
    • fail_fast — return the first error immediately.
  • Context stacks are inherited by spawned tasks.

Cancellation​

Cooperative cancellation is implemented by core.async.cancellation. A cancelled task continues running until it hits a cancel point — an .await on a cancellation-aware future, an explicit throw_if_cancelled check, or a CancelScope exit.

Core types​

// Owner — only holder can cancel.
CancellationTokenSource.new()
  .token()     -> CancellationToken                  // observer handle (clone-cheap)
  .cancel()                                          // with default CancelReason.Cancelled
  .cancel_with(reason)
  .is_cancelled() -> Bool
  .reason() -> Maybe<CancelReason>
  .linked_to(&parent)                                 // child source, propagates from parent

// Observer — read-only view.
token: CancellationToken
  .is_cancelled() -> Bool
  .reason() -> Maybe<CancelReason>
  .throw_if_cancelled() -> Result<(), CancellationError>
  .cancelled() -> CancelledFuture                    // Future<Output = CancelReason>
  .register(fn()) -> Registration                    // sync callback; RAII-deregister
  .child_source() -> CancellationTokenSource          // propagation tree
  .combine(&[t1, t2, ...]) -> CancellationToken       // any-of aggregation (static fn)
  .with_timeout(Duration) -> CancellationToken        // auto-cancel (static fn)
  .with_deadline(Instant) -> CancellationToken
  .never() -> CancellationToken                       // sentinel; never fires

Structured reasons​

type CancelReason is
    | Cancelled
    | Timeout { deadline: Instant }
    | ParentCancelled
    | Aborted(Text);

Children of a cancelled parent see ParentCancelled — not the parent's own reason. This makes structured traceability explicit.

Awaitable integration​

token.cancelled() returns a CancelledFuture that completes with the token's CancelReason. Compose with select:

select {
    r   = work().await           => Ok(r),
    res = token.cancelled().await => Err(res),
}

The future deregisters its waker on drop; no stale wake-ups.

Sync callback bridge — Registration​

let reg = token.register(|| close_file_handle(fd));
// ... do work ...
// Drop of `reg` deregisters BEFORE cancel fires. If cancel already
// happened, the callback fired synchronously inside `register()`.

Scoped cancellation — CancelScope​

let scope = CancelScope.new();
let token = scope.token();
spawn worker(token.clone());
// ... work ...
// Dropping `scope` cancels `token` (unless `scope.dismiss()` was called).

Scope variants:

CancelScope.new()                              // auto-cancel on drop
CancelScope.linked_to(&parent_token)           // child-source pattern
CancelScope.with_timeout(Duration)             // auto-cancel + timeout
scope.dismiss()                                // opt-out of drop-cancel
scope.cancel()                                 // explicit

Propagation rules (normative)​

  1. Parent → child: source.cancel() propagates to every token linked via child_source() / linked_to(). Children fire with CancelReason.ParentCancelled.
  2. Combine: tokens from CancellationToken.combine(&inputs) fire when any input fires.
  3. Idempotent: subsequent calls to cancel_with() after a cancel are no-ops; the first call's reason wins.
  4. Registered callbacks and wakers are drained under lock, then invoked without the lock held (no re-entrancy).
  5. Dropped registrations / futures deregister automatically.

Async signal subscription — core.signal​

Async-aware wrapper around core.sys.signal — exposes OS signals as awaitable futures and AsyncIterator streams for ergonomic composition with select, nursery, and cancellation tokens.

// Wait for a single Ctrl-C
ctrl_c().await;                                  // -> ()

// Wait for a single SIGTERM (K8s pod-eviction trigger)
terminate().await;                               // -> ()

// Wait for SIGHUP (reload-config convention)
hup().await;                                     // -> ()

// Wait for any shutdown signal; returns which one fired
let sig: Signal = shutdown_signals().await;      // Int | Term | Hup

// Arbitrary signal set as a Stream of arrivals
let mut stream = signal_stream(&[Signal.Usr1, Signal.Usr2]);
for await sig in stream {
    handle(sig);
}

Idiomatic shutdown — race server work vs signal:

select {
    _ = ctrl_c().await           => drain_and_exit(),
    _ = shutdown_signals().await => drain_and_exit(),
    r = run_server().await       => handle_result(r),
}

Architecture​

Invoking any allocating or lock-taking operation from inside a POSIX signal handler is undefined — the handler may preempt the mainline thread mid-malloc, mid-mutex-unlock, etc. core.signal uses the standard self-pipe / atomic-flag pattern:

  1. The OS-level signal handler (registered once per subscribed signal via core.sys.signal.on_signal) does only an async-signal-safe atomic store into a SignalFlag (set bit).
  2. A single background poller task polls these flags every ~20 ms, clears any set flags, and fans out to subscribers via a BroadcastSender<Signal>. The poller runs in normal runtime context, so broadcasting and waker operations are safe.

Trade-off: up to ~20 ms signal-to-subscriber latency — acceptable for shutdown, reload, and heartbeat use cases. A future upgrade to Linux signalfd(2), kqueue EVFILT_SIGNAL, or Windows APC delivery will collapse latency to zero behind the same public API.

Timers​

sleep(duration) -> Sleep                   // await to suspend
sleep_ms(ms)                               sleep_secs(secs)
sleep_until(deadline: Instant) -> SleepUntil

delay(future) -> Delay<F>                  // delay future by a duration
timeout(future, duration) -> Timeout<F>    // -> Result<T, TimeoutError>
debounce(future) -> Debounce<F>            // suppress rapid calls
throttle(future) -> Throttle<F>            // rate-limit

interval(duration) -> Interval              // stream firing on schedule
let mut ticker = Interval.new(500.millis());
loop {
    ticker.tick().await;
    update_ui();
}

Runtime and executor​

type RuntimeConfig is { ... };
type AsyncRuntime is { ... };              // `core.async.executor`
type LocalExecutor is { ... };
type TimeoutError is ();
type ExecutionEnv is { ... };               // θ+ context

RuntimeBuilder.new() -> RuntimeBuilder
builder.stack_size(bytes).max_tasks(n)
       .enable_work_stealing().enable_io().enable_time()
       .build() -> AsyncRuntime

rt.block_on(future) -> Output              // rt: AsyncRuntime
rt.spawn(future) -> JoinHandle<T>
rt.shutdown()                              // no deadline variant exists

Global helpers​

block_on(future) -> Output                 // uses default runtime
spawn(future) -> JoinHandle<T>
current_runtime() -> Maybe<&AsyncRuntime>

LocalExecutor​

Single-threaded executor for !Send futures:

let exec = LocalExecutor.new();
exec.spawn_local(future);
exec.spawn_local(main_future);
// `run_until_complete(&mut self) -> Int` takes NO future and returns the
// number of tasks it drove. Spawn everything first, then drive.
// There is no `run_until`.
let completed = exec.run_until_complete();
warning

Known limitation: the sequence above panics as soon as there is a task to drive.

Panic: GetFieldNamed: receiver is not a heap object (kind=nil)

Each call is fine alone — new(), a lone spawn_local, and run_until_complete on an empty executor all return. The panic needs a spawned task that actually gets driven, so a probe that stopped short of that would report the API healthy. block_on fails too, differently: method 'poll' not found on receiver of runtime kind Int.

What drives a future today, run before it is recommended here:

mount core.async.{spawn};

async fn work(n: Int) -> Int { return n * 2; }

fn main() {
    let handle = spawn(async {
        let v = work(3).await;
        print(f"spawned={v}");
    });
    handle.await;
    print("joined");
}
spawned=6
joined

spawn returns a handle and awaiting the handle drives the task to completion — awaiting works from a synchronous fn main. That is the shape the shipped async example uses, and it is the one to reach for until this note goes away.

Spawn configuration​

type SpawnConfig is { ... };                // builder
// The no-recovery arm is `NoRecovery`, not `None`, and `Fallback`
// carries an Int rather than a closure.
type RecoveryStrategy is
    | NoRecovery
    | Retry(RetryConfig)
    | CircuitBreaker(CircuitBreakerConfig)
    | Fallback(Int)
    | Supervised;
type RestartPolicy is Permanent | Transient | Temporary;
type IsolationLevel is Shared | SendOnly | Full;
type Priority is Low | Normal | High | Critical;

let cfg = SpawnConfig.new()
    .with_priority(Priority.High)
    .with_isolation(IsolationLevel.Full)
    .with_recovery(RecoveryStrategy.Retry(RetryConfig.default()))
    .with_timeout_ms(5000)
    .with_name("worker-42");

let handle = spawn_with(cfg, task());

Retry and circuit breaker​

Every field name here was wrong, and so were both constructors: RetryConfig has FOUR fields and a single default().

type RetryConfig is {
    max_retries: Int,        // not `max_attempts`
    initial_delay_ms: Int,   // not `initial_backoff_ms`
    max_delay_ms: Int,       // not `max_backoff_ms`
    backoff_factor: Int,     // an Int multiplier, not a Float
};

RetryConfig.default()        // 3 attempts, 100 ms initial, 2x, 5 s cap

There is no RetryConfig.fixed and no RetryConfig.exponential; build the record directly when the default does not suit. The two free functions take positional arguments — the language has no named ones:

execute_with_retry(f, max_attempts: Int, backoff_ms: Int) -> Result<T, Text>
execute_with_retry_config(f, config: RetryConfig)          -> Result<T, Text>

Circuit breaker​

type CircuitBreakerConfig is {
    failure_threshold: Int,
    reset_timeout_ms: Int,
    half_open_max_calls: Int,
};
type CircuitState is Closed | Open | HalfOpen;

let breaker = CircuitBreaker.new(CircuitBreakerConfig {
    failure_threshold: 5,
    reset_timeout_ms: 30_000,
    half_open_max_calls: 1,
});

if breaker.is_call_allowed() {
    match call_remote().await {
        Result.Ok(v)  => { breaker.record_success(); Result.Ok(v) }
        Result.Err(e) => { breaker.record_failure(); Result.Err(e) }
    }
} else {
    Result.Err(Error.new("circuit open"))
}

Parallel helpers​

Data-parallel patterns, implemented in a portable way that the work- stealing runtime can pick up.

parallel_map(items, worker_count, |x| f(x)) -> List<U>
parallel_filter_map(items, worker_count, |x| maybe_transform(x)) -> List<U>
parallel_for_each(items, worker_count, |x| side_effect(x))
parallel_reduce(items, worker_count, |a, b| combine(a, b)) -> Maybe<T>

worker_count = 0 means "default to num_cpus()".

Low-level intrinsics (intrinsics.vr)​

type Executor is { ... };                  // opaque handle

Executor.current() -> Maybe<Executor>
Executor.in_async_context() -> Bool

spawn_with_env(future) -> JoinHandle<T>
executor_spawn(&exec, future) -> JoinHandle<T>
executor_block_on(future) -> Output
future_poll_sync(&mut future) -> Maybe<Output>      // single poll, sync

async_sleep_ms(ms) / async_sleep_ns(ns)

User code rarely touches these; they exist for runtime authors.

Waker and Context​

type Waker is { ... };
type Context<'a> is { waker: &'a Waker };
type RawWaker     is { ... };
type RawWakerVTable is { ... };

noop_waker() -> Waker
Context.from_waker(&waker) -> Context<'a>

waker.wake()         // consume, enqueue task
waker.wake_by_ref()  // enqueue without consuming
waker.clone()

Context inheritance across .await and spawn​

  • .await preserves the current context stack verbatim.
  • spawn snapshots the parent's context stack at spawn time.
  • nursery { spawn ... } — tasks inherit the nursery's contexts.
  • Channels do not propagate contexts (they're pure data pipes).

See Language → context system for the rules.

semaphore — cooperative task limiter​

mount core.async.semaphore.{AsyncSemaphore, SemaphorePermit};

let sem = AsyncSemaphore.new(10);      // cap at 10 concurrent ops

for url in urls {
    let permit = sem.acquire().await?;
    spawn(async move {
        let _p = permit;                // held for task lifetime (RAII)
        fetch(&url).await;
    });
}

Async-task counting semaphore — waiters park via Future / Waker instead of blocking an OS thread (unlike core.sync.Semaphore which futex-blocks). FIFO waiter fairness; try_acquire() non-blocking fast path; add_permits(n) runtime resize; close() causes pending + future acquire calls to fail with SemaphoreError.Closed.

Typical deployments:

  • bounded outbound fan-out (N concurrent HTTP fetches)
  • DB connection-pool checkout
  • rate-limit async CPU-bound tasks (N parallel inferences)
  • producer/consumer backpressure without a channel

backoff — retry delay policies​

mount core.async.backoff.{Backoff, BackoffStrategy};

let mut bo = Backoff.exponential_full_jitter(
    Duration.from_millis(100),
    Duration.from_secs(30),
).with_max_attempts(5);

loop {
    match try_operation() {
        Ok(r)    => return r,
        Err(_)   => match bo.next_delay() {
            Some(d) => async_sleep(d).await,
            None    => return Err(MaxAttemptsReached),
        },
    }
}

Four industry-standard strategies:

StrategyFormulaNotes
ExponentialNoJitterbase × 2^ndeterministic
ExponentialFullJitterrand(0, base × 2^n)AWS default
ExponentialDecorrelatedrand(base, prev × 3)AWS whitepaper; best for large fleets
FibonacciFullJitterbase × F(n+1) jitteredgentler ramp

Overflow-guarded integer arithmetic over microseconds. Once base × 2^attempt would overflow UInt64, the raw value saturates at cap_us — pathological max_attempts values plateau at the configured ceiling instead of wrapping.

See also​