chore(web): refactor websocket state machine

4 files changed, 271 insertions, 154 deletions

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<void>;
+
+type PushHandler = (
+  resourceId: string,
+  stateId: string,
+  service: string,
+  data: unknown
+) => Promise<void>;
+
+enum State {
+  Initial,
+  Opening,
+  TabVisible,
+  TabHiddenIdle,
+  TabHiddenWaiting,
+  Error,
+  TimedOut,
+}
 
 export class XtextWebSocketClient {
   nextMessageId = 0;
 
-  closing = false;
-
   connection!: WebSocket;
 
   pendingRequests = new Map<string, PendingRequest>();
@@ -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<unknown> {
     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;
+    }
+  }
+}