diff --git a/docs/systems/federation.md b/docs/systems/federation.md index dd9d0236..47751731 100644 --- a/docs/systems/federation.md +++ b/docs/systems/federation.md @@ -313,6 +313,8 @@ Admin-initiated paths (`/peer/initiate`, `/approve`) do NOT call `ensurePeered`. HMAC-authenticated in **both directions**: the request is signed (only a peer holding the shared secret may call it — unknown/revoked peers → 403, bad signature → 401, missing headers → 400) **and the response body `{ instanceId }` is HMAC-signed** with the same secret (`X-Federation-Signature/Timestamp/Nonce` response headers). The caller (`fetchPeerEpoch(peer)` in `utils/federationEpoch.ts`) verifies that response signature with the same secret before trusting the value, then writes it to `federation_peers.peer_instance_id`. Response-signing (not TLS-only) is deliberate: a poisoned baseline could drive a spurious data-heal on a live peer, so the newly-trusted epoch is authenticated (design §9). `fetchPeerEpoch` **fails safe** — a `404` from a not-yet-upgraded peer, an absent/invalid response signature, or a network/timeout error all return `null` (10s timeout via `AbortSignal.timeout`); the caller treats `null` as "retry on the next tick," never as an error to surface. This is the deterministic populator of the epoch baseline (the bounded periodic epoch-refresh, design §3.2), independent of organic relay traffic. +**Deterministic epoch-refresh driver (`refreshPeerEpochs()` in `utils/federationEpoch.ts`).** Selects every `active` peer whose `peer_instance_id IS NULL`, calls `fetchPeerEpoch(peer)` once each, and on a non-null result writes the epoch via `UPDATE ... SET peer_instance_id WHERE id = ? AND peer_instance_id IS NULL`. The trailing `IS NULL` guard makes it **populate-if-null only** — it can never overwrite a baseline another path (relay envelope, handshake) already established — and makes it **self-terminating**: once a peer's `peer_instance_id` is set, the `IS NULL` filter excludes it, so it is never fetched again. A `null` from `fetchPeerEpoch` (404 / bad-sig / network) is a benign `continue` with no error log-spam, retried next tick. Wired into the federation worker in two places: once at `startFederationWorkers()` startup and once at the end of `processHealthCheckTick()` (the existing 15-minute health-check tick), both as `refreshPeerEpochs().catch(() => {})`. This guarantees the trusted baseline is populated within one refresh cycle of an upgrade, independent of user/relay activity — the load-bearing populator that relay-only population cannot cover for idle peers. + ### S2S Identity Deletion (`DELETE /api/federation/identity`) Allows a home instance to remove a user's replicated identity from a remote instance. @@ -1563,6 +1565,7 @@ All workers are started by `startFederationWorkers()` on server boot and stopped | Outbox delivery | 10s | 50 | 30s | `processOutboxTick` | | File download | 30s | 5 | 60s | `processFileQueueTick` | | Health check | 15min | all unreachable | 10s | `processHealthCheckTick` | +| Epoch-refresh baseline | Startup + 15min (end of health tick) | active peers w/ `peer_instance_id IS NULL` | 10s per peer | `refreshPeerEpochs` (populate-if-null, self-terminating) | | Janitor | 1h | -- | -- | `runFederationJanitor` (sync) | | Startup bootstrap sync | Once at startup | -- | 30s per page | `startupBootstrapSync` → `onPeerActivated` | diff --git a/packages/server/src/utils/federationEpoch.test.ts b/packages/server/src/utils/federationEpoch.test.ts index a0079f6d..1813eca7 100644 --- a/packages/server/src/utils/federationEpoch.test.ts +++ b/packages/server/src/utils/federationEpoch.test.ts @@ -2,6 +2,7 @@ import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; import Fastify, { type FastifyInstance } from 'fastify'; import Database from 'better-sqlite3'; import { drizzle } from 'drizzle-orm/better-sqlite3'; +import { eq } from 'drizzle-orm'; import fs from 'node:fs'; import path from 'node:path'; import { fileURLToPath } from 'node:url'; @@ -320,3 +321,93 @@ describe('fetchPeerEpoch — signs request, verifies signed response, fails safe expect(result).toBeNull(); }); }); + +describe('refreshPeerEpochs — deterministic populate-if-null baseline (self-terminating)', () => { + // Drives the REAL refreshPeerEpochs → fetchPeerEpoch → verifySignature round-trip. + // fetchPeerEpoch is deliberately NOT stubbed: a signing/arg-order mismatch must + // fail these assertions loudly rather than degrade to a silent null (which would + // masquerade as a benign 404 and quietly disable the whole refresh). + beforeEach(() => { + // Local instance epoch must be readable (getOurOrigin does not need it, but the + // module is shared; seed for parity with real boot state). + seedInstanceSettings(LOCAL_EPOCH); + seedActivePeer(); + }); + + afterEach(() => { + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + }); + + /** A response body signed with `secret` over exactly the bytes we return. */ + function signedEpochResponse(instanceId: string, secret: string): Response { + const responseBody = JSON.stringify({ instanceId }); + const sigHeaders = buildFederationHeaders(responseBody, secret, PEER_ORIGIN); + return new Response(responseBody, { status: 200, headers: sigHeaders }); + } + + function readPeerInstanceId(): string | null { + const row = testDb + .select({ peerInstanceId: schema.federationPeers.peerInstanceId }) + .from(schema.federationPeers) + .where(eq(schema.federationPeers.id, 'peer-remote')) + .get(); + return row?.peerInstanceId ?? null; + } + + it('populates peer_instance_id from a validly-signed response, then self-terminates', async () => { + const fetchMock = vi.fn(async () => signedEpochResponse('E1', PEER_SECRET)); + vi.stubGlobal('fetch', fetchMock); + + const { refreshPeerEpochs } = await import('./federationEpoch.js'); + await refreshPeerEpochs(); + + expect(readPeerInstanceId()).toBe('E1'); + expect(fetchMock).toHaveBeenCalledTimes(1); + + // Second pass: the peer is now non-null, so the IS NULL filter excludes it — + // no further fetch is issued. Self-termination is structural, not incidental. + await refreshPeerEpochs(); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(readPeerInstanceId()).toBe('E1'); + }); + + it('leaves the baseline NULL when the response signature is invalid (tampered)', async () => { + // Signed with a different secret → verification fails → fetchPeerEpoch returns null. + const fetchMock = vi.fn(async () => signedEpochResponse('E1', 'a-different-secret')); + vi.stubGlobal('fetch', fetchMock); + + const { refreshPeerEpochs } = await import('./federationEpoch.js'); + await refreshPeerEpochs(); + + expect(readPeerInstanceId()).toBeNull(); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it('leaves the baseline NULL and does not throw on a 404 (peer not yet upgraded)', async () => { + const fetchMock = vi.fn(async () => new Response('Not found', { status: 404 })); + vi.stubGlobal('fetch', fetchMock); + + const { refreshPeerEpochs } = await import('./federationEpoch.js'); + await expect(refreshPeerEpochs()).resolves.toBeUndefined(); + + expect(readPeerInstanceId()).toBeNull(); + }); + + it('never overwrites an already-populated baseline (populate-if-null only)', async () => { + testDb.update(schema.federationPeers) + .set({ peerInstanceId: 'pre-existing' }) + .where(eq(schema.federationPeers.id, 'peer-remote')) + .run(); + + const fetchMock = vi.fn(async () => signedEpochResponse('E1', PEER_SECRET)); + vi.stubGlobal('fetch', fetchMock); + + const { refreshPeerEpochs } = await import('./federationEpoch.js'); + await refreshPeerEpochs(); + + // Already non-null → excluded by the IS NULL filter → no fetch, value untouched. + expect(fetchMock).not.toHaveBeenCalled(); + expect(readPeerInstanceId()).toBe('pre-existing'); + }); +}); diff --git a/packages/server/src/utils/federationEpoch.ts b/packages/server/src/utils/federationEpoch.ts index 5f35d0b0..bbb82388 100644 --- a/packages/server/src/utils/federationEpoch.ts +++ b/packages/server/src/utils/federationEpoch.ts @@ -1,4 +1,4 @@ -import { eq } from 'drizzle-orm'; +import { and, eq, isNull } from 'drizzle-orm'; import { getDb, schema } from '../db/index.js'; import { buildFederationHeaders, verifySignature, getOurOrigin } from './federationAuth.js'; @@ -86,3 +86,52 @@ export async function fetchPeerEpoch(peer: PeerForEpoch): Promise return null; } } + +/** + * Deterministic baseline populator: for each `active` peer whose + * `peer_instance_id` is still NULL, fetch its authenticated epoch once and store + * it. This is the load-bearing guarantee (design §3.2) — it populates the + * trusted baseline within one refresh cycle of an upgrade, independent of any + * user/relay activity, closing the window that relay-only population leaves for + * idle peers. + * + * Populate-if-null ONLY: the `UPDATE ... WHERE peer_instance_id IS NULL` guard + * makes it structurally impossible to overwrite a baseline that another path + * (relay, handshake) already established. Self-terminating: once a peer's + * `peer_instance_id` is set, the `isNull` filter excludes it, so it is never + * fetched again. + * + * Staggered-rollout tolerant: `fetchPeerEpoch` returns `null` for a 404 + * (not-yet-upgraded peer), a bad/absent response signature, or a network error. + * All of those are benign no-ops — we simply skip the peer and retry on the next + * tick, with no error log-spam. No exception escapes this function. + */ +export async function refreshPeerEpochs(): Promise { + const db = getDb(); + const peers = db + .select({ + id: schema.federationPeers.id, + origin: schema.federationPeers.origin, + hmacSecret: schema.federationPeers.hmacSecret, + }) + .from(schema.federationPeers) + .where(and( + eq(schema.federationPeers.status, 'active'), + isNull(schema.federationPeers.peerInstanceId), + )) + .all(); + + for (const peer of peers) { + const epoch = await fetchPeerEpoch(peer); + if (!epoch) continue; // 404 / bad-sig / network → retry next tick, no log-spam. + + // Populate-if-null only: the IS NULL guard never overwrites a non-null baseline. + db.update(schema.federationPeers) + .set({ peerInstanceId: epoch }) + .where(and( + eq(schema.federationPeers.id, peer.id), + isNull(schema.federationPeers.peerInstanceId), + )) + .run(); + } +} diff --git a/packages/server/src/utils/federationWorker.ts b/packages/server/src/utils/federationWorker.ts index 674c55d5..32b9fc6a 100644 --- a/packages/server/src/utils/federationWorker.ts +++ b/packages/server/src/utils/federationWorker.ts @@ -15,6 +15,7 @@ import { startupBootstrapSync, onPeerDeactivated } from './federationPeerActivat import { probePeerReachable, markPeerRecovered } from './federationRecovery.js'; import { backfillReplicatedProfileAssets } from '../routes/federation.js'; import { invokePermanentFailureCallback } from './federationRollback.js'; +import { refreshPeerEpochs } from './federationEpoch.js'; import fs from 'node:fs'; import path from 'node:path'; import crypto from 'node:crypto'; @@ -1162,6 +1163,14 @@ async function processHealthCheckTick(): Promise { console.warn(`[federation-worker] Auto-rotation failed for peer ${peer.origin}: ${message}`); } } + + // ── Deterministic baseline epoch-refresh ──────────────────────────────────── + // Populate-if-null, self-terminating: fill peer_instance_id for active peers + // whose baseline is still NULL (design §3.2). Runs every tick so the baseline + // is established within one 15-minute cycle of an upgrade, independent of any + // relay/user activity. Best-effort — a failed fetch is a benign no-op retried + // next tick, so it never disturbs the rest of the health-check work. + await refreshPeerEpochs().catch(() => {}); } // ─── Federated Call Health Sweep ──────────────────────────────────────────── @@ -1235,6 +1244,12 @@ export function startFederationWorkers(): void { console.error('[federation-worker] federatedCallSentinel tick failed:', err) ); }, FEDERATED_CALL_SENTINEL_MS); + // Deterministic baseline epoch-refresh at startup (design §3.2): populate + // peer_instance_id for any active peer whose baseline is still NULL, so an + // instance that upgrades sees its peers' epochs within one cycle regardless of + // traffic. Best-effort, self-terminating (populate-if-null). + refreshPeerEpochs().catch(() => {}); + // Bootstrap sync for freshly-peered rows (async, non-blocking) startupBootstrapSync().catch((err) => { console.error('[federation-worker] Startup bootstrap sync error:', err);