fix(memory-core): a Windows sharing violation on the head wake ticket no longer rejects the waiter

packages/memory-core/src/locks/recall-wake-domain.test.ts

@@ -13,6 +13,7 @@ import {
   type RecallWakeLease,
 } from "./index"
 import { createLockRecord, type LockRecord } from "./lock-record"
+import { setRecallWakeTicketFsForTests } from "./recall-wake-domain"
 
 const dirs: string[] = []
 afterEach(async () => {
@@ -34,17 +35,6 @@ function withinMs<T>(promise: Promise<T>, label: string, ms = 5_000): Promise<T>
   return Promise.race([promise, deadline]).finally(() => clearTimeout(timer))
 }
 
-/**
- * True when the promise has already settled. Both racers sit one microtask hop deep and the settled
- * branch is subscribed first, so an already-settled promise always beats the sentinel; a pending one
- * never does.
- */
-async function settledAlready(promise: Promise<unknown>): Promise<boolean> {
-  const sentinel = Symbol("pending")
-  const winner = await Promise.race([promise.then(() => "settled", () => "settled"), Promise.resolve().then(() => sentinel)])
-  return winner !== sentinel
-}
-
 async function tickets(dir: string): Promise<string[]> {
   return (await readdir(recallWakeTicketDirectory(dir))).filter((name) => name.endsWith(".ticket"))
 }
@@ -92,12 +82,13 @@ describe("recall-wake lock domain: counting FIFO lease", () => {
 
   test("#given ten contenders and two slots #when slots are released one at a time #then every contender acquires, in ticket order, with never more than two leases live", async () => {
     const dir = await fixture()
+    const controller = new AbortController()
     const order: number[] = []
     let live = 0
     let peak = 0
     const leases: Array<RecallWakeLease | undefined> = []
     const waits = Array.from({ length: 10 }, (_, index) =>
-      acquireRecallWakeLease(dir, { waitTimeoutMs: 10_000, retryDelayMs: 10 }).then((lease) => {
+      acquireRecallWakeLease(dir, { waitTimeoutMs: 10_000, retryDelayMs: 10, signal: controller.signal }).then((lease) => {
         order.push(index)
         leases[index] = lease
         live += 1
@@ -106,25 +97,29 @@ describe("recall-wake lock domain: counting FIFO lease", () => {
       }),
     )
 
-    await withinMs(Promise.all([waits[0], waits[1]]), "the first two leases")
-    expect(order).toEqual([0, 1])
-    expect(await settledAlready(waits[2] ?? Promise.resolve())).toBe(false)
-
-    // Each release admits exactly the next ticket, never a later one.
-    for (let next = 2; next < 10; next += 1) {
-      const releasing = leases[next - 2]
-      if (releasing === undefined) throw new Error(`lease ${next - 2} was never acquired`)
-      live -= 1
-      expect(await releasing.release()).toBe(true)
-      await withinMs(waits[next] ?? Promise.resolve(), `lease ${next}`)
-      expect(order[next]).toBe(next)
-      if (next + 1 < 10) expect(await settledAlready(waits[next + 1] ?? Promise.resolve())).toBe(false)
+    try {
+      await withinMs(Promise.all([waits[0], waits[1]]), "the first two leases")
+      expect(order).toEqual([0, 1])
+
+      // Each release admits exactly the next ticket, never a later one.
+      for (let next = 2; next < 10; next += 1) {
+        const releasing = leases[next - 2]
+        if (releasing === undefined) throw new Error(`lease ${next - 2} was never acquired`)
+        live -= 1
+        expect(await releasing.release()).toBe(true)
+        await withinMs(waits[next] ?? Promise.resolve(), `lease ${next}`)
+        expect(order).toEqual(Array.from({ length: next + 1 }, (_, index) => index))
+        expect(live).toBe(2)
+      }
+
+      expect(peak).toBe(2)
+      expect(await tickets(dir)).toEqual([])
+      for (const lease of leases.slice(8)) expect(await lease?.release()).toBe(true)
+    } finally {
+      controller.abort(new Error("FIFO test teardown"))
+      await Promise.allSettled(waits)
+      await Promise.all(leases.map(async (lease) => { await lease?.release() }))
     }
-
-    expect(order).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
-    expect(peak).toBe(2)
-    expect(await tickets(dir)).toEqual([])
-    for (const lease of leases.slice(8)) expect(await lease?.release()).toBe(true)
   })
 
   test("#given a dead owner recorded on both slots #when a contender arrives #then both are reclaimed on the liveness proof and the new owner is recorded", async () => {
@@ -199,6 +194,55 @@ describe("recall-wake lock domain: counting FIFO lease", () => {
     expect(await tickets(dir)).toEqual([path.basename(liveTicket)])
   })
 
+  test("#given a live head ticket with a transient sharing violation #when a follower inspects it #then the ticket keeps its place and both waiters acquire in order", async () => {
+    const dir = await fixture()
+    const controller = new AbortController()
+    const holder = await acquireRecallWakeLease(dir, { ...fast, maxConcurrent: 1 })
+    const sharingObserved = Promise.withResolvers<void>()
+    let sharingInjected = false
+    const restore = setRecallWakeTicketFsForTests({
+      readFile: async (filePath, encoding) => {
+        if (!sharingInjected && filePath.endsWith(".ticket")) {
+          sharingInjected = true
+          sharingObserved.resolve()
+          throw Object.assign(new Error("ticket is transitioning"), { code: "EPERM" })
+        }
+        return await readFile(filePath, encoding)
+      },
+      isSharingError: (error) => error instanceof Error && "code" in error && error.code === "EPERM",
+    })
+    // The waiters' budgets are circuit breakers, not the behavior under test: each must outlast the
+    // whole hand-off (sharing violation, holder release, head acquire and release) on a slow runner.
+    const waiting = { waitTimeoutMs: 10_000, retryDelayMs: 10, maxConcurrent: 1, signal: controller.signal }
+    const headWait = acquireRecallWakeLease(dir, waiting)
+    const followerWait = acquireRecallWakeLease(dir, waiting)
+    const followerResult = followerWait.then(
+      (lease) => ({ status: "acquired", lease } as const),
+      (error: unknown) => ({ status: "rejected", error } as const),
+    )
+    let headLease: RecallWakeLease | undefined
+    let followerLease: RecallWakeLease | undefined
+
+    try {
+      await withinMs(sharingObserved.promise, "the injected ticket sharing violation")
+      expect(await holder.release()).toBe(true)
+      headLease = await withinMs(headWait, "the head waiter")
+      expect(await headLease.release()).toBe(true)
+
+      const follower = await withinMs(followerResult, "the follower waiter")
+      expect(follower.status).toBe("acquired")
+      if (follower.status !== "acquired") throw follower.error
+      followerLease = follower.lease
+      expect(await followerLease.release()).toBe(true)
+      expect(await tickets(dir)).toEqual([])
+    } finally {
+      controller.abort(new Error("sharing test teardown"))
+      await Promise.allSettled([headWait, followerWait])
+      await Promise.all([holder.release(), headLease?.release(), followerLease?.release()])
+      restore()
+    }
+  })
+
   test("#given a waiting contender #when its signal aborts #then the wait ends promptly with the abort reason and no ticket is left behind", async () => {
     const dir = await fixture()
     const first = await acquireRecallWakeLease(dir, fast)

packages/memory-core/src/locks/recall-wake-domain.ts

@@ -67,6 +67,20 @@ let ticketSequence = 0
 /** In-process publication order: a later ticket is never visible before an earlier one. */
 let publishing: Promise<unknown> = Promise.resolve()
 
+export interface RecallWakeTicketFs {
+  readonly readFile?: (filePath: string, encoding: "utf8") => Promise<string>
+  readonly isSharingError?: (error: unknown) => boolean
+}
+
+let ticketFs: RecallWakeTicketFs = {}
+
+/** Test seam for deterministic Windows ticket-sharing coverage; production uses resilient fs. */
+export function setRecallWakeTicketFsForTests(next: RecallWakeTicketFs | undefined): () => void {
+  const previous = ticketFs
+  ticketFs = next ?? {}
+  return () => { ticketFs = previous }
+}
+
 function ticketName(): string {
   ticketSequence = (ticketSequence + 1) % 1_000_000
   const issued = String(Date.now()).padStart(16, "0")
@@ -100,15 +114,25 @@ async function unlinkIfPresent(filePath: string): Promise<void> {
   }
 }
 
+function isTicketReadSharingError(error: unknown): boolean {
+  if (ticketFs.isSharingError !== undefined) return ticketFs.isSharingError(error)
+  if (process.platform !== "win32") return false
+  const code = errorCode(error)
+  return code === "EBUSY" || code === "EPERM" || code === "EACCES"
+}
+
 /** Reaps the head ticket when its owner is proven dead; an unreadable or unparsable ticket keeps its place. */
 async function reapDeadHead(ticketDirectory: string, head: string): Promise<boolean> {
   const ticketPath = path.join(ticketDirectory, head)
   let raw: string
   try {
-    raw = await readFile(ticketPath, "utf8")
+    raw = await (ticketFs.readFile ?? readFile)(ticketPath, "utf8")
   } catch (error) {
     // Gone already: its owner acquired or withdrew between our readdir and this read.
     if (errorCode(error) === "ENOENT") return true
+    // Windows can deny the open while the ticket owner is publishing or withdrawing it. The
+    // unreadable ticket remains the queue head; wait for the next normal poll instead of rejecting.
+    if (isTicketReadSharingError(error)) return false
     throw error
   }
   const owner = parseLockRecord(raw)

packages/omo-senpi/src/components/memory/kibitzer/wake-slot.test.ts

@@ -39,17 +39,6 @@ function acquired(admission: KibitzerWakeAdmission): KibitzerWakeLease {
   return admission.lease
 }
 
-/**
- * True when the promise has already settled. Both racers sit one microtask hop deep and the settled
- * branch is subscribed first, so an already-settled promise always beats the sentinel; a pending one
- * never does.
- */
-async function settledAlready(promise: Promise<unknown>): Promise<boolean> {
-  const sentinel = Symbol("pending")
-  const winner = await Promise.race([promise.then(() => "settled", () => "settled"), Promise.resolve().then(() => sentinel)])
-  return winner !== sentinel
-}
-
 describe("createKibitzerWakeSlot", () => {
   test("#given the production defaults #when a slot is built from config #then the wait is bounded and the poll is a delay, never a spin", async () => {
     const slot = createKibitzerWakeSlot({ locksDirectory: await locksDirectory(), maxConcurrent: 2 })
@@ -62,38 +51,44 @@ describe("createKibitzerWakeSlot", () => {
   test("#given ten sessions contending for two slots #when leases are released one at a time #then admission is FIFO and never more than two wakes are live", async () => {
     const dir = await locksDirectory()
     const slot = createKibitzerWakeSlot({ locksDirectory: dir, maxConcurrent: 2, waitTimeoutMs: 10_000, pollMs: 10 })
+    const controller = new AbortController()
     const order: number[] = []
     const leases: Array<KibitzerWakeLease | undefined> = []
     let live = 0
     let peak = 0
     const waits = Array.from({ length: 10 }, (_, session) =>
-      slot.acquire().then((admission) => {
+      slot.acquire(controller.signal).then((admission) => {
+        const lease = acquired(admission)
         order.push(session)
-        leases[session] = acquired(admission)
+        leases[session] = lease
         live += 1
         peak = Math.max(peak, live)
         return admission
       }),
     )
 
-    await withinMs(Promise.all([waits[0], waits[1]]), "the first two admissions", 5_000)
-    expect(order).toEqual([0, 1])
-    expect(await settledAlready(waits[2] ?? Promise.resolve())).toBe(false)
-
-    for (let next = 2; next < 10; next += 1) {
-      const releasing = leases[next - 2]
-      if (releasing === undefined) throw new Error(`session ${next - 2} never acquired`)
-      live -= 1
-      expect(await releasing.release()).toBe(true)
-      await withinMs(waits[next] ?? Promise.resolve(), `admission ${next}`, 5_000)
-      expect(order[next]).toBe(next)
-      if (next + 1 < 10) expect(await settledAlready(waits[next + 1] ?? Promise.resolve())).toBe(false)
+    try {
+      await withinMs(Promise.all([waits[0], waits[1]]), "the first two admissions", 5_000)
+      expect(order).toEqual([0, 1])
+
+      for (let next = 2; next < 10; next += 1) {
+        const releasing = leases[next - 2]
+        if (releasing === undefined) throw new Error(`session ${next - 2} never acquired`)
+        live -= 1
+        expect(await releasing.release()).toBe(true)
+        await withinMs(waits[next] ?? Promise.resolve(), `admission ${next}`, 5_000)
+        expect(order).toEqual(Array.from({ length: next + 1 }, (_, index) => index))
+        expect(live).toBe(2)
+      }
+
+      expect(peak).toBe(2)
+      expect(await tickets(dir)).toEqual([])
+      for (const lease of leases.slice(8)) expect(await lease?.release()).toBe(true)
+    } finally {
+      controller.abort()
+      await Promise.allSettled(waits)
+      await Promise.all(leases.map(async (lease) => { await lease?.release() }))
     }
-
-    expect(order).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
-    expect(peak).toBe(2)
-    expect(await tickets(dir)).toEqual([])
-    for (const lease of leases.slice(8)) expect(await lease?.release()).toBe(true)
   })
 
   test("#given both slots held #when a wake waits its whole budget #then it is reported busy with the time it waited, leaves no ticket, and a released slot admits it", async () => {