From 3b1a0b64a3c7d40b48d4952629d3a5f90f1c0261 Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Wed, 1 Jul 2026 21:58:03 +0200 Subject: [PATCH] feat(federation): relay envelope populates peer epoch baseline --- docs/systems/federation.md | 2 + packages/server/src/routes/federation.ts | 19 +++++ .../server/src/utils/federationEpoch.test.ts | 81 +++++++++++++++++++ .../server/src/utils/federationWorker.test.ts | 18 +++++ packages/server/src/utils/federationWorker.ts | 7 +- 5 files changed, 126 insertions(+), 1 deletion(-) diff --git a/docs/systems/federation.md b/docs/systems/federation.md index 47751731..655f22f2 100644 --- a/docs/systems/federation.md +++ b/docs/systems/federation.md @@ -406,6 +406,8 @@ Two layers of replay protection: **Important:** The body is re-serialized server-side. This means Fastify's JSON parsing and re-stringification must produce identical output to the sender's `JSON.stringify`. In practice this works because both sides use standard `JSON.stringify` with no custom replacers. +**Relay-envelope epoch (fast-path baseline population, design §3.2).** `FederationRelayRequest` carries `sourceInstanceId?: string` — the sender stamps its current epoch (`getInstanceId()`) when building the request in `federationWorker.ts`. Because the whole body is HMAC-verified above (step 4), a valid relay authentically carries the sender's current incarnation id. Immediately after the signature check passes (and only there — the authenticated boundary), the receiver runs **populate-if-null**: `if (sourceInstanceId && peer.peerInstanceId IS NULL) UPDATE federation_peers SET peer_instance_id = WHERE id = ? AND peer_instance_id IS NULL`. This is the *fast-path* baseline populator — it fills the trusted epoch the instant organic traffic flows, usually before the deterministic 15-minute `refreshPeerEpochs` backstop fires. It **never overwrites** a non-null baseline: a differing incarnation implies a different HMAC secret that would have failed verification, so a valid relay can never carry an epoch differing from an established baseline. Runs independent of per-event processing and does not affect relay accept/reject. Backward-compatible: older peers omit `sourceInstanceId` → the update is skipped (no-op). + --- ## 3. Identity Resolution diff --git a/packages/server/src/routes/federation.ts b/packages/server/src/routes/federation.ts index 7e7c6661..40fc2152 100644 --- a/packages/server/src/routes/federation.ts +++ b/packages/server/src/routes/federation.ts @@ -2259,6 +2259,25 @@ export async function federationRoutes(app: FastifyInstance): Promise { return reply.code(401).send({ error: 'Invalid signature', statusCode: 401 }); } + // 1b-epoch. Fast-path baseline population (design §3.2). The signature just + // verified proves the peer holds the current shared secret, so the epoch it + // carries in `sourceInstanceId` is authentic. Populate-if-null ONLY: a valid + // relay can never carry an epoch differing from a non-null baseline (a + // different incarnation implies a different secret that fails HMAC), so we + // only ever fill a NULL — never overwrite. This is independent of per-event + // processing and does not affect relay accept/reject in any way. Old peers + // omit the field → skip (backward-compatible no-op). + const claimedEpoch = request.body.sourceInstanceId; + if (claimedEpoch && !peer.peerInstanceId) { + db.update(schema.federationPeers) + .set({ peerInstanceId: claimedEpoch }) + .where(and( + eq(schema.federationPeers.id, peer.id), + isNull(schema.federationPeers.peerInstanceId), + )) + .run(); + } + // 1c. Nonce-based replay protection if (fedHeaders.nonce) { if (isNonceDuplicate(peer.origin, fedHeaders.nonce)) { diff --git a/packages/server/src/utils/federationEpoch.test.ts b/packages/server/src/utils/federationEpoch.test.ts index 1813eca7..ccdea5cb 100644 --- a/packages/server/src/utils/federationEpoch.test.ts +++ b/packages/server/src/utils/federationEpoch.test.ts @@ -411,3 +411,84 @@ describe('refreshPeerEpochs — deterministic populate-if-null baseline (self-te expect(readPeerInstanceId()).toBe('pre-existing'); }); }); + +describe('POST /api/federation/relay — fast-path epoch baseline (populate-if-null)', () => { + // A verified inbound relay authentically carries the sender's current epoch in + // `sourceInstanceId` (design §3.2). On the authenticated path only, the receiver + // fills a NULL `peer_instance_id` — never overwrites a non-null baseline. + let app: FastifyInstance; + + beforeEach(async () => { + seedInstanceSettings(LOCAL_EPOCH); + seedActivePeer(); + app = await buildApp(); + }); + + afterEach(async () => { + await app.close(); + vi.restoreAllMocks(); + }); + + 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; + } + + /** Send a validly-signed relay (empty event batch) carrying `sourceInstanceId`. */ + async function injectSignedRelay(sourceInstanceId?: string): Promise { + const relay: Record = { + version: 1, + sourceInstance: PEER_ORIGIN, + events: [], + }; + if (sourceInstanceId !== undefined) relay.sourceInstanceId = sourceInstanceId; + const body = JSON.stringify(relay); + const headers = buildFederationHeaders(body, PEER_SECRET, PEER_ORIGIN); + const response = await app.inject({ + method: 'POST', + url: '/api/federation/relay', + headers, + payload: body, + }); + return response.statusCode; + } + + it('populates a NULL baseline from the epoch a verified relay carries', async () => { + expect(readPeerInstanceId()).toBeNull(); + const status = await injectSignedRelay('remote-epoch-A'); + expect(status).toBe(200); + expect(readPeerInstanceId()).toBe('remote-epoch-A'); + }); + + it('never overwrites a non-null baseline (a valid relay cannot carry a differing epoch)', async () => { + const first = await injectSignedRelay('remote-epoch-A'); + expect(first).toBe(200); + expect(readPeerInstanceId()).toBe('remote-epoch-A'); + + // A subsequent relay claiming a different epoch must leave the baseline intact. + const second = await injectSignedRelay('remote-epoch-B'); + expect(second).toBe(200); + expect(readPeerInstanceId()).toBe('remote-epoch-A'); + }); + + it('is a no-op when a pre-existing baseline is already set', async () => { + testDb.update(schema.federationPeers) + .set({ peerInstanceId: 'pre-existing' }) + .where(eq(schema.federationPeers.id, 'peer-remote')) + .run(); + + const status = await injectSignedRelay('remote-epoch-A'); + expect(status).toBe(200); + expect(readPeerInstanceId()).toBe('pre-existing'); + }); + + it('is a no-op for a backward-compatible relay that omits sourceInstanceId', async () => { + const status = await injectSignedRelay(undefined); + expect(status).toBe(200); + expect(readPeerInstanceId()).toBeNull(); + }); +}); diff --git a/packages/server/src/utils/federationWorker.test.ts b/packages/server/src/utils/federationWorker.test.ts index d0f4e3a8..fb871930 100644 --- a/packages/server/src/utils/federationWorker.test.ts +++ b/packages/server/src/utils/federationWorker.test.ts @@ -6,6 +6,7 @@ import path from 'node:path'; import { fileURLToPath } from 'node:url'; import * as schema from '../db/schema.js'; import { eq } from 'drizzle-orm'; +import { __resetInstanceIdCacheForTest } from './federationEpoch.js'; const __dirname = path.dirname(fileURLToPath(import.meta.url)); type TestDb = ReturnType>; @@ -84,6 +85,20 @@ function applyMigrations(db: Database.Database): void { } } +/** + * Seed this instance's epoch so the outbox relay builder can stamp + * `sourceInstanceId` via getInstanceId(). Resets the module cache so the fresh + * per-test DB row is read rather than a value cached from a prior test. + */ +function seedInstanceEpoch(): void { + testDb.insert(schema.instanceSettings).values({ + id: 1, + instanceId: 'worker-test-epoch', + updatedAt: Date.now(), + } as typeof schema.instanceSettings.$inferInsert).run(); + __resetInstanceIdCacheForTest(); +} + function seedPeer(id: string): void { testDb.insert(schema.federationPeers).values({ id, origin: 'https://peer.example', hmacSecret: 'secret', @@ -108,6 +123,7 @@ describe('outbox worker — duplicate rejection is terminal', () => { sqlite = new Database(':memory:'); testDb = drizzle(sqlite, { schema }); applyMigrations(sqlite); + seedInstanceEpoch(); vi.restoreAllMocks(); // Re-apply the static mocks that vi.restoreAllMocks() would undo. // isFederationRelayEnabled is mocked at module level via vi.mock (hoisted), @@ -303,6 +319,7 @@ describe('outbox worker — terminal rejection reasons + rollback invocation', ( sqlite = new Database(':memory:'); testDb = drizzle(sqlite, { schema }); applyMigrations(sqlite); + seedInstanceEpoch(); vi.restoreAllMocks(); invokeRollbackMock.mockReset(); }); @@ -431,6 +448,7 @@ describe('unreachable transition resets probe pacing', () => { sqlite = new Database(':memory:'); testDb = drizzle(sqlite, { schema }); applyMigrations(sqlite); + seedInstanceEpoch(); vi.restoreAllMocks(); }); diff --git a/packages/server/src/utils/federationWorker.ts b/packages/server/src/utils/federationWorker.ts index 32b9fc6a..7d6ffc8d 100644 --- a/packages/server/src/utils/federationWorker.ts +++ b/packages/server/src/utils/federationWorker.ts @@ -15,7 +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 { refreshPeerEpochs, getInstanceId } from './federationEpoch.js'; import fs from 'node:fs'; import path from 'node:path'; import crypto from 'node:crypto'; @@ -235,6 +235,11 @@ export async function processOutboxTick(): Promise { const request: FederationRelayRequest = { version: 1, sourceInstance: ourOrigin, + // Stamp our current epoch so a verified relay authentically carries this + // instance's incarnation id — the receiver uses it as the fast-path + // populate-if-null baseline (design §3.2). A reset instance cannot sign a + // valid relay, so this never carries a *new* epoch post-reset. + sourceInstanceId: getInstanceId(), events, };