feat(federation): implement onPeerActivated with dedup

Two-invariant handler: resetOutboxBackoff + syncPeerMutationLog.
In-flight map keyed by peerId coalesces concurrent activations —
a second call for a peer whose activation is still running shares
the same promise. Errors are swallowed and logged — the handler
never throws so fire-and-forget callers at HTTP handler sites
are safe.
This commit is contained in:
Jannis Braun
2026-04-22 00:24:57 +02:00
parent 39c43032e8
commit cca2245cdf
2 changed files with 77 additions and 3 deletions
@@ -34,6 +34,15 @@ vi.mock('../routes/federation.js', () => ({
processRelayEvents: vi.fn().mockResolvedValue({ accepted: [], rejected: [] }), processRelayEvents: vi.fn().mockResolvedValue({ accepted: [], rejected: [] }),
})); }));
vi.mock('../ws/handler.js', () => ({
connectionManager: {
sendToAdmins: vi.fn(),
getAllOnlineUserIds: () => [],
sendToUser: vi.fn(),
sendToDmMembers: vi.fn(),
},
}));
function applyMigrations(db: Database.Database): void { function applyMigrations(db: Database.Database): void {
const migrationsDir = path.resolve(__dirname, '../../drizzle'); const migrationsDir = path.resolve(__dirname, '../../drizzle');
const files = fs.readdirSync(migrationsDir).filter(f => f.endsWith('.sql')).sort(); const files = fs.readdirSync(migrationsDir).filter(f => f.endsWith('.sql')).sort();
@@ -237,3 +246,51 @@ describe('syncPeerMutationLog', () => {
expect(bodies[3]?.contextType).toBe('profile'); expect(bodies[3]?.contextType).toBe('profile');
}); });
}); });
describe('onPeerActivated', () => {
beforeEach(() => {
sqlite = new Database(':memory:');
testDb = drizzle(sqlite, { schema });
applyMigrations(sqlite);
vi.restoreAllMocks();
});
it('runs resetOutboxBackoff and syncPeerMutationLog once, even under concurrent calls', async () => {
const { onPeerActivated } = await import('./federationPeerActivation.js');
testDb.insert(schema.federationPeers).values({
id: 'peer-x', origin: 'https://peer-x.example', hmacSecret: 'secret',
status: 'active', lastSyncedAt: 0, createdAt: Date.now(),
}).run();
let fetchCount = 0;
vi.spyOn(globalThis, 'fetch').mockImplementation(async () => {
fetchCount++;
// Deliberately slow to let the second concurrent call share the in-flight promise.
await new Promise(r => setTimeout(r, 20));
return new Response(JSON.stringify({ events: [], hasMore: false, checkpoint: 0 }), { status: 200 });
});
const p1 = onPeerActivated('peer-x', 'health_check_recovery');
const p2 = onPeerActivated('peer-x', 'accept_new');
await Promise.all([p1, p2]);
// Three fetch calls for the three sync passes (dm, friend, profile) — not six.
expect(fetchCount).toBe(3);
});
it('swallows errors from syncPeerMutationLog so the handler does not throw', async () => {
const { onPeerActivated } = await import('./federationPeerActivation.js');
testDb.insert(schema.federationPeers).values({
id: 'peer-err', origin: 'https://peer-err.example', hmacSecret: 'secret',
status: 'active', lastSyncedAt: 0, createdAt: Date.now(),
}).run();
vi.spyOn(globalThis, 'fetch').mockImplementation(async () => {
throw new Error('network down');
});
await expect(onPeerActivated('peer-err', 'ensure_peered')).resolves.toBeUndefined();
});
});
@@ -42,9 +42,26 @@ export async function onPeerActivated(
peerId: string, peerId: string,
reason: PeerActivationReason, reason: PeerActivationReason,
): Promise<void> { ): Promise<void> {
// Stub — implemented in Task 4. const existing = inFlightActivation.get(peerId);
void peerId; if (existing) return existing;
void reason;
const promise = (async () => {
try {
resetOutboxBackoff(peerId);
await syncPeerMutationLog(peerId, reason);
const { connectionManager } = await import('../ws/handler.js');
connectionManager.sendToAdmins({ type: 'federation_peers_changed' as const });
} catch (err) {
console.error(`[federation] onPeerActivated(${peerId}, ${reason}) failed:`, err);
}
})();
inFlightActivation.set(peerId, promise);
try {
await promise;
} finally {
inFlightActivation.delete(peerId);
}
} }
/** /**