RpcClient.layerProtocolWorker: what's the equivalent of Piscina's abortable tasks? (terminate a stuck worker on interrupt, control respawn)

npm i [email protected] @effect/[email protected]
node main.mjs

rpcs.mjs

import { Schema } from "effect"
import { Rpc, RpcGroup } from "effect/rpc"

export class Rpcs extends RpcGroup.make(
  // Synchronous work that never yields, like jsdom on a hostile page
  Rpc.make("Spin", { payload: { ms: Schema.Number }, success: Schema.Number }),
  Rpc.make("Ping", { success: Schema.Number }),
  // The thread dies mid-request, as it would at its heap limit
  Rpc.make("Crash", {})
) {}

worker.mjs

import { threadId } from "node:worker_threads"
import * as NodeRuntime from "@effect/platform-node/NodeRuntime"
import * as NodeWorkerRunner from "@effect/platform-node/NodeWorkerRunner"
import { Effect, Layer } from "effect"
import { RpcServer } from "effect/rpc"
import { Rpcs } from "./rpcs.mjs"

const handlers = Rpcs.toLayer({
  Spin: ({ ms }) =>
    Effect.sync(() => {
      const end = Date.now() + ms
      while (Date.now() < end) {
        // never yields
      }
      return threadId
    }),
  Ping: () => Effect.succeed(threadId),
  Crash: () => Effect.sync(() => process.exit(1))
})

RpcServer.layer(Rpcs).pipe(
  Layer.provide(handlers),
  Layer.provide(RpcServer.layerProtocolWorkerRunner),
  Layer.provide(NodeWorkerRunner.layer),
  Layer.launch,
  NodeRuntime.runMain
)

broken-worker.mjs

// A worker module that can't load, e.g. a missing native dependency
throw new Error("cannot load this module")

main.mjs

import { Worker } from "node:worker_threads"
import * as NodeWorker from "@effect/platform-node/NodeWorker"
import { Duration, Effect, Exit, Fiber, Layer, Scope } from "effect"
import { RpcClient } from "effect/rpc"
import { Rpcs } from "./rpcs.mjs"

let t0 = performance.now()
const log = (...args) => console.log(`[${((performance.now() - t0) / 1000).toFixed(2)}s]`, ...args)

const threads = []
const spawner = (file) =>
  NodeWorker.layer(() => {
    const worker = new Worker(new URL(file, import.meta.url))
    const n = threads.push(worker)
    log(`spawned thread #${n}`)
    worker.on("exit", (code) => log(`thread #${n} exited (code ${code})`))
    return worker
  })

const makeClient = (scope, file) =>
  Layer.buildWithScope(RpcClient.layerProtocolWorker({ size: 1 }).pipe(Layer.provide(spawner(file))), scope).pipe(
    Effect.flatMap((context) => RpcClient.make(Rpcs).pipe(Effect.provideContext(context), Scope.provide(scope)))
  )

const section = (title) =>
  Effect.sync(() => {
    threads.length = 0
    t0 = performance.now()
    console.log(`\n=== ${title}`)
  })

const program = Effect.gen(function*() {
  yield* section("1. A request that times out leaves its thread running")
  {
    const scope = yield* Scope.make()
    const client = yield* makeClient(scope, "./worker.mjs")
    yield* client.Ping()
    log("Spin(5000) with a 500 ms timeout ->", yield* client.Spin({ ms: 5000 }).pipe(Effect.timeoutOption("500 millis")))
    const before = threads[0].performance.eventLoopUtilization()
    yield* Effect.sleep("1 second")
    const after = threads[0].performance.eventLoopUtilization(before)
    log(`thread #1 utilisation over the next second: ${after.utilization.toFixed(2)}`)
    const [took] = yield* Effect.timed(client.Ping())
    log(`the next Ping took ${Duration.format(took)}, on the same thread, behind the abandoned Spin`)
    yield* Scope.close(scope, Exit.void)
  }

  yield* section("2. Closing the scope while a task is stuck takes NodeWorker's 5 s")
  {
    const scope = yield* Scope.make()
    const client = yield* makeClient(scope, "./worker.mjs")
    yield* client.Ping()
    yield* client.Spin({ ms: 30_000 }).pipe(Effect.timeoutOption("500 millis"))
    log("Spin(30000) timed out; closing the client's scope")
    const [took] = yield* Effect.timed(Scope.close(scope, Exit.void))
    log(`scope closed after ${Duration.format(took)}`)
  }

  yield* section("3. A thread that dies is respawned a second later, with nothing waiting")
  {
    const scope = yield* Scope.make()
    const client = yield* makeClient(scope, "./worker.mjs")
    yield* client.Ping()
    log("Crash ->", (yield* Effect.exit(client.Crash())).toString().slice(0, 120))
    yield* Effect.sleep("2500 millis")
    log(`threads spawned: ${threads.length}`)
    yield* Scope.close(scope, Exit.void)
  }

  yield* section("4. A module that can't load is respawned every second, forever")
  {
    const scope = yield* Scope.make()
    const client = yield* makeClient(scope, "./broken-worker.mjs")
    const ping = yield* Effect.forkChild(Effect.exit(client.Ping()))
    yield* Effect.sleep("4500 millis")
    log(`threads spawned in 4.5 s: ${threads.length}`)
    const result = ping.pollUnsafe()
    log("the Ping sent at the start:", result ? result.toString().slice(0, 160) : "still waiting")
    yield* Fiber.interrupt(ping)
    yield* Scope.close(scope, Exit.void)
  }
})

Effect.runPromise(program).then(
  () => process.exit(0),
  (error) => {
    console.error(error)
    process.exit(1)
  }
)

Output:

=== 1. A request that times out leaves its thread running
[0.01s] spawned thread #1
[0.78s] Spin(5000) with a 500 ms timeout -> { _id: 'Option', _tag: 'None' }
[1.78s] thread #1 utilisation over the next second: 1.00
[5.28s] the next Ping took 3s 497ms 937833ns, on the same thread, behind the abandoned Spin
[5.29s] thread #1 exited (code 130)

=== 2. Closing the scope while a task is stuck takes NodeWorker's 5 s
[0.00s] spawned thread #1
[0.65s] Spin(30000) timed out; closing the client's scope
[5.65s] thread #1 exited (code 1)
[5.65s] scope closed after 5s 3ms 373292ns

=== 3. A thread that dies is respawned a second later, with nothing waiting
[0.00s] spawned thread #1
[0.19s] thread #1 exited (code 1)
[0.19s] Crash -> Failure(Cause([Fail(~effect/rpc/RpcClientError: WorkerReceiveError: The worker has exited with code: 1)]))
[1.20s] spawned thread #2
[2.70s] threads spawned: 2
[2.70s] thread #2 exited (code 130)

=== 4. A module that can't load is respawned every second, forever
[0.00s] spawned thread #1
[0.02s] thread #1 exited (code 1)
[1.02s] spawned thread #2
[1.05s] thread #2 exited (code 1)
[2.05s] spawned thread #3
[2.07s] thread #3 exited (code 1)
[3.07s] spawned thread #4
[3.08s] thread #4 exited (code 1)
[4.09s] spawned thread #5
[4.10s] thread #5 exited (code 1)
[4.50s] threads spawned in 4.5 s: 5
[4.50s] the Ping sent at the start: Success(Failure(Cause([Fail(~effect/rpc/RpcClientError: WorkerReceiveError: An error event was emitted)])))