From 1be544bbd512ebcc910f30ca387928345884b6b0 Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Tue, 21 Apr 2026 13:33:52 +0200 Subject: [PATCH 01/10] feat(shared): add dm_call_undeliverable event type --- packages/shared/src/types.ts | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/packages/shared/src/types.ts b/packages/shared/src/types.ts index 24b452e9..0d8c123a 100644 --- a/packages/shared/src/types.ts +++ b/packages/shared/src/types.ts @@ -358,6 +358,19 @@ export interface Activity { // ─── WebSocket Event Types ────────────────────────────────────────────────── +export type DmCallUndeliverableReason = + | 'peer_rejected' + | 'peer_awaiting_approval' + | 'peer_transient_failure' + | 'livekit_unavailable'; + +export interface DmCallUndeliverableFailure { + reason: DmCallUndeliverableReason; + peerOrigin?: string; + peerLabel?: string; + affectedUserIds?: string[]; +} + // Client → Server Events export type ClientEvent = | { type: 'auth'; token: string } @@ -413,6 +426,7 @@ export type ServerEvent = | { type: 'dm_call_accepted'; dmChannelId: string | null; federatedCallId?: string } | { type: 'dm_call_rejected'; dmChannelId: string } | { type: 'dm_call_ended'; dmChannelId: string } + | { type: 'dm_call_undeliverable'; dmChannelId: string | null; federatedCallId: string; terminal: boolean; failures: DmCallUndeliverableFailure[] } | { type: 'voice_status_update'; userId: string; channelId: string; isMuted: boolean; isDeafened: boolean; isCameraOn: boolean; isScreenSharing: boolean } | { type: 'dm_channel_created'; dmChannel: DmChannel } | { type: 'dm_channel_closed'; dmChannelId: string } From b22a7bd0c674168ec9920b367417f085876384ab Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Tue, 21 Apr 2026 13:37:30 +0200 Subject: [PATCH 02/10] feat(server): add racePeering helper with tests Exports `racePeering(origin, timeoutMs, ensurePeeredFn?)` that races `ensurePeered` against a deadline. On timeout, the background handshake continues (warming the peer for the next attempt) and a warn-logged .catch() prevents unhandledRejection. Injectable `ensurePeeredFn` param enables full DI in tests without mocking module internals. --- .../src/utils/federationPeering.test.ts | 49 ++++++++++++++++++- .../server/src/utils/federationPeering.ts | 35 +++++++++++++ 2 files changed, 83 insertions(+), 1 deletion(-) diff --git a/packages/server/src/utils/federationPeering.test.ts b/packages/server/src/utils/federationPeering.test.ts index e25d4495..96f31624 100644 --- a/packages/server/src/utils/federationPeering.test.ts +++ b/packages/server/src/utils/federationPeering.test.ts @@ -1,5 +1,6 @@ -import { describe, it, expect } from 'vitest'; +import { describe, it, expect, vi } from 'vitest'; import type { EnsurePeeredResult } from './federationPeering.js'; +import { racePeering } from './federationPeering.js'; describe('EnsurePeeredResult type', () => { it('active result has peerId', () => { @@ -43,3 +44,49 @@ describe('EnsurePeeredResult type', () => { } }); }); + +describe('racePeering', () => { + it('returns the ensurePeered result when it resolves before the timeout', async () => { + const stub = vi.fn(async (): Promise => ({ + status: 'active', + peerId: 'peer-1', + })); + const result = await racePeering('https://example.com', 1_000, stub); + expect(result).toEqual({ status: 'active', peerId: 'peer-1' }); + expect(stub).toHaveBeenCalledWith('https://example.com'); + }); + + it('returns timeout when ensurePeered takes longer than the deadline', async () => { + const stub = vi.fn((): Promise => new Promise(() => { + // Never resolves — simulates a slow handshake. + })); + const result = await racePeering('https://example.com', 50, stub); + expect(result).toEqual({ status: 'timeout' }); + }); + + it('returns rejected result verbatim when ensurePeered resolves with rejection', async () => { + const stub = vi.fn(async (): Promise => ({ + status: 'rejected', + error: 'peer denied', + })); + const result = await racePeering('https://example.com', 1_000, stub); + expect(result).toEqual({ status: 'rejected', error: 'peer denied' }); + }); + + it('attaches a warn-logged catch to the background promise so race-losing rejections do not emit unhandledRejection', async () => { + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); + const stub = vi.fn(() => new Promise((_, reject) => { + setTimeout(() => reject(new Error('late failure')), 30); + })); + const result = await racePeering('https://example.com', 10, stub); + expect(result).toEqual({ status: 'timeout' }); + // Give the background promise time to reject. + await new Promise(r => setTimeout(r, 50)); + expect(warnSpy).toHaveBeenCalledWith( + expect.stringContaining('background handshake'), + 'https://example.com', + expect.any(Error), + ); + warnSpy.mockRestore(); + }); +}); diff --git a/packages/server/src/utils/federationPeering.ts b/packages/server/src/utils/federationPeering.ts index 2bf49fcd..29fe7abc 100644 --- a/packages/server/src/utils/federationPeering.ts +++ b/packages/server/src/utils/federationPeering.ts @@ -210,3 +210,38 @@ async function performHandshake( export function _clearInFlightPeering(): void { inFlightPeering.clear(); } + +/** + * Race ensurePeered() against a deadline. On timeout, the background + * handshake is NOT aborted — it continues so the next attempt finds + * the peer active. A warn-logged catch is attached so a late-rejecting + * background promise does not emit an unhandledRejection. + * + * The ensurePeered implementation is injectable for testing; the default + * is the real function. + */ +export async function racePeering( + origin: string, + timeoutMs: number, + ensurePeeredFn: (origin: string) => Promise = ensurePeered, +): Promise { + const handshake = ensurePeeredFn(origin); + + // Attach a warn-logged catch so a background arm that rejects AFTER the + // race loses does not trigger unhandledRejection. This runs regardless + // of which arm wins. + handshake.catch(err => { + console.warn('[federation] background handshake after call-relay race:', origin, err); + }); + + let timeoutHandle: ReturnType | undefined; + const timeoutPromise = new Promise<{ status: 'timeout' }>(resolve => { + timeoutHandle = setTimeout(() => resolve({ status: 'timeout' }), timeoutMs); + }); + + try { + return await Promise.race([handshake, timeoutPromise]); + } finally { + if (timeoutHandle) clearTimeout(timeoutHandle); + } +} From 4ddb09edf15e781759134072c2b62a5358e96028 Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Tue, 21 Apr 2026 13:41:44 +0200 Subject: [PATCH 03/10] fix(server): racePeering normalizes handshake rejections and only warns on timeout win MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses code review on b22a7bd: (1) a rejected handshake now returns { status: 'failed', error } instead of throwing, keeping the structured contract; (2) the "background handshake" warn only fires when the timeout arm wins — not when the handshake is itself the race winner by rejection. Timing tests migrated to vi.useFakeTimers for determinism. Regression test added for the handshake-wins-by-rejection case. --- .../src/utils/federationPeering.test.ts | 32 ++++++++++++++++--- .../server/src/utils/federationPeering.ts | 26 +++++++++------ 2 files changed, 44 insertions(+), 14 deletions(-) diff --git a/packages/server/src/utils/federationPeering.test.ts b/packages/server/src/utils/federationPeering.test.ts index 96f31624..3961dc0b 100644 --- a/packages/server/src/utils/federationPeering.test.ts +++ b/packages/server/src/utils/federationPeering.test.ts @@ -57,11 +57,15 @@ describe('racePeering', () => { }); it('returns timeout when ensurePeered takes longer than the deadline', async () => { + vi.useFakeTimers(); const stub = vi.fn((): Promise => new Promise(() => { // Never resolves — simulates a slow handshake. })); - const result = await racePeering('https://example.com', 50, stub); + const racePromise = racePeering('https://example.com', 50, stub); + await vi.advanceTimersByTimeAsync(50); + const result = await racePromise; expect(result).toEqual({ status: 'timeout' }); + vi.useRealTimers(); }); it('returns rejected result verbatim when ensurePeered resolves with rejection', async () => { @@ -73,20 +77,38 @@ describe('racePeering', () => { expect(result).toEqual({ status: 'rejected', error: 'peer denied' }); }); - it('attaches a warn-logged catch to the background promise so race-losing rejections do not emit unhandledRejection', async () => { + it('attaches a warn-logged catch to the background handshake when the timeout wins', async () => { + vi.useFakeTimers(); const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); const stub = vi.fn(() => new Promise((_, reject) => { setTimeout(() => reject(new Error('late failure')), 30); })); - const result = await racePeering('https://example.com', 10, stub); + const racePromise = racePeering('https://example.com', 10, stub); + await vi.advanceTimersByTimeAsync(10); + const result = await racePromise; expect(result).toEqual({ status: 'timeout' }); - // Give the background promise time to reject. - await new Promise(r => setTimeout(r, 50)); + await vi.advanceTimersByTimeAsync(30); + // Let microtasks flush so the .catch handler runs. + await Promise.resolve(); expect(warnSpy).toHaveBeenCalledWith( expect.stringContaining('background handshake'), 'https://example.com', expect.any(Error), ); + vi.useRealTimers(); + warnSpy.mockRestore(); + }); + + it('normalizes a thrown handshake error into { status: failed } without emitting the background warn', async () => { + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); + const stub = vi.fn(async (): Promise => { + throw new Error('immediate handshake failure'); + }); + const result = await racePeering('https://example.com', 1_000, stub); + expect(result).toEqual({ status: 'failed', error: 'immediate handshake failure' }); + // The handshake rejection was the race winner — no background warn should fire. + await Promise.resolve(); + expect(warnSpy).not.toHaveBeenCalled(); warnSpy.mockRestore(); }); }); diff --git a/packages/server/src/utils/federationPeering.ts b/packages/server/src/utils/federationPeering.ts index 29fe7abc..2b582413 100644 --- a/packages/server/src/utils/federationPeering.ts +++ b/packages/server/src/utils/federationPeering.ts @@ -227,21 +227,29 @@ export async function racePeering( ): Promise { const handshake = ensurePeeredFn(origin); - // Attach a warn-logged catch so a background arm that rejects AFTER the - // race loses does not trigger unhandledRejection. This runs regardless - // of which arm wins. - handshake.catch(err => { - console.warn('[federation] background handshake after call-relay race:', origin, err); - }); - let timeoutHandle: ReturnType | undefined; const timeoutPromise = new Promise<{ status: 'timeout' }>(resolve => { timeoutHandle = setTimeout(() => resolve({ status: 'timeout' }), timeoutMs); }); + let raceResult: EnsurePeeredResult | { status: 'timeout' }; try { - return await Promise.race([handshake, timeoutPromise]); - } finally { + raceResult = await Promise.race([handshake, timeoutPromise]); + } catch (err) { + // ensurePeeredFn rejected as the race winner. Normalize to failed. if (timeoutHandle) clearTimeout(timeoutHandle); + const message = err instanceof Error ? err.message : 'Unknown handshake error'; + return { status: 'failed', error: message }; } + if (timeoutHandle) clearTimeout(timeoutHandle); + + // Only when the timeout arm won is the background handshake still running. + // Guard its eventual rejection so we don't emit unhandledRejection. + if (raceResult.status === 'timeout') { + handshake.catch(err => { + console.warn('[federation] background handshake after call-relay race:', origin, err); + }); + } + + return raceResult; } From 21f220739c76ecc23e5846fdee1f3c36db97b371 Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Tue, 21 Apr 2026 13:45:29 +0200 Subject: [PATCH 04/10] feat(server): sendCallRelay auto-peers on demand, typing passes peeringTimeoutMs:0 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit sendCallRelay now returns CallRelayResult with a typed reason on failure. When the peer is not already active (or unreachable), runs a racePeering against CALL_PEERING_TIMEOUT_MS (3s). Background handshake is not aborted on race loss — next attempt succeeds. sendTypingRelay passes peeringTimeoutMs:0 so typing never blocks on a handshake; instead a warm-up ensurePeered runs in the background for any non-active peer so the NEXT relay (message, call, or typing) benefits. --- packages/server/src/utils/federationOutbox.ts | 108 +++++++++++++++--- 1 file changed, 90 insertions(+), 18 deletions(-) diff --git a/packages/server/src/utils/federationOutbox.ts b/packages/server/src/utils/federationOutbox.ts index 9fe2bec2..ce554125 100644 --- a/packages/server/src/utils/federationOutbox.ts +++ b/packages/server/src/utils/federationOutbox.ts @@ -6,6 +6,7 @@ import crypto from 'node:crypto'; import type { FederationRelayEvent, FederationRelayParticipant, FederationRelayAttachment, DmMessageWithUser, FederationRelayRequest } from '@backspace/shared'; import { getOurOrigin, buildFederationHeaders, generateHmacSecret } from './federationAuth.js'; import { extractDomain } from '../routes/federation.js'; +import { racePeering, ensurePeered } from './federationPeering.js'; // ─── Settings Cache ────────────────────────────────────────────────────────── @@ -469,28 +470,87 @@ export function buildRelayPayload( }; } +/** 3s budget for the on-demand handshake before a call relay POST. */ +export const CALL_PEERING_TIMEOUT_MS = 3_000; + +export type CallRelayFailureReason = + | 'peer_rejected' + | 'peer_awaiting_approval' + | 'peer_transient_failure' + | 'post_failed'; + +export type CallRelayResult = + | { ok: true } + | { ok: false; reason: CallRelayFailureReason; error: string }; + /** * Send call signaling events directly to a remote peer (bypasses outbox). - * Used for time-critical call events where latency matters. - * If the HTTP POST fails, the call operation fails — no retry. + * Latency-sensitive: if no active peer exists, race an ensurePeered handshake + * against `opts.peeringTimeoutMs` (default CALL_PEERING_TIMEOUT_MS). + * + * `peeringTimeoutMs: 0` = non-blocking mode (used by sendTypingRelay): + * - If the peer is currently active, POST. Otherwise skip the POST, kick off + * ensurePeered() in the background as a warm-up, and return + * { ok:false, reason:'peer_transient_failure' }. */ export async function sendCallRelay( targetPeerOrigin: string, events: FederationRelayEvent[], -): Promise<{ ok: boolean; error?: string }> { + opts: { peeringTimeoutMs?: number } = {}, +): Promise { + const timeoutMs = opts.peeringTimeoutMs ?? CALL_PEERING_TIMEOUT_MS; const db = getDb(); - const peer = db.select() + + // ─── Fast path: peer already active or unreachable (health check handles) ── + const existing = db.select() .from(schema.federationPeers) - .where(and( - eq(schema.federationPeers.origin, targetPeerOrigin), - eq(schema.federationPeers.status, 'active'), - )) + .where(eq(schema.federationPeers.origin, targetPeerOrigin)) .get(); + let peer = existing && (existing.status === 'active' || existing.status === 'unreachable') + ? existing + : null; + if (!peer) { - return { ok: false, error: `No active peer for origin ${targetPeerOrigin}` }; + // ─── Non-blocking mode (typing): warm up in background, do not POST ── + if (timeoutMs === 0) { + ensurePeered(targetPeerOrigin).catch(err => { + console.warn('[federation] typing-triggered background handshake:', targetPeerOrigin, err); + }); + return { ok: false, reason: 'peer_transient_failure', error: 'peer not active' }; + } + + // ─── Race ensurePeered against the deadline ── + const raced = await racePeering(targetPeerOrigin, timeoutMs); + + switch (raced.status) { + case 'active': + // Re-fetch the now-active peer row for HMAC secret. + peer = db.select() + .from(schema.federationPeers) + .where(eq(schema.federationPeers.origin, targetPeerOrigin)) + .get() ?? null; + if (!peer) { + return { ok: false, reason: 'peer_transient_failure', error: 'peer row missing after handshake' }; + } + break; + case 'rejected': + return { ok: false, reason: 'peer_rejected', error: raced.error }; + case 'pending': + return { ok: false, reason: 'peer_awaiting_approval', error: raced.error }; + case 'failed': + return { ok: false, reason: 'peer_transient_failure', error: raced.error }; + case 'timeout': + return { ok: false, reason: 'peer_transient_failure', error: `handshake did not complete in ${timeoutMs}ms` }; + default: { + // Exhaustiveness check — catches any future additions to EnsurePeeredResult. + const _exhaustive: never = raced; + return { ok: false, reason: 'peer_transient_failure', error: `unexpected peering result: ${JSON.stringify(_exhaustive)}` }; + } + } } + // ─── POST ────────────────────────────────────────────────────────────────── const ourOrigin = getOurOrigin(); const body: FederationRelayRequest = { version: 1, @@ -499,7 +559,6 @@ export async function sendCallRelay( }; const bodyStr = JSON.stringify(body); - // Use pending secret during rotation (FED-011), otherwise current const signingSecret = peer.pendingHmacSecret ?? peer.hmacSecret; const headers = buildFederationHeaders(bodyStr, signingSecret, ourOrigin); @@ -511,14 +570,27 @@ export async function sendCallRelay( signal: AbortSignal.timeout(10_000), }); - if (!res.ok) { - const text = await res.text().catch(() => ''); - return { ok: false, error: `HTTP ${res.status}: ${text}` }; - } + if (res.ok) return { ok: true }; - return { ok: true }; + const text = await res.text().catch(() => ''); + if (res.status >= 400 && res.status < 500) { + return { + ok: false, + reason: 'post_failed', + error: `HTTP ${res.status}: ${text}`, + }; + } + return { + ok: false, + reason: 'peer_transient_failure', + error: `HTTP ${res.status}: ${text}`, + }; } catch (err) { - return { ok: false, error: err instanceof Error ? err.message : 'fetch_failed' }; + return { + ok: false, + reason: 'peer_transient_failure', + error: err instanceof Error ? err.message : 'fetch_failed', + }; } } @@ -700,9 +772,9 @@ export async function sendTypingRelay( }, }; - // Fire-and-forget to each remote peer + // Fire-and-forget to each remote peer; 0ms peering timeout = non-blocking warm-up. for (const peerOrigin of remoteOrigins) { - sendCallRelay(peerOrigin, [event]).catch(err => { + sendCallRelay(peerOrigin, [event], { peeringTimeoutMs: 0 }).catch(err => { console.warn(`[federation] Typing relay to ${peerOrigin} failed:`, err); }); } From 53483d6981f51fd956bd280dc48f320a8bef08a4 Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Tue, 21 Apr 2026 13:49:20 +0200 Subject: [PATCH 05/10] polish(server): align sendCallRelay timeout message with codebase convention; use .then on sendTypingRelay fire-and-forget --- packages/server/src/utils/federationOutbox.ts | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/packages/server/src/utils/federationOutbox.ts b/packages/server/src/utils/federationOutbox.ts index ce554125..36bbb37c 100644 --- a/packages/server/src/utils/federationOutbox.ts +++ b/packages/server/src/utils/federationOutbox.ts @@ -541,7 +541,7 @@ export async function sendCallRelay( case 'failed': return { ok: false, reason: 'peer_transient_failure', error: raced.error }; case 'timeout': - return { ok: false, reason: 'peer_transient_failure', error: `handshake did not complete in ${timeoutMs}ms` }; + return { ok: false, reason: 'peer_transient_failure', error: `Peering handshake did not complete within ${(timeoutMs / 1000).toFixed(1)}s` }; default: { // Exhaustiveness check — catches any future additions to EnsurePeeredResult. const _exhaustive: never = raced; @@ -774,8 +774,14 @@ export async function sendTypingRelay( // Fire-and-forget to each remote peer; 0ms peering timeout = non-blocking warm-up. for (const peerOrigin of remoteOrigins) { - sendCallRelay(peerOrigin, [event], { peeringTimeoutMs: 0 }).catch(err => { - console.warn(`[federation] Typing relay to ${peerOrigin} failed:`, err); - }); + sendCallRelay(peerOrigin, [event], { peeringTimeoutMs: 0 }) + .then(result => { + if (!result.ok) { + console.debug(`[federation] Typing relay to ${peerOrigin}: ${result.reason} ${result.error}`); + } + }) + .catch(err => { + console.warn(`[federation] Typing relay to ${peerOrigin} threw unexpectedly:`, err); + }); } } From f83c2af357c9f383dfd8e48bf3ff6cca6e07903d Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Tue, 21 Apr 2026 13:53:34 +0200 Subject: [PATCH 06/10] feat(server): aggregate call-start failures into dm_call_undeliverable sendFederatedCallStart now collects per-targeted-peer results and emits a single dm_call_undeliverable event to the caller when any targeted peer relay fails. Destroys the local ring room when no plausible recipient remains (no targeted success + no connected local ringee). LiveKit pre-flight also emits via this path with reason 'livekit_unavailable' instead of a silent console.warn, closing the 60s hang for unconfigured instances. Guards against phantom toasts when the caller cancels mid-race by checking getRoom() before emitting. --- packages/server/src/ws/events.ts | 178 ++++++++++++++++++++++++------- 1 file changed, 142 insertions(+), 36 deletions(-) diff --git a/packages/server/src/ws/events.ts b/packages/server/src/ws/events.ts index ab8a7cba..12027c30 100644 --- a/packages/server/src/ws/events.ts +++ b/packages/server/src/ws/events.ts @@ -6,7 +6,8 @@ import { connectionManager } from './handler.js'; import type { VoiceRoom, DmRoomMeta, SpaceRoomMeta } from './handler.js'; import { isMember, getChannelSpaceId, isDmMember, hasPermission, computePermissions, PermissionBits } from '../utils/permissions.js'; import { broadcastDmMessage, getDmMessageWithUser } from '../routes/dm.js'; -import { MAX_MESSAGE_LENGTH, type MessageWithUser, type Attachment, type DmMessageWithUser, type Embed, type Activity, type ActivityType, type ActivityTimestamps, type ActivityAssets, type ServerEvent } from '@backspace/shared'; +import { MAX_MESSAGE_LENGTH, type MessageWithUser, type Attachment, type DmMessageWithUser, type Embed, type Activity, type ActivityType, type ActivityTimestamps, type ActivityAssets, type ServerEvent, type DmCallUndeliverableFailure, type DmCallUndeliverableReason } from '@backspace/shared'; +import type { CallRelayFailureReason } from '../utils/federationOutbox.js'; import { ACTIVITY_LIMITS } from '@backspace/shared/src/activities.js'; import { sanitizeUser } from '../utils/sanitize.js'; import { deleteAttachmentFiles } from '../utils/fileCleanup.js'; @@ -1692,7 +1693,8 @@ function handleDmCallEnd(event: Record, userId: string): void { /** * Send S2S dm_call_start to all remote instances with DM members. - * Fire-and-forget — if delivery fails, the call still works locally. + * Fire-and-forget per relay, but aggregates per-peer results to surface + * undeliverable calls via `dm_call_undeliverable` to the caller. */ async function sendFederatedCallStart( dmChannelId: string, @@ -1746,9 +1748,33 @@ async function sendFederatedCallStart( .run(); } - // Check LiveKit configuration - if (!config.livekit.apiKey || !config.livekit.apiSecret) { + // Classify members relative to this instance. + const remoteMembers = members.filter(m => { + if (!m.homeInstance) return false; + const normalized = m.homeInstance.startsWith('http') ? m.homeInstance : `https://${m.homeInstance}`; + return normalized !== ourOrigin; + }); + const localNonCallerMembers = members.filter(m => { + if (m.userId === callerId) return false; + const home = m.homeInstance + ? (m.homeInstance.startsWith('http') ? m.homeInstance : `https://${m.homeInstance}`) + : ourOrigin; + return home === ourOrigin; + }); + const hasConnectedLocalRingee = localNonCallerMembers.some(m => + connectionManager.isUserOnline(m.userId), + ); + + // ─── LiveKit pre-flight ──────────────────────────────────────────────────── + if ((!config.livekit.apiKey || !config.livekit.apiSecret) && remoteMembers.length > 0) { console.warn('[federation] Cannot start federated call: LiveKit not configured'); + emitUndeliverableAndMaybeDestroy({ + callerId, + dmChannelId, + federatedId, + terminal: !hasConnectedLocalRingee, + failures: [{ reason: 'livekit_unavailable' }], + }); return; } @@ -1793,55 +1819,135 @@ async function sendFederatedCallStart( }, }); - // Identify remote members for targeted relay - const remoteMembers = members.filter(m => { - if (!m.homeInstance) return false; - const normalized = m.homeInstance.startsWith('http') ? m.homeInstance : `https://${m.homeInstance}`; - return normalized !== ourOrigin; - }); - // Group remote members by home instance (targeted peers) - const targetedPeers = new Set(); + const targetedPeers = new Map>(); for (const m of remoteMembers) { const origin = m.homeInstance!.startsWith('http') ? m.homeInstance! : `https://${m.homeInstance!}`; - targetedPeers.add(origin); + const bucket = targetedPeers.get(origin) ?? []; + bucket.push({ userId: m.userId, displayName: m.displayName, username: m.username }); + targetedPeers.set(origin, bucket); } - // Query ALL active federation peers for broadcast + // All active federation peers for broadcast const allPeers = db.select({ origin: schema.federationPeers.origin }) .from(schema.federationPeers) .where(eq(schema.federationPeers.status, 'active')) .all(); - // Build parallel relay promises - const relayPromises: Promise[] = []; - - // Targeted relay: peers with known remote DM members - for (const peerOrigin of targetedPeers) { - relayPromises.push( - sendCallRelay(peerOrigin, [buildRelayEvent()]).then(result => { - if (!result.ok) { - console.error(`[federation] Failed to send dm_call_start to ${peerOrigin}: ${result.error}`); - } - }) - ); + // Peer label resolution (instanceName if known) + const peerRows = db.select({ + origin: schema.federationPeers.origin, + instanceName: schema.federationPeers.instanceName, + }) + .from(schema.federationPeers) + .all(); + const peerLabelByOrigin = new Map(); + for (const row of peerRows) { + if (row.instanceName) peerLabelByOrigin.set(row.origin, row.instanceName); } - // All-peers broadcast: every other active peer + // ─── Targeted relay: fan out in parallel, await results ──────────────────── + const targetedResults: Array<{ origin: string; ok: boolean; reason?: DmCallUndeliverableReason; error?: string }> = []; + await Promise.all( + Array.from(targetedPeers.keys()).map(async peerOrigin => { + const result = await sendCallRelay(peerOrigin, [buildRelayEvent()]); + if (result.ok) { + targetedResults.push({ origin: peerOrigin, ok: true }); + } else { + const reason = mapCallReasonToEventReason(result.reason); + console.error(`[federation] dm_call_start to ${peerOrigin} failed (${result.reason}): ${result.error}`); + targetedResults.push({ origin: peerOrigin, ok: false, reason, error: result.error }); + } + }), + ); + + // ─── All-peers broadcast: fire-and-forget; failures NOT surfaced ─────────── for (const peer of allPeers) { if (targetedPeers.has(peer.origin)) continue; if (peer.origin === ourOrigin) continue; - relayPromises.push( - sendCallRelay(peer.origin, [buildRelayEvent()]).then(result => { - if (!result.ok) { - console.debug(`[federation] All-peers dm_call_start to ${peer.origin}: ${result.error || 'failed'}`); - } - }) - ); + sendCallRelay(peer.origin, [buildRelayEvent()]).then(result => { + if (!result.ok) { + console.debug(`[federation] All-peers dm_call_start to ${peer.origin}: ${result.reason} ${result.error}`); + } + }).catch(err => console.warn('[federation] all-peers broadcast threw:', err)); } - // Fire all relays in parallel — each has its own 10s timeout - await Promise.all(relayPromises); + // ─── Aggregate failures → dm_call_undeliverable ─────────────────────────── + const failedTargeted = targetedResults.filter(r => !r.ok); + if (failedTargeted.length === 0) return; + + const anyTargetedSuccess = targetedResults.some(r => r.ok); + const plausibleRecipientRemains = anyTargetedSuccess || hasConnectedLocalRingee; + + const failures: DmCallUndeliverableFailure[] = failedTargeted.map(r => { + const affected = targetedPeers.get(r.origin) ?? []; + return { + reason: r.reason!, + peerOrigin: r.origin, + peerLabel: peerLabelByOrigin.get(r.origin), + affectedUserIds: affected.map(m => m.userId), + }; + }); + + emitUndeliverableAndMaybeDestroy({ + callerId, + dmChannelId, + federatedId, + terminal: !plausibleRecipientRemains, + failures, + }); +} + +/** Map a sendCallRelay reason to the event-surface reason. */ +function mapCallReasonToEventReason(reason: CallRelayFailureReason): DmCallUndeliverableReason { + switch (reason) { + case 'peer_rejected': return 'peer_rejected'; + case 'peer_awaiting_approval': return 'peer_awaiting_approval'; + case 'peer_transient_failure': return 'peer_transient_failure'; + case 'post_failed': return 'peer_transient_failure'; // 4xx looks transient to users + } +} + +/** + * Emit dm_call_undeliverable to the caller. If terminal, also destroy the + * local ring room (which clears the ringing timer and voice WS binding) + * and broadcast dm_call_ended to non-caller DM members, mirroring the + * 60s auto-timeout's cleanup semantics (handler.ts:414-424) so any + * ringing client (e.g., a Connection WS from another instance) exits + * the ring state instead of hanging. + * + * Guard: if the caller already cancelled mid-race, the room is already + * gone — do NOT emit a phantom "could not reach" toast. + */ +function emitUndeliverableAndMaybeDestroy(args: { + callerId: string; + dmChannelId: string; + federatedId: string; + terminal: boolean; + failures: DmCallUndeliverableFailure[]; +}): void { + const { callerId, dmChannelId, federatedId, terminal, failures } = args; + + // Caller may have cancelled mid-race. If the room is gone, move on silently. + const room = connectionManager.getRoom(dmChannelId); + if (!room) return; + + if (terminal) { + connectionManager.clearVoiceWs(callerId); + connectionManager.destroyRoom(dmChannelId); + connectionManager.sendToDmMembers(dmChannelId, { + type: 'dm_call_ended', + dmChannelId, + }, callerId); + } + + connectionManager.sendToUser(callerId, { + type: 'dm_call_undeliverable', + dmChannelId, + federatedCallId: federatedId, + terminal, + failures, + }); } async function sendFederatedCallAccept(dmChannelId: string, acceptorUserId: string): Promise { From 44a44163af16429e29170697ddc0d2c29c550906 Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Tue, 21 Apr 2026 14:00:29 +0200 Subject: [PATCH 07/10] polish(server): consolidate federation_peers queries, narrow targetedPeers map, use return values from Promise.all to drop non-null assertion --- packages/server/src/ws/events.ts | 38 +++++++++++++++----------------- 1 file changed, 18 insertions(+), 20 deletions(-) diff --git a/packages/server/src/ws/events.ts b/packages/server/src/ws/events.ts index 12027c30..393ae9bf 100644 --- a/packages/server/src/ws/events.ts +++ b/packages/server/src/ws/events.ts @@ -1820,44 +1820,40 @@ async function sendFederatedCallStart( }); // Group remote members by home instance (targeted peers) - const targetedPeers = new Map>(); + const targetedPeers = new Map(); for (const m of remoteMembers) { const origin = m.homeInstance!.startsWith('http') ? m.homeInstance! : `https://${m.homeInstance!}`; const bucket = targetedPeers.get(origin) ?? []; - bucket.push({ userId: m.userId, displayName: m.displayName, username: m.username }); + bucket.push(m.userId); targetedPeers.set(origin, bucket); } - // All active federation peers for broadcast - const allPeers = db.select({ origin: schema.federationPeers.origin }) - .from(schema.federationPeers) - .where(eq(schema.federationPeers.status, 'active')) - .all(); - - // Peer label resolution (instanceName if known) + // Single query for all peers — derives both active-peer list and label map const peerRows = db.select({ origin: schema.federationPeers.origin, instanceName: schema.federationPeers.instanceName, + status: schema.federationPeers.status, }) .from(schema.federationPeers) .all(); + + const allPeers = peerRows.filter(r => r.status === 'active'); + const peerLabelByOrigin = new Map(); for (const row of peerRows) { if (row.instanceName) peerLabelByOrigin.set(row.origin, row.instanceName); } // ─── Targeted relay: fan out in parallel, await results ──────────────────── - const targetedResults: Array<{ origin: string; ok: boolean; reason?: DmCallUndeliverableReason; error?: string }> = []; - await Promise.all( + const targetedResults = await Promise.all( Array.from(targetedPeers.keys()).map(async peerOrigin => { const result = await sendCallRelay(peerOrigin, [buildRelayEvent()]); if (result.ok) { - targetedResults.push({ origin: peerOrigin, ok: true }); - } else { - const reason = mapCallReasonToEventReason(result.reason); - console.error(`[federation] dm_call_start to ${peerOrigin} failed (${result.reason}): ${result.error}`); - targetedResults.push({ origin: peerOrigin, ok: false, reason, error: result.error }); + return { origin: peerOrigin, ok: true as const }; } + const reason = mapCallReasonToEventReason(result.reason); + console.error(`[federation] dm_call_start to ${peerOrigin} failed (${result.reason}): ${result.error}`); + return { origin: peerOrigin, ok: false as const, reason, error: result.error }; }), ); @@ -1873,19 +1869,21 @@ async function sendFederatedCallStart( } // ─── Aggregate failures → dm_call_undeliverable ─────────────────────────── - const failedTargeted = targetedResults.filter(r => !r.ok); + const failedTargeted = targetedResults.filter( + (r): r is Extract => !r.ok, + ); if (failedTargeted.length === 0) return; const anyTargetedSuccess = targetedResults.some(r => r.ok); const plausibleRecipientRemains = anyTargetedSuccess || hasConnectedLocalRingee; const failures: DmCallUndeliverableFailure[] = failedTargeted.map(r => { - const affected = targetedPeers.get(r.origin) ?? []; + const affectedUserIds = targetedPeers.get(r.origin) ?? []; return { - reason: r.reason!, + reason: r.reason, peerOrigin: r.origin, peerLabel: peerLabelByOrigin.get(r.origin), - affectedUserIds: affected.map(m => m.userId), + affectedUserIds, }; }); From 6a7d1fb38cdb547ff3f2e7c93300b209a1175ae4 Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Tue, 21 Apr 2026 14:02:08 +0200 Subject: [PATCH 08/10] =?UTF-8?q?feat(web):=20handle=20dm=5Fcall=5Fundeliv?= =?UTF-8?q?erable=20=E2=80=94=20toast=20+=20tear=20down=20outgoing=20call?= =?UTF-8?q?=20on=20terminal?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/web/src/hooks/useWebSocket.ts | 57 ++++++++++++++++++++++++++ 1 file changed, 57 insertions(+) diff --git a/packages/web/src/hooks/useWebSocket.ts b/packages/web/src/hooks/useWebSocket.ts index 9ee59274..5cd33c53 100644 --- a/packages/web/src/hooks/useWebSocket.ts +++ b/packages/web/src/hooks/useWebSocket.ts @@ -114,6 +114,43 @@ function buildWsUrl(origin: string): string { return `${protocol}//${url.host}/ws`; } +// ─── Call relay helpers ─────────────────────────────────────────────────────── + +function buildCallUndeliverableToast( + failures: Array<{ reason: string; peerOrigin?: string; peerLabel?: string }>, + terminal: boolean, +): string { + const primary = failures[0]; + const labelFor = (f: { peerLabel?: string; peerOrigin?: string }) => + f.peerLabel ?? f.peerOrigin?.replace(/^https?:\/\//, '') ?? 'the remote instance'; + + if (!terminal) { + const labels = failures.map(labelFor).join(', '); + return `Some participants could not be reached: ${labels}.`; + } + + if (failures.length > 1) { + const labels = failures.map(labelFor).join(', '); + return `Could not reach ${failures.length} instances: ${labels}.`; + } + + if (!primary) return 'Call could not be placed.'; + + const label = labelFor(primary); + switch (primary.reason) { + case 'peer_rejected': + return `Cannot reach ${label} — this instance requires manual peering approval.`; + case 'peer_awaiting_approval': + return `Waiting for ${label} admin to approve your instance. Calls will work once approved.`; + case 'peer_transient_failure': + return `Could not reach ${label}. Try again in a moment.`; + case 'livekit_unavailable': + return 'Voice is not configured on this instance.'; + default: + return `Call to ${label} could not be placed.`; + } +} + // ─── Event handling ─────────────────────────────────────────────────────────── const HOME_ORIGIN = ''; @@ -954,6 +991,26 @@ function handleEvent(origin: string, event: ServerEvent): void { break; } + case 'dm_call_undeliverable': { + if (!isHome && !activePeerOrigins.has(origin)) break; + + const { setIncomingCall, setOutgoingCall, setActiveDmCall, disconnectFn, clearFederatedCallData } = useVoiceStore.getState(); + const { addToast } = useUIStore.getState(); + + if (event.terminal) { + // Tear down local outbound call state — mirrors dm_call_ended. + setIncomingCall(null); + setOutgoingCall(null); + setActiveDmCall(null); + clearFederatedCallData(); + if (disconnectFn) disconnectFn(); + } + + const msg = buildCallUndeliverableToast(event.failures, event.terminal); + addToast(msg, event.terminal ? 'warning' : 'info', 8_000); + break; + } + // ─── DM channel events (all origins) ──────────────────────────────────── case 'dm_channel_created': { From 740dae298d06ddc394d132cdb7a8f34f6056127b Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Tue, 21 Apr 2026 14:05:59 +0200 Subject: [PATCH 09/10] docs: document call-relay auto-peering and dm_call_undeliverable surface --- docs/systems/federation.md | 23 ++++++++++++++++++++--- docs/systems/voice.md | 8 +++++++- docs/systems/websocket.md | 1 + 3 files changed, 28 insertions(+), 4 deletions(-) diff --git a/docs/systems/federation.md b/docs/systems/federation.md index a663b850..5e86cea6 100644 --- a/docs/systems/federation.md +++ b/docs/systems/federation.md @@ -403,7 +403,7 @@ The `(source_instance, source_message_id)` pair is checked before insertion. Dup **`sendTypingRelay()` (`federationOutbox.ts`):** - Fetches channel's `federatedId` and `getDmParticipants()` for target resolution - Builds `FederationRelayEvent` with `typing: { homeUserId, homeInstance, username }` -- Reuses `sendCallRelay()` for the actual POST to each remote peer origin +- Calls `sendCallRelay(origin, [event], { peeringTimeoutMs: 0 })` for each remote peer origin — non-active peers are skipped and a background `ensurePeered` warm-up is kicked off instead **Inbound (`federation.ts`):** - `processDmTypingStartEvent` → look up channel by `federatedId`, resolve user via `resolveLocalUser()` (no stub creation for ephemeral events), broadcast `dm_typing` to local members @@ -1024,7 +1024,20 @@ All events carry standard relay fields: `eventType`, `messageId`, `encryptionVer ### Direct Delivery (No Outbox) -Call signaling is time-critical and bypasses the outbox entirely. `sendCallRelay()` sends a synchronous HTTP POST to the peer's `/api/federation/relay` endpoint using existing HMAC signing (`buildFederationHeaders`). If delivery fails, the call operation fails — there is no retry. +**`sendCallRelay(targetPeerOrigin, events, opts?)`** (`federationOutbox.ts`): + +- Latency-sensitive: returns `CallRelayResult = { ok: true } | { ok: false; reason: CallRelayFailureReason; error: string }`. +- Peering resolution: + 1. If the peer row is `active` or `unreachable`, POST directly (the health check restores `unreachable` peers; re-handshaking is wasteful). + 2. Otherwise race `ensurePeered` against `opts.peeringTimeoutMs` (default `CALL_PEERING_TIMEOUT_MS = 3_000` ms). The background handshake is **not** aborted on race loss — a warn-logged catch is attached so a late-rejecting background promise does not emit `unhandledRejection`. +- Peer-state → reason mapping is exhaustive over the `EnsurePeeredResult` union (`active` / `rejected` / `pending` / `failed`) plus the external `timeout` branch. TypeScript `never` check in the switch default catches future additions. +- Non-blocking mode: `peeringTimeoutMs: 0` (used by typing) skips the POST for non-active peers, kicks off `ensurePeered` as a background warm-up, returns `peer_transient_failure` silently. + +**`sendTypingRelay(dmChannelId, eventType, userId)`**: + +- Fire-and-forget to each remote DM participant's home instance via `sendCallRelay(origin, [event], { peeringTimeoutMs: 0 })`. Typing is an ephemeral hint — lost packets are acceptable and there is no user-facing failure surface. + +**Call-start failure surfacing.** `sendFederatedCallStart` aggregates targeted-peer results and emits `dm_call_undeliverable` to the caller for failed targeted peers. See `docs/systems/voice.md` and `docs/systems/websocket.md` for the event contract. ### Call Flows @@ -1157,4 +1170,8 @@ If the `federation_mutation_log` table exists but is empty, populates it with `c ## Known Issues -No critical known issues. See `docs/federation-production-roadmap.md` for open items (FED-001 through FED-013). +See `docs/federation-production-roadmap.md` for open items (FED-001 through FED-013). + +- **Accept-relay failure dead end.** If Bob on B accepts a call from Alice on A and the `dm_call_accept` S2S relay back to A fails, Alice's client does not exit the `outgoingCall` state until the 60 s ring timeout fires `dm_call_ended`. The client clears `outgoingCall` only on `dm_call_accepted | rejected | ended` (`useWebSocket.ts`), with no LiveKit participant-join fallback. Surfacing this requires a B-side event and call-state rollback; deferred. +- **End-relay failure dead end.** Similar mechanism, lower severity because LiveKit `ParticipantDisconnected` typically unwinds the voice UI on the host side; local DM call state still lingers to the 60 s timeout. Deferred. +- **Path-B reject-relay failure dead end.** Third-instance user (Carol on C) rings for a call hosted on A via Path B; if her `dm_call_reject` C→A relay fails, A never deducts her from the pending ringees, so her name persists in the caller's "still ringing" set until the 60 s timeout. Same class as accept-failure; deferred. diff --git a/docs/systems/voice.md b/docs/systems/voice.md index f9fe3417..155171b0 100644 --- a/docs/systems/voice.md +++ b/docs/systems/voice.md @@ -56,7 +56,13 @@ DM calls work across federated instances. The caller's instance hosts the LiveKi ### Universal Relay -All `dm_call_*` signaling events (`start`, `accept`, `reject`, `end`) are relayed to every active federation peer in parallel via `Promise.all`. Each `sendCallRelay` call has a 10-second timeout. This is a synchronous HTTP POST to the peer's federation endpoint — it bypasses the outbox worker entirely because call signaling is latency-sensitive and must not be queued. +All `dm_call_*` signaling events (`start`, `accept`, `reject`, `end`) are relayed to every active federation peer in parallel. Each `sendCallRelay` call has a 10-second HTTP timeout. This bypasses the outbox worker — call signaling is latency-sensitive. + +**Auto-peering at send time.** If the target origin has no active peer record, `sendCallRelay` races an `ensurePeered` handshake against a 3 s deadline (`CALL_PEERING_TIMEOUT_MS`). On success the relay POSTs normally; on timeout it returns `peer_transient_failure` without aborting the background handshake, so a subsequent attempt typically succeeds. Typing (`sendTypingRelay`) passes `peeringTimeoutMs: 0` — the POST is skipped for non-active peers and a warm-up `ensurePeered` runs in the background. + +**Call-start failure surface.** `sendFederatedCallStart` aggregates results from its targeted-peer relays (peers whose origin matches a DM member's `homeInstance`) and emits a single `dm_call_undeliverable` event to the caller when any targeted peer fails. `terminal: true` (no successful targeted relay AND no connected local non-caller member) also destroys the local ring room so the caller's UI clears immediately; `terminal: false` leaves the call ringing for reachable recipients. The all-peers broadcast (Path-B fallback to non-member-hosting peers) logs failures only — they are not surfaced. + +Accept and end relay failures are NOT surfaced today; see the federation doc's "Known issues" section for the deferred accept-failure dead-end. ### Dual-Path Processing diff --git a/docs/systems/websocket.md b/docs/systems/websocket.md index 5a51a7bf..9f160208 100644 --- a/docs/systems/websocket.md +++ b/docs/systems/websocket.md @@ -170,6 +170,7 @@ reason: `'displaced'` (new tab) | `'session_closed'` | `dm_call_accepted` | dmChannelId?, federatedCallId? | DM members | | `dm_call_rejected` | dmChannelId?, federatedCallId? | DM members | | `dm_call_ended` | dmChannelId?, federatedCallId? | DM members | +| `dm_call_undeliverable` | Sent to caller when a call-start relay to one or more targeted peers fails. `failures[]` enumerates each failed peer with a `reason` (`peer_rejected` / `peer_awaiting_approval` / `peer_transient_failure` / `livekit_unavailable`). `terminal: true` means the local ring was destroyed; `false` means the call continues for reachable recipients. | caller only | ### Social | type | fields | scope | From 9cdc5921d9e56695a426e7bdfaf6e4e126ce7e83 Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Tue, 21 Apr 2026 14:08:31 +0200 Subject: [PATCH 10/10] docs: clarify that livekit_unavailable is emitted from sendFederatedCallStart, not sendCallRelay --- docs/systems/federation.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/systems/federation.md b/docs/systems/federation.md index 5e86cea6..9a91b57f 100644 --- a/docs/systems/federation.md +++ b/docs/systems/federation.md @@ -1030,7 +1030,7 @@ All events carry standard relay fields: `eventType`, `messageId`, `encryptionVer - Peering resolution: 1. If the peer row is `active` or `unreachable`, POST directly (the health check restores `unreachable` peers; re-handshaking is wasteful). 2. Otherwise race `ensurePeered` against `opts.peeringTimeoutMs` (default `CALL_PEERING_TIMEOUT_MS = 3_000` ms). The background handshake is **not** aborted on race loss — a warn-logged catch is attached so a late-rejecting background promise does not emit `unhandledRejection`. -- Peer-state → reason mapping is exhaustive over the `EnsurePeeredResult` union (`active` / `rejected` / `pending` / `failed`) plus the external `timeout` branch. TypeScript `never` check in the switch default catches future additions. +- Peer-state → reason mapping is exhaustive over the `EnsurePeeredResult` union (`active` / `rejected` / `pending` / `failed`) plus the external `timeout` branch. TypeScript `never` check in the switch default catches future additions. Note: the `livekit_unavailable` reason in `DmCallUndeliverableReason` is emitted separately from `sendFederatedCallStart`'s LiveKit pre-flight in `ws/events.ts`, not from this switch — `sendCallRelay` only produces `CallRelayFailureReason` values (`peer_rejected` / `peer_awaiting_approval` / `peer_transient_failure` / `post_failed`). - Non-blocking mode: `peeringTimeoutMs: 0` (used by typing) skips the POST for non-active peers, kicks off `ensurePeered` as a background warm-up, returns `peer_transient_failure` silently. **`sendTypingRelay(dmChannelId, eventType, userId)`**: