Attempt reconnect via the FB channel on init sync stall

ts/fast-sync/chunking.test.ts

@@ -4,6 +4,7 @@ import {
     getStringChunk,
     receiveInChucks,
 } from './chunking'
+import Interruptable from './interruptable'
 
 describe('Fast sync channel package chunking', () => {
     it('should calculate the right chunk count for strings exactly fitting the chunk size', () => {
@@ -31,8 +32,25 @@ describe('Fast sync channel package chunking', () => {
     it('should correctly receive data in chunks', async () => {
         const chunks = [`chunk:0:3:ab`, `chunk:1:3:cde`, `chunk:2:3:fghij`]
 
-        expect(await receiveInChucks(async () => chunks.shift()!)).toEqual(
-            'abcdefghij',
-        )
+        const interruptable = new Interruptable()
+        expect(
+            await receiveInChucks(async () => chunks.shift()!, interruptable),
+        ).toEqual('abcdefghij')
+    })
+
+    it('should throw an exception when cancelled', async () => {
+        const chunks = [`chunk:0:3:ab`, `chunk:1:3:cde`, `chunk:2:3:fghij`]
+        let index = -1
+
+        const interruptable = new Interruptable({ throwOnCancelled: true })
+        await expect(
+            receiveInChucks(async () => {
+                ++index
+                if (index === 1) {
+                    await interruptable.cancel()
+                }
+                return chunks.shift()!
+            }, interruptable),
+        ).rejects.toThrow('Tried to execute code on a cancelled interruptable')
     })
 })

ts/fast-sync/chunking.ts

@@ -1,4 +1,5 @@
 import { splitWithTail } from './utils'
+import Interruptable from './interruptable'
 
 export function calculateStringChunkCount(
     s: string,
@@ -17,21 +18,20 @@ export function getStringChunk(
 
 export async function receiveInChucks(
     receiveChunk: () => Promise<string>,
+    interruptable: Interruptable,
 ): Promise<string> {
     let data: string[] = []
     let expectedChunkCount: null | number = null
 
-    while (true) {
-        const chunk = await receiveChunk()
-
+    const processChunk = (chunk: string) => {
         const [
             chunkConfirmation,
             chunkIndexString,
             chunkCountString,
             chunkContent,
         ] = splitWithTail(chunk, ':', 4)
         if (chunkConfirmation !== 'chunk') {
-            throw new Error(`Invalid WebRTC package received: ${chunk}`)
+            throw new Error(`Invalid WebRTC chunk package received: ${chunk}`)
         }
 
         const chunkIndex = parseInt(chunkIndexString)
@@ -67,10 +67,44 @@ export async function receiveInChucks(
         }
 
         data.push(chunkContent)
-        if (data.length === expectedChunkCount) {
-            break
-        }
+        return { finished: data.length === expectedChunkCount }
     }
 
+    let running = true
+    await interruptable.whileLoop(
+        async () => running,
+        async () => {
+            const chunk = (await interruptable.execute(receiveChunk))!
+            const result = await interruptable.execute(async () =>
+                processChunk(chunk),
+            )
+            if (result?.finished) {
+                running = false
+            }
+        },
+    )
+
     return data.join('')
 }
+
+export async function sendInChunks(
+    message: string,
+    send: (chunk: string) => Promise<void>,
+    options: {
+        interruptable: Interruptable
+        chunkSize: number
+    },
+) {
+    const chunkCount = calculateStringChunkCount(message, options)
+    let chunkIndex = -1
+    await options.interruptable.whileLoop(
+        async () => chunkIndex < chunkCount,
+        async () => {
+            chunkIndex += 1
+            const chunkContent = getStringChunk(message, chunkIndex, {
+                chunkSize: options.chunkSize,
+            })
+            await send(`chunk:${chunkIndex}:${chunkCount}:${chunkContent}`)
+        },
+    )
+}

ts/fast-sync/index.test.ts

@@ -117,7 +117,9 @@ async function createWebRTCSyncChannels(options: {
     ])
 
     return {
-        senderChannel: new WebRTCFastSyncChannel({ peer: peers[0] }),
+        senderChannel: new WebRTCFastSyncChannel({
+            peer: peers[0],
+        }),
         receiverChannel: new WebRTCFastSyncChannel({
             peer: peers[1],
         }),

ts/fast-sync/index.ts

@@ -3,6 +3,7 @@ import TypedEmitter from 'typed-emitter'
 import StorageManager from '@worldbrain/storex'
 import {
     FastSyncInfo,
+    FastSyncChannelEvents,
     FastSyncProgress,
     FastSyncChannel,
     FastSyncRole,
@@ -29,15 +30,12 @@ export interface FastSyncPreSendProcessorParams {
     object: any
 }
 
-export interface FastSyncEvents {
+export interface FastSyncEvents extends FastSyncChannelEvents {
     prepared: (event: { syncInfo: FastSyncInfo; role: FastSyncRole }) => void
     progress: (event: {
         progress: FastSyncProgress
         role: FastSyncRole
     }) => void
-    stalled: () => void
-    paused: () => void
-    resumed: () => void
     roleSwitch: (event: { before: FastSyncRole; after: FastSyncRole }) => void
     error: (event: { error: string }) => void
 }
@@ -59,6 +57,7 @@ export class FastSync {
 
     constructor(private options: FastSyncOptions) {
         this.totalObjectsProcessed = 0
+        this.forwardReconnectEvents()
     }
 
     get state() {
@@ -104,9 +103,9 @@ export class FastSync {
     async send(options: { role: FastSyncRole; fastSyncInfo?: FastSyncInfo }) {
         const { channel } = this.options
 
-        channel.events.on('stalled', () => this.events.emit('stalled'))
         const interruptable = (this.interruptable = new Interruptable())
         this._state = 'running'
+
         try {
             const syncInfo =
                 options.fastSyncInfo ||
@@ -185,6 +184,15 @@ export class FastSync {
         })
     }
 
+    private forwardReconnectEvents() {
+        const events: any[] = ['stalled', 'reconnect', 'reconnected']
+        for (const event of events) {
+            this.options.channel.events.on(event, (e: any) =>
+                this.events.emit(event, e),
+            )
+        }
+    }
+
     async _preproccesObjects(params: { collection: string; objects: any[] }) {
         const preSendProcessor = this.options.preSendProcessor
         if (!preSendProcessor) {
@@ -213,11 +221,10 @@ export class FastSync {
             this._state = state === 'paused' ? 'paused' : 'running'
             this.events.emit(state)
         }
-        this.options.channel.events.on('stalled', () =>
-            this.events.emit('stalled'),
-        )
+
         this.options.channel.events.on('paused', stateChangeHandler('paused'))
         this.options.channel.events.on('resumed', stateChangeHandler('resumed'))
+
         try {
             const syncInfo = await this.options.channel.receiveSyncInfo()
             this.events.emit('prepared', { syncInfo, role: options.role })

ts/fast-sync/interruptable.ts

@@ -1,9 +1,13 @@
 import { ResolvablePromise, resolvablePromise } from './utils'
 
+export class InterruptableCancelledError extends Error {}
+
 export default class Interruptable {
     cancelled: boolean = false
     private pausePromise: ResolvablePromise<void> | null = null // only set if paused, resolved when pause ends
 
+    constructor(private options?: { throwOnCancelled: boolean }) {}
+
     get paused(): boolean {
         return !!this.pausePromise
     }
@@ -73,7 +77,7 @@ export default class Interruptable {
         }
     }
 
-    async execute(f: () => Promise<void>) {
+    async execute<ReturnValue>(f: () => Promise<ReturnValue | undefined>) {
         if (await this._shouldCancelAfterWaitingForPause()) {
             return
         }
@@ -85,6 +89,11 @@ export default class Interruptable {
         if (this.pausePromise) {
             await this.pausePromise.promise
         }
+        if (this.cancelled && this.options?.throwOnCancelled) {
+            throw new InterruptableCancelledError(
+                'Tried to execute code on a cancelled interruptable',
+            )
+        }
         return this.cancelled
     }
 }

ts/fast-sync/types.ts

@@ -4,17 +4,23 @@ export type FastSyncRole = 'sender' | 'receiver'
 export const flippedRole = (role: FastSyncRole): FastSyncRole =>
     role === 'sender' ? 'receiver' : 'sender'
 export type FastSyncOrder = 'receive-first' | 'send-first'
-export type FastSyncPackage<UserPackageType = any> =
-    | { type: 'batch'; batch: any }
-    | { type: 'confirm' }
-    | { type: 'state-change'; state: 'paused' | 'running' }
-    | { type: 'sync-info'; info: FastSyncInfo }
-    | { type: 'finish' }
-    | { type: 'user-package'; package: UserPackageType }
+export type FastSyncPackage<
+    UserPackageType = any,
+    WithIndex extends boolean = true
+> = (WithIndex extends true ? { index: number } : {}) &
+    (
+        | { type: 'sync-info'; info: FastSyncInfo }
+        | { type: 'batch'; batch: any }
+        | { type: 'finish' }
+        | { type: 'state-change'; state: 'paused' | 'running' }
+        | { type: 'user-package'; package: UserPackageType }
+        | { type: 'confirm' }
+    )
+
 export interface FastSyncChannelEvents {
     stalled: () => void
-    paused: () => void
     resumed: () => void
+    paused: () => void
 }
 export interface FastSyncChannel<UserPackageType = any> {
     timeoutInMiliseconds: number

ts/index.test.ts

@@ -1366,7 +1366,7 @@ if (process.env.TEST_SYNC_FIRESTORE === 'true') {
                     {
                         sharedSyncLog: ({ storageManager }) =>
                             new SharedSyncLogStorage({
-                                storageManager,
+                                storageManager: storageManager as any,
                                 autoPkType: 'string',
                                 excludeTimestampChecks:
                                     !options || !options.includeTimestampChecks,