feat(federation-worker): treat receiver-ack 4xx reasons as terminal + invoke rollback

packages/server/src/utils/federationWorker.test.ts

@@ -64,6 +64,13 @@ vi.mock('../utils/federationAuthFailure.js', () => ({
   AUTH_FAILURE_THRESHOLD: 5,
 }));
 
+const invokeRollbackMock = vi.fn();
+vi.mock('./federationRollback.js', () => ({
+  invokePermanentFailureCallback: invokeRollbackMock,
+  registerPermanentFailureCallback: vi.fn(),
+  _resetCallbacks: vi.fn(),
+}));
+
 function applyMigrations(db: Database.Database): void {
   const migrationsDir = path.resolve(__dirname, '../../drizzle');
   const files = fs.readdirSync(migrationsDir).filter(f => f.endsWith('.sql')).sort();
@@ -84,10 +91,10 @@ function seedPeer(id: string): void {
   }).run();
 }
 
-function seedOutboxEntry(id: string, peerId: string, entityId: string): void {
+function seedOutboxEntry(id: string, peerId: string, entityId: string, eventType = 'create'): void {
   testDb.insert(schema.federationOutbox).values({
     id, peerId, contextId: 'ch-1', entityId,
-    contextType: 'dm', eventType: 'create', payload: JSON.stringify({
+    contextType: 'dm', eventType, payload: JSON.stringify({
       message: { userId: 'u', homeUserId: 'u', homeInstance: 'test.example', content: 'hi', replyToId: null, editedAt: null, createdAt: Date.now() },
     }),
     encryptionVersion: 0, attempts: 0, nextRetryAt: Date.now() - 1000,
@@ -288,3 +295,133 @@ describe('federatedCallSentinel', () => {
     );
   });
 });
+
+// ─── Terminal rejection reasons + rollback invocation ────────────────────────
+
+describe('outbox worker — terminal rejection reasons + rollback invocation', () => {
+  beforeEach(() => {
+    sqlite = new Database(':memory:');
+    testDb = drizzle(sqlite, { schema });
+    applyMigrations(sqlite);
+    vi.restoreAllMocks();
+    invokeRollbackMock.mockReset();
+  });
+
+  afterEach(() => {
+    sqlite.close();
+  });
+
+  it('recipient_not_found for friend_request_create is terminal AND invokes rollback', async () => {
+    seedPeer('peer-r1');
+    seedOutboxEntry('entry-r1', 'peer-r1', 'msg-1', 'friend_request_create');
+
+    vi.spyOn(globalThis, 'fetch').mockImplementation(async () =>
+      new Response(JSON.stringify({
+        accepted: [],
+        rejected: [{ messageId: 'msg-1', reason: 'recipient_not_found' }],
+      }), { status: 200, headers: { 'Content-Type': 'application/json' } }),
+    );
+
+    const workerModule = await import('./federationWorker.js');
+    const processOutboxTick = (workerModule as { processOutboxTick?: () => Promise<void> }).processOutboxTick!;
+    await processOutboxTick();
+
+    const remaining = testDb.select().from(schema.federationOutbox)
+      .where(eq(schema.federationOutbox.id, 'entry-r1')).get();
+    expect(remaining).toBeUndefined();
+
+    expect(invokeRollbackMock).toHaveBeenCalledTimes(1);
+    expect(invokeRollbackMock).toHaveBeenCalledWith('friend_request_create', 'msg-1', 'recipient_not_found');
+  });
+
+  it('recipient_not_found for dm_message_create: row deleted, rollback still invoked (no-op inside registry)', async () => {
+    seedPeer('peer-r2');
+    seedOutboxEntry('entry-r2', 'peer-r2', 'msg-2', 'dm_message_create');
+
+    vi.spyOn(globalThis, 'fetch').mockImplementation(async () =>
+      new Response(JSON.stringify({
+        accepted: [],
+        rejected: [{ messageId: 'msg-2', reason: 'recipient_not_found' }],
+      }), { status: 200, headers: { 'Content-Type': 'application/json' } }),
+    );
+
+    const workerModule = await import('./federationWorker.js');
+    const processOutboxTick = (workerModule as { processOutboxTick?: () => Promise<void> }).processOutboxTick!;
+    await processOutboxTick();
+
+    const remaining = testDb.select().from(schema.federationOutbox)
+      .where(eq(schema.federationOutbox.id, 'entry-r2')).get();
+    expect(remaining).toBeUndefined();
+
+    // Worker always calls invoke; whether a callback is registered is the registry's concern
+    expect(invokeRollbackMock).toHaveBeenCalledTimes(1);
+    expect(invokeRollbackMock).toHaveBeenCalledWith('dm_message_create', 'msg-2', 'recipient_not_found');
+  });
+
+  it('duplicate rejection is terminal but does NOT invoke rollback callback', async () => {
+    seedPeer('peer-r3');
+    seedOutboxEntry('entry-r3', 'peer-r3', 'msg-3', 'friend_request_create');
+
+    vi.spyOn(globalThis, 'fetch').mockImplementation(async () =>
+      new Response(JSON.stringify({
+        accepted: [],
+        rejected: [{ messageId: 'msg-3', reason: 'duplicate' }],
+      }), { status: 200, headers: { 'Content-Type': 'application/json' } }),
+    );
+
+    const workerModule = await import('./federationWorker.js');
+    const processOutboxTick = (workerModule as { processOutboxTick?: () => Promise<void> }).processOutboxTick!;
+    await processOutboxTick();
+
+    const remaining = testDb.select().from(schema.federationOutbox)
+      .where(eq(schema.federationOutbox.id, 'entry-r3')).get();
+    expect(remaining).toBeUndefined();
+
+    expect(invokeRollbackMock).not.toHaveBeenCalled();
+  });
+
+  it('attribution_mismatch is terminal AND invokes rollback', async () => {
+    seedPeer('peer-r4');
+    seedOutboxEntry('entry-r4', 'peer-r4', 'msg-4', 'friend_request_create');
+
+    vi.spyOn(globalThis, 'fetch').mockImplementation(async () =>
+      new Response(JSON.stringify({
+        accepted: [],
+        rejected: [{ messageId: 'msg-4', reason: 'attribution_mismatch' }],
+      }), { status: 200, headers: { 'Content-Type': 'application/json' } }),
+    );
+
+    const workerModule = await import('./federationWorker.js');
+    const processOutboxTick = (workerModule as { processOutboxTick?: () => Promise<void> }).processOutboxTick!;
+    await processOutboxTick();
+
+    const remaining = testDb.select().from(schema.federationOutbox)
+      .where(eq(schema.federationOutbox.id, 'entry-r4')).get();
+    expect(remaining).toBeUndefined();
+
+    expect(invokeRollbackMock).toHaveBeenCalledTimes(1);
+    expect(invokeRollbackMock).toHaveBeenCalledWith('friend_request_create', 'msg-4', 'attribution_mismatch');
+  });
+
+  it('non-terminal rejection (processing_error) does NOT invoke rollback and retains the outbox row', async () => {
+    seedPeer('peer-r5');
+    seedOutboxEntry('entry-r5', 'peer-r5', 'msg-5', 'friend_request_create');
+
+    vi.spyOn(globalThis, 'fetch').mockImplementation(async () =>
+      new Response(JSON.stringify({
+        accepted: [],
+        rejected: [{ messageId: 'msg-5', reason: 'processing_error' }],
+      }), { status: 200, headers: { 'Content-Type': 'application/json' } }),
+    );
+
+    const workerModule = await import('./federationWorker.js');
+    const processOutboxTick = (workerModule as { processOutboxTick?: () => Promise<void> }).processOutboxTick!;
+    await processOutboxTick();
+
+    const remaining = testDb.select().from(schema.federationOutbox)
+      .where(eq(schema.federationOutbox.id, 'entry-r5')).get();
+    expect(remaining).toBeDefined();
+
+    expect(invokeRollbackMock).not.toHaveBeenCalled();
+  });
+});

packages/server/src/utils/federationWorker.ts

@@ -12,6 +12,7 @@ import { connectionManager } from '../ws/handler.js';
 import { generateThumbnail } from './thumbnail.js';
 import type { FederationRelayRequest, FederationRelayResponse, FederationRelayEvent } from '@backspace/shared';
 import { onPeerActivated, startupBootstrapSync, onPeerDeactivated } from './federationPeerActivation.js';
+import { invokePermanentFailureCallback } from './federationRollback.js';
 import fs from 'node:fs';
 import path from 'node:path';
 import crypto from 'node:crypto';
@@ -49,6 +50,24 @@ const BACKOFF_SCHEDULE_MS: readonly number[] = [
 const MAX_FILE_ATTEMPTS = 10;
 const PEER_UNREACHABLE_THRESHOLD = 10;
 
+/**
+ * Outbox rejection reasons that the receiver has acknowledged as permanently
+ * undeliverable. These cause the outbox entry to be deleted (no retry) and
+ * trigger the registered permanent-failure callback for the eventType.
+ *
+ * 'duplicate' is treated as terminal-but-no-rollback (the receiver already has
+ * the event; nothing to roll back locally).
+ *
+ * 5xx responses, network errors, and timeouts are NOT in this set — they are
+ * transient and retried via the existing backoff schedule.
+ */
+const TERMINAL_REJECTION_REASONS = new Set<string>([
+  'duplicate',            // peer already has it (existing behavior)
+  'recipient_not_found',  // receiver doesn't know the target user
+  'attribution_mismatch', // payload claims a homeInstance the source can't authoritatively speak for
+  'unknown_event_type',   // peer doesn't understand this eventType — never will
+]);
+
 // ─── Worker State ───────────────────────────────────────────────────────────
 
 let outboxTimer: ReturnType<typeof setTimeout> | null = null;
@@ -230,15 +249,25 @@ export async function processOutboxTick(): Promise<void> {
       if (response.ok) {
         const result = await response.json() as FederationRelayResponse;
 
-        // Terminal outcomes = accepted + duplicate-rejected.
+        // Terminal rejection reasons: receiver acknowledged the event is permanently
-        // `duplicate` means the peer already has the message (e.g., delivered
+        // undeliverable. Retrying will fail forever — remove from outbox.
-        // earlier via outbox or pulled via sync). Retrying will fail with
+        // Non-`duplicate` terminals additionally invoke any registered permanent-
-        // `duplicate` forever until TTL expires — treat it as effectively-
+        // failure callback for the eventType so the originator can roll back local
-        // accepted and remove the outbox entry.
+        // state (e.g., friend_request_create deletes the local friend_requests row).
         const terminalEntityIds = new Set<string>(result.accepted);
+        const terminalForRollback: Array<{ messageId: string; reason: string; eventType: string | null }> = [];
+
         for (const rejection of result.rejected) {
-          if (rejection.reason === 'duplicate') {
+          if (TERMINAL_REJECTION_REASONS.has(rejection.reason)) {
             terminalEntityIds.add(rejection.messageId);
+            if (rejection.reason !== 'duplicate') {
+              const entry = peerEntries.find(e => e.entityId === rejection.messageId);
+              terminalForRollback.push({
+                messageId: rejection.messageId,
+                reason: rejection.reason,
+                eventType: entry?.eventType ?? null,
+              });
+            }
           }
         }
 
@@ -254,13 +283,22 @@ export async function processOutboxTick(): Promise<void> {
           }
         }
 
-        // Log rejected entries. Duplicate is terminal (outbox entry already
+        // Invoke registered rollback callbacks AFTER deleting the outbox row,
-        // removed above) — log at info level. Other reasons are transient /
+        // so the rollback runs in a clean state. The registry catches and logs
-        // retained for retry — log at warn level.
+        // callback errors — they cannot prevent outbox cleanup.
+        for (const { messageId, reason, eventType } of terminalForRollback) {
+          if (eventType) {
+            invokePermanentFailureCallback(eventType, messageId, reason);
+          }
+        }
+
+        // Log rejected entries. Terminal reasons (incl. 'duplicate') are logged
+        // at info level — outbox entry already removed. Non-terminals stay in
+        // outbox for retry and log at warn level.
         for (const rejection of result.rejected) {
-          if (rejection.reason === 'duplicate') {
+          if (TERMINAL_REJECTION_REASONS.has(rejection.reason)) {
             console.log(
-              `[federation-worker] Peer ${peerOrigin} rejected message ${rejection.messageId} as duplicate — outbox entry removed (terminal)`,
+              `[federation-worker] Peer ${peerOrigin} terminal rejection ${rejection.messageId}: ${rejection.reason} — outbox entry removed`,
             );
           } else {
             console.warn(