process.ts: the runtime

1. Two sources per process
2. The mailbox
3. The drive loop
4. Ownership
5. Dispose ordering
6. The outside face
7. Instrumentation
8. Tests

packages/core/src/process.ts and four collaborators. Imports graph.ts and reconcile.ts; imported by registry.ts and index.ts. This is where an async generator becomes a running thing with state, a mailbox, a lifetime, and children.

Every write to the scope pointer is in scope.ts, and every emit site is in process.ts; the other files hold no per-process state.

spawnProcess builds five collaborating pieces:

flowchart TD
    subgraph handle["the handle (outside)"]
        READ["read()<br/>+ pending / stale / error<br/>+ cast / call / iterate / dispose"]
    end
    subgraph runtime["the runtime"]
        MB["Mailbox&lt;In&gt;<br/>FIFO queue + parked takers"]
        DRIVE["drive()<br/>the resume loop"]
        SRC["source: values"]
        META["source: pending/stale/errored"]
        CORE["ProcessCore<br/>children, dispose, settled"]
    end
    GEN[["your async generator"]]

    READ -->|cast / call| MB
    MB -->|for await| GEN
    DRIVE -->|g.next| GEN
    GEN -->|yield| DRIVE
    DRIVE -->|publish| SRC
    DRIVE -->|publish| META
    SRC --> READ
    META --> READ
    DRIVE --- CORE

Every yield goes through a graph source. The local update path therefore matches the wire path: reconcile runs on either type of yield (when the source has readers or a sink wants the ops; see publication.md). That is why a remote process behaves like a local one, and why remote reads are as fine-grained as local ones.

Two sources per process

Values and lifecycle are separate sources, so a reader that only watches pending doesn't wake on every value, and vice versa.

meta holds { pending, stale, errored }. setMeta merges through nextMeta, which hands back the current object when no field changed, and returns early on that identity, so no-op transitions publish nothing.

The watcher count the registry refcounts is the sum of both sources' watchers (valueWatchers + metaWatchers); either kind of subscription keeps a shared process alive.

error is a closure variable rather than part of the published metadata. The getter reads meta().errored purely to establish the subscription, then returns the raw error. Errors are arbitrary values, not Json.

The mailbox

mailbox.ts. A FIFO queue plus a FIFO of parked takers. Both are Fifos: an array with a moving head index that compacts once the consumed prefix reaches half the array, so a dequeue is O(1) amortized at any depth (Array#shift is O(n) and made a 100k-deep drain quadratic; process.lifecycle.test.ts drains 200k queued casts under a time bound). A dequeued slot is cleared at once, so a consumed message is not retained by the backing array (process.leaks.test.ts). Delivery rules:

• A push with a non-latest taker parked hands the message over directly.
• A push with a latest taker parked queues it and schedules a microtask drain, so same-tick casts can supersede each other before the taker sees one. Without that deferral, latest() would hand over the first message of a burst instead of the newest.
• take(latest) with a non-empty queue either shifts one message or drains to the newest (dropping the rest through onDrop).
• take on an empty queue reports idle (pending: false) and parks.
• An iterator's return() withdraws its parked taker, which ends instead of taking the next message. A wrapper reading its own mailbox beside the generator (durable() does) relies on this to let go when its instance ends, so the first message after a crash is not taken and requeued last (mailbox.test.ts, "Self iterators").

bound (mailbox: n) drops the oldest message on overflow, warns once, and routes the dropped message through onDrop. Drop-oldest, not drop-newest, ensuring that the latest input survives sustained overload.

onDrop is also how a dropped call rejects rather than hanging forever: every in-flight call is registered in pendingCalls keyed by its message object, so dropping that object rejects its promise (rejectDropped in calls.ts).

close() resolves parked takers as done and drops the queue, ending the generator's for await and rejecting every call still queued. It takes an optional wrapper around the taker resolution; disposal uses it to resume the body in scope (see Ownership).

The drive loop

stateDiagram-v2
    [*] --> running: spawn
    running --> running: yield → publish → meta{pending:false, stale:false}
    running --> done: generator returns
    running --> crashed: throw, no restart budget
    running --> restarting: throw, restart 'on-crash'
    restarting --> running: fresh AbortController<br/>new generator from the same args<br/>same mailbox; queued casts replay
    running --> disposed: dispose()
    done --> [*]
    crashed --> [*]
    disposed --> [*]

Each iteration awaits g.next() inside step() and hands the value to publish, which reconciles it into the value source and clears the metadata flags. r.done ends the loop. A process that returns is over, and its children go with it. That is why a view process that owns state must idle on its mailbox instead of returning.

The call proc(self, args) sits inside the loop's try: a throw while binding the generator's parameters (a destructuring default, say) is a crash like any other, not a process stuck at pending with an unhandled rejection.

Every exit from running goes through transition(to, meta): it sets the phase, publishes the final metadata, and ends every open async iterator. That is the only place the phase changes, and the loop publishes a value only while the phase is running. A body that was parked at a foreign await when it was disposed may still reach a yield on its way out; that value is dropped, so it cannot flip stale back to false. Iterators end through the closer list rather than by watching metadata, so a process that returns while already idle (no metadata change at all) still ends its iterators.

On a throw (and only while running), crash records the error, aborts the signal, removes queued calls from the mailbox (retainCasts), rejects every pending call (rejectAll), and disposes the crashed instance's children. Then it either restarts (restarts < maxRestarts, default 3; crash returns true and the loop runs the generator again) or settles as crashed with stale: true, errored: true. Readers keep the last good value throughout because a crash makes the value stale rather than empty. The phase, not the error value, says whether the process crashed, so throw undefined is a crash like any other.

Restart is the Erlang position: re-run from the init args, not from the crashed state. The mailbox survives, so queued casts replay into the fresh instance. Calls do not: every call is registered in pendingCalls under its message object, and on a crash the mailbox is filtered against that map before the calls reject, so a call queued behind the crash is never executed by the restarted instance (process.lifecycle.test.ts).

Observing crashes

onProcessError(handler) (in instrument.ts, exported from the package root) installs one global handler that sees every crash, including those a restart recovers from, as (error, name), where name is the proc's function name. It returns a remover. There is no default handler; without one, a crash is visible only on the handle (error, stale) and in rejected calls, as before. The handler runs a microtask after the crash, so a handler that throws surfaces as an unhandled rejection instead of wedging the drive loop. One slot rather than a listener set keeps it cheap; fan out in userland if you need more.

Ownership

currentScope is an ambient module-level pointer in scope.ts. step() sets it only around a resumption, through withScope:

export function withScope<T>(scope: ProcessCore | null, fn: () => T): T {
  const prev = currentScope
  currentScope = scope
  try {
    return fn()          // a resumption returns at the generator's first await/yield
  } finally {
    currentScope = prev
  }
}

const step = (g: AsyncGenerator<T>) => withScope(core, () => g.next())

fn() returns as soon as the generator body hits its first await or yield, so the scope covers exactly the synchronous window of that resumption. A bare spawn after an intervening await in the same step runs unowned.

That window is why self.spawn exists. It does not consult the ambient scope: each instance's Self carries a spawn that runs withScope(core, () => spawnProcess(...)), so the child attaches to this process from anywhere in the body. The instance is identified by the restart count it started with; a spawn from an instance that has crashed (and been replaced), or from any instance after drive() has reaped its children (reaped), disposes the child at once rather than adopting it into a lifetime that is already over. channel()'s spawn is the same idea with the channel as owner: children are tracked in a set, removed as they settle, and disposed when the channel ends. durable() forwards its outer self.spawn unchanged, so a durable body's children belong to the hosting process.

The ambient window remains for a bare spawn, and its rule is one line: spawn before awaiting, or use self.spawn. unscoped(), which is withScope(null, fn), is the explicit escape hatch: the registry wraps every spawn in it so shared state is never owned by whichever caller happened to look it up first.

Spawns during teardown. The one resumption the runtime extends the scope to is the one disposal causes by closing the mailbox. A body parked on its mailbox (the usual place) resumes when close() resolves its taker, and that resumption is a promise reaction outside any step(). Disposal therefore passes resumeWithin(core, resolveTakers) as the mailbox's close wrapper, which brackets the resolution with two microtasks, scope on and scope off. Microtasks run FIFO, so the body's resumption runs between them, and a finally that spawns before its own first await attaches the child to the dying process. drive() disposes it with the other children, and asyncDispose waits for it. A finally reached after a foreign await (a fetch, a timer) is not covered for a bare spawn; self.spawn there is owned, because reaped is set only after the generator has settled.

Dispose ordering

The order is a contract, not an implementation detail:

sequenceDiagram
    participant U as caller
    participant P as process
    participant G as generator
    participant C as owned children

    U->>P: dispose()
    P->>P: transition(disposed): meta, iterators end
    P->>P: detach from parent
    P->>P: 1. mailbox.close(), resume in scope
    Note over P,G: the body's for-await ends,<br/>queued calls reject
    P->>P: 2. controller.abort()
    P->>G: 3. g.return() inside step()
    Note over G: finally blocks run
    G-->>P: generator settles
    P->>P: gen = null, rejectAll
    P->>C: 4. disposeChildren()
    C-->>P: settled
    P->>U: completion resolves (asyncDispose)

Symbol.dispose starts teardown synchronously and returns; it cannot make an awaited promise settle. Symbol.asyncDispose awaits completion, which resolves only after the generator has settled and every owned child's finalizer has settled (await Promise.all([...childSettlements])).

An operation that ignores self.signal and never settles will therefore block async disposal because cancellation is cooperative.

Disposing an already-finished process still runs the teardown that is left: mark stale (through transition) and dispose children.

The outside face

The handle is the read function, with everything else installed onto it:

const read = (): T | undefined => src() as unknown as T | undefined
Object.defineProperties(read, { pending: …, stale: …, error: … })
p['cast'] = …; p['call'] = …
p[Symbol.asyncIterator] = …; p[Symbol.dispose] = …; p[Symbol.asyncDispose] = …

Calling the handle is a source read, so it is tracked inside a derive, effect, or binding and a snapshot read anywhere else. The process handle inherits path precision for free from graph.ts.

call(msg) builds { ...msg, reply }, registers the rejector under that exact object, and pushes it. The generator sees a message carrying reply; calling it deletes the entry and resolves. Calls on a non-running process reject immediately.

Symbol.asyncIterator returns an independent, lossy subscription: one buffered slot, overwritten by newer values, deduplicated with Object.is. It subscribes to the value source as a subtree dependency (reading the whole object, letting it escape), so it wakes on every yield. Latest-value delivery is the default for state synchronization and is also what the wire host needs, because patches computed between consecutively observed snapshots always compose.

The iterator is iterate() in iterate.ts, shared with derive. The owner passes a pull function and a set of closers; transition calls every closer when the process leaves running. A closer pulls once more, untracked, before ending, so a value yielded just before the process returned is delivered before done even if the effect flush has not run yet. An iterator opened on a finished process yields the last value, then done. Concurrent next() calls queue and settle in call order.

Instrumentation

instrument(sink) stores one module-level sink in instrument.ts (a second call replaces the first; the remover only clears its own); process.ts reads that live binding at each site. Every event site is written sink?.({ ... }). An optional call short-circuits its arguments, so with no sink the event object is never built and, for yield, the extra reconcile(untracked(src), next) never runs. That diff is the price of inspecting: the graph's publish diffs the same pair but keeps its patch private, and graph.ts stays out of this. instrument.test.ts pins it by counting how often a yielded record's keys are listed: once without a sink, twice with one.

Where the events come from:

• spawn: after the ownership link, before drive() starts, so it precedes everything the body does, including the spawns of its first step. id is a module counter (lastId), assigned for every spawn so an owner's id is known even when the sink was installed after the owner spawned. The registry passes its definition name through internal.key.
• cast: post, which the handle's cast and self.cast share.
• call and reply: the handle's call numbers the request from lastCall (only while a sink is installed; 0 otherwise) and the reply closure reports under the same number.
• yield: inside publish, before the source publishes, only while running.
• status: in setMeta, after the no-change check.
• crash and restart: in crash, the crash first.
• exit: in transition, before the final setMeta, so the reason precedes the flags it causes.

The sink runs synchronously inside the runtime, so it must not throw, and whatever it casts is itself reported re-entrantly. @nonchalant/inspect handles that by skipping the processes it owns (see inspect.md).

Tests

mailbox.test.ts (Fifo order and compaction boundaries, the delivery rules without a process), scope.test.ts (which scope is current, when, and for how long, including the resumeWithin bracket), process.helpers.test.ts (validateSpawnOpts accept/reject tables, nextMeta, and the reply bookkeeping in calls.ts), process.test.ts (lifecycle, mailbox order, latest() conflation, crash and restart, ownership, call rejection paths), process.lifecycle.test.ts (how a process ends: iterators after return and crash, queued calls across a restart, yields after dispose, teardown spawns, parameter-binding throws, onProcessError, mailbox depth), instrument.test.ts (the exact event sequence, ids and owners, the cost with no sink), process.leaks.test.ts (nothing retained after disposal. This test needs gc({ execution: 'async' }), since plain gc() false-fails under V8 conservative stack scanning), types.check.ts (the @ts-expect-error lines are regression checks for the type surface).

Next: registry.md covers naming, sharing, and eviction.

This page's source on GitHub