Merge pull request #4 from hiveteams/fix/large-collab-ops

Fix/large collab ops

packages/backend/src/AutomergeCollaboration.ts

@@ -7,6 +7,7 @@ import flatten from 'lodash/flatten'
 import { SyncDoc, CollabAction, toJS } from '@hiveteams/collab-bridge'
 import { debugCollabBackend } from './utils/debug'
 import AutomergeBackend from './AutomergeBackend'
+import getActiveConnections from './utils/getActiveConnections'
 
 export interface IAutomergeMetaData {
   docId: string
@@ -344,11 +345,7 @@ export default class AutomergeCollaboration {
   garbageNsp = (socket: SocketIO.Socket) => {
     const { name: docId } = socket.nsp
 
-    // This is the only way to synchronously check the number of active Automerge.Connections
+    const activeConnectionsCount = getActiveConnections(this.backend, docId)
-    // for this docId.
-    // @ts-ignore
-    const activeConnectionsCount = this.backend.documentSetMap[docId]?.handlers
-      .size
 
     debugCollabBackend(
       'Garbage namespace activeConnections=%s',

packages/backend/src/utils/getActiveConnections.ts

@@ -0,0 +1,17 @@
+import AutomergeBackend from '../AutomergeBackend'
+
+/**
+ * Get the number of active connections for the specified docId
+ */
+const getActiveConnections = (backend: AutomergeBackend, docId: string) => {
+  const automergeDocument = backend.documentSetMap[docId]
+
+  if (!automergeDocument) return 0
+
+  // This is the only way to synchronously check the number of active Automerge.Connections
+  // for this docId.
+  // @ts-ignore
+  return automergeDocument.handlers.size
+}
+
+export default getActiveConnections

packages/client/package.json

@@ -22,7 +22,7 @@
     "build:types": "tsc --emitDeclarationOnly",
     "build:js": "babel src --out-dir lib --extensions \".ts,.tsx\" --source-maps inline",
     "watch": "yarn build:js -w",
-    "test": "jest"
+    "test": "DEBUG=app* jest"
   },
   "dependencies": {
     "@babel/plugin-proposal-optional-chaining": "^7.9.0",

packages/client/src/client.spec.ts

@@ -4,10 +4,18 @@ import fs from 'fs'
 import isEqual from 'lodash/isEqual'
 import { createEditor, Editor, Element, Node, Transforms } from 'slate'
 import { createDoc, SyncDoc, toJS, toSlateOp } from '@hiveteams/collab-bridge'
-import AutomergeCollaboration from '@hiveteams/collab-backend/lib/AutomergeCollaboration'
+import AutomergeCollaboration, {
+  IAutomergeMetaData
+} from '@hiveteams/collab-backend/lib/AutomergeCollaboration'
 import withIOCollaboration from './withIOCollaboration'
-import { AutomergeOptions, SocketIOPluginOptions } from './interfaces'
+import {
+  AutomergeEditor,
+  AutomergeOptions,
+  SocketIOPluginOptions,
+  WithSocketIOEditor
+} from './interfaces'
 import { getTarget } from '@hiveteams/collab-bridge/src/path'
+import getActiveConnections from '@hiveteams/collab-backend/src/utils/getActiveConnections'
 
 const connectionSlug = 'test'
 const docId = `/${connectionSlug}`
@@ -41,6 +49,7 @@ const server = createServer(function(req, res) {
 })
 
 const defaultSlateJson = [{ type: 'paragraph', children: [{ text: '' }] }]
+let operationTraces: IAutomergeMetaData[] = []
 const collabBackend = new AutomergeCollaboration({
   entry: server,
   defaultValue: defaultSlateJson,
@@ -50,6 +59,10 @@ const collabBackend = new AutomergeCollaboration({
   },
   async onDocumentLoad(pathname) {
     return defaultSlateJson
+  },
+  onTrace(metaData, computationFn) {
+    operationTraces.push(metaData)
+    computationFn()
   }
 })
 
@@ -60,6 +73,19 @@ describe('automerge editor client tests', () => {
     server.listen(5000, () => done())
   })
 
+  let collabEditors: (Editor & WithSocketIOEditor & AutomergeEditor)[] = []
+  afterEach(done => {
+    // Clear our operation traces after each test
+    operationTraces = []
+
+    // Destroy any created collab editors after each test
+    collabEditors.forEach(editor => editor.destroy())
+    collabEditors = []
+
+    // Ensure that the collab document has been cleaned up on the backend
+    waitForCondition(() => !collabBackend.backend.getDocument(docId)).then(done)
+  })
+
   const createCollabEditor = async (
     editorOptions?: Partial<AutomergeOptions> & Partial<SocketIOPluginOptions>
   ) => {
@@ -82,6 +108,7 @@ describe('automerge editor client tests', () => {
     })
     editor.connect()
 
+    collabEditors.push(editor)
     await promise
     return editor
   }
@@ -101,8 +128,6 @@ describe('automerge editor client tests', () => {
       const serverDoc = toJS(collabBackend.backend.getDocument(docId))
       return serverDoc.children.length === 2
     })
-
-    editor.destroy()
   })
 
   it('should sync updates across two clients', async () => {
@@ -115,9 +140,6 @@ describe('automerge editor client tests', () => {
       const serverDoc = toJS(collabBackend.backend.getDocument(docId))
       return serverDoc.children.length === 2 && editor2.children.length === 2
     })
-
-    editor1.destroy()
-    editor2.destroy()
   })
 
   it('should sync offline changes on reconnect', async () => {
@@ -143,9 +165,6 @@ describe('automerge editor client tests', () => {
     })
 
     expect(Node.string(editor2.children[2])).toEqual('offline')
-
-    editor1.destroy()
-    editor2.destroy()
   })
 
   it('should work with concurrent edits', async () => {
@@ -166,9 +185,6 @@ describe('automerge editor client tests', () => {
     })
 
     expect(isEqual(editor1.children, editor2.children)).toBeTruthy()
-
-    editor1.destroy()
-    editor2.destroy()
   })
 
   it('should work with concurrent insert text operations', async () => {
@@ -192,9 +208,6 @@ describe('automerge editor client tests', () => {
     })
 
     expect(isEqual(editor1.children, editor2.children)).toBeTruthy()
-
-    editor1.destroy()
-    editor2.destroy()
   })
 
   it('should not throw deep nested tree error', () => {
@@ -265,6 +278,38 @@ describe('automerge editor client tests', () => {
     expect(target).toEqual(null)
   })
 
+  it('should reconnect with no opCount', async () => {
+    const editor1 = await createCollabEditor({ resetOnReconnect: true })
+
+    await waitForCondition(() => {
+      return getActiveConnections(collabBackend.backend, docId) === 1
+    })
+
+    editor1.disconnect()
+
+    await waitForCondition(
+      () => getActiveConnections(collabBackend.backend, docId) === 0
+    )
+
+    editor1.connect()
+
+    await waitForCondition(
+      () => getActiveConnections(collabBackend.backend, docId) === 1
+    )
+
+    // Wait for a few seconds to allow the client and server to synchronize their
+    // document states
+    await new Promise(res => setTimeout(res, 3000))
+
+    // Expect that reconnecting with resetOnReconnect option set to true
+    // does not result in any operations being sent from the client to the server
+    expect(
+      operationTraces.some(
+        trace => trace.opCount !== undefined && trace.opCount > 0
+      )
+    ).toBeFalsy
+  })
+
   afterAll(() => {
     collabBackend.destroy()
     server.close()

packages/client/src/interfaces.ts

@@ -38,6 +38,7 @@ export interface SocketIOPluginOptions {
   onConnect?: () => void
   onDisconnect?: () => void
   onError?: (msg: string | Error, data: any) => void
+  resetOnReconnect?: boolean
 }
 
 export interface WithSocketIOEditor {

packages/client/src/withSocketIO.ts

@@ -17,7 +17,14 @@ const withSocketIO = <T extends AutomergeEditor>(
   slateEditor: T,
   options: SocketIOPluginOptions & AutomergeOptions
 ) => {
-  const { onConnect, onDisconnect, connectOpts, url } = options
+  const {
+    onConnect,
+    onDisconnect,
+    connectOpts,
+    url,
+    docId,
+    resetOnReconnect
+  } = options
   const editor = slateEditor as T & WithSocketIOEditor & AutomergeEditor
   let socket: SocketIOClient.Socket
 
@@ -31,6 +38,17 @@ const withSocketIO = <T extends AutomergeEditor>(
     // On socket io connect, open a new automerge connection
     socket.on('connect', () => {
       editor.clientId = socket.id
+
+      // If the resetOnReconnect option is true we should close our connection
+      // and remove our document from the docSet if the user has already received
+      // a document from our collab server
+      if (resetOnReconnect && editor.docSet.getDoc(docId)) {
+        if (editor.connection) {
+          editor.connection.close()
+        }
+        editor.docSet.removeDoc(docId)
+      }
+
       editor.openConnection()
       onConnect && onConnect()
     })