diff --git a/packages/workspace-fs/src/watch-idle-keep-alive.test.ts b/packages/workspace-fs/src/watch-idle-keep-alive.test.ts new file mode 100644 index 00000000000..e27b404d8ce --- /dev/null +++ b/packages/workspace-fs/src/watch-idle-keep-alive.test.ts @@ -0,0 +1,355 @@ +import { afterEach, describe, expect, it } from "bun:test"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { FsWatcherManager, type FsWatcherManagerOptions } from "./watch"; +import type { NativeWatchBackend, NativeWatchRequest } from "./watch-backend"; + +interface FakeSubscription { + rootPath: string; + request: NativeWatchRequest; + disposed: boolean; +} + +function createFakeBackend() { + const subscriptions: FakeSubscription[] = []; + let unsubscribeGate: Promise = Promise.resolve(); + const backend: NativeWatchBackend = { + name: "fake", + async subscribe(request) { + const subscription: FakeSubscription = { + rootPath: request.rootPath, + request, + disposed: false, + }; + subscriptions.push(subscription); + return { + async unsubscribe() { + await unsubscribeGate; + subscription.disposed = true; + }, + }; + }, + }; + return { + backend, + subscriptions, + live: (rootPath: string) => + subscriptions.filter((s) => s.rootPath === rootPath && !s.disposed), + holdUnsubscribes() { + let release: () => void = () => {}; + unsubscribeGate = new Promise((resolve) => { + release = resolve; + }); + return release; + }, + }; +} + +const tempRoots: string[] = []; +const managers: FsWatcherManager[] = []; + +afterEach(async () => { + await Promise.all(managers.splice(0).map((m) => m.close())); + await Promise.all( + tempRoots + .splice(0) + .map((rootPath) => fs.rm(rootPath, { recursive: true, force: true })), + ); +}); + +async function createTempRoot(): Promise { + const tempPath = await fs.mkdtemp(path.join(os.tmpdir(), "watch-idle-")); + const rootPath = await fs.realpath(tempPath); + tempRoots.push(rootPath); + return rootPath; +} + +function createManager( + backend: NativeWatchBackend, + options: FsWatcherManagerOptions = {}, +): FsWatcherManager { + const manager = new FsWatcherManager({ + debounceMs: 10, + backend, + idleKeepAliveMs: 100, + ...options, + }); + managers.push(manager); + return manager; +} + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +async function waitUntil(condition: () => boolean, label: string) { + const deadline = Date.now() + 2_000; + while (!condition()) { + if (Date.now() > deadline) { + throw new Error(`Timed out waiting for ${label}`); + } + await sleep(10); + } +} + +describe("FsWatcherManager idle keep-alive", () => { + it("reuses the warm native subscription when re-subscribed within the grace period", async () => { + const fake = createFakeBackend(); + const rootPath = await createTempRoot(); + const manager = createManager(fake.backend); + + const unsubscribe = await manager.subscribe( + { absolutePath: rootPath }, + () => {}, + ); + await unsubscribe(); + + const seen: string[] = []; + await manager.subscribe({ absolutePath: rootPath }, (batch) => { + for (const event of batch.events) seen.push(event.absolutePath); + }); + + expect(fake.subscriptions).toHaveLength(1); + expect(fake.live(rootPath)).toHaveLength(1); + + const filePath = path.join(rootPath, "a.ts"); + fake.subscriptions[0]?.request.onEvents([ + { type: "create", path: filePath }, + ]); + await waitUntil(() => seen.includes(filePath), "event on warm watcher"); + }); + + it("disposes the idle watcher once the grace period elapses", async () => { + const fake = createFakeBackend(); + const rootPath = await createTempRoot(); + const manager = createManager(fake.backend); + + const unsubscribe = await manager.subscribe( + { absolutePath: rootPath }, + () => {}, + ); + await unsubscribe(); + expect(fake.live(rootPath)).toHaveLength(1); + + await waitUntil(() => fake.live(rootPath).length === 0, "idle expiry"); + + await manager.subscribe({ absolutePath: rootPath }, () => {}); + expect(fake.subscriptions).toHaveLength(2); + expect(fake.live(rootPath)).toHaveLength(1); + }); + + it("does not expire a watcher that was re-subscribed during its grace period", async () => { + const fake = createFakeBackend(); + const rootPath = await createTempRoot(); + const manager = createManager(fake.backend); + + const unsubscribe = await manager.subscribe( + { absolutePath: rootPath }, + () => {}, + ); + await unsubscribe(); + await manager.subscribe({ absolutePath: rootPath }, () => {}); + + await sleep(250); + expect(fake.subscriptions).toHaveLength(1); + expect(fake.live(rootPath)).toHaveLength(1); + }); + + it("evicts the oldest idle watcher beyond the idle cap", async () => { + const fake = createFakeBackend(); + const [rootA, rootB, rootC] = await Promise.all([ + createTempRoot(), + createTempRoot(), + createTempRoot(), + ]); + const manager = createManager(fake.backend, { + idleKeepAliveMs: 60_000, + maxIdleWatchers: 2, + }); + + const unsubscribeA = await manager.subscribe( + { absolutePath: rootA }, + () => {}, + ); + const unsubscribeB = await manager.subscribe( + { absolutePath: rootB }, + () => {}, + ); + const unsubscribeC = await manager.subscribe( + { absolutePath: rootC }, + () => {}, + ); + + await unsubscribeA(); + await unsubscribeB(); + expect(fake.live(rootA)).toHaveLength(1); + expect(fake.live(rootB)).toHaveLength(1); + + await unsubscribeC(); + await waitUntil(() => fake.live(rootA).length === 0, "eviction of A"); + expect(fake.live(rootB)).toHaveLength(1); + expect(fake.live(rootC)).toHaveLength(1); + }); + + it("does not count a re-subscribed watcher against the idle cap", async () => { + const fake = createFakeBackend(); + const [rootA, rootB] = await Promise.all([ + createTempRoot(), + createTempRoot(), + ]); + const manager = createManager(fake.backend, { + idleKeepAliveMs: 60_000, + maxIdleWatchers: 1, + }); + + const unsubscribeA = await manager.subscribe( + { absolutePath: rootA }, + () => {}, + ); + const unsubscribeB = await manager.subscribe( + { absolutePath: rootB }, + () => {}, + ); + + await unsubscribeA(); + await manager.subscribe({ absolutePath: rootA }, () => {}); + await unsubscribeB(); + + await sleep(50); + expect(fake.live(rootA)).toHaveLength(1); + expect(fake.live(rootB)).toHaveLength(1); + }); + + it("close() disposes idle watchers and leaves no expiry behind", async () => { + const fake = createFakeBackend(); + const rootPath = await createTempRoot(); + const manager = createManager(fake.backend); + + const unsubscribe = await manager.subscribe( + { absolutePath: rootPath }, + () => {}, + ); + await unsubscribe(); + await manager.close(); + expect(fake.live(rootPath)).toHaveLength(0); + + await manager.subscribe({ absolutePath: rootPath }, () => {}); + await sleep(250); + expect(fake.subscriptions).toHaveLength(2); + expect(fake.live(rootPath)).toHaveLength(1); + }); + + it("disposes immediately when the keep-alive is disabled", async () => { + const fake = createFakeBackend(); + const rootPath = await createTempRoot(); + const manager = createManager(fake.backend, { idleKeepAliveMs: 0 }); + + const unsubscribe = await manager.subscribe( + { absolutePath: rootPath }, + () => {}, + ); + await unsubscribe(); + expect(fake.live(rootPath)).toHaveLength(0); + }); + + it("gives a subscribe racing an in-flight expiry a fresh watcher that survives the old teardown", async () => { + const fake = createFakeBackend(); + const rootPath = await createTempRoot(); + const manager = createManager(fake.backend, { idleKeepAliveMs: 30 }); + + const unsubscribe = await manager.subscribe( + { absolutePath: rootPath }, + () => {}, + ); + const releaseUnsubscribe = fake.holdUnsubscribes(); + await unsubscribe(); + await sleep(80); + + await manager.subscribe({ absolutePath: rootPath }, () => {}); + releaseUnsubscribe(); + + await waitUntil( + () => fake.subscriptions[0]?.disposed === true, + "old teardown", + ); + expect(fake.subscriptions).toHaveLength(2); + expect(fake.live(rootPath)).toHaveLength(1); + }); + + it("releases a watcher whose root was deleted instead of keeping it idle", async () => { + const fake = createFakeBackend(); + const rootPath = await createTempRoot(); + const manager = createManager(fake.backend, { recoveryPollMs: 60_000 }); + + const unsubscribe = await manager.subscribe( + { absolutePath: rootPath }, + () => {}, + ); + fake.subscriptions[0]?.request.onEvents([ + { type: "delete", path: rootPath }, + ]); + await waitUntil(() => fake.live(rootPath).length === 0, "suspension"); + await unsubscribe(); + + await manager.subscribe({ absolutePath: rootPath }, () => {}); + expect(fake.subscriptions).toHaveLength(2); + expect(fake.live(rootPath)).toHaveLength(1); + }); + + it("attaches afresh when the idle watcher's root was deleted while idle", async () => { + const fake = createFakeBackend(); + const rootPath = await createTempRoot(); + const manager = createManager(fake.backend, { + idleKeepAliveMs: 60_000, + recoveryPollMs: 60_000, + }); + + const unsubscribe = await manager.subscribe( + { absolutePath: rootPath }, + () => {}, + ); + await unsubscribe(); + fake.subscriptions[0]?.request.onEvents([ + { type: "delete", path: rootPath }, + ]); + await waitUntil(() => fake.live(rootPath).length === 0, "suspension"); + + await manager.subscribe({ absolutePath: rootPath }, () => {}); + expect(fake.subscriptions).toHaveLength(2); + expect(fake.live(rootPath)).toHaveLength(1); + }); + + it("attaches afresh when the root was replaced while idle and no delete event arrived", async () => { + const fake = createFakeBackend(); + const rootPath = await createTempRoot(); + const manager = createManager(fake.backend, { idleKeepAliveMs: 60_000 }); + + const unsubscribe = await manager.subscribe( + { absolutePath: rootPath }, + () => {}, + ); + await unsubscribe(); + await fs.rm(rootPath, { recursive: true, force: true }); + await fs.mkdir(rootPath); + + await manager.subscribe({ absolutePath: rootPath }, () => {}); + expect(fake.subscriptions).toHaveLength(2); + expect(fake.live(rootPath)).toHaveLength(1); + }); + + it("ignores a repeated unsubscribe so it cannot extend the grace period", async () => { + const fake = createFakeBackend(); + const rootPath = await createTempRoot(); + const manager = createManager(fake.backend); + + const unsubscribe = await manager.subscribe( + { absolutePath: rootPath }, + () => {}, + ); + await unsubscribe(); + await sleep(60); + await unsubscribe(); + + await sleep(70); + expect(fake.live(rootPath)).toHaveLength(0); + }); +}); diff --git a/packages/workspace-fs/src/watch-overflow-backoff.test.ts b/packages/workspace-fs/src/watch-overflow-backoff.test.ts index e6d51eaa2d2..3cd20ab03e8 100644 --- a/packages/workspace-fs/src/watch-overflow-backoff.test.ts +++ b/packages/workspace-fs/src/watch-overflow-backoff.test.ts @@ -275,6 +275,7 @@ describe("FSEvents overflow rescan backoff", () => { overflowRescanInitialMs: 50, overflowRescanMaxMs: 200, overflowBackoffResetMs: 10_000, + idleKeepAliveMs: 0, }); const events: FsWatchEvent[] = []; const unsubscribe = await manager.subscribe( diff --git a/packages/workspace-fs/src/watch.ts b/packages/workspace-fs/src/watch.ts index bb52ec4b2b5..713ad691167 100644 --- a/packages/workspace-fs/src/watch.ts +++ b/packages/workspace-fs/src/watch.ts @@ -65,6 +65,11 @@ const OVERFLOW_BACKOFF_RESET_MS = 120_000; const PROBE_PREFIX = ".superset-watcher-probe-"; const PROBE_TIMEOUT_MS = 4_000; +// Re-attaching a large worktree is a full crawl, and switching workspaces +// releases one; the cap keeps idle cost flat as worktree count grows. +const IDLE_KEEP_ALIVE_MS = 60_000; +const MAX_IDLE_WATCHERS = 3; + // Backslash-escape every character picomatch (parcel's glob engine) treats as // magic, so an absolute path is matched literally when embedded in a glob. // Mirrors the metacharacter set `is-glob`/picomatch@2 recognize. @@ -153,6 +158,12 @@ interface WatcherState { realPath: string; realPathNormalized: string; realPathDiffers: boolean; + /** + * The attached root directory's identity. A root deleted and recreated + * while nobody listens can go unreported, leaving a live-looking + * subscription on a dead tree; reuse compares against this first. + */ + rootIdentity: string; /** Null while suspended (root deleted, polling for recreation). */ subscription: NativeWatchSubscription | null; recoveryTimer: ReturnType | null; @@ -179,6 +190,8 @@ interface WatcherState { overflowsCoalesced: number; lastOverflowAt: number; listeners: Set; + /** Disposes the watcher once it has sat without listeners for the keep-alive. */ + idleTimer: ReturnType | null; /** * Root-relative directories pruned from the native subscription beyond the * static defaults (nested repos + gitignored dirs), kept queryable so @@ -217,6 +230,11 @@ async function unsubscribeQuietly( ]); } +async function readRootIdentity(rootPath: string): Promise { + const stats = await stat(rootPath); + return `${stats.dev}:${stats.ino}:${stats.birthtimeMs}`; +} + function internalToFsWatchEvent(event: InternalWatchEvent): FsWatchEvent { return { kind: event.kind, @@ -270,6 +288,10 @@ export interface FsWatcherManagerOptions { overflowRescanInitialMs?: number; overflowRescanMaxMs?: number; overflowBackoffResetMs?: number; + /** How long a watcher with no listeners stays attached; 0 disposes at once. */ + idleKeepAliveMs?: number; + /** Most listener-less watchers kept attached; the longest-idle goes first. */ + maxIdleWatchers?: number; } export class FsWatcherManager { @@ -292,7 +314,11 @@ export class FsWatcherManager { private readonly overflowRescanInitialMs: number; private readonly overflowRescanMaxMs: number; private readonly overflowBackoffResetMs: number; + private readonly idleKeepAliveMs: number; + private readonly maxIdleWatchers: number; private readonly watchers = new Map(); + /** Listener-less watchers still attached, oldest-idle first. */ + private readonly idleWatchers = new Set(); /** * One-shot dedup so a single ENOSPC report doesn't spam logs across every * watcher creation that follows it. Mirrors VS Code's `enospcErrorLogged` @@ -334,6 +360,8 @@ export class FsWatcherManager { options.overflowRescanMaxMs ?? OVERFLOW_RESCAN_MAX_MS; this.overflowBackoffResetMs = options.overflowBackoffResetMs ?? OVERFLOW_BACKOFF_RESET_MS; + this.idleKeepAliveMs = options.idleKeepAliveMs ?? IDLE_KEEP_ALIVE_MS; + this.maxIdleWatchers = options.maxIdleWatchers ?? MAX_IDLE_WATCHERS; } async subscribe( @@ -342,6 +370,14 @@ export class FsWatcherManager { ): Promise<() => Promise> { const absolutePath = normalizeAbsolutePath(options.absolutePath); options.signal?.throwIfAborted(); + const idle = this.watchers.get(absolutePath); + if (idle?.listeners.size === 0 && !(await this.isStillAttached(idle))) { + // A concurrent subscribe may have claimed it during the stat. + if (idle.listeners.size === 0) { + await this.releaseWatcher(idle); + } + } + options.signal?.throwIfAborted(); let state = this.watchers.get(absolutePath); let pending = this.pending.get(absolutePath); if (!state && !pending) { @@ -368,6 +404,7 @@ export class FsWatcherManager { pending = { controller, listeners, promise }; this.pending.set(absolutePath, pending); } + if (state) this.markActive(state); const listeners = state?.listeners ?? pending?.listeners; if (!listeners) throw new Error("Missing watcher subscription state"); listeners.add(listener); @@ -383,10 +420,16 @@ export class FsWatcherManager { pending.controller.abort(); } const current = this.watchers.get(absolutePath); - if (current?.listeners === listeners) { - this.watchers.delete(absolutePath); - await this.disposeWatcherState(current); + if (current?.listeners !== listeners) return; + if ( + this.idleKeepAliveMs <= 0 || + this.maxIdleWatchers <= 0 || + current.subscription === null + ) { + await this.releaseWatcher(current); + return; } + await this.markIdle(current); }; const onAbort = () => { void release(); @@ -402,6 +445,17 @@ export class FsWatcherManager { } } + private async isStillAttached(state: WatcherState): Promise { + if (state.subscription === null) { + return false; + } + try { + return (await readRootIdentity(state.realPath)) === state.rootIdentity; + } catch { + return false; + } + } + async close(): Promise { const pending = [...this.pending.values()]; this.pending.clear(); @@ -409,11 +463,48 @@ export class FsWatcherManager { await Promise.allSettled(pending.map((entry) => entry.promise)); const states = Array.from(this.watchers.values()); this.watchers.clear(); + this.idleWatchers.clear(); await Promise.all(states.map((state) => this.disposeWatcherState(state))); } + private markActive(state: WatcherState): void { + if (state.idleTimer) { + clearTimeout(state.idleTimer); + state.idleTimer = null; + } + this.idleWatchers.delete(state); + } + + private async markIdle(state: WatcherState): Promise { + state.idleTimer = setTimeout(() => { + state.idleTimer = null; + void this.releaseWatcher(state); + }, this.idleKeepAliveMs); + state.idleTimer.unref?.(); + this.idleWatchers.add(state); + + const excess = this.idleWatchers.size - this.maxIdleWatchers; + const evicted = Array.from(this.idleWatchers).slice(0, Math.max(0, excess)); + await Promise.all(evicted.map((idle) => this.releaseWatcher(idle))); + } + + private async releaseWatcher(state: WatcherState): Promise { + if (this.watchers.get(state.absolutePath) !== state) { + return; + } + // Remove from the map before touching the native layer so a fresh + // subscribe can never reuse a state whose teardown is in flight. + this.watchers.delete(state.absolutePath); + this.markActive(state); + await this.disposeWatcherState(state); + } + private async disposeWatcherState(state: WatcherState): Promise { state.controller.abort(); + if (state.idleTimer) { + clearTimeout(state.idleTimer); + state.idleTimer = null; + } if (state.flushTimer) { clearTimeout(state.flushTimer); state.flushTimer = null; @@ -583,6 +674,7 @@ export class FsWatcherManager { realPath: normalizedPath, realPathNormalized: normalizedPath, realPathDiffers: false, + rootIdentity: "", subscription: null, recoveryTimer: null, recovering: false, @@ -596,6 +688,7 @@ export class FsWatcherManager { overflowsCoalesced: 0, lastOverflowAt: 0, listeners: new Set(), + idleTimer: null, prunedRelPrefixes: [], filePaths: new Map(), directoryPaths: new Set(), @@ -635,6 +728,7 @@ export class FsWatcherManager { state.realPath = realPath; state.realPathNormalized = realPathNormalized; state.realPathDiffers = realPathDiffers; + state.rootIdentity = await readRootIdentity(realPath); const generation = ++state.generation; // Nested git repos/worktrees (agent tools pile these up — a full repo copy