From 146ef9d6ed1af547294fa52fd88b9ee4617b38b3 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 17 Aug 2026 17:15:06 +0000 Subject: [PATCH 1/5] fix(stack): settle public service state before stopService returns stopService returned before the background allStateChanges fiber had re-projected the public state, so a getState immediately after a stop could still observe the previous status (e.g. Dormant instead of Stopped). Every start path already settles the projection before returning via waitForTargets; do the same in stopService. This removes the 20ms sleep the restart-preservation unit test used to paper over the race, which intermittently failed under CI load. Also make the functions-reload test's write-failure injection privilege-independent: replace the workspace directory with a plain file instead of relying on permission bits, which root bypasses. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01McsNM9yxC5tiY6SusBALoq --- packages/stack/src/LocalStack.ts | 3 +++ packages/stack/src/Stack.unit.test.ts | 19 +++++++++---------- 2 files changed, 12 insertions(+), 10 deletions(-) diff --git a/packages/stack/src/LocalStack.ts b/packages/stack/src/LocalStack.ts index e2f7e8d24f..91b69ee2ae 100644 --- a/packages/stack/src/LocalStack.ts +++ b/packages/stack/src/LocalStack.ts @@ -806,6 +806,9 @@ export const localStackLayer = ( ).toReversed()) { yield* runtime.orchestrator.stopService(target); } + // Settle the public projection before returning so callers observe + // the stop immediately, matching the start/restart/waitReady paths. + yield* syncRuntimeProjectedStates(runtime); }).pipe(withLifecycleLock), restartService: (name) => Effect.gen(function* () { diff --git a/packages/stack/src/Stack.unit.test.ts b/packages/stack/src/Stack.unit.test.ts index c5b569e441..ebad7c5423 100644 --- a/packages/stack/src/Stack.unit.test.ts +++ b/packages/stack/src/Stack.unit.test.ts @@ -3,7 +3,7 @@ import { BunServices } from "@effect/platform-bun"; import { buildGraph } from "@supabase/process-compose"; import { createHmac } from "node:crypto"; import { mkdtempSync } from "node:fs"; -import { chmod, readFile, rm } from "node:fs/promises"; +import { readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { Deferred, Effect, Exit, Fiber, Layer, Stream } from "effect"; @@ -249,13 +249,18 @@ describe("Stack", () => { ).toBe("StackBuildError"); expect((yield* readRuntimeConfig).env.SHARED).toBe("replacement-secret"); + // Replace the workspace directory with a plain file so the config write + // fails for any user — permission bits alone are bypassed by root. const runtimeDirectory = join(runtimeRoot, "edge-runtime"); - yield* Effect.promise(() => chmod(runtimeDirectory, 0o500)); + yield* Effect.promise(async () => { + await rm(runtimeDirectory, { recursive: true, force: true }); + await writeFile(runtimeDirectory, ""); + }); const failedBundle = functionsBundle(runtimeRoot, "failed-secret"); const error = yield* stack.reloadFunctions({ functions: failedBundle }).pipe(Effect.flip); expect(error._tag).toBe("StackBuildError"); - yield* Effect.promise(() => chmod(runtimeDirectory, 0o700)); + yield* Effect.promise(() => rm(runtimeDirectory)); yield* stack.reloadFunctions(); expect((yield* readRuntimeConfig).env.SHARED).toBe("replacement-secret"); @@ -270,12 +275,7 @@ describe("Stack", () => { ).toBe(false); }).pipe( Effect.provide(layer), - Effect.ensuring( - Effect.promise(async () => { - await chmod(join(runtimeRoot, "edge-runtime"), 0o700).catch(() => {}); - await rm(runtimeRoot, { recursive: true, force: true }); - }), - ), + Effect.ensuring(Effect.promise(() => rm(runtimeRoot, { recursive: true, force: true }))), Effect.timeout("5 seconds"), ); }); @@ -1104,7 +1104,6 @@ describe("Stack", () => { const stack = yield* Stack; yield* stack.start(); yield* stack.stopService("auth"); - yield* Effect.sleep("20 millis"); expect((yield* stack.getState("auth")).status).toBe("Stopped"); yield* stack.stop(); From ddafa47e8f3be52ec1eb9d26ce8d6ac56bc0d550 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 08:36:09 +0000 Subject: [PATCH 2/5] test(stack): root-proof the repair write-failure injection The moved-workspace repair test blocked the partial write with chmod, which root bypasses, so the repair unexpectedly succeeded when the suite ran in a privileged sandbox. Gate writeFileString at the FileSystem seam instead (same pattern as the lifecycle race test) so the injected failure is deterministic for any user. Also resolve the develop merge in favor of asserting getState directly after stopService, now that stopService settles the public projection before returning. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01McsNM9yxC5tiY6SusBALoq --- ...naged-manager-recovery.integration.test.ts | 54 ++++++++++++++----- 1 file changed, 42 insertions(+), 12 deletions(-) diff --git a/packages/stack/src/managed-manager-recovery.integration.test.ts b/packages/stack/src/managed-manager-recovery.integration.test.ts index 4e9435154c..ea00ec469c 100644 --- a/packages/stack/src/managed-manager-recovery.integration.test.ts +++ b/packages/stack/src/managed-manager-recovery.integration.test.ts @@ -1,15 +1,18 @@ import { it } from "@effect/vitest"; import { NodeFileSystem, NodePath } from "@effect/platform-node"; -import { Cause, Deferred, Effect, Exit, Fiber, Layer, ManagedRuntime } from "effect"; -import { HttpServer } from "effect/unstable/http"; import { - chmodSync, - mkdirSync, - mkdtempSync, - realpathSync, - renameSync, - writeFileSync, -} from "node:fs"; + Cause, + Deferred, + Effect, + Exit, + Fiber, + FileSystem, + Layer, + ManagedRuntime, + PlatformError, +} from "effect"; +import { HttpServer } from "effect/unstable/http"; +import { mkdirSync, mkdtempSync, realpathSync, renameSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, describe, expect } from "vitest"; @@ -131,6 +134,33 @@ describe("managed stack recovery journeys", () => { it.live("repairs a moved workspace without changing stack id or ports", () => { const { layer, stateRoot } = setup(); + // Permission bits cannot block writes when tests run as root, so gate the + // FileSystem seam instead to force the partial-repair failure. + const blockedWrites = { root: undefined as string | undefined }; + const blockingFileSystemLayer = Layer.effect( + FileSystem.FileSystem, + Effect.gen(function* () { + const base = yield* FileSystem.FileSystem; + return { + ...base, + writeFileString: ( + path: string, + data: string, + options?: Parameters[2], + ) => + blockedWrites.root !== undefined && path.startsWith(blockedWrites.root) + ? Effect.fail( + PlatformError.systemError({ + _tag: "PermissionDenied", + module: "FileSystem", + method: "writeFileString", + pathOrDescriptor: path, + }), + ) + : base.writeFileString(path, data, options), + } satisfies FileSystem.FileSystem; + }), + ).pipe(Layer.provide(NodeFileSystem.layer)); return Effect.scoped( Effect.gen(function* () { const root = mkdtempSync(join(tmpdir(), "managed-repair-test-")); @@ -177,9 +207,9 @@ describe("managed stack recovery journeys", () => { const blockedId = [originalId, secondaryId].sort().at(-1); if (blockedId === undefined) throw new Error("expected affected stack"); const blockedRoot = managedStackPaths(stateRoot, blockedId).root; - chmodSync(blockedRoot, 0o500); + blockedWrites.root = blockedRoot; const failed = yield* manager.repairWorkspace(discovery.repair).pipe(Effect.exit); - chmodSync(blockedRoot, 0o700); + blockedWrites.root = undefined; expect(Exit.isFailure(failed)).toBe(true); const firstUpdatedId = [originalId, secondaryId].sort().at(0); if (firstUpdatedId === undefined) throw new Error("expected affected stack"); @@ -204,7 +234,7 @@ describe("managed stack recovery journeys", () => { }), ).pipe( Effect.provide(layer), - Effect.provide(NodeFileSystem.layer), + Effect.provide(blockingFileSystemLayer), Effect.provide(NodePath.layer), Effect.provide(gitConfigStoreLayer), Effect.provide(controlTransportLayer), From 82a4ef2145a46dc855aa411b2f5c393e5ffb6a62 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 09:53:36 +0000 Subject: [PATCH 3/5] fix(stack): fall through control endpoint candidates on port collisions The control endpoint was derived from two bytes of the stack id onto a single fixed loopback port, so two live stacks could birthday-collide and the later acquirer failed with ControlAddressConflictError. This intermittently broke parallel integration runs (managed-manager-ports "keeps automatic ports exclusive while stopped" on CI) and can equally break real stacks on one machine. Derive a deterministic sequence of candidate endpoints instead. An acquirer first scans all candidates for a live owner of its id (the scan read doubles as the attach handshake, keeping the one-read contract scripted-owner tests rely on), then binds the first free candidate, skipping ports occupied by other stacks or unrelated listeners. Acquisition only conflicts when every candidate is occupied; a protocol mismatch still fails closed. probeControl scans the same sequence and returns the owner's actual endpoint, which connectManagedStack now uses instead of re-deriving the primary port. Exact service-port requests reserve every candidate of every stack. The concurrent-start liveness test now holds only the status probe's first read in flight, since acquisition legitimately reads candidates before binding while real transport reads are bounded by a timeout. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01McsNM9yxC5tiY6SusBALoq --- packages/stack/docs/architecture.md | 31 +-- packages/stack/src/discovery.ts | 2 +- .../src/managed-control.integration.test.ts | 64 +++-- ...naged-manager-projects.integration.test.ts | 20 +- packages/stack/src/managed/control.ts | 220 +++++++++++++----- packages/stack/src/managed/lifecycle.ts | 35 +-- packages/stack/src/managed/manager.ts | 14 +- 7 files changed, 262 insertions(+), 124 deletions(-) diff --git a/packages/stack/docs/architecture.md b/packages/stack/docs/architecture.md index 742cc3bbd6..d89005bcc9 100644 --- a/packages/stack/docs/architecture.md +++ b/packages/stack/docs/architecture.md @@ -101,9 +101,10 @@ runtime logs, and `runtime/` for supervisor-owned runtime files. The `ManagedStackManager` is the only component that writes `stack.json`. Control ownership is the liveness and mutation authority. `acquireControl` -returns `Owned` for the process that bound the deterministic endpoint or -`Attached` for a live owner. An attached caller uses the owner's endpoint for -runtime requests; it never edits the document directly. +returns `Owned` for the process that bound one of the deterministic endpoint +candidates or `Attached` for a live owner found on any candidate. An attached +caller uses the owner's actual endpoint for runtime requests; it never edits +the document directly. ### Start and attach @@ -121,8 +122,8 @@ runtime requests; it never edits the document directly. `stack.start()` over the control transport when service startup is needed. `connectManagedStack` reads the document, probes the deterministic endpoint -without binding it, and returns a `RemoteStack` only when the owner reports a -ready running state. Read-only status and discovery therefore do not claim an +candidates without binding them, and returns a `RemoteStack` against the +owner's actual endpoint only when the owner reports a ready running state. Read-only status and discovery therefore do not claim an endpoint; mutating operations acquire control ownership. ### Update, stop, and delete @@ -200,14 +201,18 @@ persisted endpoint, or another stack's reservation under the normal exact-port rules. A persisted automatic assignment in the control range is invalid and fails loudly rather than being silently migrated. -The control endpoint is derived from the stack id and served on loopback. A -persisted endpoint is accepted only when it matches that derivation. A rare -hash collision or unrelated listener makes control acquisition fail with a -typed conflict; a read-only probe treats the address as non-live and never -claims it. An exact service port can still equal the future endpoint of an -identity that has never started, so that low-probability conflict is rejected -when ownership is acquired rather than forbidding every explicit port in the -reserved range. +The control endpoint is derived from the stack id and served on loopback. The +derivation yields a short deterministic candidate sequence rather than a +single port: an owner binds the first free candidate, skipping candidates +occupied by other stacks or unrelated listeners, and readers scan the same +sequence and match the published `ownershipId`. A hash collision between two +stack ids therefore degrades to the collided stack binding its next candidate +instead of failing. Acquisition fails with a typed conflict only when every +candidate is occupied by a foreign listener; a read-only probe treats an +address with no matching owner as non-live and never claims it. An exact +service port can still equal a candidate of an identity that has never +started, so every stack's full candidate set is reserved against exact-port +requests rather than forbidding every explicit port in the reserved range. This is deliberately a small single-user localhost mechanism. The control protocol has no token authentication; ownership, endpoint identity, and diff --git a/packages/stack/src/discovery.ts b/packages/stack/src/discovery.ts index cf21f60d15..cddeec124e 100644 --- a/packages/stack/src/discovery.ts +++ b/packages/stack/src/discovery.ts @@ -74,7 +74,7 @@ const liveStatus = ( ): Effect.Effect => manager .probeControl(document.id) - .pipe(Effect.map((status) => status?.state === "running" && status.ready)); + .pipe(Effect.map((probe) => probe?.status.state === "running" && probe.status.ready)); export const listStacks = (opts: { readonly cacheRoot: string; diff --git a/packages/stack/src/managed-control.integration.test.ts b/packages/stack/src/managed-control.integration.test.ts index 0dbd8fe5e2..c3649c0058 100644 --- a/packages/stack/src/managed-control.integration.test.ts +++ b/packages/stack/src/managed-control.integration.test.ts @@ -7,10 +7,13 @@ import { describe, expect } from "vitest"; import { DaemonServer } from "./DaemonServer.ts"; import { acquireControl, + CONTROL_CANDIDATE_COUNT, controlEndpoint, + controlEndpointCandidates, ControlBindError, ControlTransport, ControlTransportError, + probeControl, } from "./managed/control.ts"; import { controlTransportLayer } from "./platform-node.ts"; import { httpTransportClientLayer } from "./HttpTransportClient.ts"; @@ -103,10 +106,15 @@ const spawnBoundChild = (port: number) => { }; describe("managed control endpoint", () => { - it.live("derives one deterministic loopback endpoint from the ownership id", () => { + it.live("derives deterministic loopback candidates from the ownership id", () => { return Effect.sync(() => { const endpoint = Effect.runSync(controlEndpoint(STACK_ID)); expect(endpoint.url).toBe("http://127.0.0.1:13737"); + const candidates = Effect.runSync(controlEndpointCandidates(STACK_ID)); + expect(candidates).toHaveLength(CONTROL_CANDIDATE_COUNT); + expect(candidates.map(({ port }) => port)).toEqual( + Array.from({ length: CONTROL_CANDIDATE_COUNT }, (_, offset) => 13737 + offset), + ); }); }); @@ -254,7 +262,7 @@ describe("managed control endpoint", () => { ), ); - it.live("rejects a valid owner with a colliding deterministic endpoint", () => + it.live("claims the next candidate when another stack owns the first", () => Effect.scoped( live( Effect.gen(function* () { @@ -272,15 +280,20 @@ describe("managed control endpoint", () => { ), ); yield* Effect.promise(() => daemonRuntime.runPromise(DaemonServer)); - const contender = yield* acquireControl({ stackId: COLLIDING_STACK_ID }).pipe( - Effect.exit, - ); - expect(Exit.isFailure(contender)).toBe(true); - if (Exit.isFailure(contender)) { - expect(Cause.squash(contender.cause)).toMatchObject({ - _tag: "ControlAddressConflictError", - }); - } + const contender = yield* acquireControl({ stackId: COLLIDING_STACK_ID }); + if (contender._tag !== "Owned") throw new Error("expected contender ownership"); + expect(contender.endpoint.port).not.toBe(owner.endpoint.port); + + // Readers locate each owner at its actual candidate. + const ownerProbe = yield* probeControl(STACK_ID); + expect(ownerProbe?.endpoint.port).toBe(owner.endpoint.port); + const contenderProbe = yield* probeControl(COLLIDING_STACK_ID); + expect(contenderProbe?.endpoint.port).toBe(contender.endpoint.port); + + // A second caller for the collided stack attaches to its owner. + const attached = yield* acquireControl({ stackId: COLLIDING_STACK_ID }); + expect(attached._tag).toBe("Attached"); + expect(attached.endpoint.port).toBe(contender.endpoint.port); yield* Effect.promise(() => daemonRuntime.dispose()); }), ), @@ -306,15 +319,37 @@ describe("managed control endpoint", () => { ), ); - it.live("rejects an unrelated listener without taking it over", () => + it.live("claims the next candidate without taking over an unrelated listener", () => live( Effect.scoped( Effect.gen(function* () { - const endpoint = yield* controlEndpoint(STACK_ID); + const candidates = yield* controlEndpointCandidates(STACK_ID); const unrelated = yield* Effect.acquireRelease( - Effect.promise(() => listenRaw(endpoint.port)), + Effect.promise(() => listenRaw(candidates[0]!.port)), (server) => Effect.promise(() => closeRaw(server)), ); + const owner = yield* acquireControl({ stackId: STACK_ID }); + if (owner._tag !== "Owned") throw new Error("expected control ownership"); + expect(owner.endpoint.port).toBe(candidates[1]!.port); + expect(unrelated.listening).toBe(true); + const probe = yield* probeControl(STACK_ID); + expect(probe?.endpoint.port).toBe(candidates[1]!.port); + }), + ), + ), + ); + + it.live("fails once every candidate is occupied by unrelated listeners", () => + live( + Effect.scoped( + Effect.gen(function* () { + const candidates = yield* controlEndpointCandidates(STACK_ID); + yield* Effect.forEach(candidates, (candidate) => + Effect.acquireRelease( + Effect.promise(() => listenRaw(candidate.port)), + (server) => Effect.promise(() => closeRaw(server)), + ), + ); const result = yield* acquireControl({ stackId: STACK_ID, }).pipe( @@ -325,7 +360,6 @@ describe("managed control endpoint", () => { ); expect(result._tag).toBe("Left"); if (result._tag === "Left") expect(result.error._tag).toBe("ControlAddressConflictError"); - expect(unrelated.listening).toBe(true); }), ), ), diff --git a/packages/stack/src/managed-manager-projects.integration.test.ts b/packages/stack/src/managed-manager-projects.integration.test.ts index c14e1f8311..328e810e8e 100644 --- a/packages/stack/src/managed-manager-projects.integration.test.ts +++ b/packages/stack/src/managed-manager-projects.integration.test.ts @@ -53,17 +53,23 @@ describe("managed stack projects journeys", () => { const baseTransport = yield* ControlTransport; const readStarted = yield* Deferred.make(); const continueRead = yield* Deferred.make(); + // Hold only the status probe's first read in flight. Later reads (the + // concurrent acquire scans its endpoint candidates before binding) must + // pass through, mirroring the real transport's bounded read timeout. let gateReads = false; + let gatedRead = false; const gatedTransport = Layer.succeed(ControlTransport, { ...baseTransport, read: (endpoint) => - gateReads - ? Effect.gen(function* () { - yield* Deferred.succeed(readStarted, void 0); - yield* Deferred.await(continueRead); - return yield* baseTransport.read(endpoint); - }) - : baseTransport.read(endpoint), + Effect.suspend(() => { + if (!gateReads || gatedRead) return baseTransport.read(endpoint); + gatedRead = true; + return Effect.gen(function* () { + yield* Deferred.succeed(readStarted, void 0); + yield* Deferred.await(continueRead); + return yield* baseTransport.read(endpoint); + }); + }), }); yield* Effect.scoped( diff --git a/packages/stack/src/managed/control.ts b/packages/stack/src/managed/control.ts index 122b92dbff..af784bb1c5 100644 --- a/packages/stack/src/managed/control.ts +++ b/packages/stack/src/managed/control.ts @@ -69,7 +69,7 @@ export class ControlAddressConflictError extends Data.TaggedError("ControlAddres readonly cause: unknown; }> { override get message(): string { - return `Control endpoint ${this.endpoint.url} is occupied by a non-Supabase listener`; + return `Control endpoint ${this.endpoint.url} is occupied by another listener`; } } @@ -148,20 +148,41 @@ const ownershipBytes = (ownershipId: string): ReadonlyArray => { return bytes; }; -/** Derives a deterministic loopback address and reserved port from a stack id. */ -export const controlEndpoint = ( +/** + * Number of deterministic endpoints derived per ownership id. Two ids can + * hash to the same primary port, so owners fall through to the next + * candidate and readers scan the same sequence, matching on `ownershipId`. + */ +export const CONTROL_CANDIDATE_COUNT = 8; + +const CONTROL_RANGE_SIZE = CONTROL_PORT_RANGE.max - CONTROL_PORT_RANGE.min + 1; + +const endpointForValue = (value: number): ControlEndpoint => { + const port = CONTROL_PORT_RANGE.min + (value % CONTROL_RANGE_SIZE); + const host = "127.0.0.1"; + return { hostname: host, port, url: `http://${host}:${port}` }; +}; + +/** Derives the deterministic loopback endpoint candidates for a stack id. */ +export const controlEndpointCandidates = ( ownershipId: string, -): Effect.Effect => { +): Effect.Effect, InvalidControlOwnershipIdError> => { if (!CONTROL_ID_PATTERN.test(ownershipId)) return invalidId(ownershipId); const bytes = ownershipBytes(ownershipId); const value = (bytes[3]! << 8) | bytes[4]!; - const port = - CONTROL_PORT_RANGE.min + (value % (CONTROL_PORT_RANGE.max - CONTROL_PORT_RANGE.min + 1)); - const host = "127.0.0.1"; - const url = `http://${host}:${port}`; - return Effect.succeed({ hostname: host, port, url }); + return Effect.succeed( + Array.from({ length: CONTROL_CANDIDATE_COUNT }, (_, offset) => + endpointForValue(value + offset), + ), + ); }; +/** Derives the primary deterministic endpoint (first candidate) for a stack id. */ +export const controlEndpoint = ( + ownershipId: string, +): Effect.Effect => + Effect.map(controlEndpointCandidates(ownershipId), (candidates) => candidates[0]!); + const decodeOwnerStatus = ( endpoint: ControlEndpoint, value: unknown, @@ -224,22 +245,40 @@ const readOwnerStatus = ( ), ); -/** Reads an existing owner without ever claiming its deterministic endpoint. */ +/** A located owner: its published status and the candidate it bound. */ +export interface ControlProbe { + readonly status: ControlOwnerStatus; + readonly endpoint: ControlEndpoint; +} + +/** Reads an existing owner wherever it bound, without claiming an endpoint. */ export const probeControl = ( ownershipId: string, -): Effect.Effect< - ControlOwnerStatus | undefined, - InvalidControlOwnershipIdError, - ControlTransport -> => +): Effect.Effect => Effect.gen(function* () { - const endpoint = yield* controlEndpoint(ownershipId); + const candidates = yield* controlEndpointCandidates(ownershipId); const transport = yield* ControlTransport; - return yield* readOwnerStatus(endpoint, ownershipId, transport).pipe( - Effect.catch(() => Effect.succeed(undefined)), - ); + for (const endpoint of candidates) { + const status = yield* readOwnerStatus(endpoint, ownershipId, transport).pipe( + Effect.catch(() => Effect.succeed(undefined)), + ); + if (status !== undefined) return { status, endpoint }; + } + return undefined; }); +const makeAttached = ( + endpoint: ControlEndpoint, + ownershipId: string, + transport: ControlTransportShape, +): ControlAttached => ({ + _tag: "Attached", + ownershipId, + endpoint, + ownerStatus: readOwnerStatus(endpoint, ownershipId, transport), + requestStop: transport.requestStop(endpoint), +}); + const attach = ( endpoint: ControlEndpoint, ownershipId: string, @@ -252,13 +291,7 @@ const attach = ( | ControlAddressConflictError > => readOwnerStatus(endpoint, ownershipId, transport).pipe( - Effect.map(() => ({ - _tag: "Attached" as const, - ownershipId, - endpoint, - ownerStatus: readOwnerStatus(endpoint, ownershipId, transport), - requestStop: transport.requestStop(endpoint), - })), + Effect.map(() => makeAttached(endpoint, ownershipId, transport)), ); const makeOwned = ( @@ -295,8 +328,33 @@ const makeOwned = ( }); }; -const acquireAtEndpoint = ( - endpoint: ControlEndpoint, +/** + * Finds the candidate a live owner of `ownershipId` bound, if any. Foreign + * owners, non-Supabase listeners, and free ports are skipped; a protocol + * mismatch fails closed because a newer owner of this very stack may be + * publishing there, and claiming another candidate beside it would split + * ownership across versions. + */ +const scanForOwner = ( + candidates: ReadonlyArray, + ownershipId: string, + transport: ControlTransportShape, +): Effect.Effect => + Effect.gen(function* () { + for (const endpoint of candidates) { + const found = yield* readOwnerStatus(endpoint, ownershipId, transport).pipe( + Effect.map(() => true), + Effect.catchTag("ControlTransportError", () => Effect.succeed(false)), + Effect.catchTag("ControlProtocolError", () => Effect.succeed(false)), + Effect.catchTag("ControlAddressConflictError", () => Effect.succeed(false)), + ); + if (found) return endpoint; + } + return undefined; + }); + +const acquireAtCandidates = ( + candidates: ReadonlyArray, ownershipId: string, status: ControlOwnerStatus, transport: ControlTransportShape, @@ -321,36 +379,78 @@ const acquireAtEndpoint = ( | ControlUnavailableError, import("effect/Scope").Scope > = Effect.gen(function* () { - const bound = yield* transport - .bind( - endpoint, - () => Ref.getUnsafe(statusRef), - () => { - Effect.runSync(Deferred.succeed(stopRequested, void 0)); - }, - ) - .pipe(Effect.result); - if (Result.isSuccess(bound)) { - const owned = yield* makeOwned( + // An existing owner may hold any candidate: an earlier occupant can have + // freed a lower port since the owner bound. Attach before claiming one so + // a stack never ends up with two owners on different candidates. The scan + // read doubles as the attach handshake, so an owner is read exactly once. + const ownerEndpoint = yield* scanForOwner(candidates, ownershipId, transport); + if (ownerEndpoint !== undefined) { + return makeAttached(ownerEndpoint, ownershipId, transport); + } + let pending: ControlUnavailableError | undefined; + let conflict: ControlAddressConflictError | undefined; + for (const endpoint of candidates) { + const bound = yield* transport + .bind( + endpoint, + () => Ref.getUnsafe(statusRef), + () => { + Effect.runSync(Deferred.succeed(stopRequested, void 0)); + }, + ) + .pipe(Effect.result); + if (Result.isSuccess(bound)) { + const owned = yield* makeOwned( + endpoint, + ownershipId, + bound.success, + statusRef, + stopRequested, + ); + yield* Effect.addFinalizer(() => owned.close); + return owned; + } + const error = bound.failure; + if (error.reason !== "in-use") return yield* Effect.fail(error); + // The address was taken between the scan and the bind: attach if the + // occupant is our owner, retry the walk if it is not serving yet, and + // move to the next candidate if it belongs to someone else. + const attached: ControlAcquisition | undefined = yield* attach( endpoint, ownershipId, - bound.success, - statusRef, - stopRequested, + transport, + ).pipe( + Effect.map((acquisition): ControlAcquisition | undefined => acquisition), + Effect.catchTag("ControlAddressConflictError", (cause) => + Effect.sync(() => { + conflict = cause; + return undefined; + }), + ), + Effect.catchTag("ControlProtocolError", (cause) => + Effect.sync(() => { + conflict = new ControlAddressConflictError({ endpoint, cause }); + return undefined; + }), + ), + Effect.catchTag("ControlTransportError", (cause) => + cause.reason === "unreachable" + ? Effect.sync(() => { + pending = unavailable(endpoint, cause); + return undefined; + }) + : Effect.fail(cause), + ), ); - yield* Effect.addFinalizer(() => owned.close); - return owned; + if (attached !== undefined) return attached; } - const error = bound.failure; - if (error.reason !== "in-use") return yield* Effect.fail(error); - return yield* attach(endpoint, ownershipId, transport).pipe( - Effect.mapError((cause) => - cause._tag === "ControlTransportError" && cause.reason === "unreachable" - ? unavailable(endpoint, cause) - : cause._tag === "ControlProtocolError" - ? new ControlAddressConflictError({ endpoint, cause }) - : cause, - ), + if (pending !== undefined) return yield* Effect.fail(pending); + return yield* Effect.fail( + conflict ?? + new ControlAddressConflictError({ + endpoint: candidates[0]!, + cause: new Error("Every control endpoint candidate is occupied"), + }), ); }); @@ -362,12 +462,14 @@ const acquireAtEndpoint = ( while: (error) => error._tag === "ControlUnavailableError", }), Effect.catchTag("ControlUnavailableError", (error) => - Effect.fail(new ControlAddressConflictError({ endpoint, cause: error.cause })), + Effect.fail( + new ControlAddressConflictError({ endpoint: error.endpoint, cause: error.cause }), + ), ), ); }; -/** Acquires the deterministic loopback listener or attaches to its owner. */ +/** Acquires a deterministic loopback listener or attaches to its owner. */ export const acquireControl = ( input: ControlOwnershipInput, ): Effect.Effect< @@ -381,10 +483,10 @@ export const acquireControl = ( ControlTransport | import("effect/Scope").Scope > => Effect.gen(function* () { - const endpoint = yield* controlEndpoint(input.stackId); + const candidates = yield* controlEndpointCandidates(input.stackId); const transport = yield* ControlTransport; - return yield* acquireAtEndpoint( - endpoint, + return yield* acquireAtCandidates( + candidates, input.stackId, defaultStatus(input.stackId, input.initialStatus), transport, diff --git a/packages/stack/src/managed/lifecycle.ts b/packages/stack/src/managed/lifecycle.ts index 90cb704711..2391eb6624 100644 --- a/packages/stack/src/managed/lifecycle.ts +++ b/packages/stack/src/managed/lifecycle.ts @@ -9,13 +9,12 @@ import { HttpTransportClient, HttpTransportClientError } from "../HttpTransportC import type { ManagedStackDocument } from "./document.ts"; import { ManagedStackAttachedError, - ManagedStackControlRequiredError, ManagedStackManager, workspaceRepairConflict, type ManagedStackManagerError, type ManagedStackLaunchUpdate, } from "./manager.ts"; -import { ControlTransportError, controlEndpoint, type ControlEndpoint } from "./control.ts"; +import { ControlTransportError } from "./control.ts"; import { ManagedStackNotStoppedError, type ManagedPortIntentDocument, @@ -70,28 +69,11 @@ export const resolveManagedDocument = ( return document === undefined ? yield* Effect.fail(noRunningStack(input)) : document; }); -const runtimeEndpoint = ( - document: ManagedStackDocument, - input: ManagedLifecycleInput, -): Effect.Effect => - Effect.gen(function* () { - if ( - (document.lifecycle !== "running" && document.lifecycle !== "starting") || - (document.lifecycle === "running" && document.runtime?.controlEndpoint === undefined) - ) { - return yield* Effect.fail(noRunningStack(input)); - } - const endpoint = yield* controlEndpoint(document.id).pipe( - Effect.mapError(() => new ManagedStackControlRequiredError({ stackId: document.id })), - ); - return endpoint; - }); - class ManagedStopPending extends Data.TaggedError("ManagedStopPending")<{}> {} class ManagedStopOwnerTerminal extends Data.TaggedError("ManagedStopOwnerTerminal")<{}> {} class ManagedDeletePending extends Data.TaggedError("ManagedDeletePending")<{}> {} -/** Connect to the deterministic endpoint persisted by the managed supervisor. */ +/** Connect to the control endpoint the managed supervisor actually bound. */ export const connectManagedStack = ( input: ManagedLifecycleInput, ): Effect.Effect< @@ -101,14 +83,19 @@ export const connectManagedStack = ( > => Effect.gen(function* () { const document = yield* resolveManagedDocument(input); + if ( + (document.lifecycle !== "running" && document.lifecycle !== "starting") || + (document.lifecycle === "running" && document.runtime?.controlEndpoint === undefined) + ) { + return yield* Effect.fail(noRunningStack(input)); + } const manager = yield* ManagedStackManager; - const status = yield* manager.probeControl(document.id); - if (status?.state !== "running" || !status.ready) { + const probe = yield* manager.probeControl(document.id); + if (probe === undefined || probe.status.state !== "running" || !probe.status.ready) { return yield* Effect.fail(noRunningStack(input)); } - const endpoint = yield* runtimeEndpoint(document, input); const client = yield* HttpTransportClient; - return RemoteStack.layer(endpoint).pipe( + return RemoteStack.layer(probe.endpoint).pipe( Layer.provide(Layer.succeed(HttpTransportClient, client)), ); }); diff --git a/packages/stack/src/managed/manager.ts b/packages/stack/src/managed/manager.ts index 585ecaabfc..4c118de157 100644 --- a/packages/stack/src/managed/manager.ts +++ b/packages/stack/src/managed/manager.ts @@ -23,12 +23,12 @@ import { import { acquireControl, CONTROL_PORT_RANGE, - controlEndpoint, + controlEndpointCandidates, ControlTransport, probeControl, type ControlAcquisition, - type ControlOwnerStatus, type ControlOwnership, + type ControlProbe, } from "./control.ts"; import { discoverEnvironment, @@ -204,7 +204,7 @@ export interface ManagedStackManagerShape { ) => Effect.Effect; readonly probeControl: ( stackId: string, - ) => Effect.Effect; + ) => Effect.Effect; readonly readStack: ( request: ReadStackRequest, ) => Effect.Effect; @@ -435,7 +435,9 @@ const makeManager = ( ); } const strictReserved = new Set(); - const exactReserved = new Set([(yield* controlEndpoint(request.stackId)).port]); + const exactReserved = new Set( + (yield* controlEndpointCandidates(request.stackId)).map(({ port }) => port), + ); const owners = new Map< number, ReadonlyArray<{ @@ -444,7 +446,9 @@ const makeManager = ( }> >(); for (const listing of listings.filter(isHealthyDocument)) { - exactReserved.add((yield* controlEndpoint(listing.document.id)).port); + for (const candidate of yield* controlEndpointCandidates(listing.document.id)) { + exactReserved.add(candidate.port); + } if (listing.document.id === request.stackId) continue; for (const assignment of listing.document.ports) { const liveExact = From 65662f9ce6b503d316795ae6fdbfbc77d3cccccd Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 10:18:38 +0000 Subject: [PATCH 4/5] test(stack): harden managed suites against parallel-run port and watcher races Three independent flakes surfaced while stress-running the integration suite under CPU load: - Managed tests persisted catalog default ports (54321...) on first automatic allocation, and sticky reuse re-reserves them exactly, so a restarting stack raced every other file's default-port availability probes. Automatic selection now takes an injectable preferCatalogDefaults option (production keeps the defaults) and the managed test layers disable it, keeping test stacks on isolated ports without changing the spec'd no-relocation reuse semantics. - The supervisor test file watchers treated any watcher error as fatal, but the runtime's directory watcher can report ENOENT when a watched entry (an atomic-write temp file) vanishes mid-scan. The helpers now re-arm the watch on ENOENT and re-check, keeping their timeouts as the only failure path. - Sub-second synchronization timeouts in the recovery and lifecycle tests (1-2s awaits and polls) expired under load; they are guards, not timing assertions, and now allow 5-15 seconds. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01McsNM9yxC5tiY6SusBALoq --- ...aged-manager-lifecycle.integration.test.ts | 6 +- ...naged-manager-recovery.integration.test.ts | 10 +-- packages/stack/src/managed/manager.ts | 21 ++++-- packages/stack/src/managed/port-plan.ts | 12 +++- .../stack/src/supervisor.integration.test.ts | 64 +++++++++++++++---- .../stack/tests/helpers/managed-manager.ts | 2 +- .../stack/tests/helpers/supervisor-child.ts | 20 +++++- 7 files changed, 103 insertions(+), 32 deletions(-) diff --git a/packages/stack/src/managed-manager-lifecycle.integration.test.ts b/packages/stack/src/managed-manager-lifecycle.integration.test.ts index f75f0df0d0..dcba0d8407 100644 --- a/packages/stack/src/managed-manager-lifecycle.integration.test.ts +++ b/packages/stack/src/managed-manager-lifecycle.integration.test.ts @@ -133,7 +133,9 @@ describe("managed stack lifecycle journeys", () => { ? Effect.succeed(current) : Effect.fail(new Error("stop pending")), ), - Effect.retry(Schedule.spaced("10 millis").pipe(Schedule.upTo({ duration: "2 seconds" }))), + Effect.retry( + Schedule.spaced("10 millis").pipe(Schedule.upTo({ duration: "10 seconds" })), + ), ); yield* owner.close; yield* Fiber.join(stopFiber); @@ -179,7 +181,7 @@ describe("managed stack lifecycle journeys", () => { } satisfies FileSystem.FileSystem; }), ).pipe(Layer.provide(NodeFileSystem.layer)); - const managerLayer = managedStackManagerLayer({ stateRoot }).pipe( + const managerLayer = managedStackManagerLayer({ stateRoot, preferCatalogDefaults: false }).pipe( Layer.provide(gatedFileSystemLayer), Layer.provide(NodePath.layer), Layer.provide(gitConfigStoreLayer), diff --git a/packages/stack/src/managed-manager-recovery.integration.test.ts b/packages/stack/src/managed-manager-recovery.integration.test.ts index 32742c8fa4..fd761bf0fe 100644 --- a/packages/stack/src/managed-manager-recovery.integration.test.ts +++ b/packages/stack/src/managed-manager-recovery.integration.test.ts @@ -52,7 +52,7 @@ const acquireIsolatedCollisionOwner = () => const stackId = randomBytes(32).toString("hex"); const collidingStackId = `${stackId.slice(0, 10)}${randomBytes(27).toString("hex")}`; const acquisition = yield* acquireControl({ stackId }).pipe( - Effect.timeout("1 second"), + Effect.timeout("5 seconds"), Effect.exit, ); if (Exit.isSuccess(acquisition) && acquisition.value._tag === "Owned") { @@ -69,7 +69,7 @@ const acquireIsolatedStackOwner = (workspacePath: string) => const stackName = `test-${randomBytes(8).toString("hex")}`; const stackId = deriveStackId(environment.identity, stackName); const acquisition = yield* acquireControl({ stackId }).pipe( - Effect.timeout("1 second"), + Effect.timeout("5 seconds"), Effect.exit, ); if (Exit.isSuccess(acquisition) && acquisition.value._tag === "Owned") { @@ -128,7 +128,7 @@ describe("managed stack recovery journeys", () => { } satisfies FileSystem.FileSystem; }), ).pipe(Layer.provide(NodeFileSystem.layer)); - const managerLayer = managedStackManagerLayer({ stateRoot }).pipe( + const managerLayer = managedStackManagerLayer({ stateRoot, preferCatalogDefaults: false }).pipe( Layer.provide(gatedFileSystemLayer), Layer.provide(NodePath.layer), Layer.provide(gitConfigStoreLayer), @@ -224,11 +224,11 @@ describe("managed stack recovery journeys", () => { ownership: stackOwner.ownership, }) .pipe(Effect.forkScoped); - yield* Deferred.await(repairRead).pipe(Effect.timeout("1 second")); + yield* Deferred.await(repairRead).pipe(Effect.timeout("10 seconds")); expect(yield* manager.inspectStack(stackId)).toBeUndefined(); yield* repairOwner.close; yield* Effect.promise(() => repairDaemon.dispose()); - const started = yield* Fiber.join(startFiber).pipe(Effect.timeout("2 seconds")); + const started = yield* Fiber.join(startFiber).pipe(Effect.timeout("15 seconds")); expect(started.stack.id).toBe(stackId); yield* releaseLease(started); }), diff --git a/packages/stack/src/managed/manager.ts b/packages/stack/src/managed/manager.ts index a9a5103e70..31d1731f28 100644 --- a/packages/stack/src/managed/manager.ts +++ b/packages/stack/src/managed/manager.ts @@ -368,10 +368,17 @@ const conflictError = ( ownerKey: owner.ports.find((candidate) => candidate.port === assignment.port)?.key, }); +interface ManagedStackManagerOptions { + readonly stateRoot: string; + /** Test seam: disable well-known default ports for automatic allocation. */ + readonly preferCatalogDefaults?: boolean; +} + const makeManager = ( - stateRoot: string, + options: ManagedStackManagerOptions, ): Effect.Effect => Effect.gen(function* () { + const { stateRoot, preferCatalogDefaults = true } = options; const fileSystem = yield* FileSystem.FileSystem; const pathService = yield* Path.Path; const gitConfig = yield* GitConfigStore; @@ -458,6 +465,7 @@ const makeManager = ( disabledFields: request.portDocument.disabledFields, intents: resolvePortIntents(request.portDocument), persisted, + preferCatalogDefaults, }); const requests = portRequests(plan); const exactRequests = requests.filter((item) => item.selection.kind === "exact"); @@ -1003,11 +1011,12 @@ const makeManager = ( }); /** Internal manager layer. Platform layers provide filesystem, Git, and control transport. */ -export const managedStackManagerLayer = (options: { - readonly stateRoot: string; -}): Layer.Layer => - Layer.effect(ManagedStackManager, makeManager(options.stateRoot)); +export const managedStackManagerLayer = ( + options: ManagedStackManagerOptions, +): Layer.Layer => + Layer.effect(ManagedStackManager, makeManager(options)); export const makeManagedStackManager = ( stateRoot: string, -): Effect.Effect => makeManager(stateRoot); +): Effect.Effect => + makeManager({ stateRoot }); diff --git a/packages/stack/src/managed/port-plan.ts b/packages/stack/src/managed/port-plan.ts index 8c78cfb22c..0c249e2a97 100644 --- a/packages/stack/src/managed/port-plan.ts +++ b/packages/stack/src/managed/port-plan.ts @@ -49,6 +49,12 @@ export interface ManagedPortPlanInput { readonly disabledFields?: ReadonlyArray; readonly intents: ReadonlyArray; readonly persisted?: ReadonlyArray; + /** + * Seed automatic selections with the catalog's well-known default ports. + * Tests disable this so parallel suites do not contend on the same + * defaults, which sticky reuse later re-reserves exactly. + */ + readonly preferCatalogDefaults?: boolean; } const automaticSelection = (preferred: number | undefined): PortSelection => @@ -57,6 +63,8 @@ const automaticSelection = (preferred: number | undefined): PortSelection => /** Build sticky durable selections and runtime-only requests from resolved intent. */ export const planManagedPorts = (input: ManagedPortPlanInput): ManagedPortPlan => { const persisted = input.persisted ?? []; + const preferredFor = (entry: (typeof PORT_CATALOG)[PortField]): number | undefined => + (input.preferCatalogDefaults ?? true) ? entry.preferred : undefined; const persistedByKey = new Map(persisted.map((assignment) => [assignment.key, assignment])); const intentsByField = new Map(input.intents.map((request) => [request.field, request])); const activeKeys = new Set(); @@ -91,7 +99,7 @@ export const planManagedPorts = (input: ManagedPortPlanInput): ManagedPortPlan = field, key: entry.configKey, intent, - selection: automaticSelection(entry.preferred), + selection: automaticSelection(preferredFor(entry)), newlyAllocatedAutomatic: true, }); } @@ -99,7 +107,7 @@ export const planManagedPorts = (input: ManagedPortPlanInput): ManagedPortPlan = } if (entry.persistence === "runtime") { - runtimeOnly.push({ field, selection: automaticSelection(entry.preferred) }); + runtimeOnly.push({ field, selection: automaticSelection(preferredFor(entry)) }); } } diff --git a/packages/stack/src/supervisor.integration.test.ts b/packages/stack/src/supervisor.integration.test.ts index a87342d194..ac8e6d0d9e 100644 --- a/packages/stack/src/supervisor.integration.test.ts +++ b/packages/stack/src/supervisor.integration.test.ts @@ -89,6 +89,43 @@ const workspace = async (): Promise<{ throw new Error("Unable to allocate a free supervisor control endpoint after 32 attempts"); }; +/** + * Watches a directory, re-arming on ENOENT watcher errors: the runtime's + * directory watcher can report ENOENT when a watched entry (for example an + * atomic-write temp file) vanishes mid-scan. Callers keep their own timeout + * as the guard. Returns a close function. + */ +const watchDirectoryWithRetry = ( + directory: string, + onEvent: () => void, + onError: (cause: unknown) => void, +): (() => void) => { + let watcher: FSWatcher | undefined; + let closed = false; + const arm = () => { + if (closed) return; + try { + watcher = watch(directory, () => onEvent()); + watcher.once("error", (cause) => { + watcher?.close(); + if (cause instanceof Error && "code" in cause && cause.code === "ENOENT") { + arm(); + onEvent(); + return; + } + onError(cause); + }); + } catch (cause) { + onError(cause); + } + }; + arm(); + return () => { + closed = true; + watcher?.close(); + }; +}; + const waitForFile = (path: string): Promise => new Promise((resolve, reject) => { if (existsSync(path)) { @@ -97,20 +134,21 @@ const waitForFile = (path: string): Promise => } let settled = false; let timeout: ReturnType | undefined; - let watcher: FSWatcher | undefined; + let stopWatching: (() => void) | undefined; const settle = (continuation: () => void) => { if (settled) return; settled = true; if (timeout !== undefined) clearTimeout(timeout); - watcher?.close(); + stopWatching?.(); continuation(); }; - watcher = watch(dirname(path), () => { - if (existsSync(path)) settle(resolve); - }); - watcher.once("error", (cause) => { - settle(() => reject(cause)); - }); + stopWatching = watchDirectoryWithRetry( + dirname(path), + () => { + if (existsSync(path)) settle(resolve); + }, + (cause) => settle(() => reject(cause instanceof Error ? cause : new Error(String(cause)))), + ); timeout = setTimeout( () => settle(() => reject(new Error(`timed out waiting for file ${path}`))), FILE_WAIT_TIMEOUT_MS, @@ -509,14 +547,14 @@ const waitForStackDocument = async ( if (existing?.lifecycle === lifecycle) return existing; return new Promise((resolve, reject) => { - let watcher: FSWatcher | undefined; + let stopWatching: (() => void) | undefined; let timeout: ReturnType | undefined; let settled = false; const settle = (continuation: () => void) => { if (settled) return; settled = true; if (timeout !== undefined) clearTimeout(timeout); - watcher?.close(); + stopWatching?.(); continuation(); }; const check = () => { @@ -525,9 +563,9 @@ const waitForStackDocument = async ( settle(() => resolve(document)); } }; - const fail = (cause: unknown) => settle(() => reject(cause)); - watcher = watch(stackDirectory, () => check()); - watcher.once("error", fail); + const fail = (cause: unknown) => + settle(() => reject(cause instanceof Error ? cause : new Error(String(cause)))); + stopWatching = watchDirectoryWithRetry(stackDirectory, check, fail); timeout = setTimeout( () => fail(new Error(`timed out waiting for stack document lifecycle ${lifecycle}`)), FILE_WAIT_TIMEOUT_MS, diff --git a/packages/stack/tests/helpers/managed-manager.ts b/packages/stack/tests/helpers/managed-manager.ts index 69dca80120..d5a1d9cc39 100644 --- a/packages/stack/tests/helpers/managed-manager.ts +++ b/packages/stack/tests/helpers/managed-manager.ts @@ -29,7 +29,7 @@ export const setupManagedManager = (roots: Array) => { const workspace = join(root, "workspace"); mkdirSync(workspace); const stateRoot = join(root, "state"); - const layer = managedStackManagerLayer({ stateRoot }); + const layer = managedStackManagerLayer({ stateRoot, preferCatalogDefaults: false }); return { layer, stateRoot, workspace }; }; diff --git a/packages/stack/tests/helpers/supervisor-child.ts b/packages/stack/tests/helpers/supervisor-child.ts index 6f28597d78..3883a4d19a 100644 --- a/packages/stack/tests/helpers/supervisor-child.ts +++ b/packages/stack/tests/helpers/supervisor-child.ts @@ -156,9 +156,23 @@ const waitForAttachedBeforeReadyRelease = (): Effect.Effect => { const resolveIfReleased = () => { if (existsSync(releaseFile)) settle(Effect.void); }; - try { + // Re-arm on ENOENT watcher errors: the runtime's directory watcher can + // report ENOENT when a watched entry vanishes mid-scan. + const arm = () => { + if (settled) return; watcher = watch(dirname(releaseFile), () => resolveIfReleased()); - watcher.once("error", (cause) => settle(Effect.die(cause))); + watcher.once("error", (cause) => { + watcher?.close(); + if (cause instanceof Error && "code" in cause && cause.code === "ENOENT") { + arm(); + resolveIfReleased(); + return; + } + settle(Effect.die(cause)); + }); + }; + try { + arm(); writeFileSync(readyFile, "ready"); resolveIfReleased(); } catch (cause) { @@ -205,7 +219,7 @@ const testPlatform = (): "node" | "bun" => process.env["SUPABASE_STACK_TEST_PLATFORM"] === "bun" ? "bun" : "node"; const managerLayer = (stateRoot: string, platform: "node" | "bun") => - managedStackManagerLayer({ stateRoot }).pipe( + managedStackManagerLayer({ stateRoot, preferCatalogDefaults: false }).pipe( Layer.provide( platform === "bun" ? Layer.mergeAll(BunFileSystem.layer, gitConfigStoreLayer, bunControlTransportLayer) From 5d80e4a06774d6e4e73e5207792bbf67a90cbd38 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 10:48:24 +0000 Subject: [PATCH 5/5] fix(stack): keep control probes one-shot and bound retries by duration Stress runs under CPU oversubscription surfaced three settling hazards around the control endpoint: - Status probes reused pooled keep-alive connections, so a closed listener kept answering probes on the poller's own hot connection and an attach-scanning acquirer livelocked as Attached instead of binding the freed endpoint. Control reads and stop requests now force one-shot connections in both transports. - The acquire retry for a bound-but-not-serving listener was bounded by attempt count, but each attempt can spend the 500 ms transport timeout per read, letting one acquire call stretch far beyond the intended ~1.5 s window and starve its caller's budget. The retry is now bounded by duration. - The workspace-repair fence in startStack gave up after ~5 s, which a realistic repair (Git operations included) can exceed; it now waits up to 30 s. Test guards that raced these budgets under load (2 s conflict-report timeout, fence join guards) are widened to stay guards rather than timing assertions. The integration suite now passes 8/8 under a 2x CPU-oversubscribed stress gauntlet. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01McsNM9yxC5tiY6SusBALoq --- packages/stack/src/managed-control.integration.test.ts | 2 +- .../stack/src/managed-manager-recovery.integration.test.ts | 4 ++-- packages/stack/src/managed/control.ts | 7 ++++--- packages/stack/src/managed/manager.ts | 7 ++++++- packages/stack/src/platform-bun.ts | 5 +++++ packages/stack/src/platform-node.ts | 5 +++++ 6 files changed, 23 insertions(+), 7 deletions(-) diff --git a/packages/stack/src/managed-control.integration.test.ts b/packages/stack/src/managed-control.integration.test.ts index c3649c0058..7fd41e05b7 100644 --- a/packages/stack/src/managed-control.integration.test.ts +++ b/packages/stack/src/managed-control.integration.test.ts @@ -384,7 +384,7 @@ describe("managed control endpoint", () => { requestStop: () => Effect.void, }); const exit = yield* acquireControl({ stackId: STACK_ID }).pipe( - Effect.timeout("2 seconds"), + Effect.timeout("10 seconds"), Effect.exit, Effect.provide(unavailable), ); diff --git a/packages/stack/src/managed-manager-recovery.integration.test.ts b/packages/stack/src/managed-manager-recovery.integration.test.ts index fd761bf0fe..9ccfffc7e6 100644 --- a/packages/stack/src/managed-manager-recovery.integration.test.ts +++ b/packages/stack/src/managed-manager-recovery.integration.test.ts @@ -224,11 +224,11 @@ describe("managed stack recovery journeys", () => { ownership: stackOwner.ownership, }) .pipe(Effect.forkScoped); - yield* Deferred.await(repairRead).pipe(Effect.timeout("10 seconds")); + yield* Deferred.await(repairRead).pipe(Effect.timeout("30 seconds")); expect(yield* manager.inspectStack(stackId)).toBeUndefined(); yield* repairOwner.close; yield* Effect.promise(() => repairDaemon.dispose()); - const started = yield* Fiber.join(startFiber).pipe(Effect.timeout("15 seconds")); + const started = yield* Fiber.join(startFiber).pipe(Effect.timeout("60 seconds")); expect(started.stack.id).toBe(stackId); yield* releaseLease(started); }), diff --git a/packages/stack/src/managed/control.ts b/packages/stack/src/managed/control.ts index af784bb1c5..65efbdab0b 100644 --- a/packages/stack/src/managed/control.ts +++ b/packages/stack/src/managed/control.ts @@ -456,9 +456,10 @@ const acquireAtCandidates = ( return attempt.pipe( Effect.retry({ - // Leave explicit margin inside the parent's 35-second startup handshake, - // even when every owner probe consumes its 500 ms transport timeout. - schedule: Schedule.spaced("50 millis").pipe(Schedule.upTo({ times: 30 })), + // Bound by duration, not attempts: an attempt's own reads can each + // consume the 500 ms transport timeout, and a count-based budget would + // stretch a single acquire far past the parent's startup handshake. + schedule: Schedule.spaced("50 millis").pipe(Schedule.upTo({ duration: "1500 millis" })), while: (error) => error._tag === "ControlUnavailableError", }), Effect.catchTag("ControlUnavailableError", (error) => diff --git a/packages/stack/src/managed/manager.ts b/packages/stack/src/managed/manager.ts index 31d1731f28..7746b11c8a 100644 --- a/packages/stack/src/managed/manager.ts +++ b/packages/stack/src/managed/manager.ts @@ -726,7 +726,12 @@ const makeManager = ( ), ), Effect.retry({ - schedule: Schedule.spaced("20 millis").pipe(Schedule.upTo({ times: 250 })), + // A held repair fence means another process is actively + // repairing this workspace; wait out a realistic repair + // (Git operations included) instead of failing after ~5s. + schedule: Schedule.spaced("20 millis").pipe( + Schedule.upTo({ duration: "30 seconds" }), + ), while: (error) => error instanceof ManagedWorkspaceRepairConflictError, }), ), diff --git a/packages/stack/src/platform-bun.ts b/packages/stack/src/platform-bun.ts index dbe2176aef..ebbeee33fc 100644 --- a/packages/stack/src/platform-bun.ts +++ b/packages/stack/src/platform-bun.ts @@ -82,6 +82,10 @@ const controlTransport: ControlTransport["Service"] = { try: async () => { const response = await fetch(`http://127.0.0.1:${endpoint.port}${CONTROL_STATUS_PATH}`, { signal: AbortSignal.timeout(500), + // One-shot connection: a pooled keep-alive connection would let a + // closed listener keep answering status probes while the probes + // themselves keep the connection alive. + headers: { connection: "close" }, }); if (!response.ok) throw new Error(`Control status request returned ${response.status}`); return await response.json(); @@ -107,6 +111,7 @@ const controlTransport: ControlTransport["Service"] = { const response = await fetch(`http://127.0.0.1:${endpoint.port}${CONTROL_STOP_PATH}`, { method: "POST", signal: AbortSignal.timeout(500), + headers: { connection: "close" }, }); if (!response.ok) throw new Error(`Control stop request returned ${response.status}`); }, diff --git a/packages/stack/src/platform-node.ts b/packages/stack/src/platform-node.ts index 4f43b33fad..583db3c324 100644 --- a/packages/stack/src/platform-node.ts +++ b/packages/stack/src/platform-node.ts @@ -78,6 +78,10 @@ const controlTransport: ControlTransport["Service"] = { port: endpoint.port, path: CONTROL_STATUS_PATH, method: "GET", + // One-shot connection: a pooled keep-alive connection would + // let a closed listener keep answering status probes while + // the probes themselves keep the connection alive. + agent: false, }, (response) => { let body = ""; @@ -142,6 +146,7 @@ const controlTransport: ControlTransport["Service"] = { port: endpoint.port, path: CONTROL_STOP_PATH, method: "POST", + agent: false, }, (response) => { response.resume();