fix(api): derive voice reconciliation candidate ttl from real sweep spacing (#1810)

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,

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;

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);
+	});
+});

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<void> {
 		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<void> {
+		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<boolean> {
 		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;

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) {

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;

packages/config/src/config_loader/EnvironmentOverrides.ts

@@ -96,6 +96,30 @@ const NAMED_FLUXER_ENV_OVERRIDES: Record<string, NamedEnvOverride> = {
 		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,