fix(sync): only clear queue after successful send

Also add a unit test for the websocket polyfill

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

src/services/WebSocketPolyfill.js

@@ -83,21 +83,24 @@ export default function initWebSocketPolyfill(syncService, fileId, initialSessio
 			// data.forEach(logStep)
 
 			queue.push(...data)
-			let outbox
+			let outbox = []
 			return syncService.sendSteps(() => {
 				const data = {
 					steps: this.#steps,
 					awareness: this.#awareness,
 					version: this.#version,
 				}
-				outbox = queue.splice(0, queue.length)
+				outbox = [...queue]
 				logger.debug('sending steps ', data)
 				return data
-			})?.catch(err => {
-				logger.error(err)
-				// Prefix the queue with the steps in outbox to send them again
-				queue.splice(0, 0, ...outbox)
-			})
+			})?.then(ret => {
+				// only keep the steps that were not send yet
+				queue.splice(0,
+					queue.length,
+					...queue.filter(s => !outbox.includes(s)),
+				)
+				return ret
+			}, err => logger.error(err))
 		}
 
 		get #steps() {
@@ -130,14 +133,16 @@ export default function initWebSocketPolyfill(syncService, fileId, initialSessio
 						awareness: this.#awareness,
 						version: this.#version,
 					}
-					outbox = queue.splice(0, queue.length)
+					outbox = [...queue]
 					logger.debug('sending final steps ', data)
 					return data
-				})?.catch(err => {
-					logger.error(err)
-					// Prefix the queue with the steps in outbox to send them again
-					queue.splice(0, 0, ...outbox)
-				})
+				})?.then(() => {
+					// only keep the steps that were not send yet
+					queue.splice(0,
+						queue.length,
+						...queue.filter(s => !outbox.includes(s)),
+					)
+				}, err => logger.error(err))
 			}
 		}
 

src/tests/services/WebsocketPolyfill.spec.js

@@ -0,0 +1,114 @@
+import initWebSocketPolyfill from '../../services/WebSocketPolyfill.js'
+
+describe('Init function', () => {
+
+	it('returns a websocket polyfill class', () => {
+		const syncService = { on: jest.fn(), open: jest.fn() }
+		const Polyfill = initWebSocketPolyfill(syncService)
+		const websocket = new Polyfill('url')
+		expect(websocket).toBeInstanceOf(Polyfill)
+	})
+
+	it('registers handlers', () => {
+		const syncService = { on: jest.fn(), open: jest.fn() }
+		const Polyfill = initWebSocketPolyfill(syncService)
+		const websocket = new Polyfill('url')
+		expect(syncService.on).toHaveBeenCalled()
+	})
+
+	it('opens sync service', () => {
+		const syncService = { on: jest.fn(), open: jest.fn() }
+		const fileId = 123
+		const initialSession = { }
+		const Polyfill = initWebSocketPolyfill(syncService, fileId, initialSession)
+		const websocket = new Polyfill('url')
+		expect(syncService.open).toHaveBeenCalledWith({ fileId, initialSession })
+	})
+
+	it('sends steps to sync service', async () => {
+		const syncService = {
+			on: jest.fn(),
+			open: jest.fn(),
+			sendSteps: async getData => getData(),
+		}
+		const queue = [ 'initial' ]
+		const data = { dummy: 'data' }
+		const Polyfill = initWebSocketPolyfill(syncService, null, null, queue)
+		const websocket = new Polyfill('url')
+		const result = websocket.send(data)
+		expect(result).toBeInstanceOf(Promise)
+		expect(queue).toEqual([ 'initial' , data ])
+		const dataSendOut = await result
+		expect(queue).toEqual([])
+		expect(dataSendOut).toHaveProperty('awareness')
+		expect(dataSendOut).toHaveProperty('steps')
+		expect(dataSendOut).toHaveProperty('version')
+	})
+
+	it('handles early reject', async () => {
+		const syncService = {
+			on: jest.fn(),
+			open: jest.fn(),
+			sendSteps: jest.fn().mockRejectedValue('error before reading steps in sync service'),
+		}
+		const queue = [ 'initial' ]
+		const data = { dummy: 'data' }
+		const Polyfill = initWebSocketPolyfill(syncService, null, null, queue)
+		const websocket = new Polyfill('url')
+		const result = websocket.send(data)
+		expect(queue).toEqual([ 'initial' , data ])
+		expect(result).toBeInstanceOf(Promise)
+		const returned = await result
+		expect(returned).toBeUndefined()
+		expect(queue).toEqual([ 'initial' , data ])
+	})
+
+	it('handles reject after reading data', async () => {
+		const syncService = {
+			on: jest.fn(),
+			open: jest.fn(),
+			sendSteps: jest.fn().mockImplementation( async getData => {
+				getData()
+				throw 'error when sending in sync service'
+			}),
+		}
+		const queue = [ 'initial' ]
+		const data = { dummy: 'data' }
+		const Polyfill = initWebSocketPolyfill(syncService, null, null, queue)
+		const websocket = new Polyfill('url')
+		const result = websocket.send(data)
+		expect(queue).toEqual([ 'initial' , data ])
+		expect(result).toBeInstanceOf(Promise)
+		const returned = await result
+		expect(returned).toBeUndefined()
+		expect(queue).toEqual([ 'initial' , data ])
+	})
+
+	it('queue survives a close', async () => {
+		const syncService = {
+			on: jest.fn(),
+			open: jest.fn(),
+			sendSteps: jest.fn().mockImplementation( async getData => {
+				getData()
+				throw 'error when sending in sync service'
+			}),
+			sendStepsNow: jest.fn().mockImplementation( async getData => {
+				getData()
+				throw 'sendStepsNow error when sending'
+			}),
+			off: jest.fn(),
+			close: jest.fn( async data => data ),
+		}
+		const queue = [ 'initial' ]
+		const data = { dummy: 'data' }
+		const Polyfill = initWebSocketPolyfill(syncService, null, null, queue)
+		const websocket = new Polyfill('url')
+		websocket.onclose = jest.fn()
+		await websocket.send(data)
+		const promise = websocket.close()
+		expect(queue).toEqual([ 'initial' , data ])
+		await promise
+		expect(queue).toEqual([ 'initial' , data ])
+	})
+
+})