fix(process-compose): reset user-stopped services in startService

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<number>;
           stopped: Deferred.Deferred<void>;
           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<string, ServiceSignals>();
 
@@ -114,6 +119,7 @@ export class Orchestrator extends Context.Service<
             completed: Deferred.makeUnsafe<number>(),
             stopped: Deferred.makeUnsafe<void>(),
             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<void> =>
+          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<void>();
+              if (Deferred.isDoneUnsafe(svc.healthy)) svc.healthy = Deferred.makeUnsafe<void>();
+              if (Deferred.isDoneUnsafe(svc.completed)) {
+                svc.completed = Deferred.makeUnsafe<number>();
+              }
+              if (Deferred.isDoneUnsafe(svc.stopped)) svc.stopped = Deferred.makeUnsafe<void>();
+              svc.stoppedByUser = false;
+            }
+          });
+
         const stopForRestart = (name: string): Effect.Effect<void> =>
           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);
               }
             }),
 

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",