Skip to content

fix(stack): reserve ports through service startup #6045

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jgoux wants to merge 8 commits into
base: lazy-stack/04-realtime-websocket
Choose a base branch
from
Open

fix(stack): reserve ports through service startup #6045

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
9d23fdb
fix(stack): reserve ports through service startup
jgoux Aug 3, 2026
8bba36c
fix(stack): harden managed port lock acquisition
jgoux Aug 4, 2026
1ad28fc
fix(stack): fail interrupted supervised startup
jgoux Aug 4, 2026
2662171
fix(stack): re-reserve ports across service restarts
jgoux Aug 4, 2026
8d3df92
fix(stack): harden managed port lock handoff
jgoux Aug 4, 2026
3d45f36
fix(stack): publish complete lock generations atomically
jgoux Aug 4, 2026
bf4fa2e
fix(stack): clean up interrupted lock installation
jgoux Aug 4, 2026
8d6ede9
fix(process-compose): fail services on pre-start errors
jgoux Aug 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
139 changes: 110 additions & 29 deletions packages/process-compose/src/Orchestrator.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,6 @@
import { access, mkdtemp, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import {
Cause,
Deferred,
Expand All @@ -6,6 +9,7 @@ import {
Exit,
FiberMap,
Layer,
Schedule,
Context,
Semaphore,
Stream,
Expand All @@ -15,7 +19,13 @@ import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
import { buildGraph, type ResolvedGraph } from "./DependencyGraph.ts";
import { type HealthProbeCallbacks, runHealthProbe } from "./HealthProbe.ts";
import { LogBuffer } from "./LogBuffer.ts";
import type { HookTrigger, OrchestratorConfig, RestartPolicy, ServiceDef } from "./ServiceDef.ts";
import type {
HookTrigger,
OrchestratorConfig,
RestartPolicy,
ServiceDef,
ServiceStartOptions,
} from "./ServiceDef.ts";
import { defaults } from "./ServiceDef.ts";
import { initial } from "./ServiceState.ts";
import { makeSupervisedCommand, usesSupervisor } from "./Supervisor.ts";
Expand Down Expand Up @@ -43,11 +53,17 @@ const waitForProcessToStop = (handle: {
export class Orchestrator extends Context.Service<
Orchestrator,
{
readonly start: () => Effect.Effect<void>;
readonly startService: (name: string) => Effect.Effect<void, ServiceNotFoundError>;
readonly start: (options?: ServiceStartOptions) => Effect.Effect<void>;
readonly startService: (
name: string,
options?: ServiceStartOptions,
) => Effect.Effect<void, ServiceNotFoundError>;
readonly stop: () => Effect.Effect<void>;
readonly stopService: (name: string) => Effect.Effect<void, ServiceNotFoundError>;
readonly restartService: (name: string) => Effect.Effect<void, ServiceNotFoundError>;
readonly restartService: (
name: string,
options?: ServiceStartOptions,
) => Effect.Effect<void, ServiceNotFoundError>;
readonly updateServiceDefinition: (
name: string,
def: ServiceDef,
Expand Down Expand Up @@ -180,11 +196,22 @@ export class Orchestrator extends Context.Service<
const shouldRestartOnUnhealthy = (policy: RestartPolicy): boolean => policy !== "no";

// The full lifecycle loop for a single service
const runService = (def: ServiceDef): Effect.Effect<void, SpawnError> =>
const runService = (
def: ServiceDef,
options?: ServiceStartOptions,
): Effect.Effect<void, SpawnError> =>
Effect.gen(function* () {
let restartCount = 0;
const maxRestarts = def.maxRestarts ?? defaults.maxRestarts;
const restartPolicy = def.restart ?? defaults.restart;
const prepareStart = () =>
options
?.beforeStart?.(def.name)
.pipe(
Effect.mapError((cause) => new SpawnError({ service: def.command, cause })),
) ?? Effect.void;

yield* prepareStart();

// Re-create signals on each run (needed for restarts)
const resetSignals = Effect.sync(() => {
Expand Down Expand Up @@ -241,17 +268,46 @@ export class Orchestrator extends Context.Service<
Effect.gen(function* () {
const unhealthyRestart = Deferred.makeUnsafe<void>();
const supervised = usesSupervisor(def);
const spawnGate =
supervised && options?.beforeSpawn !== undefined
? yield* Effect.tryPromise({
try: async () => {
const directory = await mkdtemp(
join(tmpdir(), "process-compose-spawn-gate-"),
);
return {
directory,
requestPath: join(directory, "request"),
releasePath: join(directory, "release"),
};
},
catch: (cause) => new SpawnError({ service: def.command, cause }),
})
: undefined;
if (spawnGate !== undefined) {
yield* Effect.addFinalizer(() =>
Effect.promise(() =>
rm(spawnGate.directory, { recursive: true, force: true }),
),
);
}

// Build command
const cmd = supervised
? makeSupervisedCommand(def)
? makeSupervisedCommand(def, spawnGate)
: ChildProcess.make(def.command, def.args ?? [], {
cwd: def.cwd,
env: def.env,
extendEnv: true,
stdin: "ignore",
});

// Release external resources such as port reservations only
// once dependencies are satisfied and spawning is imminent.
if (!supervised) {
yield* options?.beforeSpawn?.(def.name) ?? Effect.void;
}

// Spawn the process
const handle = yield* spawner
.spawn(cmd)
Expand Down Expand Up @@ -307,6 +363,22 @@ export class Orchestrator extends Context.Service<
),
);

if (spawnGate !== undefined) {
yield* Effect.tryPromise({
try: () => access(spawnGate.requestPath),
catch: (cause) => cause,
}).pipe(
Effect.retry(Schedule.spaced("10 millis")),
Effect.timeout("10 seconds"),
Effect.mapError((cause) => new SpawnError({ service: def.command, cause })),
Comment thread
jgoux marked this conversation as resolved.
);
yield* options?.beforeSpawn?.(def.name) ?? Effect.void;
Comment thread
jgoux marked this conversation as resolved.
yield* Effect.tryPromise({
try: () => writeFile(spawnGate.releasePath, "release", { flag: "wx" }),
catch: (cause) => new SpawnError({ service: def.command, cause }),
});
}

// Transition to Running
yield* sendEvent(def.name, {
_tag: "ProcessSpawned",
Expand Down Expand Up @@ -493,6 +565,10 @@ export class Orchestrator extends Context.Service<
while (shouldRestart(result) && (maxRestarts === 0 || restartCount < maxRestarts)) {
restartCount++;

// The previous process scope has closed, so external resources can
// be reserved safely for the duration of this restart's backoff.
yield* prepareStart();

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Fail unhealthy restarts when port re-reservation fails

When an unhealthy service loses its port before this restart-time beforeStart call, the reservation failure occurs while its state is still Unhealthy. runServiceSafe() then emits SpawnFailed, but the transition table accepts that event only from Pending or Starting, so the fiber exits with no process while the old completed healthy signal can make a later waitReady() report success. Fresh evidence beyond the prior initial pre-start failure thread is that this call runs inside the unhealthy restart loop before RestartTriggered; transition the service to a restart state first or handle SpawnFailed from Unhealthy.

AGENTS.md reference: AGENTS.md:L126-L126

Useful? React with 👍 / 👎.


if (result._tag === "UnhealthyRestart") {
yield* appendRecentServiceLogs(
def.name,
Expand All @@ -516,11 +592,11 @@ export class Orchestrator extends Context.Service<
}
});

const runServiceSafe = (def: ServiceDef) =>
runService(def).pipe(
const runServiceSafe = (def: ServiceDef, options?: ServiceStartOptions) =>
runService(def, options).pipe(
Effect.catch((error) =>
sendEvent(def.name, {
_tag: "DependencyFailed",
_tag: "SpawnFailed",
error: `Spawn failed: ${error.service} - ${String(error.cause)}`,
}).pipe(Effect.asVoid),
),
Expand Down Expand Up @@ -551,23 +627,26 @@ export class Orchestrator extends Context.Service<

const restartClosureFor = (name: string): ReadonlyArray<ServiceDef> => {
const names = new Set<string>([name]);
const collectDependents = (current: string): void => {
const visited = new Set<string>();
const collectDependents = (current: string): boolean => {
if (visited.has(current)) return names.has(current);
visited.add(current);
let hasActiveDependent = false;
for (const dependent of graph.dependentsOf(current)) {
if (names.has(dependent.name)) continue;
names.add(dependent.name);
collectDependents(dependent.name);
const dependentService = services.get(dependent.name);
const dependentIsActive =
FiberMap.hasUnsafe(fibers, dependent.name) ||
(dependentService?.requested === true && dependentService.stoppedByUser !== true);
const descendantIsActive = collectDependents(dependent.name);
if (dependentIsActive || descendantIsActive) {
names.add(dependent.name);
hasActiveDependent = true;
}
}
return hasActiveDependent;
};
collectDependents(name);
return graph.startOrder.filter((def) => {
const service = services.get(def.name);
return (
names.has(def.name) &&
(def.name === name ||
FiberMap.hasUnsafe(fibers, def.name) ||
(service?.requested === true && service.stoppedByUser !== true))
);
});
return graph.startOrder.filter((def) => names.has(def.name));
};

const waitReadySingle = (def: ServiceDef): Effect.Effect<void, ServiceReadyError> =>
Expand Down Expand Up @@ -665,16 +744,16 @@ export class Orchestrator extends Context.Service<
});

return {
start: () =>
start: (options) =>
Effect.gen(function* () {
for (const def of graph.startOrder) {
const service = services.get(def.name);
if (service !== undefined) service.requested = true;
yield* FiberMap.run(fibers, def.name, runServiceSafe(def));
yield* FiberMap.run(fibers, def.name, runServiceSafe(def, options));
}
}),

startService: (name: string) =>
startService: (name: string, options) =>
Effect.gen(function* () {
const def = lookupDef(name);
if (def === undefined) {
Expand Down Expand Up @@ -723,7 +802,9 @@ export class Orchestrator extends Context.Service<
resetNames.add(d.name);
}
if (service !== undefined) service.requested = true;
yield* FiberMap.run(fibers, d.name, runServiceSafe(d), { onlyIfMissing: true });
yield* FiberMap.run(fibers, d.name, runServiceSafe(d, options), {
onlyIfMissing: true,
});
Comment thread
jgoux marked this conversation as resolved.
}

// A caller may retry a dependency directly while dependents started by an
Expand All @@ -743,7 +824,7 @@ export class Orchestrator extends Context.Service<
yield* FiberMap.remove(fibers, d.name);
yield* resetService(d.name);
service.requested = true;
yield* FiberMap.run(fibers, d.name, runServiceSafe(d), {
yield* FiberMap.run(fibers, d.name, runServiceSafe(d, options), {
onlyIfMissing: true,
});
}
Expand Down Expand Up @@ -816,7 +897,7 @@ export class Orchestrator extends Context.Service<
yield* sendEvent(name, { _tag: "ProcessExited", exitCode: 143 });
}),

restartService: (name: string) =>
restartService: (name: string, options) =>
Effect.gen(function* () {
const def = lookupDef(name);
if (def === undefined) {
Expand All @@ -833,7 +914,7 @@ export class Orchestrator extends Context.Service<
for (const affectedDef of affected) {
const service = services.get(affectedDef.name);
if (service !== undefined) service.requested = true;
yield* FiberMap.run(fibers, affectedDef.name, runServiceSafe(affectedDef));
yield* FiberMap.run(fibers, affectedDef.name, runServiceSafe(affectedDef, options));

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Re-reserve ports before tearing down the restart closure

When restarting a running service with active dependents, restartService() completes every stopForRestart() at lines 908-910 before these relaunch fibers can execute beforeStart and reacquire their leases. Ports belonging to dependents stopped early in the closure can consequently remain unreserved throughout the rest of the shutdown sequence, allowing another process to bind them and make the eventual restart fail. Fresh evidence after the prior restart-lease thread is this current stop-all-then-relaunch ordering; reacquire each reservation immediately when its process scope closes rather than waiting until all affected services are stopped.

AGENTS.md reference: AGENTS.md:L123-L126

Useful? React with 👍 / 👎.

}
}),

Expand Down
Loading
Loading