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"); 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/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); } })(); 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(); } }