From f0fe642f4d6a13d082ef7a1ccc75730268cff4ce Mon Sep 17 00:00:00 2001 From: Martijn Walraven Date: Sun, 5 Jul 2026 20:50:53 +0200 Subject: [PATCH] fix(process-compose): reset user-stopped services in startService MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit stopService marks a service stoppedByUser and drives its state machine to Stopped — a state the transition table only exits via RestartTriggered. startService re-ran the fiber (the process starts and serves) but never reset, so every subsequent event was dropped by applyEvent: the service stays Stopped for getState/state streams forever, the unless-stopped restart policy stays disarmed, and the stack README's documented stop→start cycle ("Restart it (blocks until ready)") never worked through startService. The fix mirrors restartService's remove-reset-run sequence (stopForRestart): the fiber-map entry is removed before the reset — a terminal fiber not yet pruned would make the onlyIfMissing run a no-op and strand the freshly reset service in Pending — then the state machine resets, then the run proceeds. Gated on terminal state (Stopped/Failed) only: a service still Stopping under an in-flight stopService is not reset out from under that stop. Hardened against the PR review's concurrency findings: the start-path reset (resetForStart) replaces a signal Deferred only when already resolved — an unresolved signal can have live waiters (a dependent's dependency await, a waitReady caller) that replacement would orphan, while a resolved one has no waiters by definition and must be replaced so the fresh life can re-attest it. And each service's start step serializes on a startGate semaphore, so concurrent startService calls coalesce on one fresh run instead of the second removing the first's fiber. Regression tests: the stop→start cycle, the failed→start heal, a probed dependency that failed before ever becoming healthy unblocking its waiting dependent through the preserved signal, and concurrent starts spawning exactly once. Co-Authored-By: Claude Fable 5 --- packages/process-compose/src/Orchestrator.ts | 56 ++++++++- .../src/Orchestrator.unit.test.ts | 112 ++++++++++++++++++ 2 files changed, 167 insertions(+), 1 deletion(-) diff --git a/packages/process-compose/src/Orchestrator.ts b/packages/process-compose/src/Orchestrator.ts index ad24308ffa..2723481f94 100644 --- a/packages/process-compose/src/Orchestrator.ts +++ b/packages/process-compose/src/Orchestrator.ts @@ -7,6 +7,7 @@ import { FiberMap, Layer, Context, + Semaphore, Stream, SubscriptionRef, } from "effect"; @@ -101,6 +102,10 @@ export class Orchestrator extends Context.Service< completed: Deferred.Deferred; stopped: Deferred.Deferred; stoppedByUser: boolean; + /** Serializes startService's per-service step: a second concurrent + * start must observe the first's reset state and coalesce on its + * fresh run instead of removing it. */ + readonly startGate: Semaphore.Semaphore; } const services = new Map(); @@ -114,6 +119,7 @@ export class Orchestrator extends Context.Service< completed: Deferred.makeUnsafe(), stopped: Deferred.makeUnsafe(), stoppedByUser: false, + startGate: Semaphore.makeUnsafe(1), }); } @@ -521,6 +527,30 @@ export class Orchestrator extends Context.Service< } }); + // The start-path reset. Same state-machine and user-stop reset as + // resetService, but a signal Deferred is replaced only when already + // resolved: an unresolved signal can have live waiters (a dependent + // fiber's dependency await, a waitReady caller) that replacement + // would orphan — the re-run resolves the same object instead. A + // resolved signal has no waiters by definition and must be replaced + // so the fresh life can re-attest it. restartService keeps the + // replace-all resetService: its restart closure re-runs every + // transitive dependent, so their waits are re-established anyway. + const resetForStart = (name: string): Effect.Effect => + Effect.gen(function* () { + const svc = services.get(name); + if (svc) { + yield* SubscriptionRef.set(svc.state, initial(name)); + if (Deferred.isDoneUnsafe(svc.started)) svc.started = Deferred.makeUnsafe(); + if (Deferred.isDoneUnsafe(svc.healthy)) svc.healthy = Deferred.makeUnsafe(); + if (Deferred.isDoneUnsafe(svc.completed)) { + svc.completed = Deferred.makeUnsafe(); + } + if (Deferred.isDoneUnsafe(svc.stopped)) svc.stopped = Deferred.makeUnsafe(); + svc.stoppedByUser = false; + } + }); + const stopForRestart = (name: string): Effect.Effect => Effect.gen(function* () { yield* sendEvent(name, { _tag: "StopRequested" }); @@ -613,7 +643,31 @@ export class Orchestrator extends Context.Service< } const order = graph.startOrderFor(name); for (const d of order) { - yield* FiberMap.run(fibers, d.name, runServiceSafe(d), { onlyIfMissing: true }); + // The transition table has no edge out of Stopped/Failed + // except RestartTriggered — without a reset the re-run + // fiber's events are all dropped by applyEvent, so the + // process serves while every state read stays Stopped + // forever. The fiber-map entry must go before the reset: a + // terminal fiber not yet pruned would make `onlyIfMissing` + // skip the re-run, stranding the freshly reset service in + // Pending. restartService does the same remove-reset-run + // sequence (stopForRestart). Gated on terminal state only: + // a service still Stopping under an in-flight stopService + // (stoppedByUser already set, its remove not yet run) must + // not be reset out from under that stop — the stop + // completes to Stopped and a subsequent start heals it. + const svc = services.get(d.name); + const startStep = Effect.gen(function* () { + if (svc !== undefined) { + const state = yield* SubscriptionRef.get(svc.state); + if (state.status === "Stopped" || state.status === "Failed") { + yield* FiberMap.remove(fibers, d.name); + yield* resetForStart(d.name); + } + } + yield* FiberMap.run(fibers, d.name, runServiceSafe(d), { onlyIfMissing: true }); + }); + yield* svc === undefined ? startStep : svc.startGate.withPermits(1)(startStep); } }), diff --git a/packages/process-compose/src/Orchestrator.unit.test.ts b/packages/process-compose/src/Orchestrator.unit.test.ts index 22a45f266b..5dd8dd6aa4 100644 --- a/packages/process-compose/src/Orchestrator.unit.test.ts +++ b/packages/process-compose/src/Orchestrator.unit.test.ts @@ -410,6 +410,118 @@ describe("Orchestrator", () => { }).pipe(Effect.provide(layer), Effect.scoped); }); + it.live("startService after stopService resets the user-stopped state machine", () => { + const { layer, proc } = setupOrchestrator([svc("a")], { + exitDelay: "5 seconds", + }); + return Effect.gen(function* () { + const orc = yield* Orchestrator; + yield* orc.start(); + yield* waitForHealthy(orc, "a"); + yield* orc.stopService("a"); + const stopped = yield* orc.getState("a"); + expect(stopped.status).toBe("Stopped"); + // Without the reset, the re-run fiber's events are all dropped by + // the transition table (no edge out of Stopped except + // RestartTriggered): the process would serve while every state read + // stays Stopped forever. + yield* orc.startService("a"); + yield* proc.waitForSpawnCount(2); + yield* waitForHealthy(orc, "a"); + const healed = yield* orc.getState("a"); + expect(healed.status).toBe("Healthy"); + }).pipe(Effect.provide(layer), Effect.scoped); + }); + + it.live("startService after terminal failure resets and re-runs", () => { + let spawns = 0; + const { layer, proc } = setupOrchestrator([svc("a", { restart: "no" })], { + getExitCode: () => (++spawns === 1 ? 1 : 0), + exitDelay: "200 millis", + }); + return Effect.gen(function* () { + const orc = yield* Orchestrator; + yield* orc.start(); + yield* waitForFailed(orc, "a"); + // Failed is terminal like Stopped (no edge out except + // RestartTriggered), and the failed run's fiber may still occupy + // the fiber map: without the remove-and-reset, startService's + // onlyIfMissing run would be skipped and the service would strand. + yield* orc.startService("a"); + yield* proc.waitForSpawnCount(2); + yield* waitForHealthy(orc, "a"); + const healed = yield* orc.getState("a"); + expect(healed.status).toBe("Healthy"); + }).pipe(Effect.provide(layer), Effect.scoped); + }); + + it.live("startService of a failed dependency unblocks a waiting dependent", () => { + let dbLives = 0; + let dbExits = 0; + const { layer } = setupOrchestrator( + [ + svc("db", { + restart: "no", + healthCheck: { + probe: { _tag: "Exec", command: "db-probe", args: [] }, + periodSeconds: 0.05, + successThreshold: 1, + failureThreshold: 999, + }, + }), + svc("api", { dependencies: [{ service: "db", condition: "healthy" }] }), + ], + { + exitDelay: "5 seconds", + onSpawn: (record) => { + if (record.command === "db") dbLives++; + }, + perService: { + // First life: db exits 1 before its probe ever succeeds, so it + // goes Failed with `healthy` UNRESOLVED while api's fiber waits + // on that exact Deferred. Second life: serves and probes healthy. + db: { exitDelay: "400 millis", getExitCode: () => (++dbExits === 1 ? 1 : 0) }, + "db-probe": { exitDelay: "1 millis", getExitCode: () => (dbLives >= 2 ? 0 : 1) }, + }, + }, + ); + return Effect.gen(function* () { + const orc = yield* Orchestrator; + yield* orc.start(); + yield* waitForFailed(orc, "db"); + // api is awaiting db's ORIGINAL healthy signal. The start-path reset + // must not replace an unresolved signal: db's second life resolves + // the same object, or api hangs to its dependency timeout. + yield* orc.startService("db"); + yield* waitForHealthy(orc, "db"); + yield* waitForHealthy(orc, "api"); + }).pipe(Effect.provide(layer), Effect.scoped); + }); + + it.live("concurrent startService calls coalesce on one fresh run", () => { + const { layer, proc } = setupOrchestrator([svc("a")], { + exitDelay: "5 seconds", + }); + return Effect.gen(function* () { + const orc = yield* Orchestrator; + yield* orc.start(); + yield* waitForHealthy(orc, "a"); + yield* orc.stopService("a"); + const stopped = yield* orc.getState("a"); + expect(stopped.status).toBe("Stopped"); + // The per-service start gate serializes the reset-and-run step: the + // second caller observes the reset state and coalesces via + // onlyIfMissing instead of removing the first caller's fresh fiber. + yield* Effect.all([orc.startService("a"), orc.startService("a")], { + concurrency: "unbounded", + }); + yield* proc.waitForSpawnCount(2); + yield* waitForHealthy(orc, "a"); + expect(proc.spawned.length).toBe(2); + expect(proc.killed.length).toBe(1); + }).pipe(Effect.provide(layer), Effect.scoped); + }); + it.live("stop() interrupts all service fibers", () => { const { layer, proc } = setupOrchestrator([svc("a"), svc("b")], { exitDelay: "5 seconds",