Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 19 additions & 0 deletions packages/core/src/background-job.ts
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,25 @@ export const make = Effect.gen(function* () {
scope: yield* Scope.Scope,
}

yield* Effect.addFinalizer(() =>
SynchronizedRef.get(state.jobs).pipe(
Effect.flatMap((jobs) =>
Effect.forEach(
[...jobs.values()],
(job) => {
if (job.info.status !== "running") return Effect.void
return Deferred.succeed(job.done, {
...job.info,
status: "cancelled" as const,
completed_at: Date.now(),
}).pipe(Effect.ignore)
},
{ discard: true },
),
),
),
)

const settle = Effect.fn("BackgroundJob.settle")(function* (
id: string,
token: object,
Expand Down
16 changes: 15 additions & 1 deletion packages/opencode/src/effect/instance-registry.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,20 @@ export function registerDisposer(disposer: (directory: string) => Promise<void>)
}
}

const DISPOSE_TIMEOUT_MS = 10_000

export async function disposeInstance(directory: string) {
await Promise.allSettled([...disposers].map((disposer) => disposer(directory)))
await Promise.allSettled(
[...disposers].map((disposer) =>
Promise.race([
disposer(directory),
new Promise<void>((_, reject) =>
setTimeout(
() => reject(new Error(`Disposer timed out after ${DISPOSE_TIMEOUT_MS}ms for ${directory}`)),
DISPOSE_TIMEOUT_MS,
),
),
]),
),
)
}
6 changes: 5 additions & 1 deletion packages/opencode/src/tool/task.ts
Original file line number Diff line number Diff line change
Expand Up @@ -347,7 +347,11 @@ export const TaskTool = Tool.define(
(_, exit) =>
Effect.gen(function* () {
if (Exit.hasInterrupts(exit))
yield* Effect.all([cancel, background.cancel(nextSession.id)], { discard: true })
yield* Effect.all([cancel, background.cancel(nextSession.id)], { discard: true }).pipe(
Effect.ignore,
Effect.forkDetach,
Effect.asVoid,
)
}).pipe(
Effect.ensuring(
Effect.sync(() => {
Expand Down
36 changes: 33 additions & 3 deletions packages/opencode/test/background/job.test.ts
Original file line number Diff line number Diff line change
@@ -1,8 +1,10 @@
import { describe, expect } from "bun:test"
import { afterEach, describe, expect } from "bun:test"
import { LayerNode } from "@opencode-ai/core/effect/layer-node"
import { Deferred, Effect } from "effect"
import { Deferred, Effect, Scope, Exit } from "effect"
import { BackgroundJob } from "@/background/job"
import { testEffect } from "../lib/effect"
import { BackgroundJob as CoreBackgroundJob } from "@opencode-ai/core/background-job"
import { testEffect, awaitWithTimeout } from "../lib/effect"
import { disposeAllInstances } from "../fixture/fixture"

const it = testEffect(LayerNode.compile(BackgroundJob.node))

Expand Down Expand Up @@ -241,4 +243,32 @@ describe("background.job", () => {
expect((yield* jobs.get(job.id))?.metadata?.value).toBe("initial")
}),
)

it.live("scope close resolves pending done deferreds", () =>
Effect.gen(function* () {
const scope = yield* Scope.make()
const jobs = yield* CoreBackgroundJob.make.pipe(Effect.provideService(Scope.Scope, scope))

const waiterDone = yield* Deferred.make<boolean>()

const job = yield* jobs.start({ type: "test", run: Effect.never })
expect(job.status).toBe("running")

yield* jobs
.wait({ id: job.id })
.pipe(
Effect.flatMap((result) => Deferred.succeed(waiterDone, result.info?.status === "cancelled")),
Effect.forkDetach,
)
yield* Effect.yieldNow

yield* Scope.close(scope, Exit.void)

const resolved = yield* awaitWithTimeout(
Deferred.await(waiterDone),
"waiter should resolve after scope close",
)
expect(resolved).toBe(true)
}),
)
})
113 changes: 113 additions & 0 deletions packages/opencode/test/project/instance-lifecycle.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,113 @@
import { afterEach, describe, expect } from "bun:test"
import { Effect, Exit, Layer } from "effect"
import { BackgroundJob } from "@/background/job"
import { InstanceStore } from "../../src/project/instance-store"
import { InstanceBootstrap } from "../../src/project/bootstrap-service"
import { registerDisposer } from "../../src/effect/instance-registry"
import {
tmpdirScoped,
provideInstanceEffect,
reloadInstance,
disposeAllInstances,
} from "../fixture/fixture"
import { testEffect, awaitWithTimeout } from "../lib/effect"
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"

afterEach(async () => {
await disposeAllInstances()
})

const noopBootstrap = Layer.succeed(InstanceBootstrap.Service, InstanceBootstrap.Service.of({ run: Effect.void }))

const it = testEffect(
Layer.mergeAll(InstanceStore.defaultLayer, BackgroundJob.defaultLayer, CrossSpawnSpawner.defaultLayer).pipe(
Layer.provide(noopBootstrap),
),
)

describe("instance lifecycle", () => {
it.live("disposal completes when background jobs are active", () =>
Effect.gen(function* () {
const dir = yield* tmpdirScoped({ git: true })
const store = yield* InstanceStore.Service

yield* store.load({ directory: dir })

yield* Effect.gen(function* () {
const jobs = yield* BackgroundJob.Service
yield* jobs.start({ type: "test", run: Effect.never })
}).pipe(provideInstanceEffect(dir))

yield* awaitWithTimeout(store.disposeAll(), "disposal deadlocked with active background job")
}),
)

it.live("reload completes when background jobs are active", () =>
Effect.gen(function* () {
const dir = yield* tmpdirScoped({ git: true })
const store = yield* InstanceStore.Service

yield* store.load({ directory: dir })

yield* Effect.gen(function* () {
const jobs = yield* BackgroundJob.Service
yield* jobs.start({ type: "test", run: Effect.never })
}).pipe(provideInstanceEffect(dir))

yield* awaitWithTimeout(
reloadInstance({ directory: dir }),
"reload deadlocked with active background job",
)
}),
)

it.live("disposal completes when a fiber is blocked on background.wait", () =>
Effect.gen(function* () {
const dir = yield* tmpdirScoped({ git: true })
const store = yield* InstanceStore.Service

yield* store.load({ directory: dir })

yield* Effect.gen(function* () {
const jobs = yield* BackgroundJob.Service
const jobId = "lifecycle-wait-test"
yield* jobs.start({ id: jobId, type: "test", run: Effect.never })

yield* Effect.acquireUseRelease(
Effect.void,
() => jobs.wait({ id: jobId }),
(_, exit) =>
Exit.hasInterrupts(exit) ? jobs.cancel(jobId).pipe(Effect.ignore, Effect.forkDetach, Effect.asVoid) : Effect.void,
).pipe(Effect.forkScoped)

yield* Effect.yieldNow
}).pipe(provideInstanceEffect(dir))

yield* awaitWithTimeout(store.disposeAll(), "disposal deadlocked with fiber blocked on background.wait")
}),
)

it.live("disposal timeout prevents a deadlocked disposer from hanging indefinitely", () =>
Effect.gen(function* () {
const dir = yield* tmpdirScoped({ git: true })
const store = yield* InstanceStore.Service

yield* store.load({ directory: dir })

const off = registerDisposer(
() => new Promise<void>(() => {}),
)

try {
yield* awaitWithTimeout(
store.disposeAll(),
"disposal should have completed via timeout",
"15 seconds",
)
} finally {
off()
}
}),
{ timeout: 20_000 },
)
})
Loading