From 5c1b42938e556d5c5ac78fb38e2065eddf6fc759 Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Wed, 22 Apr 2026 00:15:48 +0200 Subject: [PATCH] feat(federation): implement syncPeerMutationLog Three-pass pull-sync (dm, friend, profile) from peer's /api/federation/sync endpoint, paginated. Seeds sinceTimestamp from peer.lastSyncedAt so a recovered peer pulls only the delta. Updates lastSyncedAt to Date.now() on full success; leaves it untouched on transient failure so the next activation retries the same window. Replaces the body of the soon-to-be-removed runInitialSyncForNewPeers. --- .../utils/federationPeerActivation.test.ts | 106 ++++++++++++++++++ .../src/utils/federationPeerActivation.ts | 60 +++++++++- 2 files changed, 163 insertions(+), 3 deletions(-) diff --git a/packages/server/src/utils/federationPeerActivation.test.ts b/packages/server/src/utils/federationPeerActivation.test.ts index 04b77809..2406973f 100644 --- a/packages/server/src/utils/federationPeerActivation.test.ts +++ b/packages/server/src/utils/federationPeerActivation.test.ts @@ -18,6 +18,22 @@ vi.mock('../db/index.js', () => ({ schema, })); +vi.mock('../utils/federationOutbox.js', () => ({ + isFederationRelayEnabled: () => true, +})); + +vi.mock('../utils/federationAuth.js', () => ({ + getOurOrigin: () => 'https://local.example', + buildFederationHeaders: (_body: string, _secret: string, _origin: string) => ({ + 'Content-Type': 'application/json', + 'X-Federation-Origin': _origin, + }), +})); + +vi.mock('../routes/federation.js', () => ({ + processRelayEvents: vi.fn().mockResolvedValue({ accepted: [], rejected: [] }), +})); + function applyMigrations(db: Database.Database): void { const migrationsDir = path.resolve(__dirname, '../../drizzle'); const files = fs.readdirSync(migrationsDir).filter(f => f.endsWith('.sql')).sort(); @@ -93,3 +109,93 @@ describe('resetOutboxBackoff', () => { expect(() => resetOutboxBackoff('peer-empty')).not.toThrow(); }); }); + +describe('syncPeerMutationLog', () => { + beforeEach(() => { + sqlite = new Database(':memory:'); + testDb = drizzle(sqlite, { schema }); + applyMigrations(sqlite); + // Instance settings row is created by the baseline migration; defaults have + // federation_relay_enabled = 1, so no explicit update is needed. Confirm: + testDb.update(schema.instanceSettings) + .set({ federationRelayEnabled: 1 }) + .where(eq(schema.instanceSettings.id, 1)) + .run(); + vi.restoreAllMocks(); + }); + + it('seeds sinceTimestamp from peer.lastSyncedAt for each pass', async () => { + const { syncPeerMutationLog } = await import('./federationPeerActivation.js'); + testDb.insert(schema.federationPeers).values({ + id: 'peer-1', origin: 'https://peer-1.example', hmacSecret: 'secret', + status: 'active', lastSyncedAt: 5000, createdAt: Date.now(), + }).run(); + + const fetchSpy = vi.spyOn(globalThis, 'fetch').mockImplementation(async () => + new Response(JSON.stringify({ events: [], hasMore: false, checkpoint: 5000 }), { status: 200 }) + ); + + await syncPeerMutationLog('peer-1', 'health_check_recovery'); + + // Three passes: dm (no contextType), friend, profile + expect(fetchSpy).toHaveBeenCalledTimes(3); + for (const call of fetchSpy.mock.calls) { + const body = JSON.parse(call[1]?.body as string) as { sinceTimestamp: number }; + expect(body.sinceTimestamp).toBe(5000); + } + const calls = fetchSpy.mock.calls.map(c => JSON.parse(c[1]?.body as string) as { contextType?: string }); + expect(calls[0]!.contextType).toBeUndefined(); // DM pass (no contextType filter) + expect(calls[1]!.contextType).toBe('friend'); + expect(calls[2]!.contextType).toBe('profile'); + }); + + it('advances lastSyncedAt on success', async () => { + const { syncPeerMutationLog } = await import('./federationPeerActivation.js'); + testDb.insert(schema.federationPeers).values({ + id: 'peer-2', origin: 'https://peer-2.example', hmacSecret: 'secret', + status: 'active', lastSyncedAt: 0, createdAt: Date.now(), + }).run(); + + vi.spyOn(globalThis, 'fetch').mockImplementation(async () => + new Response(JSON.stringify({ events: [], hasMore: false, checkpoint: 1000 }), { status: 200 }) + ); + + const before = Date.now(); + await syncPeerMutationLog('peer-2', 'startup_bootstrap'); + const after = Date.now(); + + const row = testDb.select().from(schema.federationPeers) + .where(eq(schema.federationPeers.id, 'peer-2')).get(); + expect(row?.lastSyncedAt).toBeGreaterThanOrEqual(before); + expect(row?.lastSyncedAt).toBeLessThanOrEqual(after); + }); + + it('does NOT update lastSyncedAt on transient failure', async () => { + const { syncPeerMutationLog } = await import('./federationPeerActivation.js'); + testDb.insert(schema.federationPeers).values({ + id: 'peer-3', origin: 'https://peer-3.example', hmacSecret: 'secret', + status: 'active', lastSyncedAt: 42_000, createdAt: Date.now(), + }).run(); + + vi.spyOn(globalThis, 'fetch').mockImplementation(async () => + new Response('internal error', { status: 500 }) + ); + + await syncPeerMutationLog('peer-3', 'ensure_peered'); + const row = testDb.select().from(schema.federationPeers) + .where(eq(schema.federationPeers.id, 'peer-3')).get(); + expect(row?.lastSyncedAt).toBe(42_000); + }); + + it('does nothing when peer is not active', async () => { + const { syncPeerMutationLog } = await import('./federationPeerActivation.js'); + testDb.insert(schema.federationPeers).values({ + id: 'peer-4', origin: 'https://peer-4.example', hmacSecret: 'secret', + status: 'pending', lastSyncedAt: 0, createdAt: Date.now(), + }).run(); + + const fetchSpy = vi.spyOn(globalThis, 'fetch'); + await syncPeerMutationLog('peer-4', 'health_check_recovery'); + expect(fetchSpy).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/server/src/utils/federationPeerActivation.ts b/packages/server/src/utils/federationPeerActivation.ts index 4c2c2c8c..0d07ffdf 100644 --- a/packages/server/src/utils/federationPeerActivation.ts +++ b/packages/server/src/utils/federationPeerActivation.ts @@ -74,9 +74,63 @@ export async function syncPeerMutationLog( peerId: string, reason: PeerActivationReason, ): Promise { - // Stub — implemented in Task 3. - void peerId; - void reason; + if (!isFederationRelayEnabled()) return; + + const db = getDb(); + const peer = db.select().from(schema.federationPeers) + .where(eq(schema.federationPeers.id, peerId)).get(); + if (!peer || peer.status !== 'active') return; + + const ourOrigin = getOurOrigin(); + const signingSecret = (peer.pendingHmacSecret && peer.secretRotationAt) + ? peer.pendingHmacSecret + : peer.hmacSecret; + + console.log(`[federation] Sync-pull from ${peer.origin} (reason=${reason}, since=${peer.lastSyncedAt ?? 0})`); + + let totalEvents = 0; + + async function runPass(contextType?: 'friend' | 'profile'): Promise { + let since = peer!.lastSyncedAt ?? 0; + while (true) { + const bodyObj: Record = { sinceTimestamp: since, limit: 100 }; + if (contextType) bodyObj.contextType = contextType; + const body = JSON.stringify(bodyObj); + const headers = buildFederationHeaders(body, signingSecret, ourOrigin); + const resp = await fetch(`${peer!.origin}/api/federation/sync`, { + method: 'POST', headers, body, + signal: AbortSignal.timeout(30_000), + }); + if (!resp.ok) { + console.warn(`[federation] Sync-pull ${contextType ?? 'dm'} pass HTTP ${resp.status} for ${peer!.origin}`); + return false; + } + const data = await resp.json() as { events: FederationRelayEvent[]; hasMore: boolean; checkpoint: number }; + if (data.events.length === 0) return true; + const { processRelayEvents } = await import('../routes/federation.js'); + await processRelayEvents(data.events, peer!.origin, peer!.origin, db); + totalEvents += data.events.length; + since = data.checkpoint; + if (!data.hasMore) return true; + } + } + + try { + if (!(await runPass())) return; + if (!(await runPass('friend'))) return; + if (!(await runPass('profile'))) return; + + db.update(schema.federationPeers) + .set({ lastSyncedAt: Date.now() }) + .where(eq(schema.federationPeers.id, peer.id)) + .run(); + + if (totalEvents > 0) { + console.log(`[federation] Sync-pull from ${peer.origin} replayed ${totalEvents} events`); + } + } catch (err) { + console.error(`[federation] Sync-pull from ${peer.origin} failed:`, err); + } } /**