opensource/kilocode
3
0
Fork 0
mirror of https://github.com/Kilo-Org/kilocode.git synced 2026-08-29 03:44:06 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: make settings save concurrency failure-safe

Browse Source
This commit is contained in:
marius-kilocode
2026-07-10 13:17:07 +02:00
parent 8b466010c5
commit 45ced299e8
11 changed files with 151 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/faster-vscode-settings-save.md
+1 -1
View File
@@ -3,4 +3,4 @@
"@kilocode/cli": patch
---
Speed up VS Code settings saves by draining and disposing worktree instances concurrently, then finishing once config writes succeed.
Speed up VS Code settings saves by draining pending prompts and disposing worktree instances concurrently.
packages/kilo-vscode/src/KiloProvider.ts
+22 -37
View File
@@ -59,7 +59,7 @@ import type { RemoteStatusService } from "./services/RemoteStatusService"
import { resolveProjectDirectory } from "./project-directory"
import { seedSessionStatuses } from "./session-status"
import { normalizeEnhancePromptErrorMessage } from "./enhance-prompt-error"
import { deadline, retry } from "./services/cli-backend/retry"
import { retry } from "./services/cli-backend/retry"
import { slimInfo, slimPart, slimParts } from "./kilo-provider/slim-metadata"
import { handleSidebarWorktreeMessage } from "./kilo-provider/sidebar-worktree"
import { parseMessageFiles, type MessageFile } from "./kilo-provider/message-files"
@@ -361,8 +361,6 @@ export class KiloProvider implements vscode.WebviewViewProvider, TelemetryProper
private cachedMcpStatusMessage: unknown = null
/** Ref-count of in-flight handleUpdateConfig calls; prevents fetchAndSendConfig from sending stale data */
private pending = 0
private configRevision = 0
private refreshWait = 5_000
private configWarningsShown = false
/** Cached notificationsLoaded payload */
private cachedNotificationsMessage: NotificationsMessage | null = null
@@ -2434,13 +2432,11 @@ export class KiloProvider implements vscode.WebviewViewProvider, TelemetryProper
try {
const workspaceDir = this.getWorkspaceDirectory()
const revision = this.configRevision
const [{ data: config }, { data: global }, { data: overlay }] = await Promise.all([
retry(() => this.client!.config.get({ directory: workspaceDir }, { throwOnError: true })),
this.client.global.config.get({ throwOnError: true }),
this.client.config.overlay({ directory: workspaceDir, scope: "project" }, { throwOnError: true }),
])
if (revision !== this.configRevision) return
this.cachedGlobalConfig = global ?? null
const message = {
@@ -2461,10 +2457,8 @@ export class KiloProvider implements vscode.WebviewViewProvider, TelemetryProper
/** Fetch global-only config (no project/managed layers) for settings export. */
private async fetchAndSendGlobalConfig(): Promise<void> {
if (!this.client || this.connectionState !== "connected") return
const revision = this.configRevision
try {
const { data: config } = await this.client.global.config.get({ throwOnError: true })
if (revision !== this.configRevision) return
this.cachedGlobalConfig = config ?? null
this.postMessage({ type: "globalConfigLoaded", config })
} catch (error) {
@@ -2547,7 +2541,6 @@ export class KiloProvider implements vscode.WebviewViewProvider, TelemetryProper
*/
private async fetchAndSendConfigUpdated(): Promise<void> {
if (!this.client || this.connectionState !== "connected") return
const revision = ++this.configRevision
try {
const dir = this.getWorkspaceDirectory()
const [{ data: config }, { data: global }, { data: overlay }] = await Promise.all([
@@ -2555,7 +2548,6 @@ export class KiloProvider implements vscode.WebviewViewProvider, TelemetryProper
this.client.global.config.get({ throwOnError: true }),
this.client.config.overlay({ directory: dir, scope: "project" }, { throwOnError: true }),
])
if (revision !== this.configRevision) return
this.cachedGlobalConfig = global ?? null
this.cachedConfigMessage = {
type: "configLoaded",
@@ -2918,7 +2910,6 @@ export class KiloProvider implements vscode.WebviewViewProvider, TelemetryProper
const hasProject = Object.keys(project).length > 0 || projectUnset.length > 0
this.pending++
this.configRevision++
const dir = this.getWorkspaceDirectory()
try {
@@ -2936,32 +2927,17 @@ export class KiloProvider implements vscode.WebviewViewProvider, TelemetryProper
)
}
} catch (error) {
this.cachedConfigMessage = null
this.cachedGlobalConfig = null
this.pending--
await deadline(
this.fetchAndSendConfigUpdated(),
this.refreshWait,
"Timed out refreshing config after a failed save",
).catch((err) => console.error("[Kilo New] KiloProvider: Failed to refresh config after a failed save:", err))
this.postConfigFailure(error)
this.pending--
return
}
const revision = ++this.configRevision
this.postMessage({ type: "configSaved" })
try {
const [{ data: merged }, { data: global }, { data: overlay }] = await deadline(
Promise.all([
retry(() => this.client!.config.get({ directory: dir }, { throwOnError: true })),
this.client.global.config.get({ throwOnError: true }),
this.client.config.overlay({ directory: dir, scope: "project" }, { throwOnError: true }),
]),
this.refreshWait,
"Timed out refreshing saved config",
)
if (revision !== this.configRevision) return
const [{ data: merged }, { data: global }, { data: overlay }] = await Promise.all([
retry(() => this.client!.config.get({ directory: dir }, { throwOnError: true })),
this.client.global.config.get({ throwOnError: true }),
this.client.config.overlay({ directory: dir, scope: "project" }, { throwOnError: true }),
])
this.cachedGlobalConfig = global ?? null
this.cachedConfigMessage = {
type: "configLoaded",
@@ -2986,12 +2962,21 @@ export class KiloProvider implements vscode.WebviewViewProvider, TelemetryProper
])
} catch (error) {
console.error("[Kilo New] KiloProvider: Config write succeeded but post-write refresh failed:", error)
// The webview already holds the optimistic state acknowledged by configSaved.
// Never replay stale scoped config when the authoritative refresh is unavailable.
if (revision === this.configRevision) {
this.cachedConfigMessage = null
this.cachedGlobalConfig = null
}
const patch =
partial.indexing === undefined && project.indexing === undefined
? { ...partial, ...project }
: { ...partial, ...project, indexing: { ...(partial.indexing ?? {}), ...(project.indexing ?? {}) } }
const cached = (this.cachedConfigMessage as { config?: unknown } | null)?.config
const features = (this.cachedConfigMessage as { features?: unknown } | null)?.features
const optimistic =
cached && typeof cached === "object" ? { ...(cached as Record<string, unknown>), ...patch } : patch
this.postMessage({
type: "configUpdated",
config: optimistic,
globalConfig: this.cachedGlobalConfig ?? undefined,
settings: { maxCost: this.maxCostSetting(), languageCommitMessage: this.commitMessageLanguageSetting() },
features: features ?? configFeatures(optimistic as Config),
})
this.requirements.clear()
} finally {
this.pending--
packages/kilo-vscode/src/services/cli-backend/retry.ts
+2 -18
View File
@@ -1,5 +1,5 @@
// retry() is replicated from packages/core/src/util/retry.ts to avoid adding @opencode-ai/core
// as a dependency of the extension. Keep retry() in sync with the original.
// Replicated from packages/core/src/util/retry.ts to avoid adding @opencode-ai/core
// as a dependency of the extension. Keep in sync with the original.
const TRANSIENT = [
"load failed",
@@ -32,19 +32,3 @@ export async function retry<T>(fn: () => Promise<T>, attempts = 3, delay = 500):
}
throw last
}
export function deadline<T>(task: Promise<T>, delay: number, message: string): Promise<T> {
return new Promise((resolve, reject) => {
const timer = setTimeout(() => reject(new Error(message)), delay)
task.then(
(value) => {
clearTimeout(timer)
resolve(value)
},
(err) => {
clearTimeout(timer)
reject(err)
},
)
})
}
packages/kilo-vscode/tests/unit/config-utils.test.ts
+4 -26
View File
@@ -198,22 +198,17 @@ describe("ConfigState", () => {
expect(s.config.username).toBe("bob") // server update applied
})
it("waits for explicit save confirmation before clearing the draft", () => {
it("clears draft when update confirms our save", () => {
const s = new ConfigState()
s.handleConfigLoaded({ snapshot: true })
s.updateConfig({ snapshot: false })
s.saveConfig()
expect(s.saving).toBe(true)
// An unrelated config push must not confirm the in-flight write.
// Server confirms the write
s.handleConfigUpdated({ snapshot: false })
expect(s.config.snapshot).toBe(false)
expect(s.dirty).toBe(true)
expect(s.saving).toBe(true)
s.handleConfigSaved()
expect(s.dirty).toBe(false)
expect(s.saving).toBe(false)
expect(Object.keys(s.draft).length).toBe(0)
@@ -225,9 +220,8 @@ describe("ConfigState", () => {
s.updateConfig({ default_agent: null })
s.saveConfig()
// The refresh returns config without default_agent before the write ack.
// Server confirms the write by returning config without default_agent.
s.handleConfigUpdated({})
s.handleConfigSaved()
expect(s.config.default_agent).toBeUndefined()
expect(s.dirty).toBe(false)
@@ -264,20 +258,6 @@ describe("ConfigState", () => {
expect(s.saved.agent?.code?.prompt).toBeUndefined()
expect(s.config.agent?.code?.prompt).toBeUndefined()
})
it("rejects programmatic edits while the settings UI is inert", () => {
const s = new ConfigState()
s.handleConfigLoaded({ snapshot: true, username: "alice" })
s.updateConfig({ snapshot: false })
s.saveConfig()
s.updateConfig({ username: "bob" })
s.handleConfigSaved()
expect(s.config).toEqual({ snapshot: false, username: "alice" })
expect(s.saved).toEqual({ snapshot: false, username: "alice" })
expect(s.dirty).toBe(false)
})
})
describe("configSaveFailed while a save is in-flight", () => {
@@ -309,7 +289,6 @@ describe("ConfigState", () => {
s.saveConfig()
s.saveConfig()
s.handleConfigUpdated({ snapshot: false })
s.handleConfigSaved()
expect(s.saving).toBe(false)
expect(s.dirty).toBe(false)
@@ -380,9 +359,8 @@ describe("ConfigState", () => {
s.updateConfig({ agent: { explore: { model: null } } })
s.saveConfig()
// Backend removed the override, then explicitly confirms the write.
// Backend removed the override and pushes the stripped config back.
s.handleConfigUpdated({ agent: { explore: {} } })
s.handleConfigSaved()
expect(s.config.agent?.explore?.model).toBeUndefined()
expect(s.dirty).toBe(false)
packages/kilo-vscode/tests/unit/kilo-provider-indexing-refresh.test.ts
-129
View File
@@ -8,10 +8,6 @@ type Internals = {
connectionState: "connecting" | "connected" | "disconnected" | "error"
currentSession: { id: string } | null
cachedIndexingStatusMessage: unknown
pending: number
configRevision: number
refreshWait: number
postMessage: (message: unknown) => void
handleEvent: (event: unknown, directory?: string) => void
reloadAfterAuthChange: () => Promise<void>
handleUpdateConfig: (
@@ -20,8 +16,6 @@ type Internals = {
globalUnset?: string[][],
projectUnset?: string[][],
) => Promise<void>
fetchAndSendGlobalConfig: () => Promise<void>
fetchAndSendConfigUpdated: () => Promise<void>
fetchAndSendConfig: () => Promise<void>
fetchAndSendProviders: () => Promise<void>
fetchAndSendAgents: () => Promise<void>
@@ -53,7 +47,6 @@ function createConnection() {
}
return {
client,
drains: () => drains,
patches: () => patches,
service: {
@@ -116,128 +109,6 @@ describe("KiloProvider indexing refresh", () => {
expect(indexing).toBe(0)
})
it("confirms saved config when the post-write refresh stalls", async () => {
const conn = createConnection()
conn.client.config.get = async () => new Promise<never>(() => {})
const provider = new KiloProvider({} as never, conn.service as never)
const internal = provider as unknown as Internals
const messages: Array<{ type?: string; writes: number }> = []
internal.connectionState = "connected"
internal.refreshWait = 0
internal.postMessage = (message) =>
messages.push({ ...(message as { type?: string }), writes: conn.patches().length })
await internal.handleUpdateConfig(
{ indexing: { provider: "kilo" } },
{ commit_message: { prompt: "Use conventional commits" } },
)
expect(messages).toEqual([{ type: "configSaved", writes: 2 }])
expect(internal.pending).toBe(0)
})
it("does not confirm a partially written scoped save", async () => {
const conn = createConnection()
conn.client.config.overlayUpdate = async (patch: unknown) => {
conn.patches().push(patch)
if ((patch as { scope?: string }).scope === "project") throw new Error("project write failed")
return { data: {} }
}
const provider = new KiloProvider({} as never, conn.service as never)
const internal = provider as unknown as Internals
const messages: Array<{ type?: string }> = []
internal.connectionState = "connected"
internal.postMessage = (message) => messages.push(message as { type?: string })
await internal.handleUpdateConfig({ snapshot: true }, { commit_message: { prompt: "test" } })
expect(messages.map((message) => message.type)).toEqual(["configUpdated", "configUpdateFailed"])
expect(internal.pending).toBe(0)
})
it("reports a partial write failure when the recovery refresh stalls", async () => {
const conn = createConnection()
conn.client.config.overlayUpdate = async (patch: unknown) => {
conn.patches().push(patch)
if ((patch as { scope?: string }).scope === "project") throw new Error("project write failed")
return { data: {} }
}
conn.client.config.get = async () => new Promise<never>(() => {})
const provider = new KiloProvider({} as never, conn.service as never)
const internal = provider as unknown as Internals
const messages: Array<{ type?: string }> = []
internal.connectionState = "connected"
internal.refreshWait = 0
internal.postMessage = (message) => messages.push(message as { type?: string })
await internal.handleUpdateConfig({ snapshot: true }, { commit_message: { prompt: "test" } })
expect(messages.map((message) => message.type)).toEqual(["configUpdateFailed"])
expect(internal.pending).toBe(0)
})
it("drops a stale config refresh that finishes after a newer one", async () => {
const conn = createConnection()
let release!: () => void
const gate = new Promise<void>((resolve) => {
release = resolve
})
let calls = 0
conn.client.config.get = async () => {
calls += 1
if (calls === 1) {
await gate
return { data: { snapshot: true } }
}
return { data: { snapshot: false } }
}
const provider = new KiloProvider({} as never, conn.service as never)
const internal = provider as unknown as Internals
const messages: Array<{ type?: string; config?: Config }> = []
internal.connectionState = "connected"
internal.postMessage = (message) => messages.push(message as { type?: string; config?: Config })
const stale = internal.fetchAndSendConfigUpdated()
await Bun.sleep(0)
await internal.fetchAndSendConfigUpdated()
release()
await stale
expect(messages).toHaveLength(1)
expect(messages[0]?.config?.snapshot).toBe(false)
expect(internal.configRevision).toBe(2)
})
it("drops a stale global-only config response", async () => {
const conn = createConnection()
let release!: () => void
const gate = new Promise<void>((resolve) => {
release = resolve
})
conn.client.global.config.get = async () => {
await gate
return { data: { snapshot: true } }
}
const provider = new KiloProvider({} as never, conn.service as never)
const internal = provider as unknown as Internals
const messages: Array<{ type?: string }> = []
internal.connectionState = "connected"
internal.postMessage = (message) => messages.push(message as { type?: string })
const stale = internal.fetchAndSendGlobalConfig()
await Bun.sleep(0)
internal.configRevision++
release()
await stale
expect(messages).toHaveLength(0)
})
it("refreshes providers when prompt-training model visibility changes", async () => {
const conn = createConnection()
const provider = new KiloProvider({} as never, conn.service as never)
packages/kilo-vscode/webview-ui/src/components/settings/Settings.tsx
-2
View File
@@ -158,8 +158,6 @@ const Settings: Component<SettingsProps> = (props) => {
variant="settings"
value={active()}
onChange={onTabChange}
inert={saving()}
aria-busy={saving()}
style={{ flex: 1, overflow: "hidden" }}
>
<Tabs.List>
packages/kilo-vscode/webview-ui/src/context/config.tsx
+32 -29
View File
@@ -130,34 +130,42 @@ export const ConfigProvider: ParentComponent = (props) => {
return
}
if (message.type === "configUpdated") {
// Config pushes can come from unrelated saves or disposal events. Re-apply
// local drafts and wait for configSaved before acknowledging this write.
setConfig(resolveConfig(message.config, draft(), has(draft() as Record<string, unknown>)))
if (message.globalConfig !== undefined) {
setGlobalConfig(mergeScopedConfig(message.globalConfig, globalDraft()))
setSavedGlobal(message.globalConfig)
if (saving()) {
// This configUpdated is the confirmation of our saveConfig() write.
// Clear the draft now that the server has confirmed the write.
setSaving(false)
setDraft({})
setGlobalDraft({})
setProjectDraft({})
setSaveError(null)
setConfig(message.config)
if (message.globalConfig !== undefined) {
setGlobalConfig(mergeScopedConfig(message.globalConfig, globalDraft()))
setSavedGlobal(message.globalConfig)
}
if (message.projectConfig !== undefined) {
setProjectConfig(message.projectConfig)
setSavedProject(message.projectConfig)
}
setFeatures(message.features)
} else {
// configUpdated from a different source (e.g. PermissionDock save).
// Re-apply the draft on top so pending settings changes are preserved.
setConfig(resolveConfig(message.config, draft(), has(draft() as Record<string, unknown>)))
if (message.globalConfig !== undefined) {
setGlobalConfig(mergeScopedConfig(message.globalConfig, globalDraft()))
setSavedGlobal(message.globalConfig)
}
if (message.projectConfig !== undefined) {
setProjectConfig(mergeScopedConfig(message.projectConfig, projectDraft()))
setSavedProject(message.projectConfig)
}
setFeatures(message.features)
}
if (message.projectConfig !== undefined) {
setProjectConfig(mergeScopedConfig(message.projectConfig, projectDraft()))
setSavedProject(message.projectConfig)
}
setFeatures(message.features)
if (message.settings) mergeSettings(message.settings)
setSaved(message.config)
return
}
if (message.type === "configSaved") {
if (!saving()) return
setSaving(false)
setDraft({})
setGlobalDraft({})
setProjectDraft({})
setSaveError(null)
setSaved(config())
setSavedGlobal(globalConfig())
setSavedProject(projectConfig())
return
}
if (message.type === "configUpdateFailed") {
// The write was rejected (e.g. schema validation) — surface the error
// and keep the draft + isDirty so the user can correct and retry.
@@ -205,7 +213,6 @@ export const ConfigProvider: ParentComponent = (props) => {
})
function updateConfig(partial: Partial<Config>) {
if (saving()) return
// Optimistically update local state with deep merge + null stripping
setConfig((prev) => stripNulls(deepMerge(prev, partial)))
// Accumulate in draft — will be sent on saveConfig()
@@ -216,28 +223,24 @@ export const ConfigProvider: ParentComponent = (props) => {
}
function updateGlobalConfig(partial: Partial<Config>) {
if (saving()) return
setGlobalConfig((prev) => mergeScopedConfig(prev, partial))
setGlobalDraft((prev) => deepMerge(prev as Config, partial))
setSaveError(null)
}
function updateProjectConfig(partial: Partial<Config>) {
if (saving()) return
setProjectConfig((prev) => mergeScopedConfig(prev, partial))
setProjectDraft((prev) => deepMerge(prev as Config, partial))
setSaveError(null)
}
function updateSetting(key: string, value: unknown) {
if (saving()) return
setSettings((prev) => ({ ...prev, [key]: value }))
setSettingsDraft((prev) => ({ ...prev, [key]: value }))
setSaveError(null)
}
function saveConfig() {
if (saving()) return
const changes = draft()
const globals = globalDraft()
const projects = projectDraft()
@@ -247,7 +250,7 @@ export const ConfigProvider: ParentComponent = (props) => {
const projectDirty = has(projects as Record<string, unknown>)
const settingsDirty = has(pending)
if (!configDirty && !globalDirty && !projectDirty && !settingsDirty) return
// Don't clear draft/isDirty yet — wait for configSaved confirmation.
// Don't clear draft/isDirty yet — wait for configUpdated confirmation.
// If the write fails, the save bar stays visible so the user can retry.
setSaving(true)
setSaveError(null)
packages/kilo-vscode/webview-ui/src/types/messages/extension-messages.ts
-5
View File
@@ -539,10 +539,6 @@ export interface ConfigUpdatedMessage {
features: FeatureFlags
}
export interface ConfigSavedMessage {
type: "configSaved"
}
export interface ConfigUpdateFailedMessage {
type: "configUpdateFailed"
message: string
@@ -1137,7 +1133,6 @@ export type ExtensionMessage =
| ClaudeCompatSettingLoadedMessage
| ConfigLoadedMessage
| ConfigUpdatedMessage
| ConfigSavedMessage
| ConfigUpdateFailedMessage
| GlobalConfigLoadedMessage
| NotificationSettingsLoadedMessage
packages/kilo-vscode/webview-ui/src/utils/config-utils.ts
+8 -2
View File
@@ -100,7 +100,6 @@ export class ConfigState {
/** Accumulate a partial change (same as the toggle click path). */
updateConfig(partial: Partial<Config>) {
if (this.saving) return
this.config = stripNulls(deepMerge(this.config, partial))
this.draft = deepMerge(this.draft as Config, partial)
this.dirty = true
@@ -116,7 +115,14 @@ export class ConfigState {
/** Handle an incoming configUpdated push from the extension. */
handleConfigUpdated(server: Config) {
this.config = resolveConfig(server, this.draft, this.dirty)
if (this.saving) {
this.saving = false
this.draft = {}
this.dirty = false
this.config = server
} else {
this.config = resolveConfig(server, this.draft, this.dirty)
}
this.saved = server
}
packages/opencode/src/project/instance-store.ts
+14 -8
View File
@@ -100,10 +100,12 @@ export const layer: Layer.Layer<Service, never, Project.Service | InstanceBootst
const disposeEntry = Effect.fnUntraced(function* (directory: string, entry: Entry, ctx: InstanceContext) {
if (cache.get(directory) !== entry) return false
yield* disposeContext(ctx)
if (cache.get(directory) !== entry) return false
cache.delete(directory)
return true
// kilocode_change start - remove disposed entries even when event publication fails
const exit = yield* Effect.exit(disposeContext(ctx))
const removed = yield* removeEntry(directory, entry)
yield* exit
return removed
// kilocode_change end
})
const load = (input: LoadInput): Effect.Effect<InstanceContext> => {
@@ -163,7 +165,8 @@ export const layer: Layer.Layer<Service, never, Project.Service | InstanceBootst
const disposeAllOnce = Effect.fnUntraced(function* () {
yield* Effect.logInfo("disposing all instances")
yield* Effect.forEach(
// kilocode_change start - dispose independent worktrees concurrently without interrupting siblings
const exits = yield* Effect.forEach(
[...cache.entries()],
(item) =>
Effect.gen(function* () {
@@ -176,9 +179,12 @@ export const layer: Layer.Layer<Service, never, Project.Service | InstanceBootst
return
}
yield* disposeEntry(item[0], item[1], exit.value)
}),
{ discard: true, concurrency: 4 }, // kilocode_change - dispose independent worktrees concurrently
)
}).pipe(Effect.exit),
{ concurrency: 4 },
).pipe(Effect.uninterruptible)
const failure = exits.find(Exit.isFailure)
if (failure) yield* failure
// kilocode_change end
})
const cachedDisposeAll = yield* Effect.cachedWithTTL(disposeAllOnce(), Duration.zero)
packages/opencode/test/kilocode/project/instance-store.test.ts
+68 -2
View File
@@ -1,6 +1,7 @@
import { describe, expect } from "bun:test"
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
import { Deferred, Effect, Fiber, Layer } from "effect"
import { Deferred, Effect, Exit, Fiber, Layer, Scope } from "effect"
import { GlobalBus } from "../../../src/bus/global"
import { registerDisposer } from "../../../src/effect/instance-registry"
import { InstanceBootstrap } from "../../../src/project/bootstrap-service"
import { InstanceStore } from "../../../src/project/instance-store"
@@ -41,11 +42,76 @@ describe("InstanceStore disposal", () => {
yield* Effect.forEach(dirs, (directory) => store.load({ directory }), { discard: true })
const fiber = yield* store.disposeAll().pipe(Effect.forkScoped)
yield* awaitWithTimeout(Deferred.await(ready), "instance disposal remained serial", "1 second")
yield* awaitWithTimeout(Deferred.await(ready), "instance disposal remained serial")
expect(started).toEqual(new Set(dirs))
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(fiber)
}),
)
it.live("finishes sibling disposal when an event listener throws", () =>
Effect.gen(function* () {
const dirs = yield* Effect.all(
Array.from({ length: 4 }, () => tmpdirScoped({ git: true })),
{ concurrency: "unbounded" },
)
const store = yield* InstanceStore.Service
const before = yield* Effect.forEach(dirs, (directory) => store.load({ directory }))
const disposed = new Set<string>()
const listener = (event: { directory?: string; payload?: { type?: string } }) => {
if (event.payload?.type === "server.instance.disposed" && event.directory === dirs[0]) {
throw new Error("listener failed")
}
}
yield* register(async (directory) => {
if (dirs.includes(directory)) disposed.add(directory)
})
GlobalBus.on("event", listener)
yield* Effect.addFinalizer(() => Effect.sync(() => GlobalBus.off("event", listener)))
const exit = yield* Effect.exit(store.disposeAll())
GlobalBus.off("event", listener)
expect(Exit.isFailure(exit)).toBe(true)
expect(disposed).toEqual(new Set(dirs))
const after = yield* Effect.forEach(dirs, (directory) => store.load({ directory }))
for (const [index, ctx] of after.entries()) {
expect(ctx).not.toBe(before[index])
}
}),
)
it.live("finishes queued disposal when the caller is interrupted", () =>
Effect.gen(function* () {
const dirs = yield* Effect.all(
Array.from({ length: 5 }, () => tmpdirScoped({ git: true })),
{ concurrency: "unbounded" },
)
const store = yield* InstanceStore.Service
const ready = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const started = new Set<string>()
yield* Effect.addFinalizer(() => Deferred.succeed(release, undefined).pipe(Effect.ignore))
yield* register(async (directory) => {
if (!dirs.includes(directory)) return
started.add(directory)
if (started.size === 4) Deferred.doneUnsafe(ready, Effect.void)
await Effect.runPromise(Deferred.await(release))
})
yield* Effect.forEach(dirs, (directory) => store.load({ directory }), { discard: true })
const disposal = yield* store.disposeAll().pipe(Effect.forkScoped)
yield* awaitWithTimeout(Deferred.await(ready), "bounded disposal did not start")
const scope = yield* Scope.Scope
const interrupted = yield* Fiber.interrupt(disposal).pipe(Effect.forkIn(scope, { startImmediately: true }))
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(interrupted)
expect(started).toEqual(new Set(dirs))
}),
)
})