diff --git a/src/node/services/heartbeatService.test.ts b/src/node/services/heartbeatService.test.ts index 39b02c0dd2..51f83de656 100644 --- a/src/node/services/heartbeatService.test.ts +++ b/src/node/services/heartbeatService.test.ts @@ -18,6 +18,7 @@ import type { BackgroundProcessManager } from "./backgroundProcessManager"; import type { ExtensionMetadataService } from "./ExtensionMetadataService"; import { advanceAnchoredDeadline, HeartbeatService } from "./heartbeatService"; import type { HistoryService } from "./historyService"; +import { IdleDispatcher } from "./idleDispatcher"; import type { InitStateManager } from "./initStateManager"; import type { TaskService } from "./taskService"; import { makeAgentTaskIntegrationFake } from "./taskWorkspaceSeam.testUtils"; @@ -584,6 +585,60 @@ describe("HeartbeatService", () => { expect(internals.queuedWorkspaceIds.size).toBe(0); expect(executeHeartbeatMock).not.toHaveBeenCalled(); }); + + test("start failure releases earlier acquisitions and leaves the service restartable", () => { + const dispatcher = new IdleDispatcher(); + const emitter = new EventEmitter(); + let failListenerRegistration = true; + const realOn = emitter.on.bind(emitter); + // Fail only the SECOND listener registration ("metadata") so the test + // also covers partial-acquisition rollback: the already-registered + // "activity" listener must be released, not leaked across retries. + emitter.on = ((event: string, listener: (...args: unknown[]) => void) => { + if (failListenerRegistration && event === "metadata") { + throw new Error("listener registration failed"); + } + return realOn(event, listener); + }) as typeof emitter.on; + const failingWorkspaceService = Object.assign(emitter, { + getChatHistory: getChatHistoryMock, + executeHeartbeat: executeHeartbeatMock, + isBusyForMessage: isBusyForMessageMock, + }) as unknown as WorkspaceService; + + const failingService = new HeartbeatService( + mockConfig, + mockExtensionMetadata, + failingWorkspaceService, + mockTaskService, + dispatcher + ); + + expect(() => failingService.start()).toThrow(); + + // The "activity" listener registered before the failing "metadata" + // registration must have been rolled back — a leak here would double up + // event handling after a successful retry. + expect(emitter.listenerCount("activity")).toBe(0); + + // The idle-consumer registration acquired before the failing step must + // have been released: re-registering the same consumer name would + // otherwise trip the dispatcher's duplicate-registration assert. + const disposeProbe = dispatcher.registerConsumer({ + name: "heartbeat", + priority: 50, + buildPayload: () => Promise.resolve(null), + }); + disposeProbe(); + + // The rollback restores the stopped state, so start() succeeds once the + // failure cause is fixed. + failListenerRegistration = false; + failingService.start(); + expect(emitter.listenerCount("activity")).toBe(1); + failingService.stop(); + expect(emitter.listenerCount("activity")).toBe(0); + }); }); describe("event handling", () => { diff --git a/src/node/services/heartbeatService.ts b/src/node/services/heartbeatService.ts index d8b7a6b68f..9172b47a33 100644 --- a/src/node/services/heartbeatService.ts +++ b/src/node/services/heartbeatService.ts @@ -1,3 +1,4 @@ +import { Duration, Effect, Exit, Schedule, Scope, type Fiber } from "effect"; import assert from "@/common/utils/assert"; import type { MuxMessage } from "@/common/types/message"; import type { ProjectsConfig, Workspace } from "@/common/types/project"; @@ -61,8 +62,22 @@ export class HeartbeatService { private timelineRecorder: TimelineRecorder = NOOP_TIMELINE_RECORDER; - private startupTimeout: ReturnType | null = null; - private checkInterval: ReturnType | null = null; + // The scheduler runs as a single Effect fiber forked into `lifecycleScope`: + // sleep(STARTUP_DELAY_MS), then tick immediately and every CHECK_INTERVAL_MS. + // The legacy two-field shape is preserved because it is the observable + // lifecycle contract (tests pin the null/non-null progression): + // `startupTimeout` holds the fiber while the startup delay is pending and + // `checkInterval` holds it once the periodic ticker is live. + private startupTimeout: Fiber.Fiber | null = null; + private checkInterval: Fiber.Fiber | null = null; + /** + * Owns every resource start() acquires — idle-consumer registration, + * workspace event listeners, and the scheduler fiber. Closing it releases + * them in reverse acquisition order (fiber interrupt, listeners off, + * consumer dispose — the same order the hand-rolled stop() used) and is + * guaranteed to run them even when a later startup step throws. + */ + private lifecycleScope: Scope.Closeable | null = null; private stopped = true; private readonly nextEligibleAtByWorkspaceId = new Map(); @@ -114,25 +129,90 @@ export class HeartbeatService { this.stopped = false; this.lifecycleVersion += 1; - this.heartbeatConsumerDisposer = this.idleDispatcher.registerConsumer({ - name: HEARTBEAT_IDLE_CONSUMER_NAME, - priority: HEARTBEAT_IDLE_CONSUMER_PRIORITY, - buildPayload: (workspaceId) => this.buildHeartbeatDispatchPayload(workspaceId), - }); - this.workspaceService.on("activity", this.onActivity); - this.workspaceService.on("metadata", this.onMetadata); - - this.startupTimeout = setTimeout(() => { - if (this.stopped) { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + const scope = Scope.makeUnsafe(); + this.lifecycleScope = scope; + + const scheduler = Effect.gen(function* () { + yield* Effect.sleep(Duration.millis(STARTUP_DELAY_MS)); + // Defensive parity with the legacy setTimeout callback: interruption via + // stop() already prevents this resumption, so a stopped service must + // never transition into the ticking phase even if a wake-up raced it. + if (self.stopped) { return; } + // Startup delay elapsed: the same fiber now becomes the periodic ticker. + self.checkInterval = self.startupTimeout; + self.startupTimeout = null; + // Effect.repeat runs the first tick immediately (matching the legacy + // direct tick() call when the startup timer fired), then Schedule.fixed + // reproduces setInterval cadence: wall-clock anchored, no burst catch-up. + // tick() is synchronous fire-and-forget, so the body never delays a slot. + yield* Effect.sync(() => self.tick()).pipe( + Effect.repeat(Schedule.fixed(Duration.millis(CHECK_INTERVAL_MS))) + ); + }); + + const acquireResources = Effect.gen(function* () { + yield* Effect.acquireRelease( + Effect.sync(() => { + self.heartbeatConsumerDisposer = self.idleDispatcher.registerConsumer({ + name: HEARTBEAT_IDLE_CONSUMER_NAME, + priority: HEARTBEAT_IDLE_CONSUMER_PRIORITY, + buildPayload: (workspaceId) => self.buildHeartbeatDispatchPayload(workspaceId), + }); + }), + () => + Effect.sync(() => { + self.heartbeatConsumerDisposer?.(); + self.heartbeatConsumerDisposer = null; + }) + ); + // One acquireRelease per listener: a combined acquisition would install + // its finalizer only after BOTH .on() calls succeed, so a throw from the + // second registration (e.g. a `newListener` hook) would leak the first + // listener across start() retries (Codex P2 on #4031). + yield* Effect.acquireRelease( + Effect.sync(() => { + self.workspaceService.on("activity", self.onActivity); + }), + () => + Effect.sync(() => { + self.workspaceService.off("activity", self.onActivity); + }) + ); + yield* Effect.acquireRelease( + Effect.sync(() => { + self.workspaceService.on("metadata", self.onMetadata); + }), + () => + Effect.sync(() => { + self.workspaceService.off("metadata", self.onMetadata); + }) + ); + self.startupTimeout = yield* Effect.forkIn(scheduler, scope); + }); + + try { + // Runs synchronously: acquisitions are Effect.sync and forkIn executes + // the scheduler up to its first sleep before returning, so the startup + // timer is registered before start() returns (same observable ordering + // as the previous setTimeout call). + Effect.runSync(Scope.provide(scope)(acquireResources)); + } catch (error) { + // Guaranteed cleanup on partial startup failure: close the scope so the + // finalizers registered before the failing step run (the hand-rolled + // version leaked earlier acquisitions here), restore the stopped state + // so a later start() retry is possible, then surface the error. + this.lifecycleScope = null; this.startupTimeout = null; - this.tick(); - this.checkInterval = setInterval(() => { - this.tick(); - }, CHECK_INTERVAL_MS); - }, STARTUP_DELAY_MS); + this.checkInterval = null; + this.stopped = true; + Effect.runSync(Scope.close(scope, Exit.void)); + throw error; + } log.info("HeartbeatService started", { startupDelayMs: STARTUP_DELAY_MS, @@ -144,19 +224,25 @@ export class HeartbeatService { this.stopped = true; this.lifecycleVersion += 1; - if (this.startupTimeout) { - clearTimeout(this.startupTimeout); - this.startupTimeout = null; - } - if (this.checkInterval) { - clearInterval(this.checkInterval); - this.checkInterval = null; - } - - this.workspaceService.off("activity", this.onActivity); - this.workspaceService.off("metadata", this.onMetadata); - this.heartbeatConsumerDisposer?.(); - this.heartbeatConsumerDisposer = null; + // Captured before teardown for the shutdown log below. + const schedulerPhase = + this.checkInterval != null + ? "ticking" + : this.startupTimeout != null + ? "startup_delay" + : "not_started"; + + if (this.lifecycleScope) { + const scope = this.lifecycleScope; + this.lifecycleScope = null; + // Releases everything start() acquired, in reverse acquisition order: + // scheduler fiber interrupt (synchronously clearing its pending timer), + // listeners off, consumer disposer — completing synchronously because + // the fiber only ever suspends on its clock timer. + Effect.runSync(Scope.close(scope, Exit.void)); + } + this.startupTimeout = null; + this.checkInterval = null; this.nextEligibleAtByWorkspaceId.clear(); this.trackedIntervalMsByWorkspaceId.clear(); @@ -166,7 +252,7 @@ export class HeartbeatService { this.isProcessingQueue = false; this.tickInFlight = false; - log.info("HeartbeatService stopped"); + log.info("HeartbeatService stopped", { schedulerPhase }); } private tick(): void { diff --git a/src/node/services/idleCompactionService.ts b/src/node/services/idleCompactionService.ts index 36060941c5..75abaf5d34 100644 --- a/src/node/services/idleCompactionService.ts +++ b/src/node/services/idleCompactionService.ts @@ -1,3 +1,4 @@ +import { Duration, Effect, Exit, Schedule, Scope } from "effect"; import assert from "@/common/utils/assert"; import type { Config } from "@/node/config"; import type { HistoryService } from "./historyService"; @@ -44,8 +45,12 @@ export class IdleCompactionService { private readonly historyService: HistoryService; private readonly extensionMetadata: ExtensionMetadataService; private readonly executeIdleCompaction: (workspaceId: string) => Promise; - private initialTimeout: ReturnType | null = null; - private checkInterval: ReturnType | null = null; + /** + * Owns the checker fiber forked by start(): sleep(INITIAL_CHECK_DELAY_MS), + * then check immediately and every CHECK_INTERVAL_MS. Closing the scope in + * stop() interrupts the fiber, synchronously clearing its pending timer. + */ + private lifecycleScope: Scope.Closeable | null = null; private readonly queue: QueuedIdleCompaction[] = []; private readonly queuedWorkspaceIds = new Set(); private readonly activeWorkspaceIds = new Set(); @@ -76,14 +81,27 @@ export class IdleCompactionService { start(): void { this.stopped = false; - // First check after delay to let startup settle. - this.initialTimeout = setTimeout(() => { - void this.checkAllWorkspaces(); - // Then periodically. - this.checkInterval = setInterval(() => { - void this.checkAllWorkspaces(); - }, CHECK_INTERVAL_MS); - }, INITIAL_CHECK_DELAY_MS); + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + const scope = Scope.makeUnsafe(); + this.lifecycleScope = scope; + + const checker = Effect.gen(function* () { + // First check after delay to let startup settle. + yield* Effect.sleep(Duration.millis(INITIAL_CHECK_DELAY_MS)); + // Effect.repeat runs the first check immediately (matching the legacy + // direct call when the initial timer fired), then Schedule.fixed + // reproduces setInterval cadence. The check stays fire-and-forget so a + // slow sweep never delays the next cadence slot (same as setInterval). + yield* Effect.sync(() => { + void self.checkAllWorkspaces(); + }).pipe(Effect.repeat(Schedule.fixed(Duration.millis(CHECK_INTERVAL_MS)))); + }); + // Runs synchronously up to the checker's first sleep, so the initial-delay + // timer is registered before start() returns (same as the previous + // setTimeout call). + Effect.runSync(Effect.forkIn(checker, scope)); + log.info("IdleCompactionService started", { initialDelayMs: INITIAL_CHECK_DELAY_MS, intervalMs: CHECK_INTERVAL_MS, @@ -96,13 +114,12 @@ export class IdleCompactionService { stop(): void { this.stopped = true; - if (this.initialTimeout) { - clearTimeout(this.initialTimeout); - this.initialTimeout = null; - } - if (this.checkInterval) { - clearInterval(this.checkInterval); - this.checkInterval = null; + if (this.lifecycleScope) { + const scope = this.lifecycleScope; + this.lifecycleScope = null; + // Interrupts the checker fiber, synchronously clearing its pending + // timer — the fiber only ever suspends on its clock timer. + Effect.runSync(Scope.close(scope, Exit.void)); } // Best-effort queue reset: do not start new compactions after stop(). diff --git a/src/node/services/idleDispatcher.ts b/src/node/services/idleDispatcher.ts index 6cb3c95f53..f2dd272600 100644 --- a/src/node/services/idleDispatcher.ts +++ b/src/node/services/idleDispatcher.ts @@ -1,3 +1,4 @@ +import { Duration, Effect, type Fiber } from "effect"; import assert from "@/common/utils/assert"; import { log } from "./log"; @@ -29,7 +30,13 @@ interface IdleDispatcherOptions { interface PendingDispatchRequest { readonly sources: Set; readonly resolvers: Array<() => void>; - debounceTimer: ReturnType | null; + /** + * Debounce fiber: sleeps for `debounceMs` (Effect's clock registers a plain + * `setTimeout` under the hood), then marks the workspace ready. Non-null + * exactly while the debounce window is open — later requests for the same + * workspace coalesce into this pending entry instead of re-arming it. + */ + debounceFiber: Fiber.Fiber | null; } export class IdleDispatcher { @@ -89,14 +96,24 @@ export class IdleDispatcher { pending.sources.add(source); pending.resolvers.push(resolve); - if (pending.debounceTimer != null) { + if (pending.debounceFiber != null) { return; } - pending.debounceTimer = setTimeout(() => { - pending.debounceTimer = null; - this.markWorkspaceReady(workspaceId); - }, this.debounceMs); + // Effect.runFork executes synchronously up to the sleep, so the debounce + // timer is registered before this callback returns (same observable + // ordering as the previous setTimeout call); a zero-duration sleep still + // defers to a timer tick rather than firing inline. + pending.debounceFiber = Effect.runFork( + Effect.sleep(Duration.millis(this.debounceMs)).pipe( + Effect.flatMap(() => + Effect.sync(() => { + pending.debounceFiber = null; + this.markWorkspaceReady(workspaceId); + }) + ) + ) + ); }); } @@ -109,7 +126,7 @@ export class IdleDispatcher { const pending: PendingDispatchRequest = { sources: new Set(), resolvers: [], - debounceTimer: null, + debounceFiber: null, }; this.pendingByWorkspaceId.set(workspaceId, pending); return pending;