From a0833a0fad26784ba71a7c715924965f332a8a2f Mon Sep 17 00:00:00 2001 From: flint Date: Tue, 13 Oct 2026 22:50:41 +0000 Subject: [PATCH] fix(mail): MailClient resolves one mail root per operation for the lock and the paths it guards Refs #636. MailClient pinned its promotion lock root at construction while its mailbox file paths and replay-state path were joined from the configured mail directory at each use. If the mail root was re-pointed or moved while a long-running client was alive, the promotion lock stayed under the old root while the files and the replay state followed the new one, so a second process holding the new root's lock did not exclude promotion and a delivery could repeat. Resolve the mailbox root once per operation and derive the promotion lock path, the mailbox file paths and the replay-state path from that one value, so a re-pointed root moves all three together. Test: a real-filesystem case re-points the mail root under a live client while a real second process holds the new root's lock; the client does not promote under the old root's lock, and the one envelope is delivered once. --- .../fixed-mailclient-one-mail-root.md | 1 + packages/agent/src/io/mail.ts | 102 +++++++++-------- .../test/mail-client-repoint-root.test.ts | 107 ++++++++++++++++++ .../test/mail-interrupted-recovery.test.ts | 5 +- .../agent/test/mail-promote-guard.test.ts | 3 +- 5 files changed, 168 insertions(+), 50 deletions(-) create mode 100644 .changelog/unreleased/fixed-mailclient-one-mail-root.md create mode 100644 packages/agent/test/mail-client-repoint-root.test.ts diff --git a/.changelog/unreleased/fixed-mailclient-one-mail-root.md b/.changelog/unreleased/fixed-mailclient-one-mail-root.md new file mode 100644 index 00000000..28fc8046 --- /dev/null +++ b/.changelog/unreleased/fixed-mailclient-one-mail-root.md @@ -0,0 +1 @@ +- **MailClient derives its promotion lock, mailbox paths and replay state from one root resolved per operation.** A mail root re-pointed or moved while a client is alive now moves all three together: the promotion lock is taken under the root in force, not one resolved at construction. diff --git a/packages/agent/src/io/mail.ts b/packages/agent/src/io/mail.ts index 2e116092..bddccfcb 100644 --- a/packages/agent/src/io/mail.ts +++ b/packages/agent/src/io/mail.ts @@ -102,12 +102,6 @@ function lstatOrAbsent(path: string): { dev: number; ino: number } | null { * Writes outgoing mail to mailDir/outbox/new. */ export class MailClient { - private mailboxRoot: string; - private readonly mailboxLockRoot: string; - private inboxNew: string; - private inboxCur: string; - private inboxDlq: string; - private outboxNew: string; private readonly flairClient: FlairClient; constructor( @@ -126,30 +120,38 @@ export class MailClient { ); } this.flairClient = flairClient; - this.mailboxRoot = join(mailDir, agentId); - // cli#616: promotion locks the mailbox under the resolved real path of the - // configured mail directory (see mail-lock-namespace.ts). Placement recovery - // locks `mailboxRoot` as given; both reach one filesystem lock. - this.mailboxLockRoot = mailboxLockRoot(mailDir, agentId); - this.inboxNew = join(mailDir, agentId, "new"); - this.inboxCur = join(mailDir, agentId, "cur"); - this.inboxDlq = join(mailDir, agentId, "dlq"); - this.outboxNew = join(mailDir, agentId, "outbox"); - for (const dir of [this.inboxNew, this.inboxCur, this.inboxDlq, this.outboxNew]) { - mkdirSync(dir, { recursive: true }); - } + // cli#636: create the maildir for the root configured now. Every operation + // derives its own root instead; dir creation is not a construction-time pin. + this.mailboxRoot(); + } + + /** + * cli#636: the mailbox root for the current operation, resolved once from the + * configured `mailDir` and `agentId` (symlinks and `.`/`..` resolved) with its + * maildir subdirectories created. The promotion lock, the mailbox file paths + * and the replay-state path all derive from this one value, so a mail root + * re-pointed or moved while the client is alive moves all three together. It + * is never pinned at construction. cli#616: the resolved real path comes from + * mail-lock-namespace.ts so promotion and placement recovery reach one lock. + */ + private mailboxRoot(): string { + const root = mailboxLockRoot(this.mailDir, this.agentId); + for (const dir of ["new", "cur", "dlq", "outbox"]) mkdirSync(join(root, dir), { recursive: true }); + return root; } async checkNewMail(): Promise { const messages: MailMessage[] = []; - await this.recoverInterruptedPlacements(messages); - if (!existsSync(this.inboxNew)) return messages; + const root = this.mailboxRoot(); + await this.recoverInterruptedPlacements(messages, root); + const inboxNew = join(root, "new"); + if (!existsSync(inboxNew)) return messages; - const files = readdirSync(this.inboxNew).filter((f) => !f.startsWith(".") && !f.includes("/") && !f.includes("\\")); + const files = readdirSync(inboxNew).filter((f) => !f.startsWith(".") && !f.includes("/") && !f.includes("\\")); for (const file of files) { const started = Date.now(); - const srcPath = join(this.inboxNew, file); + const srcPath = join(inboxNew, file); let body: string; try { @@ -186,7 +188,7 @@ export class MailClient { } if (!verifyResult.pass) { - this.deadLetter(file, srcPath, verifyResult, started); + this.deadLetter(file, srcPath, verifyResult, started, root); continue; } @@ -208,10 +210,10 @@ export class MailClient { } try { - const committed = await this.commitToCur(file, srcPath, body, verifyResult.envelope); + const committed = await this.commitToCur(root, file, srcPath, body, verifyResult.envelope); if (committed === "already-delivered") continue; if (committed) { - this.deadLetter(file, srcPath, { ...committed, from: verifyResult.envelope.from }, started); + this.deadLetter(file, srcPath, { ...committed, from: verifyResult.envelope.from }, started, root); continue; } const from = verifyResult.envelope.from; @@ -244,12 +246,13 @@ export class MailClient { /** Commit under the mailbox lock; attempt placement rollback on append failure. */ private async commitToCur( + root: string, file: string, srcPath: string, body: string, envelope: Envelope, ): Promise<{ pass: false; class: MailboxRejectClass; reason: string } | "already-delivered" | null> { - const lock: MailLock | null = await acquireMailLock(this.mailboxLockRoot); + const lock: MailLock | null = await acquireMailLock(root); if (!lock) throw new Error("mailbox lock busy; not promoted"); let relayLock: MailLock | null; try { relayLock = tryAcquireRelayMoveLock(JSON.parse(body), this.mailDir); } @@ -257,9 +260,9 @@ export class MailClient { if (!relayLock) { lock.release(); throw new Error("relay acceptance busy; not promoted"); } try { if (readFileSync(srcPath, "utf-8") !== body) throw new Error("source changed during promotion; not promoted"); - const replay = mailboxReplayStore(this.mailboxRoot); + const replay = mailboxReplayStore(root); if (replay.hasPendingFile(file)) throw new Error(`pending placement unresolved: ${file}`); - const dstPath = join(this.inboxCur, file); + const dstPath = join(root, "cur", file); if (sameFileByDeviceInode(srcPath, dstPath)) { replay.isConsumed(envelope.messageId); unlinkSync(srcPath); @@ -308,13 +311,13 @@ export class MailClient { * including an older intent without identity, leaves the intent pending and * withholds delivery; the new/ scan cannot replace that intent. */ - private async recoverInterruptedPlacements(messages: MailMessage[]): Promise { + private async recoverInterruptedPlacements(messages: MailMessage[], root: string): Promise { let pending: Array<{ messageId: string; file: string; phase: PlacementPhase }>; let listing: MailLock | null = null; try { - listing = await acquireMailLock(this.mailboxRoot); + listing = await acquireMailLock(root); if (!listing) return; // lock unavailable; defer recovery - pending = pendingPlacementEntries(this.mailboxRoot); + pending = pendingPlacementEntries(root); } catch (err) { this.events?.emit({ type: "mail.receive", @@ -331,7 +334,7 @@ export class MailClient { for (const { messageId, file, phase } of pending) { if (!resumableOnRecovery(phase)) continue; - const curPath = join(this.inboxCur, file); + const curPath = join(root, "cur", file); try { let bytes: Buffer; try { @@ -339,13 +342,13 @@ export class MailClient { } catch (err) { if ((err as NodeJS.ErrnoException).code !== "ENOENT") throw err; // Observed absent; re-check under the lock before finishing the intent. - await this.finishUnlinkedPlacement(messageId, file); + await this.finishUnlinkedPlacement(messageId, file, root); continue; } const outcome = await this.verifyMailBody(bytes.toString("utf8")); if (!outcome.pass) continue; // withhold: the cur copy is not a verified message const trustTier = verifiedMailTier(outcome.envelope, this.mailDir); - const delivered = await this.completeInterruptedPlacement(messageId, file, bytes, outcome.envelope, trustTier); + const delivered = await this.completeInterruptedPlacement(messageId, file, root, bytes, outcome.envelope, trustTier); if (delivered) messages.push(delivered); } catch (err) { this.events?.emit({ @@ -361,14 +364,14 @@ export class MailClient { } /** Finish a still-pending intent only when cur/ is absent under the lock. */ - private async finishUnlinkedPlacement(messageId: string, file: string): Promise { - const lock = await acquireMailLock(this.mailboxRoot); + private async finishUnlinkedPlacement(messageId: string, file: string, root: string): Promise { + const lock = await acquireMailLock(root); if (!lock) return; try { - const replay = mailboxReplayStore(this.mailboxRoot); + const replay = mailboxReplayStore(root); if (!replay.hasPendingPlacement(messageId, file)) return; if (!resumableOnRecovery(replay.placementPhase(file))) return; - if (lstatOrAbsent(join(this.inboxCur, file)) !== null) return; + if (lstatOrAbsent(join(root, "cur", file)) !== null) return; replay.finishPlacement(messageId, file); } finally { lock.release(); @@ -381,14 +384,15 @@ export class MailClient { private async completeInterruptedPlacement( messageId: string, file: string, + root: string, bytes: Buffer, envelope: Envelope, trustTier: MailMessage["trustTier"], ): Promise { - const lock = await acquireMailLock(this.mailboxRoot); + const lock = await acquireMailLock(root); if (!lock) throw new Error("mailbox lock busy; interrupted placement recovery deferred"); try { - const replay = mailboxReplayStore(this.mailboxRoot); + const replay = mailboxReplayStore(root); // Match the intent, recorded device/inode and pre-verification raw bytes. if (!replay.hasPendingPlacement(messageId, file)) return null; if (!resumableOnRecovery(replay.placementPhase(file))) return null; @@ -396,7 +400,7 @@ export class MailClient { if (!replay.ownsPendingPlacement(messageId, file)) return null; let current: Buffer; try { - current = readFileSync(join(this.inboxCur, file)); + current = readFileSync(join(root, "cur", file)); } catch (err) { if ((err as NodeJS.ErrnoException).code === "ENOENT") return null; throw err; @@ -424,8 +428,10 @@ export class MailClient { srcPath: string, rejected: { class: MailboxRejectClass; reason: string; from?: string }, started: number, + root: string, ): void { - const dlqPath = join(this.inboxDlq, file); + const dlqDir = join(root, "dlq"); + const dlqPath = join(dlqDir, file); let relayLock: MailLock | null = null; try { let record: unknown; @@ -433,7 +439,7 @@ export class MailClient { relayLock = tryAcquireRelayMoveLock(record, this.mailDir); if (!relayLock) return; if (srcPath !== dlqPath) renameSync(srcPath, dlqPath); - writeRejectSidecar(this.inboxDlq, file, rejected.class, rejected.reason); + writeRejectSidecar(dlqDir, file, rejected.class, rejected.reason); } catch (err) { console.error(`[MailClient] failed to dead-letter ${file}: ${sanitizeError(err)}`); } finally { relayLock?.release(); } @@ -451,6 +457,7 @@ export class MailClient { async sendMail(to: string, body: string): Promise { const started = Date.now(); try { + const root = this.mailboxRoot(); const { writeFileSync } = await import("node:fs"); // Sign the body as this agent — the v1 envelope a promote() reader // verifies. signMailBody THROWS when no key exists, so an unsigned body @@ -458,7 +465,7 @@ export class MailClient { const signed = this.signMailBody(to, body); const filename = `${Date.now()}-${Math.random().toString(36).slice(2)}.json`; writeFileSync( - join(this.outboxNew, filename), + join(root, "outbox", filename), JSON.stringify({ from: this.agentId, to, body: signed, sentAt: new Date().toISOString() }, null, 2), "utf-8" ); @@ -513,10 +520,11 @@ export class MailClient { /** Deliver all messages in outbox to recipient inboxes (local relay). */ deliverOutbox(): void { - if (!existsSync(this.outboxNew)) return; - const files = readdirSync(this.outboxNew).filter(f => !f.startsWith(".")); + const outboxNew = join(this.mailboxRoot(), "outbox"); + if (!existsSync(outboxNew)) return; + const files = readdirSync(outboxNew).filter(f => !f.startsWith(".")); for (const file of files) { - const srcPath = join(this.outboxNew, file); + const srcPath = join(outboxNew, file); try { const raw = readFileSync(srcPath, "utf-8"); const msg = JSON.parse(raw) as { to: string; body: string; sentAt: string }; diff --git a/packages/agent/test/mail-client-repoint-root.test.ts b/packages/agent/test/mail-client-repoint-root.test.ts new file mode 100644 index 00000000..a3e31497 --- /dev/null +++ b/packages/agent/test/mail-client-repoint-root.test.ts @@ -0,0 +1,107 @@ +/** + * mail-client-repoint-root.test.ts — cli#636. + * + * The promotion lock, the mailbox file paths and the replay-state path derive + * from one root resolved per operation, never pinned at construction. This test + * re-points the configured `mailDir` (a symlink) under a live MailClient, while + * a second process holds the NEW root's mailbox lock. The client must not + * promote under the OLD root's lock, and once the lock is released the one + * envelope is delivered exactly once. + */ +import { afterEach, beforeEach, expect, test } from "bun:test"; +import { spawn } from "node:child_process"; +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, symlinkSync, unlinkSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { MailClient } from "../src/io/mail.js"; +import { mailboxLockRoot } from "../src/lib/mail-lock-namespace.js"; +import { senderKeyResolver, signedRelayRecord } from "./helpers/signed-relay-record.js"; + +const AGENT = "mailbox"; +const BRANCH = "remote"; +const ID = "reserved"; + +let root: string; +let storeOld: string; +let storeNew: string; +let alias: string; +let savedHome: string | undefined; + +beforeEach(() => { + root = mkdtempSync(join(tmpdir(), "mail-client-repoint-root-")); + storeOld = join(root, "store-old"); + storeNew = join(root, "store-new"); + mkdirSync(storeOld, { recursive: true }); + mkdirSync(storeNew, { recursive: true }); + alias = join(root, "mail"); + symlinkSync(storeOld, alias, "dir"); + savedHome = process.env.HOME; + process.env.HOME = root; +}); + +afterEach(() => { + if (savedHome === undefined) delete process.env.HOME; + else process.env.HOME = savedHome; + rmSync(root, { recursive: true, force: true }); +}); + +/** Hold a mailbox lock in a real second process and release it on stdin. */ +function holdMailboxLock(lockRoot: string): Promise<{ release: () => Promise }> { + const script = join(root, "holder.ts"); + writeFileSync( + script, + ` +const { acquireMailLockSync } = await import(${JSON.stringify(new URL("../src/lib/mail-lock.js", import.meta.url).href)}); +const lock = acquireMailLockSync(process.env.HOLD_ROOT); +if (!lock) throw new Error("holder could not acquire"); +process.stdout.write("ready"); +process.stdin.once("data", () => { lock.release(); process.exit(0); }); +`, + ); + const child = spawn("bun", [script], { env: { ...process.env, HOLD_ROOT: lockRoot }, stdio: ["pipe", "pipe", "inherit"] }); + const exited = new Promise((resolve) => child.once("exit", resolve)); + return new Promise((resolve, reject) => { + child.once("error", reject); + child.once("exit", (code) => reject(new Error(`holder exited ${code} before ready`))); + child.stdout.once("data", (chunk: Buffer) => { + if (chunk.toString() === "ready") { + resolve({ release: async () => { child.stdin.end("release"); await exited; } }); + } else reject(new Error(`holder printed ${JSON.stringify(chunk.toString())}`)); + }); + }); +} + +test("a live client does not promote under the old root's lock, and delivers the envelope once", async () => { + // The client is constructed while the alias points at the old store. + const mail = new MailClient(alias, undefined, AGENT, senderKeyResolver()); + + // Re-point the mail root under the live client. The new root is a real + // mailbox: its maildir exists, and one record is seeded in new/. A client that + // resolves the root per operation reads and promotes here. + unlinkSync(alias); + symlinkSync(storeNew, alias, "dir"); + const raw = signedRelayRecord(AGENT, "repointed", { branchId: BRANCH, id: ID }); + const source = join(storeNew, AGENT, "new", "r.json"); + mkdirSync(join(storeNew, AGENT, "new"), { recursive: true }); + for (const dir of ["cur", "dlq", "outbox"]) mkdirSync(join(storeNew, AGENT, dir), { recursive: true }); + writeFileSync(source, raw); + + // A real second process holds the new root's mailbox lock. + const holder = await holdMailboxLock(mailboxLockRoot(storeNew, AGENT)); + try { + // It must not promote under the old root's lock. + expect(await mail.checkNewMail()).toEqual([]); + expect(existsSync(source)).toBe(true); + expect(existsSync(join(storeNew, AGENT, "cur", "r.json"))).toBe(false); + } finally { + await holder.release(); + } + + // After release the one envelope is delivered exactly once. + expect((await mail.checkNewMail()).map((message) => message.filename)).toEqual(["r.json"]); + expect(existsSync(source)).toBe(false); + expect(readFileSync(join(storeNew, AGENT, "cur", "r.json"), "utf8")).toBe(raw); + + // The replay gate refuses the same envelope again. + expect(await mail.checkNewMail()).toEqual([]); +}); diff --git a/packages/agent/test/mail-interrupted-recovery.test.ts b/packages/agent/test/mail-interrupted-recovery.test.ts index 048c094a..07ecbfff 100644 --- a/packages/agent/test/mail-interrupted-recovery.test.ts +++ b/packages/agent/test/mail-interrupted-recovery.test.ts @@ -10,6 +10,7 @@ import * as ed from "@noble/ed25519"; import { MailClient } from "../src/io/mail.js"; import { hasCommittedMessageId, mailboxReplayStore } from "../src/lib/mailbox-policy.js"; import { acquireMailLock } from "../src/lib/mail-lock.js"; +import { mailboxLockRoot } from "../src/lib/mail-lock-namespace.js"; import { signEnvelope, type FlairClient } from "../src/lib/signEnvelope.js"; const AGENT = "agent-a"; @@ -232,8 +233,8 @@ describe("MailClient defers unresolved placements", () => { const client = new MailClient(mailDir, undefined, AGENT, verifier()); mailboxReplayStore(mailboxRoot).beginPlacement(messageId, "m1.json", source); copyFileSync(source, curPath()); - await (client as unknown as { finishUnlinkedPlacement(id: string, file: string): Promise }) - .finishUnlinkedPlacement(messageId, "m1.json"); + await (client as unknown as { finishUnlinkedPlacement(id: string, file: string, root: string): Promise }) + .finishUnlinkedPlacement(messageId, "m1.json", mailboxLockRoot(mailDir, AGENT)); expect(mailboxReplayStore(mailboxRoot).hasPendingPlacement(messageId, "m1.json")).toBe(true); }); diff --git a/packages/agent/test/mail-promote-guard.test.ts b/packages/agent/test/mail-promote-guard.test.ts index f59e4b44..1531380e 100644 --- a/packages/agent/test/mail-promote-guard.test.ts +++ b/packages/agent/test/mail-promote-guard.test.ts @@ -28,6 +28,7 @@ import { parseFlairPublicKey } from "../src/lib/public-key.js"; import { FlairContextProvider } from "../src/io/flair.js"; import { MailClient } from "../src/io/mail.js"; import { acquireMailLockSync } from "../src/lib/mail-lock.js"; +import { mailboxLockRoot } from "../src/lib/mail-lock-namespace.js"; import { mailboxReplayStore } from "../src/lib/mailbox-policy.js"; const AGENT = "mailbox"; @@ -546,7 +547,7 @@ describe("agent MailClient promotion is fail-closed (cli#380 F1)", () => { return rename(...args); }); try { - await expect(client["commitToCur"]("m1.json", join(inbox("new"), "m1.json"), + await expect(client["commitToCur"](mailboxLockRoot(tmpDir, AGENT), "m1.json", join(inbox("new"), "m1.json"), readFileSync(join(inbox("new"), "m1.json"), "utf-8"), env)).rejects.toThrow(/append fault.*rollback fault/); expect(readFileSync(join(tmpDir, AGENT, "consumed.jsonl"), "utf-8")).toBe(""); } finally {