Capability Spec Delta: queue-distribution-worker-thread (enterprise-queue-distribution)

Package: @suaveplan/queue-distribution-worker-thread Status: PROPOSED

Every requirement below carries a **Verified by:** line at authoring time. (See openspec/AGENTS.md Section 13.)

ADDED Requirements

Requirement: WorkerFactory-driven communicator covering Web-Worker and `worker_threads`

The package SHALL expose WorkerThreadCommunicator implementing WorkerCommunicator from @suaveplan/queue-distribution-http (re-exported WorkerCommunicator interface). The constructor SHALL accept an injectable WorkerFactory of shape () => { postMessage(msg, transfer?); addEventListener(kind, handler); terminate(); } so the same class works against globalThis.Worker (browser) and a thin worker_threads wrapper (Node/Bun) without importing platform APIs at module scope. The implementation SHALL maintain a Map<requestId, { resolve, reject, timeoutHandle }> of outstanding requests. Each outgoing message SHALL include an id (generated via injected IdGenerator), a kind: "task" | "status" | "health" | "ping", and a payload validated by Zod. Reply correlation SHALL match id; unknown id replies SHALL be logged and dropped. Reply payloads SHALL be Zod-validated against per-kind schemas; failures SHALL reject the in-flight request with QueueDistributionWorkerThreadError(WT.MESSAGE_MALFORMED). Each request SHALL be subject to a per-call timeout (default defaultTimeoutMs) honoured by the injected Clock; expiry SHALL reject with WT.REQUEST_TIMEOUT and remove the entry from the map.

Scenario: round-trip correlation

WHEN sendTask/checkStatus/checkHealth/ping are invoked concurrently and the underlying worker replies in arbitrary order
THEN each promise resolves with its own reply payload (matched by id)

Scenario: malformed reply throws WT.MESSAGE_MALFORMED

WHEN the underlying worker posts back { id: <known>, kind: "task", payload: { something: "bad" } } not conforming to the task-reply Zod schema
THEN the in-flight sendTask promise rejects with QueueDistributionWorkerThreadError(WT.MESSAGE_MALFORMED)

Scenario: request timeout

WHEN a FakeClock is injected and the worker never replies within defaultTimeoutMs
THEN after the clock advances past the deadline, the call rejects with QueueDistributionWorkerThreadError(WT.REQUEST_TIMEOUT) AND the outstanding-request map no longer contains the entry

Verified by: src/communicator/worker-thread/worker-thread-communicator.test.ts::"correlates concurrent replies by id" AND src/communicator/worker-thread/worker-thread-communicator.test.ts::"rejects with WT.MESSAGE_MALFORMED when reply fails Zod validation" AND src/communicator/worker-thread/worker-thread-communicator.test.ts::"rejects with WT.REQUEST_TIMEOUT and clears the outstanding-request map"

Requirement: WorkerThreadWorker — `AbstractWorker` subclass dispatching to a backing worker

The package SHALL ship WorkerThreadWorker extending AbstractWorker from @suaveplan/queue-distribution. Constructor SHALL accept an injectable WorkerFactory and instantiate a fresh WorkerThreadCommunicator per worker, AND a single backing worker thread. doStart() SHALL register error / exit listeners on the backing worker — worker.exit events SHALL markUnhealthy("Worker exited") and flip status to OFFLINE; uncaught error events SHALL emit the WT.WORKER_CRASHED OTel counter. doStop(force) SHALL terminate() the backing worker, fail every outstanding request with WT.WORKER_TERMINATED, and clear the outstanding-request map. doExecuteTask SHALL delegate to communicator.sendTask(<no-endpoint>, task, input, signal). getCapabilities() SHALL return { isRemote: false, isThreaded: true }. getType() SHALL return "worker-thread".

Scenario: backing worker crash promotes to UNHEALTHY

WHEN the backing worker emits an error event
THEN the OTel queue_dist_wt.worker.crashed_total counter increments and the worker is marked UNHEALTHY

Scenario: terminate fails outstanding requests with WT.WORKER_TERMINATED

WHEN two sendTask calls are in flight and worker.stop(true) is invoked
THEN both pending promises reject with QueueDistributionWorkerThreadError(WT.WORKER_TERMINATED) AND worker.status === OFFLINE

Verified by: src/worker/worker-thread/worker-thread-worker.test.ts::"backing worker crash flips status to UNHEALTHY and increments WT.WORKER_CRASHED counter" AND src/worker/worker-thread/worker-thread-worker.test.ts::"terminate fails outstanding requests with WT.WORKER_TERMINATED"

Requirement: Stable error registry for the worker-thread plugin

The package SHALL register every error code through defineCodes("WT", [...]) from @suaveplan/error and expose QueueDistributionWorkerThreadError extends SuaveplanError<WtErrorCode>. The exhaustive set is: WORKER_CRASHED, MESSAGE_MALFORMED, REQUEST_TIMEOUT, WORKER_TERMINATED. Every code SHALL be thrown by ≥1 production call path.

Scenario: every registered code is thrown

WHEN the test suite iterates over every code in the registry
THEN a corresponding production throw site exists

Verified by: src/errors/queue-distribution-worker-thread-error.test.ts::"every registered WT.* code is thrown by a production call path"

Requirement: OpenTelemetry instrumentation

The package SHALL emit, via @suaveplan/opentelemetry under meter name "@suaveplan/queue-distribution-worker-thread":

histogram: queue_dist_wt.request.duration_ms (attributes kind, outcome)
counter: queue_dist_wt.request_total (attributes kind, outcome)
counter: queue_dist_wt.worker.crashed_total
counter: queue_dist_wt.worker.terminated_total
span: queue_dist_wt.sendTask

Scenario: histogram emission on sendTask round-trip

WHEN sendTask resolves successfully under an in-memory MeterProvider
THEN one observation is recorded on queue_dist_wt.request.duration_ms with kind === "task" AND outcome === "success"

Verified by: src/communicator/worker-thread/worker-thread-communicator.spec-conformance.test.ts::"emits queue_dist_wt.request.duration_ms histogram with kind and outcome attributes"

6.3 KiB Raw Permalink Blame History