fix(drive): stop realtime cursor writes from warning during a sync

src/handwritten/commands/drive/state.ts

@@ -139,38 +139,63 @@ export async function ensureDriveRealtimeState(root: string): Promise<DriveState
   })
 }
 
-export async function writeDriveRealtimeState(root: string, realtime: DriveRealtimeState): Promise<void> {
+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<void>
+}
+
+// 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<void> {
   await withDriveLock(root, async () => {
     const current = await readDriveState(root)
     await writeDriveState(root, { ...current, realtime })
-  })
+  }, lockOptions)
 }
 
-export async function withDriveLock<T>(root: string, fn: () => Promise<T>): Promise<T> {
+export async function withDriveLock<T>(root: string, fn: () => Promise<T>, options: DriveLockOptions = {}): Promise<T> {
   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<void>((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<ReturnType<typeof acquire>> | 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(() => {})
   }
 }

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")

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")