- commit
- 991a9800ad5cd866d7741b15e9fbf6a5db233c61
- parent
- 6f51f921c4f972ec5449aa74fb1b2e4af8cc2d33
- author
- Dami <dami.etoile@gmail.com>
| M | docs/ | 2 | +- |
| M | docs/ | 2 | +- |
| M | src/ | 2 | +- |
| M | src/ | 53 | ++++++++++++- |
| A | test/ | 131 | ++++++++++++++++++++++++++++++ |
@@ -41,7 +41,7 @@ Custom hostnames (`domains.ts`) are registered with Cloudflare for SaaS from the
4141 - **Fuel** (`fuel.ts`): usage meters, credits, prices.4242 - **Identity** (`identity.ts`): the relay's own keypair and everything it signs.4343 44-The constructor runs schema setup inside `blockConcurrencyWhile`, then handles fetches. Websockets use the hibernation API. Per-connection state (NIP-42 challenge, authenticated pubkeys, open subscriptions, the client's address) is stored on the socket with `serializeAttachment`, so it survives hibernation. Negentropy sessions are in memory only and end with `closed:` when the object sleeps.44+The constructor runs schema setup inside `blockConcurrencyWhile`, then handles fetches. Websockets use the hibernation API. Per-connection state (NIP-42 challenge, authenticated pubkeys, open subscriptions, the filters of open syncs, the client's address) is stored on the socket with `serializeAttachment`, so it survives hibernation. A negentropy session's item list is in memory only; the next NEG-MSG after a wake reads it again from the store by the remembered filter, asking the read rule again first, so a sync carries on across a sleep. A socket whose attachment is over its limit keeps the session in memory alone, and that one ends with `closed:` when the object sleeps.4545 4646 The write path is one gate plus the client policy. The gate refuses malformed, expired, banned and out-of-context events (an `h` tag for another group) and, for client writes, the unclaimed and expired-lease states, fuel and NIP-70. The client policy adds kind rules, retention windows, the write rule, proof of work and per-member caps. Host-side writes, such as pulls and the relay's own records, pass the gate only. Kind 24133, NIP-46 signer traffic, passes everything but bans and rate limits on the way in: it is ephemeral, never stored and encrypted end to end, so the relay can carry a remote signer's session for the console, including for someone about to claim it. On the way out it is a private kind, delivered to a socket authenticated as its sender or its `p` recipient, or to a subscription that names one of those keys in `#p` or `authors`, under any read rule and without AUTH: the payload is encrypted end to end, and signers and their clients hold no key the relay knows. Amber probes a relay this way before it will list it, and Amethyst's bunker listens under a transport key it never authenticates with. A bare subscription to the kind sees nothing, so a members-only relay reveals signer traffic only to someone who already holds the key it is addressed to. A subscription to 24133 alone may be opened under any read rule. The console's remote-signer client answers the challenge with the session's own key.4747 @@ -53,7 +53,7 @@ Nothing else is required: no origin, no certificates, no servers. An idle relay
5353 ## Operate5454 5555 - `wrangler tail` streams logs, including receipt validation.56-- Deploys roll to objects within seconds. Live websockets survive when the object hibernates; a mid-flight negentropy sync does not.56+- Deploys roll to objects within seconds. Live websockets survive when the object hibernates, and so does a mid-flight negentropy sync: the next message rebuilds the session from the store.5757 - The `usage` table grows one row per relay per month. No cleanup is needed.5858 - To take a relay down as the operator, use the owner's typed-name `deleterelay` call or the console.5959 @@ -110,7 +110,7 @@ export function landing(req: Request, env: Env): Response {
110110 const bad = new URL(req.url).searchParams.get("bad");111111 const days = leaseDays(env);112112 const interfaces = table(["Endpoint", "Protocol", "Auth", "Notes"], [113- [`wss://<name>.${domain}`, "NIP-01, 09, 13, 17, 29, 40, 42, 43, 45, 46 transport, 50, 56, 59, 62, 67, 70, 77", "optional NIP-42", "Sockets hibernate. Negentropy sessions end when the object hibernates."],113+ [`wss://<name>.${domain}`, "NIP-01, 09, 13, 17, 29, 40, 42, 43, 45, 46 transport, 50, 56, 59, 62, 67, 70, 77", "optional NIP-42", "Sockets hibernate; a negentropy session carries on after a wake."],114114 ["GET / (accept nostr+json)", "NIP-11", "none", "limitation, retention, payments_url, self"],115115 ["POST /", "NIP-86", "NIP-98, owner", "claim and the management methods"],116116 ["POST /lease", "lease", "optional NIP-98", `A temporary relay for ${leaseDays(env)} days. A signature reserves the claim for that key.`],@@ -94,6 +94,9 @@ interface ConnState {
9494 ip: string; // the client's address as the worker saw it; "unknown" outside Cloudflare9595 authed: string[];9696 subs: Record<string, Filter[]>;97+ // Open NIP-77 syncs by id, each with its filter: enough to rebuild the98+ // session's item list after a wake. Absent on sockets from before.99+ syncs?: Record<string, Filter>;97100 }98101 99102 // Succession bookkeeping, kept in storage next to the policy.@@ -451,7 +454,13 @@ export class Relay extends DurableObject<Env> {
451454 delete s.subs[id];452455 changed = true;453456 }454- if (this.settings.mayRead(s.authed)) this.syncs.delete(ws);457+ if (this.settings.mayRead(s.authed)) {458+ this.syncs.delete(ws);459+ if (s.syncs && Object.keys(s.syncs).length) {460+ delete s.syncs;461+ changed = true;462+ }463+ }455464 if (changed) this.persist(ws, s);456465 }457466 }@@ -1650,8 +1659,27 @@ export class Relay extends DurableObject<Env> {
16501659 }16511660 let sess = sessions.get(id);16521661 let msgHex = "";1662+ // The responder's side of a sync is the sorted item list and nothing1663+ // else, and the list comes from the filter. So the filter rides on the1664+ // socket's attachment with the subscriptions, and a session whose1665+ // Negentropy object went with the object's memory is rebuilt from it1666+ // when the next message arrives, rather than ended with closed:.1667+ const remember = (f: Filter | null) => {1668+ const next = { ...(s.syncs ?? {}) };1669+ if (f) next[id] = f;1670+ else delete next[id];1671+ if (Object.keys(next).length) s.syncs = next;1672+ else delete s.syncs;1673+ // Past the attachment limit the filter stays in memory only, and the1674+ // session ends when the object sleeps, as every session once did.1675+ if (!this.persist(ws, s) && f) {1676+ delete s.syncs![id];1677+ if (Object.keys(s.syncs!).length === 0) delete s.syncs;1678+ }1679+ };16531680 if (verb === "NEG-CLOSE") {16541681 sessions.delete(id);1682+ if (s.syncs?.[id]) remember(null);16551683 return;16561684 }16571685 if (verb === "NEG-OPEN") {@@ -1660,13 +1688,33 @@ export class Relay extends DurableObject<Env> {
16601688 if (typeof f === "string") return this.send(ws, "NEG-ERR", id, "invalid: bad filter: " + f);16611689 const { reason } = this.allowFilters(s, [f]);16621690 if (reason) return this.send(ws, "NEG-ERR", id, reason);1663- if (!sessions.has(id) && sessions.size >= this.settings.policy.maxSubs) return this.send(ws, "NEG-ERR", id, "blocked: too many open syncs");1691+ const open = new Set([...sessions.keys(), ...Object.keys(s.syncs ?? {})]);1692+ if (!open.has(id) && open.size >= this.settings.policy.maxSubs) return this.send(ws, "NEG-ERR", id, "blocked: too many open syncs");16641693 const items = this.store.syncItems(f, { pubkeys: s.authed }, MAX_SYNC, now());16651694 if (items === "too big") return this.send(ws, "NEG-ERR", id, ERR_TOO_BIG, MAX_SYNC);16661695 sess = new Negentropy(items, sha256);16671696 sessions.set(id, sess);1697+ remember(f);16681698 msgHex = typeof rest[1] === "string" ? rest[1] : "";16691699 } else {1700+ const f = s.syncs?.[id];1701+ if (!sess && f) {1702+ // A wake since NEG-OPEN. The read rule is asked again with what the1703+ // socket has proved by now, and the items are read afresh, so the1704+ // session carries on from the store as it is.1705+ const { reason } = this.allowFilters(s, [f]);1706+ if (reason) {1707+ remember(null);1708+ return this.send(ws, "NEG-ERR", id, reason);1709+ }1710+ const items = this.store.syncItems(f, { pubkeys: s.authed }, MAX_SYNC, now());1711+ if (items === "too big") {1712+ remember(null);1713+ return this.send(ws, "NEG-ERR", id, ERR_TOO_BIG, MAX_SYNC);1714+ }1715+ sess = new Negentropy(items, sha256);1716+ sessions.set(id, sess);1717+ }16701718 if (!sess) return this.send(ws, "NEG-ERR", id, "closed: no such sync, send NEG-OPEN first");16711719 msgHex = typeof rest[0] === "string" ? rest[0] : "";16721720 }@@ -1676,6 +1724,7 @@ export class Relay extends DurableObject<Env> {
16761724 this.send(ws, "NEG-MSG", id, bytesToHex(reply ?? new Uint8Array()));16771725 } catch (err) {16781726 sessions.delete(id);1727+ if (s.syncs?.[id]) remember(null);16791728 this.send(ws, "NEG-ERR", id, "invalid: " + (err instanceof Error ? err.message : String(err)));16801729 }16811730 }@@ -0,0 +1,131 @@
1+// NIP-77 across a wake: the filter of an open sync rides on the socket's2+// attachment, so a session whose Negentropy object left with the object's3+// memory carries on from the store instead of ending with closed:.4+import { SELF, env, runInDurableObject } from "cloudflare:test";5+import { describe, it, expect } from "vitest";6+import { finalizeEvent, generateSecretKey, type Event } from "nostr-tools/pure";7+import { getToken } from "nostr-tools/nip98";8+import { sha256 } from "@noble/hashes/sha2.js";9+import type { Relay } from "../src/relay.ts";10+import { Negentropy, bytesToHex, hexToBytes, type SyncItem } from "../src/negentropy.ts";11+12+const now = () => Math.floor(Date.now() / 1000);13+const ev = (sk: Uint8Array, content: string, created_at: number) => finalizeEvent({ kind: 1, content, tags: [], created_at }, sk);14+const item = (e: Event): SyncItem => ({ timestamp: e.created_at, id: hexToBytes(e.id) });15+16+async function rpc(host: string, sk: Uint8Array, method: string, ...params: unknown[]) {17+ const url = `http://${host}/`;18+ const payload = { method, params };19+ const token = await getToken(url, "POST", (e) => finalizeEvent(e, sk), true, payload);20+ const resp = await SELF.fetch(url, { method: "POST", headers: { "content-type": "application/nostr+json+rpc", authorization: token }, body: JSON.stringify(payload) });21+ return { status: resp.status, ...(await resp.json<any>()) };22+}23+24+class WS {25+ private queue: any[][] = [];26+ private waiters: ((m: any[]) => void)[] = [];27+ constructor(public ws: WebSocket) {28+ ws.accept();29+ ws.addEventListener("message", (e) => {30+ const m = JSON.parse(e.data as string);31+ const w = this.waiters.shift();32+ if (w) w(m);33+ else this.queue.push(m);34+ });35+ }36+ static async connect(host: string): Promise<WS> {37+ const resp = await SELF.fetch(`http://${host}/`, { headers: { upgrade: "websocket" } });38+ const c = new WS(resp.webSocket!);39+ await c.expect("AUTH");40+ return c;41+ }42+ send(...m: unknown[]) {43+ this.ws.send(JSON.stringify(m));44+ }45+ recv(): Promise<any[]> {46+ const m = this.queue.shift();47+ if (m) return Promise.resolve(m);48+ return new Promise((res) => this.waiters.push(res));49+ }50+ async expect(type: string) {51+ const m = await this.recv();52+ expect(m[0], JSON.stringify(m)).toBe(type);53+ return m;54+ }55+ async ok(e: Event) {56+ this.send("EVENT", e);57+ const m = await this.expect("OK");58+ return { ok: m[2] as boolean, msg: m[3] as string };59+ }60+}61+62+// wake empties the relay's in-memory sync sessions, which is what a63+// hibernation does to them; the sockets and their attachments stay.64+const wake = (name: string) => runInDurableObject(env.RELAY.getByName(name), async (r: Relay) => (r as unknown as { syncs: Map<unknown, unknown> }).syncs.clear());65+// persisted reads the sync ids on the first socket's attachment.66+const persisted = (name: string) => runInDurableObject(env.RELAY.getByName(name), async (r: Relay) => Object.keys(((r as unknown as { ctx: DurableObjectState }).ctx.getWebSockets()[0].deserializeAttachment() as { syncs?: Record<string, unknown> }).syncs ?? {}));67+68+// seed claims a relay and stores three notes; the client holds two of them69+// and one the relay never saw.70+async function seed(name: string) {71+ const host = name + ".bind.ws";72+ const owner = generateSecretKey();73+ const author = generateSecretKey();74+ await rpc(host, owner, "claim");75+ const c = await WS.connect(host);76+ const t = now() - 100;77+ const notes = [ev(author, "one", t), ev(author, "two", t + 1), ev(author, "three", t + 2)];78+ for (const n of notes) expect((await c.ok(n)).ok).toBe(true);79+ const stray = ev(author, "stray", t + 3);80+ const mine = [notes[0], notes[1], stray];81+ return { host, owner, c, notes, stray, mine };82+}83+84+describe("NIP-77 across a wake", () => {85+ it("carries a sync on from the store after the session left memory", async () => {86+ const { host, c, notes, stray, mine } = await seed("syncwake");87+ const neg = new Negentropy(mine.map(item), sha256);88+ c.send("NEG-OPEN", "s1", { kinds: [1] }, bytesToHex(neg.initiate()));89+ const first = await c.expect("NEG-MSG");90+ expect(await persisted("syncwake")).toEqual(["s1"]);91+92+ await wake("syncwake");93+ // The same opening message again, now against a rebuilt session, draws94+ // the same answer, and reconciling it names exactly what each side lacks.95+ const again = new Negentropy(mine.map(item), sha256);96+ c.send("NEG-MSG", "s1", bytesToHex(again.initiate()));97+ const second = await c.expect("NEG-MSG");98+ expect(second[2]).toBe(first[2]);99+ const r = again.reconcile(hexToBytes(second[2]));100+ expect(r.reply).toBeNull();101+ expect(r.need).toEqual([notes[2].id]);102+ expect(r.have).toEqual([stray.id]);103+104+ // NEG-CLOSE takes the filter off the socket; after that a wake ends it.105+ c.send("NEG-CLOSE", "s1");106+ await new Promise((res) => setTimeout(res, 50));107+ expect(await persisted("syncwake")).toEqual([]);108+ await wake("syncwake");109+ c.send("NEG-MSG", "s1", bytesToHex(again.initiate()));110+ const err = await c.expect("NEG-ERR");111+ expect(err[2]).toMatch(/^closed: no such sync/);112+ });113+114+ it("counts remembered syncs against the cap and drops them with the read rule", async () => {115+ const { host, owner, c, mine } = await seed("synccap");116+ await rpc(host, owner, "setpolicy", { maxSubs: 1 });117+ const neg = new Negentropy(mine.map(item), sha256);118+ c.send("NEG-OPEN", "s1", { kinds: [1] }, bytesToHex(neg.initiate()));119+ await c.expect("NEG-MSG");120+ await wake("synccap");121+ c.send("NEG-OPEN", "s2", { kinds: [1] }, bytesToHex(neg.initiate()));122+ expect((await c.expect("NEG-ERR"))[2]).toBe("blocked: too many open syncs");123+124+ // Reads tightened to members: the socket has proved nothing, so the125+ // remembered sync goes with the subscriptions it never had.126+ await rpc(host, owner, "setpolicy", { reads: "members" });127+ expect(await persisted("synccap")).toEqual([]);128+ c.send("NEG-MSG", "s1", bytesToHex(neg.initiate()));129+ expect((await c.expect("NEG-ERR"))[2]).toMatch(/^closed: no such sync/);130+ });131+});