feat(federation): lean federationRecovery module (probe + markPeerRecovered)

packages/server/src/utils/federationRecovery.test.ts

@@ -0,0 +1,80 @@
+import { describe, it, expect, beforeEach, vi } from 'vitest';
+import Database from 'better-sqlite3';
+import { drizzle } from 'drizzle-orm/better-sqlite3';
+import fs from 'node:fs';
+import path from 'node:path';
+import { fileURLToPath } from 'node:url';
+import * as schema from '../db/schema.js';
+import { eq } from 'drizzle-orm';
+
+const __dirname = path.dirname(fileURLToPath(import.meta.url));
+type TestDb = ReturnType<typeof drizzle<typeof schema>>;
+let sqlite: Database.Database;
+let testDb: TestDb;
+
+vi.mock('../db/index.js', () => ({ getDb: () => testDb, schema }));
+
+const onPeerActivated = vi.fn();
+vi.mock('./federationPeerActivation.js', () => ({ onPeerActivated }));
+
+function applyMigrations(db: Database.Database): void {
+  const dir = path.resolve(__dirname, '../../drizzle');
+  for (const f of fs.readdirSync(dir).filter(x => x.endsWith('.sql')).sort()) {
+    const sql = fs.readFileSync(path.join(dir, f), 'utf8');
+    for (const stmt of sql.split(/-->\s*statement-breakpoint/)) {
+      const clean = stmt.trim();
+      if (clean) db.exec(clean);
+    }
+  }
+}
+
+function seedUnreachable(id: string, attempts = 0, lastProbeAt: number | null = null): void {
+  testDb.insert(schema.federationPeers).values({
+    id, origin: 'https://peer.example', hmacSecret: 'secret',
+    status: 'unreachable', consecutiveFailures: 10,
+    probeAttempts: attempts, lastProbeAt,
+    lastSyncedAt: Date.now(), createdAt: Date.now(),
+  }).run();
+}
+
+describe('federationRecovery primitives', () => {
+  beforeEach(() => {
+    sqlite = new Database(':memory:');
+    testDb = drizzle(sqlite, { schema });
+    applyMigrations(sqlite);
+    vi.clearAllMocks();
+  });
+
+  it('probePeerReachable returns true on a 200 from /api/instance/info', async () => {
+    const spy = vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('{}', { status: 200 }));
+    const { probePeerReachable } = await import('./federationRecovery.js');
+    await expect(probePeerReachable('https://peer.example')).resolves.toBe(true);
+    expect(spy).toHaveBeenCalledWith('https://peer.example/api/instance/info', expect.anything());
+  });
+
+  it('probePeerReachable returns false on a non-ok response', async () => {
+    vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('', { status: 502 }));
+    const { probePeerReachable } = await import('./federationRecovery.js');
+    await expect(probePeerReachable('https://peer.example')).resolves.toBe(false);
+  });
+
+  it('probePeerReachable returns false on network error', async () => {
+    vi.spyOn(globalThis, 'fetch').mockRejectedValue(new Error('ENOTFOUND'));
+    const { probePeerReachable } = await import('./federationRecovery.js');
+    await expect(probePeerReachable('https://peer.example')).resolves.toBe(false);
+  });
+
+  it('markPeerRecovered flips status to active, resets pacing + counters, calls onPeerActivated', async () => {
+    seedUnreachable('peer-rec', 3, Date.now());
+    const { markPeerRecovered } = await import('./federationRecovery.js');
+    await markPeerRecovered('peer-rec');
+    const row = testDb.select().from(schema.federationPeers)
+      .where(eq(schema.federationPeers.id, 'peer-rec')).get()!;
+    expect(row.status).toBe('active');
+    expect(row.consecutiveFailures).toBe(0);
+    expect(row.probeAttempts).toBe(0);
+    expect(row.lastProbeAt).toBeNull();
+    expect(row.lastSeenAt).toBeGreaterThan(0);
+    expect(onPeerActivated).toHaveBeenCalledWith('peer-rec', 'health_check_recovery');
+  });
+});

packages/server/src/utils/federationRecovery.ts

@@ -0,0 +1,45 @@
+import { getDb } from '../db/index.js';
+import * as schema from '../db/schema.js';
+import { eq } from 'drizzle-orm';
+import { onPeerActivated } from './federationPeerActivation.js';
+
+/** Reachability-probe timeout (ms). */
+export const RECOVERY_PROBE_TIMEOUT_MS = 10_000;
+
+/**
+ * Liveness probe shared by the recovery tick and the manual recheck endpoint.
+ * GET {origin}/api/instance/info with a 10s timeout. No HMAC — reachability is
+ * not trust; a recovered-but-HMAC-broken peer still transitions to
+ * needs_attention via the auth-failure path on the next real delivery.
+ */
+export async function probePeerReachable(origin: string, signal?: AbortSignal): Promise<boolean> {
+  try {
+    const timeout = AbortSignal.timeout(RECOVERY_PROBE_TIMEOUT_MS);
+    const response = await fetch(`${origin}/api/instance/info`, {
+      signal: signal ? AbortSignal.any([signal, timeout]) : timeout,
+    });
+    return response.ok;
+  } catch {
+    return false;
+  }
+}
+
+/**
+ * Transition an unreachable peer back to active and reset recovery pacing.
+ * onPeerActivated broadcasts federation_peers_changed to admins. Rotation fields
+ * are intentionally untouched — recovery is orthogonal to rotation.
+ */
+export async function markPeerRecovered(peerId: string): Promise<void> {
+  const db = getDb();
+  db.update(schema.federationPeers)
+    .set({
+      status: 'active',
+      consecutiveFailures: 0,
+      lastSeenAt: Date.now(),
+      probeAttempts: 0,
+      lastProbeAt: null,
+    })
+    .where(eq(schema.federationPeers.id, peerId))
+    .run();
+  await onPeerActivated(peerId, 'health_check_recovery');
+}