diff --git a/packages/core/src/session/run-coordinator-machine.ts b/packages/core/src/session/run-coordinator-machine.ts new file mode 100644 index 0000000000..5afa1cdbd0 --- /dev/null +++ b/packages/core/src/session/run-coordinator-machine.ts @@ -0,0 +1,358 @@ +/** @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 5068b913ad..021c8e0c59 100644 --- a/packages/core/src/session/run-coordinator.ts +++ b/packages/core/src/session/run-coordinator.ts @@ -1,267 +1,240 @@ export * as SessionRunCoordinator from "./run-coordinator" -import { Cause, Context, Deferred, Effect, Exit, Fiber, FiberSet, Layer, Scope } from "effect" +import { Cause, Context, Deferred, Effect, 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 = "run" | "wake" +export type Mode = SessionRunCoordinatorMachine.Mode -/** Why one drain generation should run. Explicit runs dominate advisory wakes when demands coalesce. */ -type Demand = { readonly _tag: "run" } | { readonly _tag: "wake"; readonly seq?: number } - -/** - * Runs at most one drain chain per key while allowing different keys to drain concurrently. - * - * For each key: - * - * idle --run/wake--> draining --run/wake--> draining + one coalesced rerun --> idle - * - * `run` is an explicit drain request. It starts a chain or joins the current chain and - * upgrades a pending follow-up so the caller receives explicit-run semantics. - * - * `wake` reports that durable work may now be available. It starts a chain while idle or - * requests one coalesced follow-up while draining. Repeated wakes collapse together. - * - * `interrupt` stops the current ownership chain. Advisory wakes from before the interrupt - * boundary are suppressed; advisory wakes after the boundary run after cleanup. - */ export interface Coordinator { - /** Starts or joins one explicit drain generation. */ readonly run: (key: Key) => Effect.Effect - /** Coalesces one wake-up after durable work is recorded. */ readonly wake: (key: Key, seq?: number) => Effect.Effect - /** Waits until the current ownership chain settles. */ readonly awaitIdle: (key: Key) => Effect.Effect - /** Interrupts the active ownership chain without automatically draining pending wakes. */ readonly interrupt: (key: Key, seq?: number) => Effect.Effect } -/** One Session's process-local execution lane: one active demand and at most one coalesced follow-up. */ -type Entry = { +type Chain = { readonly done: Deferred.Deferred readonly settled: Deferred.Deferred> - current: Demand - pending?: Demand - explicitWaiter?: Deferred.Deferred - interruptSeq?: number - owner?: Fiber.Fiber - stopping: boolean } -/** Combines follow-up demand: runs dominate, while wakes retain the newest durable admission sequence. */ -const coalesce = (left: Demand | undefined, right: Demand): Demand => { - if (left?._tag === "run" || right._tag === "run") return { _tag: "run" } - return { _tag: "wake", seq: maxSeq(left?.seq, right.seq) } -} - -const maxSeq = (left: number | undefined, right: number | undefined) => { - if (left === undefined) return right - if (right === undefined) return left - return Math.max(left, right) -} - -/** Constructs a scoped coordinator. Every in-memory transition is synchronous. */ export const make = (options: { readonly drain: (key: Key, mode: Mode) => Effect.Effect readonly onFailure?: (key: Key, cause: Cause.Cause) => Effect.Effect }): Effect.Effect, never, Scope.Scope> => Effect.gen(function* () { - const active = new Map>() - const interruptSeq = new Map() + const state = yield* SynchronizedRef.make(SessionRunCoordinatorMachine.initial()) const report = yield* FiberSet.makeRuntime() const fork = yield* FiberSet.makeRuntime() + const chains = new Map>() + const waiters = new Map>() + const owners = new Map>>() const shutdown = Deferred.makeUnsafe() - let closed = false - yield* Effect.addFinalizer(() => - Effect.sync(() => { - closed = true - Deferred.doneUnsafe(shutdown, Effect.void) - active.clear() - interruptSeq.clear() - }), - ) - const makeEntry = (current: Demand, explicitWaiter?: Deferred.Deferred): Entry => ({ - done: Deferred.makeUnsafe(), - settled: Deferred.makeUnsafe>(), - current, - explicitWaiter, - stopping: false, - }) - - const start = (key: Key, entry: Entry, demand: Demand, successor = false) => { - const ready = Deferred.makeUnsafe() - const drain = Effect.suspend(() => options.drain(key, demand._tag)) - // Initial work retains immediate-start behavior but cannot run before ownership is published. - // Observer-started successors yield once so synchronous drains cannot recurse on the JS stack. - const owner = fork( - (successor - ? Effect.yieldNow.pipe(Effect.andThen(drain)) - : Deferred.await(ready).pipe(Effect.andThen(drain)) - ).pipe( - Effect.onExit((exit) => Effect.sync(() => settle(key, entry, demand, exit))), - Effect.exit, - Effect.asVoid, - ), - ) - entry.owner = owner - if (!successor) Deferred.doneUnsafe(ready, Effect.void) + 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 settle = (key: Key, entry: Entry, demand: Demand, exit: Exit.Exit) => { - if (closed) { - Deferred.doneUnsafe(entry.done, exit) - Deferred.doneUnsafe(entry.settled, Effect.succeed(exit)) - return - } - if (demand._tag === "run" && entry.explicitWaiter !== undefined) { - Deferred.doneUnsafe(entry.explicitWaiter, exit) - entry.explicitWaiter = undefined - } - if (entry.stopping && demand._tag === "wake" && entry.explicitWaiter !== undefined) { - Deferred.doneUnsafe(entry.explicitWaiter, exit) - entry.explicitWaiter = undefined - } - if (active.get(key) !== entry) { - Deferred.doneUnsafe(entry.done, exit) - Deferred.doneUnsafe(entry.settled, Effect.succeed(exit)) - return - } - if (exit._tag === "Success" && !entry.stopping) { - if (entry.pending !== undefined) { - const pending = entry.pending - entry.pending = undefined - entry.current = pending - start(key, entry, pending, true) - return - } - active.delete(key) - Deferred.doneUnsafe(entry.done, exit) - Deferred.doneUnsafe(entry.settled, Effect.succeed(exit)) - return - } - - const successor = entry.pending !== undefined ? makeEntry(entry.pending, entry.explicitWaiter) : undefined - if (successor === undefined) active.delete(key) - else active.set(key, successor) - if (successor !== undefined) start(key, successor, successor.current, true) - Deferred.doneUnsafe(entry.done, exit) - Deferred.doneUnsafe(entry.settled, Effect.succeed(exit)) - if ( - exit._tag === "Failure" && - !(entry.stopping && Cause.hasInterruptsOnly(exit.cause)) && - demand._tag === "wake" && - options.onFailure !== undefined - ) { - report(Effect.suspend(() => options.onFailure!(key, exit.cause))) - } + 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 wake = (key: Key, seq?: number) => - Effect.sync(() => { - if (closed) return - if (!isAfterInterrupt(key, seq)) return - const entry = active.get(key) - if (entry !== undefined) { - if (!acceptsWake(entry, seq)) return - entry.pending = coalesce(entry.pending, { _tag: "wake", seq }) - return - } + type RuntimeResponse = + | { readonly _tag: "None" | "Idle" | "Closed" } + | { readonly _tag: "Await"; readonly deferred: Deferred.Deferred } + | { readonly _tag: "Retry" | "Observe"; readonly deferred: Deferred.Deferred> } - const next = makeEntry({ _tag: "wake", seq }) - active.set(key, next) - start(key, next, next.current) + 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 + }) }) + type Execution = { readonly _tag: "General" } | { readonly _tag: "Settlement"; readonly exit: Exit.Exit } + + 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 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))), + ) + } + + const dispatch = (event: SessionRunCoordinatorMachine.Event) => + Effect.uninterruptible( + transition(event).pipe(Effect.flatMap((result) => execute(result.actions, { _tag: "General" }))), + ) + + 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, + ) + }), + ) + }), + ) + + const wake = (key: Key, seq?: number) => { + return Effect.uninterruptible( + Effect.suspend(() => { + return transition({ _tag: "Wake", key, seq }).pipe( + Effect.flatMap((result) => execute(result.actions, { _tag: "General" })), + ) + }), + ) + } + + const interrupt = (key: Key, seq?: number) => dispatch({ _tag: "Interrupt", key, seq }) + const awaitIdle = (key: Key): Effect.Effect => Effect.gen(function* () { - let firstFailure: Cause.Cause | undefined - while (!closed) { - const entry = active.get(key) - if (entry === undefined) break + let failure: Cause.Cause | undefined + while (true) { + const observation = yield* transition({ _tag: "Observe", key }) + if (observation.response._tag !== "Observe") break const exit = yield* Effect.raceFirst( - Deferred.await(entry.settled), + Deferred.await(observation.response.deferred), Deferred.await(shutdown).pipe(Effect.as(Exit.void)), ) - if (closed) break - if (exit._tag === "Failure" && firstFailure === undefined) firstFailure = exit.cause + if (exit._tag === "Failure" && failure === undefined) failure = exit.cause } - if (firstFailure !== undefined) return yield* Effect.failCause(firstFailure) + if (failure !== undefined) return yield* Effect.failCause(failure) + return undefined }) - const interrupt = (key: Key, seq?: number): Effect.Effect => - Effect.suspend(() => { - const entry = active.get(key) - const latest = interruptSeq.get(key) - if (seq !== undefined && latest !== undefined && seq <= latest) - return entry?.stopping && entry.owner !== undefined ? Fiber.interrupt(entry.owner) : Effect.void - if (seq !== undefined) interruptSeq.set(key, seq) - if (entry?.owner === undefined) return Effect.void - if ( - seq !== undefined && - entry.current._tag === "wake" && - entry.current.seq !== undefined && - entry.current.seq > seq - ) - return Effect.void - if (entry.stopping) { - entry.interruptSeq = maxSeq(entry.interruptSeq, seq) - suppressPendingAtOrBefore(entry, seq) - return Fiber.interrupt(entry.owner) - } - entry.stopping = true - entry.interruptSeq = seq - suppressPendingAtOrBefore(entry, seq) - return Fiber.interrupt(entry.owner) - }) + yield* Effect.addFinalizer(() => + transition({ _tag: "Close" }).pipe(Effect.andThen(Effect.sync(() => Deferred.doneUnsafe(shutdown, Effect.void)))), + ) - return { run, wake, awaitIdle, interrupt } + return { run, wake, interrupt, awaitIdle } - function run(key: Key): Effect.Effect { - return Effect.uninterruptibleMask((restore) => { - if (closed) return Effect.interrupt - const entry = active.get(key) - if (entry !== undefined) { - if (entry.stopping) { - return restore(Deferred.await(entry.settled).pipe(Effect.andThen(run(key)))) - } - if (entry.current._tag === "wake") { - entry.pending = coalesce(entry.pending, { _tag: "run" }) - entry.explicitWaiter ??= Deferred.makeUnsafe() - return restore(awaitRun(entry.explicitWaiter)) - } - return restore(awaitRun(entry.done)) - } - - const next = makeEntry({ _tag: "run" }) - active.set(key, next) - start(key, next, next.current) - return restore(awaitRun(next.done)) - }) - } - - function awaitRun(done: Deferred.Deferred): Effect.Effect { - return Effect.raceFirst(Deferred.await(done), Deferred.await(shutdown).pipe(Effect.andThen(Effect.interrupt))) - } - - function acceptsWake(entry: Entry, seq: number | undefined) { - return !entry.stopping || (entry.interruptSeq !== undefined && seq !== undefined && seq > entry.interruptSeq) - } - - function isAfterInterrupt(key: Key, seq: number | undefined) { - const latest = interruptSeq.get(key) - return latest === undefined || (seq !== undefined && seq > latest) - } - - function suppressPendingAtOrBefore(entry: Entry, seq: number | undefined) { - if ( - entry.pending?._tag === "wake" && - seq !== undefined && - entry.pending.seq !== undefined && - entry.pending.seq > seq - ) - return - entry.pending = undefined + function awaitResult(deferred: Deferred.Deferred) { + return Effect.raceFirst(Deferred.await(deferred), Deferred.await(shutdown).pipe(Effect.andThen(Effect.interrupt))) } }) diff --git a/packages/core/test/session-run-coordinator-machine.test.ts b/packages/core/test/session-run-coordinator-machine.test.ts new file mode 100644 index 0000000000..87a2523414 --- /dev/null +++ b/packages/core/test/session-run-coordinator-machine.test.ts @@ -0,0 +1,213 @@ +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 39fb5779d3..429d3123ff 100644 --- a/packages/core/test/session-run-coordinator.test.ts +++ b/packages/core/test/session-run-coordinator.test.ts @@ -29,6 +29,51 @@ describe("SessionRunCoordinator", () => { ), ) + it.effect("allocates fresh ownership when one run effect is reused", () => + Effect.scoped( + Effect.gen(function* () { + let runs = 0 + const coordinator = yield* SessionRunCoordinator.make({ drain: () => Effect.sync(() => runs++) }) + const run = coordinator.run("session") + + yield* run + yield* run + + expect(runs).toBe(2) + }), + ), + ) + + it.effect("captures awaitIdle chains safely while settlement races", () => + Effect.scoped( + Effect.gen(function* () { + const iterations = 500 + const gates = Array.from({ length: iterations }, () => Deferred.makeUnsafe()) + let runs = 0 + const coordinator = yield* SessionRunCoordinator.make({ + drain: () => + Effect.suspend(() => { + const gate = gates[runs++] + return gate === undefined ? Effect.die("Missing test gate") : Deferred.await(gate) + }), + }) + + for (let index = 0; index < iterations; index++) { + const run = yield* coordinator.run("session").pipe(Effect.forkChild) + yield* Effect.yieldNow + const idle = yield* coordinator.awaitIdle("session").pipe(Effect.forkChild({ startImmediately: true })) + const gate = gates[index] + if (gate === undefined) yield* Effect.die("Missing test gate") + yield* Deferred.succeed(gate, undefined) + yield* Effect.all([Fiber.join(run), Fiber.join(idle)]) + } + + expect(runs).toBe(iterations) + return undefined + }), + ), + ) + it.effect("starts a drain when woken while idle", () => Effect.scoped( Effect.gen(function* () { @@ -122,6 +167,106 @@ describe("SessionRunCoordinator", () => { ), ) + it.effect("preserves a newer wake coalesced behind a pending explicit run", () => + Effect.scoped( + Effect.gen(function* () { + const firstStarted = yield* Deferred.make() + const secondStarted = yield* Deferred.make() + const modes: SessionRunCoordinator.Mode[] = [] + const coordinator = yield* SessionRunCoordinator.make({ + drain: (_key, mode) => + Effect.sync(() => modes.push(mode)).pipe( + Effect.flatMap((run) => + run === 1 + ? Deferred.succeed(firstStarted, undefined).pipe(Effect.andThen(Effect.never)) + : Deferred.succeed(secondStarted, undefined), + ), + ), + }) + + yield* coordinator.wake("session", 1) + yield* Deferred.await(firstStarted) + const run = yield* coordinator.run("session").pipe(Effect.exit, Effect.forkChild) + yield* Effect.yieldNow + yield* coordinator.wake("session", 3) + yield* coordinator.interrupt("session", 2) + yield* Deferred.await(secondStarted) + yield* coordinator.awaitIdle("session").pipe(Effect.exit) + + const runExit = yield* Fiber.join(run) + expect(Exit.isFailure(runExit) && Cause.hasInterruptsOnly(runExit.cause)).toBeTrue() + expect(modes).toEqual(["wake", "wake"]) + }), + ), + ) + + it.effect("preserves a newer wake from an interrupted active combined demand", () => + Effect.scoped( + Effect.gen(function* () { + const firstGate = yield* Deferred.make() + const secondStarted = yield* Deferred.make() + const thirdStarted = yield* Deferred.make() + const modes: SessionRunCoordinator.Mode[] = [] + const coordinator = yield* SessionRunCoordinator.make({ + drain: (_key, mode) => + Effect.sync(() => modes.push(mode)).pipe( + Effect.flatMap((run) => { + if (run === 1) return Deferred.await(firstGate) + if (run === 2) return Deferred.succeed(secondStarted, undefined).pipe(Effect.andThen(Effect.never)) + return Deferred.succeed(thirdStarted, undefined) + }), + ), + }) + + yield* coordinator.wake("session", 1) + const run = yield* coordinator.run("session").pipe(Effect.exit, Effect.forkChild) + yield* Effect.yieldNow + yield* coordinator.wake("session", 3) + yield* Deferred.succeed(firstGate, undefined) + yield* Deferred.await(secondStarted) + yield* coordinator.interrupt("session", 2) + yield* Deferred.await(thirdStarted) + yield* coordinator.awaitIdle("session").pipe(Effect.exit) + + const runExit = yield* Fiber.join(run) + expect(Exit.isFailure(runExit) && Cause.hasInterruptsOnly(runExit.cause)).toBeTrue() + expect(modes).toEqual(["wake", "run", "wake"]) + }), + ), + ) + + it.effect("suppresses an older wake from an interrupted active combined demand", () => + Effect.scoped( + Effect.gen(function* () { + const firstGate = yield* Deferred.make() + const secondStarted = yield* Deferred.make() + const modes: SessionRunCoordinator.Mode[] = [] + const coordinator = yield* SessionRunCoordinator.make({ + drain: (_key, mode) => + Effect.sync(() => modes.push(mode)).pipe( + Effect.flatMap((run) => { + if (run === 1) return Deferred.await(firstGate) + return Deferred.succeed(secondStarted, undefined).pipe(Effect.andThen(Effect.never)) + }), + ), + }) + + yield* coordinator.wake("session", 1) + const run = yield* coordinator.run("session").pipe(Effect.exit, Effect.forkChild) + yield* Effect.yieldNow + yield* coordinator.wake("session", 2) + yield* Deferred.succeed(firstGate, undefined) + yield* Deferred.await(secondStarted) + yield* coordinator.interrupt("session", 2) + yield* coordinator.awaitIdle("session").pipe(Effect.exit) + + const runExit = yield* Fiber.join(run) + expect(Exit.isFailure(runExit) && Cause.hasInterruptsOnly(runExit.cause)).toBeTrue() + expect(modes).toEqual(["wake", "run"]) + }), + ), + ) + it.effect("interrupts only the requested key", () => Effect.scoped( Effect.gen(function* () { @@ -847,6 +992,37 @@ describe("SessionRunCoordinator", () => { }), ) + it.effect("settles a post-stop run waiter when its owning scope closes", () => + Effect.gen(function* () { + const scope = yield* Scope.make() + const started = yield* Deferred.make() + const cleanupStarted = yield* Deferred.make() + const cleanupGate = yield* Deferred.make() + const coordinator = yield* SessionRunCoordinator.make({ + drain: () => + Deferred.succeed(started, undefined).pipe( + Effect.andThen(Effect.never), + Effect.onInterrupt(() => + Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))), + ), + ), + }).pipe(Scope.provide(scope)) + + yield* coordinator.wake("session") + yield* Deferred.await(started) + const interrupt = yield* coordinator.interrupt("session").pipe(Effect.forkChild) + yield* Deferred.await(cleanupStarted) + const run = yield* coordinator.run("session").pipe(Effect.forkChild) + const close = yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild) + + const runExit = yield* Fiber.await(run) + expect(Exit.isFailure(runExit) && Cause.hasInterruptsOnly(runExit.cause)).toBeTrue() + yield* Deferred.succeed(cleanupGate, undefined) + yield* Fiber.join(interrupt) + yield* Fiber.join(close) + }), + ) + it.effect("does not start work after its owning scope closes", () => Effect.gen(function* () { const scope = yield* Scope.make()