fix(sync): keep queue around during reconnects

When yjs does not receive awareness updates
it will close and reopen the websocket.
Keep the content of the queue, i.e. the outgoing steps
so they can be send out once the connection is back.

Signed-off-by: Max <[email protected]>

cypress/e2e/api/SyncServiceProvider.spec.js

@@ -60,6 +60,7 @@ describe('Sync service provider', function() {
 	 * @param {object} ydoc Yjs document
 	 */
 	function createProvider(ydoc) {
+		const queue = []
 		const syncService = new SyncService({
 			serialize: () => 'Serialized',
 			getDocumentState: () => null,
@@ -70,6 +71,7 @@ describe('Sync service provider', function() {
 			syncService,
 			fileId,
 			initialSession: null,
+			queue,
 			disableBc: true,
 		})
 	}

src/components/Editor.vue

@@ -329,6 +329,7 @@ export default {
 	},
 	created() {
 		this.$ydoc = new Doc()
+		this.$queue = []
 		// The following can be useful for debugging ydoc updates
 		// this.$ydoc.on('update', function(update, origin, doc, tr) {
 		//   console.debug('ydoc update', update, origin, doc, tr)
@@ -381,6 +382,7 @@ export default {
 				ydoc: this.$ydoc,
 				syncService: this.$syncService,
 				fileId: this.fileId,
+				queue: this.$queue,
 				initialSession: this.initialSession,
 			})
 			this.$providers.push(syncServiceProvider)

src/services/SyncServiceProvider.js

@@ -30,15 +30,16 @@ import { logger } from '../helpers/logger.js'
  * @param {object} options.ydoc - the Ydoc
  * @param {object} options.syncService - sync service to build upon
  * @param {number} options.fileId - file id of the file to open
+ * @param {number} options.queue - queue for outgoing steps
  * @param {object} options.initialSession - initialSession to start from
  * @param {boolean} options.disableBc - disable broadcast channel synchronization (default: disabled in debug mode, enabled otherwise)
  */
-export default function createSyncServiceProvider({ ydoc, syncService, fileId, initialSession, disableBc }) {
+export default function createSyncServiceProvider({ ydoc, syncService, fileId, initialSession, queue, disableBc }) {
 	if (!fileId) {
 		// We need a file id as a unique identifier for y.js as otherwise state might leak between different files
 		throw new Error('fileId is required')
 	}
-	const WebSocketPolyfill = initWebSocketPolyfill(syncService, fileId, initialSession)
+	const WebSocketPolyfill = initWebSocketPolyfill(syncService, fileId, initialSession, queue)
 	disableBc = disableBc ?? !!window?._oc_debug
 	const websocketProvider = new WebsocketProvider(
 		'ws://localhost:1234',

src/services/WebSocketPolyfill.js

@@ -28,8 +28,9 @@ import { encodeArrayBuffer, decodeArrayBuffer } from '../helpers/base64.js'
  * @param {object} syncService - the sync service to build upon
  * @param {number} fileId - id of the file to open
  * @param {object} initialSession - initial session to open
+ * @param {object[]} queue - queue for the outgoing steps
  */
-export default function initWebSocketPolyfill(syncService, fileId, initialSession) {
+export default function initWebSocketPolyfill(syncService, fileId, initialSession, queue) {
 	return class WebSocketPolyfill {
 
 		#url
@@ -41,11 +42,9 @@ export default function initWebSocketPolyfill(syncService, fileId, initialSessio
 		onclose
 		onopen
 		#handlers
-		#queue
 
 		constructor(url) {
 			this.url = url
-			this.#queue = []
 			logger.debug('WebSocketPolyfill#constructor', { url, fileId, initialSession })
 			this.#registerHandlers({
 				opened: ({ version, session }) => {
@@ -83,32 +82,32 @@ export default function initWebSocketPolyfill(syncService, fileId, initialSessio
 			// Useful for debugging what steps are sent and how they were initiated
 			// data.forEach(logStep)
 
-			this.#queue.push(...data)
+			queue.push(...data)
 			let outbox = []
 			return syncService.sendSteps(() => {
-				outbox = [...this.#queue]
+				outbox = [...queue]
 				const data = {
 					steps: this.#steps,
 					awareness: this.#awareness,
 					version: this.#version,
 				}
-				this.#queue = []
+				queue = []
 				logger.debug('sending steps ', data)
 				return data
 			})?.catch(err => {
 				logger.error(err)
 				// try to send the steps again
-				this.#queue = [...outbox, ...this.#queue]
+				queue = [...outbox, ...queue]
 			})
 		}
 
 		get #steps() {
-			return this.#queue.map(s => encodeArrayBuffer(s))
+			return queue.map(s => encodeArrayBuffer(s))
 				.filter(s => s < 'AQ')
 		}
 
 		get #awareness() {
-			return this.#queue.map(s => encodeArrayBuffer(s))
+			return queue.map(s => encodeArrayBuffer(s))
 				.findLast(s => s > 'AQ') || ''
 		}
 
@@ -124,21 +123,21 @@ export default function initWebSocketPolyfill(syncService, fileId, initialSessio
 		}
 
 		#sendRemainingSteps() {
-			if (this.#queue.length) {
+			if (queue.length) {
 				let outbox = []
 				return syncService.sendStepsNow(() => {
-					outbox = [...this.#queue]
+					outbox = [...queue]
 					const data = {
 						steps: this.#steps,
 						awareness: this.#awareness,
 						version: this.#version,
 					}
-					this.#queue = []
+					queue = []
 					logger.debug('sending final steps ', data)
 					return data
 				})?.catch(err => {
 					logger.error(err)
-					this.#queue = [...outbox, ...this.#queue]
+					queue = [...outbox, ...queue]
 				})
 			}
 		}