fix(federation): per-event fault isolation in syncPeerMutationLog (#25)

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

@@ -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 <origin> replayed <N> events` log line is suffixed with `(<K> 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`)
 

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', () => {

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);