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.7import { hasTag, isPrivate, now, type Event } from "./event.ts";8import type { Filter } from "./filter.ts";9import { Socket, dial, checkPullURL, runPullRound, newPullProgress, type PullConnect, type PullSource, type PullFilter, type PullJob, type PullResult } from "./pull.ts";10import type { Relay } from "./relay.ts";11import { runMirrorRound } from "./site-mirror.ts";12import { runImportRound } from "./imports.ts";1314export type JobKind = "pull" | "push" | "import" | "mirror";15export type JobLabel = "pull" | "backfill" | "push" | "replica" | "import" | "mirror";16export const EVERY = [0, 1, 6, 24] as const; // hours; 0 runs once1718export interface JobResult {19 finishedAt: number;20 error: string;21 rounds: number;22 stored: number;23 skipped: number;24 blobs: number;25 sent: number;26 refused: number;27 duplicates?: number;28 sources?: PullSource[];29}3031export interface Job {32 id: string;33 kind: JobKind;34 label: JobLabel;35 relays: string[]; // sources for a pull, targets for a push36 filter: PullFilter;37 every: number; // hours; 0 once38 createdAt: number;39 nextRun: number; // unix seconds; 0 when a once job has finished40 running: boolean;41 startedAt: number;42 // Progress of the current run.43 rounds: number;44 failures: number;45 relayIndex: number; // pull: which source is being synced46 cursor: number; // push: last sequence number forwarded, kept across runs47 // Newer push jobs keep independent cursors. `cursor` remains for old job48 // readers and is the greatest cursor reached by any target.49 targetCursors?: Record<string, number>;50 targetStatus?: Record<string, { status: "pending" | "accepted" | "rejected"; error: string; at: number }>;51 targetAttempts?: Record<string, number>;52 stored: number;53 skipped: number;54 blobs: number;55 sent: number;56 refused: number;57 duplicates?: number; // import: events the relay already had58 site?: string; // mirror: the manifest event id59 object?: string; // import: the R2 object's id60 size?: number; // import: the object's size in bytes61 carry?: string; // import: the partial last line of the previous round, base6462 last: JobResult | null;63 pullSources?: PullSource[];64}6566export const MAX_STANDING = 5;67export const MAX_JOBS = 20;68const KEEP_FINISHED = 10;69const PUSH_BATCH = 200;70const PUSH_PER_TARGET = 50;71const PUSH_TIMEOUT_MS = 15_000;72const REFUSALS_PER_TARGET = 5;7374const hex64 = (v: unknown): v is string => typeof v === "string" && /^[0-9a-f]{64}$/.test(v);7576export type JobSpec = Omit<Job, "id" | "createdAt" | "nextRun">;7778// checkJob validates a job request. Returns the job's fields or a reason.79export function checkJob(raw: unknown, relay: Relay): JobSpec | string {80 const r = (raw && typeof raw === "object" ? raw : {}) as Record<string, unknown>;81 const kind = r.kind === "pull" || r.kind === "push" ? r.kind : null;82 if (!kind) return "invalid: kind must be pull or push";83 const label: JobLabel = kind === "pull" && (r.label === "backfill" || r.label === "replica") ? r.label : kind;84 const list = Array.isArray(r.relays) ? r.relays : [];85 const relays: string[] = [];86 for (const u of list) {87 if (typeof u !== "string") return "invalid: relays must be URLs";88 const url = u.trim();89 const bad = checkPullURL(url, relay.slug, relay.domain);90 if (bad) return kind === "push" ? bad.replace("pull from itself", "push to itself") : bad;91 if (!relays.includes(url)) relays.push(url);92 }93 if (relays.length === 0) return "invalid: at least one relay is needed";94 if (relays.length > 10) return "invalid: at most ten relays per job";95 const f = (r.filter && typeof r.filter === "object" ? r.filter : {}) as Record<string, unknown>;96 const filter: PullFilter = {};97 if (Array.isArray(f.authors) && f.authors.length) {98 if (!f.authors.every(hex64) || f.authors.length > 50) return "invalid: authors must be up to fifty hex pubkeys";99 filter.authors = [...new Set(f.authors as string[])];100 }101 if (Array.isArray(f.kinds) && f.kinds.length) {102 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";103 filter.kinds = [...new Set(f.kinds as number[])];104 }105 if (f.since !== undefined && f.since !== null && f.since !== 0) {106 if (!Number.isInteger(f.since) || (f.since as number) < 0) return "invalid: since must be a unix time";107 filter.since = f.since as number;108 }109 const every = r.every === undefined ? 0 : Number(r.every);110 if (!(EVERY as readonly number[]).includes(every)) return "invalid: every must be 0, 1, 6 or 24 hours";111 if (kind === "push" && relay.settings.policy.reads === "members" && filter.kinds?.some(isPrivate)) return "restricted: a members-only relay does not rebroadcast private kinds";112 return { kind, label, relays, filter, every, running: false, startedAt: 0, rounds: 0, failures: 0, relayIndex: 0, cursor: 0, targetCursors: {}, targetStatus: {}, targetAttempts: {}, stored: 0, skipped: 0, blobs: 0, sent: 0, refused: 0, last: null };113}114115// relaysFromList reads the owner's kind 10002 stored on this relay and116// returns its relays, both markers, without this relay itself.117export function relaysFromList(relay: Relay, pubkey: string): string[] {118 const rows = relay.store.query({ kinds: [10002], authors: [pubkey], tags: {} }, { pubkeys: [], all: true }, 1, now()).rows;119 if (rows.length === 0) return [];120 const e = JSON.parse(rows[0]) as Event;121 const out: string[] = [];122 for (const t of e.tags) {123 if (t[0] !== "r" || typeof t[1] !== "string") continue;124 const url = t[1].trim().replace(/\/+$/, "");125 if (checkPullURL(url, relay.slug, relay.domain)) continue;126 if (!out.includes(url)) out.push(url);127 }128 return out;129}130131// runRound does one round of a job. more: call again for this run.132export async function runRound(relay: Relay, job: Job): Promise<{ more: boolean; error: string }> {133 if (job.kind === "pull") return runPullSourceRound(relay, job);134 if (job.kind === "mirror") return runMirrorRound(relay, job);135 if (job.kind === "import") return runImportRound(relay, job);136 return runPushRound(relay, job);137}138139// A pull job syncs its sources one after another; each source is a pull140// in the sense of pull.ts, with the job's filter.141export async function runPullSourceRound(relay: Relay, job: Job, connect?: PullConnect): Promise<{ more: boolean; error: string }> {142 if (job.relayIndex >= job.relays.length) return { more: false, error: "" };143 const sources = job.pullSources ??= job.relays.map((url) => ({ url, ...newPullProgress(), stored: 0, skipped: 0, blobs: 0 }));144 const source = sources[job.relayIndex];145 const sub: PullJob = { url: source.url, startedAt: job.startedAt, rounds: 0, stored: 0, skipped: 0, blobs: 0, failures: 0, progress: source };146 if (job.filter.authors || job.filter.kinds || job.filter.since) sub.filter = job.filter;147 const r = await runPullRound(relay, sub, connect);148 job.rounds++;149 job.stored += sub.stored;150 job.skipped += sub.skipped;151 job.blobs += sub.blobs;152 source.stored += sub.stored;153 source.skipped += sub.skipped;154 source.blobs += sub.blobs;155 if (r.error) {156 source.failures++;157 const refused = /auth-required:|query refused:|files refused:/.test(r.error);158 if (!refused && source.failures < 3) return r;159 source.status = refused ? "refused" : "failed";160 source.error = r.error;161 job.relayIndex++;162 } else {163 source.failures = 0;164 source.error = "";165 if (!r.more) job.relayIndex++;166 }167 return { more: job.relayIndex < job.relays.length, error: "" };168}169170// A push round takes the next batch past the cursor and offers it to every171// target. Events nobody else may publish (NIP-70 protected) are skipped, as172// are private kinds on a members-only relay. Refusals are counted; a target173// that refuses several in a row is left alone for the round.174async function runPushRound(relay: Relay, job: Job): Promise<{ more: boolean; error: string }> {175 job.rounds++;176 const f: Filter = { tags: {} };177 if (job.filter.authors) f.authors = job.filter.authors;178 if (job.filter.kinds) f.kinds = job.filter.kinds;179 if (job.filter.since) f.since = job.filter.since;180 job.targetCursors ??= {};181 job.targetStatus ??= {};182 job.targetAttempts ??= {};183 const membersOnly = relay.settings.policy.reads === "members";184 let more = false, failed = "", reached = 0;185 const legacyCursor = job.cursor;186 // Materialize every target's starting point before any target can advance187 // the legacy cursor. This is the compatibility bridge for old jobs.188 for (const url of job.relays) if (job.targetCursors[url] === undefined) job.targetCursors[url] = legacyCursor;189 for (const url of job.relays) {190 if ((job.targetAttempts[url] ?? 0) >= 3) continue;191 const cursor = job.targetCursors[url] ?? 0;192 const rows = relay.store.after(cursor, f, PUSH_BATCH, now());193 if (rows.length === 0) { job.targetStatus[url] = { status: "accepted", error: "", at: now() }; continue; }194 more = more || rows.length === PUSH_BATCH;195 const events: Event[] = [];196 for (const row of rows) {197 const e = JSON.parse(row.raw) as Event;198 if (hasTag(e, "-") || (membersOnly && isPrivate(e.kind))) { job.skipped++; continue; }199 events.push(e);200 }201 try {202 const refusedBefore = job.refused;203 await pushTo(relay, url, events, job);204 reached++;205 job.targetCursors[url] = rows[rows.length - 1].seq;206 job.targetStatus[url] = { status: job.refused > refusedBefore ? "rejected" : "accepted", error: job.refused > refusedBefore ? "Some events were explicitly refused." : "", at: now() };207 job.targetAttempts[url] = 0;208 job.cursor = Math.max(job.cursor, rows[rows.length - 1].seq);209 } catch (err) {210 failed = url + ": " + (err instanceof Error ? err.message : String(err));211 job.targetAttempts[url] = (job.targetAttempts[url] ?? 0) + 1;212 job.targetStatus[url] = { status: job.targetAttempts[url] >= 3 ? "rejected" : "pending", error: failed, at: now() };213 more = true;214 }215 }216 if (reached === 0 && failed) return { more: true, error: failed };217 return { more, error: "" };218}219220async function pushTo(relay: Relay, url: string, events: Event[], job: Job): Promise<void> {221 if (events.length === 0) return;222 const sock = new Socket(await dial(relay, url));223 try {224 let refusedInARow = 0;225 for (let i = 0; i < events.length; i += PUSH_PER_TARGET) {226 const batch = events.slice(i, i + PUSH_PER_TARGET);227 const pending = new Set(batch.map((e) => e.id));228 for (const e of batch) sock.send("EVENT", e);229 const deadline = Date.now() + PUSH_TIMEOUT_MS;230 while (pending.size && Date.now() < deadline) {231 const m = await sock.recv(deadline - Date.now());232 if (m[0] !== "OK" || typeof m[1] !== "string" || !pending.has(m[1])) continue;233 pending.delete(m[1]);234 const msg = String(m[3] ?? "");235 if (m[2] === true || msg.startsWith("duplicate:")) {236 job.sent++;237 refusedInARow = 0;238 } else {239 job.refused++;240 refusedInARow++;241 }242 }243 if (pending.size) {244 job.refused += pending.size;245 throw new Error("the relay stopped answering");246 }247 if (refusedInARow >= REFUSALS_PER_TARGET) throw new Error("target refused too many events");248 }249 } finally {250 sock.close();251 }252}253254// startRun resets the counters of a run. The push cursor is kept so a255// standing push forwards only what arrived since.256export function startRun(job: Job, t: number) {257 job.running = true;258 job.startedAt = t;259 job.rounds = 0;260 job.failures = 0;261 job.relayIndex = 0;262 job.stored = 0;263 job.skipped = 0;264 job.blobs = 0;265 job.sent = 0;266 job.refused = 0;267 job.duplicates = 0;268 if (job.kind === "push") job.targetAttempts = {};269 if (job.kind === "pull") job.pullSources = job.relays.map((url) => ({ url, ...newPullProgress(), stored: 0, skipped: 0, blobs: 0 }));270}271272// finishRun closes the current run and schedules the next one.273export function finishRun(job: Job, error: string, t: number) {274 if (!error && job.kind === "push" && Object.values(job.targetStatus ?? {}).some((s) => s.status === "rejected")) error = "Some delivery targets refused events or exhausted retries; inspect target results.";275 const incomplete = job.pullSources?.filter((s) => ["partial", "refused", "failed"].includes(s.status));276 if (!error && incomplete?.length) error = `${incomplete.length} import source(s) incomplete; inspect Source results for details.`;277 job.last = { finishedAt: t, error, rounds: job.rounds, stored: job.stored, skipped: job.skipped, blobs: job.blobs, sent: job.sent, refused: job.refused, duplicates: job.duplicates ?? 0, ...(job.pullSources ? { sources: structuredClone(job.pullSources) } : {}) };278 job.running = false;279 job.nextRun = job.every > 0 ? t + job.every * 3600 : 0;280}281282// pruneFinished drops the oldest finished once jobs beyond a few, so the283// list stays a list.284export function pruneFinished(jobs: Job[]): Job[] {285 const done = jobs.filter((j) => !j.running && j.every === 0 && j.nextRun === 0).sort((a, b) => (b.last?.finishedAt ?? 0) - (a.last?.finishedAt ?? 0));286 const drop = new Set(done.slice(KEEP_FINISHED).map((j) => j.id));287 return jobs.filter((j) => !drop.has(j.id));288}289290// pullView presents pull jobs in the shape the first console and tests291// used: the running pull, if any, and the last finished one.292export function pullView(jobs: Job[]): { running: PullJob | null; last: PullResult | null } {293 const pulls = jobs.filter((j) => j.kind === "pull");294 const run = pulls.find((j) => j.running || (j.nextRun > 0 && j.nextRun <= now() && j.every === 0));295 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;296 // Latest finished wins; on the same second the job added later wins.297 let done: Job | undefined;298 for (const j of pulls) if (j.last && !j.running && (!done || (j.last.finishedAt >= (done.last as JobResult).finishedAt))) done = j;299 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;300 return { running, last };301}302303export function newJobID(): string {304 const b = crypto.getRandomValues(new Uint8Array(6));305 return [...b].map((x) => x.toString(16).padStart(2, "0")).join("");306}