diff --git a/fluxer_api/src/api/Config.ts b/fluxer_api/src/api/Config.ts index 96726807e..817ffcf4e 100644 --- a/fluxer_api/src/api/Config.ts +++ b/fluxer_api/src/api/Config.ts @@ -1,7 +1,7 @@ // SPDX-License-Identifier: AGPL-3.0-or-later -import {resolveDownloadsProvider} from '@fluxer/config/src/S3DownloadsProvider'; import type {MasterConfig} from '@fluxer/config/src/MasterConfig'; +import {resolveDownloadsProvider} from '@fluxer/config/src/S3DownloadsProvider'; import {parseIpAddress} from '@fluxer/ip_utils/src/IpAddress'; import {parseGeoipSourceConfig, resolveGeoipRuntimeSourceConfig} from '@pkgs/geoip/src/GeoipStartup'; import type {APIConfig, BlueskyOAuthConfig} from './config/APIConfig'; @@ -463,6 +463,14 @@ export function buildAPIConfigFromMaster(master: MasterConfig): APIConfig { taskName: apiWorkerConfig?.task as WorkerTaskName | undefined, enableCronScheduler: apiWorkerConfig?.enable_cron_scheduler, enableVoiceReconciliation: apiWorkerConfig?.enable_voice_reconciliation ?? true, + voiceReconciliation: { + intervalMs: apiWorkerConfig?.voice_reconciliation?.interval_ms, + staggerDelayMs: apiWorkerConfig?.voice_reconciliation?.stagger_delay_ms, + lockTtlSeconds: apiWorkerConfig?.voice_reconciliation?.lock_ttl_seconds, + cadenceTtlSeconds: apiWorkerConfig?.voice_reconciliation?.cadence_ttl_seconds, + gatewayOnlyGraceMs: apiWorkerConfig?.voice_reconciliation?.gateway_only_grace_ms, + liveKitOnlyGraceMs: apiWorkerConfig?.voice_reconciliation?.livekit_only_grace_ms, + }, laneConcurrencyOverrides: { realtime: apiWorkerConfig?.lane_concurrency_overrides?.realtime, unfurl: apiWorkerConfig?.lane_concurrency_overrides?.unfurl, diff --git a/fluxer_api/src/api/config/APIConfig.ts b/fluxer_api/src/api/config/APIConfig.ts index 461cc5392..9f028aecf 100644 --- a/fluxer_api/src/api/config/APIConfig.ts +++ b/fluxer_api/src/api/config/APIConfig.ts @@ -361,6 +361,14 @@ export interface APIConfig { taskName?: WorkerTaskName; enableCronScheduler?: boolean; enableVoiceReconciliation: boolean; + voiceReconciliation: { + intervalMs: number | undefined; + staggerDelayMs: number | undefined; + lockTtlSeconds: number | undefined; + cadenceTtlSeconds: number | undefined; + gatewayOnlyGraceMs: number | undefined; + liveKitOnlyGraceMs: number | undefined; + }; laneConcurrencyOverrides: { realtime?: number; unfurl?: number; diff --git a/fluxer_api/src/api/voice/VoiceReconciliationWorker.test.ts b/fluxer_api/src/api/voice/VoiceReconciliationWorker.test.ts new file mode 100644 index 000000000..214136e47 --- /dev/null +++ b/fluxer_api/src/api/voice/VoiceReconciliationWorker.test.ts @@ -0,0 +1,40 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {describe, expect, it} from 'vitest'; +import {candidateTtlSecondsFor} from './VoiceReconciliationWorker'; + +const INTERVAL_MS = 15000; +const GATEWAY_ONLY_GRACE_MS = 10000; + +function ttlFor(observedSweepSpacingMs: number): number { + return candidateTtlSecondsFor({ + intervalMs: INTERVAL_MS, + observedSweepSpacingMs, + graceMs: GATEWAY_ONLY_GRACE_MS, + }); +} + +describe('candidateTtlSecondsFor', () => { + it('outlives the gap between two consecutive observations of the same key', () => { + for (const observedSweepSpacingMs of [0, 45_000, 136_000, 300_000, 596_000, 900_000]) { + expect(ttlFor(observedSweepSpacingMs) * 1000).toBeGreaterThan(observedSweepSpacingMs); + } + }); + + it('outlives a sweep gap far longer than the tick interval', () => { + expect(ttlFor(596_000) * 1000).toBeGreaterThan(596_000); + }); + + it('grows with the observed sweep spacing rather than the tick interval', () => { + expect(ttlFor(596_000)).toBeGreaterThan(ttlFor(136_000)); + expect(ttlFor(136_000)).toBeGreaterThan(ttlFor(0)); + }); + + it('keeps a floor that survives a single long sweep before any spacing is observed', () => { + expect(ttlFor(0)).toBeGreaterThanOrEqual(300); + }); + + it('stays bounded so a stale candidate cannot outlive its connection indefinitely', () => { + expect(ttlFor(Number.MAX_SAFE_INTEGER)).toBeLessThanOrEqual(3600); + }); +}); diff --git a/fluxer_api/src/api/voice/VoiceReconciliationWorker.ts b/fluxer_api/src/api/voice/VoiceReconciliationWorker.ts index bde8cbcf5..a3e738170 100644 --- a/fluxer_api/src/api/voice/VoiceReconciliationWorker.ts +++ b/fluxer_api/src/api/voice/VoiceReconciliationWorker.ts @@ -96,9 +96,24 @@ const DEFAULT_STAGGER_DELAY_MS = 25; const DEFAULT_LOCK_TTL_SECONDS = 180; const DEFAULT_GATEWAY_ONLY_GRACE_MS = 10000; const DEFAULT_LIVEKIT_ONLY_GRACE_MS = 60000; +const MIN_CANDIDATE_TTL_SECONDS = 300; +const MAX_CANDIDATE_TTL_SECONDS = 3600; +const CANDIDATE_TTL_SWEEP_MULTIPLIER = 3; +const LAST_SWEEP_KEY_TTL_SECONDS = 86400; const ROOM_KEY_PREFIX = 'voice:room:server:'; +export function candidateTtlSecondsFor(input: { + intervalMs: number; + observedSweepSpacingMs: number; + graceMs: number; +}): number { + const spacingMs = Math.max(input.intervalMs, input.observedSweepSpacingMs); + const ttlSeconds = Math.ceil((spacingMs * CANDIDATE_TTL_SWEEP_MULTIPLIER + input.graceMs * 2) / 1000); + return Math.min(MAX_CANDIDATE_TTL_SECONDS, Math.max(MIN_CANDIDATE_TTL_SECONDS, ttlSeconds)); +} + const VOICE_RECONCILIATION_LOCK_KEY = 'voice:reconcile:lock'; const VOICE_RECONCILIATION_CADENCE_KEY = 'voice:reconcile:cadence'; +const VOICE_RECONCILIATION_LAST_SWEEP_KEY = 'voice:reconcile:last-sweep-at'; const GATEWAY_ONLY_CANDIDATE_KEY_PREFIX = 'voice:reconcile:gateway-only:'; const LIVEKIT_ONLY_CANDIDATE_KEY_PREFIX = 'voice:reconcile:livekit-only:'; @@ -115,8 +130,7 @@ export class VoiceReconciliationWorker { private readonly cadenceTtlSeconds: number; private readonly gatewayOnlyGraceMs: number; private readonly liveKitOnlyGraceMs: number; - private readonly gatewayOnlyCandidateTtlSeconds: number; - private readonly liveKitOnlyCandidateTtlSeconds: number; + private observedSweepSpacingMs = 0; private intervalHandle: NodeJS.Timeout | null = null; private reconciling = false; private reconciliationLockLost = false; @@ -136,14 +150,6 @@ export class VoiceReconciliationWorker { this.cadenceTtlSeconds = options.cadenceTtlSeconds ?? Math.max(1, Math.ceil((this.intervalMs * 3) / 1000)); this.gatewayOnlyGraceMs = options.gatewayOnlyGraceMs ?? DEFAULT_GATEWAY_ONLY_GRACE_MS; this.liveKitOnlyGraceMs = options.liveKitOnlyGraceMs ?? DEFAULT_LIVEKIT_ONLY_GRACE_MS; - this.gatewayOnlyCandidateTtlSeconds = Math.max( - 60, - Math.ceil((this.intervalMs * 4 + this.gatewayOnlyGraceMs * 4) / 1000), - ); - this.liveKitOnlyCandidateTtlSeconds = Math.max( - 60, - Math.ceil((this.intervalMs * 4 + this.liveKitOnlyGraceMs * 4) / 1000), - ); } start(): void { @@ -156,6 +162,7 @@ export class VoiceReconciliationWorker { intervalMs: this.intervalMs, gatewayOnlyGraceMs: this.gatewayOnlyGraceMs, liveKitOnlyGraceMs: this.liveKitOnlyGraceMs, + gatewayOnlyCandidateTtlSeconds: this.candidateTtlSeconds(this.gatewayOnlyGraceMs), }, 'Starting VoiceReconciliationWorker', ); @@ -175,6 +182,7 @@ export class VoiceReconciliationWorker { async reconcile(): Promise { const startTime = Date.now(); + await this.recordSweepSpacing(startTime); const discovery = await this.discoverActiveRooms(); this.logger.info( { @@ -233,6 +241,8 @@ export class VoiceReconciliationWorker { const durationMs = Date.now() - startTime; this.logger.info( { + observedSweepSpacingMs: this.observedSweepSpacingMs, + gatewayOnlyCandidateTtlSeconds: this.candidateTtlSeconds(this.gatewayOnlyGraceMs), roomsChecked, totalConfirmed, totalRepaired, @@ -300,6 +310,37 @@ export class VoiceReconciliationWorker { } } + private candidateTtlSeconds(graceMs: number): number { + return candidateTtlSecondsFor({ + intervalMs: this.intervalMs, + observedSweepSpacingMs: this.observedSweepSpacingMs, + graceMs, + }); + } + + private async recordSweepSpacing(startedAt: number): Promise { + try { + const previous = await this.kvClient.get(VOICE_RECONCILIATION_LAST_SWEEP_KEY); + const previousAt = previous === null ? Number.NaN : Number(previous); + if (Number.isFinite(previousAt) && startedAt > previousAt) { + this.observedSweepSpacingMs = Math.max(this.observedSweepSpacingMs, startedAt - previousAt); + const gatewayOnlyCandidateTtlSeconds = this.candidateTtlSeconds(this.gatewayOnlyGraceMs); + if (this.observedSweepSpacingMs >= gatewayOnlyCandidateTtlSeconds * 1000) { + this.logger.warn( + { + observedSweepSpacingMs: this.observedSweepSpacingMs, + gatewayOnlyCandidateTtlSeconds, + }, + 'Reconciliation sweeps are further apart than the candidate TTL; divergent voice states will be deferred forever', + ); + } + } + await this.kvClient.setex(VOICE_RECONCILIATION_LAST_SWEEP_KEY, LAST_SWEEP_KEY_TTL_SECONDS, String(startedAt)); + } catch (error) { + this.logger.warn({error}, 'Failed to record reconciliation sweep spacing'); + } + } + private async acquireCadenceLease(): Promise { try { return await this.kvClient.setnx(VOICE_RECONCILIATION_CADENCE_KEY, '1', this.cadenceTtlSeconds); @@ -993,7 +1034,7 @@ export class VoiceReconciliationWorker { } await this.kvClient.setex( key, - this.gatewayOnlyCandidateTtlSeconds, + this.candidateTtlSeconds(this.gatewayOnlyGraceMs), Number.isFinite(firstSeen) ? String(firstSeen) : String(now), ); return false; @@ -1024,7 +1065,7 @@ export class VoiceReconciliationWorker { } await this.kvClient.setex( key, - this.liveKitOnlyCandidateTtlSeconds, + this.candidateTtlSeconds(this.liveKitOnlyGraceMs), Number.isFinite(firstSeen) ? String(firstSeen) : String(now), ); return false; diff --git a/fluxer_api/src/api/worker/WorkerDependencies.ts b/fluxer_api/src/api/worker/WorkerDependencies.ts index 900e98e7d..f30ca8caf 100644 --- a/fluxer_api/src/api/worker/WorkerDependencies.ts +++ b/fluxer_api/src/api/worker/WorkerDependencies.ts @@ -243,6 +243,12 @@ export async function initializeWorkerDependencies(snowflakeService: ISnowflakeS voiceRoomStore, kvClient, logger: Logger, + intervalMs: Config.worker.voiceReconciliation.intervalMs, + staggerDelayMs: Config.worker.voiceReconciliation.staggerDelayMs, + lockTtlSeconds: Config.worker.voiceReconciliation.lockTtlSeconds, + cadenceTtlSeconds: Config.worker.voiceReconciliation.cadenceTtlSeconds, + gatewayOnlyGraceMs: Config.worker.voiceReconciliation.gatewayOnlyGraceMs, + liveKitOnlyGraceMs: Config.worker.voiceReconciliation.liveKitOnlyGraceMs, }) : null; if (Config.voice.enabled && voiceTopology !== null) { diff --git a/packages/config/src/MasterConfig.ts b/packages/config/src/MasterConfig.ts index 872fab115..c25778de8 100644 --- a/packages/config/src/MasterConfig.ts +++ b/packages/config/src/MasterConfig.ts @@ -112,6 +112,14 @@ export interface MasterConfig { task?: string; enable_cron_scheduler?: boolean; enable_voice_reconciliation?: boolean; + voice_reconciliation?: { + interval_ms?: number; + stagger_delay_ms?: number; + lock_ttl_seconds?: number; + cadence_ttl_seconds?: number; + gateway_only_grace_ms?: number; + livekit_only_grace_ms?: number; + }; lane_concurrency_overrides?: { realtime?: number; unfurl?: number; diff --git a/packages/config/src/config_loader/EnvironmentOverrides.ts b/packages/config/src/config_loader/EnvironmentOverrides.ts index eadb90753..f4f477c2e 100644 --- a/packages/config/src/config_loader/EnvironmentOverrides.ts +++ b/packages/config/src/config_loader/EnvironmentOverrides.ts @@ -96,6 +96,30 @@ const NAMED_FLUXER_ENV_OVERRIDES: Record = { path: ['services', 'api', 'worker', 'enable_voice_reconciliation'], parse: parseEnvValue, }, + FLUXER_API_WORKER_VOICE_RECONCILIATION_INTERVAL_MS: { + path: ['services', 'api', 'worker', 'voice_reconciliation', 'interval_ms'], + parse: parseEnvValue, + }, + FLUXER_API_WORKER_VOICE_RECONCILIATION_STAGGER_DELAY_MS: { + path: ['services', 'api', 'worker', 'voice_reconciliation', 'stagger_delay_ms'], + parse: parseEnvValue, + }, + FLUXER_API_WORKER_VOICE_RECONCILIATION_LOCK_TTL_SECONDS: { + path: ['services', 'api', 'worker', 'voice_reconciliation', 'lock_ttl_seconds'], + parse: parseEnvValue, + }, + FLUXER_API_WORKER_VOICE_RECONCILIATION_CADENCE_TTL_SECONDS: { + path: ['services', 'api', 'worker', 'voice_reconciliation', 'cadence_ttl_seconds'], + parse: parseEnvValue, + }, + FLUXER_API_WORKER_VOICE_RECONCILIATION_GATEWAY_ONLY_GRACE_MS: { + path: ['services', 'api', 'worker', 'voice_reconciliation', 'gateway_only_grace_ms'], + parse: parseEnvValue, + }, + FLUXER_API_WORKER_VOICE_RECONCILIATION_LIVEKIT_ONLY_GRACE_MS: { + path: ['services', 'api', 'worker', 'voice_reconciliation', 'livekit_only_grace_ms'], + parse: parseEnvValue, + }, FLUXER_API_WORKER_LANE_CONCURRENCY_OVERRIDES: { path: ['services', 'api', 'worker', 'lane_concurrency_overrides'], parse: parseEnvValue,