Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .changelog/unreleased/fixed-mailclient-one-mail-root.md
Original file line number Diff line number Diff line change
@@ -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.
102 changes: 55 additions & 47 deletions packages/agent/src/io/mail.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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<MailMessage[]> {
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 {
Expand Down Expand Up @@ -186,7 +188,7 @@ export class MailClient {
}

if (!verifyResult.pass) {
this.deadLetter(file, srcPath, verifyResult, started);
this.deadLetter(file, srcPath, verifyResult, started, root);
continue;
}

Expand All @@ -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;
Expand Down Expand Up @@ -244,22 +246,23 @@ 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); }
catch (error) { lock.release(); throw error; }
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);
Expand Down Expand Up @@ -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<void> {
private async recoverInterruptedPlacements(messages: MailMessage[], root: string): Promise<void> {
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",
Expand All @@ -331,21 +334,21 @@ 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 {
bytes = readFileSync(curPath);
} 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({
Expand All @@ -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<void> {
const lock = await acquireMailLock(this.mailboxRoot);
private async finishUnlinkedPlacement(messageId: string, file: string, root: string): Promise<void> {
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();
Expand All @@ -381,22 +384,23 @@ export class MailClient {
private async completeInterruptedPlacement(
messageId: string,
file: string,
root: string,
bytes: Buffer,
envelope: Envelope,
trustTier: MailMessage["trustTier"],
): Promise<MailMessage | null> {
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;
if (envelope.messageId !== messageId) return null;
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;
Expand Down Expand Up @@ -424,16 +428,18 @@ 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;
try { record = JSON.parse(readFileSync(srcPath, "utf-8")); } catch {}
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(); }
Expand All @@ -451,14 +457,15 @@ export class MailClient {
async sendMail(to: string, body: string): Promise<void> {
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
// is never written.
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"
);
Expand Down Expand Up @@ -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 };
Expand Down
107 changes: 107 additions & 0 deletions packages/agent/test/mail-client-repoint-root.test.ts
Original file line number Diff line number Diff line change
@@ -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<void> }> {
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<number | null>((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([]);
});
5 changes: 3 additions & 2 deletions packages/agent/test/mail-interrupted-recovery.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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<void> })
.finishUnlinkedPlacement(messageId, "m1.json");
await (client as unknown as { finishUnlinkedPlacement(id: string, file: string, root: string): Promise<void> })
.finishUnlinkedPlacement(messageId, "m1.json", mailboxLockRoot(mailDir, AGENT));
expect(mailboxReplayStore(mailboxRoot).hasPendingPlacement(messageId, "m1.json")).toBe(true);
});

Expand Down
Loading
Loading