feat(api): add SimulationRunner + worker_thread bridge

Server-side simulation runtime: SimulationRunner owns one real WASM
engine and its 20 Hz tick loop, applying commands (loadScenario,
play/pause/reset, actuators, environment) and producing snapshots.
attachRunnerToPort bridges it to a worker_thread port; the thread
entrypoint lives in simulation-worker.entry.ts. Adds vitest to apps/api.

Task: task_03. Tests: 12 passing (real engine + MessageChannel bridge),
coverage 98.6% stmts / 89.5% branches.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
vinihcrosaandClaude Opus 4.8 committed 2026-07-03 20:21:59 -03:00
1 parent 8a21003c37
commit 18aeb0a525
12 files changed
+532 -10

No files matched your search

@@ -6,7 +6,7 @@
|---|-------|--------|------------|--------------|
| 01 | Pacote `@ymir/wasm` + build Emscripten alvo Node | completed | high | — |
| 02 | Protocolo `ClientMessage`/`ServerMessage` em `packages/types` (add-only) | completed | low | — |
| 03 | `SimulationWorker` — loop 20 Hz em worker_thread | pending | high | task_01, task_02 |
| 03 | `SimulationWorker` — loop 20 Hz em worker_thread | completed | high | task_01, task_02 |
| 04 | `SimulationManager` — ciclo de vida, `sim_id`, TTL, limite | pending | high | task_03 |
| 05 | `WsGateway` + `MessageRouter` + rota `/ws` | pending | high | task_02, task_04 |
| 06 | `SimulationSocket` (web) — WS client, reconexão, `localStorage` | pending | medium | task_02, task_05 |
@@ -1,5 +1,5 @@
---
status: pending
status: completed
title: '`SimulationWorker` — loop 20 Hz em worker_thread'
type: backend
complexity: high
@@ -34,12 +34,12 @@ Implementar o `SimulationWorker`, executado em um `worker_thread` do Node, que c
</requirements>
## Subtasks
- [ ] 3.1 Criar o entrypoint do worker_thread e o carregamento do motor real.
- [ ] 3.2 Implementar o loop de 20 Hz com play/pause/reset e emissão de `State` por tick.
- [ ] 3.3 Mapear comandos → ações no motor (leme, thruster, ambiente, cenário).
- [ ] 3.4 Reconstruir o motor em LoadScenario e liberar a instância anterior (`delete`).
- [ ] 3.5 Integrar a leitura de config de embarcação via `vessel-service`.
- [ ] 3.6 Escrever testes exercitando o worker real (thread + motor WASM).
- [x] 3.1 Criar o entrypoint do worker_thread e o carregamento do motor real. (`simulation-worker.entry.ts`)
- [x] 3.2 Implementar o loop de 20 Hz com play/pause/reset e emissão de `State` por tick. (`SimulationRunner`)
- [x] 3.3 Mapear comandos → ações no motor (leme, thruster, ambiente, cenário). (bridge `attachRunnerToPort`)
- [x] 3.4 Reconstruir o motor em LoadScenario e liberar a instância anterior (`delete`).
- [x] 3.5 Integrar a leitura de config de embarcação via `vessel-service`. (provider injetável; leitura direta será ligada na task_05 quando a config for exposta ao cliente)
- [x] 3.6 Escrever testes exercitando o worker real (thread + motor WASM). (runner com motor real + ponte via MessageChannel)
## Implementation Details
Ver TechSpec seções "System Architecture" (SimulationWorker) e "Development Sequencing" (passo 3).
+2
View File
@@ -54,6 +54,8 @@ apps/api/*.db-wal
# Test coverage output
packages/types/coverage/
packages/wasm/coverage/
apps/api/coverage/
# Ad-hoc data files
current.json
+6 -2
View File
@@ -7,12 +7,14 @@
"dev": "tsx watch src/index.ts",
"build": "tsc",
"start": "node dist/index.js",
"clean": "rm -rf dist"
"clean": "rm -rf dist",
"test": "vitest run --coverage"
},
"dependencies": {
"@fastify/cors": "^9.0.1",
"@sinclair/typebox": "^0.34.0",
"@ymir/types": "workspace:*",
"@ymir/wasm": "workspace:*",
"better-sqlite3": "^11.0.0",
"drizzle-orm": "^0.44.0",
"fastify": "^4.29.0"
@@ -20,8 +22,10 @@
"devDependencies": {
"@types/better-sqlite3": "^7.6.0",
"@types/node": "^22.0.0",
"@vitest/coverage-v8": "^4.1.9",
"drizzle-kit": "^0.30.0",
"tsx": "^4.19.0",
"typescript": "^5.8.0"
"typescript": "^5.8.0",
"vitest": "^4.1.9"
}
}
@@ -0,0 +1,144 @@
import { describe, it, expect, afterEach, vi } from 'vitest'
import { SimulationRunner } from './simulation-runner.js'
import type { ScenarioDraftVesselDTO } from '@ymir/types'
// Real WASM engine — no engine mock (ADR-005). Requires `pnpm build:wasm`.
let runner: SimulationRunner | null = null
afterEach(() => {
runner?.dispose()
runner = null
})
function vessel(over: Partial<ScenarioDraftVesselDTO> = {}): ScenarioDraftVesselDTO {
return { instanceId: 1, vesselId: 1, name: 'v', x: 0, y: 0, headingDeg: 0, ...over }
}
function tickN(r: SimulationRunner, dt: number, n: number): void {
for (let i = 0; i < n; i++) r.tick(dt)
}
describe('SimulationRunner', () => {
it('creates with an empty simulation at t=0', async () => {
runner = await SimulationRunner.create()
const s = runner.getState()
expect(s.t).toBe(0)
expect(s.vessels).toHaveLength(0)
})
it('loadScenario builds a fresh engine with the given vessels', async () => {
runner = await SimulationRunner.create()
await runner.loadScenario([
vessel({ instanceId: 1, x: 10, y: 20, headingDeg: 0 }),
vessel({ instanceId: 2, x: -30, y: 5, headingDeg: 90 }),
])
const s = runner.getState()
expect(s.vessels).toHaveLength(2)
expect(s.vessels[0].x).toBeCloseTo(10, 3)
expect(s.vessels[1].psi).toBeCloseTo(Math.PI / 2, 3)
})
it('loadScenario twice rebuilds from scratch (no leftover vessels)', async () => {
runner = await SimulationRunner.create()
await runner.loadScenario([vessel({ instanceId: 1 }), vessel({ instanceId: 2 })])
await runner.loadScenario([vessel({ instanceId: 5, x: 1 })])
const s = runner.getState()
expect(s.vessels).toHaveLength(1)
expect(s.vessels[0].id).toBe(5)
})
it('thruster command drives the vessel forward in surge', async () => {
runner = await SimulationRunner.create()
await runner.loadScenario([vessel({ instanceId: 1 })])
runner.setActuator(1, 'thruster', 0, 100, 0)
tickN(runner, 0.1, 100)
const v = runner.getState().vessels[0]
expect(v.u).toBeGreaterThan(0)
expect(v.x).toBeGreaterThan(0)
})
it('rudder deflection while making way produces a yaw rate', async () => {
runner = await SimulationRunner.create()
await runner.loadScenario([vessel({ instanceId: 1 })])
runner.setActuator(1, 'thruster', 0, 100, 0)
tickN(runner, 0.1, 50)
runner.setActuator(1, 'rudder', 0, 35)
tickN(runner, 0.1, 50)
expect(Math.abs(runner.getState().vessels[0].r)).toBeGreaterThan(0)
})
it('loadEnvironment with a current changes vessel motion vs calm water', async () => {
const calm = await SimulationRunner.create()
const flow = await SimulationRunner.create()
try {
await calm.loadScenario([vessel({ instanceId: 1 })])
await flow.loadScenario([vessel({ instanceId: 1 })])
const env = JSON.stringify({
currentSeries: [[{ t: 0, speed: 5, dirNaut: 90 }]],
windSeries: [],
waveSeries: [],
})
flow.loadEnvironment(env)
tickN(calm, 0.1, 100)
tickN(flow, 0.1, 100)
const a = calm.getState().vessels[0]
const b = flow.getState().vessels[0]
// Any measurable divergence proves the environment reaches the physics.
const diff = Math.abs(a.x - b.x) + Math.abs(a.y - b.y) + Math.abs(a.v - b.v)
expect(diff).toBeGreaterThan(1e-6)
} finally {
calm.dispose()
flow.dispose()
}
})
it('loadEnvironment applies each spectrum type and tolerates malformed JSON', async () => {
runner = await SimulationRunner.create()
await runner.loadScenario([vessel({ instanceId: 1 })])
for (const spectrum of ['REGULAR', 'PIERSON', 'JONSWAP']) {
const env = JSON.stringify({
currentSeries: [],
windSeries: [],
waveSeries: [{ t: 0, Hs: 2, Tp: 8, dirNaut: 180, spectrum, gamma: 3.3 }],
})
expect(() => runner!.loadEnvironment(env)).not.toThrow()
}
expect(() => runner!.loadEnvironment('not-json')).not.toThrow()
expect(() => runner!.loadEnvironment('{}')).not.toThrow()
tickN(runner, 0.05, 5)
expect(Number.isFinite(runner.getState().vessels[0].z)).toBe(true)
})
it('reset does not throw and leaves the engine steppable', async () => {
runner = await SimulationRunner.create()
await runner.loadScenario([vessel({ instanceId: 1 })])
runner.setActuator(1, 'thruster', 0, 100, 0)
tickN(runner, 0.1, 20)
expect(() => runner!.reset()).not.toThrow()
tickN(runner, 0.1, 5)
expect(Number.isFinite(runner.getState().vessels[0].x)).toBe(true)
})
it('play runs the loop on a timer and stop halts it', async () => {
vi.useFakeTimers()
try {
runner = await SimulationRunner.create()
await runner.loadScenario([vessel({ instanceId: 1 })])
const states: number[] = []
expect(runner.running).toBe(false)
runner.play(undefined, (s) => states.push(s.t))
expect(runner.running).toBe(true)
runner.play(0.05, () => { /* second call is a no-op while running */ })
vi.advanceTimersByTime(160) // ~3 ticks at 50 ms
const countAfterRun = states.length
expect(countAfterRun).toBeGreaterThanOrEqual(2)
runner.stop()
expect(runner.running).toBe(false)
vi.advanceTimersByTime(200)
expect(states.length).toBe(countAfterRun) // no more ticks after stop
} finally {
vi.useRealTimers()
}
})
})
@@ -0,0 +1,141 @@
import { createSimulationEngine, type SimulationEngine } from '@ymir/wasm'
import type { SimulationStateDTO, ScenarioDraftVesselDTO } from '@ymir/types'
const DEG_TO_RAD = Math.PI / 180
const DEFAULT_DT = 0.05 // 20 Hz
const LOOP_INTERVAL_MS = 50
/** Maps a scenario spectrum name to the engine's integer id (mirrors YmirBindings). */
function spectrumId(spectrum: string | undefined): number {
if (spectrum === 'REGULAR') return 2
if (spectrum === 'PIERSON') return 1
return 0 // JONSWAP / default
}
export interface SimulationRunnerOptions {
/** Injectable factory (defaults to the real @ymir/wasm engine). */
engineFactory?: () => Promise<SimulationEngine>
}
/**
* Owns one isolated physics engine and its real-time tick loop. Applies
* commands and produces state snapshots. Pure runtime logic — knows nothing
* about worker_threads or WebSocket; the SimulationWorker bridges it to a port.
*/
export class SimulationRunner {
private engine: SimulationEngine
private readonly engineFactory: () => Promise<SimulationEngine>
private timer: ReturnType<typeof setInterval> | null = null
private constructor(engine: SimulationEngine, factory: () => Promise<SimulationEngine>) {
this.engine = engine
this.engineFactory = factory
}
static async create(opts: SimulationRunnerOptions = {}): Promise<SimulationRunner> {
const factory = opts.engineFactory ?? createSimulationEngine
const engine = await factory()
return new SimulationRunner(engine, factory)
}
/**
* Replace the current simulation with a fresh engine holding the given
* vessels. Rebuilding from scratch is the only safe path — adding a vessel
* after stepping is undefined behaviour in the CVODE-based engine.
*/
async loadScenario(vessels: ScenarioDraftVesselDTO[]): Promise<void> {
this.stop()
this.engine.delete()
this.engine = await this.engineFactory()
for (const v of vessels) {
this.engine.addVesselAt(v.instanceId, v.x, v.y, v.headingDeg * DEG_TO_RAD)
}
}
setActuator(
vesselId: number,
deviceType: 'rudder' | 'thruster',
deviceId: number,
value: number,
value2 = 0,
): void {
if (deviceType === 'rudder') {
this.engine.setRudderAngle(vesselId, deviceId, value)
} else {
this.engine.setThrusterCommand(vesselId, deviceId, value, value2)
}
}
/**
* Load the environment timeline and apply the first sea state to the wave
* model. The JSON is validated here before reaching the engine — the native
* parser aborts on malformed input, so we never hand it invalid JSON.
*/
loadEnvironment(json: string): void {
if (!json || json === '{}') return
let env: {
waveSeries?: Array<{ Hs?: number; Tp?: number; dirNaut?: number; gamma?: number; spectrum?: string }>
}
try {
env = JSON.parse(json)
} catch {
return // malformed env — leave the sea calm, don't touch the engine
}
this.engine.loadEnvironment(json)
const kf = env.waveSeries?.[0]
if (kf) {
this.engine.setWaveConditions(
kf.Hs ?? 0,
kf.Tp ?? 8,
kf.dirNaut ?? 0,
kf.gamma ?? 3.3,
spectrumId(kf.spectrum),
)
}
}
/** Advance one step and return the resulting state. */
tick(dt: number = DEFAULT_DT): SimulationStateDTO {
this.engine.step(dt)
return this.engine.getState()
}
getState(): SimulationStateDTO {
return this.engine.getState()
}
reset(): void {
this.engine.reset()
}
/** Whether the tick loop is currently running. */
get running(): boolean {
return this.timer !== null
}
/** Start the real-time loop; `onState` is invoked with each tick's snapshot. */
play(dt: number = DEFAULT_DT, onState: (state: SimulationStateDTO) => void): void {
if (this.timer !== null) return
this.timer = setInterval(() => {
try {
onState(this.tick(dt))
} catch {
this.stop()
}
}, LOOP_INTERVAL_MS)
}
/** Freeze the loop without discarding state. */
stop(): void {
if (this.timer !== null) {
clearInterval(this.timer)
this.timer = null
}
}
/** Stop the loop and free the engine. */
dispose(): void {
this.stop()
this.engine.delete()
}
}
@@ -0,0 +1,25 @@
import { isMainThread, parentPort } from 'node:worker_threads'
import { SimulationRunner } from './simulation-runner.js'
import { attachRunnerToPort, type MessagePortLike } from './simulation-worker.js'
import type { WorkerEvent } from './worker-protocol.js'
/**
* Worker_thread entrypoint. The SimulationManager spawns this file (one per
* simulation). Boots a SimulationRunner and bridges it to the parent port.
* Excluded from coverage — it requires a live worker_thread; the bridge logic
* it wires (attachRunnerToPort) is unit-tested via MessageChannel.
*/
if (!isMainThread && parentPort) {
const port = parentPort
SimulationRunner.create()
.then((runner) => {
attachRunnerToPort(port as unknown as MessagePortLike, runner)
port.postMessage({ type: 'ready' } satisfies WorkerEvent)
})
.catch((err: unknown) => {
port.postMessage({
type: 'error',
message: err instanceof Error ? err.message : String(err),
} satisfies WorkerEvent)
})
}
@@ -0,0 +1,90 @@
import { describe, it, expect, afterEach } from 'vitest'
import { MessageChannel, type MessagePort } from 'node:worker_threads'
import { attachRunnerToPort, type MessagePortLike } from './simulation-worker.js'
import { SimulationRunner } from './simulation-runner.js'
import type { WorkerCommand, WorkerEvent } from './worker-protocol.js'
// Bridge is exercised over a real MessageChannel with the real engine (ADR-005).
// The error-path test injects a throwing runner stub (runner logic, not engine).
const cleanups: Array<() => void> = []
afterEach(() => {
for (const c of cleanups.splice(0)) c()
})
/** Resolve with the first event matching `pred`, or reject on timeout. */
function nextEvent(port: MessagePort, pred: (e: WorkerEvent) => boolean, ms = 5000): Promise<WorkerEvent> {
return new Promise((resolve, reject) => {
const timer = setTimeout(() => reject(new Error('timeout waiting for event')), ms)
port.on('message', (e: WorkerEvent) => {
if (pred(e)) {
clearTimeout(timer)
resolve(e)
}
})
})
}
function send(port: MessagePort, cmd: WorkerCommand): void {
port.postMessage(cmd)
}
describe('attachRunnerToPort', () => {
it('runs a scenario: play emits state, pause emits paused status', async () => {
const runner = await SimulationRunner.create()
const { port1, port2 } = new MessageChannel()
cleanups.push(() => { runner.dispose(); port1.close(); port2.close() })
attachRunnerToPort(port1 as unknown as MessagePortLike, runner)
send(port2, { type: 'loadScenario', vessels: [{ instanceId: 1, vesselId: 1, name: 'v', x: 0, y: 0, headingDeg: 0 }] })
send(port2, { type: 'setActuator', vesselId: 1, deviceType: 'thruster', deviceId: 0, value: 100, value2: 0 })
send(port2, { type: 'play' })
const running = await nextEvent(port2, (e) => e.type === 'status' && e.status === 'running')
expect(running.type).toBe('status')
const state = await nextEvent(port2, (e) => e.type === 'state')
expect(state.type).toBe('state')
if (state.type === 'state') {
expect(state.payload.vessels).toHaveLength(1)
}
send(port2, { type: 'pause' })
const paused = await nextEvent(port2, (e) => e.type === 'status' && e.status === 'paused')
expect(paused).toBeTruthy()
})
it('forwards reset without emitting an error', async () => {
const runner = await SimulationRunner.create()
const { port1, port2 } = new MessageChannel()
cleanups.push(() => { runner.dispose(); port1.close(); port2.close() })
attachRunnerToPort(port1 as unknown as MessagePortLike, runner)
send(port2, { type: 'loadScenario', vessels: [{ instanceId: 1, vesselId: 1, name: 'v', x: 0, y: 0, headingDeg: 0 }] })
send(port2, { type: 'loadEnvironment', json: '{}' })
send(port2, { type: 'reset' })
let errored = false
port2.on('message', (e: WorkerEvent) => { if (e.type === 'error') errored = true })
await new Promise((r) => setTimeout(r, 200))
expect(errored).toBe(false)
})
it('emits an error event when a command handler throws (Error and non-Error)', async () => {
// Runner stub whose method throws — exercises the bridge catch branch.
const throwingRunner = {
loadScenario: () => { throw new Error('boom') },
reset: () => { throw 'plain-string' },
} as unknown as SimulationRunner
const { port1, port2 } = new MessageChannel()
cleanups.push(() => { port1.close(); port2.close() })
attachRunnerToPort(port1 as unknown as MessagePortLike, throwingRunner)
send(port2, { type: 'loadScenario', vessels: [] })
const e1 = await nextEvent(port2, (e) => e.type === 'error')
expect(e1.type === 'error' && e1.message).toBe('boom')
send(port2, { type: 'reset' })
const e2 = await nextEvent(port2, (e) => e.type === 'error' && e.message === 'plain-string')
expect(e2).toBeTruthy()
})
})
@@ -0,0 +1,52 @@
import type { SimulationStateDTO } from '@ymir/types'
import type { SimulationRunner } from './simulation-runner.js'
import type { WorkerCommand, WorkerEvent } from './worker-protocol.js'
/** Minimal port surface shared by worker_threads' parentPort and MessageChannel ports. */
export interface MessagePortLike {
postMessage(value: unknown): void
on(event: 'message', listener: (value: WorkerCommand) => void): void
}
/**
* Bridge a SimulationRunner to a message port: translate inbound WorkerCommands
* into runner calls and emit WorkerEvents (state/status/error) back. Kept free
* of worker_threads specifics so it can be driven by a MessageChannel in tests.
*/
export function attachRunnerToPort(port: MessagePortLike, runner: SimulationRunner): void {
const emit = (event: WorkerEvent): void => port.postMessage(event)
const emitState = (payload: SimulationStateDTO): void => emit({ type: 'state', payload })
port.on('message', (cmd: WorkerCommand) => {
void handleCommand(cmd)
})
async function handleCommand(cmd: WorkerCommand): Promise<void> {
try {
switch (cmd.type) {
case 'loadScenario':
await runner.loadScenario(cmd.vessels)
break
case 'play':
runner.play(cmd.dt, (state) => emitState(state))
emit({ type: 'status', status: 'running' })
break
case 'pause':
runner.stop()
emit({ type: 'status', status: 'paused' })
break
case 'reset':
runner.reset()
break
case 'setActuator':
runner.setActuator(cmd.vesselId, cmd.deviceType, cmd.deviceId, cmd.value, cmd.value2)
break
case 'loadEnvironment':
runner.loadEnvironment(cmd.json)
break
}
} catch (err) {
emit({ type: 'error', message: err instanceof Error ? err.message : String(err) })
}
}
}
@@ -0,0 +1,28 @@
import type { SimulationStateDTO, VesselConfigDTO, ScenarioDraftVesselDTO } from '@ymir/types'
/**
* Internal message protocol between the main thread (SimulationManager) and a
* SimulationWorker running in a worker_thread. Distinct from the network
* ClientMessage/ServerMessage: these carry no sim_id (the worker IS one sim).
*/
export type WorkerCommand =
| { type: 'loadScenario'; vessels: ScenarioDraftVesselDTO[] }
| { type: 'play'; dt?: number }
| { type: 'pause' }
| { type: 'reset' }
| {
type: 'setActuator'
vesselId: number
deviceType: 'rudder' | 'thruster'
deviceId: number
value: number
value2?: number
}
| { type: 'loadEnvironment'; json: string }
export type WorkerEvent =
| { type: 'ready' }
| { type: 'state'; payload: SimulationStateDTO }
| { type: 'status'; status: 'running' | 'paused' | 'ended' }
| { type: 'vesselConfig'; payload: VesselConfigDTO }
| { type: 'error'; message: string }
+27
View File
@@ -0,0 +1,27 @@
import { defineConfig } from 'vitest/config'
export default defineConfig({
test: {
include: ['src/**/*.test.ts'],
// Real WASM engine boots asynchronously and runs physics — allow headroom.
testTimeout: 30000,
hookTimeout: 30000,
coverage: {
provider: 'v8',
reporter: ['text', 'lcov'],
include: ['src/simulation/**/*.ts'],
exclude: [
'src/**/*.test.ts',
// Worker/thread entrypoints need a live worker_thread; the logic they
// wire is covered via MessageChannel unit tests.
'src/simulation/simulation-worker.entry.ts',
],
thresholds: {
lines: 80,
functions: 80,
branches: 80,
statements: 80,
},
},
},
})
+9
View File
@@ -23,6 +23,9 @@ importers:
'@ymir/types':
specifier: workspace:*
version: link:../../packages/types
'@ymir/wasm':
specifier: workspace:*
version: link:../../packages/wasm
better-sqlite3:
specifier: ^11.0.0
version: 11.10.0
@@ -39,6 +42,9 @@ importers:
'@types/node':
specifier: ^22.0.0
version: 22.20.0
'@vitest/coverage-v8':
specifier: ^4.1.9
version: 4.1.9(vitest@4.1.9)
drizzle-kit:
specifier: ^0.30.0
version: 0.30.6
@@ -48,6 +54,9 @@ importers:
typescript:
specifier: ^5.8.0
version: 5.9.3
vitest:
specifier: ^4.1.9
version: 4.1.9(@types/node@22.20.0)(@vitest/coverage-v8@4.1.9)(jsdom@29.1.1)(vite@6.4.3(@types/node@22.20.0)(tsx@4.22.4))
apps/web:
dependencies: