Servers, workflows, and agents

A process with a mailbox, addressed by name, is what Erlang calls an actor, and actors are a well-proven way to build servers.

The previous page put chat rooms on a server as plain processes. This page adds what backends need beyond that: state that survives a crash, timers that survive a restart, agents that call tools and wait for people, and queues.

Losing a chat room when the server restarts may be fine. Losing an order halfway through checkout is not. @nonchalant/durable wraps a process so it can recover. durable(proc, options) returns an ordinary process, so it can be registered, reached over a connection, and bound to a view like any other.

The process inside keeps the usual shape, but recovery doesn't resume a paused generator: it starts the generator again from the last saved state and redelivers the message that was in progress. So the body starts from d.restored, runs its side effects through d.step, and keeps the steps of each message in the same order so they can be matched up on replay. Anything it keeps in a local variable and doesn't yield is gone after a restart.

const order: DurableProc<Order, OrderMsg, { id: string }> = async function* (self, _args, d) {
  let s: Order = d.restored ?? { status: 'open', total: 0, charged: 0 }   // the last saved state
  yield s
  for await (const msg of self) {
    switch (msg.type) {
      case 'add':
        s = { ...s, total: s.total + msg.price }
        break
      case 'checkout': {
        if (s.status !== 'open') continue
        const receipt = await d.step('charge', (key) => payments.charge(s.total, { key }))
        s = { ...s, status: 'charged', charged: receipt.amount }
        break
      }
    }
    yield s
  }
}

const orders = registry({
  order: define(durable(order, { store, key: (a: { id: string }) => a.id })),
})

Here is what gets saved, and when:

1. Each message is saved before the process handles it.
2. Each d.step result is saved as soon as the step completes.
3. When the process asks for its next message, its new state and its position in the message log are saved together, in one transaction.

Published is not saved. A yield reaches the page as soon as it happens, but it is saved only at step 3. If the server dies in between, the page may briefly show a state that recovery then rolls back and rebuilds. A caller that needs to know something was saved should use call: an answer is sent only after it is saved. The same rule means an answer waits for the end of its handler, so reply and finish rather than reply and then wait.

If the server dies partway through a message, recovery loads the last saved state and delivers that message again. The handler runs again, but a step that already completed returns its saved result instead of running, so the card isn't charged twice. A step that was interrupted mid-flight does run again. For that case each step gets a stable key to pass to the outside service, as payments.charge does above, so the service can recognize the retry.

sequenceDiagram
  participant C as caller
  participant D as durable wrapper
  participant S as store
  participant P as your process
  C->>D: cast, or call with a callId
  D->>S: append(message)
  D->>P: deliver
  P->>D: step("charge")
  D->>S: putStep(result)
  P->>D: yield state
  P->>D: request the next message
  D->>S: commit(state, cursor) - together, or not at all
  Note over D,S: a crash before that commit delivers the message again
  Note over D,S: saved steps answer from the journal instead of running

Durable execution is usually done in one of two ways, each with a cost. Nonchalant makes a different trade:

Calls that survive retries

A call to a durable process carries an ID, and the answer is saved under it, so retrying with the same ID returns the saved answer instead of doing the work twice. Inside a durable process, d.call makes the ID for you from the caller, the current message, and the call's name, so a replay after a crash asks with the same ID:

const receipt = await d.call('reserve', (callId) =>
  vault.call({ type: 'reserve', amount: 100, callId }))

When a server hosts durable processes for many users, its principal option gives each user's call IDs their own namespace, so one client can't collide with, or read, another's saved answers.

Timers that survive restarts

d.sleep(name, ms) saves its deadline instead of holding a timer. If the process has been evicted, or the server restarted, by the time the deadline arrives, a scheduler asks the store which processes are due and looks each one up. The lookup restores the process, which finds its deadline passed and carries on.

await d.sleep('cool-off', 3_600_000)   // the deadline is saved, not the timer

const timers = scheduler({ store, wake: (id) => orders.lookup('order', { id }) })

Two schedulers sharing a store won't both wake the same process, and a wake-up that fails before saving comes due again.

Storage is yours

The wrapper talks to storage through an interface of eight methods, so it isn't tied to any database. The package ships an in-memory store for tests and development. An adapter for a real database proves itself by passing @nonchalant/durable/conformance, the storage contract written as a test suite. examples/durable-sqlite does that for Node's built-in SQLite, then crashes a process mid-message, reopens the file, and checks that the charge wasn't repeated.

If two servers ever activate the same order, every write carries an epoch number, and the older activation's writes are refused instead of overwriting the newer one's. Deciding which server should own an order in the first place (a lease, a consistent hash) is left to you.

Not included: a worker fleet, cron, retry policies beyond restart and maxAttempts, or an execution dashboard. The scheduler wakes saved deadlines and nothing else.

An agent takes a question, calls tools, reads the results, and reports progress as it goes. Each of those is something a process already does, so an agent is a process. The one below is also durable.

Tools are processes, reached with call(). Their state is visible like any other (the search tool counts its calls), and the registry can put a tool in the same thread or across a connection.

Human approval is a call that hasn't been answered yet. Type refund 20. The approval tool keeps the request, with its reply, until you click. The agent is simply awaiting its call, and the list of waiting requests is process state, shown by an ordinary view:

const approvals: Proc<{ pending: number }, ApprovalMsg, void> = async function* (self) {
  let waiting: Request[] = []
  yield { pending: 0 }
  for await (const msg of self) {
    switch (msg.type) {
      case 'request':
        waiting = [...waiting, msg]            // the message carries its own reply
        break
      case 'decide': {
        const [head, ...rest] = waiting
        if (head === undefined) continue
        head.reply(msg.ok)                     // the agent's await resolves here
        waiting = rest
        break
      }
    }
    yield { pending: waiting.length }
  }
}

A crash doesn't lose the run. Press kill it while a question is running. The agent's process is disposed, and the next lookup starts a replacement that recovers from its journal without asking the model again.

Tests need no model. The model is passed in as an argument, so examples/agent/agent.test.ts runs the loop with a stub, no DOM, and no network, and checks the states it yields.

Several agents

In the multi-agent example, a supervisor sends a brief to a researcher, passes the result to a writer, and asks a person for approval before publishing. The usual orchestration patterns turn out to be ordinary process code:

flowchart TB
  V["the page: bindings on the supervisor's state"] --- B
  B["supervisor (durable)"] -- "d.call research" --> R["researcher (durable)"]
  B -- "d.call write" --> W["writer (durable)"]
  B -- "d.step approve" --> A["approvals: a reply parked until a person clicks"]
  R -- "call" --> SR["search tool"]
  R -- "call" --> M["budget: one limit for the whole system"]
  W -- "call" --> M
  W -- "call" --> MO["the model"]

Backends often need publish/subscribe and work queues. The messaging example defines both behind interfaces, with in-memory versions.

A subscription can itself be a process: it subscribes when it starts, forwards each event to its own mailbox with self.cast, and unsubscribes when it's disposed. That turns an external stream into state a view can read, and looking it up by topic lets every reader of a topic share one subscription.

A queue worker is a process too: it takes a job with a time-limited lease, handles it, and acknowledges it, or releases it on failure. In the example, press kill while a worker holds a job. The job is never acknowledged, so its lease runs out and another worker picks it up, with its attempts count raised by one.

Read more

Processes on the server covers durability in full, including what happens to a message that keeps crashing its process and how to change the shape of saved state. Hosting safely covers putting a host on the internet.