From 232fbcafa863a3c28ab907b112c5257f0b6dc8f1 Mon Sep 17 00:00:00 2001 From: Kristóf Marussy Date: Tue, 26 Oct 2021 21:40:36 +0200 Subject: chore(web): refactor websocket state machine --- language-web/src/main/js/editor/PendingRequest.ts | 10 +- language-web/src/main/js/editor/XtextClient.ts | 48 ++- .../src/main/js/editor/XtextWebSocketClient.ts | 334 +++++++++++++-------- language-web/src/main/js/utils/Timer.ts | 33 ++ 4 files changed, 271 insertions(+), 154 deletions(-) create mode 100644 language-web/src/main/js/utils/Timer.ts (limited to 'language-web/src/main/js') diff --git a/language-web/src/main/js/editor/PendingRequest.ts b/language-web/src/main/js/editor/PendingRequest.ts index 784f06ec..49d4c36c 100644 --- a/language-web/src/main/js/editor/PendingRequest.ts +++ b/language-web/src/main/js/editor/PendingRequest.ts @@ -9,16 +9,24 @@ export class PendingRequest { private readonly rejectCallback: (reason?: unknown) => void; + private readonly timeoutCallback: () => void; + private resolved = false; private timeoutId: NodeJS.Timeout; - constructor(resolve: (value: unknown) => void, reject: (reason?: unknown) => void) { + constructor( + resolve: (value: unknown) => void, + reject: (reason?: unknown) => void, + timeout: () => void, + ) { this.resolveCallback = resolve; this.rejectCallback = reject; + this.timeoutCallback = timeout; this.timeoutId = setTimeout(() => { if (!this.resolved) { this.reject(new Error('Request timed out')); + this.timeoutCallback(); } }, REQUEST_TIMEOUT_MS); } diff --git a/language-web/src/main/js/editor/XtextClient.ts b/language-web/src/main/js/editor/XtextClient.ts index 1c6c0ae6..5216154e 100644 --- a/language-web/src/main/js/editor/XtextClient.ts +++ b/language-web/src/main/js/editor/XtextClient.ts @@ -8,6 +8,7 @@ import { nanoid } from 'nanoid'; import type { EditorStore } from './EditorStore'; import { getLogger } from '../logging'; +import { Timer } from '../utils/Timer'; import { isDocumentStateResult, isServiceConflictResult, @@ -36,7 +37,9 @@ export class XtextClient { dirtyChanges: ChangeDesc; - updateTimeout: NodeJS.Timeout | null = null; + updateTimer = new Timer(() => { + this.handleUpdate(); + }, UPDATE_TIMEOUT_MS); store: EditorStore; @@ -46,15 +49,11 @@ export class XtextClient { this.store = store; this.dirtyChanges = this.newEmptyChangeDesc(); this.webSocketClient = new XtextWebSocketClient( - () => { - this.updateFullText().catch((error) => { - log.error('Unexpected error during initial update', error); - }); + async () => { + await this.updateFullText(); }, - (resource, stateId, service, push) => { - this.onPush(resource, stateId, service, push).catch((error) => { - log.error('Unexected error during push message handling', error); - }); + async (resource, stateId, service, push) => { + await this.onPush(resource, stateId, service, push); }, ); } @@ -62,9 +61,8 @@ export class XtextClient { onTransaction(transaction: Transaction): void { const { changes } = transaction; if (!changes.empty) { - this.webSocketClient.ensureOpen(); this.dirtyChanges = this.dirtyChanges.composeDesc(changes.desc); - this.scheduleUpdate(); + this.updateTimer.reschedule(); } } @@ -118,22 +116,16 @@ export class XtextClient { return this.pendingUpdate.composeDesc(this.dirtyChanges); } - private scheduleUpdate() { - if (this.updateTimeout !== null) { - clearTimeout(this.updateTimeout); + private handleUpdate() { + if (!this.webSocketClient.isOpen || this.dirtyChanges.empty) { + return; } - this.updateTimeout = setTimeout(() => { - this.updateTimeout = null; - if (!this.webSocketClient.isOpen || this.dirtyChanges.empty) { - return; - } - if (!this.pendingUpdate) { - this.updateDeltaText().catch((error) => { - log.error('Unexpected error during scheduled update', error); - }); - } - this.scheduleUpdate(); - }, UPDATE_TIMEOUT_MS); + if (!this.pendingUpdate) { + this.updateDeltaText().catch((error) => { + log.error('Unexpected error during scheduled update', error); + }); + } + this.updateTimer.reschedule(); } private newEmptyChangeDesc() { @@ -169,7 +161,7 @@ export class XtextClient { return; } const delta = this.computeDelta(); - log.debug('Editor delta', delta); + log.trace('Editor delta', delta); await this.withUpdate(async () => { const result = await this.webSocketClient.send({ resource: this.resourceName, @@ -231,7 +223,7 @@ export class XtextClient { this.pendingUpdate = null; switch (newStateId) { case UpdateAction.ForceReconnect: - this.webSocketClient.forceReconnectDueToError(); + this.webSocketClient.handleApplicationError(); break; case UpdateAction.FullTextUpdate: await this.updateFullText(); diff --git a/language-web/src/main/js/editor/XtextWebSocketClient.ts b/language-web/src/main/js/editor/XtextWebSocketClient.ts index c034f8c8..6766029b 100644 --- a/language-web/src/main/js/editor/XtextWebSocketClient.ts +++ b/language-web/src/main/js/editor/XtextWebSocketClient.ts @@ -2,6 +2,7 @@ import { nanoid } from 'nanoid'; import { getLogger } from '../logging'; import { PendingRequest } from './PendingRequest'; +import { Timer } from '../utils/Timer'; import { isErrorResponse, isOkResponse, @@ -14,7 +15,11 @@ const XTEXT_SUBPROTOCOL_V1 = 'tools.refinery.language.web.xtext.v1'; const WEBSOCKET_CLOSE_OK = 1000; -const RECONNECT_DELAY_MS = 1000; +const RECONNECT_DELAY_MS = [1000, 5000, 30_000]; + +const MAX_RECONNECT_DELAY_MS = RECONNECT_DELAY_MS[RECONNECT_DELAY_MS.length - 1]; + +const MAX_APP_ERROR_COUNT = 1; const BACKGROUND_IDLE_TIMEOUT_MS = 5 * 60 * 1000; @@ -22,15 +27,28 @@ const PING_TIMEOUT_MS = 10 * 1000; const log = getLogger('XtextWebSocketClient'); -type ReconnectHandler = () => void; - -type PushHandler = (resourceId: string, stateId: string, service: string, data: unknown) => void; +type ReconnectHandler = () => Promise; + +type PushHandler = ( + resourceId: string, + stateId: string, + service: string, + data: unknown +) => Promise; + +enum State { + Initial, + Opening, + TabVisible, + TabHiddenIdle, + TabHiddenWaiting, + Error, + TimedOut, +} export class XtextWebSocketClient { nextMessageId = 0; - closing = false; - connection!: WebSocket; pendingRequests = new Map(); @@ -39,161 +57,162 @@ export class XtextWebSocketClient { onPush: PushHandler; - reconnectTimeout: NodeJS.Timeout | null = null; + state = State.Initial; - idleTimeout: NodeJS.Timeout | null = null; + appErrorCount = 0; - pingTimeout: NodeJS.Timeout | null = null; + reconnectTryCount = 0; + + idleTimer = new Timer(() => { + this.handleIdleTimeout(); + }, BACKGROUND_IDLE_TIMEOUT_MS); + + pingTimer = new Timer(() => { + this.sendPing(); + }, PING_TIMEOUT_MS); + + reconnectTimer = new Timer(() => { + this.handleReconnect(); + }); constructor(onReconnect: ReconnectHandler, onPush: PushHandler) { this.onReconnect = onReconnect; this.onPush = onPush; document.addEventListener('visibilitychange', () => { - this.scheduleIdleTimeout(); + this.handleVisibilityChange(); }); this.reconnect(); } - get isOpen(): boolean { - return this.connection.readyState === WebSocket.OPEN; + private get isLogicallyClosed(): boolean { + return this.state === State.Error || this.state === State.TimedOut; } - get isClosed(): boolean { - return this.connection.readyState === WebSocket.CLOSING - || this.connection.readyState === WebSocket.CLOSED; - } - - ensureOpen(): void { - if (this.isClosed) { - this.closing = false; - this.reconnect(); - } + get isOpen(): boolean { + return this.state === State.TabVisible + || this.state === State.TabHiddenIdle + || this.state === State.TabHiddenWaiting; } private reconnect() { - this.reconnectTimeout = null; + if (this.isOpen || this.state === State.Opening) { + log.error('Trying to reconnect from', this.state); + return; + } + this.state = State.Opening; const webSocketServer = window.origin.replace(/^http/, 'ws'); const webSocketUrl = `${webSocketServer}/xtext-service`; this.connection = new WebSocket(webSocketUrl, XTEXT_SUBPROTOCOL_V1); this.connection.addEventListener('open', () => { if (this.connection.protocol !== XTEXT_SUBPROTOCOL_V1) { log.error('Unknown subprotocol', this.connection.protocol, 'selected by server'); - this.forceReconnectDueToError(); - return; + this.handleProtocolError(); } - log.info('Connected to xtext web services'); - this.scheduleIdleTimeout(); - this.schedulePingTimeout(); - this.onReconnect(); + if (document.visibilityState === 'hidden') { + this.handleTabHidden(); + } else { + this.handleTabVisibleConnected(); + } + log.info('Connected to websocket'); + this.nextMessageId = 0; + this.appErrorCount = 0; + this.reconnectTryCount = 0; + this.pingTimer.schedule(); + this.onReconnect().catch((error) => { + log.error('Unexpected error in onReconnect handler', error); + }); }); this.connection.addEventListener('error', (event) => { log.error('Unexpected websocket error', event); - this.forceReconnectDueToError(); + this.handleProtocolError(); }); this.connection.addEventListener('message', (event) => { this.handleMessage(event.data); }); this.connection.addEventListener('close', (event) => { - if (!this.closing || event.code !== WEBSOCKET_CLOSE_OK) { - log.error('Websocket closed undexpectedly', event.code, event.reason); + if (this.isLogicallyClosed && event.code === WEBSOCKET_CLOSE_OK + && this.pendingRequests.size === 0) { + log.info('Websocket closed'); + return; } - this.cleanupAndMaybeReconnect(); + log.error('Websocket closed unexpectedly', event.code, event.reason); + this.handleProtocolError(); }); } - private scheduleIdleTimeout() { + private handleVisibilityChange() { if (document.visibilityState === 'hidden') { - if (this.idleTimeout !== null) { - return; + if (this.state === State.TabVisible) { + this.handleTabHidden(); } - log.info('Lost visibility, will disconnect in', BACKGROUND_IDLE_TIMEOUT_MS, 'ms'); - this.idleTimeout = setTimeout(() => { - this.idleTimeout = null; - if (!this.isClosed && document.visibilityState === 'hidden') { - log.info('Closing websocket connection due to inactivity'); - this.close(); - } - }, BACKGROUND_IDLE_TIMEOUT_MS); - } else { - log.info('Gained visibility, connection will be kept alive'); - if (this.idleTimeout !== null) { - clearTimeout(this.idleTimeout); - this.idleTimeout = null; - } - this.ensureOpen(); + return; } - } - - private schedulePingTimeout() { - if (this.pingTimeout !== null) { + this.idleTimer.cancel(); + if (this.state === State.TabHiddenIdle || this.state === State.TabHiddenWaiting) { + this.handleTabVisibleConnected(); return; } - this.pingTimeout = setTimeout(() => { - if (this.isClosed) { - return; - } - if (this.isOpen) { - const ping = nanoid(); - log.trace('ping:', ping); - this.pingTimeout = null; - this.send({ - ping, - }).then((result) => { - if (!isPongResult(result) || result.pong !== ping) { - log.error('invalid pong'); - this.forceReconnectDueToError(); - } - log.trace('pong:', ping); - }).catch((error) => { - log.error('ping error', error); - this.forceReconnectDueToError(); - }); - } - this.schedulePingTimeout(); - }, PING_TIMEOUT_MS); + if (this.state === State.TimedOut) { + this.reconnect(); + } } - private cleanupAndMaybeReconnect() { - this.cleanup(); - if (!this.closing) { - this.delayedReconnect(); - } + private handleTabHidden() { + log.trace('Tab became hidden while websocket is connected'); + this.state = State.TabHiddenIdle; + this.idleTimer.schedule(); } - private cleanup() { - this.pendingRequests.forEach((pendingRequest) => { - pendingRequest.reject(new Error('Websocket closed')); - }); - this.pendingRequests.clear(); - if (this.idleTimeout !== null) { - clearTimeout(this.idleTimeout); - this.idleTimeout = null; - } - if (this.pingTimeout !== null) { - clearTimeout(this.pingTimeout); - this.pingTimeout = null; + private handleTabVisibleConnected() { + log.trace('Tab became visible while websocket is connected'); + this.state = State.TabVisible; + } + + private handleIdleTimeout() { + log.trace('Waiting for pending tasks before disconnect'); + if (this.state === State.TabHiddenIdle) { + this.state = State.TabHiddenWaiting; + this.handleWaitingForDisconnect(); } } - private delayedReconnect() { - if (this.reconnectTimeout !== null) { - clearTimeout(this.reconnectTimeout); - this.reconnectTimeout = null; + private handleWaitingForDisconnect() { + if (this.state !== State.TabHiddenWaiting) { + return; } - this.reconnectTimeout = setTimeout(() => { - log.info('Attempting to reconnect websocket'); - this.reconnect(); - }, RECONNECT_DELAY_MS); + const pending = this.pendingRequests.size; + if (pending === 0) { + log.info('Closing idle websocket'); + this.state = State.TimedOut; + this.closeConnection(1000, 'idle timeout'); + return; + } + log.info('Waiting for', pending, 'pending requests before closing websocket'); } - public forceReconnectDueToError(): void { - this.closeConnection(); - this.cleanupAndMaybeReconnect(); + private sendPing() { + if (!this.isOpen) { + return; + } + const ping = nanoid(); + log.trace('Ping', ping); + this.send({ ping }).then((result) => { + if (isPongResult(result) && result.pong === ping) { + log.trace('Pong', ping); + this.pingTimer.schedule(); + } else { + log.error('Invalid pong'); + this.handleProtocolError(); + } + }).catch((error) => { + log.error('Error while waiting for ping', error); + this.handleProtocolError(); + }); } send(request: unknown): Promise { if (!this.isOpen) { - throw new Error('Connection is not open'); + throw new Error('Not open'); } const messageId = this.nextMessageId.toString(16); if (messageId in this.pendingRequests) { @@ -209,16 +228,20 @@ export class XtextWebSocketClient { id: messageId, request, } as IXtextWebRequest); - return new Promise((resolve, reject) => { - this.connection.send(message); - this.pendingRequests.set(messageId, new PendingRequest(resolve, reject)); + const promise = new Promise((resolve, reject) => { + this.pendingRequests.set(messageId, new PendingRequest(resolve, reject, () => { + this.removePendingRequest(messageId); + })); }); + log.trace('Sending message', message); + this.connection.send(message); + return promise; } private handleMessage(messageStr: unknown) { if (typeof messageStr !== 'string') { log.error('Unexpected binary message', messageStr); - this.forceReconnectDueToError(); + this.handleProtocolError(); return; } log.trace('Incoming websocket message', messageStr); @@ -227,7 +250,7 @@ export class XtextWebSocketClient { message = JSON.parse(messageStr); } catch (error) { log.error('Json parse error', error); - this.forceReconnectDueToError(); + this.handleProtocolError(); return; } if (isOkResponse(message)) { @@ -236,21 +259,28 @@ export class XtextWebSocketClient { this.rejectRequest(message.id, new Error(`${message.error} error: ${message.message}`)); if (message.error === 'server') { log.error('Reconnecting due to server error: ', message.message); - this.forceReconnectDueToError(); + this.handleApplicationError(); } } else if (isPushMessage(message)) { - this.onPush(message.resource, message.stateId, message.service, message.push); + this.onPush( + message.resource, + message.stateId, + message.service, + message.push, + ).catch((error) => { + log.error('Unexpected error in onPush handler', error); + }); } else { log.error('Unexpected websocket message', message); - this.forceReconnectDueToError(); + this.handleProtocolError(); } } private resolveRequest(messageId: string, value: unknown) { const pendingRequest = this.pendingRequests.get(messageId); - this.pendingRequests.delete(messageId); if (pendingRequest) { pendingRequest.resolve(value); + this.removePendingRequest(messageId); return; } log.error('Trying to resolve unknown request', messageId, 'with', value); @@ -258,24 +288,78 @@ export class XtextWebSocketClient { private rejectRequest(messageId: string, reason?: unknown) { const pendingRequest = this.pendingRequests.get(messageId); - this.pendingRequests.delete(messageId); if (pendingRequest) { pendingRequest.reject(reason); + this.removePendingRequest(messageId); return; } log.error('Trying to reject unknown request', messageId, 'with', reason); } - private closeConnection() { - if (!this.isClosed) { - log.info('Closing websocket connection'); - this.connection.close(1000, 'end session'); + private removePendingRequest(messageId: string) { + this.pendingRequests.delete(messageId); + this.handleWaitingForDisconnect(); + } + + private handleProtocolError() { + if (this.isLogicallyClosed) { + return; } + this.abortPendingRequests(); + this.closeConnection(1000, 'reconnecting due to protocol error'); + log.error('Reconnecting after delay due to protocol error'); + this.handleErrorState(); } - close(): void { - this.closing = true; - this.closeConnection(); - this.cleanup(); + handleApplicationError(): void { + if (this.isLogicallyClosed) { + return; + } + this.abortPendingRequests(); + this.closeConnection(1000, 'reconnecting due to application error'); + this.appErrorCount += 1; + if (this.appErrorCount <= MAX_APP_ERROR_COUNT) { + log.error('Immediately reconnecting due to application error'); + this.state = State.Initial; + this.reconnect(); + } else { + log.error('Reconnecting after delay due to application error'); + this.handleErrorState(); + } + } + + private abortPendingRequests() { + this.pendingRequests.forEach((request) => { + request.reject(new Error('Websocket disconnect')); + }); + this.pendingRequests.clear(); + } + + private closeConnection(code: number, reason: string) { + this.pingTimer.cancel(); + const { readyState } = this.connection; + if (readyState !== WebSocket.CLOSING && readyState !== WebSocket.CLOSED) { + this.connection.close(code, reason); + } + } + + private handleErrorState() { + this.state = State.Error; + this.reconnectTryCount += 1; + const delay = RECONNECT_DELAY_MS[this.reconnectTryCount - 1] || MAX_RECONNECT_DELAY_MS; + log.info('Reconnecting in', delay, 'ms'); + this.reconnectTimer.schedule(delay); + } + + private handleReconnect() { + if (this.state !== State.Error) { + log.error('Unexpected reconnect in', this.state); + return; + } + if (document.visibilityState === 'hidden') { + this.state = State.TimedOut; + } else { + this.reconnect(); + } } } diff --git a/language-web/src/main/js/utils/Timer.ts b/language-web/src/main/js/utils/Timer.ts new file mode 100644 index 00000000..efde6633 --- /dev/null +++ b/language-web/src/main/js/utils/Timer.ts @@ -0,0 +1,33 @@ +export class Timer { + readonly callback: () => void; + + readonly defaultTimeout: number; + + timeout: NodeJS.Timeout | null = null; + + constructor(callback: () => void, defaultTimeout = 0) { + this.callback = () => { + this.timeout = null; + callback(); + }; + this.defaultTimeout = defaultTimeout; + } + + schedule(timeout: number | null = null): void { + if (this.timeout === null) { + this.timeout = setTimeout(this.callback, timeout || this.defaultTimeout); + } + } + + reschedule(timeout: number | null = null): void { + this.cancel(); + this.schedule(timeout); + } + + cancel(): void { + if (this.timeout !== null) { + clearTimeout(this.timeout); + this.timeout = null; + } + } +} -- cgit v1.2.3-54-g00ecf