From 3e79530389b30f68328a53b25c377eca24738fb7 Mon Sep 17 00:00:00 2001 From: Hampus Date: Sat, 12 Sep 2026 01:36:04 +0200 Subject: [PATCH] fix(app): defer non-critical gateway dispatches (#2715) --- .../GatewayDispatchScheduling.test.ts | 71 ++++++++++++++++ .../transport/GatewayDispatchScheduling.ts | 42 +++++++++ .../gateway/transport/GatewaySocket.ts | 85 +++++++++++++++++-- 3 files changed, 192 insertions(+), 6 deletions(-) create mode 100644 fluxer_app/src/features/gateway/transport/GatewayDispatchScheduling.test.ts create mode 100644 fluxer_app/src/features/gateway/transport/GatewayDispatchScheduling.ts diff --git a/fluxer_app/src/features/gateway/transport/GatewayDispatchScheduling.test.ts b/fluxer_app/src/features/gateway/transport/GatewayDispatchScheduling.test.ts new file mode 100644 index 000000000..cecd6ea3e --- /dev/null +++ b/fluxer_app/src/features/gateway/transport/GatewayDispatchScheduling.test.ts @@ -0,0 +1,71 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import { + DISPATCH_IDLE_MIN_BUDGET_MS, + isCriticalGatewayDispatch, + selectGatewayDispatchFlushMode, + shouldRetryIdleWait, + shouldSkipIdleWait, +} from '@app/features/gateway/transport/GatewayDispatchScheduling'; +import {describe, expect, it} from 'vitest'; + +describe('selectGatewayDispatchFlushMode', () => { + it('flushes transport events that carry no dispatch type immediately', () => { + expect(selectGatewayDispatchFlushMode(null)).toBe('immediate'); + }); + + it('flushes the session-establishing dispatches immediately', () => { + expect(selectGatewayDispatchFlushMode('READY')).toBe('immediate'); + expect(selectGatewayDispatchFlushMode('RESUMED')).toBe('immediate'); + }); + + it('defers ordinary dispatches so an in-flight REST page commits first', () => { + expect(selectGatewayDispatchFlushMode('MESSAGE_CREATE')).toBe('deferred'); + expect(selectGatewayDispatchFlushMode('TYPING_START')).toBe('deferred'); + expect(selectGatewayDispatchFlushMode('PRESENCE_UPDATE')).toBe('deferred'); + }); +}); + +describe('isCriticalGatewayDispatch', () => { + it('treats voice and call dispatches as critical', () => { + expect(isCriticalGatewayDispatch('VOICE_STATE_UPDATE')).toBe(true); + expect(isCriticalGatewayDispatch('VOICE_SERVER_UPDATE')).toBe(true); + expect(isCriticalGatewayDispatch('CALL_CREATE')).toBe(true); + expect(isCriticalGatewayDispatch('SESSIONS_REPLACE')).toBe(true); + }); + + it('does not treat ordinary dispatches as critical', () => { + expect(isCriticalGatewayDispatch('MESSAGE_CREATE')).toBe(false); + expect(isCriticalGatewayDispatch('TYPING_START')).toBe(false); + expect(isCriticalGatewayDispatch(null)).toBe(false); + }); +}); + +describe('shouldSkipIdleWait', () => { + it('skips the idle wait when critical work is queued', () => { + expect(shouldSkipIdleWait(true, true)).toBe(true); + }); + + it('skips the idle wait when the environment has no idle callback', () => { + expect(shouldSkipIdleWait(false, false)).toBe(true); + }); + + it('waits for idle for ordinary work in a capable environment', () => { + expect(shouldSkipIdleWait(false, true)).toBe(false); + }); +}); + +describe('shouldRetryIdleWait', () => { + it('does not retry once the idle callback has timed out', () => { + expect(shouldRetryIdleWait(true, 0)).toBe(false); + }); + + it('retries when the granted budget is below the minimum', () => { + expect(shouldRetryIdleWait(false, DISPATCH_IDLE_MIN_BUDGET_MS / 2)).toBe(true); + }); + + it('flushes when the granted budget is sufficient', () => { + expect(shouldRetryIdleWait(false, DISPATCH_IDLE_MIN_BUDGET_MS)).toBe(false); + expect(shouldRetryIdleWait(false, 50)).toBe(false); + }); +}); diff --git a/fluxer_app/src/features/gateway/transport/GatewayDispatchScheduling.ts b/fluxer_app/src/features/gateway/transport/GatewayDispatchScheduling.ts new file mode 100644 index 000000000..34d508de7 --- /dev/null +++ b/fluxer_app/src/features/gateway/transport/GatewayDispatchScheduling.ts @@ -0,0 +1,42 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +export const DISPATCH_FLUSH_DELAY_MS = 50; +export const DISPATCH_IDLE_TIMEOUT_MS = 1000; +export const DISPATCH_IDLE_RETRY_TIMEOUT_MS = 200; +export const DISPATCH_IDLE_MIN_BUDGET_MS = 1000 / 60 / 8; + +const IMMEDIATE_DISPATCH_TYPES = new Set(['READY', 'RESUMED']); + +const CRITICAL_DISPATCH_TYPES = new Set([ + 'READY', + 'RESUMED', + 'VOICE_STATE_UPDATE', + 'VOICE_SERVER_UPDATE', + 'CALL_CREATE', + 'CALL_UPDATE', + 'CALL_DELETE', + 'SESSIONS_REPLACE', + 'STREAM_CREATE', + 'STREAM_UPDATE', + 'STREAM_DELETE', + 'STREAM_SERVER_UPDATE', +]); + +export type GatewayDispatchFlushMode = 'immediate' | 'deferred'; + +export function selectGatewayDispatchFlushMode(dispatchType: string | null): GatewayDispatchFlushMode { + if (dispatchType == null) return 'immediate'; + return IMMEDIATE_DISPATCH_TYPES.has(dispatchType) ? 'immediate' : 'deferred'; +} + +export function isCriticalGatewayDispatch(dispatchType: string | null): boolean { + return dispatchType != null && CRITICAL_DISPATCH_TYPES.has(dispatchType); +} + +export function shouldSkipIdleWait(criticalWorkScheduled: boolean, idleCallbackSupported: boolean): boolean { + return criticalWorkScheduled || !idleCallbackSupported; +} + +export function shouldRetryIdleWait(didTimeout: boolean, timeRemainingMs: number): boolean { + return !didTimeout && timeRemainingMs < DISPATCH_IDLE_MIN_BUDGET_MS; +} diff --git a/fluxer_app/src/features/gateway/transport/GatewaySocket.ts b/fluxer_app/src/features/gateway/transport/GatewaySocket.ts index 90c545245..5585034ff 100644 --- a/fluxer_app/src/features/gateway/transport/GatewaySocket.ts +++ b/fluxer_app/src/features/gateway/transport/GatewaySocket.ts @@ -9,6 +9,15 @@ import { isGatewayCompressionError, } from '@app/features/gateway/transport/GatewayCompression'; import GatewayConnection from '@app/features/gateway/transport/GatewayConnection'; +import { + DISPATCH_FLUSH_DELAY_MS, + DISPATCH_IDLE_RETRY_TIMEOUT_MS, + DISPATCH_IDLE_TIMEOUT_MS, + isCriticalGatewayDispatch, + selectGatewayDispatchFlushMode, + shouldRetryIdleWait, + shouldSkipIdleWait, +} from '@app/features/gateway/transport/GatewayDispatchScheduling'; import { formatGatewayReadyTimings, type GatewayTimings, @@ -180,6 +189,8 @@ export class GatewaySocket extends EventEmitter { private shouldReconnectImmediately = false; private deferredEmitQueue: Array<() => void> = []; private deferredEmitTimeoutId: number | null = null; + private deferredEmitIdleId: number | null = null; + private criticalWorkScheduled = false; private payloadDecompressor: GatewayCompression | null = null; private compressionFallbackInProgress = false; @@ -200,15 +211,77 @@ export class GatewaySocket extends EventEmitter { this.deferredEmitQueue.push(() => { (this.emit as (event: K, ...args: GatewaySocketEventArgs) => boolean)(event, ...args); }); - if (this.deferredEmitTimeoutId != null) return; + const dispatchType = event === 'dispatch' ? (args[0] as string) : null; + this.scheduleDeferredFlush(dispatchType); + } + + private scheduleDeferredFlush(dispatchType: string | null): void { + if (isCriticalGatewayDispatch(dispatchType)) { + this.criticalWorkScheduled = true; + } + if (selectGatewayDispatchFlushMode(dispatchType) === 'immediate') { + this.clearDeferredFlushWork(); + this.flushDeferredEmits(); + return; + } + if (this.hasDeferredFlushWork()) return; this.deferredEmitTimeoutId = window.setTimeout(() => { this.deferredEmitTimeoutId = null; - const queue = this.deferredEmitQueue; - this.deferredEmitQueue = []; - for (const emitFn of queue) { - emitFn(); + if (typeof document !== 'undefined' && document.visibilityState === 'hidden') { + this.flushDeferredEmits(); + return; } - }, 0); + this.queueDeferredIdleFlush(); + }, DISPATCH_FLUSH_DELAY_MS); + } + + private queueDeferredIdleFlush(): void { + if (shouldSkipIdleWait(this.criticalWorkScheduled, typeof window.requestIdleCallback === 'function')) { + this.flushDeferredEmits(); + return; + } + this.deferredEmitIdleId = window.requestIdleCallback( + (deadline) => { + this.deferredEmitIdleId = null; + if (!shouldRetryIdleWait(deadline.didTimeout, deadline.timeRemaining())) { + this.flushDeferredEmits(); + return; + } + this.deferredEmitIdleId = window.requestIdleCallback( + () => { + this.deferredEmitIdleId = null; + this.flushDeferredEmits(); + }, + {timeout: DISPATCH_IDLE_RETRY_TIMEOUT_MS}, + ); + }, + {timeout: DISPATCH_IDLE_TIMEOUT_MS}, + ); + } + + private hasDeferredFlushWork(): boolean { + return this.deferredEmitTimeoutId != null || this.deferredEmitIdleId != null; + } + + private clearDeferredFlushWork(): void { + if (this.deferredEmitTimeoutId != null) { + clearTimeout(this.deferredEmitTimeoutId); + this.deferredEmitTimeoutId = null; + } + if (this.deferredEmitIdleId != null && typeof window.cancelIdleCallback === 'function') { + window.cancelIdleCallback(this.deferredEmitIdleId); + } + this.deferredEmitIdleId = null; + } + + private flushDeferredEmits(): void { + this.clearDeferredFlushWork(); + this.criticalWorkScheduled = false; + const queue = this.deferredEmitQueue; + this.deferredEmitQueue = []; + for (const emitFn of queue) { + emitFn(); + } } connect(): void {