test(files): open the read-held thread's FIFO before it says it is reading (#2296)

changelog.d/file-pool-test-hang.md

@@ -0,0 +1,4 @@
+### Maintenance
+- The file-thread test that holds a thread inside a read can no longer hang the unit run: its
+  stand-in thread opens the FIFO it waits on before saying it is reading, and the test releases it
+  with a write that never waits, so a thread ended in between no longer blocks the test forever.

server/file-pool.test.ts

@@ -13,8 +13,9 @@ const waitAsync = (Atomics as unknown as { waitAsync: (cells: Int32Array, index:
 
 /** A held answer: the thread runs its task, then waits until `release`. */
 const gate = () => {
-  const cells = new Int32Array(new SharedArrayBuffer(8));
+  const cells = new Int32Array(new SharedArrayBuffer(12));
   cells[1] = -1;
+  cells[2] = -1;
   return {
     buffer: cells.buffer as SharedArrayBuffer,
     release: () => { Atomics.store(cells, 0, 1); Atomics.notify(cells, 0); },
@@ -24,6 +25,11 @@ const gate = () => {
       if ((waited.async ? await waited.value : waited.value) === "timed-out") throw new Error("the held thread never ran its task");
       return cells[1]!;
     },
+    /** The FIFO a FIFO gate's thread, ended inside its read, never closed: closed once that thread is gone. */
+    closeFifo: () => {
+      const fd = Atomics.exchange(cells, 2, -1);
+      if (fd >= 0) closeSync(fd);
+    },
   };
 };
 const held = (task: FileTask, buffer: SharedArrayBuffer): FileTask & { gate: SharedArrayBuffer } => ({ ...task, gate: buffer });
@@ -171,6 +177,7 @@ describe("file threads", () => {
     closeSync(writer);
     await fileThreadsClosed();
     expect(() => fstatSync(fd)).toThrow();
+    hold.closeFifo();
   });
 
   it("keeps a descriptor its thread is reading open until that thread is gone, when the last server stops", async () => {
@@ -189,10 +196,13 @@ describe("file threads", () => {
     const other = openSync(big, "r");
     expect(other).not.toBe(fd);
     closeSync(other);
-    // the read returns, the thread is gone, and only then is its descriptor closed
-    closeSync(openSync(fifo, "w"));
+    // the read returns (a write that never waits: the thread opened the FIFO before it said it ran), the thread is gone, and only then is its descriptor closed
+    const writer = openSync(fifo, fsConstants.O_WRONLY | fsConstants.O_NONBLOCK);
+    writeSync(writer, "x");
+    closeSync(writer);
     await fileThreadsClosed();
     expect(() => fstatSync(fd)).toThrow();
+    hold.closeFifo();
   });
 
   it("ends its threads, a busy one too, when the last server holding them stops", async () => {

server/file-worker.fake.ts

@@ -6,15 +6,21 @@
  * - with a `gate` (a SharedArrayBuffer of two Int32 cells), records in cell 1 the descriptor the
  *   task opened (`Atomics.notify` on it; -2 when it opened none) and answers only once cell 0 is
  *   set, as a call that returns late would;
- * - with a `readGate` (two Int32 cells and a FIFO's path), stops inside the file read itself: it
- *   records the descriptor being read in cell 1 (`Atomics.notify`), then opens the FIFO for reading,
- *   a real system call that blocks until a writer opens it (which ending the thread cannot cut
- *   short), and then reads, as a read on a stalled disk would;
+ * - with a `readGate` (three Int32 cells and a FIFO's path), stops inside the file read itself: it
+ *   opens the FIFO (read and write, which never waits), records the descriptor being read in cell 1
+ *   (`Atomics.notify`), then blocks in a read of the FIFO, a real system call that ending the thread
+ *   cannot cut short, and then reads the file, as a read on a stalled disk would. Opened before
+ *   cell 1 is set, the FIFO always has this reader once the test sees it, so the test's release (a
+ *   write that does not wait) never waits for this thread, even one ended in the meantime;
  * - with an `authorizeGate` (the same shape), stops once the file is authorized, before the task
  *   answers, as a check on a stalled disk would: it opens the FIFO (read and write, which never
  *   waits), records the descriptor the task named in its cell (`opened`) in cell 1, then blocks in a
  *   read of the FIFO. Opened before cell 1 is set, the FIFO always has this reader once the test
  *   sees it, so the test's release (a write that does not wait) never waits for this thread.
+ *
+ * Both FIFO gates name their FIFO descriptor in cell 2: a thread ended inside its read never runs its
+ * `finally` (Bun stops it as the read returns), so the test closes that descriptor once the thread
+ * is gone. Whichever side takes it from the cell first (`Atomics.exchange`) closes it.
  */
 import { closeSync, constants, openSync, readSync } from "node:fs";
 
@@ -25,28 +31,29 @@ declare const self: Worker;
 
 type Gate = { cells: SharedArrayBuffer; fifo: string };
 
+/** Opens the gate's FIFO (which never waits), names it in cell 2 and `fd` in cell 1 (`Atomics.notify`), then blocks in a read of it. */
+const holdOnFifo = (gate: Gate, fd: number): void => {
+  const cells = new Int32Array(gate.cells);
+  const fifo = openSync(gate.fifo, constants.O_RDWR);
+  Atomics.store(cells, 2, fifo);
+  Atomics.store(cells, 1, fd);
+  Atomics.notify(cells, 1);
+  try { readSync(fifo, new Uint8Array(1), 0, 1, null); } finally { if (Atomics.exchange(cells, 2, -1) >= 0) closeSync(fifo); }
+};
+
 self.onmessage = (event: MessageEvent<FileTask & { gate?: SharedArrayBuffer; readGate?: Gate; authorizeGate?: Gate }>) => {
   const task = event.data;
   if ("path" in task && (task.path.endsWith("/stuck") || (task.op === "open" && task.path.endsWith("/held-open")))) return;
   const readGate = task.readGate;
   if (readGate !== undefined) {
     pageFileHooks.readSync = (fd, buffer, offset, length, position) => {
-      const cells = new Int32Array(readGate.cells);
-      Atomics.store(cells, 1, fd);
-      Atomics.notify(cells, 1);
-      closeSync(openSync(readGate.fifo, "r"));
+      holdOnFifo(readGate, fd);
       return readSync(fd, buffer, offset, length, position);
     };
   }
   const authorizeGate = task.authorizeGate;
   if (authorizeGate !== undefined) {
-    pageFileHooks.afterAuthorize = () => {
-      const cells = new Int32Array(authorizeGate.cells);
-      const fifo = openSync(authorizeGate.fifo, constants.O_RDWR);
-      Atomics.store(cells, 1, "opened" in task && task.opened !== undefined ? task.opened[0]! : -2);
-      Atomics.notify(cells, 1);
-      try { readSync(fifo, new Uint8Array(1), 0, 1, null); } finally { closeSync(fifo); }
-    };
+    pageFileHooks.afterAuthorize = () => holdOnFifo(authorizeGate, "opened" in task && task.opened !== undefined ? task.opened[0]! : -2);
   }
   let result;
   try { result = runFileTask(task); } finally { delete pageFileHooks.readSync; delete pageFileHooks.afterAuthorize; }