refactor: replace disposeAll dedup slot with cachedWithTTL
The manual Deferred slot + uninterruptibleMask + identity check collapses into Effect.cachedWithTTL(_, Duration.zero): concurrent callers share the in-flight execution, and the cache expires on completion so the next call runs fresh. Adds a test pinning the re-arm semantic.
This commit is contained in:
parent
8a63cbe79c
commit
1b146ad094
2 changed files with 42 additions and 36 deletions
|
|
@ -3,7 +3,7 @@ import { WorkspaceContext } from "@/control-plane/workspace-context"
|
|||
import { disposeInstance } from "@/effect/instance-registry"
|
||||
import { makeRuntime } from "@/effect/run-service"
|
||||
import { AppFileSystem } from "@opencode-ai/core/filesystem"
|
||||
import { Context, Deferred, Effect, Exit, Layer, Scope } from "effect"
|
||||
import { Context, Deferred, Duration, Effect, Exit, Layer, Scope } from "effect"
|
||||
import { context, type InstanceContext } from "./instance-context"
|
||||
import * as Project from "./project"
|
||||
|
||||
|
|
@ -33,9 +33,6 @@ export const layer: Layer.Layer<Service, never, Project.Service> = Layer.effect(
|
|||
const project = yield* Project.Service
|
||||
const scope = yield* Scope.Scope
|
||||
const cache = new Map<string, Entry>()
|
||||
const disposal = {
|
||||
all: undefined as Deferred.Deferred<void> | undefined,
|
||||
}
|
||||
|
||||
const boot = Effect.fn("InstanceStore.boot")(function* (input: LoadInput & { directory: string }) {
|
||||
const ctx =
|
||||
|
|
@ -146,41 +143,29 @@ export const layer: Layer.Layer<Service, never, Project.Service> = Layer.effect(
|
|||
yield* disposeEntry(ctx.directory, entry, ctx).pipe(Effect.asVoid)
|
||||
})
|
||||
|
||||
const disposeAll = Effect.fn("InstanceStore.disposeAll")(function* () {
|
||||
return yield* Effect.uninterruptibleMask((restore) =>
|
||||
Effect.gen(function* () {
|
||||
const existing = disposal.all
|
||||
if (existing) return yield* restore(Deferred.await(existing))
|
||||
|
||||
const done = Deferred.makeUnsafe<void>()
|
||||
const entries = [...cache.entries()]
|
||||
disposal.all = done
|
||||
const exit = yield* Effect.gen(function* () {
|
||||
yield* Effect.logInfo("disposing all instances")
|
||||
yield* Effect.forEach(
|
||||
entries,
|
||||
(item) =>
|
||||
Effect.gen(function* () {
|
||||
const exit = yield* Deferred.await(item[1].deferred).pipe(Effect.exit)
|
||||
if (Exit.isFailure(exit)) {
|
||||
yield* Effect.logWarning("instance dispose failed", { key: item[0], cause: exit.cause })
|
||||
yield* removeEntry(item[0], item[1])
|
||||
return
|
||||
}
|
||||
yield* disposeEntry(item[0], item[1], exit.value)
|
||||
}),
|
||||
{ discard: true },
|
||||
)
|
||||
}).pipe(Effect.exit)
|
||||
yield* Deferred.done(done, exit).pipe(Effect.asVoid)
|
||||
if (disposal.all === done) {
|
||||
disposal.all = undefined
|
||||
}
|
||||
return yield* restore(Deferred.await(done))
|
||||
}),
|
||||
const disposeAllOnce = Effect.fnUntraced(function* () {
|
||||
yield* Effect.logInfo("disposing all instances")
|
||||
yield* Effect.forEach(
|
||||
[...cache.entries()],
|
||||
(item) =>
|
||||
Effect.gen(function* () {
|
||||
const exit = yield* Deferred.await(item[1].deferred).pipe(Effect.exit)
|
||||
if (Exit.isFailure(exit)) {
|
||||
yield* Effect.logWarning("instance dispose failed", { key: item[0], cause: exit.cause })
|
||||
yield* removeEntry(item[0], item[1])
|
||||
return
|
||||
}
|
||||
yield* disposeEntry(item[0], item[1], exit.value)
|
||||
}),
|
||||
{ discard: true },
|
||||
)
|
||||
})
|
||||
|
||||
const cachedDisposeAll = yield* Effect.cachedWithTTL(disposeAllOnce(), Duration.zero)
|
||||
const disposeAll = Effect.fn("InstanceStore.disposeAll")(function* () {
|
||||
return yield* cachedDisposeAll
|
||||
})
|
||||
|
||||
yield* Effect.addFinalizer(() => disposeAll().pipe(Effect.ignore))
|
||||
|
||||
return Service.of({
|
||||
|
|
|
|||
|
|
@ -215,6 +215,27 @@ describe("InstanceStore", () => {
|
|||
}),
|
||||
)
|
||||
|
||||
it.live("re-arms disposeAll after completion", () =>
|
||||
Effect.gen(function* () {
|
||||
const dir1 = yield* tmpdirScoped({ git: true })
|
||||
const dir2 = yield* tmpdirScoped({ git: true })
|
||||
const store = yield* InstanceStore.Service
|
||||
const disposed: Array<string> = []
|
||||
const off = registerDisposer(async (directory) => {
|
||||
disposed.push(directory)
|
||||
})
|
||||
yield* Effect.addFinalizer(() => Effect.sync(off))
|
||||
|
||||
yield* store.load({ directory: dir1 })
|
||||
yield* store.disposeAll()
|
||||
expect(disposed).toEqual([dir1])
|
||||
|
||||
yield* store.load({ directory: dir2 })
|
||||
yield* store.disposeAll()
|
||||
expect(disposed).toEqual([dir1, dir2])
|
||||
}),
|
||||
)
|
||||
|
||||
it.live("keeps Instance.provide as the legacy ALS wrapper", () =>
|
||||
Effect.gen(function* () {
|
||||
const dir = yield* tmpdirScoped({ git: true })
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue