Throttle live (codesandbox#3722)

* throttle live based on connections

* ops, forgot to change back

* modified

* refactor throttling

* another iteration

* presence and blocking instead, not working though

* fix presence

* fix typo

* open up when saving

* fix bug

* ensure revision when ack

* add as experiment

* reset files with operation error

* properly propagate errors

* do not force experiment :D

* final thing

* rename stuff

* comments

* use 2s throttling

* do not wait to be synchronized

* wait for operation flush also on save clicked

* fix comments from ives

* even smoother

Author: christianalfoni

packages/app/src/app/overmind/effects/live/clients.ts

@@ -1,6 +1,6 @@
-import { TextOperation } from 'ot';
 import { logBreadcrumb } from '@codesandbox/common/lib/utils/analytics/sentry';
 import { Blocker, blocker } from 'app/utils/blocker';
+import { TextOperation } from 'ot';
 
 import { OTClient, synchronized_ } from './ot/client';
 

packages/app/src/app/overmind/effects/live/index.ts

@@ -12,7 +12,7 @@ import {
 import _debug from '@codesandbox/common/lib/utils/debug';
 import { Blocker, blocker } from 'app/utils/blocker';
 import { camelizeKeys } from 'humps';
-import { TextOperation, SerializedTextOperation } from 'ot';
+import { SerializedTextOperation, TextOperation } from 'ot';
 import { Channel, Presence, Socket } from 'phoenix';
 import uuid from 'uuid';
 
@@ -45,13 +45,16 @@ declare global {
   }
 }
 
+const TIME_TO_THROTTLE_SOLO_MODE_SENDS = 2000;
+
 class Live {
   private identifier = uuid.v4();
   private pendingMessages = new Map();
   private debug = _debug('cs:socket');
   private channel: Channel | null;
   private messageIndex = 0;
   private clients: ReturnType<typeof clientsFactory>;
+  private awaitSendTimer: number;
   private socket: Socket;
   /*
     Since in "Solo mode" we want to batch up operations and other events later,
@@ -86,15 +89,32 @@ class Live {
   private connectionsCount = 0;
   private setAwaitSend() {
     this.awaitSend = blocker();
+    clearTimeout(this.awaitSendTimer);
+    this.awaitSendTimer = window.setTimeout(async () => {
+      if (this.connectionsCount === 1) {
+        // We await the currently resolved blocker before setting it back,
+        // so that messages gets through
+        await this.resolveAwaitSend();
+        this.setAwaitSend();
+      }
+    }, TIME_TO_THROTTLE_SOLO_MODE_SENDS);
   }
 
   private resolveAwaitSend() {
     if (!this.awaitSend) {
-      return;
+      return Promise.resolve();
     }
     const awaitSend = this.awaitSend;
     this.awaitSend = null;
     awaitSend.resolve();
+    return awaitSend.promise;
+  }
+
+  private async awaitSynchronizedModule(moduleShortid: string) {
+    const client = this.clients.get(moduleShortid);
+    if (client.awaitSynchronized) {
+      await client.awaitSynchronized.promise;
+    }
   }
 
   private onSendOperation = async (
@@ -168,7 +188,7 @@ class Live {
 
           const waitTime = 500 + 5000 * Math.random();
 
-          setTimeout(() => {
+          window.setTimeout(() => {
             this.socket.connect();
           }, waitTime);
         }
@@ -224,9 +244,9 @@ class Live {
           const currentCount = this.connectionsCount;
 
           this.connectionsCount = this.presence.list().length;
-          if (currentCount >= 2 && this.connectionsCount < 2) {
+          if (currentCount !== 1 && this.connectionsCount === 1) {
             this.setAwaitSend();
-          } else if (currentCount < 2 && this.connectionsCount >= 2) {
+          } else if (currentCount === 1 && this.connectionsCount > 1) {
             this.resolveAwaitSend();
           }
         });
@@ -315,13 +335,10 @@ class Live {
     */
     if (this.isLiveBlockerExperiement() && this.awaitSend) {
       this.resolveAwaitSend();
-      const client = this.clients.get(module.shortid);
-      if (client.awaitSynchronized) {
-        await client.awaitSynchronized.promise;
-      }
+      this.awaitSynchronizedModule(module.shortid);
       this.setAwaitSend();
-      // Send the save message
     }
+    // Send the save message
   }
 
   sendModuleUpdate(module: Module) {
@@ -485,6 +502,12 @@ class Live {
     return this.clients.getAll();
   }
 
+  applyClient(moduleShortid: string, operation: SerializedTextOperation) {
+    return this.clients
+      .get(moduleShortid)
+      .applyClient(TextOperation.fromJSON(operation));
+  }
+
   applyServer(moduleShortid: string, operation: SerializedTextOperation) {
     return this.clients
       .get(moduleShortid)

packages/app/src/app/overmind/namespaces/editor/actions.ts

@@ -370,6 +370,10 @@ export const saveClicked: AsyncAction = withOwnedSandbox(
         state.editor.changedModuleShortids.includes(module.shortid)
       );
 
+      await Promise.all(
+        changedModules.map(module => effects.live.saveModule(module))
+      );
+
       const updatedModules = await effects.api.saveModules(
         sandbox.id,
         changedModules