Publish a transitive owner to a running managed task’s scope.

Only a normal exit of owner will prove its subtree drained; any other exit settles the task as DrainProofLost. Synchronous: returns once the scope has the owner under monitor, or Refused at once if the scope has already exited. May be called from any process holding the ledger, any number of times, for as many owners as the task discovers.

Examples

case weft.adopt(ledger, owner: http_owner, cancel: stop_http) {
  weft.Adopted -> begin_http()
  weft.Refused -> Nil
}

Publish a leaf owner to a running managed task’s scope: an owner that provably owns nothing further, completed by any exit. The prepared_leaf exemption, for owners discovered mid-run.

Examples

case weft.adopt_leaf(ledger, owner: pump, cancel: stop_pump) {
  weft.Adopted -> start_pump()
  weft.Refused -> Nil
}

Fire a cancel signal, ending every run watching it.

Firing a signal that has already fired does nothing. Runs watching the signal report Abandoned for the tasks that were running and NeverStarted for the tasks that had not begun.

Examples

let stop = weft.cancel_signal()
process.spawn(fn() { weft.new(tasks) |> weft.cancel_with(stop) |> weft.start })

// From any process, at any time, including while the run's caller is
// blocked inside `start`:
weft.cancel(stop)

Cancel a detached run without giving up its account.

Unlike a fold reducer’s Halt, which discards what it has not seen, this keeps the deliveries coming: running tasks settle as Abandoned, unstarted ones as NeverStarted, and managed tasks by whatever their owners’ exits prove. Idempotent — cancelling a cancelled or finished run does nothing.

Examples

weft.cancel_detached(detached)
// ... keep pulling until AllDelivered ...

Bound how long a cancellation waits for managed owners to exit, in milliseconds.

Without a grace, cancelling a managed task means asking its owner to stop and waiting for the owner’s exit however long that takes, because “cancelled” without drain proof is not a fact this module will invent. With one, an owner still alive when the grace expires settles as CancellationUnconfirmed, the run finishes bounded — and the scope exits abnormally, because it can no longer vouch for the subtree it was witnessing. The grace bounds the wait, and what it buys in boundedness it pays for in the scope’s exit verdict.

One grace timer serves the whole cancellation rather than one per task: it is an acknowledgement window on the teardown, not a per-task timeout, and it is armed when cancellation begins, from whichever direction it came.

Examples

let outcomes =
  weft.new_prepared(tasks)
  |> weft.deadline(30_000)
  |> weft.cancel_grace(2000)
  |> weft.start

Create a fresh cancel signal.

The returned value can be passed to cancel_with on any number of runs and shared with any number of processes. It costs one idle process, which lives until cancel is called on it.

Examples

let stop = weft.cancel_signal()

let outcomes =
  weft.new(tasks)
  |> weft.cancel_with(stop)
  |> weft.start

Cancel the run when pid exits, for any reason.

This is the consumer’s death made a cancellation cause without the consumer having to be the process that started the run: a stream whose receiver is gone has nobody to deliver to and should stop its work. The watch is a monitor, never a link, so firing it ends the run and touches nothing else. A pid already dead when the run starts cancels it before any task is spawned, exactly like a spent cancel signal.

Examples

let witness =
  weft.new_prepared([weft.managed(serve)])
  |> weft.cancel_when_exits(consumer)
  |> weft.start_witnessed

Watch a cancel signal for the duration of the run.

This is the only way to stop a run from outside the process that started it, because that process is blocked inside start or fold and cannot act on its own behalf.

Examples

let stop = weft.cancel_signal()

let outcomes =
  weft.new(tasks)
  |> weft.cancel_with(stop)
  |> weft.start

Give the whole run a wall-clock budget, in milliseconds.

Hitting the deadline is not an error, it is a reason some entries in the account are Abandoned (started, then cut short) or NeverStarted (never got a slot). The timer is cancelled and flushed if the run finishes first, so it can never fire into a later receive.

Examples

let outcomes =
  weft.new(tasks)
  |> weft.deadline(30_000)
  |> weft.start

The errors of the tasks that returned Error, in the order given.

Tasks that crashed, were abandoned, or never started are not errors the tasks chose to return, so they appear here in no form at all; reach for partition when you need to see them.

Examples

let outcomes = [weft.Failed(0, "a"), weft.Crashed(1, process.Killed)]
assert weft.failures(outcomes) == ["a"]

Run tasks concurrently and return the first one to succeed, falling back to the full account if none did. The losers are killed and joined before a success returns.

A Failed or Crashed task does not end the run here; that is the whole point of the function, and it is why first_ok leaves on_failure at KeepGoing. An empty list is answerable, unlike in race, because the error channel already exists: it yields Error([]).

If something outside the run destroys the scope, the Error carries only the outcomes heard before the death rather than a total account; reach for start, which fills the gap in, when that distinction matters.

Examples

let body = weft.first_ok([fn() { fetch(mirror_a) }, fn() { fetch(mirror_b) }])
// -> Ok(<-body->)

// Nothing succeeded, so the caller gets the whole account in input order.
let answer = weft.first_ok([fn() { Error("a") }, fn() { Error("b") }])
// -> Error([weft.Failed(0, "a"), weft.Failed(1, "b")])

Consume outcomes in completion order as they land.

This is how a run whose results do not all fit in memory is processed, and how a caller acts on early results without waiting for the slow tail. The scope holds each outcome until the reducer asks for it, so at most one is ever in flight and a slow reducer stops new work from starting rather than filling a mailbox.

A reducer returning Halt ends the run: the remaining tasks are killed and joined before fold returns, so no worker survives the call. The outcomes of those tasks are not delivered — fold returns an accumulator and has nowhere to put them — so a caller that wants the full account of a run it stopped early should use start with a cancel signal instead.

Examples

// Count successes without holding every value.
let successes =
  weft.new(tasks)
  |> weft.fold(from: 0, with: fn(count, outcome) {
    case outcome {
      weft.Completed(..) -> weft.Continue(count + 1)
      weft.Failed(..) | weft.Crashed(..) -> weft.Continue(count)
      weft.Abandoned(..) | weft.NeverStarted(..) -> weft.Continue(count)
      weft.DrainProofLost(..) -> weft.Continue(count)
      weft.CancellationUnconfirmed(..) -> weft.Continue(count)
    }
  })

// Stop as soon as three values are in hand; the rest are cancelled.
let first_three =
  weft.new(tasks)
  |> weft.fold(from: [], with: fn(found, outcome) {
    let found = case outcome {
      weft.Completed(value:, ..) -> [value, ..found]
      weft.Failed(..) | weft.Crashed(..) -> found
      weft.Abandoned(..) | weft.NeverStarted(..) -> found
      weft.DrainProofLost(..) | weft.CancellationUnconfirmed(..) -> found
    }
    case list.length(found) >= 3 {
      True -> weft.Halt(found)
      False -> weft.Continue(found)
    }
  })

Set how many tasks may occupy a slot at once.

A slot is held from the moment a task is spawned until its outcome has been handed to the consumer, so this bounds running work and completed-but- unconsumed results together. A max below one is raised to one; there is no unbounded setting, by design.

Examples

let outcomes =
  weft.new(tasks)
  |> weft.limit(8)
  |> weft.start

Prepare a managed task whose owners are discovered while it runs.

This is prepared_task for work whose owners do not exist yet when the run starts: a request worker that prepares a transport, learns the pid holding its socket, and must have that pid witnessed before the socket is allowed to open. begin receives a Ledger; every adopt or adopt_leaf on it publishes one more owner to the scope, and the task’s slot is held and its outcome withheld until the worker and every adopted owner have exited. A task that adopts nothing behaves exactly like task.

Publication is the parked-work pattern applied mid-run: prepare the owner parked, adopt it, and release it only on Adopted. A Refused means the run is already cancelling — the scope retains the owner and asks it to stop, so nothing leaks, but the caller must not begin the work.

Examples

let outcomes =
  weft.new_prepared([
    weft.managed(fn(ledger) {
      let #(owner, cancel, release) = prepare(request)
      case weft.adopt(ledger, owner:, cancel:) {
        weft.Adopted -> Ok(release())
        weft.Refused -> Error(Cancelled)
      }
    }),
  ])
  |> weft.start

Apply a fallible function to every item, bounded, and return the account.

Exactly weft.new over the items with limit and start, which is the shape most callers want and the one it is easiest to get wrong by mapping an unbounded async over a list.

Examples

let outcomes = weft.map(urls, limit: 8, with: fetch)
let #(bodies, rest) = weft.partition(outcomes)

Begin a run over tasks, bounded by default.

The default limit is the number of schedulers online, not unbounded: spawning one process per item is fine for pure computation and actively harmful for anything holding a socket, a file handle, a database connection or a rate-limited quota. Someone who wants a wider fan-out says so with limit.

The default failure policy is KeepGoing.

Examples

let outcomes =
  urls
  |> list.map(fn(url) { fn() { fetch(url) } })
  |> weft.new
  |> weft.start

Begin a run over prepared tasks, plain and managed mixed freely.

This is new for tasks built with task, prepared_task and prepared_leaf. The defaults are the same: a limit of the schedulers online, KeepGoing on failure, no deadline, and no cancellation grace — a cancelled managed task is awaited until its owner exits, however long that takes, unless cancel_grace bounds it.

Examples

let outcomes =
  weft.new_prepared([
    weft.task(fn() { fetch(url) }),
    weft.prepared_task(owner:, cancel:, begin:),
  ])
  |> weft.start

Set what a failing task does to the tasks beside it.

Examples

// Stop at the first `Failed` or `Crashed`; the rest are reported as
// `Abandoned` or `NeverStarted` rather than being lost.
let outcomes =
  weft.new(tasks)
  |> weft.on_failure(weft.CancelSiblings)
  |> weft.start

Split an account into the values that succeeded and everything else.

Both halves keep the order they were given in. The second half stays a list of Outcome rather than a list of errors, because “did not succeed” covers four different facts and flattening them is how the accounting gets thrown away one call site at a time.

Examples

let outcomes = [weft.Completed(0, "a"), weft.Failed(1, "no")]
assert weft.partition(outcomes) == #(["a"], [weft.Failed(1, "no")])

Prepare a managed task whose owner provably owns nothing further.

The one difference from prepared_task: any exit of the owner completes the drain obligation, normal or not. This is the declared exemption for single-process owners — an observer, a pump, a relay — where an ordinary crash is an ordinary crash and must not be read as a lost proof over descendants that never existed. Everything else — the monitor-before-begin ordering, the held slot, the cancel helper — is identical.

Examples

let outcomes =
  weft.new_prepared([
    weft.prepared_leaf(owner: pump, cancel: stop_pump, begin: run),
  ])
  |> weft.start

Prepare a managed task: work whose lifecycle outlives its worker.

owner is the pid whose exit proves the task’s transitive work is gone — only a normal exit proves it; any other exit becomes DrainProofLost. cancel asks the subtree to stop; it must be safe to call more than once and safe to call after the work has already finished, and it runs on a disposable helper so a crash inside it cannot cost the run its witness. begin is the worker’s blocking path, and it runs only after the scope holds the owner under monitor.

The task’s slot is held, and its outcome withheld, until both the worker and the owner have exited. An owner already dead when the run starts yields DrainProofLost without begin ever running: proof that was never on file was never proof.

Examples

// `prepare` parks the real request and hands back its owning pid, a
// cancel capability, and the closure that lets it loose.
let #(owner, cancel, begin) = prepare(request)

let outcomes =
  weft.new_prepared([weft.prepared_task(owner:, cancel:, begin:)])
  |> weft.cancel_grace(2000)
  |> weft.start

Collect one outcome from a detached run, waiting at most within milliseconds.

Must be called from the process that called start_detached — the handle’s outbox belongs to it. Demand is unary and idempotent: a pull that times out leaves its demand standing, and the next pull re-grants it harmlessly, so at most one outcome is ever in flight and none is ever dropped or doubled.

Examples

case weft.pull(detached, within: 1000) {
  weft.PulledOutcome(outcome) -> act_on(outcome)
  weft.NotYet -> check_something_else()
  weft.AllDelivered -> done()
  weft.RunLost(reason) -> escalate(reason)
}

Run tasks concurrently and return the first one to complete, however it completed. The losers are killed and joined before this returns.

The first task is a separate argument from the rest because a race over nothing has no answer: Outcome has no variant meaning “there was nothing to race”, and wrapping the result in a Result would push an error case onto every call site that already knows its list is not empty. Making the empty case unrepresentable is cheaper than making every caller handle it. A caller holding a runtime list matches on it once and decides for itself what nothing should mean.

Note the difference from first_ok: this returns the first task to finish, even if it finished by failing or crashing. Conflating the two is a common bug, which is why they are separate functions.

Examples

// Whichever mirror answers first, even if it answers with an error.
let outcome = weft.race(fn() { fetch(mirror_a) }, [fn() { fetch(mirror_b) }])
// -> weft.Completed(index: 1, value: <-body->)

The scope process behind a detached run.

This is the pid an outer witness monitors, and the pid a nested run publishes as its owner: a detached run handed to prepared_task as owner: weft.scope_pid(inner) composes ownership — the outer scope’s proof for that task is the inner scope’s own drain verdict, with no translation code on either side.

Examples

let inner = weft.new_prepared(children) |> weft.start_detached

let outer_task =
  weft.prepared_task(
    owner: weft.scope_pid(inner),
    cancel: fn() { weft.cancel_detached(inner) },
    begin: fn() { collect(inner) },
  )

Run everything to completion and return every outcome in input order.

Blocks the calling process until the run is over. When it returns, every task has exactly one outcome and no worker from the run is alive.

Ordered versus unordered falls out of which terminal verb you call rather than out of a flag: start sorts the account back into input order, fold hands outcomes over in completion order as they land.

Examples

let outcomes =
  weft.new([fn() { Ok(1) }, fn() { Error("no") }])
  |> weft.start
// -> [weft.Completed(0, 1), weft.Failed(1, "no")]

let #(values, rest) =
  weft.new(tasks)
  |> weft.limit(4)
  |> weft.on_failure(weft.CancelSiblings)
  |> weft.start
  |> weft.partition

Start a run without blocking, and hand back its handle.

The run is the same in every respect — same scope, same account, same ownership guarantees — only the consumption differs: nothing is delivered until the holder asks with pull, so the run’s backpressure now reaches whatever pace the holder pulls at.

Examples

let detached =
  weft.new_prepared(tasks)
  |> weft.cancel_grace(2000)
  |> weft.start_detached

let watch = process.monitor(weft.scope_pid(detached))
// ... pull outcomes as the mood takes you ...

Start a run and push every outcome to sink as an ordinary message, from a relay process this function spawns and returns.

This is the adapter for consumers that are actors: an actor must not block its receive loop inside pull, so the relay owns the pulling and the actor merely receives. Each outcome arrives as PulledOutcome, followed by exactly one AllDelivered or RunLost, after which the relay exits normally.

What is traded away is stated plainly: push delivery makes the sink’s mailbox the buffer, so the engine’s backpressure ends at the relay. That is the same trade every actor’s mailbox already makes, and the limit still bounds how much work runs at once — only the finished outcomes queue without bound.

The relay is linked to the caller, and the scope to the relay, so the ownership chain survives: a dead consumer takes the relay with it, the scope traps that exit, cancels, drains its owners, and exits.

Examples

let relay =
  weft.new_prepared(tasks)
  |> weft.start_relayed(to: sink)
// The actor's selector now receives weft.Pulled(a, e) messages.

Start a run whose only report is the scope’s exit, and hand back the scope’s pid.

Nothing is delivered to anyone: each outcome is discarded the moment it is sealed, and its slot returned. What remains is the part a drain witness needs — the scope is alive exactly while any worker, any adopted owner, or any cancel helper is, and it exits normally only if every proof landed. This is the shape for a run started purely to witness work: the caller monitors the returned pid, cancels through a signal or cancel_when_exits, and reads the verdict off the DOWN.

The scope is linked to the caller, as every scope is: a dead caller cancels the run, the owners are asked to stop, and the scope drains before it exits. Use cancel_when_exits to name a consumer that is not the caller.

Examples

let scope =
  weft.new_prepared([weft.managed(run_request)])
  |> weft.cancel_when_exits(consumer)
  |> weft.start_witnessed

let watch = process.monitor(scope)
// A normal DOWN proves everything the request started is gone.

Prepare a plain task: no owner, no drain obligation, exactly the behaviour of a closure given to new.

Examples

let outcomes =
  weft.new_prepared([weft.task(fn() { Ok(1) })])
  |> weft.start

The values of the tasks that returned Ok, in the order given.

Examples

let outcomes = [weft.Completed(0, 1), weft.Abandoned(1), weft.Completed(2, 3)]
assert weft.values(outcomes) == [1, 3]