feat(journal): federation protocol doc + delivery observability

Task group 4 of federation-hardening.

4.1 — FEDERATION.md at the repo root: actor discovery (WebFinger, actor,
NodeInfo), object/activity types with real JSON examples (Note, Create,
Delete, the narrow follow-graph inbox), addressing, HTTP-Signature
expectations, the two-layer dedup contract, durable delivery/retry
policy, and blocklist moderation semantics — precise enough for another
implementation to interoperate. Linked from README and docs/architecture.

4.2 — three prom-client metrics + a journal dashboard row:
- `federation_delivery_total{outcome}` — incremented in deliver-activity
  (delivered/skipped/failed).
- `federation_inbox_dropped_total{reason}` — incremented at every inbox
  drop (duplicate | blocked); this is the counter deferred from task 3.2.
- `federation_queue_depth` — gauge sampled at scrape time in
  /api/metrics from PgBossMessageQueue.getDepth(); the restart-loss
  regression detector.
Grafana journal.json gains a Federation row (delivery rate, queue depth,
inbox drops); the logs panels shift down to make room.

Verified: dashboard JSON valid; journal typecheck + lint clean; unit
suite 357 passing (route-template guard unaffected).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

apps/journal/app/jobs/deliver-activity.ts

@@ -15,6 +15,7 @@ import {
   type DeliveryPayload,
 } from "../lib/federation-delivery.server.ts";
 import { logger } from "../lib/logger.server.ts";
+import { federationDeliveryTotal } from "../lib/metrics.server.ts";
 
 /**
  * Outbound pacing (spec 5.5): never exceed 1 request/second per remote
@@ -46,8 +47,10 @@ export const deliverActivityJob = defineJournalJob({
     for (const job of jobs) {
       const p = job.data;
       try {
-        await deliverOne(p);
+        const outcome = await deliverOne(p);
+        federationDeliveryTotal.inc({ outcome });
       } catch (err) {
+        federationDeliveryTotal.inc({ outcome: "failed" });
         logger.warn(
           { err, action: p.action, objectIri: p.objectIri, recipient: p.recipientActorIri },
           "deliver-activity attempt failed (pg-boss will retry until budget exhausted)",
@@ -58,7 +61,7 @@ export const deliverActivityJob = defineJournalJob({
   },
 });
 
-async function deliverOne(p: DeliveryPayload): Promise<void> {
+async function deliverOne(p: DeliveryPayload): Promise<"delivered" | "skipped"> {
   const federation = getFederation();
   const ctx = federation.createContext(new URL(getOrigin()), undefined);
 
@@ -75,7 +78,7 @@ async function deliverOne(p: DeliveryPayload): Promise<void> {
       .limit(1);
     if (!row) {
       logger.info({ objectIri: p.objectIri }, "deliver-activity: activity gone or non-public; skipping");
-      return;
+      return "skipped";
     }
     // Spec 9.3: flipping the profile to private stops federation — also
     // for deliveries already enqueued when the flip happened.
@@ -86,7 +89,7 @@ async function deliverOne(p: DeliveryPayload): Promise<void> {
       .limit(1);
     if (!owner || owner.profileVisibility !== "public") {
       logger.info({ objectIri: p.objectIri }, "deliver-activity: owner no longer public; skipping");
-      return;
+      return "skipped";
     }
     activity = activityToCreate(row as FederatableActivity, p.ownerUsername);
   } else {
@@ -127,4 +130,5 @@ async function deliverOne(p: DeliveryPayload): Promise<void> {
     { action: p.action, objectIri: p.objectIri, recipient: p.recipientActorIri },
     "deliver-activity: delivered",
   );
+  return "delivered";
 }

apps/journal/app/lib/federation.server.ts

@@ -38,6 +38,7 @@ import { PostgresKvStore } from "./federation-kv.server.ts";
 import { PgBossMessageQueue } from "./federation-queue.server.ts";
 import { markInboundActivityProcessed } from "./federation-replay.server.ts";
 import { isBlockedIri } from "./federation-blocklist.server.ts";
+import { federationInboxDroppedTotal } from "./metrics.server.ts";
 import { ensureUserKeypair, loadUserKeypair } from "./federation-keys.server.ts";
 import { activityToCreate, activityToNote } from "./federation-objects.server.ts";
 import {
@@ -273,8 +274,8 @@ function buildFederation(): Federation<void> {
       // when the local target is public; otherwise drop (the actor
       // already 404s for private users).
       if (follow.id == null || follow.actorId == null || follow.objectId == null) return;
-      if (await isBlockedIri(follow.actorId.href)) return; // blocked instance: silent 202 drop
-      if (!(await markInboundActivityProcessed(follow.id.href)).fresh) return; // replay: drop
+      if (await isBlockedIri(follow.actorId.href)) { federationInboxDroppedTotal.inc({ reason: "blocked" }); return; } // silent 202 drop
+      if (!(await markInboundActivityProcessed(follow.id.href)).fresh) { federationInboxDroppedTotal.inc({ reason: "duplicate" }); return; } // replay: drop
       const parsed = ctx.parseUri(follow.objectId);
       if (parsed?.type !== "actor") return;
       const { outcome } = await recordRemoteFollow(follow.actorId.href, parsed.identifier);
@@ -295,8 +296,8 @@ function buildFederation(): Federation<void> {
       // Spec 4.3: Undo(Follow) removes the follow row. Other Undos are
       // acknowledged and dropped.
       if (undo.actorId == null) return;
-      if (await isBlockedIri(undo.actorId.href)) return; // blocked instance: silent 202 drop
-      if (undo.id != null && !(await markInboundActivityProcessed(undo.id.href)).fresh) return; // replay: drop
+      if (await isBlockedIri(undo.actorId.href)) { federationInboxDroppedTotal.inc({ reason: "blocked" }); return; } // silent 202 drop
+      if (undo.id != null && !(await markInboundActivityProcessed(undo.id.href)).fresh) { federationInboxDroppedTotal.inc({ reason: "duplicate" }); return; } // replay: drop
       const undoObjectId = undo.objectId; // capture before dereference (see Accept)
       const object = await undo.getObject(ctx);
       if (object instanceof Follow && object.objectId != null) {
@@ -325,8 +326,8 @@ function buildFederation(): Federation<void> {
       // Spec 4.4: a remote accepted our outgoing Follow — settle the
       // Pending row and trigger the first outbox poll for that actor.
       if (accept.actorId == null) return;
-      if (await isBlockedIri(accept.actorId.href)) return; // blocked instance: silent 202 drop
-      if (accept.id != null && !(await markInboundActivityProcessed(accept.id.href)).fresh) return; // replay: drop
+      if (await isBlockedIri(accept.actorId.href)) { federationInboxDroppedTotal.inc({ reason: "blocked" }); return; } // silent 202 drop
+      if (accept.id != null && !(await markInboundActivityProcessed(accept.id.href)).fresh) { federationInboxDroppedTotal.inc({ reason: "duplicate" }); return; } // replay: drop
       // Capture the raw object reference BEFORE dereferencing:
       // getObject() memoizes the fetched document, after which objectId
       // reports the fetched object's id (fragment stripped) instead of
@@ -366,8 +367,8 @@ function buildFederation(): Federation<void> {
     .on(Reject, async (ctx, reject) => {
       // Spec 4.5: remote refused our Follow — drop the Pending row.
       if (reject.actorId == null) return;
-      if (await isBlockedIri(reject.actorId.href)) return; // blocked instance: silent 202 drop
-      if (reject.id != null && !(await markInboundActivityProcessed(reject.id.href)).fresh) return; // replay: drop
+      if (await isBlockedIri(reject.actorId.href)) { federationInboxDroppedTotal.inc({ reason: "blocked" }); return; } // silent 202 drop
+      if (reject.id != null && !(await markInboundActivityProcessed(reject.id.href)).fresh) { federationInboxDroppedTotal.inc({ reason: "duplicate" }); return; } // replay: drop
       const objectId = reject.objectId; // capture before dereference (see Accept)
       const object = await reject.getObject(ctx);
       let localUser: Awaited<ReturnType<typeof findLocalPublicUserByIri>> = null;

apps/journal/app/lib/metrics.server.ts

@@ -48,6 +48,42 @@ export const demoBotSyntheticActivitiesTotal = getOrCreate(
     }),
 );
 
+// --- Federation metrics (spec: federation-operations "Federation delivery
+// observability") ---------------------------------------------------------
+
+/** Outbound delivery attempts by outcome (delivered | skipped | failed). */
+export const federationDeliveryTotal = getOrCreate(
+  "federation_delivery_total",
+  () =>
+    new client.Counter({
+      name: "federation_delivery_total",
+      help: "Outbound federation delivery attempts by outcome",
+      labelNames: ["outcome"] as const,
+    }),
+);
+
+/** Inbound activities dropped, by reason (duplicate | blocked). */
+export const federationInboxDroppedTotal = getOrCreate(
+  "federation_inbox_dropped_total",
+  () =>
+    new client.Counter({
+      name: "federation_inbox_dropped_total",
+      help: "Inbound federation activities dropped before side effects, by reason",
+      labelNames: ["reason"] as const,
+    }),
+);
+
+/** Messages waiting in the durable Fedify queue. Set at scrape time by the
+ * metrics route (the restart-loss regression detector). */
+export const federationQueueDepth = getOrCreate(
+  "federation_queue_depth",
+  () =>
+    new client.Gauge({
+      name: "federation_queue_depth",
+      help: "Messages waiting in the durable Fedify (pg-boss) message queue",
+    }),
+);
+
 export const registry = client.register;
 
 // --- Route label normalization -------------------------------------------

apps/journal/app/routes/api.metrics.ts

@@ -1,6 +1,19 @@
-import { registry } from "~/lib/metrics.server";
+import { registry, federationQueueDepth } from "~/lib/metrics.server";
 
 export async function loader() {
+  // Sample the durable Fedify queue depth at scrape time (the
+  // restart-loss regression detector). Best-effort: if federation is off
+  // or the boss isn't up yet, leave the last value rather than fail the
+  // scrape.
+  if (process.env.FEDERATION_ENABLED === "true") {
+    try {
+      const { PgBossMessageQueue } = await import("~/lib/federation-queue.server");
+      const depth = await new PgBossMessageQueue().getDepth();
+      federationQueueDepth.set(depth.queued);
+    } catch {
+      // boss not initialized / transient DB error — keep the prior gauge value
+    }
+  }
   const metrics = await registry.metrics();
   return new Response(metrics, {
     headers: { "Content-Type": registry.contentType },