diff --git a/packages/server/src/utils/federationPeerActivation.test.ts b/packages/server/src/utils/federationPeerActivation.test.ts index 2406973f..d3e154d1 100644 --- a/packages/server/src/utils/federationPeerActivation.test.ts +++ b/packages/server/src/utils/federationPeerActivation.test.ts @@ -115,12 +115,6 @@ describe('syncPeerMutationLog', () => { 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(); }); @@ -198,4 +192,48 @@ describe('syncPeerMutationLog', () => { await syncPeerMutationLog('peer-4', 'health_check_recovery'); expect(fetchSpy).not.toHaveBeenCalled(); }); + + it('advances sinceTimestamp within a pass using data.checkpoint when hasMore is true', async () => { + const { syncPeerMutationLog } = await import('./federationPeerActivation.js'); + testDb.insert(schema.federationPeers).values({ + id: 'peer-5', origin: 'https://peer-5.example', hmacSecret: 'secret', + status: 'active', lastSyncedAt: 100, createdAt: Date.now(), + }).run(); + + let call = 0; + const fetchSpy = vi.spyOn(globalThis, 'fetch').mockImplementation(async () => { + call++; + // First DM call: one event, hasMore=true, checkpoint advances to 2500 + // Second DM call: empty, hasMore=false, ends the DM pass + // Remaining calls (friend, profile): empty/done immediately + if (call === 1) { + return new Response( + JSON.stringify({ + events: [{ eventType: 'create', messageId: 'm1', timestamp: 200, encryptionVersion: 0 }], + hasMore: true, + checkpoint: 2500, + }), + { status: 200 }, + ); + } + return new Response(JSON.stringify({ events: [], hasMore: false, checkpoint: call === 2 ? 2500 : 100 }), { status: 200 }); + }); + + await syncPeerMutationLog('peer-5', 'health_check_recovery'); + + // Call 1: DM pass, since=100 (peer.lastSyncedAt) + // Call 2: DM pass continuation, since=2500 (advanced by previous checkpoint) + // Call 3: friend pass, since=100 (re-seeded from peer.lastSyncedAt) + // Call 4: profile pass, since=100 (re-seeded from peer.lastSyncedAt) + expect(fetchSpy).toHaveBeenCalledTimes(4); + const bodies = fetchSpy.mock.calls.map(c => JSON.parse(c[1]?.body as string) as { sinceTimestamp: number; contextType?: string }); + expect(bodies[0]?.sinceTimestamp).toBe(100); + expect(bodies[0]?.contextType).toBeUndefined(); + expect(bodies[1]?.sinceTimestamp).toBe(2500); // advanced by checkpoint from call 1 + expect(bodies[1]?.contextType).toBeUndefined(); + expect(bodies[2]?.sinceTimestamp).toBe(100); // friend pass re-seeds from peer.lastSyncedAt + expect(bodies[2]?.contextType).toBe('friend'); + expect(bodies[3]?.sinceTimestamp).toBe(100); // profile pass re-seeds from peer.lastSyncedAt + expect(bodies[3]?.contextType).toBe('profile'); + }); }); diff --git a/packages/server/src/utils/federationPeerActivation.ts b/packages/server/src/utils/federationPeerActivation.ts index 0d07ffdf..8a5da288 100644 --- a/packages/server/src/utils/federationPeerActivation.ts +++ b/packages/server/src/utils/federationPeerActivation.ts @@ -81,34 +81,42 @@ export async function syncPeerMutationLog( .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; + const activePeer = peer; // narrowed by the guard above - console.log(`[federation] Sync-pull from ${peer.origin} (reason=${reason}, since=${peer.lastSyncedAt ?? 0})`); + const ourOrigin = getOurOrigin(); + const signingSecret = (activePeer.pendingHmacSecret && activePeer.secretRotationAt) + ? activePeer.pendingHmacSecret + : activePeer.hmacSecret; + + console.log(`[federation] Sync-pull from ${activePeer.origin} (reason=${reason}, since=${activePeer.lastSyncedAt ?? 0})`); let totalEvents = 0; + type SyncRequestBody = { + sinceTimestamp: number; + limit: number; + contextType?: 'friend' | 'profile'; + }; + async function runPass(contextType?: 'friend' | 'profile'): Promise { - let since = peer!.lastSyncedAt ?? 0; + let since = activePeer.lastSyncedAt ?? 0; while (true) { - const bodyObj: Record = { sinceTimestamp: since, limit: 100 }; + const bodyObj: SyncRequestBody = { 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`, { + const resp = await fetch(`${activePeer.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}`); + console.warn(`[federation] Sync-pull ${contextType ?? 'dm'} pass HTTP ${resp.status} for ${activePeer.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); + await processRelayEvents(data.events, activePeer.origin, activePeer.origin, db); totalEvents += data.events.length; since = data.checkpoint; if (!data.hasMore) return true; @@ -122,14 +130,14 @@ export async function syncPeerMutationLog( db.update(schema.federationPeers) .set({ lastSyncedAt: Date.now() }) - .where(eq(schema.federationPeers.id, peer.id)) + .where(eq(schema.federationPeers.id, activePeer.id)) .run(); if (totalEvents > 0) { - console.log(`[federation] Sync-pull from ${peer.origin} replayed ${totalEvents} events`); + console.log(`[federation] Sync-pull from ${activePeer.origin} replayed ${totalEvents} events`); } } catch (err) { - console.error(`[federation] Sync-pull from ${peer.origin} failed:`, err); + console.error(`[federation] Sync-pull from ${activePeer.origin} failed:`, err); } }