From e6bb88bc1d8d680900d16645524e74b811181d5b Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 6 Jun 2026 23:13:52 -0400 Subject: [PATCH] refactor(core): specialize session run coordination --- .../src/session/run-coordinator-machine.ts | 358 ------------- packages/core/src/session/run-coordinator.ts | 491 +++++++++++------- .../session-run-coordinator-machine.test.ts | 213 -------- .../core/test/session-run-coordinator.test.ts | 75 ++- 4 files changed, 390 insertions(+), 747 deletions(-) delete mode 100644 packages/core/src/session/run-coordinator-machine.ts delete mode 100644 packages/core/test/session-run-coordinator-machine.test.ts diff --git a/packages/core/src/session/run-coordinator-machine.ts b/packages/core/src/session/run-coordinator-machine.ts deleted file mode 100644 index 5afa1cdbd0e..00000000000 --- a/packages/core/src/session/run-coordinator-machine.ts +++ /dev/null @@ -1,358 +0,0 @@ -/** @internal Pure state machine for the process-local Session run coordinator. */ -export * as SessionRunCoordinatorMachine from "./run-coordinator-machine" - -/** @internal */ -export type Mode = "run" | "wake" - -/** @internal */ -export type Demand = { - readonly explicit: boolean - readonly wakeSeq?: number - readonly unsequencedWake: boolean -} - -type NonEmptyDemand = Demand & - ({ readonly explicit: true } | { readonly wakeSeq: number } | { readonly unsequencedWake: true }) - -/** @internal */ -export const Demand = { - empty: { explicit: false, unsequencedWake: false } satisfies Demand, - explicit: { explicit: true, unsequencedWake: false } satisfies NonEmptyDemand, - wake: (seq?: number): NonEmptyDemand => - seq === undefined - ? { explicit: false, unsequencedWake: true } - : { explicit: false, wakeSeq: seq, unsequencedWake: false }, - combine: (left: Demand, right: Demand): Demand => ({ - explicit: left.explicit || right.explicit, - wakeSeq: - left.wakeSeq === undefined - ? right.wakeSeq - : right.wakeSeq === undefined - ? left.wakeSeq - : Math.max(left.wakeSeq, right.wakeSeq), - unsequencedWake: left.unsequencedWake || right.unsequencedWake, - }), - afterBoundary: (demand: Demand, boundary?: number): Demand => ({ - explicit: false, - wakeSeq: - boundary !== undefined && demand.wakeSeq !== undefined && demand.wakeSeq > boundary ? demand.wakeSeq : undefined, - unsequencedWake: false, - }), - nonEmpty: (demand: Demand): demand is NonEmptyDemand => - demand.explicit || demand.wakeSeq !== undefined || demand.unsequencedWake, - mode: (demand: Demand): Mode => (demand.explicit ? "run" : "wake"), -} - -type Running = { - readonly _tag: "Running" - readonly chain: number - readonly attempt: number - readonly current: NonEmptyDemand - readonly pending: Demand - readonly waiter?: number -} - -type Stopping = { - readonly _tag: "Stopping" - readonly chain: number - readonly attempt: number - readonly current: NonEmptyDemand - readonly pending: Demand - readonly waiter?: number - readonly stopBoundary?: number -} - -/** @internal */ -export type Lane = Running | Stopping - -/** @internal */ -export type State = { - readonly closed: boolean - readonly nextID: number - readonly lanes: ReadonlyMap - readonly interruptSeq: ReadonlyMap -} - -/** @internal */ -export const initial = (): State => ({ closed: false, nextID: 1, lanes: new Map(), interruptSeq: new Map() }) - -/** @internal */ -export type Outcome = "Success" | "Failure" | "Interrupted" - -/** @internal */ -export type Event = - | { readonly _tag: "Close" } - | { readonly _tag: "Run"; readonly key: Key } - | { readonly _tag: "Wake"; readonly key: Key; readonly seq?: number } - | { readonly _tag: "Interrupt"; readonly key: Key; readonly seq?: number } - | { readonly _tag: "Observe"; readonly key: Key } - | { - readonly _tag: "Settled" - readonly key: Key - readonly chain: number - readonly attempt: number - readonly outcome: Outcome - } - -/** @internal */ -export type Action = - | { - readonly _tag: "Start" - readonly key: Key - readonly chain: number - readonly attempt: number - readonly demand: NonEmptyDemand - readonly successor: boolean - } - | { readonly _tag: "Interrupt"; readonly attempt: number } - | { readonly _tag: "CompleteChain"; readonly chain: number } - | { readonly _tag: "CompleteWaiter"; readonly waiter: number } - | { readonly _tag: "Report"; readonly key: Key } - -/** @internal */ -export type Response = - | { readonly _tag: "None" } - | { readonly _tag: "AwaitChain"; readonly chain: number } - | { readonly _tag: "AwaitWaiter"; readonly waiter: number } - | { readonly _tag: "RetryAfter"; readonly chain: number } - | { readonly _tag: "ObserveChain"; readonly chain: number } - | { readonly _tag: "Idle" } - | { readonly _tag: "Closed" } - -/** @internal */ -export type Transition = { - readonly state: State - readonly actions: ReadonlyArray> - readonly response: Response -} - -const none: Response = { _tag: "None" } - -/** @internal */ -export const reduce = (state: State, event: Event): Transition => { - if (event._tag === "Close") - return { state: { ...state, closed: true, lanes: new Map(), interruptSeq: new Map() }, actions: [], response: none } - if (state.closed) return { state, actions: [], response: event._tag === "Run" ? { _tag: "Closed" } : none } - if (event._tag === "Run") return run(state, event) - if (event._tag === "Wake") return wake(state, event) - if (event._tag === "Interrupt") return interrupt(state, event) - if (event._tag === "Observe") { - const lane = state.lanes.get(event.key) - return { - state, - actions: [], - response: lane === undefined ? { _tag: "Idle" } : { _tag: "ObserveChain", chain: lane.chain }, - } - } - return settled(state, event) -} - -const run = (state: State, event: Extract, { _tag: "Run" }>): Transition => { - const lane = state.lanes.get(event.key) - if (lane?._tag === "Stopping") return { state, actions: [], response: { _tag: "RetryAfter", chain: lane.chain } } - if (lane !== undefined && lane.current.explicit) - return { state, actions: [], response: { _tag: "AwaitChain", chain: lane.chain } } - if (lane !== undefined) { - const [allocated, waiter] = lane.waiter === undefined ? allocate(state) : [state, lane.waiter] - return { - state: setLane(allocated, event.key, { ...lane, pending: Demand.combine(lane.pending, Demand.explicit), waiter }), - actions: [], - response: { _tag: "AwaitWaiter", waiter }, - } - } - const [withChain, chain] = allocate(state) - const [allocated, attempt] = allocate(withChain) - const next: Lane = { - _tag: "Running", - chain, - attempt, - current: Demand.explicit, - pending: Demand.empty, - } - return { - state: setLane(allocated, event.key, next), - actions: [ - { - _tag: "Start", - key: event.key, - chain: next.chain, - attempt: next.attempt, - demand: next.current, - successor: false, - }, - ], - response: { _tag: "AwaitChain", chain: next.chain }, - } -} - -const wake = (state: State, event: Extract, { _tag: "Wake" }>): Transition => { - const boundary = state.interruptSeq.get(event.key) - if (boundary !== undefined && (event.seq === undefined || event.seq <= boundary)) - return { state, actions: [], response: none } - const lane = state.lanes.get(event.key) - if (lane !== undefined) { - if ( - lane._tag === "Stopping" && - (lane.stopBoundary === undefined || event.seq === undefined || event.seq <= lane.stopBoundary) - ) - return { state, actions: [], response: none } - return { - state: setLane(state, event.key, { ...lane, pending: Demand.combine(lane.pending, Demand.wake(event.seq)) }), - actions: [], - response: none, - } - } - const [withChain, chain] = allocate(state) - const [allocated, attempt] = allocate(withChain) - const next: Lane = { - _tag: "Running", - chain, - attempt, - current: Demand.wake(event.seq), - pending: Demand.empty, - } - return { - state: setLane(allocated, event.key, next), - actions: [ - { - _tag: "Start", - key: event.key, - chain: next.chain, - attempt: next.attempt, - demand: next.current, - successor: false, - }, - ], - response: none, - } -} - -const interrupt = (state: State, event: Extract, { _tag: "Interrupt" }>): Transition => { - const latest = state.interruptSeq.get(event.key) - const lane = state.lanes.get(event.key) - if (event.seq !== undefined && latest !== undefined && event.seq <= latest) - return { - state, - actions: lane?._tag === "Stopping" ? [{ _tag: "Interrupt", attempt: lane.attempt }] : [], - response: none, - } - const bounded = event.seq === undefined ? state : setInterruptSeq(state, event.key, event.seq) - if (lane === undefined) return { state: bounded, actions: [], response: none } - if ( - !lane.current.explicit && - event.seq !== undefined && - lane.current.wakeSeq !== undefined && - lane.current.wakeSeq > event.seq - ) - return { state: bounded, actions: [], response: none } - const pending = Demand.combine( - Demand.afterBoundary(lane.current, event.seq), - Demand.afterBoundary(lane.pending, event.seq), - ) - return { - state: setLane(bounded, event.key, { - _tag: "Stopping", - chain: lane.chain, - attempt: lane.attempt, - current: lane.current, - pending, - waiter: lane.waiter, - stopBoundary: lane._tag === "Stopping" ? maxSeq(lane.stopBoundary, event.seq) : event.seq, - }), - actions: [{ _tag: "Interrupt", attempt: lane.attempt }], - response: none, - } -} - -const settled = (state: State, event: Extract, { _tag: "Settled" }>): Transition => { - const lane = state.lanes.get(event.key) - if (lane?.chain !== event.chain || lane.attempt !== event.attempt) return { state, actions: [], response: none } - const completesWaiter = lane.current.explicit || (lane._tag === "Stopping" && !lane.current.explicit) - const waiterActions: ReadonlyArray> = - completesWaiter && lane.waiter !== undefined ? [{ _tag: "CompleteWaiter", waiter: lane.waiter }] : [] - const waiter = completesWaiter ? undefined : lane.waiter - if (event.outcome === "Success" && lane._tag === "Running" && Demand.nonEmpty(lane.pending)) { - const [allocated, attempt] = allocate(state) - const next = { ...lane, attempt, current: lane.pending, pending: Demand.empty, waiter } - return { - state: setLane(allocated, event.key, next), - actions: [ - ...waiterActions, - { - _tag: "Start", - key: event.key, - chain: next.chain, - attempt: next.attempt, - demand: next.current, - successor: true, - }, - ], - response: none, - } - } - const report: ReadonlyArray> = - event.outcome !== "Success" && - !(lane._tag === "Stopping" && event.outcome === "Interrupted") && - !lane.current.explicit - ? [{ _tag: "Report", key: event.key }] - : [] - if (!Demand.nonEmpty(lane.pending)) - return { - state: deleteLane(state, event.key), - actions: [...waiterActions, { _tag: "CompleteChain", chain: lane.chain }, ...report], - response: none, - } - const [withChain, chain] = allocate(state) - const [allocated, attempt] = allocate(withChain) - const next: Lane = { - _tag: "Running", - chain, - attempt, - current: lane.pending, - pending: Demand.empty, - waiter, - } - return { - state: setLane(allocated, event.key, next), - actions: [ - ...waiterActions, - { - _tag: "Start", - key: event.key, - chain: next.chain, - attempt: next.attempt, - demand: next.current, - successor: true, - }, - { _tag: "CompleteChain", chain: lane.chain }, - ...report, - ], - response: none, - } -} - -const maxSeq = (left?: number, right?: number) => - left === undefined ? right : right === undefined ? left : Math.max(left, right) - -const allocate = (state: State): readonly [State, number] => [ - { ...state, nextID: state.nextID + 1 }, - state.nextID, -] - -const setLane = (state: State, key: Key, lane: Lane): State => { - const lanes = new Map(state.lanes) - lanes.set(key, lane) - return { ...state, lanes } -} - -const deleteLane = (state: State, key: Key): State => { - const lanes = new Map(state.lanes) - lanes.delete(key) - return { ...state, lanes } -} - -const setInterruptSeq = (state: State, key: Key, seq: number): State => { - const interruptSeq = new Map(state.interruptSeq) - interruptSeq.set(key, seq) - return { ...state, interruptSeq } -} diff --git a/packages/core/src/session/run-coordinator.ts b/packages/core/src/session/run-coordinator.ts index 021c8e0c59a..5a93cb86127 100644 --- a/packages/core/src/session/run-coordinator.ts +++ b/packages/core/src/session/run-coordinator.ts @@ -1,11 +1,23 @@ export * as SessionRunCoordinator from "./run-coordinator" -import { Cause, Context, Deferred, Effect, Exit, Fiber, FiberSet, Layer, Scope, SynchronizedRef } from "effect" +import { + Cause, + Context, + Data, + Deferred, + Effect, + Equal, + Exit, + Fiber, + FiberSet, + Layer, + Scope, + SynchronizedRef, +} from "effect" import { SessionRunner } from "./runner" import { SessionSchema } from "./schema" -import { SessionRunCoordinatorMachine } from "./run-coordinator-machine" -export type Mode = SessionRunCoordinatorMachine.Mode +export type Mode = "run" | "wake" export interface Coordinator { readonly run: (key: Key) => Effect.Effect @@ -14,9 +26,102 @@ export interface Coordinator { readonly interrupt: (key: Key, seq?: number) => Effect.Effect } -type Chain = { - readonly done: Deferred.Deferred - readonly settled: Deferred.Deferred> +/** @internal */ +export class Demand extends Data.Class<{ + readonly explicit: boolean + readonly wakeSeq?: number + readonly unsequencedWake: boolean +}> { + static readonly empty = new Demand({ explicit: false, wakeSeq: undefined, unsequencedWake: false }) + static readonly run = nonEmpty(new Demand({ explicit: true, wakeSeq: undefined, unsequencedWake: false })) + + static wake(seq?: number) { + return nonEmpty(new Demand({ explicit: false, wakeSeq: seq, unsequencedWake: seq === undefined })) + } + + combine(other: Demand) { + return new Demand({ + explicit: this.explicit || other.explicit, + wakeSeq: + this.wakeSeq === undefined + ? other.wakeSeq + : other.wakeSeq === undefined + ? this.wakeSeq + : Math.max(this.wakeSeq, other.wakeSeq), + unsequencedWake: this.unsequencedWake || other.unsequencedWake, + }) + } + + afterBoundary(boundary?: number) { + return new Demand({ + explicit: false, + wakeSeq: + boundary !== undefined && this.wakeSeq !== undefined && this.wakeSeq > boundary ? this.wakeSeq : undefined, + unsequencedWake: false, + }) + } + + isNonEmpty(): this is NonEmptyDemand { + return this.explicit || this.wakeSeq !== undefined || this.unsequencedWake + } + + get mode(): Mode { + return this.explicit ? "run" : "wake" + } +} + +type NonEmptyDemand = Demand & + ({ readonly explicit: true } | { readonly wakeSeq: number } | { readonly unsequencedWake: true }) + +function nonEmpty(demand: Demand): NonEmptyDemand { + if (!demand.isNonEmpty()) throw new Error("Session run demand must not be empty") + return demand +} + +type Lifecycle = + | { readonly _tag: "Running"; readonly token: object; readonly owner: Deferred.Deferred> } + | { + readonly _tag: "Stopping" + readonly token: object + readonly owner: Deferred.Deferred> + readonly boundary?: number + } + +type Lane = { + readonly current: NonEmptyDemand + readonly pending: Demand + readonly lifecycle: Lifecycle + readonly terminal: Deferred.Deferred> + readonly waiter?: Deferred.Deferred> +} + +type State = { + readonly closed: boolean + readonly lanes: ReadonlyMap> + readonly interruptSeq: ReadonlyMap +} + +type Start = { + readonly key: Key + readonly demand: NonEmptyDemand + readonly successor: boolean + readonly token: object + readonly owner: Deferred.Deferred> + readonly ready: Deferred.Deferred + readonly terminal: Deferred.Deferred> +} + +type RunRequest = + | { readonly _tag: "Closed" } + | { readonly _tag: "Await"; readonly terminal: Deferred.Deferred> } + | { readonly _tag: "Retry"; readonly terminal: Deferred.Deferred> } + | { readonly _tag: "Start"; readonly start: Start; readonly terminal: Deferred.Deferred> } + +type Completion = { + readonly start?: Start + readonly terminal?: Deferred.Deferred> + readonly waiter?: Deferred.Deferred> + readonly report?: Cause.Cause } export const make = (options: { @@ -24,218 +129,256 @@ export const make = (options: { readonly onFailure?: (key: Key, cause: Cause.Cause) => Effect.Effect }): Effect.Effect, never, Scope.Scope> => Effect.gen(function* () { - const state = yield* SynchronizedRef.make(SessionRunCoordinatorMachine.initial()) - const report = yield* FiberSet.makeRuntime() + const state = yield* SynchronizedRef.make>({ + closed: false, + lanes: new Map(), + interruptSeq: new Map(), + }) const fork = yield* FiberSet.makeRuntime() - const chains = new Map>() - const waiters = new Map>() - const owners = new Map>>() const shutdown = Deferred.makeUnsafe() - const chain = (chainID: number) => { - const existing = chains.get(chainID) - if (existing !== undefined) return existing - const created = { done: Deferred.makeUnsafe(), settled: Deferred.makeUnsafe>() } - chains.set(chainID, created) - return created - } - const waiter = (waiterID: number) => { - const existing = waiters.get(waiterID) - if (existing !== undefined) return existing - const created = Deferred.makeUnsafe() - waiters.set(waiterID, created) - return created - } - const owner = (attemptID: number) => { - const existing = owners.get(attemptID) - if (existing !== undefined) return existing - const created = Deferred.makeUnsafe>() - owners.set(attemptID, created) - return created + const updateLane = (current: State, key: Key, lane?: Lane): State => { + const lanes = new Map(current.lanes) + if (lane === undefined) lanes.delete(key) + else lanes.set(key, lane) + return { ...current, lanes } } - const requireChain = (chainID: number) => { - const existing = chains.get(chainID) - if (existing !== undefined) return existing - throw new Error(`Missing Session run chain ${chainID}`) - } - const requireWaiter = (waiterID: number) => { - const existing = waiters.get(waiterID) - if (existing !== undefined) return existing - throw new Error(`Missing Session run waiter ${waiterID}`) + const start = (input: { + readonly state: State + readonly key: Key + readonly demand: NonEmptyDemand + readonly terminal?: Deferred.Deferred> + readonly waiter?: Deferred.Deferred> + readonly successor?: boolean + }) => { + const instruction: Start = { + key: input.key, + demand: input.demand, + successor: input.successor ?? false, + token: {}, + owner: Deferred.makeUnsafe>(), + ready: Deferred.makeUnsafe(), + terminal: input.terminal ?? Deferred.makeUnsafe>(), + } + return { + state: updateLane(input.state, input.key, { + current: input.demand, + pending: Demand.empty, + lifecycle: { _tag: "Running", token: instruction.token, owner: instruction.owner }, + terminal: instruction.terminal, + waiter: input.waiter, + }), + start: instruction, + result: instruction.terminal, + } } - type RuntimeResponse = - | { readonly _tag: "None" | "Idle" | "Closed" } - | { readonly _tag: "Await"; readonly deferred: Deferred.Deferred } - | { readonly _tag: "Retry" | "Observe"; readonly deferred: Deferred.Deferred> } - - const transition = (event: SessionRunCoordinatorMachine.Event) => - SynchronizedRef.modifyEffect(state, (current) => { - const result = SessionRunCoordinatorMachine.reduce(current, event) - return Effect.sync(() => { - result.actions.forEach((action) => { - if (action._tag !== "Start") return - chain(action.chain) - owner(action.attempt) - }) - if (result.response._tag === "AwaitChain") chain(result.response.chain) - if (result.response._tag === "AwaitWaiter") waiter(result.response.waiter) - const response: RuntimeResponse = - result.response._tag === "AwaitChain" - ? { _tag: "Await", deferred: requireChain(result.response.chain).done } - : result.response._tag === "AwaitWaiter" - ? { _tag: "Await", deferred: requireWaiter(result.response.waiter) } - : result.response._tag === "RetryAfter" - ? { _tag: "Retry", deferred: requireChain(result.response.chain).settled } - : result.response._tag === "ObserveChain" - ? { _tag: "Observe", deferred: requireChain(result.response.chain).settled } - : { _tag: result.response._tag } - return [{ actions: result.actions, response }, result.state] as const - }) + const launch = (instruction: Start) => + Effect.gen(function* () { + const fiber = fork( + Deferred.await(instruction.ready).pipe( + Effect.andThen(instruction.successor ? Effect.yieldNow : Effect.void), + Effect.andThen(Effect.suspend(() => options.drain(instruction.key, instruction.demand.mode))), + Effect.onExit((exit) => complete(instruction.key, instruction.token, exit)), + Effect.exit, + Effect.asVoid, + ), + ) + yield* Deferred.succeed(instruction.owner, fiber) + yield* Deferred.succeed(instruction.ready, undefined) }) - type Execution = { readonly _tag: "General" } | { readonly _tag: "Settlement"; readonly exit: Exit.Exit } + const complete = (key: Key, token: object, exit: Exit.Exit): Effect.Effect => { + return SynchronizedRef.modify(state, (current): readonly [Completion, State] => { + const lane = current.lanes.get(key) + if (lane === undefined || lane.lifecycle.token !== token) return [{}, current] - const execute = ( - actions: ReadonlyArray>, - execution: Execution, - ): Effect.Effect => - Effect.forEach( - actions, - (action): Effect.Effect => { - if (action._tag === "Start") { - requireChain(action.chain) - const ownerDeferred = owners.get(action.attempt) - if (ownerDeferred === undefined) return Effect.die(`Missing Session run attempt ${action.attempt}`) - const ready = Deferred.makeUnsafe() - const drain = Effect.suspend(() => - options.drain(action.key, SessionRunCoordinatorMachine.Demand.mode(action.demand)), - ) - const fiber = fork( - (action.successor - ? Effect.yieldNow.pipe(Effect.andThen(drain)) - : Deferred.await(ready).pipe(Effect.andThen(drain)) - ).pipe( - Effect.onExit((result) => settle(action.key, action.chain, action.attempt, result)), - Effect.exit, - Effect.asVoid, - ), - ) - Deferred.doneUnsafe(ownerDeferred, Effect.succeed(fiber)) - if (!action.successor) Deferred.doneUnsafe(ready, Effect.void) - return Effect.void - } - if (action._tag === "Interrupt") { - const ownerDeferred = owners.get(action.attempt) - return ownerDeferred === undefined - ? Effect.void - : Deferred.await(ownerDeferred).pipe(Effect.flatMap(Fiber.interrupt)) - } - if (execution._tag !== "Settlement") return Effect.die("Settlement action requires a settlement context") - if (action._tag === "CompleteWaiter") { - const deferred = requireWaiter(action.waiter) - waiters.delete(action.waiter) - Deferred.doneUnsafe(deferred, execution.exit) - return Effect.void - } - if (action._tag === "CompleteChain") { - const deferreds = requireChain(action.chain) - chains.delete(action.chain) - Deferred.doneUnsafe(deferreds.done, execution.exit) - Deferred.doneUnsafe(deferreds.settled, Effect.succeed(execution.exit)) - return Effect.void - } - if (action._tag === "Report") { - const onFailure = options.onFailure - if (execution.exit._tag === "Success") return Effect.die("Failure report requires a failed settlement") - if (onFailure === undefined) return Effect.void - const cause = execution.exit.cause - report(Effect.suspend(() => onFailure(action.key, cause))) - } - return Effect.void - }, - { discard: true }, - ).pipe(Effect.asVoid) + const deliberateInterrupt = + lane.lifecycle._tag === "Stopping" && exit._tag === "Failure" && Cause.hasInterruptsOnly(exit.cause) + const report = + exit._tag === "Failure" && !deliberateInterrupt && !lane.current.explicit ? exit.cause : undefined + const completesWaiter = lane.current.explicit || (lane.lifecycle._tag === "Stopping" && !lane.current.explicit) + const waiter = completesWaiter ? undefined : lane.waiter - const settle = (key: Key, chainID: number, attemptID: number, exit: Exit.Exit) => { - return transition({ - _tag: "Settled", - key, - chain: chainID, - attempt: attemptID, - outcome: exit._tag === "Success" ? "Success" : Cause.hasInterruptsOnly(exit.cause) ? "Interrupted" : "Failure", - }).pipe( - Effect.flatMap((result) => execute(result.actions, { _tag: "Settlement", exit })), - Effect.ensuring(Effect.sync(() => owners.delete(attemptID))), - ) + if (exit._tag === "Success" && lane.lifecycle._tag === "Running" && lane.pending.isNonEmpty()) { + const next = start({ + state: current, + key, + demand: lane.pending, + terminal: lane.terminal, + waiter, + successor: true, + }) + return [{ start: next.start, waiter: completesWaiter ? lane.waiter : undefined, report }, next.state] + } + + const next = lane.pending.isNonEmpty() + ? start({ state: current, key, demand: lane.pending, waiter, successor: true }) + : { state: updateLane(current, key) } + return [ + { + start: "start" in next ? next.start : undefined, + terminal: lane.terminal, + waiter: completesWaiter ? lane.waiter : undefined, + report, + }, + next.state, + ] + }).pipe(Effect.flatMap((instruction) => executeCompletion(key, exit, instruction))) } - const dispatch = (event: SessionRunCoordinatorMachine.Event) => - Effect.uninterruptible( - transition(event).pipe(Effect.flatMap((result) => execute(result.actions, { _tag: "General" }))), + const executeCompletion = (key: Key, exit: Exit.Exit, instruction: Completion) => + Effect.gen(function* () { + if (instruction.start !== undefined) yield* launch(instruction.start) + if (instruction.waiter !== undefined) yield* Deferred.succeed(instruction.waiter, exit) + if (instruction.terminal !== undefined) yield* Deferred.succeed(instruction.terminal, exit) + if (instruction.report !== undefined && options.onFailure !== undefined) { + const onFailure = options.onFailure + const cause = instruction.report + fork(Effect.suspend(() => onFailure(key, cause)).pipe(Effect.exit, Effect.asVoid)) + } + }) + + const awaitTerminal = (terminal: Deferred.Deferred>) => + Effect.raceFirst( + Deferred.await(terminal).pipe( + Effect.flatMap( + Exit.match({ + onSuccess: Effect.succeed, + onFailure: Effect.failCause, + }), + ), + ), + Deferred.await(shutdown).pipe(Effect.andThen(Effect.interrupt)), ) const run = (key: Key): Effect.Effect => Effect.suspend(() => Effect.uninterruptibleMask((restore) => { - return transition({ _tag: "Run", key }).pipe( - Effect.flatMap((result) => { - return execute(result.actions, { _tag: "General" }).pipe( - Effect.andThen( - result.response._tag === "Await" - ? awaitResult(result.response.deferred) - : result.response._tag === "Retry" - ? Effect.raceFirst( - Deferred.await(result.response.deferred).pipe(Effect.as(true)), - Deferred.await(shutdown).pipe(Effect.as(false)), - ).pipe(Effect.flatMap((settled) => (settled ? run(key) : Effect.interrupt))) - : Effect.interrupt, - ), - restore, - ) + return SynchronizedRef.modify(state, (current): readonly [RunRequest, State] => { + if (current.closed) return [{ _tag: "Closed" }, current] + const lane = current.lanes.get(key) + if (lane?.lifecycle._tag === "Stopping") return [{ _tag: "Retry", terminal: lane.terminal }, current] + if (lane?.current.explicit) return [{ _tag: "Await", terminal: lane.terminal }, current] + if (lane !== undefined) { + const terminal = lane.waiter ?? Deferred.makeUnsafe>() + const pending = lane.pending.combine(Demand.run) + if (Equal.equals(pending, lane.pending) && lane.waiter !== undefined) + return [{ _tag: "Await", terminal }, current] + return [{ _tag: "Await", terminal }, updateLane(current, key, { ...lane, pending, waiter: terminal })] + } + const next = start({ state: current, key, demand: Demand.run }) + return [{ _tag: "Start", start: next.start, terminal: next.result }, next.state] + }).pipe( + Effect.flatMap((request) => { + if (request._tag === "Closed") return Effect.interrupt + if (request._tag === "Start") + return launch(request.start).pipe(Effect.andThen(awaitTerminal(request.terminal))) + if (request._tag === "Await") return awaitTerminal(request.terminal) + return Effect.raceFirst( + Deferred.await(request.terminal).pipe(Effect.as(true)), + Deferred.await(shutdown).pipe(Effect.as(false)), + ).pipe(Effect.flatMap((retry) => (retry ? run(key) : Effect.interrupt))) }), + restore, ) }), ) - const wake = (key: Key, seq?: number) => { - return Effect.uninterruptible( + const wake = (key: Key, seq?: number) => + Effect.uninterruptible( Effect.suspend(() => { - return transition({ _tag: "Wake", key, seq }).pipe( - Effect.flatMap((result) => execute(result.actions, { _tag: "General" })), - ) + return SynchronizedRef.modify(state, (current): readonly [Start | undefined, State] => { + if (current.closed) return [undefined, current] + const boundary = current.interruptSeq.get(key) + if (boundary !== undefined && (seq === undefined || seq <= boundary)) return [undefined, current] + const lane = current.lanes.get(key) + if (lane === undefined) { + const next = start({ state: current, key, demand: Demand.wake(seq) }) + return [next.start, next.state] + } + if ( + lane.lifecycle._tag === "Stopping" && + (lane.lifecycle.boundary === undefined || seq === undefined || seq <= lane.lifecycle.boundary) + ) + return [undefined, current] + const pending = lane.pending.combine(Demand.wake(seq)) + if (Equal.equals(pending, lane.pending)) return [undefined, current] + return [undefined, updateLane(current, key, { ...lane, pending })] + }).pipe(Effect.flatMap((instruction) => (instruction === undefined ? Effect.void : launch(instruction)))) }), ) - } - const interrupt = (key: Key, seq?: number) => dispatch({ _tag: "Interrupt", key, seq }) + const interrupt = (key: Key, seq?: number) => + Effect.uninterruptible( + SynchronizedRef.modify(state, (current) => { + if (current.closed) return [undefined, current] as const + const latest = current.interruptSeq.get(key) + const lane = current.lanes.get(key) + if (seq !== undefined && latest !== undefined && seq <= latest) + return [lane?.lifecycle._tag === "Stopping" ? lane.lifecycle.owner : undefined, current] as const + + const bounded = (() => { + if (seq === undefined) return current + const interruptSeq = new Map(current.interruptSeq) + interruptSeq.set(key, seq) + return { ...current, interruptSeq } + })() + if (lane === undefined) return [undefined, bounded] as const + if ( + !lane.current.explicit && + seq !== undefined && + lane.current.wakeSeq !== undefined && + lane.current.wakeSeq > seq + ) + return [undefined, bounded] as const + + const pending = lane.current.afterBoundary(seq).combine(lane.pending.afterBoundary(seq)) + const boundary = + lane.lifecycle._tag === "Stopping" && lane.lifecycle.boundary !== undefined && seq !== undefined + ? Math.max(lane.lifecycle.boundary, seq) + : lane.lifecycle._tag === "Stopping" && seq === undefined + ? lane.lifecycle.boundary + : seq + return [ + lane.lifecycle.owner, + updateLane(bounded, key, { + ...lane, + pending, + lifecycle: { _tag: "Stopping", token: lane.lifecycle.token, owner: lane.lifecycle.owner, boundary }, + }), + ] as const + }).pipe( + Effect.flatMap((owner) => + owner === undefined ? Effect.void : Deferred.await(owner).pipe(Effect.flatMap(Fiber.interrupt)), + ), + ), + ) const awaitIdle = (key: Key): Effect.Effect => Effect.gen(function* () { let failure: Cause.Cause | undefined while (true) { - const observation = yield* transition({ _tag: "Observe", key }) - if (observation.response._tag !== "Observe") break + const terminal = (yield* SynchronizedRef.get(state)).lanes.get(key)?.terminal + if (terminal === undefined) break const exit = yield* Effect.raceFirst( - Deferred.await(observation.response.deferred), + Deferred.await(terminal), Deferred.await(shutdown).pipe(Effect.as(Exit.void)), ) if (exit._tag === "Failure" && failure === undefined) failure = exit.cause } if (failure !== undefined) return yield* Effect.failCause(failure) - return undefined }) yield* Effect.addFinalizer(() => - transition({ _tag: "Close" }).pipe(Effect.andThen(Effect.sync(() => Deferred.doneUnsafe(shutdown, Effect.void)))), + SynchronizedRef.modify(state, (_current) => [ + undefined, + { closed: true, lanes: new Map(), interruptSeq: new Map() } satisfies State, + ]).pipe(Effect.andThen(Deferred.succeed(shutdown, undefined))), ) return { run, wake, interrupt, awaitIdle } - - function awaitResult(deferred: Deferred.Deferred) { - return Effect.raceFirst(Deferred.await(deferred), Deferred.await(shutdown).pipe(Effect.andThen(Effect.interrupt))) - } }) export interface Interface extends Coordinator {} diff --git a/packages/core/test/session-run-coordinator-machine.test.ts b/packages/core/test/session-run-coordinator-machine.test.ts deleted file mode 100644 index 87a2523414b..00000000000 --- a/packages/core/test/session-run-coordinator-machine.test.ts +++ /dev/null @@ -1,213 +0,0 @@ -import { describe, expect, test } from "bun:test" -import { SessionRunCoordinatorMachine } from "../src/session/run-coordinator-machine" - -const Machine = SessionRunCoordinatorMachine - -describe("SessionRunCoordinatorMachine.Demand", () => { - test("empty is the combine identity", () => { - const demand = SessionRunCoordinatorMachine.Demand.combine( - SessionRunCoordinatorMachine.Demand.explicit, - SessionRunCoordinatorMachine.Demand.wake(3), - ) - - expect(SessionRunCoordinatorMachine.Demand.combine(SessionRunCoordinatorMachine.Demand.empty, demand)).toEqual( - demand, - ) - expect(SessionRunCoordinatorMachine.Demand.combine(demand, SessionRunCoordinatorMachine.Demand.empty)).toEqual( - demand, - ) - }) - - test("combine is associative, commutative, and idempotent", () => { - const left = Machine.Demand.explicit - const middle = Machine.Demand.wake() - const right = Machine.Demand.wake(3) - - expect(Machine.Demand.combine(left, right)).toEqual(Machine.Demand.combine(right, left)) - expect(Machine.Demand.combine(left, left)).toEqual(left) - expect(Machine.Demand.combine(Machine.Demand.combine(left, middle), right)).toEqual( - Machine.Demand.combine(left, Machine.Demand.combine(middle, right)), - ) - }) - - test("afterBoundary removes explicit and stale wake components", () => { - const combined = Machine.Demand.combine(Machine.Demand.explicit, Machine.Demand.wake(3)) - - expect(Machine.Demand.afterBoundary(combined, 2)).toEqual(Machine.Demand.wake(3)) - expect(Machine.Demand.nonEmpty(Machine.Demand.afterBoundary(combined, 3))).toBeFalse() - expect(Machine.Demand.nonEmpty(Machine.Demand.afterBoundary(combined))).toBeFalse() - }) - - test("mode follows only the explicit component", () => { - expect(Machine.Demand.mode(Machine.Demand.explicit)).toBe("run") - expect(Machine.Demand.mode(Machine.Demand.wake(1))).toBe("wake") - expect(Machine.Demand.mode(Machine.Demand.combine(Machine.Demand.explicit, Machine.Demand.wake(1)))).toBe("run") - }) -}) - -describe("SessionRunCoordinatorMachine.reduce", () => { - test("ignores a stale attempt from the active chain", () => { - const active = combinedActive(3) - const result = Machine.reduce(active, { - _tag: "Settled", - key: "session", - chain: 1, - attempt: 2, - outcome: "Success", - }) - - expect(result.state).toBe(active) - expect(result.actions).toEqual([]) - expect(result.response).toEqual({ _tag: "None" }) - }) - - test("ignores duplicate and foreign settlements", () => { - const active = combinedActive(3) - const foreign = Machine.reduce(active, { - _tag: "Settled", - key: "session", - chain: 99, - attempt: 100, - outcome: "Failure", - }) - const idle = Machine.reduce(Machine.initial(), { - _tag: "Settled", - key: "session", - chain: 1, - attempt: 2, - outcome: "Success", - }) - - expect(foreign).toEqual({ state: active, actions: [], response: { _tag: "None" } }) - expect(idle.actions).toEqual([]) - expect(idle.state).toEqual(Machine.initial()) - }) - - test("completes the superseded chain when creating its successor", () => { - const interrupted = Machine.reduce(combinedActive(3), { _tag: "Interrupt", key: "session", seq: 2 }) - const settled = Machine.reduce(interrupted.state, { - _tag: "Settled", - key: "session", - chain: 1, - attempt: 4, - outcome: "Interrupted", - }) - - expect(settled.actions).toContainEqual({ _tag: "CompleteChain", chain: 1 }) - expect(settled.state.lanes.get("session")?.chain).not.toBe(1) - }) - - test("returns caller observation separately from executable actions", () => { - const result = Machine.reduce(Machine.initial(), { - _tag: "Run", - key: "session", - }) - - expect(result.response).toEqual({ _tag: "AwaitChain", chain: 1 }) - expect(result.actions).toEqual([ - { - _tag: "Start", - key: "session", - chain: 1, - attempt: 2, - demand: Machine.Demand.explicit, - successor: false, - }, - ]) - expect(result.state.nextID).toBe(3) - }) - - test("allocates only identities selected by each transition", () => { - const woken = Machine.reduce(Machine.initial(), { _tag: "Wake", key: "session", seq: 1 }) - expect(woken.state.nextID).toBe(3) - - const coalesced = Machine.reduce(woken.state, { _tag: "Wake", key: "session", seq: 2 }) - expect(coalesced.state.nextID).toBe(3) - - const explicit = Machine.reduce(coalesced.state, { _tag: "Run", key: "session" }) - expect(explicit.state.nextID).toBe(4) - - const joined = Machine.reduce(explicit.state, { _tag: "Run", key: "session" }) - expect(joined.state.nextID).toBe(4) - - const continued = Machine.reduce(joined.state, { - _tag: "Settled", - key: "session", - chain: 1, - attempt: 2, - outcome: "Success", - }) - expect(continued.state.nextID).toBe(5) - expect(continued.state.lanes.get("session")?.attempt).toBe(4) - }) - - test("interrupting an active combined demand preserves its newer wake as an advisory successor", () => { - const active = combinedActive(3) - const interrupted = Machine.reduce(active, { _tag: "Interrupt", key: "session", seq: 2 }) - - expect(interrupted.actions).toEqual([{ _tag: "Interrupt", attempt: 4 }]) - expect(interrupted.state.lanes.get("session")?.pending).toEqual(Machine.Demand.wake(3)) - - const settled = Machine.reduce(interrupted.state, { - _tag: "Settled", - key: "session", - chain: 1, - attempt: 4, - outcome: "Interrupted", - }) - expect(settled.state.lanes.get("session")?.current).toEqual(Machine.Demand.wake(3)) - expect(settled.actions).toContainEqual({ - _tag: "Start", - key: "session", - chain: 5, - attempt: 6, - demand: Machine.Demand.wake(3), - successor: true, - }) - }) - - test("interrupting an active combined demand suppresses its wake at the boundary", () => { - const active = combinedActive(2) - const interrupted = Machine.reduce(active, { _tag: "Interrupt", key: "session", seq: 2 }) - const pending = interrupted.state.lanes.get("session")?.pending - - expect(pending).toBeDefined() - if (pending === undefined) throw new Error("Missing stopping lane") - expect(Machine.Demand.nonEmpty(pending)).toBeFalse() - - const settled = Machine.reduce(interrupted.state, { - _tag: "Settled", - key: "session", - chain: 1, - attempt: 4, - outcome: "Interrupted", - }) - expect(settled.state.lanes.has("session")).toBeFalse() - expect(settled.actions.some((action) => action._tag === "Start")).toBeFalse() - }) -}) - -function combinedActive(seq: number) { - const woken = Machine.reduce(Machine.initial(), { - _tag: "Wake", - key: "session", - seq: 1, - }) - const explicit = Machine.reduce(woken.state, { - _tag: "Run", - key: "session", - }) - const pending = Machine.reduce(explicit.state, { - _tag: "Wake", - key: "session", - seq, - }) - const active = Machine.reduce(pending.state, { - _tag: "Settled", - key: "session", - chain: 1, - attempt: 2, - outcome: "Success", - }) - return active.state -} diff --git a/packages/core/test/session-run-coordinator.test.ts b/packages/core/test/session-run-coordinator.test.ts index 429d3123ff3..502f6bcc858 100644 --- a/packages/core/test/session-run-coordinator.test.ts +++ b/packages/core/test/session-run-coordinator.test.ts @@ -1,10 +1,34 @@ -import { describe, expect } from "bun:test" -import { Cause, Deferred, Effect, Exit, Fiber, Layer, Scope } from "effect" +import { describe, expect, test } from "bun:test" +import { Cause, Deferred, Effect, Equal, Exit, Fiber, Layer, Scope } from "effect" import { SessionRunCoordinator } from "@opencode-ai/core/session/run-coordinator" import { testEffect } from "./lib/effect" const it = testEffect(Layer.empty) +describe("SessionRunCoordinator.Demand", () => { + const Demand = SessionRunCoordinator.Demand + + test("combines associatively with an identity", () => { + const left = Demand.run.combine(Demand.wake(1)) + const right = Demand.wake().combine(Demand.wake(3)) + + expect(Equal.equals(Demand.empty.combine(left), left)).toBeTrue() + expect(Equal.equals(left.combine(Demand.empty), left)).toBeTrue() + expect(Equal.equals(left.combine(right), right.combine(left))).toBeTrue() + expect( + Equal.equals(left.combine(right).combine(Demand.wake(2)), left.combine(right.combine(Demand.wake(2)))), + ).toBeTrue() + }) + + test("keeps only sequenced wakes newer than an interrupt boundary", () => { + const demand = Demand.run.combine(Demand.wake()).combine(Demand.wake(3)) + + expect(Equal.equals(demand.afterBoundary(2), Demand.wake(3))).toBeTrue() + expect(Equal.equals(demand.afterBoundary(3), Demand.empty)).toBeTrue() + expect(Equal.equals(demand.afterBoundary(), Demand.empty)).toBeTrue() + }) +}) + describe("SessionRunCoordinator", () => { it.effect("joins concurrent resumes for one key", () => Effect.scoped( @@ -461,6 +485,53 @@ describe("SessionRunCoordinator", () => { ), ) + it.effect("does not let an interrupted attempt completion settle its successor", () => + Effect.scoped( + Effect.gen(function* () { + const firstStarted = yield* Deferred.make() + const cleanupStarted = yield* Deferred.make() + const cleanupGate = yield* Deferred.make() + const secondStarted = yield* Deferred.make() + const secondGate = yield* Deferred.make() + const idleSettled = yield* Deferred.make() + let runs = 0 + const coordinator = yield* SessionRunCoordinator.make({ + drain: () => + Effect.sync(() => ++runs).pipe( + Effect.flatMap((run) => + run === 1 + ? Deferred.succeed(firstStarted, undefined).pipe( + Effect.andThen(Effect.never), + Effect.onInterrupt(() => + Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))), + ), + ) + : Deferred.succeed(secondStarted, undefined).pipe(Effect.andThen(Deferred.await(secondGate))), + ), + ), + }) + + yield* coordinator.wake("session", 1) + yield* Deferred.await(firstStarted) + const interrupt = yield* coordinator.interrupt("session", 2).pipe(Effect.forkChild) + yield* Deferred.await(cleanupStarted) + yield* coordinator.wake("session", 3) + yield* Deferred.succeed(cleanupGate, undefined) + yield* Fiber.join(interrupt) + yield* Deferred.await(secondStarted) + const idle = yield* coordinator + .awaitIdle("session") + .pipe(Effect.ensuring(Deferred.succeed(idleSettled, undefined)), Effect.forkChild) + + yield* Effect.yieldNow + expect(yield* Deferred.isDone(idleSettled)).toBeFalse() + yield* Deferred.succeed(secondGate, undefined) + yield* Fiber.join(idle) + expect(runs).toBe(2) + }), + ), + ) + it.effect("interrupts an explicit run queued before the interruption request", () => Effect.scoped( Effect.gen(function* () {