vrr/opencode
2
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-09-01 15:24:34 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(core): settle ownership when suspension hooks throw

Browse source 
Evaluate the suspension callback inside its cleanup guarantee so synchronous throws cannot leave ghost ownership. Cover successful hooks, returned defects, and synchronous throws.
This commit is contained in:
Kit Langton 2026-08-31 12:46:58 -04:00
parent d1ca3089ce
commit 9f867ae11f
2 changed files with 37 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

2
packages/core/src/session/run-coordinator.ts
View file

@ -110,7 +110,7 @@ export const make = <Key, E, Reason = never>(options: {
Effect.onExit((exit) =>
Effect.suspend(() => {
if (closing && execution.pendingWake)
return (options.suspended?.(key) ?? Effect.void).pipe(
return Effect.suspend(() => options.suspended?.(key) ?? Effect.void).pipe(
Effect.ensuring(Effect.sync(() => settle(key, execution, exit))),
)
settle(key, execution, exit)

66
packages/core/test/session-run-coordinator.test.ts
View file

@ -211,38 +211,44 @@ describe("SessionRunCoordinator", () => {
}
}
it.effect("suspends a settlement-window wake after its terminal hook completes", () =>
Effect.gen(function* () {
const settling = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const scope = yield* Scope.make()
yield* Effect.addFinalizer(() =>
Deferred.succeed(release, undefined).pipe(Effect.andThen(Scope.close(scope, Exit.void))),
)
const lifecycle: string[] = []
const coordinator = yield* SessionRunCoordinator.make({
drain: () => Effect.sync(() => void lifecycle.push("drained")),
settled: () =>
Deferred.succeed(settling, undefined).pipe(
Effect.andThen(Deferred.await(release)),
Effect.andThen(Effect.sync(() => void lifecycle.push("settled"))),
),
suspended: () => Effect.sync(() => void lifecycle.push("suspended")),
}).pipe(Scope.provide(scope))
for (const outcome of ["success", "effect defect", "synchronous throw"]) {
it.effect(`settles a shutdown-deferred successor after suspension hook ${outcome}`, () =>
Effect.gen(function* () {
const settling = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const scope = yield* Scope.make()
yield* Effect.addFinalizer(() =>
Deferred.succeed(release, undefined).pipe(Effect.andThen(Scope.close(scope, Exit.void))),
)
const lifecycle: string[] = []
const coordinator = yield* SessionRunCoordinator.make({
drain: () => Effect.sync(() => void lifecycle.push("drained")),
settled: () =>
Deferred.succeed(settling, undefined).pipe(
Effect.andThen(Deferred.await(release)),
Effect.andThen(Effect.sync(() => void lifecycle.push("settled"))),
),
suspended: () => {
lifecycle.push("suspended")
if (outcome === "synchronous throw") throw new Error("suspension failed")
return outcome === "effect defect" ? Effect.die(new Error("suspension failed")) : Effect.void
},
}).pipe(Scope.provide(scope))
yield* coordinator.wake("session")
yield* Deferred.await(settling)
yield* coordinator.wake("session")
const closing = yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild({ startImmediately: true }))
yield* Effect.yieldNow
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(closing)
yield* coordinator.wake("session")
yield* Deferred.await(settling)
yield* coordinator.wake("session")
const closing = yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild({ startImmediately: true }))
yield* Effect.yieldNow
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(closing)
expect(lifecycle).toEqual(["drained", "settled", "suspended"])
expect(yield* coordinator.active).toEqual(new Set())
yield* coordinator.awaitIdle("session")
}),
)
expect(lifecycle).toEqual(["drained", "settled", "suspended"])
expect(yield* coordinator.active).toEqual(new Set())
yield* coordinator.awaitIdle("session")
}),
)
}
it.effect("coalesces wakes received during active execution", () =>
Effect.gen(function* () {