mirror of
https://github.com/fluxerapp/fluxer
synced 2026-10-18 05:25:40 +09:00
fix(app): defer non-critical gateway dispatches (#2715)
This commit is contained in:
1 parent
d0c84b3d9b
commit
3e79530389
3 files changed
+192
-6
No files matched your search
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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<GatewaySocketEvents> {
|
||||
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<GatewaySocketEvents> {
|
||||
this.deferredEmitQueue.push(() => {
|
||||
(this.emit as (event: K, ...args: GatewaySocketEventArgs<K>) => 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 {
|
||||
|
||||
Reference in new issue
Block a user