mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-16 01:19:19 -04:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 6cb2f00d15 |
@@ -0,0 +1,5 @@
|
|||||||
|
---
|
||||||
|
"@opencode-ai/core": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
Simplify interrupt continuation: the steer-scoped resume decision now lives in SessionExecution as a post-cleanup inbox check, and the run coordinator drops its continuation state machine. Wakes arriving during cancellation cleanup now restart a normal full drain, and interrupting an idle session with continue now resumes pending steering input.
|
||||||
@@ -137,13 +137,13 @@ export const layer = Layer.effect(
|
|||||||
return Service.of({
|
return Service.of({
|
||||||
active: coordinator.active,
|
active: coordinator.active,
|
||||||
interrupt: (sessionID, options) =>
|
interrupt: (sessionID, options) =>
|
||||||
coordinator.interrupt(
|
Effect.gen(function* () {
|
||||||
sessionID,
|
yield* coordinator.interrupt(sessionID, "user")
|
||||||
"user",
|
if (!options?.continue) return
|
||||||
options?.continue
|
// Resume only steering input from the interrupted intent. Queued next-turn work
|
||||||
? { continue: { request: "steer", when: SessionInbox.has(db, sessionID, "steer") } }
|
// stays parked: a steer-scoped drain never promotes queue-delivery rows.
|
||||||
: undefined,
|
if (yield* SessionInbox.has(db, sessionID, "steer")) yield* coordinator.wake(sessionID, "steer")
|
||||||
),
|
}),
|
||||||
resume: coordinator.run,
|
resume: coordinator.run,
|
||||||
wake: coordinator.wake,
|
wake: coordinator.wake,
|
||||||
wakeActive: coordinator.wakeActive,
|
wakeActive: coordinator.wakeActive,
|
||||||
|
|||||||
@@ -10,25 +10,19 @@ export interface Coordinator<Key, E, Reason = never> {
|
|||||||
/** Starts an execution while idle, or joins the active execution and returns its exit. */
|
/** Starts an execution while idle, or joins the active execution and returns its exit. */
|
||||||
readonly run: (key: Key) => Effect.Effect<void, E>
|
readonly run: (key: Key) => Effect.Effect<void, E>
|
||||||
/** Rings the doorbell: an idle key starts an execution; an active one drains again before settling. */
|
/** Rings the doorbell: an idle key starts an execution; an active one drains again before settling. */
|
||||||
readonly wake: (key: Key, request?: Request) => Effect.Effect<void>
|
readonly wake: (key: Key, scope?: Promotable) => Effect.Effect<void>
|
||||||
/** Rings the current execution's doorbell with its existing request. Idle keys remain idle. */
|
/** Rings the current execution's doorbell with its existing scope. Idle keys remain idle. */
|
||||||
readonly wakeActive: (key: Key) => Effect.Effect<void>
|
readonly wakeActive: (key: Key) => Effect.Effect<void>
|
||||||
/** Stops the active execution, clears its doorbell, and waits for cleanup. No-op when idle. */
|
/** Stops the active execution, clears its doorbell, and waits for cleanup. No-op when idle. */
|
||||||
readonly interrupt: (
|
readonly interrupt: (key: Key, reason?: Reason) => Effect.Effect<void>
|
||||||
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. */
|
/** 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>
|
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
|
* 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
|
* until the key would stay idle. `pendingWake` is the doorbell: work recorded during the
|
||||||
* execution rings it with its eligibility request, and the execution loop drains again
|
* execution rings it with the scope that work needs, and the execution loop drains again
|
||||||
* instead of ending. The doorbell closes the gap between a drain's last eligibility check
|
* 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
|
* and the idle transition, since those cannot be one atomic step. `done` resolves joiners
|
||||||
* with this execution's exit.
|
* with this execution's exit.
|
||||||
@@ -36,15 +30,10 @@ export type Request = Promotable
|
|||||||
type Execution<E, Reason> = {
|
type Execution<E, Reason> = {
|
||||||
readonly done: Deferred.Deferred<void, E>
|
readonly done: Deferred.Deferred<void, E>
|
||||||
owner?: Fiber.Fiber<void>
|
owner?: Fiber.Fiber<void>
|
||||||
request: Request
|
scope: Promotable
|
||||||
pendingWake?: Request
|
pendingWake?: Promotable
|
||||||
stopping: boolean
|
stopping: boolean
|
||||||
interruptionReason?: Reason
|
interruptionReason?: Reason
|
||||||
continuation?: {
|
|
||||||
readonly request: Request
|
|
||||||
readonly when: Effect.Effect<boolean>
|
|
||||||
signaled: boolean
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -59,7 +48,7 @@ type Execution<E, Reason> = {
|
|||||||
* ```
|
* ```
|
||||||
*/
|
*/
|
||||||
export const make = <Key, E, Reason = never>(options: {
|
export const make = <Key, E, Reason = never>(options: {
|
||||||
readonly drain: (key: Key, force: boolean, request: Request) => Effect.Effect<void, E>
|
readonly drain: (key: Key, force: boolean, scope: Promotable) => Effect.Effect<void, E>
|
||||||
/** Runs once when a process-local busy period begins, before its first drain. */
|
/** Runs once when a process-local busy period begins, before its first drain. */
|
||||||
readonly started?: (key: Key) => Effect.Effect<void>
|
readonly started?: (key: Key) => Effect.Effect<void>
|
||||||
/**
|
/**
|
||||||
@@ -73,11 +62,11 @@ export const make = <Key, E, Reason = never>(options: {
|
|||||||
const fork = yield* FiberSet.makeRuntime<never, void, never>()
|
const fork = yield* FiberSet.makeRuntime<never, void, never>()
|
||||||
|
|
||||||
const loop = (key: Key, execution: Execution<E, Reason>, force: boolean): Effect.Effect<void, E> =>
|
const loop = (key: Key, execution: Execution<E, Reason>, force: boolean): Effect.Effect<void, E> =>
|
||||||
Effect.suspend(() => options.drain(key, force, execution.request)).pipe(
|
Effect.suspend(() => options.drain(key, force, execution.scope)).pipe(
|
||||||
Effect.flatMap(() =>
|
Effect.flatMap(() =>
|
||||||
Effect.suspend(() => {
|
Effect.suspend(() => {
|
||||||
if (execution.stopping || execution.pendingWake === undefined) return Effect.void
|
if (execution.stopping || execution.pendingWake === undefined) return Effect.void
|
||||||
execution.request = execution.pendingWake
|
execution.scope = execution.pendingWake
|
||||||
execution.pendingWake = undefined
|
execution.pendingWake = undefined
|
||||||
// Trampoline so drains that complete synchronously cannot grow the stack.
|
// Trampoline so drains that complete synchronously cannot grow the stack.
|
||||||
return Effect.yieldNow.pipe(Effect.andThen(loop(key, execution, false)))
|
return Effect.yieldNow.pipe(Effect.andThen(loop(key, execution, false)))
|
||||||
@@ -85,10 +74,10 @@ export const make = <Key, E, Reason = never>(options: {
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
const start = (key: Key, force: boolean, request: Request) => {
|
const start = (key: Key, force: boolean, scope: Promotable) => {
|
||||||
const execution: Execution<E, Reason> = {
|
const execution: Execution<E, Reason> = {
|
||||||
done: Deferred.makeUnsafe<void, E>(),
|
done: Deferred.makeUnsafe<void, E>(),
|
||||||
request,
|
scope,
|
||||||
stopping: false,
|
stopping: false,
|
||||||
}
|
}
|
||||||
executions.set(key, execution)
|
executions.set(key, execution)
|
||||||
@@ -104,7 +93,7 @@ export const make = <Key, E, Reason = never>(options: {
|
|||||||
execution.owner = undefined
|
execution.owner = undefined
|
||||||
}).pipe(Effect.andThen(options.settled?.(key, exit, execution.interruptionReason) ?? Effect.void)),
|
}).pipe(Effect.andThen(options.settled?.(key, exit, execution.interruptionReason) ?? Effect.void)),
|
||||||
),
|
),
|
||||||
Effect.onExit((exit) => finish(key, execution, exit)),
|
Effect.onExit((exit) => Effect.sync(() => settle(key, execution, exit))),
|
||||||
Effect.exit,
|
Effect.exit,
|
||||||
Effect.asVoid,
|
Effect.asVoid,
|
||||||
),
|
),
|
||||||
@@ -114,22 +103,12 @@ export const make = <Key, E, Reason = never>(options: {
|
|||||||
|
|
||||||
// A doorbell that survives the execution loop (rung after the loop decided to end, or
|
// 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.
|
// 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>, resume: boolean) => {
|
const settle = (key: Key, execution: Execution<E, Reason>, exit: Exit.Exit<void, E>) => {
|
||||||
if (resume && execution.continuation) start(key, false, execution.continuation.request)
|
if (execution.pendingWake) start(key, false, execution.pendingWake)
|
||||||
else if (execution.pendingWake) start(key, false, execution.pendingWake)
|
|
||||||
else executions.delete(key)
|
else executions.delete(key)
|
||||||
Deferred.doneUnsafe(execution.done, exit)
|
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> =>
|
const run = (key: Key): Effect.Effect<void, E> =>
|
||||||
Effect.suspend(() => {
|
Effect.suspend(() => {
|
||||||
const execution = executions.get(key)
|
const execution = executions.get(key)
|
||||||
@@ -141,55 +120,32 @@ export const make = <Key, E, Reason = never>(options: {
|
|||||||
return Deferred.await(start(key, true, "input").done)
|
return Deferred.await(start(key, true, "input").done)
|
||||||
})
|
})
|
||||||
|
|
||||||
const wake = (key: Key, request: Request = "input") =>
|
const wake = (key: Key, scope: Promotable = "input") =>
|
||||||
Effect.sync(() => {
|
Effect.sync(() => {
|
||||||
const execution = executions.get(key)
|
const execution = executions.get(key)
|
||||||
if (execution !== undefined) {
|
if (execution !== undefined) {
|
||||||
if (execution.stopping) {
|
// Coalesced wakes keep the widest scope: "input" subsumes "steer".
|
||||||
if (execution.continuation) execution.continuation.signaled = true
|
execution.pendingWake = execution.pendingWake === "input" ? "input" : scope
|
||||||
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
|
return
|
||||||
}
|
}
|
||||||
start(key, false, request)
|
start(key, false, scope)
|
||||||
})
|
})
|
||||||
|
|
||||||
const wakeActive = (key: Key) =>
|
const wakeActive = (key: Key) =>
|
||||||
Effect.suspend(() => {
|
Effect.suspend(() => {
|
||||||
const execution = executions.get(key)
|
const execution = executions.get(key)
|
||||||
return execution ? wake(key, execution.request) : Effect.void
|
return execution ? wake(key, execution.scope) : Effect.void
|
||||||
})
|
})
|
||||||
|
|
||||||
const interrupt = (
|
const interrupt = (key: Key, reason?: Reason): Effect.Effect<void> =>
|
||||||
key: Key,
|
|
||||||
reason?: Reason,
|
|
||||||
options?: { readonly continue?: { readonly request: Request; readonly when: Effect.Effect<boolean> } },
|
|
||||||
): Effect.Effect<void> =>
|
|
||||||
Effect.suspend(() => {
|
Effect.suspend(() => {
|
||||||
const execution = executions.get(key)
|
const execution = executions.get(key)
|
||||||
if (execution === undefined) return Effect.void
|
if (execution?.owner === undefined || execution.stopping) 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.stopping = true
|
||||||
|
// Wakes recorded so far belong to the interrupted intent; the interrupt claims them.
|
||||||
|
// Wakes arriving during cleanup are new admissions and restart normally at settle.
|
||||||
execution.pendingWake = undefined
|
execution.pendingWake = undefined
|
||||||
execution.interruptionReason = reason
|
execution.interruptionReason = reason
|
||||||
if (options?.continue) execution.continuation = { ...options.continue, signaled: false }
|
|
||||||
return Fiber.interrupt(execution.owner)
|
return Fiber.interrupt(execution.owner)
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|||||||
@@ -14,8 +14,10 @@ import { SessionExecution } from "@opencode-ai/core/session/execution"
|
|||||||
import { SessionRestart } from "@opencode-ai/core/session/execution/restart"
|
import { SessionRestart } from "@opencode-ai/core/session/execution/restart"
|
||||||
import { UserInterruptedError } from "@opencode-ai/core/session/error"
|
import { UserInterruptedError } from "@opencode-ai/core/session/error"
|
||||||
import { SessionEvent } from "@opencode-ai/core/session/event"
|
import { SessionEvent } from "@opencode-ai/core/session/event"
|
||||||
|
import { SessionInbox } from "@opencode-ai/core/session/inbox"
|
||||||
|
import { SessionMessage } from "@opencode-ai/core/session/message"
|
||||||
import { SessionRunner } from "@opencode-ai/core/session/runner/index"
|
import { SessionRunner } from "@opencode-ai/core/session/runner/index"
|
||||||
import { SessionTable } from "@opencode-ai/core/session/sql"
|
import { SessionInboxTable, SessionTable } from "@opencode-ai/core/session/sql"
|
||||||
import { SessionStore } from "@opencode-ai/core/session/store"
|
import { SessionStore } from "@opencode-ai/core/session/store"
|
||||||
import { Context, Deferred, Effect, Exit, Fiber, Layer, LayerMap, Scope } from "effect"
|
import { Context, Deferred, Effect, Exit, Fiber, Layer, LayerMap, Scope } from "effect"
|
||||||
import { eq } from "drizzle-orm"
|
import { eq } from "drizzle-orm"
|
||||||
@@ -290,6 +292,113 @@ describe("SessionExecution lifecycle", () => {
|
|||||||
)
|
)
|
||||||
})
|
})
|
||||||
|
|
||||||
|
describe("SessionExecution interrupt continuation", () => {
|
||||||
|
it.effect("resumes only steering input after an interrupt with continue", () =>
|
||||||
|
Effect.gen(function* () {
|
||||||
|
const database = yield* Database.Service
|
||||||
|
const sessionID = Session.ID.make("ses_continue_steer")
|
||||||
|
yield* seedSessions(database, [sessionID])
|
||||||
|
yield* seedInbox(database, sessionID, ["steer", "queue"])
|
||||||
|
|
||||||
|
const draining = yield* Deferred.make<void>()
|
||||||
|
const drains: Array<{ force: boolean; promotable?: SessionInbox.Promotable }> = []
|
||||||
|
const scope = yield* Scope.make()
|
||||||
|
yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
|
||||||
|
const context = yield* buildExecution(scope, (input) =>
|
||||||
|
Effect.suspend(() => {
|
||||||
|
drains.push({ force: input.force, promotable: input.promotable })
|
||||||
|
if (drains.length > 1) return Effect.void
|
||||||
|
return Deferred.succeed(draining, undefined).pipe(Effect.andThen(Effect.never))
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
const execution = Context.get(context, SessionExecution.Service)
|
||||||
|
yield* execution.resume(sessionID).pipe(Effect.forkScoped)
|
||||||
|
yield* Deferred.await(draining)
|
||||||
|
|
||||||
|
yield* execution.interrupt(sessionID, { continue: true })
|
||||||
|
yield* execution.awaitIdle(sessionID)
|
||||||
|
|
||||||
|
// The successor drain is steer-scoped: queued next-turn work stays parked.
|
||||||
|
expect(drains).toEqual([
|
||||||
|
{ force: true, promotable: "input" },
|
||||||
|
{ force: false, promotable: "steer" },
|
||||||
|
])
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
|
||||||
|
it.effect("stays parked after an interrupt with continue when only queued work remains", () =>
|
||||||
|
Effect.gen(function* () {
|
||||||
|
const database = yield* Database.Service
|
||||||
|
const sessionID = Session.ID.make("ses_continue_parked")
|
||||||
|
yield* seedSessions(database, [sessionID])
|
||||||
|
yield* seedInbox(database, sessionID, ["queue"])
|
||||||
|
|
||||||
|
const draining = yield* Deferred.make<void>()
|
||||||
|
const drains: Array<SessionInbox.Promotable | undefined> = []
|
||||||
|
const scope = yield* Scope.make()
|
||||||
|
yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
|
||||||
|
const context = yield* buildExecution(scope, (input) =>
|
||||||
|
Effect.suspend(() => {
|
||||||
|
drains.push(input.promotable)
|
||||||
|
return Deferred.succeed(draining, undefined).pipe(Effect.andThen(Effect.never))
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
const execution = Context.get(context, SessionExecution.Service)
|
||||||
|
yield* execution.resume(sessionID).pipe(Effect.forkScoped)
|
||||||
|
yield* Deferred.await(draining)
|
||||||
|
|
||||||
|
yield* execution.interrupt(sessionID, { continue: true })
|
||||||
|
yield* execution.awaitIdle(sessionID)
|
||||||
|
|
||||||
|
expect(drains).toEqual(["input"])
|
||||||
|
expect(yield* execution.active).toEqual(new Set())
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
|
||||||
|
it.effect("an idle interrupt with continue resumes pending steers", () =>
|
||||||
|
Effect.gen(function* () {
|
||||||
|
const database = yield* Database.Service
|
||||||
|
const sessionID = Session.ID.make("ses_continue_idle")
|
||||||
|
yield* seedSessions(database, [sessionID])
|
||||||
|
yield* seedInbox(database, sessionID, ["steer"])
|
||||||
|
|
||||||
|
const drains: Array<{ force: boolean; promotable?: SessionInbox.Promotable }> = []
|
||||||
|
const scope = yield* Scope.make()
|
||||||
|
yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
|
||||||
|
const context = yield* buildExecution(scope, (input) =>
|
||||||
|
Effect.sync(() => void drains.push({ force: input.force, promotable: input.promotable })),
|
||||||
|
)
|
||||||
|
const execution = Context.get(context, SessionExecution.Service)
|
||||||
|
|
||||||
|
yield* execution.interrupt(sessionID, { continue: true })
|
||||||
|
yield* execution.awaitIdle(sessionID)
|
||||||
|
|
||||||
|
expect(drains).toEqual([{ force: false, promotable: "steer" }])
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
})
|
||||||
|
|
||||||
|
function seedInbox(
|
||||||
|
database: Database.Service["Service"],
|
||||||
|
sessionID: Session.ID,
|
||||||
|
deliveries: ReadonlyArray<SessionInbox.Delivery>,
|
||||||
|
) {
|
||||||
|
return database.db
|
||||||
|
.insert(SessionInboxTable)
|
||||||
|
.values(
|
||||||
|
deliveries.map((delivery, index) => ({
|
||||||
|
id: SessionMessage.ID.create(),
|
||||||
|
session_id: sessionID,
|
||||||
|
type: "compaction" as const,
|
||||||
|
payload: {},
|
||||||
|
delivery,
|
||||||
|
enqueued_seq: index + 1,
|
||||||
|
})),
|
||||||
|
)
|
||||||
|
.run()
|
||||||
|
.pipe(Effect.orDie)
|
||||||
|
}
|
||||||
|
|
||||||
function seedSessions(
|
function seedSessions(
|
||||||
database: Database.Service["Service"],
|
database: Database.Service["Service"],
|
||||||
sessionIDs: ReadonlyArray<Session.ID>,
|
sessionIDs: ReadonlyArray<Session.ID>,
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
import { describe, expect } from "bun:test"
|
import { describe, expect } from "bun:test"
|
||||||
import { Cause, Deferred, Effect, Exit, Fiber, Layer } from "effect"
|
import { Cause, Deferred, Effect, Exit, Fiber, Layer } from "effect"
|
||||||
|
import { SessionInbox } from "@opencode-ai/core/session/inbox"
|
||||||
import { SessionRunCoordinator } from "@opencode-ai/core/session/run-coordinator"
|
import { SessionRunCoordinator } from "@opencode-ai/core/session/run-coordinator"
|
||||||
import { testEffect } from "./lib/effect"
|
import { testEffect } from "./lib/effect"
|
||||||
|
|
||||||
@@ -269,31 +270,24 @@ describe("SessionRunCoordinator", () => {
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
it.effect("replaces a settlement-window wake with a steer continuation", () =>
|
it.effect("a settlement-window wake starts a fresh execution with its own scope", () =>
|
||||||
Effect.scoped(
|
Effect.scoped(
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
const settling = yield* Deferred.make<void>()
|
const settling = yield* Deferred.make<void>()
|
||||||
const release = yield* Deferred.make<void>()
|
const release = yield* Deferred.make<void>()
|
||||||
const requests: SessionRunCoordinator.Request[] = []
|
const scopes: SessionInbox.Promotable[] = []
|
||||||
const coordinator = yield* SessionRunCoordinator.make({
|
const coordinator = yield* SessionRunCoordinator.make({
|
||||||
drain: (_key, _force, request) => Effect.sync(() => requests.push(request)),
|
drain: (_key, _force, scope) => Effect.sync(() => scopes.push(scope)),
|
||||||
settled: () => Deferred.succeed(settling, undefined).pipe(Effect.andThen(Deferred.await(release))),
|
settled: () => Deferred.succeed(settling, undefined).pipe(Effect.andThen(Deferred.await(release))),
|
||||||
})
|
})
|
||||||
|
|
||||||
yield* coordinator.wake("session", "input")
|
yield* coordinator.wake("session", "steer")
|
||||||
yield* Deferred.await(settling)
|
yield* Deferred.await(settling)
|
||||||
yield* coordinator.wake("session", "input")
|
yield* coordinator.wake("session", "input")
|
||||||
const interrupted = yield* coordinator
|
|
||||||
.interrupt("session", undefined, {
|
|
||||||
continue: { request: "steer", when: Effect.succeed(true) },
|
|
||||||
})
|
|
||||||
.pipe(Effect.forkChild)
|
|
||||||
yield* Effect.yieldNow
|
|
||||||
yield* Deferred.succeed(release, undefined)
|
yield* Deferred.succeed(release, undefined)
|
||||||
yield* Fiber.join(interrupted)
|
|
||||||
yield* coordinator.awaitIdle("session")
|
yield* coordinator.awaitIdle("session")
|
||||||
|
|
||||||
expect(requests).toEqual(["input", "steer"])
|
expect(scopes).toEqual(["steer", "input"])
|
||||||
}),
|
}),
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
@@ -371,17 +365,17 @@ describe("SessionRunCoordinator", () => {
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
it.effect("coalesces drain requests with input taking precedence", () =>
|
it.effect("coalesces drain scopes with input taking precedence", () =>
|
||||||
Effect.scoped(
|
Effect.scoped(
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
const firstStarted = yield* Deferred.make<void>()
|
const firstStarted = yield* Deferred.make<void>()
|
||||||
const release = yield* Deferred.make<void>()
|
const release = yield* Deferred.make<void>()
|
||||||
const requests: SessionRunCoordinator.Request[] = []
|
const scopes: SessionInbox.Promotable[] = []
|
||||||
const coordinator = yield* SessionRunCoordinator.make({
|
const coordinator = yield* SessionRunCoordinator.make({
|
||||||
drain: (_key, _force, request) =>
|
drain: (_key, _force, scope) =>
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
requests.push(request)
|
scopes.push(scope)
|
||||||
if (requests.length !== 1) return
|
if (scopes.length !== 1) return
|
||||||
yield* Deferred.succeed(firstStarted, undefined)
|
yield* Deferred.succeed(firstStarted, undefined)
|
||||||
yield* Deferred.await(release)
|
yield* Deferred.await(release)
|
||||||
}),
|
}),
|
||||||
@@ -394,22 +388,22 @@ describe("SessionRunCoordinator", () => {
|
|||||||
yield* Deferred.succeed(release, undefined)
|
yield* Deferred.succeed(release, undefined)
|
||||||
yield* coordinator.awaitIdle("session")
|
yield* coordinator.awaitIdle("session")
|
||||||
|
|
||||||
expect(requests).toEqual(["steer", "input"])
|
expect(scopes).toEqual(["steer", "input"])
|
||||||
}),
|
}),
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
it.effect("does not carry a completed input request into a steer drain", () =>
|
it.effect("does not carry a completed input scope into a steer drain", () =>
|
||||||
Effect.scoped(
|
Effect.scoped(
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
const firstStarted = yield* Deferred.make<void>()
|
const firstStarted = yield* Deferred.make<void>()
|
||||||
const release = yield* Deferred.make<void>()
|
const release = yield* Deferred.make<void>()
|
||||||
const requests: SessionRunCoordinator.Request[] = []
|
const scopes: SessionInbox.Promotable[] = []
|
||||||
const coordinator = yield* SessionRunCoordinator.make({
|
const coordinator = yield* SessionRunCoordinator.make({
|
||||||
drain: (_key, _force, request) =>
|
drain: (_key, _force, scope) =>
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
requests.push(request)
|
scopes.push(scope)
|
||||||
if (requests.length !== 1) return
|
if (scopes.length !== 1) return
|
||||||
yield* Deferred.succeed(firstStarted, undefined)
|
yield* Deferred.succeed(firstStarted, undefined)
|
||||||
yield* Deferred.await(release)
|
yield* Deferred.await(release)
|
||||||
}),
|
}),
|
||||||
@@ -421,7 +415,7 @@ describe("SessionRunCoordinator", () => {
|
|||||||
yield* Deferred.succeed(release, undefined)
|
yield* Deferred.succeed(release, undefined)
|
||||||
yield* coordinator.awaitIdle("session")
|
yield* coordinator.awaitIdle("session")
|
||||||
|
|
||||||
expect(requests).toEqual(["input", "steer"])
|
expect(scopes).toEqual(["input", "steer"])
|
||||||
}),
|
}),
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
@@ -431,12 +425,12 @@ describe("SessionRunCoordinator", () => {
|
|||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
const firstStarted = yield* Deferred.make<void>()
|
const firstStarted = yield* Deferred.make<void>()
|
||||||
const release = yield* Deferred.make<void>()
|
const release = yield* Deferred.make<void>()
|
||||||
const requests: SessionRunCoordinator.Request[] = []
|
const scopes: SessionInbox.Promotable[] = []
|
||||||
const coordinator = yield* SessionRunCoordinator.make({
|
const coordinator = yield* SessionRunCoordinator.make({
|
||||||
drain: (_key, _force, request) =>
|
drain: (_key, _force, scope) =>
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
requests.push(request)
|
scopes.push(scope)
|
||||||
if (requests.length !== 1) return
|
if (scopes.length !== 1) return
|
||||||
yield* Deferred.succeed(firstStarted, undefined)
|
yield* Deferred.succeed(firstStarted, undefined)
|
||||||
yield* Deferred.await(release)
|
yield* Deferred.await(release)
|
||||||
}),
|
}),
|
||||||
@@ -449,61 +443,23 @@ describe("SessionRunCoordinator", () => {
|
|||||||
yield* Deferred.succeed(release, undefined)
|
yield* Deferred.succeed(release, undefined)
|
||||||
yield* coordinator.awaitIdle("session")
|
yield* coordinator.awaitIdle("session")
|
||||||
|
|
||||||
expect(requests).toEqual(["steer", "steer"])
|
expect(scopes).toEqual(["steer", "steer"])
|
||||||
}),
|
}),
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
it.effect("coalesces overlapping interrupt continuations into one steer successor", () =>
|
it.effect("a cleanup-era wake starts a successor with its own scope", () =>
|
||||||
Effect.scoped(
|
Effect.scoped(
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
const firstStarted = yield* Deferred.make<void>()
|
const firstStarted = yield* Deferred.make<void>()
|
||||||
const cleanupStarted = yield* Deferred.make<void>()
|
const cleanupStarted = yield* Deferred.make<void>()
|
||||||
const cleanupGate = yield* Deferred.make<void>()
|
const cleanupGate = yield* Deferred.make<void>()
|
||||||
const requests: SessionRunCoordinator.Request[] = []
|
const scopes: SessionInbox.Promotable[] = []
|
||||||
const coordinator = yield* SessionRunCoordinator.make({
|
const coordinator = yield* SessionRunCoordinator.make({
|
||||||
drain: (_key, _force, request) =>
|
drain: (_key, _force, scope) =>
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
requests.push(request)
|
scopes.push(scope)
|
||||||
if (requests.length !== 1) return
|
if (scopes.length !== 1) return
|
||||||
yield* Deferred.succeed(firstStarted, undefined)
|
|
||||||
yield* Effect.never.pipe(
|
|
||||||
Effect.onInterrupt(() =>
|
|
||||||
Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
}),
|
|
||||||
})
|
|
||||||
const continuation = { continue: { request: "steer" as const, when: Effect.succeed(false) } }
|
|
||||||
|
|
||||||
yield* coordinator.wake("session")
|
|
||||||
yield* Deferred.await(firstStarted)
|
|
||||||
const first = yield* coordinator.interrupt("session", undefined, continuation).pipe(Effect.forkChild)
|
|
||||||
yield* Deferred.await(cleanupStarted)
|
|
||||||
const second = yield* coordinator.interrupt("session", undefined, continuation).pipe(Effect.forkChild)
|
|
||||||
yield* Effect.yieldNow
|
|
||||||
yield* coordinator.wake("session", "input")
|
|
||||||
yield* Deferred.succeed(cleanupGate, undefined)
|
|
||||||
yield* Effect.all([Fiber.join(first), Fiber.join(second)])
|
|
||||||
yield* coordinator.awaitIdle("session")
|
|
||||||
|
|
||||||
expect(requests).toEqual(["input", "steer"])
|
|
||||||
}),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
|
|
||||||
it.effect("a continuing interrupt replaces a cleanup-era input wake", () =>
|
|
||||||
Effect.scoped(
|
|
||||||
Effect.gen(function* () {
|
|
||||||
const firstStarted = yield* Deferred.make<void>()
|
|
||||||
const cleanupStarted = yield* Deferred.make<void>()
|
|
||||||
const cleanupGate = yield* Deferred.make<void>()
|
|
||||||
const requests: SessionRunCoordinator.Request[] = []
|
|
||||||
const coordinator = yield* SessionRunCoordinator.make({
|
|
||||||
drain: (_key, _force, request) =>
|
|
||||||
Effect.gen(function* () {
|
|
||||||
requests.push(request)
|
|
||||||
if (requests.length !== 1) return
|
|
||||||
yield* Deferred.succeed(firstStarted, undefined)
|
yield* Deferred.succeed(firstStarted, undefined)
|
||||||
yield* Effect.never.pipe(
|
yield* Effect.never.pipe(
|
||||||
Effect.onInterrupt(() =>
|
Effect.onInterrupt(() =>
|
||||||
@@ -515,45 +471,16 @@ describe("SessionRunCoordinator", () => {
|
|||||||
|
|
||||||
yield* coordinator.wake("session", "input")
|
yield* coordinator.wake("session", "input")
|
||||||
yield* Deferred.await(firstStarted)
|
yield* Deferred.await(firstStarted)
|
||||||
const plain = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
|
const interrupt = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
|
||||||
yield* Deferred.await(cleanupStarted)
|
yield* Deferred.await(cleanupStarted)
|
||||||
|
// A new admission during cancellation restarts normally: interruption only
|
||||||
|
// claims the wakes recorded before it.
|
||||||
yield* coordinator.wake("session", "input")
|
yield* coordinator.wake("session", "input")
|
||||||
const continuing = yield* coordinator
|
|
||||||
.interrupt("session", undefined, {
|
|
||||||
continue: { request: "steer", when: Effect.succeed(false) },
|
|
||||||
})
|
|
||||||
.pipe(Effect.forkChild)
|
|
||||||
yield* Effect.yieldNow
|
|
||||||
yield* Deferred.succeed(cleanupGate, undefined)
|
yield* Deferred.succeed(cleanupGate, undefined)
|
||||||
yield* Effect.all([Fiber.join(plain), Fiber.join(continuing)])
|
yield* Fiber.join(interrupt)
|
||||||
yield* coordinator.awaitIdle("session")
|
yield* coordinator.awaitIdle("session")
|
||||||
|
|
||||||
expect(requests).toEqual(["input", "steer"])
|
expect(scopes).toEqual(["input", "input"])
|
||||||
}),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
|
|
||||||
it.effect("does not start a conditional continuation without eligible work", () =>
|
|
||||||
Effect.scoped(
|
|
||||||
Effect.gen(function* () {
|
|
||||||
const started = yield* Deferred.make<void>()
|
|
||||||
const requests: SessionRunCoordinator.Request[] = []
|
|
||||||
const coordinator = yield* SessionRunCoordinator.make({
|
|
||||||
drain: (_key, _force, request) =>
|
|
||||||
Effect.sync(() => requests.push(request)).pipe(
|
|
||||||
Effect.andThen(Deferred.succeed(started, undefined)),
|
|
||||||
Effect.andThen(Effect.never),
|
|
||||||
),
|
|
||||||
})
|
|
||||||
|
|
||||||
yield* coordinator.wake("session")
|
|
||||||
yield* Deferred.await(started)
|
|
||||||
yield* coordinator.interrupt("session", undefined, {
|
|
||||||
continue: { request: "steer", when: Effect.succeed(false) },
|
|
||||||
})
|
|
||||||
yield* coordinator.awaitIdle("session")
|
|
||||||
|
|
||||||
expect(requests).toEqual(["input"])
|
|
||||||
}),
|
}),
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user