One persisted job list drives the alarm one round at a time, as the single pull did. A pull job syncs its sources in turn with an optional filter (authors, kinds, since); backfill is a pull with the owner as author from the relays in their kind 10002. A push job rebroadcasts what this relay holds to other relays, walking the store by a cursor that survives sleeps so a standing push forwards only what arrived since; NIP-70 protected events are skipped, and private kinds never leave a members-only relay. Standing jobs repeat every 1, 6 or 24 hours, five at most. pullfrom and pullstatus still answer in their old shape. The action is now jobs.
- commit
- 6a062f9e0a4e8941bdcf335058fa322990bf0514
- parent
- 1727783aecf7a113c36cfa37d08f88a9ea2862c5
- author
- Dami <dami.etoile@gmail.com>
| A | src/ | 255 | +++++++++++++++++++++++++++++ |
| M | src/ | 25 | +++- |
| M | src/ | 40 | +++- |
| M | src/ | 126 | +++++++++++--- |
| M | src/ | 4 | +- |
| M | src/ | 7 | + |
| A | test/ | 262 | ++++++++++++++++++++++++++++++ |
@@ -0,0 +1,255 @@
1+// Jobs: work the relay does on its own between requests, driven by the2+// alarm one round at a time so it survives the object sleeping. Two kinds:3+// pull (fetch what another relay has and this one lacks, see pull.ts) and4+// push (rebroadcast what this relay holds to other relays). A job runs5+// once or on a standing interval. Awake time is metered, so a job costs6+// fuel, which is the right signal for work nobody asked a client for.7+import { hasTag, isPrivate, now, type Event } from "./event.ts";8+import type { Filter } from "./filter.ts";9+import { Socket, dial, checkPullURL, runPullRound, type PullFilter, type PullJob, type PullResult } from "./pull.ts";10+import type { Relay } from "./relay.ts";11+12+export type JobKind = "pull" | "push";13+export type JobLabel = "pull" | "backfill" | "push";14+export const EVERY = [0, 1, 6, 24] as const; // hours; 0 runs once15+16+export interface JobResult {17+ finishedAt: number;18+ error: string;19+ rounds: number;20+ stored: number;21+ skipped: number;22+ blobs: number;23+ sent: number;24+ refused: number;25+}26+27+export interface Job {28+ id: string;29+ kind: JobKind;30+ label: JobLabel;31+ relays: string[]; // sources for a pull, targets for a push32+ filter: PullFilter;33+ every: number; // hours; 0 once34+ createdAt: number;35+ nextRun: number; // unix seconds; 0 when a once job has finished36+ running: boolean;37+ startedAt: number;38+ // Progress of the current run.39+ rounds: number;40+ failures: number;41+ relayIndex: number; // pull: which source is being synced42+ cursor: number; // push: last sequence number forwarded, kept across runs43+ stored: number;44+ skipped: number;45+ blobs: number;46+ sent: number;47+ refused: number;48+ last: JobResult | null;49+}50+51+export const MAX_STANDING = 5;52+export const MAX_JOBS = 20;53+const KEEP_FINISHED = 10;54+const PUSH_BATCH = 200;55+const PUSH_PER_TARGET = 50;56+const PUSH_TIMEOUT_MS = 15_000;57+const REFUSALS_PER_TARGET = 5;58+59+const hex64 = (v: unknown): v is string => typeof v === "string" && /^[0-9a-f]{64}$/.test(v);60+61+export type JobSpec = Omit<Job, "id" | "createdAt" | "nextRun">;62+63+// checkJob validates a job request. Returns the job's fields or a reason.64+export function checkJob(raw: unknown, relay: Relay): JobSpec | string {65+ const r = (raw && typeof raw === "object" ? raw : {}) as Record<string, unknown>;66+ const kind = r.kind === "pull" || r.kind === "push" ? r.kind : null;67+ if (!kind) return "invalid: kind must be pull or push";68+ const label: JobLabel = r.label === "backfill" && kind === "pull" ? "backfill" : kind;69+ const list = Array.isArray(r.relays) ? r.relays : [];70+ const relays: string[] = [];71+ for (const u of list) {72+ if (typeof u !== "string") return "invalid: relays must be URLs";73+ const url = u.trim();74+ const bad = checkPullURL(url, relay.slug, relay.domain);75+ if (bad) return kind === "push" ? bad.replace("pull from itself", "push to itself") : bad;76+ if (!relays.includes(url)) relays.push(url);77+ }78+ if (relays.length === 0) return "invalid: at least one relay is needed";79+ if (relays.length > 10) return "invalid: at most ten relays per job";80+ const f = (r.filter && typeof r.filter === "object" ? r.filter : {}) as Record<string, unknown>;81+ const filter: PullFilter = {};82+ if (Array.isArray(f.authors) && f.authors.length) {83+ if (!f.authors.every(hex64) || f.authors.length > 50) return "invalid: authors must be up to fifty hex pubkeys";84+ filter.authors = [...new Set(f.authors as string[])];85+ }86+ if (Array.isArray(f.kinds) && f.kinds.length) {87+ if (!f.kinds.every((k) => Number.isInteger(k) && (k as number) >= 0 && (k as number) <= 65535) || f.kinds.length > 50) return "invalid: kinds must be up to fifty integers";88+ filter.kinds = [...new Set(f.kinds as number[])];89+ }90+ if (f.since !== undefined && f.since !== null && f.since !== 0) {91+ if (!Number.isInteger(f.since) || (f.since as number) < 0) return "invalid: since must be a unix time";92+ filter.since = f.since as number;93+ }94+ const every = r.every === undefined ? 0 : Number(r.every);95+ if (!(EVERY as readonly number[]).includes(every)) return "invalid: every must be 0, 1, 6 or 24 hours";96+ if (kind === "push" && relay.settings.policy.reads === "members" && filter.kinds?.some(isPrivate)) return "restricted: a members-only relay does not rebroadcast private kinds";97+ return { kind, label, relays, filter, every, running: false, startedAt: 0, rounds: 0, failures: 0, relayIndex: 0, cursor: 0, stored: 0, skipped: 0, blobs: 0, sent: 0, refused: 0, last: null };98+}99+100+// relaysFromList reads the owner's kind 10002 stored on this relay and101+// returns its relays, both markers, without this relay itself.102+export function relaysFromList(relay: Relay, pubkey: string): string[] {103+ const rows = relay.store.query({ kinds: [10002], authors: [pubkey], tags: {} }, { pubkeys: [], all: true }, 1, now()).rows;104+ if (rows.length === 0) return [];105+ const e = JSON.parse(rows[0]) as Event;106+ const out: string[] = [];107+ for (const t of e.tags) {108+ if (t[0] !== "r" || typeof t[1] !== "string") continue;109+ const url = t[1].trim().replace(/\/+$/, "");110+ if (checkPullURL(url, relay.slug, relay.domain)) continue;111+ if (!out.includes(url)) out.push(url);112+ }113+ return out;114+}115+116+// runRound does one round of a job. more: call again for this run.117+export async function runRound(relay: Relay, job: Job): Promise<{ more: boolean; error: string }> {118+ if (job.kind === "pull") return runPullSourceRound(relay, job);119+ return runPushRound(relay, job);120+}121+122+// A pull job syncs its sources one after another; each source is a pull123+// in the sense of pull.ts, with the job's filter.124+async function runPullSourceRound(relay: Relay, job: Job): Promise<{ more: boolean; error: string }> {125+ if (job.relayIndex >= job.relays.length) return { more: false, error: "" };126+ const sub: PullJob = { url: job.relays[job.relayIndex], startedAt: job.startedAt, rounds: 0, stored: 0, skipped: 0, blobs: 0, failures: 0 };127+ if (job.filter.authors || job.filter.kinds || job.filter.since) sub.filter = job.filter;128+ const r = await runPullRound(relay, sub);129+ job.rounds++;130+ job.stored += sub.stored;131+ job.skipped += sub.skipped;132+ job.blobs += sub.blobs;133+ if (r.error) return r;134+ if (!r.more) job.relayIndex++;135+ return { more: job.relayIndex < job.relays.length, error: "" };136+}137+138+// A push round takes the next batch past the cursor and offers it to every139+// target. Events nobody else may publish (NIP-70 protected) are skipped, as140+// are private kinds on a members-only relay. Refusals are counted; a target141+// that refuses several in a row is left alone for the round.142+async function runPushRound(relay: Relay, job: Job): Promise<{ more: boolean; error: string }> {143+ job.rounds++;144+ const f: Filter = { tags: {} };145+ if (job.filter.authors) f.authors = job.filter.authors;146+ if (job.filter.kinds) f.kinds = job.filter.kinds;147+ if (job.filter.since) f.since = job.filter.since;148+ const rows = relay.store.after(job.cursor, f, PUSH_BATCH, now());149+ if (rows.length === 0) return { more: false, error: "" };150+ const membersOnly = relay.settings.policy.reads === "members";151+ const events: Event[] = [];152+ for (const r of rows) {153+ const e = JSON.parse(r.raw) as Event;154+ if (hasTag(e, "-") || (membersOnly && isPrivate(e.kind))) {155+ job.skipped++;156+ continue;157+ }158+ events.push(e);159+ }160+ let failed = "";161+ let reached = 0;162+ for (const url of job.relays) {163+ try {164+ await pushTo(relay, url, events, job);165+ reached++;166+ } catch (err) {167+ failed = url + ": " + (err instanceof Error ? err.message : String(err));168+ }169+ }170+ if (reached === 0 && events.length) return { more: false, error: failed };171+ job.cursor = rows[rows.length - 1].seq;172+ return { more: rows.length === PUSH_BATCH, error: "" };173+}174+175+async function pushTo(relay: Relay, url: string, events: Event[], job: Job) {176+ if (events.length === 0) return;177+ const sock = new Socket(await dial(relay, url));178+ try {179+ let refusedInARow = 0;180+ for (let i = 0; i < events.length; i += PUSH_PER_TARGET) {181+ const batch = events.slice(i, i + PUSH_PER_TARGET);182+ const pending = new Set(batch.map((e) => e.id));183+ for (const e of batch) sock.send("EVENT", e);184+ const deadline = Date.now() + PUSH_TIMEOUT_MS;185+ while (pending.size && Date.now() < deadline) {186+ const m = await sock.recv();187+ if (m[0] !== "OK" || typeof m[1] !== "string" || !pending.has(m[1])) continue;188+ pending.delete(m[1]);189+ const msg = String(m[3] ?? "");190+ if (m[2] === true || msg.startsWith("duplicate:")) {191+ job.sent++;192+ refusedInARow = 0;193+ } else {194+ job.refused++;195+ refusedInARow++;196+ }197+ }198+ if (pending.size) {199+ job.refused += pending.size;200+ throw new Error("the relay stopped answering");201+ }202+ if (refusedInARow >= REFUSALS_PER_TARGET) return;203+ }204+ } finally {205+ sock.close();206+ }207+}208+209+// startRun resets the counters of a run. The push cursor is kept so a210+// standing push forwards only what arrived since.211+export function startRun(job: Job, t: number) {212+ job.running = true;213+ job.startedAt = t;214+ job.rounds = 0;215+ job.failures = 0;216+ job.relayIndex = 0;217+ job.stored = 0;218+ job.skipped = 0;219+ job.blobs = 0;220+ job.sent = 0;221+ job.refused = 0;222+}223+224+// finishRun closes the current run and schedules the next one.225+export function finishRun(job: Job, error: string, t: number) {226+ job.last = { finishedAt: t, error, rounds: job.rounds, stored: job.stored, skipped: job.skipped, blobs: job.blobs, sent: job.sent, refused: job.refused };227+ job.running = false;228+ job.nextRun = job.every > 0 ? t + job.every * 3600 : 0;229+}230+231+// pruneFinished drops the oldest finished once jobs beyond a few, so the232+// list stays a list.233+export function pruneFinished(jobs: Job[]): Job[] {234+ const done = jobs.filter((j) => !j.running && j.every === 0 && j.nextRun === 0).sort((a, b) => (b.last?.finishedAt ?? 0) - (a.last?.finishedAt ?? 0));235+ const drop = new Set(done.slice(KEEP_FINISHED).map((j) => j.id));236+ return jobs.filter((j) => !drop.has(j.id));237+}238+239+// pullView presents pull jobs in the shape the first console and tests240+// used: the running pull, if any, and the last finished one.241+export function pullView(jobs: Job[]): { running: PullJob | null; last: PullResult | null } {242+ const pulls = jobs.filter((j) => j.kind === "pull");243+ const run = pulls.find((j) => j.running || (j.nextRun > 0 && j.nextRun <= now() && j.every === 0));244+ const running: PullJob | null = run ? { url: run.relays[run.relayIndex] ?? run.relays[0], startedAt: run.startedAt, rounds: run.rounds, stored: run.stored, skipped: run.skipped, blobs: run.blobs, failures: run.failures } : null;245+ // Latest finished wins; on the same second the job added later wins.246+ let done: Job | undefined;247+ for (const j of pulls) if (j.last && !j.running && (!done || (j.last.finishedAt >= (done.last as JobResult).finishedAt))) done = j;248+ const last: PullResult | null = done && done.last ? { url: done.relays[0], startedAt: done.startedAt, rounds: done.last.rounds, stored: done.last.stored, skipped: done.last.skipped, blobs: done.last.blobs, failures: 0, finishedAt: done.last.finishedAt, error: done.last.error } : null;249+ return { running, last };250+}251+252+export function newJobID(): string {253+ const b = crypto.getRandomValues(new Uint8Array(6));254+ return [...b].map((x) => x.toString(16).padStart(2, "0")).join("");255+}@@ -9,6 +9,7 @@ import { listInvites, mintInvite, revokeInvite } from "./invites.ts";
99 import { descriptor, type Blob } from "./blossom.ts";1010 import { isReplaceable, isProtected, publicFields } from "./settings.ts";1111 import { checkPullURL } from "./pull.ts";12+import { relaysFromList } from "./jobs.ts";1213 import { can, METHOD_ACTIONS } from "./roles.ts";1314 import { PRESETS, applyPreset } from "./presets.ts";1415 @@ -56,7 +57,7 @@ const METHODS = [
5657 "listblobs", "deleteblob",5758 "storagestats", "setretention", "listretention", "purgekind",5859 "deleterelay", "exportconfig", "importconfig", "resetrules", "listpresets", "applypreset",59- "pullfrom", "pullstatus", "transferowner",60+ "pullfrom", "pullstatus", "listjobs", "addjob", "removejob", "runjob", "backfill", "transferowner",6061 ];6162 6263 // validIP accepts an IPv4 or IPv6 address, nothing fancier: no ranges, no names.@@ -364,6 +365,28 @@ export async function manage(relay: Relay, req: Request): Promise<Response> {
364365 }365366 case "pullstatus":366367 return reply({ result: await relay.pullStatus() });368+ case "listjobs":369+ return reply({ result: await relay.jobs() });370+ case "addjob": {371+ // ({kind, relays, filter?, every?, label?}): a pull or push, once or standing.372+ const r = await relay.addJob(params[0]);373+ if (typeof r === "string") return reply({ error: r }, r.startsWith("invalid") ? 400 : 409);374+ return reply({ result: r });375+ }376+ case "removejob":377+ return reply({ result: await relay.removeJob(str(0)) });378+ case "runjob":379+ return reply({ result: await relay.runJob(str(0)) });380+ case "backfill": {381+ // (relays?): fetch the owner's own history from the relays in their382+ // kind 10002 stored here, or from the given list.383+ const given = Array.isArray(params[0]) ? (params[0] as unknown[]).filter((u): u is string => typeof u === "string" && u.trim() !== "") : [];384+ const relays = given.length ? given : relaysFromList(relay, p.owner);385+ if (relays.length === 0) return reply({ error: "invalid: no relay list (kind 10002) is stored here; give relays to fetch from" }, 400);386+ const r = await relay.addJob({ kind: "pull", label: "backfill", relays, filter: { authors: [p.owner] }, every: 0 });387+ if (typeof r === "string") return reply({ error: r }, r.startsWith("invalid") ? 400 : 409);388+ return reply({ result: r });389+ }367390 case "transferowner": {368391 // (pubkey): hands the relay to a member; the old owner stays as a moderator.369392 const pk = str(0);@@ -1,16 +1,25 @@
11 // Pull: this relay fetches what another relay has and it lacks. One round22 // is one connection: reconcile with NIP-77, take a bounded batch of the33 // missing events, and, for a relay on this host, a batch of its files.4-// The alarm calls rounds until nothing is missing, so a pull survives the5-// object sleeping, and running it again later only fetches what is new.4+// The alarm calls rounds until nothing is missing (see jobs.ts), so a pull5+// survives the object sleeping, and running it again only fetches what is6+// new. An optional filter narrows the reconciliation: authors for a7+// backfill of the owner's own history, kinds, since.68 import { sha256 } from "@noble/hashes/sha2.js";79 import { validate, now, type Event } from "./event.ts";10+import type { Filter } from "./filter.ts";811 import { Negentropy, bytesToHex, hexToBytes } from "./negentropy.ts";912 import { ERR_DUPLICATE } from "./store.ts";1013 import { validName } from "./names.ts";1114 import type { Blob } from "./blossom.ts";1215 import type { Relay } from "./relay.ts";1316 17+export interface PullFilter {18+ authors?: string[];19+ kinds?: number[];20+ since?: number;21+}22+1423 export interface PullJob {1524 url: string;1625 startedAt: number;@@ -19,6 +28,7 @@ export interface PullJob {
1928 skipped: number;2029 blobs: number;2130 failures: number;31+ filter?: PullFilter;2232 }2333 export interface PullResult extends PullJob {2434 finishedAt: number;@@ -56,7 +66,7 @@ export function checkPullURL(raw: string, self: string, domain: string): string
5666 5767 // dial opens a websocket to a relay. Our own relays are reached through5868 // their object directly, which also works in wrangler dev and tests.59-async function dial(relay: Relay, raw: string): Promise<WebSocket> {69+export async function dial(relay: Relay, raw: string): Promise<WebSocket> {6070 const u = new URL(raw);6171 const local = localName(u, relay.domain);6272 const headers: Record<string, string> = { upgrade: "websocket" };@@ -70,7 +80,7 @@ async function dial(relay: Relay, raw: string): Promise<WebSocket> {
7080 }7181 7282 // Socket turns a websocket into a queue of parsed messages.73-class Socket {83+export class Socket {7484 private queue: unknown[][] = [];7585 private waiters: { res: (m: unknown[]) => void; rej: (e: Error) => void }[] = [];7686 private closed: Error | null = null;@@ -133,11 +143,12 @@ export async function runPullRound(relay: Relay, job: PullJob): Promise<{ more:
133143 let sock: Socket | null = null;134144 try {135145 sock = new Socket(await dial(relay, job.url));136- const need = await reconcile(relay, sock);146+ const need = await reconcile(relay, sock, job.filter ?? {});137147 const want = need.slice(0, ROUND_EVENTS);138148 for (let i = 0; i < want.length; i += IDS_PER_REQ) await fetchEvents(relay, sock, want.slice(i, i + IDS_PER_REQ), job);139149 let more = need.length > want.length;140- if (!more) {150+ if (!more && !job.filter) {151+ // Files come along on a whole-relay pull; a filtered pull is about events.141152 const local = localName(new URL(job.url), relay.domain);142153 if (local) more = await copyBlobs(relay, local, job);143154 }@@ -149,14 +160,19 @@ export async function runPullRound(relay: Relay, job: PullJob): Promise<{ more:
149160 }150161 }151162 152-// reconcile runs NIP-77 as the initiator over everything and returns the163+// reconcile runs NIP-77 as the initiator over the filter and returns the153164 // ids the other side has that we do not.154-async function reconcile(relay: Relay, sock: Socket): Promise<string[]> {155- const items = relay.store.syncItems({ tags: {} }, { pubkeys: [], all: true }, MAX_ITEMS, now());165+async function reconcile(relay: Relay, sock: Socket, filter: PullFilter): Promise<string[]> {166+ const f: Filter = { tags: {} };167+ const wire: Record<string, unknown> = {};168+ if (filter.authors?.length) f.authors = wire.authors = filter.authors;169+ if (filter.kinds?.length) f.kinds = wire.kinds = filter.kinds;170+ if (filter.since) f.since = wire.since = filter.since;171+ const items = relay.store.syncItems(f, { pubkeys: [], all: true }, MAX_ITEMS, now());156172 if (items === "too big") throw new Error("this relay holds too many events to sync");157173 const neg = new Negentropy(items, sha256);158174 const id = "pull";159- sock.send("NEG-OPEN", id, {}, bytesToHex(neg.initiate()));175+ sock.send("NEG-OPEN", id, wire, bytesToHex(neg.initiate()));160176 const need: string[] = [];161177 for (;;) {162178 const m = await sock.recv();@@ -183,7 +199,9 @@ async function fetchEvents(relay: Relay, sock: Socket, ids: string[], job: PullJ
183199 if (m[0] === "CLOSED" && m[1] === sub) throw new Error("query refused: " + String(m[2]));184200 if (m[0] !== "EVENT" || m[1] !== sub) continue;185201 const e = m[2] as Event;186- if (validate(e) || !ids.includes(e.id) || !relay.settings.kindAllowed(e.kind)) {202+ const f = job.filter;203+ const outside = !!f && ((f.authors?.length && !f.authors.includes(e.pubkey)) || (f.kinds?.length && !f.kinds.includes(e.kind)) || (f.since && e.created_at < f.since));204+ if (validate(e) || !ids.includes(e.id) || outside || !relay.settings.kindAllowed(e.kind)) {187205 job.skipped++;188206 continue;189207 }@@ -21,7 +21,8 @@ import { claimFromProfile, nip05Document } from "./nip05.ts";
2121 import { checkInvite, claimInvite, invitePage, termsPage } from "./invites.ts";2222 import { verifyNIP98 } from "./manage.ts";2323 import { FAVICON_SVG } from "./ui.ts";24-import { runPullRound, type PullJob, type PullResult } from "./pull.ts";24+import type { PullJob, PullResult } from "./pull.ts";25+import { MAX_JOBS, MAX_STANDING, checkJob, finishRun, newJobID, pruneFinished, pullView, runRound, startRun, type Job, type JobSpec } from "./jobs.ts";2526 import { groupFacts, handleGroupEvent, isGroupManagement, isGroupState, isNIP43Request } from "./groups.ts";2627 import { KIND_GROUP_MEMBERS } from "./identity.ts";2728 import { SIGNER_JS } from "./gen/signer.ts";@@ -105,8 +106,8 @@ export class Relay extends DurableObject<Env> {
105106 private lnurl: LnurlParams | null = null;106107 // Outbound HTTP to the lightning provider; tests substitute a fake.107108 fetcher: Fetcher = (u, i) => fetch(u, i);108- // A pull round in flight, so overlapping alarms do not run two.109- private pulling = false;109+ // A job round in flight, so overlapping alarms do not run two.110+ private working = false;110111 111112 constructor(ctx: DurableObjectState, env: Env) {112113 super(ctx, env);@@ -359,50 +360,113 @@ export class Relay extends DurableObject<Env> {
359360 return "";360361 }361362 362- // ---- pulling from another relay ----363+ // ---- jobs: pulls, backfills, rebroadcasts, once or standing ----363364 364- // pullStart records the job and wakes the alarm, which runs it in rounds.365- async pullStart(url: string): Promise<string> {366- if (await this.ctx.storage.get<PullJob>("pull")) return "restricted: a pull is already running";365+ // jobs is the persisted list. A job left from before the list existed366+ // is folded in on first read.367+ async jobs(): Promise<Job[]> {368+ let list = (await this.ctx.storage.get<Job[]>("jobs")) ?? [];369+ const old = await this.ctx.storage.get<PullJob>("pull");370+ if (old) {371+ list = [...list, { id: newJobID(), kind: "pull", label: "pull", relays: [old.url], filter: {}, every: 0, createdAt: old.startedAt, nextRun: now(), running: false, startedAt: old.startedAt, rounds: old.rounds, failures: 0, relayIndex: 0, cursor: 0, stored: old.stored, skipped: old.skipped, blobs: old.blobs, sent: 0, refused: 0, last: null }];372+ await this.ctx.storage.delete("pull");373+ await this.ctx.storage.delete("lastPull");374+ await this.ctx.storage.put("jobs", list);375+ }376+ return list;377+ }378+379+ private async saveJobs(list: Job[]) {380+ await this.ctx.storage.put("jobs", list);381+ }382+383+ // addJob validates, records and wakes the alarm. Returns the job or a reason.384+ async addJob(raw: unknown): Promise<Job | string> {385+ const spec = checkJob(raw, this);386+ if (typeof spec === "string") return spec;387+ return this.addChecked(spec);388+ }389+390+ async addChecked(spec: JobSpec): Promise<Job | string> {367391 if (this.fuelStatus().outOfFuel) return "restricted: this relay is out of fuel";368- const job: PullJob = { url, startedAt: now(), rounds: 0, stored: 0, skipped: 0, blobs: 0, failures: 0 };369- await this.ctx.storage.put("pull", job);370- await this.ctx.storage.delete("lastPull");392+ let list = pruneFinished(await this.jobs());393+ if (spec.every > 0 && list.filter((j) => j.every > 0).length >= MAX_STANDING) return `restricted: at most ${MAX_STANDING} standing jobs`;394+ if (list.length >= MAX_JOBS) return `restricted: at most ${MAX_JOBS} jobs; remove one first`;395+ if (spec.kind === "pull" && list.some((j) => j.kind === "pull" && j.every === 0 && (j.running || j.nextRun > 0) && j.relays.join() === spec.relays.join())) return "restricted: a pull from there is already running";396+ const job: Job = { ...spec, id: newJobID(), createdAt: now(), nextRun: now() };397+ list = [...list, job];398+ await this.saveJobs(list);371399 await this.ctx.storage.setAlarm(Date.now() + 50);372- return "";400+ return job;401+ }402+403+ async removeJob(id: string): Promise<boolean> {404+ const list = await this.jobs();405+ const rest = list.filter((j) => j.id !== id);406+ if (rest.length === list.length) return false;407+ await this.saveJobs(rest);408+ return true;409+ }410+411+ // runJob makes a job due now.412+ async runJob(id: string): Promise<boolean> {413+ const list = await this.jobs();414+ const job = list.find((j) => j.id === id);415+ if (!job) return false;416+ if (!job.running) job.nextRun = now();417+ await this.saveJobs(list);418+ await this.ctx.storage.setAlarm(Date.now() + 50);419+ return true;420+ }421+422+ // pullStart and pullStatus are the first console's view: one pull at a time.423+ async pullStart(url: string): Promise<string> {424+ const r = await this.addChecked({ kind: "pull", label: "pull", relays: [url], filter: {}, every: 0, running: false, startedAt: 0, rounds: 0, failures: 0, relayIndex: 0, cursor: 0, stored: 0, skipped: 0, blobs: 0, sent: 0, refused: 0, last: null });425+ return typeof r === "string" ? r : "";373426 }374427 375428 async pullStatus(): Promise<{ running: PullJob | null; last: PullResult | null }> {376- return { running: (await this.ctx.storage.get<PullJob>("pull")) ?? null, last: (await this.ctx.storage.get<PullResult>("lastPull")) ?? null };429+ return pullView(await this.jobs());377430 }378431 379- // pullTick runs one round of the job. Returns whether another is due. A380- // few failed rounds in a row end the job with the reason.381- private async pullTick(): Promise<boolean> {382- if (this.pulling) return true;383- const job = await this.ctx.storage.get<PullJob>("pull");432+ // jobsTick runs one round of the job that is due. Returns whether another433+ // round is due soon. A few failed rounds in a row end a run with the434+ // reason; a standing job tries again at its next interval.435+ private async jobsTick(): Promise<boolean> {436+ if (this.working) return true;437+ const list = await this.jobs();438+ const t = now();439+ const job = list.find((j) => j.running) ?? list.find((j) => j.nextRun > 0 && j.nextRun <= t);384440 if (!job) return false;385- this.pulling = true;441+ if (!job.running) startRun(job, t);442+ this.working = true;386443 let r: { more: boolean; error: string };387444 try {388- r = await runPullRound(this, job);445+ r = await runRound(this, job);389446 } finally {390- this.pulling = false;447+ this.working = false;391448 }392- const finish = async (error: string) => {393- await this.ctx.storage.put("lastPull", { ...job, finishedAt: now(), error } satisfies PullResult);394- await this.ctx.storage.delete("pull");395- return false;396- };449+ // The job may have been removed while the round ran.450+ const fresh = await this.jobs();451+ const slot = fresh.findIndex((j) => j.id === job.id);452+ if (slot < 0) return fresh.some((j) => j.running || (j.nextRun > 0 && j.nextRun <= now()));397453 if (r.error) {398454 job.failures++;399- if (job.failures >= 3) return finish(r.error);455+ if (job.failures >= 3) finishRun(job, r.error, now());400456 } else {401457 job.failures = 0;402- if (!r.more) return finish("");458+ if (!r.more) finishRun(job, "", now());403459 }404- await this.ctx.storage.put("pull", job);405- return true;460+ fresh[slot] = job;461+ await this.saveJobs(fresh);462+ return fresh.some((j) => j.running || (j.nextRun > 0 && j.nextRun <= now()));463+ }464+465+ // nextJobRun is the earliest standing run, or 0.466+ private async nextJobRun(): Promise<number> {467+ let at = 0;468+ for (const j of await this.jobs()) if (j.nextRun > 0 && (at === 0 || j.nextRun < at)) at = j.nextRun;469+ return at;406470 }407471 408472 // ---- fuel plumbing ----@@ -938,7 +1002,7 @@ export class Relay extends DurableObject<Env> {
9381002 await this.teardown();9391003 return;9401004 }941- if (await this.pullTick()) {1005+ if (await this.jobsTick()) {9421006 await this.ctx.storage.setAlarm(Date.now() + 250);9431007 return;9441008 }@@ -951,6 +1015,8 @@ export class Relay extends DurableObject<Env> {
9511015 if (next > 0 && next < at) at = next;9521016 const lease = this.settings.policy.lease;9531017 if (this.settings.isLeased() && lease && lease.until < at) at = lease.until;1018+ const run = await this.nextJobRun();1019+ if (run > 0 && run < at) at = Math.max(run, t + 1);9541020 await this.ctx.storage.setAlarm(at * 1000 + 500);9551021 }9561022 @@ -15,7 +15,7 @@ export type Action =
1515 | "identity" // name, description, icon1616 | "storage"1717 | "config" // export and import18- | "pull"18+ | "jobs" // pulls, backfills, rebroadcasts1919 | "transfer"2020 | "deleteRelay";2121 @@ -48,7 +48,7 @@ export const METHOD_ACTIONS: Record<string, Action> = {
4848 changerelayname: "identity", changerelaydescription: "identity", changerelayicon: "identity",4949 storagestats: "storage", deleteblob: "storage",5050 exportconfig: "config", importconfig: "config",51- pullfrom: "pull", pullstatus: "pull",51+ pullfrom: "jobs", pullstatus: "jobs", listjobs: "jobs", addjob: "jobs", removejob: "jobs", runjob: "jobs", backfill: "jobs",5252 transferowner: "transfer",5353 deleterelay: "deleteRelay",5454 };@@ -272,6 +272,13 @@ export class Store {
272272 }273273 274274 // recent returns the newest events for the dashboard, regardless of visibility.275+ // after returns up to `limit` rows with a sequence number past `seq`, in276+ // insertion order, for jobs that walk the store with a cursor.277+ after(seq: number, f: Filter, limit: number, now: number): { seq: number; raw: string }[] {278+ const { conds, args } = this.where(f, { pubkeys: [], all: true }, now);279+ return this.x<{ seq: number; raw: string }>(`SELECT seq, raw FROM events WHERE seq>? AND ${conds.join(" AND ")} ORDER BY seq LIMIT ?`, seq, ...args, limit).toArray();280+ }281+275282 recent(limit: number, now: number): string[] {276283 return this.x<{ raw: string }>(`SELECT raw FROM events WHERE (expires = 0 OR expires > ?) ORDER BY created_at DESC, id ASC LIMIT ?`, now, limit).toArray().map((r) => r.raw);277284 }@@ -0,0 +1,262 @@
1+// Jobs: owner backfill from the relays in a kind 10002, rebroadcast to2+// other relays with a cursor, standing jobs on an interval, caps and3+// removal. Relays on this host are dialled object to object.4+import { SELF, env, runInDurableObject } from "cloudflare:test";5+import { describe, it, expect } from "vitest";6+import { finalizeEvent, generateSecretKey, getPublicKey, type Event } from "nostr-tools/pure";7+import { getToken } from "nostr-tools/nip98";8+import type { Relay } from "../src/relay.ts";9+import type { Job } from "../src/jobs.ts";10+11+const now = () => Math.floor(Date.now() / 1000);12+const ev = (sk: Uint8Array, kind: number, content: string, tags: string[][] = [], created_at = now()) => finalizeEvent({ kind, content, tags, created_at }, sk);13+const pk = (sk: Uint8Array) => getPublicKey(sk);14+15+async function rpc(host: string, sk: Uint8Array, method: string, ...params: unknown[]) {16+ const url = `http://${host}/`;17+ const payload = { method, params };18+ const token = await getToken(url, "POST", (e) => finalizeEvent(e, sk), true, payload);19+ const resp = await SELF.fetch(url, { method: "POST", headers: { "content-type": "application/nostr+json+rpc", authorization: token }, body: JSON.stringify(payload) });20+ return { status: resp.status, ...(await resp.json<any>()) };21+}22+23+class WS {24+ private queue: any[][] = [];25+ private waiters: ((m: any[]) => void)[] = [];26+ constructor(public ws: WebSocket) {27+ ws.accept();28+ ws.addEventListener("message", (e) => {29+ const m = JSON.parse(e.data as string);30+ const w = this.waiters.shift();31+ if (w) w(m);32+ else this.queue.push(m);33+ });34+ }35+ static async connect(host: string) {36+ const resp = await SELF.fetch(`http://${host}/`, { headers: { upgrade: "websocket" } });37+ const c = new WS(resp.webSocket!);38+ await c.expect("AUTH");39+ return c;40+ }41+ send(...m: unknown[]) {42+ this.ws.send(JSON.stringify(m));43+ }44+ recv(): Promise<any[]> {45+ const m = this.queue.shift();46+ if (m) return Promise.resolve(m);47+ return new Promise((res) => this.waiters.push(res));48+ }49+ async expect(type: string) {50+ const m = await this.recv();51+ expect(m[0], JSON.stringify(m)).toBe(type);52+ return m;53+ }54+ async ok(e: Event) {55+ this.send("EVENT", e);56+ const m = await this.expect("OK");57+ return { ok: m[2] as boolean, msg: m[3] as string };58+ }59+ private n = 0;60+ async req(filter: unknown) {61+ const id = "q" + ++this.n;62+ this.send("REQ", id, filter);63+ const events: Event[] = [];64+ for (;;) {65+ const m = await this.recv();66+ if (m[0] === "EVENT" && m[1] === id) events.push(m[2]);67+ else if (m[0] === "EOSE" && m[1] === id) {68+ this.send("CLOSE", id);69+ return events;70+ } else if (m[0] === "CLOSED" && m[1] === id) throw new Error(m[2]);71+ }72+ }73+}74+75+// relay claims a name and posts the given events to it.76+async function relay(host: string, owner: Uint8Array, events: Event[] = []) {77+ await rpc(host, owner, "claim");78+ const c = await WS.connect(host);79+ for (const e of events) expect((await c.ok(e)).ok, e.id).toBe(true);80+ return c;81+}82+83+const due = (j: Job) => j.running || (j.nextRun > 0 && j.nextRun <= now());84+85+// drive fires the alarm until no job is running or due, and returns the list.86+async function drive(host: string, owner: Uint8Array): Promise<Job[]> {87+ const stub = env.RELAY.getByName(host.split(".")[0]);88+ for (let i = 0; i < 80; i++) {89+ await runInDurableObject(stub, async (r: Relay) => r.alarm());90+ const jobs = (await rpc(host, owner, "listjobs")).result as Job[];91+ if (!jobs.some(due)) return jobs;92+ await new Promise((r) => setTimeout(r, 25));93+ }94+ throw new Error("jobs did not settle");95+}96+97+describe("owner backfill", () => {98+ it("fetches the owner's own history from the relays in their kind 10002, and nothing else", async () => {99+ const me = generateSecretKey();100+ const stranger = generateSecretKey();101+ const a = "hist-a.bind.ws";102+ const b = "hist-b.bind.ws";103+ const mine = [ev(me, 1, "on a", [], now() - 30), ev(me, 1, "also on a", [], now() - 20), ev(me, 30023, "an article", [["d", "x"]], now() - 10)];104+ await relay(a, generateSecretKey(), [mine[0], mine[1], ev(stranger, 1, "not mine")]);105+ await relay(b, generateSecretKey(), [mine[2], ev(stranger, 1, "also not mine")]);106+107+ const host = "hist.bind.ws";108+ const c = await relay(host, me);109+ expect((await rpc(host, me, "backfill")).status).toBe(400);110+ expect((await c.ok(ev(me, 10002, "", [["r", "wss://" + a, "read"], ["r", "wss://" + b, "write"], ["r", "wss://" + host]]))).ok).toBe(true);111+112+ const started = await rpc(host, me, "backfill");113+ expect(started.status).toBe(200);114+ expect(started.result.label).toBe("backfill");115+ expect(started.result.relays).toEqual(["wss://" + a, "wss://" + b]);116+ expect(started.result.filter).toEqual({ authors: [pk(me)] });117+ const jobs = await drive(host, me);118+ const job = jobs.find((j) => j.id === started.result.id)!;119+ expect(job.last?.error).toBe("");120+ expect(job.last?.stored).toBe(3);121+ expect(job.nextRun).toBe(0);122+ const got = await c.req({ kinds: [1, 30023] });123+ expect(got.map((e) => e.id).sort()).toEqual(mine.map((e) => e.id).sort());124+ expect(got.every((e) => e.pubkey === pk(me))).toBe(true);125+ // A member without the jobs action cannot start one; a bad list is refused.126+ expect((await rpc(host, stranger, "backfill")).status).toBe(403);127+ expect((await rpc(host, me, "backfill", ["https://" + a])).status).toBe(400);128+ });129+});130+131+describe("rebroadcast", () => {132+ it("forwards matching events to every target, counts duplicates as sent, and moves the cursor", async () => {133+ const owner = generateSecretKey();134+ const src = "blast.bind.ws";135+ const e1 = ev(owner, 1, "one", [], now() - 30);136+ const e2 = ev(owner, 1, "two", [], now() - 20);137+ const e7 = ev(owner, 7, "+", [["e", e1.id]], now() - 10);138+ const c = await relay(src, owner, [e1, e2, e7]);139+ const t1 = "target-a.bind.ws";140+ const t2 = "target-b.bind.ws";141+ const c1 = await relay(t1, generateSecretKey(), [e1]); // already there: a duplicate counts as sent142+ const c2 = await relay(t2, generateSecretKey());143+144+ const bad = await rpc(src, owner, "addjob", { kind: "push", relays: ["wss://" + src], filter: { kinds: [1] } });145+ expect(bad.error).toMatch(/itself/);146+ const started = await rpc(src, owner, "addjob", { kind: "push", relays: ["wss://" + t1, "wss://" + t2], filter: { kinds: [1] } });147+ expect(started.status, JSON.stringify(started)).toBe(200);148+ let jobs = await drive(src, owner);149+ let job = jobs.find((j) => j.id === started.result.id)!;150+ expect(job.last?.error).toBe("");151+ expect(job.last?.sent).toBe(4);152+ expect(job.last?.refused).toBe(0);153+ expect(job.cursor).toBeGreaterThan(0);154+ expect((await c1.req({ kinds: [1, 7] })).map((e) => e.id).sort()).toEqual([e1.id, e2.id].sort());155+ expect((await c2.req({ kinds: [1, 7] })).map((e) => e.id).sort()).toEqual([e1.id, e2.id].sort());156+157+ // Nothing on the source is stored by a push; a second run sends only what is new.158+ const e3 = ev(owner, 1, "three");159+ expect((await c.ok(e3)).ok).toBe(true);160+ expect((await rpc(src, owner, "runjob", started.result.id)).result).toBe(true);161+ jobs = await drive(src, owner);162+ job = jobs.find((j) => j.id === started.result.id)!;163+ expect(job.last?.sent).toBe(2);164+ expect((await c2.req({ ids: [e3.id] })).length).toBe(1);165+166+ // A target that refuses counts refusals and does not stop the others.167+ const closed = "target-c.bind.ws";168+ const cOwner = generateSecretKey();169+ await relay(closed, cOwner);170+ await rpc(closed, cOwner, "setpolicy", { writes: "owner" });171+ const again = await rpc(src, owner, "addjob", { kind: "push", relays: ["wss://" + closed], filter: { kinds: [1] } });172+ jobs = await drive(src, owner);173+ job = jobs.find((j) => j.id === again.result.id)!;174+ expect(job.last?.refused).toBe(3);175+ expect(job.last?.sent).toBe(0);176+ });177+178+ it("never rebroadcasts private kinds from a members-only relay", async () => {179+ const owner = generateSecretKey();180+ const src = "quiet.bind.ws";181+ const friend = generateSecretKey();182+ const dm = ev(owner, 4, "secret", [["p", pk(friend)]]);183+ const note = ev(owner, 1, "public");184+ await relay(src, owner);185+ await rpc(src, owner, "setpolicy", { reads: "members" });186+ const target = "loud.bind.ws";187+ const ct = await relay(target, generateSecretKey());188+ const stub = env.RELAY.getByName("quiet");189+ await runInDurableObject(stub, async (r: Relay) => {190+ expect(r.accept(dm, null).stored).toBe(true);191+ expect(r.accept(note, null).stored).toBe(true);192+ });193+ const refused = await rpc(src, owner, "addjob", { kind: "push", relays: ["wss://" + target], filter: { kinds: [4] } });194+ expect(refused.error).toMatch(/private kinds/);195+ const started = await rpc(src, owner, "addjob", { kind: "push", relays: ["wss://" + target] });196+ const jobs = await drive(src, owner);197+ const job = jobs.find((j) => j.id === started.result.id)!;198+ // The note and the relay's own signed records go; the message does not.199+ expect(job.last?.sent).toBeGreaterThanOrEqual(1);200+ expect(job.last?.skipped).toBeGreaterThanOrEqual(1);201+ expect((await ct.req({ kinds: [1, 4] })).map((e) => e.kind)).toEqual([1]);202+ });203+});204+205+describe("standing jobs", () => {206+ it("runs a recurring pull on its interval, caps standing jobs, and can be removed", async () => {207+ const owner = generateSecretKey();208+ const src = "upstream.bind.ws";209+ const e1 = ev(owner, 1, "first", [], now() - 10);210+ const cs = await relay(src, owner, [e1]);211+ const host = "mirror.bind.ws";212+ const c = await relay(host, owner);213+214+ expect((await rpc(host, owner, "addjob", { kind: "pull", relays: ["wss://" + src], every: 2 })).status).toBe(400);215+ const started = await rpc(host, owner, "addjob", { kind: "pull", relays: ["wss://" + src], every: 1 });216+ expect(started.status).toBe(200);217+ let jobs = await drive(host, owner);218+ let job = jobs.find((j) => j.id === started.result.id)!;219+ // The note plus the upstream relay's own signed records.220+ expect(job.last?.stored).toBeGreaterThanOrEqual(1);221+ expect(job.nextRun).toBeGreaterThan(now() + 3500);222+ expect(job.nextRun).toBeLessThanOrEqual(now() + 3600);223+ expect((await c.req({ ids: [e1.id] })).length).toBe(1);224+ const firstFinish = job.last!.finishedAt;225+ // The daily alarm is not scheduled past the next run.226+ const stub = env.RELAY.getByName("mirror");227+ await runInDurableObject(stub, async (_r: Relay, state) => {228+ const at = await state.storage.getAlarm();229+ expect(at).not.toBeNull();230+ expect(at!).toBeLessThanOrEqual(job.nextRun * 1000 + 500);231+ });232+233+ // Time passes: the interval elapses and the mirror picks up what is new.234+ const e2 = ev(owner, 1, "second");235+ expect((await cs.ok(e2)).ok).toBe(true);236+ await runInDurableObject(stub, async (_r: Relay, state) => {237+ const list = (await state.storage.get<Job[]>("jobs"))!;238+ list.find((j) => j.id === started.result.id)!.nextRun = now() - 1;239+ await state.storage.put("jobs", list);240+ });241+ await new Promise((r) => setTimeout(r, 1100));242+ jobs = await drive(host, owner);243+ job = jobs.find((j) => j.id === started.result.id)!;244+ expect(job.last!.finishedAt).toBeGreaterThan(firstFinish);245+ expect(job.last?.stored).toBeGreaterThanOrEqual(1);246+ expect(job.last?.stored).toBeLessThan(5);247+ expect((await c.req({ ids: [e2.id] })).length).toBe(1);248+249+ // At most five standing jobs; a once job still fits; removal frees the slot.250+ for (let i = 0; i < 4; i++) expect((await rpc(host, owner, "addjob", { kind: "pull", relays: [`wss://other-${i}.bind.ws`], every: 24 })).status).toBe(200);251+ const sixth = await rpc(host, owner, "addjob", { kind: "pull", relays: ["wss://other-9.bind.ws"], every: 24 });252+ expect(sixth.status).toBe(409);253+ expect(sixth.error).toMatch(/standing/);254+ expect((await rpc(host, owner, "removejob", started.result.id)).result).toBe(true);255+ expect((await rpc(host, owner, "removejob", started.result.id)).result).toBe(false);256+ expect(((await rpc(host, owner, "listjobs")).result as Job[]).some((j) => j.id === started.result.id)).toBe(false);257+ expect((await rpc(host, owner, "addjob", { kind: "pull", relays: ["wss://other-9.bind.ws"], every: 24 })).status).toBe(200);258+ // The old one-pull view still answers.259+ const st = (await rpc(host, owner, "pullstatus")).result;260+ expect(st.last === null || typeof st.last.stored === "number").toBe(true);261+ });262+});