e2 · workers

A service key is the same key on either side of a worker

A program that does yield* Heavy knows the contract and nothing else, so it cannot tell whether the implementation runs inline or on another thread. Moving work off the main thread is a change to the provider list. e2 has no worker API for this: the pattern is a provider that forwards calls, and examples/worker is a complete one.

The idea

One contract, three implementations, one program

The contract is shared by both sides. One side has the real implementation; the other has a provider that forwards named methods.

The contract

An interface and a key, as always

export interface HeavySvc {
  readonly digest: (input: string, rounds: number) => Fx<string, HeavyError>
  readonly countPrimes: (limit: number) => Fx<number, HeavyError>
}
export const Heavy = service<HeavySvc>()('Heavy')

// The program never mentions workers, messages or threads.
const fingerprint = fx(function* (input: string, rounds: number) {
  const heavy = yield* Heavy
  const digest = yield* heavy.digest(input, rounds)
  return `${input.length}:${digest}`
})

The providers

Choosing a thread is one entry in the list

on a workerproviders.ts

export const HeavyOnWorker = Heavy.make(function* () {
  const link = yield* openLink(new URL('./heavy.worker.ts', import.meta.url))
  return {
    digest: (input, rounds) =>
      call(link, 'digest', [input, rounds], decodeHeavyError),
    countPrimes: (limit) =>
      call(link, 'countPrimes', [limit], decodeHeavyError),
  }
})

inline, or a stubproviders.ts

export const HeavyOnMainThread = Heavy.make(function* () {
  return {
    digest: (input, rounds) => sync(() => fnv(input, rounds)),
    countPrimes: (limit) => limit < 2
      ? err(new HeavyError(`limit must be at least 2…`))
      : sync(() => sieve(limit)),
  }
})

export const HeavyStub = Heavy.of({ /* … */ })

await run(fingerprint('hello', 2_000), [HeavyOnWorker])       // the main thread stays responsive
await run(fingerprint('hello', 2_000), [HeavyOnMainThread])   // the same answer, blocking
await run(fingerprint('hello', 1), [HeavyStub])               // '5:stubbed'

The example's tests assert exactly that: the same program produces the same value either side of the boundary, the event loop keeps turning while the worker computes and stops dead while the inline version does, and the two real implementations agree on failures as well as on values. Swapping providers is only transparent if the failure paths match too.

What crosses

Service calls cross; effects do not

An effect is a description, but it is a description made of closures, so it is not structured-cloneable. What crosses is a named call with cloneable arguments, and what comes back maps onto the three ways an effect can end:

Message	Direction	Becomes
`{ kind: 'call', id, method, args }`	to the worker	a handler run with its own `AbortController`
`{ kind: 'cancel', id }`	to the worker	that controller aborted
`{ kind: 'ok', id, value }`	back	the value
`{ kind: 'err', id, error }`	back	an expected failure, decoded into its class, in the error channel
`{ kind: 'defect', id, message }`	back	a `Die`: a bug in the worker is a bug here too

Structured clone drops prototypes, so an error thrown in the worker arrives as plain data. decodeHeavyError rebuilds it; without it, the caller would hold an object that fails instanceof. That function is the one piece of the protocol a contract cannot express, and it is short enough to review.

Cancellation and lifetime

Interruption crosses the thread; the worker dies with the scope

call is an async_ whose cleanup posts cancel. Interrupting the caller, by interruptFiber, a lost race or a timeout, aborts the computation in the worker. Without it a core would keep spinning on work nobody wants.
The worker's handlers are time-sliced. A worker in a synchronous loop never reaches its event loop, so a cancel would sit undelivered. Yielding to the macrotask queue every few milliseconds is what lets it land.
openLink is an acquire. The worker is terminated when the runtime's scope closes, on success, failure or interruption alike. A runtime keeps one worker across many calls; nothing has to remember to call terminate.

const program = gen(function* () {
  const heavy = yield* Heavy
  const abandoned = yield* fork(heavy.digest('x'.repeat(64), 10_000_000))
  yield* sleep(20)
  yield* interruptFiber(abandoned)            // the worker stops computing
  return yield* heavy.digest('hello', 1_000)  // answered at once, not queued behind it
})

Designing for a boundary

Contracts that can cross

Because a service is looked up by a name and its members are reached by name, the forwarding provider can be generic: one Proxy that turns every member into a call, one handler table built from the real service on the worker side. That is application code, not core, and it works for any contract whose members have one of three shapes:

Shape	Example	Crosses as
an `Fx` value	`count: Fx<number>`	a zero-argument call
a method of cloneable arguments returning an `Fx`	`get(threadId, id): Fx<…>`	a call
a method whose last argument is a callback and which returns `Fx<never, E>`	`forThread(id, onMessage): Fx<never, E>`	a stream: each callback invocation is a message, interrupting the caller ends it

A plain value, a method returning a plain value, or a callback anywhere but last cannot cross. Those are usually worth changing anyway: a subscription that returns an unsubscribe function becomes an effect that never completes, which fork starts and structured concurrency ends.