vrr/opencode
2
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-07-26 17:45:36 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(core): start all suspended sessions promptly (#38720)

Browse source
This commit is contained in:
Kit Langton 2026-07-24 12:58:50 -04:00 • committed by GitHub
parent 7ab3dd04ad
commit 68ef893818
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 28 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

3
packages/core/src/session/execution/restart.ts
View file

@ -40,8 +40,7 @@ export const layer = Layer.effect(
// Drain failures are already logged and durably recorded by the execution layer.
yield* Effect.ignore(execution.resume(sessionID))
}),
// Each suspension is consumed atomically right before its drain; at most four drains run at once.
{ concurrency: 4, discard: true },
{ concurrency: "unbounded", discard: true },
)
}),
})

27
packages/core/test/session-execution.test.ts
View file

@ -105,6 +105,31 @@ describe("SessionExecution lifecycle", () => {
}),
)
it.effect("starts every suspended execution without waiting for earlier drains to finish", () =>
Effect.gen(function* () {
const database = yield* Database.Service
const sessionIDs = Array.from({ length: 5 }, (_, index) => SessionV2.ID.make(`ses_resume_concurrent_${index}`))
yield* seedSessions(database, sessionIDs, { time_suspended: Date.now() })
const fourStarted = yield* Deferred.make<void>()
const started: SessionV2.ID[] = []
const scope = yield* Scope.make()
yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
const context = yield* buildExecution(scope, ({ sessionID }) =>
Effect.sync(() => {
started.push(sessionID)
if (started.length === 4) Deferred.doneUnsafe(fourStarted, Effect.void)
}).pipe(Effect.andThen(Effect.never)),
)
const execution = Context.get(context, SessionExecution.Service)
const restart = Context.get(context, SessionRestart.Service)
yield* restart.resumeSuspendedSessions.pipe(Effect.forkIn(scope))
yield* Deferred.await(fourStarted)
expect([...(yield* execution.active)].toSorted()).toEqual(sessionIDs.toSorted())
}),
)
it.effect("resumes each suspended Session at most once", () =>
Effect.gen(function* () {
const database = yield* Database.Service
@ -115,9 +140,11 @@ describe("SessionExecution lifecycle", () => {
const drained: string[] = []
const scope = yield* Scope.make()
const context = yield* buildExecution(scope, ({ sessionID }) => Effect.sync(() => void drained.push(sessionID)))
const execution = Context.get(context, SessionExecution.Service)
const restart = Context.get(context, SessionRestart.Service)
yield* restart.resumeSuspendedSessions
yield* Effect.forEach([first, second], execution.awaitIdle, { discard: true })
expect(drained.toSorted()).toEqual([first, second])
expect(yield* suspensions(database)).toEqual({ [first]: false, [second]: false })