diff --git a/src/handwritten/commands/drive/state.ts b/src/handwritten/commands/drive/state.ts index 8ef2c40..00b8b24 100644 --- a/src/handwritten/commands/drive/state.ts +++ b/src/handwritten/commands/drive/state.ts @@ -139,38 +139,63 @@ export async function ensureDriveRealtimeState(root: string): Promise { +export interface DriveLockOptions { + // A sync holds the lock for the whole upload cycle. Realtime cursor + // persistence is a tiny best-effort write that should wait it out rather than + // surface a spurious warning, so it opts into a bounded retry. + retries?: number + retryDelayMs?: number + sleep?: (ms: number) => Promise +} + +// Enough attempts to outlast a normal sync cycle (~2s) before giving up. +const REALTIME_LOCK_RETRIES = 20 +const REALTIME_LOCK_RETRY_DELAY_MS = 100 + +export async function writeDriveRealtimeState( + root: string, + realtime: DriveRealtimeState, + lockOptions: DriveLockOptions = { retries: REALTIME_LOCK_RETRIES, retryDelayMs: REALTIME_LOCK_RETRY_DELAY_MS }, +): Promise { await withDriveLock(root, async () => { const current = await readDriveState(root) await writeDriveState(root, { ...current, realtime }) - }) + }, lockOptions) } -export async function withDriveLock(root: string, fn: () => Promise): Promise { +export async function withDriveLock(root: string, fn: () => Promise, options: DriveLockOptions = {}): Promise { await mkdir(join(root, DRIVE_DIR), { recursive: true }) const lockFile = join(root, DRIVE_DIR, "sync.lock") + const retries = options.retries ?? 0 + const retryDelayMs = options.retryDelayMs ?? 100 + const sleep = options.sleep ?? ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms))) const acquire = async () => { const handle = await open(lockFile, "wx") await handle.writeFile(String(process.pid)) return handle } - const fh = await acquire().catch(async (error) => { - if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error - if (!(await isLockAbandoned(lockFile))) { - throw new Error("sync lock already exists") - } - await rm(lockFile, { force: true }) - return acquire().catch((retryError) => { - if ((retryError as NodeJS.ErrnoException).code === "EEXIST") { - throw new Error("sync lock already exists") + let fh: Awaited> | undefined + for (let attempt = 0; ; attempt += 1) { + try { + fh = await acquire() + break + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error + if (await isLockAbandoned(lockFile)) { + await rm(lockFile, { force: true }) + continue } - throw retryError - }) - }) + if (attempt < retries) { + await sleep(retryDelayMs) + continue + } + throw Object.assign(new Error("sync lock already exists"), { code: "WSPC_DRIVE_LOCK_HELD" }) + } + } try { return await fn() } finally { - await fh.close().catch(() => {}) + await fh?.close().catch(() => {}) await rm(lockFile, { force: true }).catch(() => {}) } } diff --git a/test/handwritten/drive/realtime.test.ts b/test/handwritten/drive/realtime.test.ts index 296d616..1a49178 100644 --- a/test/handwritten/drive/realtime.test.ts +++ b/test/handwritten/drive/realtime.test.ts @@ -163,6 +163,11 @@ describe("drive realtime helpers", () => { const closed = Object.assign(new Error("unexpected failure"), { code: 1006 }) expect(redactedRealtimeError(closed)).toBe("realtime error (1006)") + // A sync-held lock collides with cursor persistence; its code must survive + // redaction so the warning is diagnosable instead of a bare "realtime error". + const lockHeld = Object.assign(new Error("sync lock already exists"), { code: "WSPC_DRIVE_LOCK_HELD" }) + expect(redactedRealtimeError(lockHeld)).toBe("realtime error (WSPC_DRIVE_LOCK_HELD)") + // Unsafe-looking codes (payload-shaped) are dropped, not surfaced. const leaky = Object.assign(new Error("boom"), { code: "Bearer secret-token" }) expect(redactedRealtimeError(leaky)).toBe("realtime error") diff --git a/test/handwritten/drive/state.test.ts b/test/handwritten/drive/state.test.ts index 0e6bb9e..8a852ba 100644 --- a/test/handwritten/drive/state.test.ts +++ b/test/handwritten/drive/state.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest" -import { mkdir, mkdtemp, readFile, utimes, writeFile } from "node:fs/promises" +import { mkdir, mkdtemp, readFile, rm, utimes, writeFile } from "node:fs/promises" import { join } from "node:path" import { tmpdir } from "node:os" import { DateTime } from "luxon" @@ -107,6 +107,51 @@ describe("drive state", () => { await expect(withDriveLock(root, async () => undefined)).rejects.toThrow(/sync lock already exists/) }) + it("tags lock contention with a diagnostic code", async () => { + const root = await mkdtemp(join(tmpdir(), "wspc-drive-lock-code-")) + await initDriveState(root, "lib_a") + await writeFile(join(root, ".wspc-drive", "sync.lock"), String(process.pid)) + + await expect(withDriveLock(root, async () => undefined)).rejects.toMatchObject({ + code: "WSPC_DRIVE_LOCK_HELD", + }) + }) + + it("retries a held lock and acquires it once released", async () => { + const root = await mkdtemp(join(tmpdir(), "wspc-drive-lock-retry-")) + await initDriveState(root, "lib_a") + const lockFile = join(root, ".wspc-drive", "sync.lock") + await writeFile(lockFile, String(process.pid)) + let sleeps = 0 + const sleep = async () => { + sleeps += 1 + if (sleeps === 2) await rm(lockFile, { force: true }) + } + let ran = false + + await withDriveLock(root, async () => { + ran = true + }, { retries: 5, sleep }) + + expect(ran).toBe(true) + expect(sleeps).toBe(2) + }) + + it("gives up retrying and reports contention when the lock never releases", async () => { + const root = await mkdtemp(join(tmpdir(), "wspc-drive-lock-retry-giveup-")) + await initDriveState(root, "lib_a") + await writeFile(join(root, ".wspc-drive", "sync.lock"), String(process.pid)) + let sleeps = 0 + const sleep = async () => { + sleeps += 1 + } + + await expect( + withDriveLock(root, async () => undefined, { retries: 3, sleep }), + ).rejects.toThrow(/sync lock already exists/) + expect(sleeps).toBe(3) + }) + it("rejects unsupported schema", async () => { const root = await mkdtemp(join(tmpdir(), "wspc-drive-state-bad-")) const badState = { schema_version: 2, library_id: "lib_a", entries: {}, conflicts: {} } @@ -359,12 +404,29 @@ describe("drive state", () => { await initDriveState(root, "lib_a") await withDriveLock(root, async () => { - await expect(writeDriveRealtimeState(root, { client_id: "drvcli_abc", last_cursor: "c1" })).rejects.toThrow( - /sync lock already exists/, - ) + await expect( + writeDriveRealtimeState(root, { client_id: "drvcli_abc", last_cursor: "c1" }, { retries: 2, sleep: async () => {} }), + ).rejects.toThrow(/sync lock already exists/) }) }) + it("persists realtime cursor by waiting out a concurrently held lock", async () => { + const root = await mkdtemp(join(tmpdir(), "wspc-drive-state-realtime-wait-")) + await initDriveState(root, "lib_a") + const lockFile = join(root, ".wspc-drive", "sync.lock") + await writeFile(lockFile, String(process.pid)) + let sleeps = 0 + const sleep = async () => { + sleeps += 1 + if (sleeps === 2) await rm(lockFile, { force: true }) + } + + await writeDriveRealtimeState(root, { client_id: "drvcli_abc", last_cursor: "c9" }, { retries: 10, sleep }) + + expect((await readDriveState(root)).realtime).toMatchObject({ last_cursor: "c9" }) + expect(sleeps).toBe(2) + }) + it("creates realtime client ids under the drive lock", async () => { const root = await mkdtemp(join(tmpdir(), "wspc-drive-state-ensure-realtime-lock-")) await initDriveState(root, "lib_a")