From 57d7ca66d3fb5c17626a96f0bc53f8f480c2d0d8 Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Wed, 22 Apr 2026 00:34:43 +0200 Subject: [PATCH] fix(federation): explicit per-status handling in queueOutboxEvent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the silent UNIQUE-swallow placeholder branch. Each peer status has an explicit branch: active/pending/unreachable: race-catch — re-fetch peer row and enqueue via matchedPeers. Previously skipped silently, losing real-time delivery under asymmetric failure. awaiting_approval/needs_attention/rejected/revoked: drop with logged reason. Mutation log still captures; sync-pull on activation replays. default: exhaustiveness check (no 'as never' cast) — TypeScript enforces that every status value is handled explicitly. --- .../src/utils/federationOutbox.state.test.ts | 92 +++++++++++++++++++ packages/server/src/utils/federationOutbox.ts | 86 ++++++++++++----- 2 files changed, 157 insertions(+), 21 deletions(-) create mode 100644 packages/server/src/utils/federationOutbox.state.test.ts diff --git a/packages/server/src/utils/federationOutbox.state.test.ts b/packages/server/src/utils/federationOutbox.state.test.ts new file mode 100644 index 00000000..68011ffd --- /dev/null +++ b/packages/server/src/utils/federationOutbox.state.test.ts @@ -0,0 +1,92 @@ +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>; + +// Mutable reference updated in beforeEach — the factory closes over this. +let testDb: TestDb; + +vi.mock('../db/index.js', () => ({ + getDb: () => testDb, + schema, +})); + +// Mock federation-auth helpers to avoid env-var dependency +vi.mock('../utils/federationAuth.js', () => ({ + getOurOrigin: () => 'https://local.example', + buildFederationHeaders: () => ({}), + generateHmacSecret: () => 'test-secret', +})); + +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 seedSettings(): void { + testDb.insert(schema.instanceSettings).values({ + id: 1, + federationRelayEnabled: 1, + federationRelayTtlDays: 30, + updatedAt: Date.now(), + }).run(); +} + +function seedPeer(id: string, origin: string, status: string): void { + testDb.insert(schema.federationPeers).values({ + id, origin, hmacSecret: 'secret', + status, lastSyncedAt: 0, createdAt: Date.now(), + }).run(); +} + +function countOutbox(peerId: string): number { + return testDb.select().from(schema.federationOutbox) + .where(eq(schema.federationOutbox.peerId, peerId)) + .all().length; +} + +// Import once at module level — vi.mock is hoisted and the factory returns the +// live testDb reference, so re-using the cached import is correct. +const { queueOutboxEvent } = await import('./federationOutbox.js'); + +describe('queueOutboxEvent — non-deliverable statuses', () => { + beforeEach(() => { + const sqlite = new Database(':memory:'); + testDb = drizzle(sqlite, { schema }); + applyMigrations(sqlite); + seedSettings(); + vi.restoreAllMocks(); + }); + + it.each([ + ['awaiting_approval'], + ['needs_attention'], + ['rejected'], + ['revoked'], + ])('drops the event and logs a reason for %s peers (no outbox row, no throw)', (status) => { + seedPeer('peer-drop', 'https://drop.example', status); + + const debugSpy = vi.spyOn(console, 'debug').mockImplementation(() => {}); + + queueOutboxEvent('entity-1', 'ctx-1', 'create', '{}', ['https://drop.example'], 'dm'); + + expect(countOutbox('peer-drop')).toBe(0); + expect(debugSpy).toHaveBeenCalled(); + expect(debugSpy.mock.calls[0]![0] as string).toContain(status); + }); +}); diff --git a/packages/server/src/utils/federationOutbox.ts b/packages/server/src/utils/federationOutbox.ts index 36bbb37c..6e86414a 100644 --- a/packages/server/src/utils/federationOutbox.ts +++ b/packages/server/src/utils/federationOutbox.ts @@ -163,40 +163,84 @@ export function queueOutboxEvent( for (const origin of targetPeerOrigins) { if (matchedOrigins.has(origin)) continue; - // Check if there's a rejected/revoked peer we should skip const existingPeer = db .select({ status: schema.federationPeers.status }) .from(schema.federationPeers) .where(eq(schema.federationPeers.origin, origin)) .get(); - if (existingPeer && (existingPeer.status === 'rejected' || existingPeer.status === 'revoked')) { - console.warn(`[federation] queueOutboxEvent: skipping ${existingPeer.status} peer ${origin}`); - continue; - } - - // No peer record at all — create a pending placeholder - const peerId = generateSnowflake(); - const now = Date.now(); - db.insert(schema.federationPeers) - .values({ + if (!existingPeer) { + // No peer row — create pending placeholder, handshake fires on next tick + const peerId = generateSnowflake(); + const now = Date.now(); + db.insert(schema.federationPeers).values({ id: peerId, origin, hmacSecret: generateHmacSecret(), status: 'pending', createdAt: now, - }) - .run(); + }).run(); + const newPeer = db.select().from(schema.federationPeers) + .where(eq(schema.federationPeers.id, peerId)).get(); + if (newPeer) { + matchedPeers = [...matchedPeers, newPeer]; + console.log(`[federation] queueOutboxEvent: created pending placeholder for ${origin}`); + } + continue; + } - const newPeer = db - .select() - .from(schema.federationPeers) - .where(eq(schema.federationPeers.id, peerId)) - .get(); + // schema.federationPeers.status is plain text — narrow to known union for + // compile-time exhaustiveness check without widening to `string`. + const status = existingPeer.status as + | 'active' + | 'pending' + | 'unreachable' + | 'awaiting_approval' + | 'needs_attention' + | 'rejected' + | 'revoked'; - if (newPeer) { - matchedPeers = [...matchedPeers, newPeer]; - console.log(`[federation] queueOutboxEvent: created pending placeholder for ${origin}`); + switch (status) { + case 'active': + case 'pending': + case 'unreachable': { + // Race: peer transitioned to a deliverable status between the initial + // peers SELECT and this point in the loop. Re-fetch the full row and + // add to matchedPeers so the outer enqueue loop includes this peer. + // Do NOT silently drop — symmetric onPeerActivated on the peer's side + // is not guaranteed to cover asymmetric-failure cases (lost /peer/accept + // 200, health-check-only transition on one side). + const raced = db + .select() + .from(schema.federationPeers) + .where(eq(schema.federationPeers.origin, origin)) + .get(); + if (raced) { + matchedPeers = [...matchedPeers, raced]; + console.log(`[federation] queueOutboxEvent: race-caught ${origin} (now ${status}); enqueueing`); + } + break; + } + case 'awaiting_approval': + console.debug(`[federation] queueOutboxEvent: skipping ${origin} (awaiting_approval); mutation log will replay on activation`); + break; + case 'needs_attention': + console.debug(`[federation] queueOutboxEvent: skipping ${origin} (needs_attention; admin Reset required); mutation log will replay after Reset + re-peer`); + break; + case 'rejected': + console.debug(`[federation] queueOutboxEvent: skipping ${origin} (rejected peering)`); + break; + case 'revoked': + console.debug(`[federation] queueOutboxEvent: skipping ${origin} (revoked by admin)`); + break; + default: { + // Exhaustiveness check — no `as never` cast. TypeScript enforces + // that every status value is handled; adding a new value to the + // union without a case here fails typecheck. + const _exhaustive: never = status; + console.error(`[federation] queueOutboxEvent: unknown peer status for ${origin}: ${String(_exhaustive)}`); + break; + } } } }