A long-running job — progress, cancel, approval, a partition, and a crash

How it works

Press start import. A job writes twelve records into a destination, about one a second. Three of them already exist there, so the job stops and asks before it overwrites them. While it runs, pull the client off the network, kill the worker, or both, and watch what each side does.

Everything here runs in this tab. The worker is a registry holding one durable() process, served with expose. The page reads it with connect, like a job on a server. The cable between them is an in-memory transport that can be cut. The journal (a memoryStore) and the destination live outside the worker, so they survive a kill, the way a database outlives the machine that talks to it.

One record is one message

The durable wrapper commits once per message: when the process asks for its next message, the state it produced, the cursor past the message it handled, and any call answers are written together. So the job handles each record as its own next message and casts the next one to itself. A crash then costs only the record in flight.

case 'next': {
  if (s.status !== 'running' || msg.run !== s.run || msg.at !== s.at) continue // a repeated token
  const item = s.items[s.at]
  if (item.overwrites && s.approved === null) {
    s = { ...s, status: 'waiting' }      // nothing is cast: the decision resumes the job
    break
  }
  s = mark(s, s.at, 'writing')
  yield s                                // published now; committed only with this message
  const receipt = await d.step(`write ${item.id}`, (key) => upload(site, ref, key, self.signal))
  s = advance(mark(s, s.at, 'written', receipt.key))
  if (s.status === 'running') self.cast({ type: 'next', run: s.run, at: s.at })
  break
}

Kill the worker

A restart does not resume a suspended generator. The new activation starts from the last committed snapshot and replays the messages the journal holds past the cursor. The write is a d.step, so its result is journaled when it lands, and a replay returns that result without calling the destination. Each row shows two counts kept by the destination: how often the write was attempted, and how often a row was written.

Killed while a record is at work: its write never reached the destination. The replay runs it again, so that row shows ran 2× · wrote 1×.
Killed after the write landed but before its answer was journaled: the step runs again under the same idempotency key (import#<seq>#0: the job's key, the message's sequence number, the step's position, all stable across replays). The destination has already applied that key and answers with the first receipt: ran 2× · wrote 1× · deduped 1×.
Records that finished before the kill are not attempted again.

Nothing on the worker holds the job open. A restarted worker starts a scheduler on the same store, and the store marks a key due once a message has gone unacknowledged for redeliverAfter (three seconds here, about three records). The interrupted message is overdue, so the scheduler wakes the job whether or not anybody looks it up: kill, disconnect, and restart, and it carries on unwatched. A finished job has nothing overdue and stays asleep in the store.

A step runs at least once and is recorded exactly once. The effect runs exactly once only because the destination refuses a key it has already applied. The key is what makes that refusal possible.

A yield is published before it is committed. The "writing" and "written" states you watch are yields from inside a message. If the worker dies before the process asks for its next message, the record can go back to queued when the worker restarts, then forward again as the message replays. job.test.ts pins this with a store that holds a commit back.

Approval is a call

At the gate the job yields waiting and finishes its message, so waiting is committed state. Kill the worker there and it restarts still waiting, with nothing running. Approve is a call() with a callId that names the decision. Its answer is committed with the message and released only after that commit lands. A retry with the same id gets the recorded answer, even if it asks for the opposite.

case 'decide':
  if (s.status !== 'waiting') {
    msg.reply({ ok: msg.ok, accepted: false })
    continue
  }
  s = { ...s, status: 'running', approved: msg.ok }
  msg.reply({ ok: msg.ok, accepted: true })   // released once this message commits
  self.cast({ type: 'next', run: s.run, at: s.at })
  break

Cancel

Cancel is a message, so it waits in the mailbox behind the record in flight. That record finishes and is recorded, and no record after it starts. A cancel never interrupts a write halfway. Pressed while the client is disconnected, it is queued by the client and sent after the reconnect.

Disconnect the client

The job keeps running, because nothing about it depends on the page: the worker's registry keeps it while the worker lives, and its scheduler resumes it after a restart. The page's view goes stale and keeps the last state it received (the attempt counts beside the rows still move, because the page reads the destination directly). On reconnect the client looks the job up again. The host answers with a full snapshot, which is diffed against the retained value, so only the rows that changed are touched. The approve buttons are disabled while stale: a call made while disconnected rejects at once rather than disappear.

// the worker: a registry, the durable job in it, served over the cable
const run = durable(job, { store, key: (a: { id: string }) => a.id, redeliverAfter: 3000 })
const reg = registry({ job: define<JobState, JobMsg, { id: string }>((self, args) => run(self, { ...args, site })) })
scheduler({ store, wake: (id) => reg.lookup('job', { id }) })   // resumes overdue work at boot
expose({ lookup: reg.lookup, principal: 'reader' }, cable.host)

// the page: a lookup like any other, from the other end
const remote = connect<JobSchema>(cable.client).lookup('job', { id: 'import' })
remote.call({ type: 'decide', ok: true, callId: `gate-${run}` })

The host moves every callId into its session's principal namespace. The stable principal above means a decision retried after a reconnect lands on the same record.

The log under the demo is the cable itself: every lookup, cast, call, patch, and reply, decoded with the wire's own codec. A record changing state crosses as a patch of a few ops, never the whole list; only a lookup gets a full snapshot.

job.test.ts drives all of this with a clock the test moves by hand. It asserts exact counts: effects once across a kill mid-job, a pending approval surviving a kill, cancel stopping after the record in flight, a partition going stale and converging, and a published state being taken back by a crash.