|
|
@@ -1,6 +1,7 @@
|
|
|
export * as SessionRunCoordinator from "./run-coordinator.js"
|
|
|
|
|
|
import { Deferred, Effect, Exit, Fiber, FiberSet, Scope } from "effect"
|
|
|
+import type { Promotable } from "./inbox.js"
|
|
|
|
|
|
/** Serializes execution for each key while allowing different keys to run concurrently. */
|
|
|
export interface Coordinator<Key, E, Reason = never> {
|
|
|
@@ -9,26 +10,41 @@ export interface Coordinator<Key, E, Reason = never> {
|
|
|
/** Starts an execution while idle, or joins the active execution and returns its exit. */
|
|
|
readonly run: (key: Key) => Effect.Effect<void, E>
|
|
|
/** Rings the doorbell: an idle key starts an execution; an active one drains again before settling. */
|
|
|
- readonly wake: (key: Key) => Effect.Effect<void>
|
|
|
+ readonly wake: (key: Key, request?: Request) => Effect.Effect<void>
|
|
|
+ /** Rings the current execution's doorbell with its existing request. Idle keys remain idle. */
|
|
|
+ readonly wakeActive: (key: Key) => Effect.Effect<void>
|
|
|
/** Stops the active execution, clears its doorbell, and waits for cleanup. No-op when idle. */
|
|
|
- readonly interrupt: (key: Key, reason?: Reason) => Effect.Effect<void>
|
|
|
+ readonly interrupt: (
|
|
|
+ key: Key,
|
|
|
+ reason?: Reason,
|
|
|
+ options?: { readonly continue?: { readonly request: Request; readonly when: Effect.Effect<boolean> } },
|
|
|
+ ) => Effect.Effect<void>
|
|
|
/** Resolves once no execution is active for the key. Returns immediately when already idle and never starts work. */
|
|
|
readonly awaitIdle: (key: Key) => Effect.Effect<void>
|
|
|
}
|
|
|
|
|
|
+export type Request = Promotable
|
|
|
+
|
|
|
/**
|
|
|
* One execution is a busy period for one key: one fiber that drains from the first wake
|
|
|
* until the key would stay idle. `pendingWake` is the doorbell: work recorded during the
|
|
|
- * execution rings it, and the execution loop drains again instead of ending. The doorbell
|
|
|
- * closes the gap between a drain's last eligibility check and the idle transition, since
|
|
|
- * those cannot be one atomic step. `done` resolves joiners with this execution's exit.
|
|
|
+ * execution rings it with its eligibility request, and the execution loop drains again
|
|
|
+ * instead of ending. The doorbell closes the gap between a drain's last eligibility check
|
|
|
+ * and the idle transition, since those cannot be one atomic step. `done` resolves joiners
|
|
|
+ * with this execution's exit.
|
|
|
*/
|
|
|
type Execution<E, Reason> = {
|
|
|
readonly done: Deferred.Deferred<void, E>
|
|
|
owner?: Fiber.Fiber<void>
|
|
|
- pendingWake: boolean
|
|
|
+ request: Request
|
|
|
+ pendingWake?: Request
|
|
|
stopping: boolean
|
|
|
interruptionReason?: Reason
|
|
|
+ continuation?: {
|
|
|
+ readonly request: Request
|
|
|
+ readonly when: Effect.Effect<boolean>
|
|
|
+ signaled: boolean
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
@@ -43,7 +59,7 @@ type Execution<E, Reason> = {
|
|
|
* ```
|
|
|
*/
|
|
|
export const make = <Key, E, Reason = never>(options: {
|
|
|
- readonly drain: (key: Key, force: boolean) => Effect.Effect<void, E>
|
|
|
+ readonly drain: (key: Key, force: boolean, request: Request) => Effect.Effect<void, E>
|
|
|
/** Runs once when a process-local busy period begins, before its first drain. */
|
|
|
readonly started?: (key: Key) => Effect.Effect<void>
|
|
|
/**
|
|
|
@@ -57,21 +73,22 @@ export const make = <Key, E, Reason = never>(options: {
|
|
|
const fork = yield* FiberSet.makeRuntime<never, void, never>()
|
|
|
|
|
|
const loop = (key: Key, execution: Execution<E, Reason>, force: boolean): Effect.Effect<void, E> =>
|
|
|
- Effect.suspend(() => options.drain(key, force)).pipe(
|
|
|
+ Effect.suspend(() => options.drain(key, force, execution.request)).pipe(
|
|
|
Effect.flatMap(() =>
|
|
|
Effect.suspend(() => {
|
|
|
- if (execution.stopping || !execution.pendingWake) return Effect.void
|
|
|
- execution.pendingWake = false
|
|
|
+ if (execution.stopping || execution.pendingWake === undefined) return Effect.void
|
|
|
+ execution.request = execution.pendingWake
|
|
|
+ execution.pendingWake = undefined
|
|
|
// Trampoline so drains that complete synchronously cannot grow the stack.
|
|
|
return Effect.yieldNow.pipe(Effect.andThen(loop(key, execution, false)))
|
|
|
}),
|
|
|
),
|
|
|
)
|
|
|
|
|
|
- const start = (key: Key, force: boolean) => {
|
|
|
+ const start = (key: Key, force: boolean, request: Request) => {
|
|
|
const execution: Execution<E, Reason> = {
|
|
|
done: Deferred.makeUnsafe<void, E>(),
|
|
|
- pendingWake: false,
|
|
|
+ request,
|
|
|
stopping: false,
|
|
|
}
|
|
|
executions.set(key, execution)
|
|
|
@@ -87,7 +104,7 @@ export const make = <Key, E, Reason = never>(options: {
|
|
|
execution.owner = undefined
|
|
|
}).pipe(Effect.andThen(options.settled?.(key, exit, execution.interruptionReason) ?? Effect.void)),
|
|
|
),
|
|
|
- Effect.onExit((exit) => Effect.sync(() => settle(key, execution, exit))),
|
|
|
+ Effect.onExit((exit) => finish(key, execution, exit)),
|
|
|
Effect.exit,
|
|
|
Effect.asVoid,
|
|
|
),
|
|
|
@@ -97,12 +114,22 @@ export const make = <Key, E, Reason = never>(options: {
|
|
|
|
|
|
// A doorbell that survives the execution loop (rung after the loop decided to end, or
|
|
|
// during failure or interruption cleanup) starts a fresh execution for the remaining work.
|
|
|
- const settle = (key: Key, execution: Execution<E, Reason>, exit: Exit.Exit<void, E>) => {
|
|
|
- if (execution.pendingWake) start(key, false)
|
|
|
+ const settle = (key: Key, execution: Execution<E, Reason>, exit: Exit.Exit<void, E>, resume: boolean) => {
|
|
|
+ if (resume && execution.continuation) start(key, false, execution.continuation.request)
|
|
|
+ else if (execution.pendingWake) start(key, false, execution.pendingWake)
|
|
|
else executions.delete(key)
|
|
|
Deferred.doneUnsafe(execution.done, exit)
|
|
|
}
|
|
|
|
|
|
+ const finish = (key: Key, execution: Execution<E, Reason>, exit: Exit.Exit<void, E>) => {
|
|
|
+ if (!execution.continuation) return Effect.sync(() => settle(key, execution, exit, false))
|
|
|
+ return execution.continuation.when.pipe(
|
|
|
+ Effect.flatMap((ready) =>
|
|
|
+ Effect.sync(() => settle(key, execution, exit, ready || execution.continuation?.signaled === true)),
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ }
|
|
|
+
|
|
|
const run = (key: Key): Effect.Effect<void, E> =>
|
|
|
Effect.suspend(() => {
|
|
|
const execution = executions.get(key)
|
|
|
@@ -111,26 +138,58 @@ export const make = <Key, E, Reason = never>(options: {
|
|
|
if (execution.stopping) return Deferred.await(execution.done).pipe(Effect.andThen(run(key)))
|
|
|
return Deferred.await(execution.done)
|
|
|
}
|
|
|
- return Deferred.await(start(key, true).done)
|
|
|
+ return Deferred.await(start(key, true, "input").done)
|
|
|
})
|
|
|
|
|
|
- const wake = (key: Key) =>
|
|
|
+ const wake = (key: Key, request: Request = "input") =>
|
|
|
Effect.sync(() => {
|
|
|
const execution = executions.get(key)
|
|
|
if (execution !== undefined) {
|
|
|
- execution.pendingWake = true
|
|
|
+ if (execution.stopping) {
|
|
|
+ if (execution.continuation) execution.continuation.signaled = true
|
|
|
+ else execution.continuation = { request, when: Effect.succeed(true), signaled: true }
|
|
|
+ return
|
|
|
+ }
|
|
|
+ // Coalesced wakes keep the widest request: "input" subsumes "steer".
|
|
|
+ execution.pendingWake = execution.pendingWake === "input" ? "input" : request
|
|
|
return
|
|
|
}
|
|
|
- start(key, false)
|
|
|
+ start(key, false, request)
|
|
|
+ })
|
|
|
+
|
|
|
+ const wakeActive = (key: Key) =>
|
|
|
+ Effect.suspend(() => {
|
|
|
+ const execution = executions.get(key)
|
|
|
+ return execution ? wake(key, execution.request) : Effect.void
|
|
|
})
|
|
|
|
|
|
- const interrupt = (key: Key, reason?: Reason): Effect.Effect<void> =>
|
|
|
+ const interrupt = (
|
|
|
+ key: Key,
|
|
|
+ reason?: Reason,
|
|
|
+ options?: { readonly continue?: { readonly request: Request; readonly when: Effect.Effect<boolean> } },
|
|
|
+ ): Effect.Effect<void> =>
|
|
|
Effect.suspend(() => {
|
|
|
const execution = executions.get(key)
|
|
|
- if (execution?.owner === undefined || execution.stopping) return Effect.void
|
|
|
+ if (execution === undefined) return Effect.void
|
|
|
+ if (execution.stopping) {
|
|
|
+ if (options?.continue)
|
|
|
+ execution.continuation = {
|
|
|
+ ...options.continue,
|
|
|
+ signaled: execution.continuation?.signaled ?? false,
|
|
|
+ }
|
|
|
+ return Deferred.await(execution.done).pipe(Effect.exit, Effect.asVoid)
|
|
|
+ }
|
|
|
+ if (execution.owner === undefined) {
|
|
|
+ if (!options?.continue) return Effect.void
|
|
|
+ execution.stopping = true
|
|
|
+ execution.pendingWake = undefined
|
|
|
+ execution.continuation = { ...options.continue, signaled: false }
|
|
|
+ return Deferred.await(execution.done).pipe(Effect.exit, Effect.asVoid)
|
|
|
+ }
|
|
|
execution.stopping = true
|
|
|
- execution.pendingWake = false
|
|
|
+ execution.pendingWake = undefined
|
|
|
execution.interruptionReason = reason
|
|
|
+ if (options?.continue) execution.continuation = { ...options.continue, signaled: false }
|
|
|
return Fiber.interrupt(execution.owner)
|
|
|
})
|
|
|
|
|
|
@@ -143,5 +202,5 @@ export const make = <Key, E, Reason = never>(options: {
|
|
|
return Deferred.await(execution.done).pipe(Effect.exit, Effect.andThen(awaitIdle(key)))
|
|
|
})
|
|
|
|
|
|
- return { active: Effect.sync(() => new Set(executions.keys())), run, wake, interrupt, awaitIdle }
|
|
|
+ return { active: Effect.sync(() => new Set(executions.keys())), run, wake, wakeActive, interrupt, awaitIdle }
|
|
|
})
|
Kit Langton