feat: handle resync & missing updates

Author: stanley2058

src/y-socket-io/client.js

@@ -340,19 +340,24 @@ export class SocketIOProvider extends Observable {
       )
     }
     if (resyncInterval > 0) {
-      this.resyncInterval = setInterval(() => {
-        if (this.socket.disconnected) return
-        this.socket.emit(
-          'sync-step-1',
-          Y.encodeStateVector(this.doc),
-          (/** @type {Uint8Array} */ update) => {
-            Y.applyUpdate(this.doc, new Uint8Array(update), this)
-          }
-        )
-      }, resyncInterval)
+      this.resyncInterval = setInterval(() => this.resync(), resyncInterval)
     }
   }
 
+  /**
+   * Resynchronize the document with the server by firing `sync-step-1`.
+   */
+  resync () {
+    if (this.socket.disconnected) return
+    this.socket.emit(
+      'sync-step-1',
+      Y.encodeStateVector(this.doc),
+      (/** @type {Uint8Array} */ update) => {
+        Y.applyUpdate(this.doc, new Uint8Array(update), this)
+      }
+    )
+  }
+
   /**
    * Disconnect provider's socket
    * @type {() => void}
@@ -413,6 +418,11 @@ export class SocketIOProvider extends Observable {
     super.destroy()
   }
 
+  /**
+   * @type {number}
+   * @private
+   */
+  _updateRetries = 0
   /**
    * This function is executed when the document is updated, if the instance that
    * emit the change is not this, it emit the changes by socket and broadcast channel.
@@ -421,9 +431,28 @@ export class SocketIOProvider extends Observable {
    * @param {SocketIOProvider} origin The SocketIOProvider instance that emits the change.
    * @readonly
    */
-  onUpdateDoc = (update, origin) => {
+  onUpdateDoc = async (update, origin) => {
+    if (this._updateRetries > 3) {
+      this._updateRetries = 0
+      this.disconnect()
+      this.connect()
+      return
+    }
+
     if (origin !== this) {
-      this.socket.emit('sync-update', update)
+      /** @type {boolean} */
+      const ack = await Promise.race([
+        new Promise((res) => this.socket.emit('sync-update', update, () => res(true))),
+        new Promise((res) => setTimeout(() => res(false), 3000)),
+      ])
+      if (!ack) {
+        this._updateRetries++
+        if (this.socket.disconnected) return
+        this.onUpdateDoc(update, origin)
+        return
+      } else {
+        this._updateRetries = 0
+      }
       if (this.bcconnected) {
         bc.publish(
           this._broadcastChannel,

src/y-socket-io/y-socket-io.js

@@ -330,8 +330,8 @@ export class YSocketIO {
 
     /** @type {unknown} */
     let prevMsg = null
-    socket.on('sync-update', (/** @type {ArrayBuffer} */ update) => {
-      if (isDeepStrictEqual(update, prevMsg)) return
+    socket.on('sync-update', (/** @type {ArrayBuffer} */ update, /** @type {() => void} */ ack) => {
+      if (isDeepStrictEqual(update, prevMsg)) return ack()
       assert(this.client)
       const namespace = this.getNamespaceString(socket.nsp)
       const message = Buffer.from(update.slice(0, update.byteLength))
@@ -341,6 +341,7 @@ export class YSocketIO {
           'index',
           Buffer.from(this.toRedis('sync-update', message))
         )
+        .then(() => ack())
         .catch(console.error)
       prevMsg = update
     })