feat(federation): deterministic baseline epoch-refresh worker

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');
+  });
+});

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<string | null>
     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<void> {
+  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();
+  }
+}

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<void> {
       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);