diff --git a/packages/server/src/utils/federationWorker.test.ts b/packages/server/src/utils/federationWorker.test.ts index 11d58850..94e01777 100644 --- a/packages/server/src/utils/federationWorker.test.ts +++ b/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 }).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 }).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 }).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 }).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 }).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(); + }); +}); diff --git a/packages/server/src/utils/federationWorker.ts b/packages/server/src/utils/federationWorker.ts index 3ddbf3bf..e2c89507 100644 --- a/packages/server/src/utils/federationWorker.ts +++ b/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([ + '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 | null = null; @@ -230,15 +249,25 @@ export async function processOutboxTick(): Promise { if (response.ok) { const result = await response.json() as FederationRelayResponse; - // Terminal outcomes = accepted + duplicate-rejected. - // `duplicate` means the peer already has the message (e.g., delivered - // earlier via outbox or pulled via sync). Retrying will fail with - // `duplicate` forever until TTL expires — treat it as effectively- - // accepted and remove the outbox entry. + // Terminal rejection reasons: receiver acknowledged the event is permanently + // undeliverable. Retrying will fail forever — remove from outbox. + // Non-`duplicate` terminals additionally invoke any registered permanent- + // failure callback for the eventType so the originator can roll back local + // state (e.g., friend_request_create deletes the local friend_requests row). const terminalEntityIds = new Set(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 { } } - // Log rejected entries. Duplicate is terminal (outbox entry already - // removed above) — log at info level. Other reasons are transient / - // retained for retry — log at warn level. + // Invoke registered rollback callbacks AFTER deleting the outbox row, + // so the rollback runs in a clean state. The registry catches and logs + // 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(