fix(federation): explicit per-status handling in queueOutboxEvent
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.
This commit is contained in:
@@ -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<typeof drizzle<typeof schema>>;
|
||||||
|
|
||||||
|
// 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);
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -163,40 +163,84 @@ export function queueOutboxEvent(
|
|||||||
for (const origin of targetPeerOrigins) {
|
for (const origin of targetPeerOrigins) {
|
||||||
if (matchedOrigins.has(origin)) continue;
|
if (matchedOrigins.has(origin)) continue;
|
||||||
|
|
||||||
// Check if there's a rejected/revoked peer we should skip
|
|
||||||
const existingPeer = db
|
const existingPeer = db
|
||||||
.select({ status: schema.federationPeers.status })
|
.select({ status: schema.federationPeers.status })
|
||||||
.from(schema.federationPeers)
|
.from(schema.federationPeers)
|
||||||
.where(eq(schema.federationPeers.origin, origin))
|
.where(eq(schema.federationPeers.origin, origin))
|
||||||
.get();
|
.get();
|
||||||
|
|
||||||
if (existingPeer && (existingPeer.status === 'rejected' || existingPeer.status === 'revoked')) {
|
if (!existingPeer) {
|
||||||
console.warn(`[federation] queueOutboxEvent: skipping ${existingPeer.status} peer ${origin}`);
|
// No peer row — create pending placeholder, handshake fires on next tick
|
||||||
continue;
|
const peerId = generateSnowflake();
|
||||||
}
|
const now = Date.now();
|
||||||
|
db.insert(schema.federationPeers).values({
|
||||||
// No peer record at all — create a pending placeholder
|
|
||||||
const peerId = generateSnowflake();
|
|
||||||
const now = Date.now();
|
|
||||||
db.insert(schema.federationPeers)
|
|
||||||
.values({
|
|
||||||
id: peerId,
|
id: peerId,
|
||||||
origin,
|
origin,
|
||||||
hmacSecret: generateHmacSecret(),
|
hmacSecret: generateHmacSecret(),
|
||||||
status: 'pending',
|
status: 'pending',
|
||||||
createdAt: now,
|
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
|
// schema.federationPeers.status is plain text — narrow to known union for
|
||||||
.select()
|
// compile-time exhaustiveness check without widening to `string`.
|
||||||
.from(schema.federationPeers)
|
const status = existingPeer.status as
|
||||||
.where(eq(schema.federationPeers.id, peerId))
|
| 'active'
|
||||||
.get();
|
| 'pending'
|
||||||
|
| 'unreachable'
|
||||||
|
| 'awaiting_approval'
|
||||||
|
| 'needs_attention'
|
||||||
|
| 'rejected'
|
||||||
|
| 'revoked';
|
||||||
|
|
||||||
if (newPeer) {
|
switch (status) {
|
||||||
matchedPeers = [...matchedPeers, newPeer];
|
case 'active':
|
||||||
console.log(`[federation] queueOutboxEvent: created pending placeholder for ${origin}`);
|
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;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user