refactor(core): extract session run coordinator machine
This commit is contained in:
Kit Langton
parent
1025540fcc
commit
fc0cf2a710
4 changed files with 940 additions and 220 deletions
213
packages/core/test/session-run-coordinator-machine.test.ts
Normal file
213
packages/core/test/session-run-coordinator-machine.test.ts
Normal file
|
|
@ -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<string>(), {
|
||||
_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<string>())
|
||||
})
|
||||
|
||||
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<string>(), {
|
||||
_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<string>(), { _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<string>(), {
|
||||
_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
|
||||
}
|
||||
|
|
@ -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<void>())
|
||||
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<void>()
|
||||
const secondStarted = yield* Deferred.make<void>()
|
||||
const modes: SessionRunCoordinator.Mode[] = []
|
||||
const coordinator = yield* SessionRunCoordinator.make<string, void, never>({
|
||||
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<void>()
|
||||
const secondStarted = yield* Deferred.make<void>()
|
||||
const thirdStarted = yield* Deferred.make<void>()
|
||||
const modes: SessionRunCoordinator.Mode[] = []
|
||||
const coordinator = yield* SessionRunCoordinator.make<string, void, never>({
|
||||
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<void>()
|
||||
const secondStarted = yield* Deferred.make<void>()
|
||||
const modes: SessionRunCoordinator.Mode[] = []
|
||||
const coordinator = yield* SessionRunCoordinator.make<string, void, never>({
|
||||
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<void>()
|
||||
const cleanupStarted = yield* Deferred.make<void>()
|
||||
const cleanupGate = yield* Deferred.make<void>()
|
||||
const coordinator = yield* SessionRunCoordinator.make<string, void, never>({
|
||||
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()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue