Recipes

These recipes build common patterns from the public primitives. Most have a runnable version in examples/.

Widget state with `cell`

function cell<T>(initial: T): Process<T, T> {
  return spawn<T, T, void>(async function* (self) {
    for await (const next of self) yield next
  }, undefined, { initial })
}

cell is a convenience wrapper for this common pattern. A cell created inside a view belongs to that view and is disposed with it. See examples/counter.

A reducer instead of a loop

When a process only folds messages into state, with no awaiting and at most one new state per message, reducer from @nonchalant/core writes it as a function. It is optional sugar and compiles to the same Proc as the loop:

export function cartStep(s: CartState, msg: CartMsg): CartState {
  switch (msg.type) {
    case 'add':
      return priced([...s.items, msg.item])
    case 'remove': {
      const items = s.items.filter((it) => it.name !== msg.name)
      return items.length === s.items.length ? s : priced(items)
    }
    case 'checkout':
      msg.reply({ ok: true, charged: s.total })
      return s.items.length === 0 ? s : priced([])  // the same state: no yield
  }
}

export const cart = reducer((_: { userId: string }) => priced([]), cartStep)

cartStep tests as a plain function, and cart works unchanged with define, connect, and durable. See examples/shared-cart, and Layers for what the sugar covers and when to go back to a generator.

Typeahead with the latest queued input

self.latest() reads the mailbox in "skip to newest" mode: while a search is in flight the loop isn't listening, and when it comes back it picks up the most recent query, ignoring everything in between. self.signal cancels the in-flight request if the process is disposed.

for await (const { q } of self.latest()) {
  yield { q, results, pending: true }
  try {
    results = await api.search(q, { signal: self.signal })
    yield { q, results, pending: false }
  } catch { /* aborted or failed; keep going */ }
}

This prevents a backlog, but it does not cancel the request already in flight. That request completes and publishes its yield before the newest queued query begins, which is why each yield carries its own q. If even a transient stale result matters, don't await in the loop: stamp each request with an epoch, run it in the background, self.cast the result back, and ignore results whose epoch is old. examples/typeahead.

Forms that wait for a result

Make the submit a Call message. The form gets a typed promise for its own result instead of watching status fields go by. examples/form.

A small query cache

const users = registry({
  user: define(async function* (self, { id }: { id: number }) {
    yield await fetchUser(id)
    for await (const _ of self) void _      // stay alive while anyone watches
  }, { evict: 30_000 }),
})
users.lookup('user', { id: 1 })   // deduped by key, shared, evicted when idle

Lookups with the same key share one fetch. The entry is removed 30 seconds after its last watcher leaves, and the next lookup starts a new fetch. packages/core/test/registry.test.ts verifies this lifecycle.

examples/query adds loading and error states, retries, stale-while-refetch, and mutations implemented as call() on the query. The example uses write-through updates instead of an invalidation graph, with the schema tested headlessly in examples/query/shop.test.ts and the reasoning written up right on the demo page.

Routing with a URL process

The current route is state like any other, and pages can be view processes: the router iterates the route, disposes the outgoing page (tearing down its whole scope), yields a loading state, await import()s the chunk, and yields the new page. Navigation, loading states, code splitting, and teardown in a small amount of process code. The router includes the route process, navigate, and spreadable link() attributes that replace history by default. It is defined in examples/lib/router.ts and used by examples/router. It ships in two flavors, hashRouter and historyRouter (real paths via pushState/popstate; needs history-fallback hosting).

Undo and redo by wrapping a process

A process is a function, so middleware is function composition: wrap a proc, give the inner one a private channel, keep the history in the wrapper. The same shape gives you logging, persistence, and replay. examples/undo-redo; a variant that groups a whole slider drag into one undo step is in examples/7guis/circle-drawer.ts.

A process for a drag gesture

Start a process on pointerdown, yield offsets while the pointer moves, and dispose it on pointerup. Disposal also removes the gesture's subscriptions.

el.addEventListener('pointerdown', (down: PointerEvent) => {
  const gesture = spawn(async function* (self: Self<PointerEvent>) {
    for await (const move of self)
      yield { dx: move.clientX - down.clientX, dy: move.clientY - down.clientY }
  }, undefined)
  const stop = effect(() => { const o = gesture(); if (o) applyOffset(o) })
  const onMove = (e: PointerEvent): void => gesture.cast(e)
  const onUp = (): void => {
    removeEventListener('pointermove', onMove)
    removeEventListener('pointerup', onUp)
    stop()
    gesture[Symbol.dispose]()
  }
  addEventListener('pointermove', onMove)
  addEventListener('pointerup', onUp)
})

Runnable version, with a draggable box: examples/drag.

Selecting one row of thousands

A binding wakes when a path it read changes. If every row compares against one selected id, every row read the same value, so changing the selection wakes all of them. Yield a set-shaped object instead and have each row read only its own entry. Changing the selection then removes one key and adds another, which wakes exactly two rows. This is what createSelector does in signal libraries, written as a process:

type Selection = { readonly [id: number]: true }
type SelectMsg = Cast<{ type: 'select'; id: number }>

const selection: Proc<Selection, SelectMsg, void> = async function* (self) {
  let sel: Selection = {}
  yield sel
  for await (const msg of self) {
    switch (msg.type) {
      case 'select':
        sel = { [msg.id]: true }
        break
    }
    yield sel
  }
}

const selected = spawn(selection, undefined, { initial: {} })

tr({ key: row.id, class: () => (selected()[row.id] === true ? 'danger' : '') },
  td({}, a({ onclick: () => selected.cast({ type: 'select', id: row.id }) }, row.label)))

Multi-select uses the same shape: add or remove one key per message. examples/js-framework-benchmark holds it to budget: selecting a row among 1,000 runs two bindings and writes two attributes (bench.test.ts). examples/charts stores a chart's series the same way, as ids plus a record of values, so changing one value writes one SVG attribute (charts.test.ts).

A spreadsheet built from derives

Give every cell a derive that parses its formula and reads the cells it references. Dependency tracking is automatic: edit A1 and only formulas that (indirectly) mention A1 recompute; a formula whose result didn't change wakes nobody downstream; a reference cycle comes back as #CYCLE instead of a hang. examples/7guis/cells.ts, with the test to prove each of those claims.

Using the wire between tabs

One tab wins a Web Lock and hosts over a BroadcastChannel transport; every tab connects as a client. When the hosting tab closes, another tab acquires the lock and takes over. The protocol requires a transport, not a server. examples/multi-tab.

Using the wire with a Web Worker

A transport carries ordered, reliable strings; the port to a worker is one. The worker calls expose(registry({...}), portTransport(workerEndpoint())), the tab calls connect(portTransport(new Worker(...))), and a heavy process runs off the thread that draws while its state arrives as data patches. portTransport takes a port rather than making one, so a MessagePort or worker_threads' parentPort works the same way, and it needs no reconnect story: a port does not drop.

Two things to keep in mind when a process grinds on its own:

Chunk the work and drive the next chunk with a timer, not a bare self.cast. A mailbox loop that casts to itself synchronously resolves in microtasks, and the event loop never turns again, so nothing else is heard.
Yield the small thing. Keep the working set in the process (a plain local array) and let call() fetch it when someone actually wants it; every yield is a diff that has to cross.

examples/worker.

Brokers behind a port

A bus and work queue can each be defined as an interface with an adapter. A subscription process subscribes during setup and unsubscribes on disposal. Looking it up by topic allows the registry to cache one subscription per topic. A queue worker reserves a job under a lease, handles it, and acknowledges it. If the worker dies, the lease expires and another worker can receive the job. See examples/messaging and Processes on the server.

Delegation as a call tree

Make every tool run a process that yields its own trace node, and make an agent a tool whose run spawns one run per tool call. The agent merges its children's streams and folds each snapshot into its node's children, so the root's state is the whole live call tree: one plain value that a view draws recursively and a test iterates.

n = { ...n, status: 'calling', children: [...n.children, ...queued] }
yield n
const runs = calls.map((c) => self.spawn(c.use.run, c.run, { initial: c.node }))
for await (const [i, child] of merge(runs, self.signal)) {
  n = { ...n, children: n.children.with(at + i, child) }
  yield n
}

Two details carry the lifecycle. The runs are spawned with self.spawn, so they are owned and die with the agent even though the loop has already awaited the model. And merge takes self.signal: a disposed process unwinds only once its pending await settles, and its children are disposed after that, so a merge still waiting on them must give up on abort. examples/delegation (delegation.test.ts, "aborts every run in the tree when the root is disposed").

Where are the operators?

Nonchalant does not define a large operator API. Common RxJS and FRP operations map to the existing primitives:

operator	here
`map` / `filter` / `scan` over values	`derive(() => f(p()))` or code in the process loop
`combineLatest`	one derive reading several processes
`fold` / reducers	the `for await` loop is the fold; `reducer` writes it as a function
queued-input conflation	`self.latest()`
`startWith`	`initial`
retry / error channels	`restart` policies, `stale`, `error`

Most of these mappings are short enough to keep near the code that uses them. Two patterns that carry more lifecycle logic are shown below:

// merge: pump several processes into one mailbox
function merge<T>(...sources: Process<T>[]): Process<T | undefined, T> {
  return spawn<T, T, void>(async function* (self) {
    for (const s of sources)
      void (async () => {
        for await (const v of s) {
          if (self.signal.aborted) break
          self.cast(v)
        }
      })()
    for await (const v of self) yield v
  }, undefined)
}

// debounce: yield ms after the source goes quiet
function debounced<T>(source: Process<T>, ms: number): Process<T | undefined> {
  return spawn<T, never, void>(async function* (self) {
    const settled = channel<T>(self.signal)
    let timer: ReturnType<typeof setTimeout> | undefined
    void (async () => {
      for await (const v of source) {
        if (self.signal.aborted) break
        clearTimeout(timer)
        timer = setTimeout(() => settled.cast(v), ms)
      }
    })()
    for await (const v of settled) yield v
  }, undefined)
}

A pump loop notices disposal when the source next produces a value. This is usually sufficient for UI streams. If a source may remain silent indefinitely, call it.return() from an abort listener.

A chat room over the wire

A room is a process that reduces posts into a capped history. Host it with serve, look it up by name from every tab, and the whole client-server chat uses the same process code as local state. Posts are casts, history arrives as patches, a dead server shows as stale: true until the reconnecting transport finds it again. examples/chat.

Switching a registry to a remote host

Replace a local registry with connect(webSocketTransport(url)) when the callers only depend on the registry interface. The process and view can stay the same, but deployment adds JSON boundaries, latency, disconnection, authentication, and authorization. examples/shared-cart shows the code change; Hosting safely covers the operational boundary.

Durable processes with loading, checkpoints, and eviction

Process state lives in generator locals, so durability is a contract you write, not a feature you enable: load before the first yield, checkpoint at transition boundaries, let eviction deactivate.

const accounts = registry({
  account: define(async function* (self: Self<AccountMsg>, { id }: { id: string }) {
    let s = (await store.load(id)) ?? initialAccount   // hydrate on activation
    yield s
    for await (const msg of self) {
      s = step(s, msg)                                 // pure reducer
      yield s
      await store.save(id, s)                          // checkpoint at the boundary
    }
  }, { evict: 60_000 }),                               // deactivate when idle
})

This implements a virtual-actor lifecycle in application code: lookup('account', { id }) activates the process, loading hydrates it, and evict deactivates it. The next lookup starts and hydrates it again. @nonchalant/durable adds message and effect journals plus an atomic commit of state and its cursor; see Processes on the server. Because the mailbox is the single writer, checkpoints are ordered with no application-level locking; the same shape gives event sourcing (append the message instead of saving the snapshot, replay to hydrate). To keep persistence out of the domain code, wrap the proc, using the same wrapping approach as undo and redo.

The mailbox has a single writer only while one node owns its name. If several hosts can activate the same ID, the store needs leases or fencing. The library does not provide that coordination. Hosting safely covers the boundary.

Long batch jobs, one message per item

@nonchalant/durable commits once per message: the state, the cursor past the message, and any call answers are written together when the process asks for its next message (the transaction boundary). A job that loops over N items inside one message commits nothing until the end. A crash at item N − 1 redelivers the whole message: its completed steps come back from the journal, but the restored state is from before item 0, and that one message's step journal holds N records. The loop also never reads its mailbox, so a cancel or an approval waits until the last item. Handle one item per message instead. The job casts itself a next tagged with its run and position, and each item commits on its own:

type Item = { id: string; risky: boolean }

interface Source {
  /** The item at a position. Items stay in their source; the snapshot holds a position. */
  item(at: number): Promise<Item>
  /** Write one item, refusing a key it has already applied. */
  write(id: string, key: string, signal: AbortSignal): Promise<string>
}

type Batch = {
  status: 'idle' | 'running' | 'waiting' | 'cancelled' | 'done'
  run: number                  // each start is a new run
  at: number                   // the next item to handle
  total: number
  written: number
  skipped: number
  approved: boolean | null     // the decision on the risky item at `at`
}

type BatchMsg =
  | Cast<{ type: 'start'; total: number }>
  | Cast<{ type: 'next'; run: number; at: number }>
  | Cast<{ type: 'cancel' }>
  | Call<{ type: 'decide'; ok: boolean; callId: string }, boolean>

const idle: Batch = { status: 'idle', run: 0, at: 0, total: 0, written: 0, skipped: 0, approved: null }

const batch: DurableProc<Batch, BatchMsg, { id: string; source: Source }> = async function* (self, { source }, d) {
  let s = d.restored ?? idle
  yield s
  for await (const msg of self) {
    switch (msg.type) {
      case 'start':
        if (s.status === 'running' || s.status === 'waiting') continue
        s = { ...idle, status: 'running', run: s.run + 1, total: msg.total }
        self.cast({ type: 'next', run: s.run, at: 0 })
        break
      case 'cancel':
        if (s.status !== 'running' && s.status !== 'waiting') continue
        s = { ...s, status: 'cancelled' }
        break
      case 'decide':
        if (s.status !== 'waiting') {
          msg.reply(false)
          continue
        }
        s = { ...s, status: 'running', approved: msg.ok }
        msg.reply(true)
        self.cast({ type: 'next', run: s.run, at: s.at })
        break
      case 'next': {
        if (s.status !== 'running' || msg.run !== s.run || msg.at !== s.at) continue   // a replayed or stale token
        const item = await d.step(`read ${s.at}`, () => source.item(s.at))
        if (item.risky && s.approved === null) {
          s = { ...s, status: 'waiting' }            // nothing is cast: the decision resumes the job
          break
        }
        if (item.risky && !s.approved) s = { ...s, skipped: s.skipped + 1 }
        else {
          await d.step(`write ${item.id}`, (key) => source.write(item.id, key, self.signal))
          s = { ...s, written: s.written + 1 }
        }
        const at = s.at + 1
        s = { ...s, at, approved: null, status: at < s.total ? 'running' : 'done' }
        if (s.status === 'running') self.cast({ type: 'next', run: s.run, at })
        break
      }
    }
    yield s
  }
}

The load-bearing parts:

The token ignores a duplicate. A replayed message runs its self.cast again, so after a crash there can be two next messages for one position. The first one handles the item and moves at. The second no longer matches s.at and is dropped with continue, which means no state change and no yield. The run number does the same for a token left over from a cancelled or finished run.
Progress is the state. at, written, and skipped are what a progress bar binds to, locally or over the wire, and a restart resumes from the last committed position.
Cancel and the gate are messages. They queue behind the item in flight, so they act between items and never in the middle of one. At the gate the job yields waiting and casts nothing, so the waiting is committed and survives a restart. The decision is a call whose callId names the decision (gate-${run}-${at}), so a retried click gets the recorded answer.
The write is a step, and its idempotency key goes to the destination. The item is read in a step as well, so a replay sees the same item even if the source has changed since.

Keep per-run lists out of the snapshot, or keep them small. Every commit writes the whole snapshot, so a state that lists every item (with a status for each) makes each item's commit cost grow with the length of the run. Hold a position and counters, read the items from their source by position, and cap anything you do keep, such as the ids of the last few failures. examples/job keeps its twelve records in state because twelve is small.

Each next here awaits real I/O, so the event loop turns between items. A batch of purely local work should wait on a timer between items instead, as the Web Worker recipe explains.

examples/job is this shape with a destination that counts its own writes. job.test.ts kills it mid-item, after a write landed but before its answer was journaled, and at the gate, and asserts with exact counts that each record is written once and that cancel stops after the item in flight.

Replay-safe local writes

A step executes at least once and is recorded exactly once (what it guarantees). If the process dies after fn has written but before the step's result is recorded, the replay runs fn again under the same idempotency key, key#seq#index. A remote service takes that key and refuses a repeat. A write to your own table needs the same protection when the store can't commit the write and the step's record in one transaction. When it can (a TransactionalStore, such as the SQLite example), use d.local instead; it writes once and needs none of what follows.

A versioned write (compare and set on a version column) is where this breaks. The first attempt moved the row from version 7 to 8. The replay expects 7, finds 8, and reports a conflict. The message throws on every attempt. With maxAttempts it ends up dead-lettered (poison messages). The row was changed, but the process state never recorded that change. Without maxAttempts the message is retried on every activation, and the key never gets past it.

The fix is to store, on each row, the idempotency key of the step that last changed it. A write that finds its own key there is a replay of a write that already landed, and it counts as success:

type Row = { id: string; balance: number; version: number; lastKey: string | null }

interface Table {
  get(id: string): Row | undefined
  /** Store `next` only if the row is still at version `expected`. */
  replace(next: Row, expected: number): boolean
}

declare const accounts: Table

/** Move a balance by `delta`, once per idempotency key. */
function adjust(table: Table, id: string, delta: number, key: string): number {
  const row = table.get(id)
  if (row === undefined) throw new Error(`no account ${id}`)
  if (row.lastKey === key) return row.balance        // this step already landed: success
  const next = { ...row, balance: row.balance + delta, version: row.version + 1, lastKey: key }
  if (!table.replace(next, row.version)) throw new Error(`account ${id} changed underneath`)
  return next.balance
}

type Wallet = { balance: number | null }
type WalletMsg = Cast<{ type: 'debit'; amount: number }>

const wallet: DurableProc<Wallet, WalletMsg, { id: string }> = async function* (self, { id }, d) {
  let s = d.restored ?? { balance: null }
  yield s
  for await (const msg of self) {
    switch (msg.type) {
      case 'debit': {
        const balance = await d.step('debit', (key) => adjust(accounts, id, -msg.amount, key))
        s = { ...s, balance }
        break
      }
    }
    yield s
  }
}

Without the lastKey check, the replay after a crash between replace and the step's record is the conflict described above. With it, the replay returns the balance the first attempt wrote, the step is recorded, and the message commits.

The lastKey column remembers only the last write. That is enough when one process owns the row, because the replay of a message is that process's next write: nothing else can change the row between the crash and the replay. If several writers share a row, another write can replace lastKey before the replay arrives, and the replay would apply twice. For shared rows, record the applied keys in their own table, written in the same local transaction as the row, the way the destination in examples/job keeps applied.

A deadline on `call`

A pending remote call already rejects on crash, completion, disconnect, and dispose. A host handler that never replies can still leave it pending. A deadline is a race, not a protocol feature:

function withDeadline<T>(promise: Promise<T>, ms: number): Promise<T> {
  let timer: ReturnType<typeof setTimeout> | undefined
  return Promise.race([
    promise,
    new Promise<never>((_, reject) => {
      timer = setTimeout(() => reject(new Error(`no reply within ${ms}ms`)), ms)
    }),
  ]).finally(() => clearTimeout(timer))
}

const receipt = await withDeadline(cart.call({ type: 'checkout' }), 5_000)

Reading connection status

Connection state uses the existing process metadata. When the transport drops, each remote handle retains its last value, sets stale, and records the error. A successful re-lookup clears both fields. Because stale and error are reactive, a connection indicator can be one derive:

const status = derive(() => (cart.stale ? 'reconnecting…' : 'live'))

Readers of unchanged paths are not notified during reconnection because the full snapshot is compared with the retained value (packages/wire/test/wire.test.ts, "reconnect is a re-lookup").

This page's source on GitHub