From a6c7584e0ccf39415f2f3ed2bd20683ec549b987 Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Fri, 26 Jun 2026 13:44:37 +0200 Subject: [PATCH] feat(federation): lean federationRecovery module (probe + markPeerRecovered) --- .../src/utils/federationRecovery.test.ts | 80 +++++++++++++++++++ .../server/src/utils/federationRecovery.ts | 45 +++++++++++ 2 files changed, 125 insertions(+) create mode 100644 packages/server/src/utils/federationRecovery.test.ts create mode 100644 packages/server/src/utils/federationRecovery.ts diff --git a/packages/server/src/utils/federationRecovery.test.ts b/packages/server/src/utils/federationRecovery.test.ts new file mode 100644 index 00000000..fea143cd --- /dev/null +++ b/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>; +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'); + }); +}); diff --git a/packages/server/src/utils/federationRecovery.ts b/packages/server/src/utils/federationRecovery.ts new file mode 100644 index 00000000..1bfffef7 --- /dev/null +++ b/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 { + 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 { + 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'); +}