From 061927257429d09d7cebb04ff4ea77c2ed4d19ff Mon Sep 17 00:00:00 2001 From: Daniil Trishkin Date: Mon, 7 Sep 2026 21:24:19 +0200 Subject: [PATCH 1/2] fix: skip directory fsync on Windows (EPERM) --- src/outbox.ts | 8 ++++++-- src/queue.ts | 4 ++++ 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/src/outbox.ts b/src/outbox.ts index b5b9a66..d0ff3ab 100644 --- a/src/outbox.ts +++ b/src/outbox.ts @@ -52,8 +52,12 @@ export function Outbox(opts: { storageDir: string }): OutboxInstance { } chmodSync(temp, 0o600) renameSync(temp, target) - const dirFd = openSync(directory, "r") - try { fsyncSync(dirFd) } finally { closeSync(dirFd) } + // Directory fsync is POSIX-only (EPERM on Windows); file fsync above is + // the best durability available there. + if (process.platform !== "win32") { + const dirFd = openSync(directory, "r") + try { fsyncSync(dirFd) } finally { closeSync(dirFd) } + } } function update(endpointId: string, messageId: string, values: Partial): OutboxRecord | null { diff --git a/src/queue.ts b/src/queue.ts index 9c91b18..6f6ea5e 100644 --- a/src/queue.ts +++ b/src/queue.ts @@ -167,6 +167,9 @@ function migrationStateFiles(spoolDir: string, state: SpoolState): string[] { } function syncMigrationDirectory(directory: string): void { + // Directory fsync is POSIX-only; on Windows it raises EPERM. The fsync'd + // file plus rename is the best durability available there. + if (process.platform === "win32") return const fd = openSync(directory, "r") try { fsyncSync(fd) @@ -605,6 +608,7 @@ export function MessageQueue(opts: QueueOptions): QueueInstance { } function syncDirectory(directory: string): void { + if (process.platform === "win32") return const dirFd = openSync(directory, "r") try { fsyncSync(dirFd) From 8b75c8f418ffcf194804ae9ef9cce427dbfc76bc Mon Sep 17 00:00:00 2001 From: Daniil Trishkin Date: Mon, 7 Sep 2026 21:24:20 +0200 Subject: [PATCH 2/2] test: make the suite pass on Windows --- tests/ack-integration.test.mjs | 7 +++++-- tests/config.test.mjs | 13 +++++++------ tests/outbox.test.mjs | 3 ++- tests/queue-concurrency.test.mjs | 9 +++++---- tests/queue.test.mjs | 6 ++++-- tests/real-opencode.test.mjs | 1 + tests/registry.test.mjs | 10 +++++++--- tests/transport.test.mjs | 10 ++++++---- 8 files changed, 37 insertions(+), 22 deletions(-) diff --git a/tests/ack-integration.test.mjs b/tests/ack-integration.test.mjs index 266d5a7..2ae11ea 100644 --- a/tests/ack-integration.test.mjs +++ b/tests/ack-integration.test.mjs @@ -10,6 +10,9 @@ import { Outbox } from "../dist/outbox.js" import { Sender } from "../dist/sender.js" const noopLogger = async () => {} +// Exercise the UDS listener where the platform has one; Windows gets the +// loopback TCP listener (UDS binds are unsupported there). +const listenerPlatform = process.platform === "win32" ? "win32" : "darwin" test("held accept/drop/expiry outcomes round-trip as durable final ACKs", async () => { const dir = await mkdtemp(join(tmpdir(), "peers-ack-e2e-")) @@ -17,7 +20,7 @@ test("held accept/drop/expiry outcomes round-trip as durable final ACKs", async const senderEndpoint = "session-sender" const receiverEndpoint = "session-receiver" const senderListener = InboxListener({ - token: "sender-token", maxBodyBytes: 20_000, runtimeDir: join(dir, "runtime"), processId: "sender", platform: "darwin", + token: "sender-token", maxBodyBytes: 20_000, runtimeDir: join(dir, "runtime"), processId: "sender", platform: listenerPlatform, resolveEndpoint: ({ toEndpointId }) => toEndpointId === senderEndpoint ? senderEndpoint : null, onMessage: async () => "refused", onAcknowledgement: async (ack) => { await outbox.applyAcknowledgement(ack) }, @@ -25,7 +28,7 @@ test("held accept/drop/expiry outcomes round-trip as durable final ACKs", async }) const queue = MessageQueue({ endpointId: receiverEndpoint, maxQueue: 10, maxHeld: 10, heldExpiryMs: 1_000, inboxFile: join(dir, "receiver", "inbox.json"), logger: noopLogger }) const receiverListener = InboxListener({ - token: "receiver-token", maxBodyBytes: 20_000, runtimeDir: join(dir, "runtime"), processId: "receiver", platform: "darwin", + token: "receiver-token", maxBodyBytes: 20_000, runtimeDir: join(dir, "runtime"), processId: "receiver", platform: listenerPlatform, resolveEndpoint: ({ toEndpointId }) => toEndpointId === receiverEndpoint ? receiverEndpoint : null, onMessage: async (message) => await queue.hold(message) ? "held" : "duplicate", logger: noopLogger, diff --git a/tests/config.test.mjs b/tests/config.test.mjs index d5866f4..cf1ec34 100644 --- a/tests/config.test.mjs +++ b/tests/config.test.mjs @@ -1,18 +1,19 @@ import { test } from "node:test" import assert from "node:assert/strict" +import { join } from "node:path" import { resolveConfig, defaultDataDir, validateName, defaultPeerName } from "../dist/config.js" test("defaultDataDir honors XDG_DATA_HOME and falls back to ~/.local/share", () => { assert.equal(defaultDataDir({ XDG_DATA_HOME: "/xdg" }), "/xdg") - assert.match(defaultDataDir({}), /\.local\/share$/) + assert.ok(defaultDataDir({}).endsWith(join(".local", "share"))) }) test("resolveConfig applies defaults", () => { const cfg = resolveConfig(undefined, { XDG_DATA_HOME: "/xdg" }) - assert.equal(cfg.storageDir, "/xdg/opencode-plugin-peers") - assert.equal(cfg.peersDir, "/xdg/opencode-plugin-peers/peers.d") - assert.equal(cfg.inboxFile, "/xdg/opencode-plugin-peers/inbox.json") - assert.equal(cfg.spoolDir, "/xdg/opencode-plugin-peers/spool") + assert.equal(cfg.storageDir, join("/xdg", "opencode-plugin-peers")) + assert.equal(cfg.peersDir, join("/xdg", "opencode-plugin-peers", "peers.d")) + assert.equal(cfg.inboxFile, join("/xdg", "opencode-plugin-peers", "inbox.json")) + assert.equal(cfg.spoolDir, join("/xdg", "opencode-plugin-peers", "spool")) assert.equal(cfg.inboundPolicy, "accept") assert.equal(cfg.peerPermissions, "allow") assert.equal(cfg.heartbeatMs, 10_000) @@ -41,7 +42,7 @@ test("resolveConfig merges user options", () => { assert.equal(cfg.peerPermissions, "ask") assert.equal(cfg.name, "frontend") assert.equal(cfg.maxQueue, 5) - assert.equal(cfg.peersDir, "/custom/peers.d") + assert.equal(cfg.peersDir, join("/custom", "peers.d")) }) test("validateName accepts safe names and rejects dangerous ones", () => { diff --git a/tests/outbox.test.mjs b/tests/outbox.test.mjs index 5a05a5c..856f23d 100644 --- a/tests/outbox.test.mjs +++ b/tests/outbox.test.mjs @@ -36,7 +36,8 @@ test("outbox durably separates transport receipt from final acknowledgement", as acknowledgedAt: 200, }), true) assert.equal(restarted.get("session-from", "m-1").finalStatus, "delivered") - assert.equal((await stat(join(dir, "outbox"))).mode & 0o777, 0o700) + // Windows cannot represent POSIX permission bits; stat always reports 0o666. + if (process.platform !== "win32") assert.equal((await stat(join(dir, "outbox"))).mode & 0o777, 0o700) } finally { await rm(dir, { recursive: true, force: true }) } diff --git a/tests/queue-concurrency.test.mjs b/tests/queue-concurrency.test.mjs index 4685b90..7dc12f2 100644 --- a/tests/queue-concurrency.test.mjs +++ b/tests/queue-concurrency.test.mjs @@ -5,13 +5,14 @@ import { mkdtemp, readdir, rm, stat, utimes, writeFile } from "node:fs/promises" import { tmpdir } from "node:os" import { join } from "node:path" import { setTimeout as delay } from "node:timers/promises" +import { fileURLToPath } from "node:url" -const worker = new URL("./fixtures/queue-worker.mjs", import.meta.url) -const staleRaceWorker = new URL("./fixtures/stale-lock-race-worker.mjs", import.meta.url) +const worker = fileURLToPath(new URL("./fixtures/queue-worker.mjs", import.meta.url)) +const staleRaceWorker = fileURLToPath(new URL("./fixtures/stale-lock-race-worker.mjs", import.meta.url)) function runWorker(dir, name, id, text, maxQueue, operation = "enqueue") { return new Promise((resolve, reject) => { - const child = spawn(process.execPath, [worker.pathname, dir, name, id, text, String(maxQueue), operation], { + const child = spawn(process.execPath, [worker, dir, name, id, text, String(maxQueue), operation], { stdio: ["ignore", "pipe", "pipe"], }) let stdout = "" @@ -39,7 +40,7 @@ async function releaseWorkers(dir, count) { } function startStaleRaceWorker(dir, role, messageId, payloadBytes = 0) { - const child = spawn(process.execPath, [staleRaceWorker.pathname, dir, role, messageId, String(payloadBytes)], { + const child = spawn(process.execPath, [staleRaceWorker, dir, role, messageId, String(payloadBytes)], { stdio: ["ignore", "pipe", "pipe"], }) let stdout = "" diff --git a/tests/queue.test.mjs b/tests/queue.test.mjs index 653f311..212bd08 100644 --- a/tests/queue.test.mjs +++ b/tests/queue.test.mjs @@ -7,6 +7,8 @@ import { MessageQueue, RateLimiter } from "../dist/queue.js" import { gateMessage, isLoopMessage } from "../dist/gating.js" const noopLogger = async () => {} +// Windows cannot represent POSIX permission bits; stat always reports 0o666. +const isPosix = process.platform !== "win32" const msg = (id) => ({ id, @@ -62,7 +64,7 @@ test("queue: stores queued messages in an endpoint spool", async () => { const [file] = await readdir(queuedDir) const entry = join(queuedDir, file) assert.deepEqual(JSON.parse(await readFile(entry, "utf8")).message.id, "durable") - assert.equal((await stat(entry)).mode & 0o777, 0o600) + if (isPosix) assert.equal((await stat(entry)).mode & 0o777, 0o600) } finally { await rm(dir, { recursive: true, force: true }) } @@ -148,7 +150,7 @@ test("queue: repairs permissions on an existing endpoint directory", async () => const q = MessageQueue({ endpointId: "endpoint-a", maxQueue: 1, maxHeld: 1, inboxFile: join(dir, "inbox.json"), logger: noopLogger }) assert.equal(q.enqueue(msg("secure-endpoint")), true) - assert.equal((await stat(endpointDir)).mode & 0o777, 0o700) + if (isPosix) assert.equal((await stat(endpointDir)).mode & 0o777, 0o700) } finally { await rm(dir, { recursive: true, force: true }) } diff --git a/tests/real-opencode.test.mjs b/tests/real-opencode.test.mjs index 37f4811..c41f18b 100644 --- a/tests/real-opencode.test.mjs +++ b/tests/real-opencode.test.mjs @@ -144,6 +144,7 @@ function v2Message(from, to, id, text) { } test("real OpenCode hosts provide busy exact injection, restart recovery, permission boundaries, v1 interop, and held command ACKs", { timeout: 60_000 }, async (t) => { + if (process.platform === "win32") return t.skip("transport liveness assertions are UDS-specific; POSIX-only for now") const found = spawnSync("which", ["opencode"], { encoding: "utf8" }) if (found.status !== 0 || !found.stdout.trim()) return t.skip("opencode binary is unavailable") const binary = found.stdout.trim() diff --git a/tests/registry.test.mjs b/tests/registry.test.mjs index 7238a3c..222e422 100644 --- a/tests/registry.test.mjs +++ b/tests/registry.test.mjs @@ -6,6 +6,8 @@ import { join } from "node:path" import { Registry, newInstanceId, pidAlive, uniqueName } from "../dist/registry.js" const noopLogger = async () => {} +// Windows cannot represent POSIX permission bits; stat always reports 0o666. +const isPosix = process.platform !== "win32" async function makeDir() { return mkdtemp(join(tmpdir(), "peers-registry-")) @@ -43,8 +45,10 @@ test("start writes a 0600 entry file with expected fields", async () => { assert.equal(entry.inboundPolicy, "accept") assert.equal(entry.inboxUrl, "http://127.0.0.1:5000") assert.ok(entry.heartbeatAt > 0) - const mode = (await stat(join(dir, files[0]))).mode & 0o777 - assert.equal(mode, 0o600) + if (isPosix) { + const mode = (await stat(join(dir, files[0]))).mode & 0o777 + assert.equal(mode, 0o600) + } await reg.stop() assert.deepEqual(await readdir(dir), []) } finally { @@ -192,7 +196,7 @@ test("v2 registry publishes one endpoint per session plus the most-recent v1 com entry: JSON.parse(await readFile(join(dir, file), "utf8")), mode: (await stat(join(dir, file))).mode & 0o777, }))) - assert.ok(entries.every(({ mode }) => mode === 0o600)) + if (isPosix) assert.ok(entries.every(({ mode }) => mode === 0o600)) const v2 = entries.filter(({ entry }) => entry.version === 2).map(({ entry }) => entry) assert.deepEqual(v2.map((entry) => entry.endpointId).sort(), ["session-alpha", "session-beta"]) diff --git a/tests/transport.test.mjs b/tests/transport.test.mjs index 7c3adc5..63b6f01 100644 --- a/tests/transport.test.mjs +++ b/tests/transport.test.mjs @@ -10,6 +10,8 @@ import { Sender } from "../dist/sender.js" import { LocalTransport } from "../dist/transport.js" const noopLogger = async () => {} +// Node has no Unix-domain socket support on Windows; the UDS tests are POSIX-only. +const udsSkip = process.platform === "win32" ? "Unix-domain sockets are unavailable on Windows" : false const v2Message = (toEndpointId) => ({ version: 2, @@ -22,7 +24,7 @@ const v2Message = (toEndpointId) => ({ sentAt: Date.now(), }) -test("local UDS transport authenticates and routes v2 exactly while v1 uses compatibility endpoint", async () => { +test("local UDS transport authenticates and routes v2 exactly while v1 uses compatibility endpoint", { skip: udsSkip }, async () => { const runtimeDir = await mkdtemp(join(tmpdir(), "peers-transport-")) const routed = [] const acknowledgements = [] @@ -110,7 +112,7 @@ test("local UDS transport authenticates and routes v2 exactly while v1 uses comp } }) -test("UDS startup rejects a live process-id collision without unlinking the owner", async () => { +test("UDS startup rejects a live process-id collision without unlinking the owner", { skip: udsSkip }, async () => { const runtimeDir = await mkdtemp(join(tmpdir(), "peers-transport-collision-")) const received = [] const first = InboxListener({ @@ -150,7 +152,7 @@ test("UDS startup rejects a live process-id collision without unlinking the owne } }) -test("UDS startup removes a stale socket left by a crashed owner", async () => { +test("UDS startup removes a stale socket left by a crashed owner", { skip: udsSkip }, async () => { const runtimeDir = await mkdtemp(join(tmpdir(), "peers-transport-stale-")) const socketPath = join(runtimeDir, "stale-process.sock") const child = spawn(process.execPath, [ @@ -187,7 +189,7 @@ test("UDS startup removes a stale socket left by a crashed owner", async () => { } }) -test("UDS start failures close the server and leave no socket artifact", async () => { +test("UDS start failures close the server and leave no socket artifact", { skip: udsSkip }, async () => { const runtimeDir = await mkdtemp(join(tmpdir(), "peers-transport-failure-")) const listener = InboxListener({ token: "secret",