From 2f5fac7c84a16f826bcbe0ff6ad33010361adbd9 Mon Sep 17 00:00:00 2001 From: Ali Turki Date: Sat, 18 Jul 2026 17:09:58 +0800 Subject: [PATCH 1/3] fix(tabs): detach watchers superseded during attach A tab's watch() call could resolve after its path changed, at which point the id-based guard still passed (the id is wanted and the map holds a slot) and the real unwatch was assigned to an orphaned slot, leaking an FSEvents stream for the rest of the session. Compare slot identity instead: if the map no longer holds the exact slot this attach created, unwatch immediately. That covers path swaps, tab close, unmount, and rapid successive swaps with one check. The rejection path now deletes only its own slot, so a failed stale attach cannot evict the live watcher that replaced it. Fixes #25 --- src/hooks/useTabs.test.ts | 86 ++++++++++++++++++++++++++++++++++++++- src/hooks/useTabs.ts | 18 +++++--- 2 files changed, 97 insertions(+), 7 deletions(-) diff --git a/src/hooks/useTabs.test.ts b/src/hooks/useTabs.test.ts index d3d553e..2f275d9 100644 --- a/src/hooks/useTabs.test.ts +++ b/src/hooks/useTabs.test.ts @@ -5,12 +5,26 @@ import { useTabs } from "./useTabs"; type WatchCallback = (event: { type: unknown }) => void; const watchCallbacks: WatchCallback[] = []; +interface WatchStart { + path: string; + cb: WatchCallback; + unwatch: ReturnType; +} +const watchStarts: WatchStart[] = []; +// Path -> promise the mocked watch awaits before resolving, so a test can +// hold one attach open while the tab underneath it swaps path. +const watchGates = new Map>(); + vi.mock("@tauri-apps/plugin-fs", () => ({ readTextFile: vi.fn(), writeTextFile: vi.fn(), - watch: vi.fn(async (_path: string, cb: WatchCallback) => { + watch: vi.fn(async (path: string, cb: WatchCallback) => { + const unwatch = vi.fn(); + watchStarts.push({ path, cb, unwatch }); + const gate = watchGates.get(path); + if (gate) await gate; watchCallbacks.push(cb); - return () => {}; + return unwatch; }), })); @@ -52,6 +66,8 @@ function fireExternalModify() { beforeEach(() => { watchCallbacks.length = 0; + watchStarts.length = 0; + watchGates.clear(); vi.mocked(readTextFile).mockResolvedValue(RAW); vi.mocked(writeTextFile).mockReset(); vi.mocked(writeTextFile).mockResolvedValue(); @@ -137,6 +153,72 @@ describe("useTabs edit", () => { }); }); +describe("useTabs watchers", () => { + const PATH_A = "/ws/a.md"; + const PATH_B = "/ws/b.md"; + + function findStart(path: string) { + const start = watchStarts.find((w) => w.path === path); + expect(start).toBeDefined(); + return start!; + } + + // Opens PATH_A with its watch held open, swaps the tab to PATH_B while that + // attach is still in flight, then lets the PATH_A watch resolve late. + async function swapPathMidAttach() { + let openGateA!: () => void; + watchGates.set( + PATH_A, + new Promise((resolve) => { + openGateA = resolve; + }) + ); + const hook = renderHook(() => + useTabs({ autoReloadOnExternalChange: false, isManagedPath: () => false }) + ); + await waitFor(() => expect(hook.result.current.hydrated).toBe(true)); + + act(() => hook.result.current.openInNew(PATH_A)); + await waitFor(() => expect(watchStarts.some((w) => w.path === PATH_A)).toBe(true)); + + act(() => hook.result.current.openInActive(PATH_B)); + await waitFor(() => expect(watchStarts.some((w) => w.path === PATH_B)).toBe(true)); + + openGateA(); + return hook; + } + + it("detaches a watcher superseded by a path swap during attach", async () => { + await swapPathMidAttach(); + await waitFor(() => expect(findStart(PATH_A).unwatch).toHaveBeenCalled()); + }); + + it("keeps the new path's watcher attached and working after the swap", async () => { + const { result } = await swapPathMidAttach(); + await waitFor(() => expect(findStart(PATH_A).unwatch).toHaveBeenCalled()); + await waitFor(() => expect(result.current.activeTab?.loading).toBe(false)); + + expect(findStart(PATH_B).unwatch).not.toHaveBeenCalled(); + vi.mocked(readTextFile).mockResolvedValue(CHANGED_ON_DISK); + act(() => findStart(PATH_B).cb({ type: { modify: { kind: "data" } } })); + await waitFor(() => + expect(result.current.activeTab?.pendingContent).toBe(CHANGED_ON_DISK) + ); + }); + + it("attaches exactly one watcher for a tab whose path never changes", async () => { + const { result } = await openTab(); + expect(watchStarts.filter((w) => w.path === "/ws/doc.md")).toHaveLength(1); + expect(watchStarts[0].unwatch).not.toHaveBeenCalled(); + + fireExternalModify(); + await waitFor(() => + expect(result.current.activeTab?.pendingContent).toBe(CHANGED_ON_DISK) + ); + expect(watchStarts[0].unwatch).not.toHaveBeenCalled(); + }); +}); + describe("useTabs load timeout", () => { const LOAD_TIMEOUT_MS = 15000; const STALE_RAW = "---\ntitle: Note\n---\n\n# stale from load A\n"; diff --git a/src/hooks/useTabs.ts b/src/hooks/useTabs.ts index 7125238..23155c9 100644 --- a/src/hooks/useTabs.ts +++ b/src/hooks/useTabs.ts @@ -59,6 +59,11 @@ export interface Tabs { setScrollTop: (path: string, value: number) => void; } +interface WatcherSlot { + path: string; + unwatch: UnwatchFn; +} + let tabIdSeq = 0; const nextId = () => `t${++tabIdSeq}_${Date.now()}`; @@ -442,11 +447,10 @@ export function useTabs(options: UseTabsOptions): Tabs { ); }, [options.autoReloadOnExternalChange]); - const watchersRef = useRef(new Map()); + const watchersRef = useRef(new Map()); useEffect(() => { const watchers = watchersRef.current; - const wantedIds = new Set(tabs.map((t) => t.id)); for (const [id, entry] of watchers) { const tab = tabs.find((t) => t.id === id); @@ -458,7 +462,7 @@ export function useTabs(options: UseTabsOptions): Tabs { for (const tab of tabs) { if (watchers.has(tab.id)) continue; - const slot: { path: string; unwatch: UnwatchFn } = { path: tab.path, unwatch: () => {} }; + const slot: WatcherSlot = { path: tab.path, unwatch: () => {} }; watchers.set(tab.id, slot); void (async () => { try { @@ -480,13 +484,17 @@ export function useTabs(options: UseTabsOptions): Tabs { }, { recursive: false, delayMs: 400 } ); - if (!wantedIds.has(tab.id) || !watchers.has(tab.id)) { + // A path swap retires this slot and installs a fresh one under the + // same tab id, so the id alone cannot tell a live attach from a + // superseded one. Slot identity can: anything but our own slot in + // the map means this watcher is orphaned and must detach now. + if (watchers.get(tab.id) !== slot) { void unwatch(); return; } slot.unwatch = unwatch; } catch (err) { - watchers.delete(tab.id); + if (watchers.get(tab.id) === slot) watchers.delete(tab.id); console.error("watch failed", err); } })(); From 00ec6ed3f7d11458fb422334a5e9a7b9307f1d6f Mon Sep 17 00:00:00 2001 From: Ali Turki Date: Sat, 18 Jul 2026 17:09:58 +0800 Subject: [PATCH 2/3] fix(registry): make workspace registration atomic and serialized Concurrent MCP sidecars each read the registry, added their entry, and wrote the whole file back, so simultaneous registrations silently dropped each other. Measured with 48 concurrent sidecar processes: 41-45 of 48 workspaces survived before, 48 of 48 after. Hold an exclusive advisory lock across the load-modify-save sequence, and replace the in-place write with a temp file plus rename so a reader never observes a partial file. The lock lives on a sidecar path because the rename swaps the registry inode, which would leave writers holding locks on different inodes and never contending. Fixes #26 --- src-tauri/core/src/workspace/registry.rs | 115 ++++++++++++++++++++++- 1 file changed, 113 insertions(+), 2 deletions(-) diff --git a/src-tauri/core/src/workspace/registry.rs b/src-tauri/core/src/workspace/registry.rs index 5281be8..e48fc4d 100644 --- a/src-tauri/core/src/workspace/registry.rs +++ b/src-tauri/core/src/workspace/registry.rs @@ -1,4 +1,5 @@ use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; use serde::{Deserialize, Serialize}; @@ -8,6 +9,9 @@ use crate::error::{CoreError, ErrorCode}; pub const REGISTRY_DIR: &str = ".docsreader"; pub const REGISTRY_FILE: &str = "workspaces.json"; +const LOCK_SUFFIX: &str = ".lock"; +const TEMP_SUFFIX: &str = ".tmp"; + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct WorkspaceEntry { pub slug: String, @@ -40,6 +44,59 @@ pub fn load_registry(file: &Path) -> Result, CoreError> { Ok(parsed.workspaces) } +fn sibling_path(file: &Path, suffix: &str) -> PathBuf { + let mut name = file.file_name().unwrap_or_default().to_os_string(); + name.push(suffix); + file.with_file_name(name) +} + +/// Unique per writer so two concurrent saves cannot collide on the scratch file. +fn temp_path(file: &Path) -> PathBuf { + static COUNTER: AtomicU64 = AtomicU64::new(0); + let seq = COUNTER.fetch_add(1, Ordering::Relaxed); + sibling_path(file, &format!(".{}.{seq}{TEMP_SUFFIX}", std::process::id())) +} + +/// Rename instead of writing in place: a concurrent reader sees either the whole +/// old file or the whole new one, never a truncated one. +fn write_atomically(file: &Path, raw: &str) -> Result<(), CoreError> { + let temp = temp_path(file); + if let Err(e) = std::fs::write(&temp, raw) { + let _ = std::fs::remove_file(&temp); + return Err(e.into()); + } + if let Err(e) = std::fs::rename(&temp, file) { + let _ = std::fs::remove_file(&temp); + return Err(e.into()); + } + Ok(()) +} + +/// Advisory exclusive lock on a sidecar file, released when the handle drops. +/// The sidecar is never renamed, so every writer contends on the same inode. +struct RegistryLock(std::fs::File); + +impl RegistryLock { + fn acquire(file: &Path) -> Result { + if let Some(parent) = file.parent() { + std::fs::create_dir_all(parent)?; + } + let handle = std::fs::OpenOptions::new() + .create(true) + .truncate(false) + .write(true) + .open(sibling_path(file, LOCK_SUFFIX))?; + handle.lock()?; + Ok(Self(handle)) + } +} + +impl Drop for RegistryLock { + fn drop(&mut self) { + let _ = self.0.unlock(); + } +} + pub fn save_registry(file: &Path, workspaces: &[WorkspaceEntry]) -> Result<(), CoreError> { if let Some(parent) = file.parent() { std::fs::create_dir_all(parent)?; @@ -48,12 +105,14 @@ pub fn save_registry(file: &Path, workspaces: &[WorkspaceEntry]) -> Result<(), C workspaces: workspaces.to_vec(), }) .map_err(|e| CoreError::new(ErrorCode::Io, format!("serialize registry: {e}")))?; - std::fs::write(file, raw)?; - Ok(()) + write_atomically(file, &raw) } /// Replaces any entry with the same path, so re-registering updates slug/scope. pub fn upsert_workspace(file: &Path, entry: WorkspaceEntry) -> Result<(), CoreError> { + // Held across the whole load-modify-save: concurrent sidecars would otherwise + // each read the same registry and the last save would drop the other entries. + let _lock = RegistryLock::acquire(file)?; let mut workspaces = load_registry(file)?; workspaces.retain(|w| w.path != entry.path); workspaces.push(entry); @@ -124,6 +183,58 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + #[test] + fn concurrent_upserts_keep_every_entry() { + const WRITERS: usize = 8; + let dir = test_dir("reg_concurrent"); + let file = dir.join(REGISTRY_FILE); + std::thread::scope(|scope| { + for i in 0..WRITERS { + let file = file.clone(); + scope.spawn(move || { + upsert_workspace( + &file, + entry( + &format!("ws{i}"), + &format!("/repo/ws{i}"), + WorkspaceScope::Project, + ), + ) + .unwrap(); + }); + } + }); + + let mut slugs: Vec = load_registry(&file) + .unwrap() + .into_iter() + .map(|w| w.slug) + .collect(); + slugs.sort(); + let expected: Vec = (0..WRITERS).map(|i| format!("ws{i}")).collect(); + assert_eq!(slugs, expected); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn save_leaves_no_temp_files() { + let dir = test_dir("reg_notemp"); + let file = dir.join(REGISTRY_FILE); + save_registry( + &file, + &[entry("notes", "/home/u/notes", WorkspaceScope::User)], + ) + .unwrap(); + + let leftovers: Vec<_> = std::fs::read_dir(&dir) + .unwrap() + .map(|e| e.unwrap().file_name()) + .filter(|name| name.to_string_lossy().ends_with(TEMP_SUFFIX)) + .collect(); + assert!(leftovers.is_empty(), "leftover temp files: {leftovers:?}"); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn malformed_registry_is_typed_error() { let dir = test_dir("reg_bad"); From 2a059e46d17a9358f7b700e7c48c59ebc03e96ee Mon Sep 17 00:00:00 2001 From: Ali Turki Date: Sat, 18 Jul 2026 17:09:58 +0800 Subject: [PATCH 3/3] fix(scan): fail a stalled scan instead of spinning forever A scan on unreachable storage never settles, leaving the workspace spinner running indefinitely. Fail it when no progress arrives for 60s rather than on a fixed deadline, so a large but healthy scan is never cut off: a 55k-file workspace scans in ~3s with at most 122ms between progress events, leaving a wide margin. This is the mitigation half of #27. The upstream defect it guards against (tauri-apps/tauri#12338) is still open, so #27 stays open to track that fix landing in our Tauri version. Refs #27 --- src/hooks/useLibrary.test.ts | 32 +++++++++ src/lib/scan.test.ts | 133 ++++++++++++++++++++++++++++++++++- src/lib/scan.ts | 45 +++++++++--- 3 files changed, 199 insertions(+), 11 deletions(-) diff --git a/src/hooks/useLibrary.test.ts b/src/hooks/useLibrary.test.ts index a2340fa..b3c76bb 100644 --- a/src/hooks/useLibrary.test.ts +++ b/src/hooks/useLibrary.test.ts @@ -236,6 +236,38 @@ describe("useLibrary stale-while-revalidate rescans", () => { }); }); +describe("useLibrary scan failure handling", () => { + beforeEach(() => { + registry = []; + storedRoots = []; + dismissed = []; + vi.clearAllMocks(); + }); + + it("leaves the root unstuck when the scan rejects", async () => { + const error = vi.spyOn(console, "error").mockImplementation(() => {}); + storedRoots = ["/hung/folder"]; + vi.mocked(scanDirectory).mockRejectedValue( + new Error("This folder stopped responding while being scanned.") + ); + try { + const hook = await mount(); + await waitFor(() => expect(scanDirectory).toHaveBeenCalled()); + await waitFor(() => + expect(hook.result.current.scans["/hung/folder"]?.scanning).toBe(false) + ); + expect(error).toHaveBeenCalled(); + } finally { + error.mockRestore(); + vi.mocked(scanDirectory).mockImplementation(async (root: string) => ({ + root, + files: [], + truncated: false, + })); + } + }); +}); + describe("useLibrary watch resilience", () => { beforeEach(() => { registry = []; diff --git a/src/lib/scan.test.ts b/src/lib/scan.test.ts index 7c68306..b2fa19f 100644 --- a/src/lib/scan.test.ts +++ b/src/lib/scan.test.ts @@ -1,5 +1,28 @@ -import { describe, it, expect } from "vitest"; -import { parseFrontmatter, splitFrontmatter } from "./scan"; +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; + +type ProgressListener = (event: { payload: ScanProgress }) => void; +const progressListeners: ProgressListener[] = []; +const unlisten = vi.fn(); + +vi.mock("@tauri-apps/api/core", () => ({ + invoke: vi.fn(), +})); + +vi.mock("@tauri-apps/api/event", () => ({ + listen: vi.fn(async (_event: string, cb: ProgressListener) => { + progressListeners.push(cb); + return unlisten; + }), +})); + +import { invoke } from "@tauri-apps/api/core"; +import { + parseFrontmatter, + scanDirectory, + splitFrontmatter, + type ScanProgress, + type ScanResult, +} from "./scan"; describe("splitFrontmatter", () => { const cases: Record = { @@ -32,3 +55,109 @@ describe("splitFrontmatter", () => { expect(parseFrontmatter(rebuilt).content).toBe("\n# hello edited\n"); }); }); + +const ROOT = "/ws"; +// Mirrors SCAN_IDLE_TIMEOUT_MS in scan.ts. +const IDLE_TIMEOUT_MS = 60_000; +const PAST_IDLE_MS = IDLE_TIMEOUT_MS + 1_000; +const WITHIN_IDLE_MS = IDLE_TIMEOUT_MS - 10_000; + +function emitProgress(root: string, filesFound: number) { + const payload: ScanProgress = { + root, + currentDir: ".", + filesFound, + dirsVisited: 1, + }; + for (const listener of progressListeners) listener({ payload }); +} + +function emptyResult(root: string): ScanResult { + return { root, files: [], truncated: false }; +} + +// Lets the awaits inside scanDirectory (listen, invoke) run before the +// fake clock is moved. +async function flushMicrotasks() { + await Promise.resolve(); + await Promise.resolve(); +} + +describe("scanDirectory stall guard", () => { + beforeEach(() => { + progressListeners.length = 0; + vi.clearAllMocks(); + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("rejects with a friendly message when no progress arrives", async () => { + vi.mocked(invoke).mockReturnValue(new Promise(() => {})); + const promise = scanDirectory(ROOT); + const assertion = expect(promise).rejects.toThrow(/stopped responding/i); + await flushMicrotasks(); + await vi.advanceTimersByTimeAsync(PAST_IDLE_MS); + await assertion; + await expect(promise.catch((err: Error) => err.message)).resolves.not.toBe(""); + expect(unlisten).toHaveBeenCalled(); + }); + + it("does not reject while progress keeps arriving", async () => { + vi.mocked(invoke).mockReturnValue(new Promise(() => {})); + let settled = false; + const promise = scanDirectory(ROOT); + promise.then( + () => { + settled = true; + }, + () => { + settled = true; + } + ); + await flushMicrotasks(); + + for (let step = 1; step <= 4; step += 1) { + await vi.advanceTimersByTimeAsync(WITHIN_IDLE_MS); + emitProgress(ROOT, step); + } + await flushMicrotasks(); + + expect(settled).toBe(false); + }); + + it("ignores progress belonging to another root", async () => { + vi.mocked(invoke).mockReturnValue(new Promise(() => {})); + const promise = scanDirectory(ROOT); + const assertion = expect(promise).rejects.toThrow(/stopped responding/i); + await flushMicrotasks(); + await vi.advanceTimersByTimeAsync(WITHIN_IDLE_MS); + emitProgress("/other", 1); + await vi.advanceTimersByTimeAsync(PAST_IDLE_MS); + await assertion; + }); + + it("resolves a fast scan and cleans up its listener and timer", async () => { + vi.mocked(invoke).mockResolvedValue(emptyResult(ROOT)); + const onProgress = vi.fn(); + const promise = scanDirectory(ROOT, onProgress); + await flushMicrotasks(); + await expect(promise).resolves.toEqual(emptyResult(ROOT)); + expect(unlisten).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + await vi.advanceTimersByTimeAsync(PAST_IDLE_MS * 10); + }); + + it("forwards progress for its own root to the callback", async () => { + vi.mocked(invoke).mockReturnValue(new Promise(() => {})); + const onProgress = vi.fn(); + void scanDirectory(ROOT, onProgress).catch(() => {}); + await flushMicrotasks(); + emitProgress(ROOT, 7); + expect(onProgress).toHaveBeenCalledWith( + expect.objectContaining({ root: ROOT, filesFound: 7 }) + ); + }); +}); diff --git a/src/lib/scan.ts b/src/lib/scan.ts index 61c28e9..ed81292 100644 --- a/src/lib/scan.ts +++ b/src/lib/scan.ts @@ -41,6 +41,15 @@ export type ProgressCallback = (progress: ScanProgress) => void; const BOM = ""; const FRONTMATTER_PATTERN = /^---\r?\n([\s\S]*?)\r?\n---\r?\n?/; +// Longest silence tolerated between two scan-progress events before the scan +// is treated as hung. The backend throttles progress to one event per 100ms +// (PROGRESS_INTERVAL_MS in src-tauri/core/src/scan.rs), so a live scan is +// hundreds of times more talkative than this window. +const SCAN_IDLE_TIMEOUT_MS = 60_000; + +const SCAN_STALLED_MESSAGE = + "This folder stopped responding while being scanned. It may be on a disconnected drive or still downloading from cloud storage. Check the folder is available, then try again."; + export function parseFrontmatter(source: string): { data: Record; content: string; @@ -75,16 +84,34 @@ export async function scanDirectory( root: string, onProgress?: ProgressCallback ): Promise { - let unlisten: UnlistenFn | undefined; - if (onProgress) { - unlisten = await listen("scan-progress", (event) => { - const payload = event.payload; - if (payload.root === root) onProgress(payload); - }); - } + let idleTimer: ReturnType | undefined; + let markActivity: () => void = () => {}; + + const unlisten: UnlistenFn = await listen("scan-progress", (event) => { + const payload = event.payload; + if (payload.root !== root) return; + markActivity(); + if (onProgress) onProgress(payload); + }); + + // A legitimate workspace can scan for minutes (the walker caps at 50k + // files), so the guard is an inactivity window rather than a total + // deadline: only silence distinguishes a hung scan from a slow one. + const stalled = new Promise((_resolve, reject) => { + markActivity = () => { + if (idleTimer) clearTimeout(idleTimer); + idleTimer = setTimeout(() => reject(new Error(SCAN_STALLED_MESSAGE)), SCAN_IDLE_TIMEOUT_MS); + }; + markActivity(); + }); + try { - return await invoke("scan_markdown", { path: root }); + return await Promise.race([ + invoke("scan_markdown", { path: root }), + stalled, + ]); } finally { - if (unlisten) unlisten(); + if (idleTimer) clearTimeout(idleTimer); + unlisten(); } }