From b8fc8fadffffa2aa497631a7960f2b3e2ca10f92 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ullrich=20Sch=C3=A4fer?= Date: Mon, 13 Jul 2026 21:46:46 +0200 Subject: [PATCH] feat(journal): durable Fedify message queue over pg-boss MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task group 1 of federation-hardening. Fedify was configured with `InProcessMessageQueue`, so every queued outbound delivery, its pending retry state, and inbox processing task was lost on a container restart (routine here: deploys, OOM history) — directly contradicting the social-federation spec's promise that fan-out survives a deploy. - `federation-queue.server.ts`: `PgBossMessageQueue` implementing Fedify's `MessageQueue` over the pg-boss instance the journal already runs, mirroring the `PostgresKvStore` adapter. `nativeRetrial = false` keeps Fedify the retry-policy owner; pg-boss supplies durability + delayed jobs (delay → whole-second `startAfter`, `retryLimit: 0`). - Swap it in for `InProcessMessageQueue` in `federation.server.ts`; document the two intentional queueing layers (our fan-out jobs feed Fedify; Fedify's sends now durable underneath). - `server.ts`: create the durable queue at startup when federation is on. - Broaden the structural `BossLike` in `boss.server.ts` with the work/offWork/createQueue/getQueue methods the adapter needs. - Tests: enqueue maps retry/delay correctly, consume roundtrip, restart durability (fresh listener drains a prior instance's backlog), abort stops the worker, depth reports ready vs delayed. Verified: journal typecheck + lint clean, 7/7 new unit tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/journal/app/lib/boss.server.ts | 26 +++- .../app/lib/federation-queue.server.test.ts | 129 ++++++++++++++++++ .../app/lib/federation-queue.server.ts | 77 +++++++++++ apps/journal/app/lib/federation.server.ts | 20 ++- apps/journal/server.ts | 6 + .../changes/federation-hardening/tasks.md | 6 +- 6 files changed, 252 insertions(+), 12 deletions(-) create mode 100644 apps/journal/app/lib/federation-queue.server.test.ts create mode 100644 apps/journal/app/lib/federation-queue.server.ts diff --git a/apps/journal/app/lib/boss.server.ts b/apps/journal/app/lib/boss.server.ts index b1b7880..61c0182 100644 --- a/apps/journal/app/lib/boss.server.ts +++ b/apps/journal/app/lib/boss.server.ts @@ -18,18 +18,40 @@ import { logger } from "./logger.server.ts"; import type { JobName, JobPayloads } from "../jobs/payloads.ts"; -// Structurally typed (we only need `send`) so we don't have to pull -// pg-boss into the journal app's dep graph just for the typedef. +// Structurally typed so we don't have to pull pg-boss into the journal +// app's dep graph just for the typedef. Kept to the subset feature code +// actually uses: `send` for enqueues, plus `work`/`offWork`/`createQueue`/ +// `getQueue` for the Fedify message-queue adapter (federation-queue.server.ts). export interface BossSendOptions { retryLimit?: number; retryBackoff?: boolean; retryDelay?: number; + /** Seconds to defer the job (pg-boss `startAfter`). */ + startAfter?: number; /** pg-boss dedup: collapses enqueues sharing a key into one active job. */ singletonKey?: string; } +/** A pg-boss job as our handlers see it (subset of pg-boss's `Job`). */ +export interface BossJob { + id: string; + name: string; + data: T; +} + +/** Queue depth counters (subset of pg-boss's `QueueResult`). */ +export interface BossQueueDepth { + queuedCount: number; + readyCount: number; + deferredCount: number; +} + interface BossLike { send(queueName: string, data: unknown, options?: BossSendOptions): Promise; + work(queueName: string, handler: (jobs: BossJob[]) => Promise): Promise; + offWork(queueName: string): Promise; + createQueue(queueName: string, options?: unknown): Promise; + getQueue(queueName: string): Promise; } const BOSS_KEY = Symbol.for("trails-cool.journal.pg-boss"); diff --git a/apps/journal/app/lib/federation-queue.server.test.ts b/apps/journal/app/lib/federation-queue.server.test.ts new file mode 100644 index 0000000..dc9600f --- /dev/null +++ b/apps/journal/app/lib/federation-queue.server.test.ts @@ -0,0 +1,129 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { setBoss } from "./boss.server.ts"; +import { PgBossMessageQueue, FEDERATION_QUEUE_NAME } from "./federation-queue.server.ts"; + +// A durable in-memory stand-in for pg-boss: `sent` survives across adapter +// instances (like rows in Postgres), which is what lets us model a restart +// — a fresh PgBossMessageQueue.listen() draining messages a previous +// instance enqueued. Only the methods the adapter uses are implemented. +interface SentJob { + data: unknown; + startAfter?: number; + retryLimit?: number; + consumed: boolean; +} + +class FakeBoss { + sent: SentJob[] = []; + offWorkCalls: string[] = []; + + async send(name: string, data: unknown, options?: { startAfter?: number; retryLimit?: number }) { + expect(name).toBe(FEDERATION_QUEUE_NAME); + this.sent.push({ data, startAfter: options?.startAfter, retryLimit: options?.retryLimit, consumed: false }); + return "job-id"; + } + + // Drain every ready (non-delayed, unconsumed) job through the handler, + // mirroring a worker that picks up the durable backlog on startup. + async work(name: string, handler: (jobs: Array<{ id: string; name: string; data: unknown }>) => Promise) { + expect(name).toBe(FEDERATION_QUEUE_NAME); + for (const job of this.sent) { + if (job.consumed || (job.startAfter ?? 0) > 0) continue; + job.consumed = true; + await handler([{ id: "j", name, data: job.data }]); + } + return "worker-id"; + } + + async offWork(name: string) { + this.offWorkCalls.push(name); + } + + async createQueue() {} + + async getQueue(name: string) { + expect(name).toBe(FEDERATION_QUEUE_NAME); + const pending = this.sent.filter((j) => !j.consumed); + const delayed = pending.filter((j) => (j.startAfter ?? 0) > 0).length; + return { queuedCount: pending.length, readyCount: pending.length - delayed, deferredCount: delayed }; + } +} + +let boss: FakeBoss; + +beforeEach(() => { + boss = new FakeBoss(); + setBoss(boss as unknown as Parameters[0]); +}); + +afterEach(() => { + setBoss(null); +}); + +describe("PgBossMessageQueue", () => { + it("enqueues with no retry (Fedify owns retries) and no delay by default", async () => { + const q = new PgBossMessageQueue(); + await q.enqueue({ hello: "world" }); + expect(boss.sent).toHaveLength(1); + const [job] = boss.sent; + expect(job).toMatchObject({ data: { hello: "world" }, retryLimit: 0 }); + expect(job?.startAfter).toBeUndefined(); + }); + + it("maps a Temporal delay to whole-second startAfter (rounded up)", async () => { + const q = new PgBossMessageQueue(); + const delay = { total: () => 1.2 } as unknown as Temporal.Duration; + await q.enqueue({ x: 1 }, { delay }); + expect(boss.sent[0]?.startAfter).toBe(2); + }); + + it("declares nativeRetrial=false so Fedify keeps ownership of retries", () => { + expect(new PgBossMessageQueue().nativeRetrial).toBe(false); + }); + + it("consume roundtrip: a listener receives the enqueued message payload", async () => { + const q = new PgBossMessageQueue(); + await q.enqueue({ type: "Create", id: "a" }); + const received: unknown[] = []; + const ac = new AbortController(); + // listen() resolves only on abort; register the worker (which drains + // the backlog), then stop. + const listening = q.listen((m) => void received.push(m), { signal: ac.signal }); + ac.abort(); + await listening; + expect(received).toEqual([{ type: "Create", id: "a" }]); + }); + + it("survives a restart: a fresh listener drains messages a prior instance enqueued", async () => { + // Prior process: enqueue two, never consumed (crash before delivery). + const before = new PgBossMessageQueue(); + await before.enqueue({ n: 1 }); + await before.enqueue({ n: 2 }); + + // New process: same durable store, brand-new adapter instance. + const after = new PgBossMessageQueue(); + const received: unknown[] = []; + const ac = new AbortController(); + const listening = after.listen((m) => void received.push(m), { signal: ac.signal }); + ac.abort(); + await listening; + + expect(received).toEqual([{ n: 1 }, { n: 2 }]); + }); + + it("listen resolves on abort and stops the worker", async () => { + const q = new PgBossMessageQueue(); + const ac = new AbortController(); + const listening = q.listen(() => {}, { signal: ac.signal }); + ac.abort(); + await expect(listening).resolves.toBeUndefined(); + expect(boss.offWorkCalls).toContain(FEDERATION_QUEUE_NAME); + }); + + it("reports queue depth split into ready and delayed", async () => { + const q = new PgBossMessageQueue(); + await q.enqueue({ n: 1 }); + await q.enqueue({ n: 2 }, { delay: { total: () => 30 } as unknown as Temporal.Duration }); + expect(await q.getDepth()).toEqual({ queued: 2, ready: 1, delayed: 1 }); + }); +}); diff --git a/apps/journal/app/lib/federation-queue.server.ts b/apps/journal/app/lib/federation-queue.server.ts new file mode 100644 index 0000000..d35a681 --- /dev/null +++ b/apps/journal/app/lib/federation-queue.server.ts @@ -0,0 +1,77 @@ +// Postgres-backed Fedify MessageQueue, implemented over the pg-boss +// instance the journal already runs (packages/jobs). Fedify uses this +// queue for outbound delivery (with its own retry scheduling) and inbox +// processing; the previous InProcessMessageQueue lost every queued +// message on restart (spec: federation-operations "Durable federation +// queue"). pg-boss gives us durable storage + delayed jobs; Fedify stays +// the retry-policy owner (see `nativeRetrial` below). +// +// This mirrors the PostgresKvStore adapter pattern (federation-kv.server.ts): +// a thin Fedify-interface shim over infrastructure we already operate, +// resolved lazily via getBoss() so it works in both the server-bootstrap +// module and the request-path bundle (see boss.server.ts on the two-copy +// module situation). +// +// Fedify fans a single MessageQueue instance out to three logical roles +// (inbox / outbox / fanout) and calls listen() once per role with an +// equivalent type-dispatching handler. Backing all three with one pg-boss +// queue is correct: any worker can process any message because Fedify +// routes by message type internally. + +import type { + MessageQueue, + MessageQueueEnqueueOptions, + MessageQueueListenOptions, + MessageQueueDepth, +} from "@fedify/fedify"; +import { getBoss } from "./boss.server.ts"; + +/** pg-boss queue name backing Fedify's inbox/outbox/fanout message queue. */ +export const FEDERATION_QUEUE_NAME = "federation.fedify"; + +export class PgBossMessageQueue implements MessageQueue { + // Fedify owns retry scheduling: on a failed delivery it re-enqueues with + // its own backoff policy, so pg-boss must NOT also retry (retryLimit 0 + // on enqueue) or every failure would be attempted twice. + readonly nativeRetrial = false; + + async enqueue(message: unknown, options?: MessageQueueEnqueueOptions): Promise { + // Fedify passes delay as a Temporal.Duration; pg-boss startAfter is + // whole seconds. Round up so a sub-second backoff still defers. + const seconds = options?.delay ? Math.ceil(options.delay.total("seconds")) : 0; + await getBoss().send(FEDERATION_QUEUE_NAME, message as object, { + retryLimit: 0, + ...(seconds > 0 ? { startAfter: seconds } : {}), + }); + } + + async listen( + handler: (message: unknown) => Promise | void, + options?: MessageQueueListenOptions, + ): Promise { + const boss = getBoss(); + await boss.work(FEDERATION_QUEUE_NAME, async (jobs) => { + for (const job of jobs) await handler(job.data); + }); + // Fedify's contract: listen() resolves when the signal aborts and never + // rejects; with no signal it never resolves (runs for the process + // lifetime, consuming the queue). + return new Promise((resolve) => { + const signal = options?.signal; + if (signal == null) return; + const stop = () => void boss.offWork(FEDERATION_QUEUE_NAME).finally(() => resolve()); + if (signal.aborted) stop(); + else signal.addEventListener("abort", stop, { once: true }); + }); + } + + async getDepth(): Promise { + const queue = await getBoss().getQueue(FEDERATION_QUEUE_NAME); + if (queue == null) return { queued: 0, ready: 0, delayed: 0 }; + return { + queued: queue.queuedCount, + ready: queue.readyCount, + delayed: queue.deferredCount, + }; + } +} diff --git a/apps/journal/app/lib/federation.server.ts b/apps/journal/app/lib/federation.server.ts index 021cc5f..190f37a 100644 --- a/apps/journal/app/lib/federation.server.ts +++ b/apps/journal/app/lib/federation.server.ts @@ -27,7 +27,7 @@ // exactly the "acknowledge and drop" the spec wants; the same applies // to wrapped types we inspect and ignore (e.g. Undo(Like)). import { configure, getConsoleSink } from "@logtape/logtape"; -import { createFederation, InProcessMessageQueue, type Federation } from "@fedify/fedify"; +import { createFederation, type Federation } from "@fedify/fedify"; import { Accept, Follow, Note, Person, PropertyValue, Reject, Undo } from "@fedify/fedify/vocab"; import { and, eq } from "drizzle-orm"; import { activities, users } from "@trails-cool/db/schema/journal"; @@ -35,6 +35,7 @@ import { getDb } from "./db.ts"; import { getOrigin } from "./config.server.ts"; import { localActorIri } from "./actor-iri.ts"; import { PostgresKvStore } from "./federation-kv.server.ts"; +import { PgBossMessageQueue } from "./federation-queue.server.ts"; import { ensureUserKeypair, loadUserKeypair } from "./federation-keys.server.ts"; import { activityToCreate, activityToNote } from "./federation-objects.server.ts"; import { @@ -148,12 +149,17 @@ function buildFederation(): Federation { // (e.g. the Accept we push back after a Follow) happen asynchronously // with Fedify's own retry policy, instead of inside the inbound HTTP // request — Mastodon gives deliveries a 10s read timeout, and a slow - // remote must never make us blow it. In-process is acceptable for - // the single-process journal: queued work is lost on restart, but - // every message type we enqueue is recoverable (remotes retry - // Follows; our own pg-boss jobs own activity delivery). Revisit with - // a Postgres-backed MessageQueue if that stops being true. - queue: new InProcessMessageQueue(), + // remote must never make us blow it. + // + // Two queueing layers coexist by design: our fan-out jobs + // (federation-delivery.server.ts enqueues one deliverActivity pg-boss + // job per follower) feed Fedify, and Fedify's own send/retry + // scheduling now runs on THIS durable pg-boss-backed queue instead of + // in-process. Before, in-process meant a deploy or crash dropped every + // queued delivery and pending retry (spec: fan-out survives a deploy); + // PgBossMessageQueue closes that. Collapsing the two layers into one + // is a refactor with no user-visible payoff, so they stay separate. + queue: new PgBossMessageQueue(), // Started explicitly below — keeps vitest runs (which build this // instance via handleFederationRequest) free of dangling timers. manuallyStartQueue: true, diff --git a/apps/journal/server.ts b/apps/journal/server.ts index 2fe7039..cda9678 100644 --- a/apps/journal/server.ts +++ b/apps/journal/server.ts @@ -165,6 +165,12 @@ server.listen(port, async () => { // users get keys before any federation traffic). Each run only // touches users whose public_key IS NULL, so repeats are no-ops. if (process.env.FEDERATION_ENABLED === "true") { + // Create the durable queue backing Fedify's message queue before any + // federation traffic can enqueue/consume on it. pg-boss requires queues + // to exist explicitly (v10+); createQueue is idempotent. + const { FEDERATION_QUEUE_NAME } = await import("./app/lib/federation-queue.server.ts"); + await boss.createQueue(FEDERATION_QUEUE_NAME); + await enqueue("backfill-user-keypairs", {}); logger.info("federation keypair backfill enqueued"); } diff --git a/openspec/changes/federation-hardening/tasks.md b/openspec/changes/federation-hardening/tasks.md index cc3d6b2..fe3ebbd 100644 --- a/openspec/changes/federation-hardening/tasks.md +++ b/openspec/changes/federation-hardening/tasks.md @@ -1,8 +1,8 @@ ## 1. Durable queue -- [ ] 1.1 Implement `federation-queue.server.ts`: Fedify `MessageQueue` over pg-boss (enqueue with delay, listen loop), mirroring the `PostgresKvStore` adapter pattern -- [ ] 1.2 Swap `InProcessMessageQueue` for it in `federation.server.ts`; document the two-layer queueing (fan-out jobs + Fedify queue) in a code comment -- [ ] 1.3 Tests: enqueue/delay/consume roundtrip; simulated restart (new listener picks up previously enqueued messages) +- [x] 1.1 Implement `federation-queue.server.ts`: Fedify `MessageQueue` over pg-boss (enqueue with delay, listen loop), mirroring the `PostgresKvStore` adapter pattern +- [x] 1.2 Swap `InProcessMessageQueue` for it in `federation.server.ts`; document the two-layer queueing (fan-out jobs + Fedify queue) in a code comment +- [x] 1.3 Tests: enqueue/delay/consume roundtrip; simulated restart (new listener picks up previously enqueued messages) ## 2. Replay defense