From d824fd28e73ebc969409fe946a7ae92dca7e9996 Mon Sep 17 00:00:00 2001 From: "rayson951005@gmail.com" Date: Fri, 26 Jun 2026 12:30:56 +0800 Subject: [PATCH] =?UTF-8?q?feat(resilience):=20broker=20/healthz=20+=20?= =?UTF-8?q?=E4=BC=98=E9=9B=85=E5=85=B3=E9=97=AD=20+=20pending=20=E4=B8=8A?= =?UTF-8?q?=E9=99=90=20+=20=E9=87=8D=E8=BF=9E=20jitter=20(PR11/=C2=A78.2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit v3 §11.1 bullet 12 / §8.2 A 类韧性。无感重连(PR4)+ 落盘续投(PR6 store_if_offline + SQLite WAL)核心已就位,本 PR 补四个隔离的硬化点。 v3 §11.1 bullet 12 / §8.2 A-class resilience. Seamless reconnect (PR4) + durable replay (PR6 store_if_offline + SQLite WAL) are already in place; this PR hardens the four remaining isolated gaps. - broker /healthz:GET /healthz 返回最小非敏感 JSON(ok/pid/uptimeMs/connections), watchdog/supervisor 的存活探针。绑 Tailscale 100.x 任意 tailnet 节点可达,故 严禁含 token/PII/identity(测试断言不泄漏)。 - 优雅关闭(cli/broker.ts):SIGTERM/SIGINT → broker.stop() + store.close() (checkpoint WAL,保证 pending_deliveries 跨重启存活)+ exit 0。once 防重入。 - pending_deliveries 每-target 上限 MAX_PENDING_PER_TARGET=1000(drop-oldest, 两实现一致):永不重连的成员不会把表撑爆。 - 重连 jitter(broker-client.ts):scheduleReconnect 改用 reconnectDelay 的 equal-jitter(ceiling/2 + rand·ceiling/2,≤ maxMs),避免同时掉线的多 adapter 锁步重连打爆 broker(thundering herd)。 Tests: reconnect-delay (3, 纯函数 jitter 数学:floor/上限/不超 maxMs) + broker-healthz (2: 200+非敏感 body / connections 随连接增减) + broker-graceful-shutdown (1: spawn 真 CLI → SIGTERM → exit 0) + store 契约 pending 上限 (两实现各跑)。check 全绿 1803 pass。 Backlog(§8.2 推后,文档/注释标明):drain 即删=at-most-once(ACK delete-after-ack 推后)、WS 心跳(半开/漫游检测)、自建 watchdog 自动拉起(先用 systemd Restart=always)、 关机前 broker_shutdown 通知、无感重连后抑制重复 member_joined。 🤖 Generated with [Claude Code](https://claude.com/claude-code) --- plugins/agentbridge/server/bridge-server.js | 4 +- plugins/agentbridge/server/daemon.js | 22 ++++++-- src/backbone/store.ts | 12 +++++ src/backbone/store/memory-store.ts | 6 +++ src/backbone/store/sqlite-store.ts | 10 ++++ src/broker-client.ts | 19 ++++++- src/broker.ts | 26 ++++++++- src/cli/broker.ts | 17 +++++- .../broker-graceful-shutdown.test.ts | 47 ++++++++++++++++ src/integration-test/broker-healthz.test.ts | 54 +++++++++++++++++++ src/unit-test/reconnect-delay.test.ts | 29 ++++++++++ src/unit-test/store-contract.ts | 15 ++++++ src/unit-test/store-sqlite-durable.test.ts | 48 +++++++++++++++++ 13 files changed, 299 insertions(+), 10 deletions(-) create mode 100644 src/integration-test/broker-graceful-shutdown.test.ts create mode 100644 src/integration-test/broker-healthz.test.ts create mode 100644 src/unit-test/reconnect-delay.test.ts create mode 100644 src/unit-test/store-sqlite-durable.test.ts diff --git a/plugins/agentbridge/server/bridge-server.js b/plugins/agentbridge/server/bridge-server.js index 06b07e4..01097cc 100755 --- a/plugins/agentbridge/server/bridge-server.js +++ b/plugins/agentbridge/server/bridge-server.js @@ -14707,10 +14707,10 @@ function defineNumber(value, fallback) { } var BUILD_INFO = Object.freeze({ version: defineString("0.1.24", "0.0.0-source"), - commit: defineString("8284e8a", "source"), + commit: defineString("b202e8f", "source"), bundle: defineBundle("plugin"), contractVersion: defineNumber(1, CONTRACT_VERSION), - codeHash: defineString("0a63b984bf2a", "source") + codeHash: defineString("7c98560c206e", "source") }); function sameRuntimeContract(a, b) { if (!a || !b) diff --git a/plugins/agentbridge/server/daemon.js b/plugins/agentbridge/server/daemon.js index 960ce5e..ef9f9a6 100755 --- a/plugins/agentbridge/server/daemon.js +++ b/plugins/agentbridge/server/daemon.js @@ -30,10 +30,10 @@ function defineNumber(value, fallback) { } var BUILD_INFO = Object.freeze({ version: defineString("0.1.24", "0.0.0-source"), - commit: defineString("8284e8a", "source"), + commit: defineString("b202e8f", "source"), bundle: defineBundle("plugin"), contractVersion: defineNumber(1, CONTRACT_VERSION), - codeHash: defineString("0a63b984bf2a", "source") + codeHash: defineString("7c98560c206e", "source") }); function daemonStatusBuildInfo() { return { ...BUILD_INFO }; @@ -6811,6 +6811,11 @@ class RoomManager { } // src/broker-client.ts +function reconnectDelay(baseMs, maxMs, attempt, rand) { + const ceiling = Math.min(maxMs, baseMs * 2 ** attempt); + return ceiling / 2 + rand * (ceiling / 2); +} + class BrokerClient { opts; ws = null; @@ -6830,6 +6835,7 @@ class BrokerClient { baseMs; maxMs; maxOutbox; + rand; constructor(opts) { this.opts = opts; this.log = opts.log ?? (() => {}); @@ -6837,6 +6843,7 @@ class BrokerClient { this.baseMs = opts.reconnectBaseMs ?? 250; this.maxMs = opts.reconnectMaxMs ?? 1e4; this.maxOutbox = opts.maxOutbox ?? 1000; + this.rand = opts.random ?? Math.random; } get connected() { return this.ws !== null && this.ws.readyState === WebSocket.OPEN && this.identity !== null; @@ -6990,9 +6997,9 @@ class BrokerClient { scheduleReconnect() { if (this.closed || this.reconnectTimer) return; - const delay = Math.min(this.maxMs, this.baseMs * 2 ** this.reconnectAttempt); + const delay = reconnectDelay(this.baseMs, this.maxMs, this.reconnectAttempt, this.rand()); this.reconnectAttempt++; - this.log(`reconnecting in ${delay}ms (attempt ${this.reconnectAttempt})`); + this.log(`reconnecting in ${Math.round(delay)}ms (attempt ${this.reconnectAttempt})`); this.reconnectTimer = setTimeout(() => { this.reconnectTimer = null; if (this.closed) @@ -7065,6 +7072,10 @@ import { dirname as dirname3, join as join11 } from "path"; // src/backbone/store/sqlite-store.ts import { Database } from "bun:sqlite"; +// src/backbone/store.ts +var MAX_PENDING_PER_TARGET = 1000; + +// src/backbone/store/sqlite-store.ts class SqliteStore { db; closed = false; @@ -7203,6 +7214,9 @@ class SqliteStore { } async enqueuePending(targetAgentId, envelope) { this.db.query("INSERT OR IGNORE INTO pending_deliveries(target_agent_id, idempotency_key, envelope) VALUES(?, ?, ?)").run(targetAgentId, envelope.idempotencyKey, JSON.stringify(envelope)); + this.db.query(`DELETE FROM pending_deliveries WHERE target_agent_id=? AND seq NOT IN ( + SELECT seq FROM pending_deliveries WHERE target_agent_id=? ORDER BY seq DESC LIMIT ? + )`).run(targetAgentId, targetAgentId, MAX_PENDING_PER_TARGET); } async drainPending(targetAgentId) { const rows = this.db.query("SELECT envelope FROM pending_deliveries WHERE target_agent_id=? ORDER BY seq").all(targetAgentId); diff --git a/src/backbone/store.ts b/src/backbone/store.ts index ff51765..6cbb5d9 100644 --- a/src/backbone/store.ts +++ b/src/backbone/store.ts @@ -1,5 +1,12 @@ import type { Envelope } from "./envelope"; +/** + * Max queued envelopes per offline target before the oldest is dropped (§8.2). + * Mirrors the BrokerClient outbox bound: logged, bounded loss beats unbounded + * growth for a member that never comes back. + */ +export const MAX_PENDING_PER_TARGET = 1000; + /** * Store interface (spec §6.1, §12 data model). * @@ -83,6 +90,11 @@ export interface Store { saveWhiteboard(roomId: string, whiteboard: WhiteboardRecord): Promise; // --- pending_deliveries: offline replay queue (§3.2), dedup by idempotencyKey --- + /** + * Queue an envelope for an offline target. Bounded per target at + * {@link MAX_PENDING_PER_TARGET} (drop-oldest) so a member that never reconnects + * can't grow the backlog without limit (§8.2 resilience). + */ enqueuePending(targetAgentId: string, envelope: Envelope): Promise; /** Remove and return the target's pending envelopes (deduped by idempotencyKey). */ drainPending(targetAgentId: string): Promise; diff --git a/src/backbone/store/memory-store.ts b/src/backbone/store/memory-store.ts index 8429ab8..8da9a76 100644 --- a/src/backbone/store/memory-store.ts +++ b/src/backbone/store/memory-store.ts @@ -1,4 +1,5 @@ import type { Envelope } from "../envelope"; +import { MAX_PENDING_PER_TARGET } from "../store"; import type { AgentRecord, IdentityRecord, @@ -140,6 +141,11 @@ export class InMemoryStore implements Store { if (!byKey.has(envelope.idempotencyKey)) { byKey.set(envelope.idempotencyKey, envelope); // dedup: first wins } + // Bound the per-target backlog (§8.2): Map preserves insertion order, so drop + // from the front (oldest) until within MAX_PENDING_PER_TARGET. Matches SqliteStore. + while (byKey.size > MAX_PENDING_PER_TARGET) { + byKey.delete(byKey.keys().next().value as string); + } } async drainPending(targetAgentId: string): Promise { diff --git a/src/backbone/store/sqlite-store.ts b/src/backbone/store/sqlite-store.ts index 5ea15ba..8e98398 100644 --- a/src/backbone/store/sqlite-store.ts +++ b/src/backbone/store/sqlite-store.ts @@ -1,5 +1,6 @@ import { Database } from "bun:sqlite"; import type { Envelope } from "../envelope"; +import { MAX_PENDING_PER_TARGET } from "../store"; import type { AgentRecord, IdentityRecord, @@ -242,6 +243,15 @@ export class SqliteStore implements Store { "INSERT OR IGNORE INTO pending_deliveries(target_agent_id, idempotency_key, envelope) VALUES(?, ?, ?)", ) .run(targetAgentId, envelope.idempotencyKey, JSON.stringify(envelope)); + // Bound the per-target backlog (§8.2): keep the newest MAX_PENDING_PER_TARGET, + // drop the oldest beyond it, so a never-reconnecting member can't grow the table. + this.db + .query( + `DELETE FROM pending_deliveries WHERE target_agent_id=? AND seq NOT IN ( + SELECT seq FROM pending_deliveries WHERE target_agent_id=? ORDER BY seq DESC LIMIT ? + )`, + ) + .run(targetAgentId, targetAgentId, MAX_PENDING_PER_TARGET); } async drainPending(targetAgentId: string): Promise { diff --git a/src/broker-client.ts b/src/broker-client.ts index 2940bcb..a09c71c 100644 --- a/src/broker-client.ts +++ b/src/broker-client.ts @@ -16,10 +16,23 @@ export interface BrokerClientOptions { maxOutbox?: number; /** WebSocket factory — injectable so tests can drive reconnect without a real socket. */ wsFactory?: (url: string) => WebSocket; + /** Randomness source for reconnect jitter [0,1) — injectable so tests are deterministic. */ + random?: () => number; } type EventHandler = (topic: string, envelope: Envelope) => void; +/** + * Reconnect backoff with EQUAL JITTER (§8.2). `ceiling = min(maxMs, baseMs·2^attempt)`; + * the delay is `ceiling/2 + rand·ceiling/2`, i.e. uniformly in `[ceiling/2, ceiling]`. + * Half-fixed keeps a sane minimum wait; half-random de-synchronises adapters that + * dropped together (no thundering herd). Result is always `≤ ceiling ≤ maxMs`. + */ +export function reconnectDelay(baseMs: number, maxMs: number, attempt: number, rand: number): number { + const ceiling = Math.min(maxMs, baseMs * 2 ** attempt); + return ceiling / 2 + rand * (ceiling / 2); +} + /** * Edge-side client to the control-plane broker (§5 adapter transport + §8.2 * resilience foundation). @@ -55,6 +68,7 @@ export class BrokerClient { private readonly baseMs: number; private readonly maxMs: number; private readonly maxOutbox: number; + private readonly rand: () => number; constructor(private readonly opts: BrokerClientOptions) { this.log = opts.log ?? (() => {}); @@ -62,6 +76,7 @@ export class BrokerClient { this.baseMs = opts.reconnectBaseMs ?? 250; this.maxMs = opts.reconnectMaxMs ?? 10_000; this.maxOutbox = opts.maxOutbox ?? 1000; + this.rand = opts.random ?? Math.random; } get connected(): boolean { @@ -237,9 +252,9 @@ export class BrokerClient { private scheduleReconnect(): void { if (this.closed || this.reconnectTimer) return; - const delay = Math.min(this.maxMs, this.baseMs * 2 ** this.reconnectAttempt); + const delay = reconnectDelay(this.baseMs, this.maxMs, this.reconnectAttempt, this.rand()); this.reconnectAttempt++; - this.log(`reconnecting in ${delay}ms (attempt ${this.reconnectAttempt})`); + this.log(`reconnecting in ${Math.round(delay)}ms (attempt ${this.reconnectAttempt})`); this.reconnectTimer = setTimeout(() => { this.reconnectTimer = null; if (this.closed) return; diff --git a/src/broker.ts b/src/broker.ts index a58dc71..1144bec 100644 --- a/src/broker.ts +++ b/src/broker.ts @@ -72,6 +72,10 @@ export interface BrokerOptions { export class Broker { private server: ReturnType | null = null; private nextConnId = 0; + /** Live WS connections (incremented on upgrade, decremented on close) — for /healthz. */ + private liveConnections = 0; + /** Epoch ms the server started, for /healthz uptime. 0 until start(). */ + private startedAt = 0; /** topic → (identityId → live-subscription count) — who is reachable per topic. */ private readonly topicMembers = new Map>(); private readonly transport: MessageTransport; @@ -90,14 +94,23 @@ export class Broker { // loopback rather than become an all-interfaces bind (`Bun.serve({hostname:""})`). const host = this.opts.host || "127.0.0.1"; const port = this.opts.port ?? DEFAULT_BROKER_PORT; + this.startedAt = Date.now(); // eslint-disable-next-line @typescript-eslint/no-this-alias const self = this; const server = Bun.serve({ hostname: host, port, fetch(req, server) { - if (new URL(req.url).pathname === "/ws") { + const pathname = new URL(req.url).pathname; + if (pathname === "/healthz") { + // Liveness probe for a watchdog/supervisor (§8.2). Minimal, NON-SENSITIVE + // body — the broker binds a Tailscale 100.x address reachable by any + // tailnet node, so this must never leak tokens/PII/identities. + return Response.json(self.healthBody()); + } + if (pathname === "/ws") { if (server.upgrade(req, { data: { connId: ++self.nextConnId, subs: new Map() } })) { + self.liveConnections++; return undefined; } } @@ -124,6 +137,7 @@ export class Broker { } for (const unsub of ws.data.subs.values()) unsub(); ws.data.subs.clear(); + if (self.liveConnections > 0) self.liveConnections--; self.log(`conn #${ws.data.connId} closed`); }, }, @@ -133,6 +147,16 @@ export class Broker { return { host, port: server.port ?? port }; } + /** Non-sensitive liveness body for GET /healthz (§8.2 watchdog). No tokens/PII/identities. */ + private healthBody(): { ok: true; pid: number; uptimeMs: number; connections: number } { + return { + ok: true, + pid: process.pid, + uptimeMs: this.startedAt === 0 ? 0 : Date.now() - this.startedAt, + connections: this.liveConnections, + }; + } + stop(): void { this.server?.stop(true); this.server = null; diff --git a/src/cli/broker.ts b/src/cli/broker.ts index f8bad45..14f002a 100644 --- a/src/cli/broker.ts +++ b/src/cli/broker.ts @@ -91,7 +91,22 @@ export async function runBrokerStart(argv: string[]): Promise { console.log(`AgentBridge broker 已启动，监听 ${bound.host}:${bound.port}`); console.log(`协作数据库：${dbPath}`); console.log("用 abg auth login 签发的 token 连接；Ctrl-C 停止。"); - // Bun.serve keeps the event loop alive — the process stays up until killed. + + // Graceful shutdown (§8.2): on SIGTERM/SIGINT, stop the server then close the + // Store — db.close() checkpoints the WAL so pending_deliveries survive the + // restart and a reconnecting member can still drain them. `once` so a second + // signal during teardown doesn't re-enter; exit 0 even if close() rejects. + let stopping = false; + const shutdown = (sig: string) => { + if (stopping) return; + stopping = true; + console.error(`[broker] ${sig} 收到，正在优雅关闭…`); + broker.stop(); + store.close().finally(() => process.exit(0)); + }; + process.once("SIGTERM", () => shutdown("SIGTERM")); + process.once("SIGINT", () => shutdown("SIGINT")); + // Bun.serve keeps the event loop alive — the process stays up until a signal. } export async function runBroker(args: string[]): Promise { diff --git a/src/integration-test/broker-graceful-shutdown.test.ts b/src/integration-test/broker-graceful-shutdown.test.ts new file mode 100644 index 0000000..b0bcf98 --- /dev/null +++ b/src/integration-test/broker-graceful-shutdown.test.ts @@ -0,0 +1,47 @@ +import { describe, test, expect, afterEach } from "bun:test"; +import { spawn } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +const CLI_PATH = fileURLToPath(new URL("../cli.ts", import.meta.url)); + +describe("abg broker start — graceful shutdown (§8.2)", () => { + let dir: string | undefined; + afterEach(() => { + if (dir) rmSync(dir, { recursive: true, force: true }); + dir = undefined; + }); + + test("SIGTERM stops the server, closes the Store, and exits 0", async () => { + dir = mkdtempSync(join(tmpdir(), "agentbridge-broker-sig-")); + const dbPath = join(dir, "collab.db"); + const child = spawn(process.execPath, ["run", CLI_PATH, "broker", "start", "--port", "0", "--db", dbPath], { + env: { ...process.env, AGENTBRIDGE_COLLAB_DB: dbPath }, + stdio: ["ignore", "pipe", "pipe"], + }); + + // Wait for the broker to finish binding before signalling (so the SIGTERM + // handler is installed — it's registered right after start()). + await new Promise((resolve, reject) => { + const to = setTimeout(() => reject(new Error("broker did not start in time")), 8000); + child.stdout.on("data", (d: Buffer) => { + if (String(d).includes("已启动")) { + clearTimeout(to); + resolve(); + } + }); + child.once("exit", (c) => { + clearTimeout(to); + reject(new Error(`broker exited before startup (code ${c})`)); + }); + }); + + const exitCode = await new Promise((resolve) => { + child.once("exit", (code) => resolve(code)); + child.kill("SIGTERM"); + }); + expect(exitCode).toBe(0); // graceful: exit 0, not killed by the signal + }, 15000); +}); diff --git a/src/integration-test/broker-healthz.test.ts b/src/integration-test/broker-healthz.test.ts new file mode 100644 index 0000000..fee156f --- /dev/null +++ b/src/integration-test/broker-healthz.test.ts @@ -0,0 +1,54 @@ +import { describe, test, expect, afterEach } from "bun:test"; +import { Broker } from "../broker"; +import { InMemoryStore } from "../backbone/store/memory-store"; +import { IdentityService } from "../backbone/identity-service"; +import { StorePskIdentityProvider } from "../backbone/identity/store-psk-identity-provider"; + +describe("Broker /healthz — liveness probe (§8.2 watchdog)", () => { + let stop: (() => void) | undefined; + afterEach(() => { + stop?.(); + stop = undefined; + }); + + async function start() { + const store = new InMemoryStore(); + const svc = new IdentityService(store); + await svc.registerIdentity("alice@x.com", "Alice"); + const token = await svc.issueToken("alice@x.com"); + const broker = new Broker({ store, identityProvider: new StorePskIdentityProvider(store), host: "127.0.0.1", port: 0, log: () => {} }); + const { port } = broker.start(); + stop = () => broker.stop(); + return { port, token, base: `http://127.0.0.1:${port}` }; + } + + test("GET /healthz returns 200 + a minimal non-sensitive JSON body", async () => { + const { base, token } = await start(); + const res = await fetch(`${base}/healthz`); + expect(res.status).toBe(200); + const body = await res.json(); + expect(body.ok).toBe(true); + expect(typeof body.pid).toBe("number"); + expect(typeof body.uptimeMs).toBe("number"); + expect(body.uptimeMs).toBeGreaterThanOrEqual(0); + expect(body.connections).toBe(0); // no ws connected yet + // must NOT leak secrets/PII + const raw = JSON.stringify(body); + expect(raw).not.toContain(token); + expect(raw).not.toContain("alice@x.com"); + }); + + test("connections reflects a live ws and drops when it closes", async () => { + const { base, port, token } = await start(); + const ws = new WebSocket(`ws://127.0.0.1:${port}/ws`); + await new Promise((res, rej) => { + ws.onopen = () => res(); + ws.onerror = () => rej(new Error("ws failed")); + }); + await new Promise((r) => setTimeout(r, 30)); + expect((await (await fetch(`${base}/healthz`)).json()).connections).toBe(1); + ws.close(); + await new Promise((r) => setTimeout(r, 60)); + expect((await (await fetch(`${base}/healthz`)).json()).connections).toBe(0); + }); +}); diff --git a/src/unit-test/reconnect-delay.test.ts b/src/unit-test/reconnect-delay.test.ts new file mode 100644 index 0000000..10ec3bb --- /dev/null +++ b/src/unit-test/reconnect-delay.test.ts @@ -0,0 +1,29 @@ +import { describe, test, expect } from "bun:test"; +import { reconnectDelay } from "../broker-client"; + +describe("reconnectDelay — equal-jitter backoff (§8.2)", () => { + test("rand=0 ⇒ ceiling/2 (the fixed floor); rand→1 ⇒ approaches ceiling", () => { + // attempt 0, base 100: ceiling = min(max, 100) = 100 + expect(reconnectDelay(100, 10_000, 0, 0)).toBe(50); + expect(reconnectDelay(100, 10_000, 0, 1)).toBe(100); + expect(reconnectDelay(100, 10_000, 0, 0.5)).toBe(75); + }); + + test("grows exponentially with attempt until clamped at maxMs", () => { + expect(reconnectDelay(100, 10_000, 1, 0)).toBe(100); // ceiling 200 → /2 + expect(reconnectDelay(100, 10_000, 2, 0)).toBe(200); // ceiling 400 → /2 + // attempt 10: 100*1024=102400 clamped to maxMs 10000 → /2 = 5000 + expect(reconnectDelay(100, 10_000, 10, 0)).toBe(5000); + expect(reconnectDelay(100, 10_000, 10, 1)).toBe(10_000); + }); + + test("never exceeds maxMs for any rand in [0,1)", () => { + for (const attempt of [0, 1, 5, 20, 100]) { + for (const r of [0, 0.3, 0.7, 0.9999]) { + const d = reconnectDelay(250, 10_000, attempt, r); + expect(d).toBeGreaterThanOrEqual(0); + expect(d).toBeLessThanOrEqual(10_000); + } + } + }); +}); diff --git a/src/unit-test/store-contract.ts b/src/unit-test/store-contract.ts index 993bf6b..d13ed34 100644 --- a/src/unit-test/store-contract.ts +++ b/src/unit-test/store-contract.ts @@ -1,4 +1,5 @@ import { describe, test, expect, beforeEach, afterEach } from "bun:test"; +import { MAX_PENDING_PER_TARGET } from "../backbone/store"; import type { Store } from "../backbone/store"; import { makeEnvelope } from "./backbone-fixtures"; @@ -124,6 +125,20 @@ export function runStoreContract(label: string, makeStore: () => Store) { expect((await store.drainPending("ag-3")).length).toBe(1); // other target intact }); + test("pending deliveries are bounded per target (§8.2): oldest dropped beyond MAX_PENDING_PER_TARGET", async () => { + const total = MAX_PENDING_PER_TARGET + 5; + for (let i = 0; i < total; i++) { + await store.enqueuePending("ag-cap", makeEnvelope({ idempotencyKey: `k${i}`, messageId: `m${i}` })); + } + const drained = await store.drainPending("ag-cap"); + expect(drained.length).toBe(MAX_PENDING_PER_TARGET); // capped + const keys = new Set(drained.map((e) => e.idempotencyKey)); + expect(keys.has("k0")).toBe(false); // oldest 5 dropped + expect(keys.has("k4")).toBe(false); + expect(keys.has("k5")).toBe(true); // newest kept + expect(keys.has(`k${total - 1}`)).toBe(true); + }); + test("auth tokens issue / resolve / list, re-issue re-points", async () => { expect(await store.resolveToken("tok-1")).toBeNull(); await store.issueToken("tok-1", "alice@x.com"); diff --git a/src/unit-test/store-sqlite-durable.test.ts b/src/unit-test/store-sqlite-durable.test.ts new file mode 100644 index 0000000..297d80c --- /dev/null +++ b/src/unit-test/store-sqlite-durable.test.ts @@ -0,0 +1,48 @@ +import { describe, test, expect, afterEach } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { SqliteStore } from "../backbone/store/sqlite-store"; +import { makeEnvelope } from "./backbone-fixtures"; + +// §8.2 durability: pending_deliveries (and the ledger) must survive a broker +// restart so a reconnecting member still drains what was queued while it was +// offline. The WAL is checkpointed on close(); reopening the SAME db file is the +// in-test stand-in for "stop the broker, start it again". +describe("SqliteStore — durability across close/reopen (§8.2)", () => { + let dir: string | undefined; + afterEach(() => { + if (dir) rmSync(dir, { recursive: true, force: true }); + dir = undefined; + }); + + test("pending deliveries enqueued before close are drainable after reopen", async () => { + dir = mkdtempSync(join(tmpdir(), "agentbridge-durable-")); + const dbPath = join(dir, "collab.db"); + + const s1 = new SqliteStore(dbPath); + await s1.enqueuePending("ag-9", makeEnvelope({ idempotencyKey: "k1", messageId: "m1" })); + await s1.enqueuePending("ag-9", makeEnvelope({ idempotencyKey: "k2", messageId: "m2" })); + await s1.close(); // checkpoints WAL — the graceful-shutdown path + + // "restart": a fresh Store over the same file. + const s2 = new SqliteStore(dbPath); + const drained = await s2.drainPending("ag-9"); + expect(drained.map((e) => e.idempotencyKey).sort()).toEqual(["k1", "k2"]); + await s2.close(); + }); + + test("ledger events survive a reopen", async () => { + dir = mkdtempSync(join(tmpdir(), "agentbridge-durable-")); + const dbPath = join(dir, "collab.db"); + + const s1 = new SqliteStore(dbPath); + await s1.appendEvent("r1", makeEnvelope({ messageId: "e1", timestamp: 1 })); + await s1.close(); + + const s2 = new SqliteStore(dbPath); + const events = await s2.getRecentEvents("r1", 10); + expect(events.map((e) => e.messageId)).toContain("e1"); + await s2.close(); + }); +});