feat(relay): isolate optional presence on the shared broker transport

dev/policy-relay.test.mjs

@@ -0,0 +1,110 @@
+import { test, expect } from "vitest";
+import { finalizeEvent, generateSecretKey, getPublicKey } from "nostr-tools";
+import { createHash } from "node:crypto";
+import { policyRelay } from "../tests/browser/policy-relay.mjs";
+
+// Positive controls for the modeled upstream used by the browser measurements.
+// No actual app or deployed relay is involved in these enforcement checks.
+test("modeled presence quota enforcement rejects overload across sockets and HTTP purposes", async () => {
+  const key = generateSecretKey(),
+    viewer = getPublicKey(key);
+  const report = { queries: [], unexpected: [] };
+  let clock = 1000;
+  const sign = (kind, tags = [], content = "online") =>
+    finalizeEvent(
+      { kind, tags, content, created_at: Math.floor(clock / 1000) },
+      key,
+    );
+  const relay = policyRelay({
+    viewer,
+    report,
+    pending: [],
+    answer: () => [],
+    presenceSnapshot: () => {},
+    acceptPublication: () => {},
+    enforceQuotas: true,
+    now: () => clock,
+  });
+  async function post(body, path = "/query", community = "primary") {
+    const url = `https://${community}.example${path}`;
+    const payload = JSON.stringify(body);
+    const auth = sign(27235, [
+      ["u", url],
+      ["payload", createHash("sha256").update(payload).digest("hex")],
+    ]);
+    return relay.fetch(url, {
+      body: payload,
+      headers: {
+        Authorization: `Nostr ${Buffer.from(JSON.stringify(auth)).toString("base64")}`,
+      },
+      signal: new AbortController().signal,
+    });
+  }
+  for (let i = 0; i < 300; i++) {
+    const filters =
+      i % 2
+        ? [{ kinds: [20001], authors: [viewer], limit: 1 }]
+        : [{ kinds: [9], "#h": ["alpha"], limit: 1 }];
+    expect((await post(filters)).status).toBe(200);
+  }
+  const denied = await post(sign(9, [["h", "alpha"]], "over quota"), "/events");
+  expect(denied.status).toBe(429);
+  expect((await denied.json()).error).toContain("rate-limited:");
+  expect(report.quotaRefusals.at(-1).category).toBe("ApiCalls");
+  expect(
+    (await post([{ kinds: [9], limit: 1 }], "/query", "secondary")).status,
+  ).toBe(200);
+  clock += 60000;
+  expect((await post([{ kinds: [9], limit: 1 }])).status).toBe(200);
+
+  const sockets = Array.from({ length: 8 }, () =>
+    relay.socket("wss://primary.example"),
+  );
+  for (const socket of sockets)
+    socket.send(
+      JSON.stringify(["AUTH", sign(22242, [["challenge", "policy-fixture"]])]),
+    );
+  const frames = [];
+  for (const socket of sockets)
+    socket.onmessage = ({ data }) => frames.push(JSON.parse(data));
+  for (let i = 0; i < 51; i++)
+    sockets[i % 8].send(
+      JSON.stringify([
+        "REQ",
+        `request-${i}`,
+        i % 2
+          ? { kinds: [20001], authors: [viewer], limit: 0 }
+          : { kinds: [9], "#h": ["alpha"], limit: 1 },
+      ]),
+    );
+  await Promise.resolve();
+  expect(relay.requests).toHaveLength(50);
+  expect(frames).toContainEqual([
+    "CLOSED",
+    "request-50",
+    expect.stringContaining("rate-limited:"),
+  ]);
+  expect(report.quotaRefusals.at(-1).category).toBe("WsEvents");
+  // Separate message quota: stay below each 50/5s WS burst but send 61 EVENTs/min.
+  clock += 5000;
+  for (let i = 0; i < 61; i++) {
+    if (i === 30) clock += 5000;
+    sockets[i % 8].send(JSON.stringify(["EVENT", sign(20001)]));
+  }
+  await Promise.resolve();
+  expect(report.presencePublications).toHaveLength(60);
+  expect(report.quotaRefusals.at(-1).category).toBe("Messages");
+  expect(frames).toContainEqual([
+    "OK",
+    expect.any(String),
+    false,
+    expect.stringContaining("rate-limited:"),
+  ]);
+  expect(report.quotaRefusals.map(({ category }) => category)).toEqual([
+    "ApiCalls",
+    "WsEvents",
+    "Messages",
+  ]);
+  expect(report.unexpected).toEqual([]);
+  for (const socket of sockets) socket.close();
+});

dev/relay-broker-api.test.mjs

@@ -407,3 +407,120 @@ test("both real sign and publish routes admit direct replies but reject arbitrar
     await h.close();
   }
 });
+
+const presenceSnapshot = [
+  { kinds: [20001], authors: ["a".repeat(64)], limit: 1 },
+];
+test("a held presence snapshot is additive to all six ordinary broker slots, not freed by browser abort", async () => {
+  let release;
+  const held = new Promise((resolve) => {
+    release = resolve;
+  });
+  const h = await harness(async (call, count, event) => {
+    await held;
+    return success(call, count, event);
+  });
+  const controller = new AbortController();
+  const pending = [];
+  try {
+    const p = h.post("query", presenceSnapshot, controller.signal);
+    const stopped = expect(p).rejects.toThrow();
+    await vi.waitFor(() => expect(h.calls).toHaveLength(1));
+    // Six ordinary requests truly reach the held upstream, not merely the admission queue.
+    for (let i = 0; i < 6; i++) pending.push(h.post("query", filters));
+    await vi.waitFor(() => expect(h.calls).toHaveLength(7), { timeout: 4000 });
+    expect(h.calls[1].at - h.calls[0].at).toBeLessThan(200);
+    const seventh = await h.post("publish", h.event);
+    expect(seventh.status).toBe(429);
+    expect(await seventh.json()).toEqual({
+      error: "Query concurrency limit",
+      sent: false,
+    });
+    controller.abort();
+    await stopped;
+    await delay(30);
+    // A different community still shares the process-wide additive snapshot cap.
+    const second = await fetch(`${h.base}/api/relay/secondary/query`, {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify(presenceSnapshot),
+    });
+    expect(second.status).toBe(429);
+    expect(h.calls).toHaveLength(7);
+    release();
+    await Promise.all(pending.map(async (response) => (await response).text()));
+    const recovered = await h.post("query", filters);
+    expect(recovered.status).toBe(200);
+    await recovered.text();
+  } finally {
+    release();
+    controller.abort();
+    await h.close();
+  }
+}, 10000);
+
+test("broker optional classification ignores spoofed priority and rejects mixed/extra-key capacity bypass", async () => {
+  let release;
+  const held = new Promise((resolve) => {
+    release = resolve;
+  });
+  const h = await harness(async () => {
+    await held;
+    return Response.json([]);
+  });
+  const pending = [];
+  try {
+    for (let i = 0; i < 6; i++) pending.push(h.post("query", filters));
+    await vi.waitFor(() => expect(h.calls).toHaveLength(1));
+    await delay(30);
+    for (const value of [
+      filters,
+      [{ ...presenceSnapshot[0], kinds: [20001, 9] }],
+      [{ ...presenceSnapshot[0], kinds: [20001, 20001] }],
+      [{ ...presenceSnapshot[0], since: 0 }],
+      [
+        {
+          ...presenceSnapshot[0],
+          authors: ["a".repeat(64), "a".repeat(64)],
+          limit: 2,
+        },
+      ],
+    ]) {
+      const response = await h.post("query", value, undefined, "presence");
+      expect(response.status).toBe(429);
+      expect(await response.json()).toEqual({
+        error: "Query concurrency limit",
+        sent: false,
+      });
+    }
+    pending.push(h.post("query", presenceSnapshot, undefined, "foreground"));
+    await vi.waitFor(() =>
+      expect(h.calls.some((c) => c.body[0].kinds[0] === 20001)).toBe(true),
+    );
+    release();
+    await Promise.all(pending.map(async (response) => (await response).text()));
+  } finally {
+    release();
+    await h.close();
+  }
+}, 10000);
+
+test("broker presence quota evidence shares API cooldown with ordinary publication, not local capacity", async () => {
+  const h = await harness(() =>
+    Response.json(
+      { error: "rate-limited: quota exceeded; retry in 3s" },
+      { status: 429 },
+    ),
+  );
+  try {
+    const p = await h.post("query", presenceSnapshot);
+    expect(p.status).toBe(429);
+    expect(await p.json()).toMatchObject({ quota: "api", retryAfterMs: 4000 });
+    const f = await h.post("publish", h.event);
+    expect(f.status).toBe(429);
+    expect(await f.json()).toMatchObject({ paused: true, sent: false });
+    expect(h.calls).toHaveLength(1);
+  } finally {
+    await h.close();
+  }
+});