From 15e42a7cc1ee0e731b89a3862f0cb197a8d61082 Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Wed, 22 Apr 2026 01:45:14 +0200 Subject: [PATCH] fix(federation): per-event fault isolation in syncPeerMutationLog (#25) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the batch-level processRelayEvents call with a per-event loop wrapped in try/catch. On exception: log event type, messageId, timestamp, peer origin, and the error message; continue to the next event. Previously, a single poison-pill event (e.g., UNIQUE conflict from a malformed relay payload) would throw, be caught by the outer try/catch, and block lastSyncedAt from advancing — causing every future activation to retry the same broken window indefinitely. The final 'replayed N events' log line now reports '(K skipped due to errors)' when K > 0, surfacing the count to operators. Individual event failures are logged via console.error with enough context to debug or replay manually. Trade-off documented in docs/systems/federation.md: forward progress of the sync pipeline takes priority over strict at-least-once delivery. An event that fails to process is lost to the receiver unless replayed manually. --- docs/systems/federation.md | 8 ++- .../utils/federationPeerActivation.test.ts | 54 +++++++++++++++++++ .../src/utils/federationPeerActivation.ts | 22 ++++++-- 3 files changed, 78 insertions(+), 6 deletions(-) diff --git a/docs/systems/federation.md b/docs/systems/federation.md index 45fe5bab..1a3521b6 100644 --- a/docs/systems/federation.md +++ b/docs/systems/federation.md @@ -1047,9 +1047,13 @@ Ephemeral events (`dm_typing_*`, `dm_call_*`) are fire-and-forget by design and | `'friend'` | Friend events | | `'profile'` | Profile update events | -#### Known Issues +#### Poison-Pill Event Handling -- **Poison-pill event in peer's mutation log.** If a peer's mutation log contains a row whose inbound processor throws (e.g., a UNIQUE conflict from a malformed relay payload), `syncPeerMutationLog` catches the error and declines to advance `lastSyncedAt`. Subsequent activations retry the same window and hit the same failure, effectively blocking catch-up for that peer. No automatic poison-pill skip is implemented — recovery requires either: (a) fixing the mutation log on the peer side, or (b) manually advancing `lastSyncedAt` past the offending row via DB admin. Flagged as a follow-up backlog item. +`syncPeerMutationLog` replays incoming events one at a time. If an event's inbound processor throws (e.g., UNIQUE conflict from a malformed payload, unexpected schema drift, a processor bug), the error is caught per-event, logged via `console.error` with the event's `eventType`, `messageId`, `timestamp`, peer origin, and error message, and the loop continues with the next event. `lastSyncedAt` is advanced past the failed event (using the event's own `timestamp`), so subsequent activations do not retry the poison pill. + +The final `Sync-pull from replayed events` log line is suffixed with `( skipped due to errors)` when `K > 0`, surfacing the count to operators watching logs. Individual event failures are in the same logs under `Skipping poison-pill event` — grep `console.error` / `stderr` to recover them. + +Trade-off: this policy prioritizes forward progress of the sync pipeline over strict at-least-once delivery of every mutation. An event that fails to process is silently lost to the receiving instance unless operators manually replay it (e.g., by resetting `lastSyncedAt` on the peer row or by reissuing the originating mutation on the sender). The alternative — refusing to advance on any error — caused the "stuck forever" state described in the pre-fix version of this section. ### Sync Endpoint (`POST /api/federation/sync`) diff --git a/packages/server/src/utils/federationPeerActivation.test.ts b/packages/server/src/utils/federationPeerActivation.test.ts index d1dbc5da..c684255c 100644 --- a/packages/server/src/utils/federationPeerActivation.test.ts +++ b/packages/server/src/utils/federationPeerActivation.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 * as federationRouteMock from '../routes/federation.js'; const __dirname = path.dirname(fileURLToPath(import.meta.url)); @@ -245,6 +246,59 @@ describe('syncPeerMutationLog', () => { expect(bodies[3]?.sinceTimestamp).toBe(100); // profile pass re-seeds from peer.lastSyncedAt expect(bodies[3]?.contextType).toBe('profile'); }); + + it('skips a poison-pill event, logs it, and advances past it to process subsequent events', async () => { + const { syncPeerMutationLog } = await import('./federationPeerActivation.js'); + testDb.insert(schema.federationPeers).values({ + id: 'peer-poison', origin: 'https://peer-poison.example', hmacSecret: 'secret', + status: 'active', lastSyncedAt: 100, createdAt: Date.now(), + }).run(); + + // Return a single batch of 3 events on the DM pass, then empty on friend + profile passes. + let fetchCall = 0; + vi.spyOn(globalThis, 'fetch').mockImplementation(async () => { + fetchCall++; + if (fetchCall === 1) { + return new Response(JSON.stringify({ + events: [ + { eventType: 'create', messageId: 'good-1', timestamp: 200, encryptionVersion: 0 }, + { eventType: 'create', messageId: 'poison', timestamp: 300, encryptionVersion: 0 }, + { eventType: 'create', messageId: 'good-2', timestamp: 400, encryptionVersion: 0 }, + ], + hasMore: false, + checkpoint: 400, + }), { status: 200 }); + } + return new Response(JSON.stringify({ events: [], hasMore: false, checkpoint: 100 }), { status: 200 }); + }); + + // Grab the top-level mock and override implementation per-call: + // good-1: resolves, poison: throws, good-2: resolves. + const processMock = vi.mocked(federationRouteMock.processRelayEvents); + processMock.mockClear(); + processMock.mockResolvedValueOnce(undefined as never); // good-1 + processMock.mockRejectedValueOnce(new Error('simulated processor failure')); // poison + processMock.mockResolvedValueOnce(undefined as never); // good-2 + + const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}); + + const before = Date.now(); + await syncPeerMutationLog('peer-poison', 'health_check_recovery'); + + // Verify processRelayEvents was called per-event: 3 calls for the 3 DM events. + expect(processMock).toHaveBeenCalledTimes(3); + + // Verify error logged for the poison event. + expect(errorSpy).toHaveBeenCalled(); + const errorMessages = errorSpy.mock.calls.map(c => String(c[0] ?? '')); + expect(errorMessages.some(m => m.includes('poison'))).toBe(true); + expect(errorMessages.some(m => m.includes('simulated processor failure'))).toBe(true); + + // Verify lastSyncedAt advanced despite the poison event (the critical property). + const row = testDb.select().from(schema.federationPeers) + .where(eq(schema.federationPeers.id, 'peer-poison')).get(); + expect(row?.lastSyncedAt).toBeGreaterThanOrEqual(before); + }); }); describe('onPeerActivated', () => { diff --git a/packages/server/src/utils/federationPeerActivation.ts b/packages/server/src/utils/federationPeerActivation.ts index ee34c265..a30f22d9 100644 --- a/packages/server/src/utils/federationPeerActivation.ts +++ b/packages/server/src/utils/federationPeerActivation.ts @@ -108,6 +108,7 @@ export async function syncPeerMutationLog( console.log(`[federation] Sync-pull from ${activePeer.origin} (reason=${reason}, since=${activePeer.lastSyncedAt ?? 0})`); let totalEvents = 0; + let skippedEvents = 0; type SyncRequestBody = { sinceTimestamp: number; @@ -133,8 +134,20 @@ export async function syncPeerMutationLog( 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, activePeer.origin, activePeer.origin, db); - totalEvents += data.events.length; + for (const event of data.events) { + try { + await processRelayEvents([event], activePeer.origin, activePeer.origin, db); + totalEvents += 1; + } catch (err) { + skippedEvents += 1; + const errMsg = err instanceof Error ? err.message : String(err); + console.error( + `[federation] Skipping poison-pill event during sync-pull from ${activePeer.origin}: ` + + `eventType=${event.eventType} messageId=${event.messageId} timestamp=${event.timestamp} ` + + `error=${errMsg}`, + ); + } + } since = data.checkpoint; if (!data.hasMore) return true; } @@ -150,8 +163,9 @@ export async function syncPeerMutationLog( .where(eq(schema.federationPeers.id, activePeer.id)) .run(); - if (totalEvents > 0) { - console.log(`[federation] Sync-pull from ${activePeer.origin} replayed ${totalEvents} events`); + if (totalEvents > 0 || skippedEvents > 0) { + const skipSuffix = skippedEvents > 0 ? ` (${skippedEvents} skipped due to errors)` : ''; + console.log(`[federation] Sync-pull from ${activePeer.origin} replayed ${totalEvents} events${skipSuffix}`); } } catch (err) { console.error(`[federation] Sync-pull from ${activePeer.origin} failed:`, err);