feat(federation): relay envelope populates peer epoch baseline

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<number> {
+    const relay: Record<string, unknown> = {
+      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();
+  });
+});

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<typeof drizzle<typeof schema>>;
@@ -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();
   });
 

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<void> {
     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,
     };