Proposal for chunked pulls

Author: tlonny

src/sync/impl/synchronize.js

@@ -14,12 +14,18 @@ import {
 import { ensureSameDatabase, isChangeSetEmpty, changeSetCount } from './helpers'
 import type { SyncArgs, Timestamp, SyncPullStrategy } from '../index'
 
+const liftToGenerator = async function* (promise) {
+    const result = await promise
+    yield result
+}
+
 export default async function synchronize({
   database,
   pullChanges,
   onWillApplyRemoteChanges,
   onDidPullChanges,
   pushChanges,
+  useAsyncGeneratorForPull = false,
   sendCreatedAsUpdated = false,
   migrationsEnabledAtVersion,
   log,
@@ -45,79 +51,79 @@ export default async function synchronize({
   )
   log && (log.phase = 'ready to pull')
 
-  // $FlowFixMe
-  const pullResult = await pullChanges({
-    lastPulledAt,
-    schemaVersion,
-    migration,
-  })
-  log && (log.phase = 'pulled')
+  const pullGenerator = useAsyncGeneratorForPull
+    ? pullChanges({ lastPulledAt, schemaVersion, migration })
+    : liftToGenerator(pullChanges({ lastPulledAt, schemaVersion, migration }))
 
-  let newLastPulledAt: Timestamp = (pullResult: any).timestamp
-  const remoteChangeCount = pullResult.changes ? changeSetCount(pullResult.changes) : NaN
+  for await (const pullResult of pullGenerator) {
+    log && (log.phase = 'pulled')
 
-  if (onWillApplyRemoteChanges) {
-    await onWillApplyRemoteChanges({ remoteChangeCount })
-  }
+    let newLastPulledAt: Timestamp = (pullResult: any).timestamp
+    const remoteChangeCount = pullResult.changes ? changeSetCount(pullResult.changes) : NaN
 
-  await database.write(async () => {
-    ensureSameDatabase(database, resetCount)
-    invariant(
-      lastPulledAt === (await getLastPulledAt(database)),
-      '[Sync] Concurrent synchronization is not allowed. More than one synchronize() call was running at the same time, and the later one was aborted before committing results to local database.',
-    )
+    if (onWillApplyRemoteChanges) {
+      await onWillApplyRemoteChanges({ remoteChangeCount })
+    }
 
-    if (unsafeTurbo) {
-      invariant(
-        !_unsafeBatchPerCollection,
-        'unsafeTurbo must not be used with _unsafeBatchPerCollection',
-      )
+    await database.write(async () => {
+      ensureSameDatabase(database, resetCount)
       invariant(
-        'syncJson' in pullResult || 'syncJsonId' in pullResult,
-        'missing syncJson/syncJsonId',
+        lastPulledAt === (await getLastPulledAt(database)),
+        '[Sync] Concurrent synchronization is not allowed. More than one synchronize() call was running at the same time, and the later one was aborted before committing results to local database.',
       )
-      invariant(lastPulledAt === null, 'unsafeTurbo can only be used as the first sync')
 
-      const syncJsonId = pullResult.syncJsonId || Math.floor(Math.random() * 1000000000)
-
-      if (pullResult.syncJson) {
-        await database.adapter.provideSyncJson(syncJsonId, pullResult.syncJson)
+      if (unsafeTurbo) {
+        invariant(
+          !_unsafeBatchPerCollection,
+          'unsafeTurbo must not be used with _unsafeBatchPerCollection',
+        )
+        invariant(
+          'syncJson' in pullResult || 'syncJsonId' in pullResult,
+          'missing syncJson/syncJsonId',
+        )
+        invariant(lastPulledAt === null, 'unsafeTurbo can only be used as the first sync')
+
+        const syncJsonId = pullResult.syncJsonId || Math.floor(Math.random() * 1000000000)
+
+        if (pullResult.syncJson) {
+          await database.adapter.provideSyncJson(syncJsonId, pullResult.syncJson)
+        }
+
+        const resultRest = await database.adapter.unsafeLoadFromSync(syncJsonId)
+        newLastPulledAt = resultRest.timestamp
+        onDidPullChanges && onDidPullChanges(resultRest)
       }
 
-      const resultRest = await database.adapter.unsafeLoadFromSync(syncJsonId)
-      newLastPulledAt = resultRest.timestamp
-      onDidPullChanges && onDidPullChanges(resultRest)
-    }
+      log && (log.newLastPulledAt = newLastPulledAt)
+      invariant(
+        typeof newLastPulledAt === 'number' && newLastPulledAt > 0,
+        `pullChanges() returned invalid timestamp ${newLastPulledAt}. timestamp must be a non-zero number`,
+      )
 
-    log && (log.newLastPulledAt = newLastPulledAt)
-    invariant(
-      typeof newLastPulledAt === 'number' && newLastPulledAt > 0,
-      `pullChanges() returned invalid timestamp ${newLastPulledAt}. timestamp must be a non-zero number`,
-    )
-
-    if (!unsafeTurbo) {
-      // $FlowFixMe
-      const { changes: remoteChanges, ...resultRest } = pullResult
-      log && (log.remoteChangeCount = remoteChangeCount)
-      // $FlowFixMe
-      await applyRemoteChanges(remoteChanges, {
-        db: database,
-        strategy: ((pullResult: any).experimentalStrategy: ?SyncPullStrategy),
-        sendCreatedAsUpdated,
-        log,
-        conflictResolver,
-        _unsafeBatchPerCollection,
-      })
-      onDidPullChanges && onDidPullChanges(resultRest)
-    }
+      if (!unsafeTurbo) {
+        // $FlowFixMe
+        const { changes: remoteChanges, ...resultRest } = pullResult
+        log && (log.remoteChangeCount = remoteChangeCount)
+        // $FlowFixMe
+        await applyRemoteChanges(remoteChanges, {
+          db: database,
+          strategy: ((pullResult: any).experimentalStrategy: ?SyncPullStrategy),
+          sendCreatedAsUpdated,
+          log,
+          conflictResolver,
+          _unsafeBatchPerCollection,
+        })
+        onDidPullChanges && onDidPullChanges(resultRest)
+      }
 
-    log && (log.phase = 'applied remote changes')
-    await setLastPulledAt(database, newLastPulledAt)
+      log && (log.phase = 'applied remote changes')
+      await setLastPulledAt(database, newLastPulledAt)
 
-    if (shouldSaveSchemaVersion) {
-      await setLastPulledSchemaVersion(database, schemaVersion)
-    }
-  }, 'sync-synchronize-apply')
+      if (shouldSaveSchemaVersion) {
+        await setLastPulledSchemaVersion(database, schemaVersion)
+      }
+    }, 'sync-synchronize-apply')
+  }
 
   // push phase
   if (pushChanges) {