devsyncwrld/backspace
1
0
Fork 0
Code Issues Pull Requests Actions 26 Packages Projects Releases Wiki Activity

fix(federation): address Task 3 review — pagination test + polish

Browse Source 
- Add pagination-advance test: verifies since=checkpoint on second
  iteration within a pass, and each pass re-seeds since from
  peer.lastSyncedAt (not carried from prior pass).
- Eliminate four peer! non-null assertions by capturing the narrowed
  value in activePeer after the guard.
- Tighten bodyObj type from Record<string, unknown> to a local
  SyncRequestBody type alias.
- Drop the no-op federationRelayEnabled UPDATE in test beforeEach
  (default is already 1 per baseline migration).
This commit is contained in:
Jannis Braun
2026-04-22 00:21:34 +02:00
parent 5c1b42938e
commit 39c43032e8
2 changed files with 65 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/utils/federationPeerActivation.test.ts
+44 -6
View File
@@ -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');
});
});
packages/server/src/utils/federationPeerActivation.ts
+21 -13
View File
@@ -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<boolean> {
let since = peer!.lastSyncedAt ?? 0;
let since = activePeer.lastSyncedAt ?? 0;
while (true) {
const bodyObj: Record<string, unknown> = { 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);
}
}