fix(federation): treat duplicate rejection as terminal in outbox worker

Duplicate rejection means the peer already has the message (e.g., delivered earlier via outbox AND pulled via sync in the same window). Retrying will fail identically forever until TTL expires. Before this patch: duplicate-rejected outbox entries were retained with attempts++ and exponential backoff, creating log noise and outbox bloat for up to 30 days. After: duplicate-rejected entityIds join the terminal set alongside accepted ones and are deleted from the outbox. Logged at info level ('outbox entry removed (terminal)') to distinguish from warn-level transient-rejection retries. Other rejection reasons (attribution_mismatch, processing_error, etc.) stay on the retry path; some may also be terminal but are deferred until observed accumulating.
2026-04-23 00:10:34 +02:00
parent 3252868152
commit 6ff983b46c
3 changed files with 200 additions and 15 deletions
@@ -466,8 +466,9 @@ Trigger (API/WS handler)
 5. Sign with `buildFederationHeaders(body, peerHmacSecret, ourOrigin)`
 6. POST to `{peerOrigin}/api/federation/relay`
 7. On success (200):
-   - Delete accepted entries from outbox (matched by `entityId` -> `outboxId`)
-   - Log rejected entries (remain in outbox for retry)
+   - Compute the **terminal entity set** = accepted entries ∪ duplicate-rejected entries
+   - Delete all terminal entries from outbox (matched by `entityId` -> `outboxId`)
+   - Log remaining (non-duplicate) rejected entries at `console.warn` (they stay in outbox for retry)
   - Store `result.maxUploadSize` on peer record
   - Update peer: `lastSeenAt = now`, `consecutiveFailures = 0`
 8. On failure (non-200 or network error):
@@ -476,6 +477,14 @@ Trigger (API/WS handler)
     - Increment peer `consecutiveFailures`, set `lastFailureAt`
     - If `consecutiveFailures >= PEER_UNREACHABLE_THRESHOLD (10)` -> mark peer `unreachable`

+#### Terminal rejection: `duplicate`
+
+The outbox delivery worker treats a relay response of `{ rejected: [{ reason: 'duplicate', ... }] }` as effectively-accepted — the outbox entry is deleted rather than retained for retry. The `duplicate` reason is emitted by the receiving instance's inbound processors when a row with the same `(sourceInstance, sourceMessageId)` already exists; retrying will fail identically until TTL (30 days). Since the peer already has the message, terminal removal is the correct outcome.
+
+Logged at `console.log` ("outbox entry removed (terminal)") to distinguish from retained-for-retry `console.warn` messages.
+
+Other rejection reasons (`attribution_mismatch`, `missing_*_payload`, `unknown_event_type`, `unauthorized_source`, `channel_not_found`, `participant_not_found`, `processing_error`, …) remain on the retry path. Some are arguably terminal too; treating them as such is deferred until they are observed accumulating in practice.
+
 ### Retry Backoff Schedule

 | Attempt | Delay |
@@ -0,0 +1,159 @@
+import { describe, it, expect, beforeEach, vi } from 'vitest';
+import Database from 'better-sqlite3';
+import { drizzle } from 'drizzle-orm/better-sqlite3';
+import fs from 'node:fs';
+import path from 'node:path';
+import { fileURLToPath } from 'node:url';
+import * as schema from '../db/schema.js';
+import { eq } from 'drizzle-orm';
+
+const __dirname = path.dirname(fileURLToPath(import.meta.url));
+type TestDb = ReturnType<typeof drizzle<typeof schema>>;
+let sqlite: Database.Database;
+let testDb: TestDb;
+
+vi.mock('../db/index.js', () => ({
+  getDb: () => testDb,
+  schema,
+}));
+
+vi.mock('../ws/handler.js', () => ({
+  connectionManager: {
+    sendToAdmins: vi.fn(),
+    getAllOnlineUserIds: () => [],
+    sendToUser: vi.fn(),
+    sendToDmMembers: vi.fn(),
+  },
+}));
+
+vi.mock('../utils/federationAuth.js', () => ({
+  getOurOrigin: () => 'https://test.example',
+  buildFederationHeaders: () => ({ 'Content-Type': 'application/json' }),
+  generateHmacSecret: () => 'secret',
+  ROTATION_GRACE_PERIOD_MS: 15 * 60 * 1000,
+}));
+
+vi.mock('../utils/federationOutbox.js', () => ({
+  isFederationRelayEnabled: () => true,
+  queueOutboxEvent: vi.fn(),
+  appendMutationLog: vi.fn(),
+}));
+
+vi.mock('../utils/federationPeerActivation.js', () => ({
+  onPeerActivated: vi.fn(),
+  startupBootstrapSync: vi.fn(),
+}));
+
+vi.mock('../utils/storageJanitor.js', () => ({
+  runFederationJanitor: vi.fn(),
+}));
+
+vi.mock('../utils/thumbnail.js', () => ({
+  generateThumbnail: vi.fn(),
+}));
+
+vi.mock('../routes/dm.js', () => ({
+  getDmMessageWithUser: vi.fn(),
+}));
+
+vi.mock('../utils/federationAuthFailure.js', () => ({
+  evaluateAuthFailure: vi.fn().mockReturnValue({ kind: 'increment', newAuthFailures: 1 }),
+  AUTH_FAILURE_THRESHOLD: 5,
+}));
+
+function applyMigrations(db: Database.Database): void {
+  const migrationsDir = path.resolve(__dirname, '../../drizzle');
+  const files = fs.readdirSync(migrationsDir).filter(f => f.endsWith('.sql')).sort();
+  for (const f of files) {
+    const sql = fs.readFileSync(path.join(migrationsDir, f), 'utf8');
+    const statements = sql.split(/-->\s*statement-breakpoint/);
+    for (const stmt of statements) {
+      const clean = stmt.trim();
+      if (clean) db.exec(clean);
+    }
+  }
+}
+
+function seedPeer(id: string): void {
+  testDb.insert(schema.federationPeers).values({
+    id, origin: 'https://peer.example', hmacSecret: 'secret',
+    status: 'active', lastSyncedAt: Date.now(), createdAt: Date.now(),
+  }).run();
+}
+
+function seedOutboxEntry(id: string, peerId: string, entityId: string): void {
+  testDb.insert(schema.federationOutbox).values({
+    id, peerId, contextId: 'ch-1', entityId,
+    contextType: 'dm', eventType: 'create', 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,
+    expiresAt: Date.now() + 30 * 86_400_000,
+    createdAt: Date.now(),
+  }).run();
+}
+
+describe('outbox worker — duplicate rejection is terminal', () => {
+  beforeEach(() => {
+    sqlite = new Database(':memory:');
+    testDb = drizzle(sqlite, { schema });
+    applyMigrations(sqlite);
+    vi.restoreAllMocks();
+    // Re-apply the static mocks that vi.restoreAllMocks() would undo.
+    // isFederationRelayEnabled is mocked at module level via vi.mock (hoisted),
+    // so it survives restoreAllMocks — spies created with vi.spyOn are the ones
+    // that get restored. The fetch spy is re-created per test via mockImplementation.
+  });
+
+  it('deletes the outbox entry when the peer responds with duplicate rejection', async () => {
+    seedPeer('peer-dup');
+    seedOutboxEntry('entry-dup', 'peer-dup', 'msg-already-there');
+
+    // Mock the fetch to return a relay response with duplicate rejection
+    vi.spyOn(globalThis, 'fetch').mockImplementation(async () =>
+      new Response(JSON.stringify({
+        accepted: [],
+        rejected: [{ messageId: 'msg-already-there', reason: 'duplicate' }],
+      }), { status: 200, headers: { 'Content-Type': 'application/json' } }),
+    );
+
+    const workerModule = await import('./federationWorker.js');
+    const processOutboxTick = (workerModule as { processOutboxTick?: () => Promise<void> }).processOutboxTick;
+    if (!processOutboxTick) {
+      throw new Error('processOutboxTick must be exported for this test. If not yet exported, export it.');
+    }
+    await processOutboxTick();
+
+    // Verify the outbox entry was deleted (terminal treatment)
+    const remaining = testDb.select().from(schema.federationOutbox)
+      .where(eq(schema.federationOutbox.id, 'entry-dup')).get();
+    expect(remaining).toBeUndefined();
+  });
+
+  it('retains outbox entries for non-duplicate rejection reasons (e.g., processing_error)', async () => {
+    seedPeer('peer-transient');
+    seedOutboxEntry('entry-transient', 'peer-transient', 'msg-transient');
+
+    vi.spyOn(globalThis, 'fetch').mockImplementation(async () =>
+      new Response(JSON.stringify({
+        accepted: [],
+        rejected: [{ messageId: 'msg-transient', reason: 'processing_error' }],
+      }), { status: 200, headers: { 'Content-Type': 'application/json' } }),
+    );
+
+    const workerModule = await import('./federationWorker.js');
+    const processOutboxTick = (workerModule as { processOutboxTick?: () => Promise<void> }).processOutboxTick;
+    if (!processOutboxTick) {
+      throw new Error('processOutboxTick must be exported for this test.');
+    }
+    await processOutboxTick();
+
+    const remaining = testDb.select().from(schema.federationOutbox)
+      .where(eq(schema.federationOutbox.id, 'entry-transient')).get();
+    // The entry must be retained — a 200 OK with a transient rejection reason
+    // does not delete the outbox entry. (Backoff is only applied on non-OK HTTP
+    // responses; a 200 with a rejection means the peer processed the batch but
+    // declined this particular message — the entry stays for the next tick.)
+    expect(remaining).toBeDefined();
+  });
+});
@@ -100,7 +100,7 @@ function scheduleOutboxTick(): void {
  }, OUTBOX_INTERVAL_MS);
 }

-async function processOutboxTick(): Promise<void> {
+export async function processOutboxTick(): Promise<void> {
  if (!isFederationRelayEnabled()) {
    return;
  }
@@ -230,26 +230,43 @@ async function processOutboxTick(): Promise<void> {
      if (response.ok) {
        const result = await response.json() as FederationRelayResponse;

-        // Delete accepted entries
-        if (result.accepted.length > 0) {
-          // Map accepted messageIds to outbox IDs
-          const acceptedSet = new Set(result.accepted);
-          const acceptedOutboxIds = peerEntries
-            .filter((e) => acceptedSet.has(e.entityId))
+        // 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.
+        const terminalEntityIds = new Set<string>(result.accepted);
+        for (const rejection of result.rejected) {
+          if (rejection.reason === 'duplicate') {
+            terminalEntityIds.add(rejection.messageId);
+          }
+        }
+
+        if (terminalEntityIds.size > 0) {
+          const terminalOutboxIds = peerEntries
+            .filter((e) => terminalEntityIds.has(e.entityId))
            .map((e) => e.outboxId);

-          if (acceptedOutboxIds.length > 0) {
+          if (terminalOutboxIds.length > 0) {
            db.delete(schema.federationOutbox)
-              .where(inArray(schema.federationOutbox.id, acceptedOutboxIds))
+              .where(inArray(schema.federationOutbox.id, terminalOutboxIds))
              .run();
          }
        }

-        // Log rejected entries (they remain in outbox for retry)
+        // 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.
        for (const rejection of result.rejected) {
-          console.warn(
-            `[federation-worker] Peer ${peerOrigin} rejected message ${rejection.messageId}: ${rejection.reason}`,
-          );
+          if (rejection.reason === 'duplicate') {
+            console.log(
+              `[federation-worker] Peer ${peerOrigin} rejected message ${rejection.messageId} as duplicate — outbox entry removed (terminal)`,
+            );
+          } else {
+            console.warn(
+              `[federation-worker] Peer ${peerOrigin} rejected message ${rejection.messageId}: ${rejection.reason}`,
+            );
+          }
        }

        // Store the peer's max upload size for informational display