SuavePlan/suaveplan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
suaveplan/openspec/specs/queue-distribution/spec.md
admin fbc70f96e6 Initial Commit
2026-05-27 23:29:53 +01:00

21 KiB
Raw Permalink Blame History

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

Package: @suaveplan/queue-distribution Status: PROPOSED

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

ADDED Requirements

Requirement: Distributed task contract decoupled from @suaveplan/queue.Job

The package SHALL expose DistributedTask<I, O> as the unit of work executed by workers, independent of @suaveplan/queue's Job / state-machine model. The contract has: id: string, execute(input: I, context: ExecutionContext): Promise<O>, getMetadata(): TaskMetadata, getStatus(): TaskStatus, setStatus(status: TaskStatus): void. TaskMetadata includes createdAt: Date, optional startedAt/completedAt, attempts: number, optional tags: string[], and indexed extras. TaskStatus is the closed enum PENDING | RUNNING | COMPLETED | FAILED | CANCELLED | TIMED_OUT. ExecutionContext is caller-defined.

Scenario: typed metadata round-trip

  • WHEN a caller constructs a DistributedTask<I, O> literal and passes it through Worker.executeTask
  • THEN task.getMetadata() returns a value validated by TaskMetadataSchema and getStatus() reflects the lifecycle (PENDING → RUNNING → COMPLETED/FAILED)

Scenario: generics are preserved end-to-end

  • WHEN a caller writes const w: Worker; const t: DistributedTask<number, string>; const r = await w.executeTask(t, 42)
  • THEN the inferred type of r is exactly string (no unknown, no any)

Verified by: src/task/task.test.ts::"validates TaskMetadata via TaskMetadataSchema" AND src/task/task.spec-conformance.test.ts::"preserves <I, O> generics through Worker.executeTask"

Requirement: Worker interface and status model

The package SHALL expose a Worker interface with: id: string, capacity: number, weight: number, status: WorkerStatus, activeTasks: number, executeTask<I, O>(task, input, signal?): Promise<O>, getHealth(signal?): Promise<WorkerHealth>, start(signal?): Promise<void>, stop(force?, signal?): Promise<void>, canAcceptTask(): boolean, getInfo(): WorkerInfo. WorkerStatus is the closed enum ONLINE | BUSY | OFFLINE | UNHEALTHY. WorkerHealth has isHealthy: boolean and a metrics object containing at least cpu, memory, tasksFailed, tasksSucceeded, responseTime plus open-ended string-keyed extras. WorkerInfo returns identity, type label, capacity, weight, status, active count, startedAt?, lastHealthCheck?, lastHealthStatus?, and capabilities?.

Scenario: health and info shapes parse via Zod

  • WHEN a Worker implementation returns from getHealth() / getInfo()
  • THEN the value passes WorkerHealthSchema / WorkerInfoSchema (failures throw QueueDistributionError(QUEUE_DIST.INVALID_OPTIONS))

Scenario: canAcceptTask honours status and capacity

  • WHEN status === ONLINE and activeTasks < capacity
  • THEN canAcceptTask() returns true
  • WHEN status === OFFLINE or activeTasks >= capacity
  • THEN canAcceptTask() returns false

Verified by: src/worker/worker.test.ts::"WorkerHealthSchema rejects malformed health" AND src/worker/worker.test.ts::"canAcceptTask honours ONLINE+capacity and rejects OFFLINE"

Requirement: AbstractWorker lifecycle and metrics, no module-scope clocks

The package SHALL ship an AbstractWorker base class implementing Worker with: idempotent start() (no-op when not OFFLINE) emitting WorkerStatus.ONLINE; stop(force) rejecting when activeTasks > 0 && !force with QUEUE_DIST.WORKER_BUSY_STOP; executeTask tracking activeTasks, success/failure counters, response-time total, and toggling status between ONLINE and BUSY; markUnhealthy(reason) flipping status to UNHEALTHY; getInfo() aggregating all of the above. The class SHALL accept Clock, IdGenerator, and LoggerLike via constructor options — no Date.now(), no Math.random(), no uuid import at module scope.

Scenario: executeTask refuses OFFLINE workers

  • WHEN await worker.executeTask(task, input) is called while status === OFFLINE
  • THEN the call rejects with QueueDistributionError whose code is QUEUE_DIST.WORKER_NOT_AVAILABLE

Scenario: force stop cancels active tasks

  • WHEN await worker.stop(true) is called with activeTasks === 2
  • THEN worker.status === OFFLINE AND worker.activeTasks === 0

Scenario: metrics emit through injectable Clock

  • WHEN a FakeClock returns 1000 at task start and 1750 at completion
  • THEN getHealth() returns metrics.responseTime === 750 (the difference, not wall-clock)

Verified by: src/worker/abstract/abstract.test.ts::"executeTask refuses OFFLINE worker with WORKER_NOT_AVAILABLE" AND src/worker/abstract/abstract.test.ts::"force stop drains activeTasks and transitions to OFFLINE" AND src/worker/abstract/abstract.spec-conformance.test.ts::"metrics derive responseTime from injected Clock"

Requirement: LocalWorker in-process execution

The package SHALL ship LocalWorker<Ctx> extending AbstractWorker. doExecuteTask SHALL invoke task.execute(input, this.context) where context is the ExecutionContext passed at construction. getCapabilities() SHALL return { hasFileSystemAccess: true, isRemote: false, contextType: <ctor name> }. getType() SHALL return "local".

Scenario: task receives the configured context

  • WHEN new LocalWorker(ctx).executeTask(task, input) is called
  • THEN task.execute(input, ctx) is invoked with the exact same ctx reference

Verified by: src/worker/local/local.test.ts::"executeTask passes the configured ExecutionContext to task.execute"

Requirement: Load-balancing strategy interface

The package SHALL expose LoadBalancingStrategy with selectWorker(workers, task?): Worker | null, optional notifyWorkerSelected(worker, task), optional notifyTaskCompleted(worker, task, success, durationMs), optional getName(): string. AbstractLoadBalancingStrategy provides filterAvailableWorkers(workers): Worker[] returning workers with status === ONLINE || status === BUSY, activeTasks < capacity, and canAcceptTask() === true.

Scenario: filterAvailableWorkers excludes OFFLINE and at-capacity workers

  • WHEN the input list contains workers in every WorkerStatus state and varying activeTasks
  • THEN the output contains only those satisfying the three predicates above

Verified by: src/load-balancing/base/base.test.ts::"filterAvailableWorkers returns only ONLINE/BUSY workers under capacity"

Requirement: Round-robin strategy

The package SHALL ship RoundRobinStrategy that selects workers in rotation order. currentIndex SHALL wrap to 0 when reaching the end of the available-worker list, and SHALL clamp to 0 if the available-worker list shrinks below currentIndex. reset() SHALL set currentIndex to 0. selectWorker SHALL return null when no workers are available. getName() SHALL return "round-robin".

Scenario: rotation order and reset

  • WHEN selectWorker is invoked 4 times against the same 3-worker list
  • THEN the selections are workers [0, 1, 2, 0] in order
  • WHEN reset() is then called and selectWorker invoked once more
  • THEN the selection is worker [0] again

Scenario: pool shrink is safe

  • WHEN currentIndex === 2 and the available-worker list shrinks to length === 1
  • THEN the next selectWorker returns worker [0] (no out-of-range read)

Verified by: src/load-balancing/round-robin/round-robin.test.ts::"cycles through workers in order and wraps after the end" AND src/load-balancing/round-robin/round-robin.test.ts::"clamps currentIndex when the pool shrinks below it"

Requirement: Least-busy strategy with optional performance bias

The package SHALL ship LeastBusyStrategy selecting the worker with the fewest activeTasks. When usePerformanceMetrics is true, ties on activeTasks SHALL break by lower avgExecutionTime. Further ties SHALL break by higher capacity then higher weight. notifyTaskCompleted(worker, task, success, durationMs) SHALL update an internal EWMA-style cumulative average for successful runs only. getWorkerPerformance(id) SHALL return the stored record or null. resetPerformanceMetrics() SHALL clear the registry. setUsePerformanceMetrics(enabled) SHALL toggle the flag at runtime.

Scenario: tie-break by performance when enabled

  • WHEN two workers both have activeTasks === 0, perf metrics enabled, and recorded average durations of 100 ms and 200 ms
  • THEN selectWorker returns the 100 ms worker

Scenario: tie-break by capacity when perf metrics unavailable

  • WHEN two workers both have activeTasks === 0 and no perf history, with capacity 2 and 4
  • THEN selectWorker returns the capacity === 4 worker

Verified by: src/load-balancing/least-busy/least-busy.test.ts::"ties break by lower average execution time when usePerformanceMetrics is on" AND src/load-balancing/least-busy/least-busy.test.ts::"ties break by higher capacity when no performance history exists"

Requirement: Weighted strategy with injectable RNG

The package SHALL ship WeightedStrategy that selects workers with probability proportional to weight. The RNG SHALL be supplied via constructor option (Rng from @suaveplan/types); the strategy SHALL NOT call Math.random(). When totalWeight <= 0, the strategy SHALL fall back to uniform selection. getName() SHALL return "weighted".

Scenario: deterministic distribution under fixed RNG

  • WHEN a deterministic Rng yields the sequence [0.0, 0.5, 0.9] against weights [1, 3, 6] (total 10)
  • THEN selectWorker returns workers [0, 1, 2] in that order

Scenario: zero-weight pool falls back to uniform

  • WHEN every worker has weight === 0 and the Rng yields 0.5
  • THEN selectWorker returns the middle worker in the list

Verified by: src/load-balancing/weighted/weighted.test.ts::"selects workers proportionally under a deterministic Rng" AND src/load-balancing/weighted/weighted.test.ts::"falls back to uniform when total weight is zero"

Requirement: Periodic health checker

The package SHALL ship WorkerHealthChecker (interface) and PeriodicHealthChecker (implementation). checkHealth(worker) SHALL race the worker's getHealth() against an injectable-Clock-driven timeout (default 5,000 ms). On success, results SHALL be cached for cacheTimeMs (default 5,000 ms) when cacheResults is true. Consecutive failures (unhealthyThreshold, default 3) SHALL flip worker.status to UNHEALTHY; consecutive successes (healthyThreshold, default 2) SHALL flip UNHEALTHY back to ONLINE or BUSY based on activeTasks. getHealthCheckCounts(id), resetHealthCheckCounts(id), and clearCache() SHALL be exposed.

Scenario: unhealthy after threshold consecutive failures

  • WHEN worker.getHealth() rejects on three consecutive calls (with threshold 3)
  • THEN worker.status === UNHEALTHY after the third call, AND QUEUE_DIST.HEALTH_CHECK_FAILED is emitted via the OTel counter

Scenario: recovery after threshold consecutive successes

  • WHEN a worker previously marked UNHEALTHY returns healthy on two consecutive calls (with threshold 2) and has activeTasks === 0
  • THEN worker.status === ONLINE after the second call
  • WHEN the same but activeTasks > 0
  • THEN worker.status === BUSY

Scenario: timeout uses injected Clock

  • WHEN the test injects a FakeClock and worker.getHealth() never resolves
  • THEN after advancing the clock by healthCheckTimeoutMs, checkHealth returns the fallback isHealthy: false payload AND emits the OTel queue_dist.health_check.duration_ms histogram with the elapsed time

Verified by: src/health-checker/health-checker.test.ts::"flips worker to UNHEALTHY after threshold consecutive failures" AND src/health-checker/health-checker.test.ts::"recovers worker to ONLINE/BUSY after threshold consecutive successes" AND src/health-checker/health-checker.spec-conformance.test.ts::"timeout race honours injected Clock and emits duration histogram"

Requirement: Worker pool registry and search

The package SHALL ship WorkerPool with: addWorker(worker) (throws QUEUE_DIST.WORKER_DUPLICATE on duplicate id), removeWorker(id), getWorkerById(id), getAllWorkers(), getWorkersByStatus(status), getAvailableWorkers(), findWorkers(criteria) filtering by status, capacity range, weight range, type, and a custom matcher predicate, startAllWorkers(signal?), stopAllWorkers(force?, signal?), clear(), size(), getWorkersInfo().

Scenario: duplicate add throws WORKER_DUPLICATE

  • WHEN addWorker(w) is called twice with the same w.id
  • THEN the second call throws QueueDistributionError(QUEUE_DIST.WORKER_DUPLICATE)

Scenario: findWorkers matches every criteria field

  • WHEN findWorkers({ status: ONLINE, minCapacity: 2, maxWeight: 3, type: "local", matcher: w => w.activeTasks === 0 }) is called against a mixed pool
  • THEN only workers satisfying every predicate are returned

Verified by: src/pool/pool.test.ts::"addWorker rejects duplicate ids with WORKER_DUPLICATE" AND src/pool/pool.test.ts::"findWorkers filters by every criteria field including custom matcher"

Requirement: WorkerManager — assignment, retry, health, restart

The package SHALL ship WorkerManager(pool, balancer?, healthChecker?, options?, deps?) providing:

  • assignTask<I, O>(task, input, signal?) — selects via the configured LoadBalancingStrategy, executes on the chosen worker, retries up to maxTaskRetries (default 3) on failure with backoff via @suaveplan/retry, and records exhausted tasks in failedTasks. The retry sleep SHALL be driven by the injected Clock, NOT setTimeout at module scope. The retry loop SHALL be bounded (no recursive self-call). AbortSignal cancellation MUST be honoured.
  • assignTasks<I, O>(tasks[], inputs[], signal?) — Promise.all over assignTask; mismatch in array lengths throws QUEUE_DIST.INVALID_OPTIONS.
  • startHealthChecks() / stopHealthChecks() — periodic loop driven by injected Clock.setInterval; idempotent.
  • checkAllWorkersHealth(signal?) — fan-out over all workers, returns map of id → WorkerHealth; logs a warning when online count falls below minOnlineWorkers.
  • handleWorkerFailure(worker, error) — checks isHealthy; if not, flips to UNHEALTHY and optionally restarts.
  • setLoadBalancer(balancer), getLoadBalancer(), getHealthChecker(), getWorkerPool(), getFailedTasks(), clearFailedTasks(), getTaskRetries(id).
  • getOptions() / updateOptions(partial) — option changes that affect the health-check interval SHALL restart the loop.

The constructor SHALL throw QUEUE_DIST.INVALID_OPTIONS for negative maxTaskRetries, negative taskRetryDelayMs, or minOnlineWorkers < 0.

Scenario: no available workers throws NO_AVAILABLE_WORKERS

  • WHEN pool.getAvailableWorkers() returns [] and assignTask is called
  • THEN the call rejects with QueueDistributionError(QUEUE_DIST.NO_AVAILABLE_WORKERS)

Scenario: balancer returning null throws STRATEGY_NO_WORKER_SELECTED

  • WHEN the configured LoadBalancingStrategy.selectWorker returns null despite a non-empty available pool
  • THEN the call rejects with QueueDistributionError(QUEUE_DIST.STRATEGY_NO_WORKER_SELECTED)

Scenario: retries exhausted throws TASK_RETRIES_EXHAUSTED

  • WHEN the chosen worker rejects executeTask maxTaskRetries + 1 times in a row
  • THEN the call rejects with QueueDistributionError(QUEUE_DIST.TASK_RETRIES_EXHAUSTED) AND getFailedTasks() contains the task

Scenario: AbortSignal cancellation between retries

  • WHEN an AbortController is aborted between the first failure and the retry sleep
  • THEN the call rejects with QueueDistributionError(QUEUE_DIST.TASK_CANCELLED) without sleeping the full taskRetryDelayMs

Scenario: restart limit exhaustion throws WORKER_RESTART_EXHAUSTED

  • WHEN an unhealthy worker has reached maxWorkerRestarts and the supervisor attempts another restart
  • THEN the supervisor emits QUEUE_DIST.WORKER_RESTART_EXHAUSTED via the OTel counter and SHALL NOT call worker.start() again

Verified by: src/manager/manager.test.ts::"throws NO_AVAILABLE_WORKERS when pool is empty" AND src/manager/manager.test.ts::"throws STRATEGY_NO_WORKER_SELECTED when balancer returns null" AND src/manager/manager.test.ts::"throws TASK_RETRIES_EXHAUSTED and records the task in failedTasks" AND src/manager/manager.spec-conformance.test.ts::"AbortSignal cancellation between retries throws TASK_CANCELLED" AND src/manager/manager.spec-conformance.test.ts::"supervisor stops restarting after WORKER_RESTART_EXHAUSTED"

Requirement: Stable error registry via defineCodes

The package SHALL register every error code through defineCodes("QUEUE_DIST", [...]) from @suaveplan/error and expose QueueDistributionError extends SuaveplanError<QueueDistributionErrorCode>. The exhaustive code set is: WORKER_NOT_AVAILABLE, WORKER_DUPLICATE, WORKER_BUSY_STOP, WORKER_NOT_FOUND, WORKER_RESTART_EXHAUSTED, NO_AVAILABLE_WORKERS, STRATEGY_NO_WORKER_SELECTED, TASK_RETRIES_EXHAUSTED, HEALTH_CHECK_TIMEOUT, HEALTH_CHECK_FAILED, INVALID_OPTIONS, TASK_CANCELLED. Every code SHALL be thrown by at least one production call path.

Scenario: every registered code is thrown by a production code path

  • WHEN the test suite iterates over every code in the registry
  • THEN a corresponding throw site exists, and the test forces the failure and asserts the code matches

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

Requirement: OpenTelemetry instrumentation on every public entrypoint

The package SHALL emit the following instruments via @suaveplan/opentelemetry, registered under meter name "@suaveplan/queue-distribution":

  • histograms: queue_dist.task.duration_ms, queue_dist.health_check.duration_ms, queue_dist.worker.restart.duration_ms
  • counters: queue_dist.task.assigned_total, queue_dist.task.completed_total (attribute outcome: "success" | "failure"), queue_dist.task.retried_total, queue_dist.worker.restart_total (attribute outcome: "success" | "failure" | "exhausted"), queue_dist.strategy.no_worker_total
  • up/down counter: queue_dist.workers.active
  • spans: queue_dist.manager.assign, queue_dist.worker.executeTask, queue_dist.health.check, queue_dist.worker.restart

Scenario: task duration histogram emits with expected attributes

  • WHEN manager.assignTask(task, input) completes successfully under an in-memory MeterProvider
  • THEN the queue_dist.task.duration_ms histogram has one recorded value with attribute outcome === "success", AND the queue_dist.task.completed_total counter incremented by 1 with the same attribute

Scenario: span emission

  • WHEN manager.assignTask is invoked
  • THEN a span named queue_dist.manager.assign is recorded with attributes queue_dist.worker_id and queue_dist.task_id

Verified by: src/manager/manager.spec-conformance.test.ts::"emits queue_dist.task.duration_ms and completed counter on success" AND src/manager/manager.spec-conformance.test.ts::"emits queue_dist.manager.assign span with worker_id and task_id attributes"

Requirement: Universal-tier — zero browser / Node / React deps

The package SHALL be installable in browser, Bun, and Node with no platform-specific runtime imports. The package SHALL NOT import from node:*, bun:*, react, react-dom, next/*, or any DOM-only API at module scope. sideEffects: false SHALL be declared in package.json and enforced via the tree-shake test.

Scenario: dep-graph guard rejects platform imports

  • WHEN tests/dep-graph.test.ts scans every .ts file in src/ for forbidden imports
  • THEN the count of matches is zero AND the test passes

Verified by: tests/dep-graph.test.ts::"src/ contains no node:*, bun:*, or React imports"

Requirement: Test files use the canonical runner wrapper

Every test file in src/ SHALL import runner primitives (describe, it, expect, beforeAll, afterAll, beforeEach, afterEach, vi, expectTypeOf) from @suaveplan/testing/runner, NOT from vitest or bun:test directly.

Scenario: scan shows zero direct runner imports

  • WHEN grep -rnE "from ['\"](vitest|bun:test)['\"]" packages/flow/queue-distribution/src/ --include="*.ts" --include="*.tsx" is run
  • THEN the output is empty

Verified by: tests/dep-graph.test.ts::"no test file imports vitest or bun:test directly"

REMOVED Requirements

Requirement: WorkerManager.rebalance()

Reason: Upstream body was a stub: this.logger.warn("Rebalance operation not implemented"). Per CLAUDE.md §Rules ("Complete all implementations — no stubs") and AGENTS.md §2b ("NEVER ship stubs"), stubs do not ship to genesis.

Migration: None today — the method had no real implementation upstream, so no consumer can have relied on its behaviour. If a real rebalance algorithm is needed later, open a fresh change proposal.

Reference in New Issue View Git Blame Copy Permalink