diff --git a/src/__mocks__/obsidian.ts b/src/__mocks__/obsidian.ts index f5ec0e6a..d65f060d 100644 --- a/src/__mocks__/obsidian.ts +++ b/src/__mocks__/obsidian.ts @@ -395,6 +395,18 @@ export class Component { unload(): void { this.loaded = false; this.children.forEach((child) => child.unload()); + // Drain registered event refs by calling their unload() — this matches + // real Obsidian Component behavior so that listener-leak tests work. + // Without this, code that correctly uses registerEvent() still appears + // to leak in jest because the mock never calls eventRef.unload(). + for (const ref of this._events) { + try { + ref.unload(); + } catch { + /* listener cleanup must not throw */ + } + } + this._events = []; this.onunload(); } diff --git a/src/__tests__/integration/Lifecycle.unload.test.ts b/src/__tests__/integration/Lifecycle.unload.test.ts new file mode 100644 index 00000000..9105e722 --- /dev/null +++ b/src/__tests__/integration/Lifecycle.unload.test.ts @@ -0,0 +1,98 @@ +/** + * Phase 0 W2 — verify the dataflow orchestrator cleans up cleanly. + * + * Background: prior to W2, src/index.ts:1758 fired `dataflowOrchestrator.cleanup()` + * without awaiting it, so workers / event refs / debounced timers could outlive + * Obsidian's onunload contract. The fix exposes `plugin.unloadComplete` as a + * promise that resolves once async cleanup work has settled. + * + * This test boots a real orchestrator via the W0 fixture, simulates teardown, + * and asserts the event-bus listener counts on the fake app are zero. We don't + * test the plugin class directly (that requires booting the entire plugin under + * jsdom which the test infrastructure isn't set up for), but the orchestrator + * is the only async work in onunload — its cleanup is what unloadComplete + * actually awaits. + */ + +import { buildOrchestrator } from "./_fixtures/buildOrchestrator"; + +describe("Orchestrator lifecycle (W2)", () => { + it("registers workspace listeners and removes them on cleanup", async () => { + const fx = await buildOrchestrator({ + files: { "a.md": "- [ ] one", "b.md": "- [x] two" }, + }); + + // After construction the orchestrator subscribes to workspace events + // for things like ICS updates and write operations. The exact count is + // an implementation detail; just assert it's > 0 so the test would + // catch a regression where listeners stop attaching at all (which would + // silently break the dataflow). + const beforeCleanup = fx.app.__totalListenerCount(); + expect(beforeCleanup).toBeGreaterThan(0); + + await fx.dispose(); + + // After dispose, every listener attached via the orchestrator's + // eventRefs (which cleanup walks via app.workspace.offref) must be gone. + // Vault listeners attached by sub-sources should also be gone. + const afterCleanup = fx.app.__totalListenerCount(); + expect(afterCleanup).toBe(0); + }); + + it("dispose is idempotent and does not throw on a re-dispose", async () => { + const fx = await buildOrchestrator({ files: { "a.md": "- [ ] one" } }); + await fx.dispose(); + // Second dispose path: orchestrator was already nulled out, so the + // fixture's stored reference still points to the old instance. Calling + // cleanup() on it again should not throw. + await expect( + (fx.orchestrator as any).cleanup(), + ).resolves.not.toThrow(); + }); + + it("file events fire to listeners before dispose, but not after", async () => { + const fx = await buildOrchestrator({ + files: { "a.md": "- [ ] one" }, + }); + + const events: string[] = []; + fx.vault.on("modify", (file: any) => { + events.push(`modify:${file.path}`); + }); + + await fx.writeFile("a.md", "- [x] one"); + expect(events).toContain("modify:a.md"); + + await fx.dispose(); + + // Note: the W0 fixture's vault still has its own listener bus. The + // "modify" listener we attached in this test is on the fake vault, + // not on the orchestrator's eventRefs, so it's still active here. + // What we care about is that the orchestrator-level listeners are gone + // (covered by the first test) and that the orchestrator no longer + // processes new vault events that arrive post-cleanup. + events.length = 0; + await fx.writeFile("a.md", "- [ ] one"); + // Test-local listener still fires, that's fine. + expect(events).toEqual(["modify:a.md"]); + }); + + it("cleanup awaits Repository.cleanup (the async cost in onunload)", async () => { + const fx = await buildOrchestrator(); + // Spy on the repository cleanup method (accessed via the QueryAPI's + // repository) to confirm it's awaited. + const repo = (fx.orchestrator as any).repository; + expect(repo).toBeDefined(); + const repoCleanupSpy = jest.spyOn(repo, "cleanup"); + + await fx.dispose(); + + expect(repoCleanupSpy).toHaveBeenCalledTimes(1); + // And the spy must have been called and resolved BEFORE dispose returned. + // Jest's awaited spies satisfy that just by virtue of dispose having + // returned. If repo.cleanup() were fire-and-forget the spy would still + // be in a pending state when dispose returned, but Jest can't observe + // that distinction directly — the orchestrator's own `await` enforces it. + repoCleanupSpy.mockRestore(); + }); +}); diff --git a/src/__tests__/integration/Rebuild.failure.test.ts b/src/__tests__/integration/Rebuild.failure.test.ts new file mode 100644 index 00000000..955d6ece --- /dev/null +++ b/src/__tests__/integration/Rebuild.failure.test.ts @@ -0,0 +1,93 @@ +/** + * Phase 0 W2-bis — verify that a failed rebuild doesn't leave a partial cache. + * + * The fix in src/dataflow/Orchestrator.ts wraps rebuild() in try/catch. On any + * thrown error, every cache namespace is cleared so the next plugin load + * triggers a clean rebuild from disk instead of trusting a partial cache. + * + * The test injects a failure into processBatch (the most realistic failure + * point) by spying on the orchestrator method, then asserts: + * - rebuild() rejects with the original error + * - storage.clearNamespace was called for all 4 namespaces + */ + +import { buildOrchestrator } from "./_fixtures/buildOrchestrator"; + +describe("Orchestrator.rebuild failure handling (W2-bis)", () => { + it("clears all cache namespaces on rebuild failure and re-throws", async () => { + const fx = await buildOrchestrator({ + files: { "a.md": "- [ ] one", "b.md": "- [x] two" }, + }); + + // Reach into the orchestrator's storage to spy on namespace clears. + const storage = (fx.orchestrator as any).storage; + expect(storage).toBeDefined(); + const clearSpy = jest.spyOn(storage, "clearNamespace"); + + // Inject a failure: replace processBatch with a thrower. This is the + // most realistic failure mode (a worker error, a vault read error, + // or a parse error mid-batch). + const boom = new Error("simulated worker crash"); + (fx.orchestrator as any).processBatch = jest.fn(async () => { + throw boom; + }); + + await expect(fx.orchestrator.rebuild()).rejects.toBe(boom); + + // All four namespaces must have been cleared as the last-resort path. + const clearedNamespaces = clearSpy.mock.calls.map( + (call: any[]) => call[0], + ); + expect(clearedNamespaces).toEqual( + expect.arrayContaining(["raw", "augmented", "project", "consolidated"]), + ); + + clearSpy.mockRestore(); + await fx.dispose(); + }); + + it("does not clear namespaces on a successful rebuild", async () => { + const fx = await buildOrchestrator({ + files: { "a.md": "- [ ] one" }, + }); + + const storage = (fx.orchestrator as any).storage; + const clearSpy = jest.spyOn(storage, "clearNamespace"); + + // processBatch is the slow path; replace with a no-op so rebuild + // completes quickly without exercising worker plumbing. + (fx.orchestrator as any).processBatch = jest.fn(async () => {}); + + await expect(fx.orchestrator.rebuild()).resolves.toBeUndefined(); + + // On success, the catch branch should NOT have run, so clearNamespace + // should not have been called from rebuild's catch handler. Note: if + // repository.clear() internally calls clearNamespace, this assertion + // would fail; our implementation calls repository.clear() (a different + // path), not storage.clearNamespace, so this is safe. + expect(clearSpy).not.toHaveBeenCalled(); + + clearSpy.mockRestore(); + await fx.dispose(); + }); + + it("re-throws even when last-resort cleanup itself fails", async () => { + const fx = await buildOrchestrator({ files: { "a.md": "- [ ] one" } }); + + const storage = (fx.orchestrator as any).storage; + // Make every clearNamespace call fail. The catch-of-catch path should + // still re-throw the original error. + jest.spyOn(storage, "clearNamespace").mockRejectedValue( + new Error("storage is on fire"), + ); + + const boom = new Error("primary failure"); + (fx.orchestrator as any).processBatch = jest.fn(async () => { + throw boom; + }); + + await expect(fx.orchestrator.rebuild()).rejects.toBe(boom); + + await fx.dispose(); + }); +}); diff --git a/src/__tests__/integration/WorkerTimeout.test.ts b/src/__tests__/integration/WorkerTimeout.test.ts new file mode 100644 index 00000000..c531ad1f --- /dev/null +++ b/src/__tests__/integration/WorkerTimeout.test.ts @@ -0,0 +1,178 @@ +/** + * Phase 0 W3 — verify TaskWorkerManager per-task timeout fires, terminates the + * hung worker, and rejects with WorkerTimeoutError; verify WorkerOrchestrator + * counts the timeout in its metrics and falls back to main thread. + * + * Two test groups: + * 1. TaskWorkerManager: monkey-patch the underlying worker.worker.postMessage + * to a no-op so the worker never responds, then drive the per-task timeout + * with a short workerTimeoutMs and assert the rejection shape. + * 2. WorkerOrchestrator: stub a TaskWorkerManager whose processFile rejects + * with WorkerTimeoutError, assert metrics.taskWorkerTimeouts increments + * and the main-thread fallback runs. + */ + +import { TaskWorkerManager } from "@/dataflow/workers/TaskWorkerManager"; +import { WorkerOrchestrator } from "@/dataflow/workers/WorkerOrchestrator"; +import { WorkerTimeoutError } from "@/dataflow/workers/errors"; + +// Build a minimal TFile-shaped object — TaskWorkerManager only reads .path, +// .extension, and .stat from it during processFile. +function fakeFile(path: string, content = "- [ ] task") { + return { + path, + basename: path.replace(/\.md$/, ""), + extension: "md", + name: path.split("/").pop() ?? path, + parent: null, + stat: { mtime: 1, ctime: 1, size: content.length }, + } as any; +} + +describe("TaskWorkerManager per-task timeout (W3)", () => { + it("rejects active task with WorkerTimeoutError when worker hangs", async () => { + // Minimal vault/metadataCache mocks. Only methods TaskWorkerManager + // reaches into during processFile/getTaskMetadata are needed. + const vault: any = { + cachedRead: async () => "- [ ] task", + }; + const metadataCache: any = { + getFileCache: () => null, + }; + + // Construct manager with a 100ms timeout — long enough that the + // jest event loop can schedule it, short enough that the test stays fast. + const mgr = new TaskWorkerManager(vault, metadataCache, { + maxWorkers: 1, + cpuUtilization: 1, + workerTimeoutMs: 100, + }); + + // Force-spawn one worker by reaching into the private state, then + // monkey-patch its underlying worker.postMessage to do nothing so the + // worker never responds. (Workers are constructed eagerly in the + // constructor via initializeWorkers().) + const workersMap: Map = (mgr as any).workers; + expect(workersMap.size).toBeGreaterThan(0); + for (const w of workersMap.values()) { + w.worker.postMessage = jest.fn(); // black hole + } + + const file = fakeFile("hang.md"); + const result = mgr.processFile(file).then( + (value) => ({ kind: "resolved", value }), + (error) => ({ kind: "rejected", error }), + ); + + // Wait long enough for the timeout to fire (100ms + slack) + await new Promise((resolve) => setTimeout(resolve, 200)); + const settled = await result; + + expect(settled.kind).toBe("rejected"); + const err = (settled as any).error; + expect(err).toBeInstanceOf(WorkerTimeoutError); + expect(err.filePath).toBe("hang.md"); + expect(err.timeoutMs).toBe(100); + + // Timeout was counted on the manager. + expect(mgr.getTimeoutCount()).toBe(1); + + // Cleanup + (mgr as any).onunload(); + }); + + it("spawns a replacement worker after a timeout so the pool stays warm", async () => { + const vault: any = { + cachedRead: async () => "- [ ] task", + }; + const metadataCache: any = { + getFileCache: () => null, + }; + + const mgr = new TaskWorkerManager(vault, metadataCache, { + maxWorkers: 1, + cpuUtilization: 1, + workerTimeoutMs: 80, + }); + + const workersMap: Map = (mgr as any).workers; + const initialIds = new Set([...workersMap.keys()]); + for (const w of workersMap.values()) { + w.worker.postMessage = jest.fn(); + } + + await mgr + .processFile(fakeFile("hang.md")) + .catch(() => undefined); // we expect rejection + + // Give the timeout time to fire and replacement to spawn + await new Promise((resolve) => setTimeout(resolve, 200)); + + // After the timeout: the pool size should be back to maxWorkers (1) + // and the worker IDs should NOT all match the initial set (one was + // replaced). The timeout counter should be exactly 1. + expect(workersMap.size).toBe(1); + expect(mgr.getTimeoutCount()).toBe(1); + const finalIds = new Set([...workersMap.keys()]); + const allReplaced = [...finalIds].every((id) => !initialIds.has(id)); + expect(allReplaced).toBe(true); + + (mgr as any).onunload(); + }); +}); + +describe("WorkerOrchestrator counts WorkerTimeoutError separately (W3)", () => { + it("increments metrics.taskWorkerTimeouts and falls back to main thread", async () => { + // Stub a TaskWorkerManager whose processFile always rejects with the + // timeout error. The orchestrator should catch it, bump the counter, + // and fall through to parseFileTasksMainThread. + const stubTaskMgr: any = { + processFile: jest.fn(async (file: any) => { + throw new WorkerTimeoutError(file.path, 100); + }), + processBatch: jest.fn(), + isProcessingBatchTask: () => false, + getPendingTaskCount: () => 0, + getBatchProgress: () => ({ current: 0, total: 0, percentage: 0 }), + getStats: () => ({}), + }; + const stubProjectMgr: any = { + getProjectData: jest.fn(), + getBatchProjectData: jest.fn(), + isWorkersEnabled: () => true, + getMemoryStats: () => ({}), + }; + + const orch = new WorkerOrchestrator(stubTaskMgr, stubProjectMgr, { + enableWorkerProcessing: true, + }); + + // Patch the orchestrator's main-thread fallback so the test doesn't + // need a real ConfigurableTaskParser. We just need to confirm it's + // called and returns something. + const fallbackResult: any[] = [{ id: "fallback", content: "ok" }]; + (orch as any).parseFileTasksMainThread = jest.fn(async () => { + return fallbackResult; + }); + + const file = { + path: "hang.md", + extension: "md", + stat: { mtime: 1, ctime: 1, size: 1 }, + } as any; + const result = await orch.parseFileTasks(file); + + // Fallback was reached + expect(result).toBe(fallbackResult); + expect( + (orch as any).parseFileTasksMainThread, + ).toHaveBeenCalledTimes(1); + + // Metric was incremented + const metrics = orch.getMetrics(); + expect(metrics.taskWorkerTimeouts).toBe(1); + // And the generic failure count went up too — timeouts are a subset + // of failures, not a replacement for them. + expect(metrics.taskParsingFailures).toBeGreaterThanOrEqual(1); + }); +}); diff --git a/src/__tests__/integration/_fixtures/buildOrchestrator.ts b/src/__tests__/integration/_fixtures/buildOrchestrator.ts index 4133b452..dad242f3 100644 --- a/src/__tests__/integration/_fixtures/buildOrchestrator.ts +++ b/src/__tests__/integration/_fixtures/buildOrchestrator.ts @@ -86,7 +86,17 @@ class EventBus { on(name: string, handler: (...args: any[]) => void) { if (!this.handlers.has(name)) this.handlers.set(name, new Set()); this.handlers.get(name)!.add(handler); - return { name, handler } as any; + // Return an EventRef-shaped object. Both `name`/`handler` (used by + // our offref()) AND a working `unload()` method (used by the obsidian + // mock's Component.unload to drain registered events) must be present. + const self = this; + return { + name, + handler, + unload() { + self.off(name, handler); + }, + } as any; } off(name: string, handler: (...args: any[]) => void) { diff --git a/src/dataflow/Orchestrator.ts b/src/dataflow/Orchestrator.ts index 0caca624..7f59f7b9 100644 --- a/src/dataflow/Orchestrator.ts +++ b/src/dataflow/Orchestrator.ts @@ -1699,36 +1699,67 @@ export class DataflowOrchestrator { } /** - * Clear all data and rebuild + * Clear all data and rebuild. + * + * If rebuild fails partway through (e.g. a worker crash, vault read error, + * or persistence failure), we don't want to leave the cache in a partial + * state — the next plugin load would think it's loading a complete index + * but actually be missing files. As a last-resort, on any thrown error we + * clear every cache namespace so the next load triggers a clean rebuild + * from disk. The error is re-thrown so the caller can react. + * + * This is W2-bis in the v10 Phase 0 plan. It's intentionally minimal — + * snapshot/restore is deferred to a future phase. */ async rebuild(): Promise { - // Clear all data - await this.repository.clear(); + try { + // Clear all data + await this.repository.clear(); - // Process all markdown and canvas files - const files = this.vault.getMarkdownFiles(); - const canvasFiles = this.vault - .getFiles() - .filter((f) => f.extension === "canvas"); + // Process all markdown and canvas files + const files = this.vault.getMarkdownFiles(); + const canvasFiles = this.vault + .getFiles() + .filter((f) => f.extension === "canvas"); - const allFiles = [...files, ...canvasFiles]; + const allFiles = [...files, ...canvasFiles]; - // Process in batches for performance - const BATCH_SIZE = 50; - for (let i = 0; i < allFiles.length; i += BATCH_SIZE) { - const batch = allFiles.slice(i, i + BATCH_SIZE); - await this.processBatch(batch); + // Process in batches for performance + const BATCH_SIZE = 50; + for (let i = 0; i < allFiles.length; i += BATCH_SIZE) { + const batch = allFiles.slice(i, i + BATCH_SIZE); + await this.processBatch(batch); + } + + // Persist the rebuilt index + await this.repository.persist(); + + // Emit ready event + emit(this.app, Events.CACHE_READY, { + initial: false, + timestamp: Date.now(), + seq: Seq.next(), + }); + } catch (err) { + console.error( + "[DataflowOrchestrator] Rebuild failed; clearing caches to force a clean rebuild on next load:", + err, + ); + // Last-resort cleanup: clear every namespace so the next plugin load + // can't read a partial cache that looks complete. + try { + await this.storage.clearNamespace("raw"); + await this.storage.clearNamespace("augmented"); + await this.storage.clearNamespace("project"); + await this.storage.clearNamespace("consolidated"); + } catch (clearErr) { + console.error( + "[DataflowOrchestrator] Last-resort cache clear ALSO failed; cache may be in inconsistent state:", + clearErr, + ); + } + throw err; } - - // Persist the rebuilt index - await this.repository.persist(); - - // Emit ready event - emit(this.app, Events.CACHE_READY, { - initial: false, - timestamp: Date.now(), - seq: Seq.next(), - }); } /** diff --git a/src/dataflow/indexer/Repository.ts b/src/dataflow/indexer/Repository.ts index ed9ab3d6..42ba6b98 100644 --- a/src/dataflow/indexer/Repository.ts +++ b/src/dataflow/indexer/Repository.ts @@ -562,11 +562,27 @@ export class Repository { } /** - * Cleanup and ensure all pending data is persisted + * Cleanup and ensure all pending data is persisted. + * + * IMPORTANT: also unloads the underlying TaskIndexer (a Component) so its + * vault listeners (modify/delete/create) are deregistered. Prior to this, + * those listeners leaked across plugin reloads — TaskIndexer was constructed + * eagerly in the Repository constructor but never explicitly unloaded. + * See W2 in the v10 Phase 0 plan. */ async cleanup(): Promise { // Execute any pending persist operations await this.executePersist(); + + // Tear down the indexer's Component lifecycle so its registered events + // (vault.on modify/delete/create in TaskIndexer.setupEventListeners) + // are released. The Component base class' unload() walks _events and + // calls offref on each. + try { + (this.indexer as any).unload?.(); + } catch (error) { + console.error("[Repository] Error unloading TaskIndexer:", error); + } } /** diff --git a/src/dataflow/workers/TaskWorkerManager.ts b/src/dataflow/workers/TaskWorkerManager.ts index 9ad47d2d..70f2f7c1 100644 --- a/src/dataflow/workers/TaskWorkerManager.ts +++ b/src/dataflow/workers/TaskWorkerManager.ts @@ -28,6 +28,7 @@ import { // @ts-ignore Ignore type error for worker import import TaskWorker from "./TaskIndex.worker"; import { Deferred, deferred } from "./deferred-promise"; +import { WorkerTimeoutError } from "./errors"; // Using similar queue structure as importer.ts import { Queue } from "@datastructures-js/queue"; @@ -42,6 +43,14 @@ export interface WorkerPoolOptions { cpuUtilization: number; /** Whether to enable debug logging */ debug?: boolean; + /** + * Per-task timeout in milliseconds. If a worker doesn't post a response + * within this window the worker is terminated, the active task's promise + * is rejected with WorkerTimeoutError, and a replacement worker is spawned. + * Default: 8000 (8s) — enough headroom for a worst-case 200KB markdown + * parse on a slow desktop. See W3 in v10 Phase 0 plan. + */ + workerTimeoutMs?: number; /** Settings for the task indexer */ settings?: { preferMetadataFormat: "dataview" | "tasks"; @@ -62,6 +71,14 @@ export interface WorkerPoolOptions { }; } +/** + * Default per-task timeout in ms for the task worker pool. A debug override + * can be supplied via `(globalThis as any).__taskGeniusDebug?.workerTimeoutMs` + * which the constructor reads at boot — used by integration tests to shrink + * the timeout to a couple hundred ms. + */ +export const DEFAULT_TASK_WORKER_TIMEOUT_MS = 8000; + /** * Default worker pool options */ @@ -103,6 +120,12 @@ interface PoolWorker { availableAt: number; /** The active task this worker is processing, if any */ active?: [TFile, Deferred, number, TaskPriority]; + /** + * Per-task timeout handle. Set when a task is dispatched to this worker + * (in schedule()), cleared when finish() runs, fires onTaskTimeout() if + * the worker doesn't respond within workerTimeoutMs. See W3 in plan. + */ + timeoutHandle?: ReturnType; } /** @@ -169,6 +192,10 @@ export class TaskWorkerManager extends Component { private initialized: boolean = false; /** Reference to task indexer for cache checking */ private taskIndexer?: any; + /** Effective per-task timeout in ms (resolved from options + debug override) */ + private workerTimeoutMs: number; + /** Count of timed-out tasks. Exposed via getStats() and the orchestrator metrics. */ + private timeoutCount: number = 0; /** Performance statistics */ private stats = { filesSkipped: 0, @@ -189,10 +216,36 @@ export class TaskWorkerManager extends Component { this.vault = vault; this.metadataCache = metadataCache; + // Resolve effective per-task timeout. Precedence: + // 1. options.workerTimeoutMs (explicit) + // 2. debug override on globalThis (for tests) + // 3. DEFAULT_TASK_WORKER_TIMEOUT_MS + const debugOverride = + (globalThis as any).__taskGeniusDebug?.workerTimeoutMs; + this.workerTimeoutMs = + this.options.workerTimeoutMs ?? + (typeof debugOverride === "number" ? debugOverride : undefined) ?? + DEFAULT_TASK_WORKER_TIMEOUT_MS; + // Initialize workers up to max this.initializeWorkers(); } + /** + * Get the count of timed-out tasks since the last reset. + * Used by WorkerOrchestrator metrics. + */ + public getTimeoutCount(): number { + return this.timeoutCount; + } + + /** + * Reset the timeout counter (e.g. when metrics are reset). + */ + public resetTimeoutCount(): void { + this.timeoutCount = 0; + } + /** * Set file parsing configuration */ @@ -689,6 +742,14 @@ export class TaskWorkerManager extends Component { const {file, promise, priority} = queueItem; worker.active = [file, promise, 0, priority]; // 0 表示重试次数 + // Arm a per-task timeout. If the worker doesn't respond within + // workerTimeoutMs, onTaskTimeout will terminate it, reject the promise + // with WorkerTimeoutError, spawn a replacement, and continue scheduling. + // Cleared in finish() and on direct terminate(). + worker.timeoutHandle = setTimeout( + () => this.onTaskTimeout(worker), + this.workerTimeoutMs, + ); try { this.getTaskMetadata(file) @@ -722,6 +783,10 @@ export class TaskWorkerManager extends Component { }) .catch((error) => { console.error(`Error reading file ${file.path}:`, error); + if (worker.timeoutHandle) { + clearTimeout(worker.timeoutHandle); + worker.timeoutHandle = undefined; + } promise.reject(error); worker.active = undefined; @@ -733,6 +798,10 @@ export class TaskWorkerManager extends Component { }); } catch (error) { console.error(`Error processing file ${file.path}:`, error); + if (worker.timeoutHandle) { + clearTimeout(worker.timeoutHandle); + worker.timeoutHandle = undefined; + } promise.reject(error); worker.active = undefined; @@ -751,6 +820,14 @@ export class TaskWorkerManager extends Component { worker: PoolWorker, data: IndexerResult ): Promise { + // Cancel the timeout before doing any work — the worker did respond, + // so we no longer want to consider it hung. Even if the response is + // itself an error, the timeout path is no longer correct here. + if (worker.timeoutHandle) { + clearTimeout(worker.timeoutHandle); + worker.timeoutHandle = undefined; + } + if (!worker.active) { console.log("Received a stale worker message. Ignoring.", data); return; @@ -881,6 +958,13 @@ export class TaskWorkerManager extends Component { * Terminate a worker */ private terminate(worker: PoolWorker): void { + // Always clear the timeout — orphaned timers are exactly what plugin + // reload races complain about. + if (worker.timeoutHandle) { + clearTimeout(worker.timeoutHandle); + worker.timeoutHandle = undefined; + } + worker.worker.terminate(); if (worker.active) { @@ -891,6 +975,68 @@ export class TaskWorkerManager extends Component { this.log(`Terminated worker #${worker.id}`); } + /** + * Called when a worker doesn't respond within workerTimeoutMs. + * + * Strategy: kill the worker, reject the active promise with + * WorkerTimeoutError so the WorkerOrchestrator can fall back to main-thread + * processing for this file, then spawn a replacement worker so the pool + * stays warm. Without termination, a hung worker would hold its slot and + * starve the queue until the existing 30s circuit breaker tripped. + */ + private onTaskTimeout(worker: PoolWorker): void { + // Defensive: if the timeout fires after finish() already cleared the + // handle, do nothing. + if (!worker.active) return; + + const [file, promise] = worker.active; + console.warn( + `[TaskWorkerManager] Worker #${worker.id} timed out on ${file.path} after ${this.workerTimeoutMs}ms; terminating and replacing.`, + ); + + this.timeoutCount++; + + // Terminate the hung worker. Note: terminate() will also clear the + // handle (no-op here since we just fired) and reject worker.active[1] + // with "Terminated", but we want a more specific error, so we reject + // here first then null out worker.active before terminate runs. + try { + worker.worker.terminate(); + } catch (e) { + console.error("[TaskWorkerManager] Error terminating worker:", e); + } + + // Remove from pool + this.workers.delete(worker.id); + + // Reject with the typed error so the orchestrator can distinguish + // timeouts from other failures. + promise.reject(new WorkerTimeoutError(file.path, this.workerTimeoutMs)); + this.outstanding.delete(file.path); + worker.active = undefined; + worker.timeoutHandle = undefined; + + // Keep the pool warm: if we're below capacity (we just dropped one), + // spawn a replacement so subsequent tasks aren't starved. + if (this.active && this.workers.size < this.options.maxWorkers) { + try { + const replacement = this.newWorker(); + this.workers.set(replacement.id, replacement); + this.log( + `Spawned replacement worker #${replacement.id} after timeout`, + ); + } catch (e) { + console.error( + "[TaskWorkerManager] Failed to spawn replacement worker after timeout:", + e, + ); + } + } + + // Drain the queue with whatever workers remain. + this.schedule(); + } + /** * Clean up existing workers without affecting the active state */ diff --git a/src/dataflow/workers/WorkerOrchestrator.ts b/src/dataflow/workers/WorkerOrchestrator.ts index 8e3e5380..f5d536db 100644 --- a/src/dataflow/workers/WorkerOrchestrator.ts +++ b/src/dataflow/workers/WorkerOrchestrator.ts @@ -5,6 +5,7 @@ import { TaskWorkerManager, DEFAULT_WORKER_OPTIONS } from "./TaskWorkerManager"; import { ProjectDataWorkerManager } from "./ProjectDataWorkerManager"; import { MetadataParseMode } from "../../types/TaskParserConfig"; import { ConfigurableTaskParser } from "@/dataflow/core/ConfigurableTaskParser"; +import { WorkerTimeoutError } from "./errors"; /** * WorkerOrchestrator - Unified task and project worker management @@ -24,6 +25,9 @@ export class WorkerOrchestrator { private metrics = { taskParsingSuccess: 0, taskParsingFailures: 0, + // Subset of taskParsingFailures: hung-worker timeouts (W3). Tracked + // separately so we can tell "worker hung" from "worker errored". + taskWorkerTimeouts: 0, projectDataSuccess: 0, projectDataFailures: 0, averageTaskParsingTime: 0, @@ -87,8 +91,12 @@ export class WorkerOrchestrator { error ); - // Update failure metrics + // Update failure metrics. Track timeouts separately so observability + // can distinguish "worker hung" from "worker threw" (W3). this.metrics.taskParsingFailures++; + if (error instanceof WorkerTimeoutError) { + this.metrics.taskWorkerTimeouts++; + } this.handleWorkerFailure(); // Fallback to main thread @@ -132,8 +140,12 @@ export class WorkerOrchestrator { error ); - // Update failure metrics + // Update failure metrics. A timeout in a batch is counted once for + // the offending file, not for the whole batch. this.metrics.taskParsingFailures += files.length; + if (error instanceof WorkerTimeoutError) { + this.metrics.taskWorkerTimeouts++; + } this.handleWorkerFailure(); // Fallback to main thread @@ -220,7 +232,12 @@ export class WorkerOrchestrator { } /** - * Generic retry mechanism with exponential backoff + * Generic retry mechanism with exponential backoff. + * + * Skips retries for WorkerTimeoutError: a worker that just hung is being + * replaced anyway, and the file content is the most likely cause of the + * hang. Retrying would just waste 7s of exponential backoff before falling + * back to main thread, which is the right answer for hung tasks. (W3) */ private async retryOperation( operation: () => Promise, @@ -248,6 +265,11 @@ export class WorkerOrchestrator { error ); + // Don't retry timeout errors — see method docs. + if (error instanceof WorkerTimeoutError) { + break; + } + // If this is the last attempt, don't wait if (attempt === maxRetries) { break; @@ -525,6 +547,7 @@ export class WorkerOrchestrator { this.metrics = { taskParsingSuccess: 0, taskParsingFailures: 0, + taskWorkerTimeouts: 0, projectDataSuccess: 0, projectDataFailures: 0, averageTaskParsingTime: 0, diff --git a/src/dataflow/workers/errors.ts b/src/dataflow/workers/errors.ts new file mode 100644 index 00000000..438a643c --- /dev/null +++ b/src/dataflow/workers/errors.ts @@ -0,0 +1,24 @@ +/** + * Worker-related error types. + * + * Phase 0 W3: distinguish "worker hung past its timeout" from other worker + * failures so the WorkerOrchestrator can: + * - increment a dedicated metric (`workerTimeouts`) separate from generic + * parsing failures + * - decide whether the fallback to main thread is appropriate + * - allow tests to assert specifically on the timeout path + */ + +export class WorkerTimeoutError extends Error { + readonly filePath: string; + readonly timeoutMs: number; + + constructor(filePath: string, timeoutMs: number) { + super( + `Task worker timed out after ${timeoutMs}ms while processing ${filePath}`, + ); + this.name = "WorkerTimeoutError"; + this.filePath = filePath; + this.timeoutMs = timeoutMs; + } +} diff --git a/src/index.ts b/src/index.ts index b332bfae..b80b7bf6 100644 --- a/src/index.ts +++ b/src/index.ts @@ -159,6 +159,13 @@ export default class TaskProgressBarPlugin extends Plugin { // Write API for dataflow architecture writeAPI?: WriteAPI; + // Resolves once all async cleanup work scheduled by onunload() has settled. + // Obsidian's onunload() signature is sync (void), so async cleanup work has + // to be fired off without awaiting. Tests and any code that needs to know + // when the plugin is fully torn down should `await plugin.unloadComplete`. + // Reset on each onload(); see W2 in the v10 Phase 0 plan. + public unloadComplete: Promise = Promise.resolve(); + // Notification manager (desktop) notificationManager?: DesktopIntegrationManager; @@ -1746,30 +1753,49 @@ export default class TaskProgressBarPlugin extends Plugin { } onunload() { - // Clean up global suggest manager + // Synchronous cleanup paths run immediately. Asynchronous cleanup + // (currently just dataflowOrchestrator.cleanup() which awaits Repository + // persistence) is gathered into a single promise exposed as + // `unloadComplete` so tests and any external observer can await full + // teardown. Obsidian itself never awaits this — its onunload signature + // is sync — but at least listeners + workers + persistence get a chance + // to settle before the next plugin lifecycle, instead of racing. + const asyncTasks: Array> = []; + + // Clean up global suggest manager (sync) if (this.globalSuggestManager) { this.globalSuggestManager.cleanup(); } // Bases views are automatically unregistered by Obsidian when plugin unloads - // Clean up dataflow orchestrator (experimental) + // Clean up dataflow orchestrator (async — capture into asyncTasks) if (this.dataflowOrchestrator) { - this.dataflowOrchestrator.cleanup().catch((error) => { - console.error( - "Error cleaning up dataflow orchestrator:", - error, - ); - }); - // Set to undefined to prevent any further access + const orch = this.dataflowOrchestrator; + // Null out immediately so any other code path that fires during + // teardown can't reach into a half-cleaned-up orchestrator. this.dataflowOrchestrator = undefined; + asyncTasks.push( + orch.cleanup().catch((error) => { + console.error( + "Error cleaning up dataflow orchestrator:", + error, + ); + }), + ); } - // Clean up MCP server manager (desktop only) + // Clean up MCP server manager (desktop only, sync) if (this.mcpServerManager) { this.mcpServerManager.cleanup(); } + // Task Genius Icon Manager cleanup is handled automatically by Component system + + // Expose a promise so tests / external observers can know when async + // cleanup is fully done. Never rejects — individual catches above + // already log errors. + this.unloadComplete = Promise.all(asyncTasks).then(() => undefined); } /**