Add unsafePullChangesAsyncGenerator to SyncArgs

tlonny · tlonny · commit a3673d65b231 · 2025-11-02T17:41:16.000Z
unsafePullChangesAsyncGenerator is an alternative to pullChanges (now
made optional), that returns an AsyncGenerator of PullResults. This
should make large updates/initial syncs more tractable as the sync can
be split up into several smaller chunks.

  1. Updated TS types to be robust discriminated union (i.e. either
     pullChanges OR unsafePullChangesAsyncGenerator HAVE to be present)
  2. Updated Flow types to make pullChanges and
     unsafePullChangesAsyncGenerator optional
  3. Added an invariant check to ensure there is at least one pull
     strategy provided (for JS lads)
  4. Added code that "lifts" the result from pullChanges into an
     AsyncGenerator - ensuring only 1 code path going forward
  5. Added a loop that pulls of the generator and performs DB writes
     until the generator is exhausted.
  6. Added tests
  7. Added babel support for async generators
diff --git a/babel.config.js b/babel.config.js
@@ -41,6 +41,7 @@ const plugins = [
   '@babel/plugin-proposal-nullish-coalescing-operator',
   '@babel/plugin-transform-shorthand-properties',
   '@babel/plugin-transform-spread',
+  '@babel/plugin-transform-async-generator-functions',
   [
     '@babel/plugin-proposal-object-rest-spread',
     {
diff --git a/package.json b/package.json
@@ -88,6 +88,7 @@
     "@babel/plugin-syntax-flow": "^7.24.7",
     "@babel/plugin-syntax-jsx": "^7.24.7",
     "@babel/plugin-transform-arrow-functions": "^7.24.7",
+    "@babel/plugin-transform-async-generator-functions": "^7.25.9",
     "@babel/plugin-transform-async-to-generator": "^7.24.7",
     "@babel/plugin-transform-block-scoping": "^7.24.7",
     "@babel/plugin-transform-classes": "^7.24.7",
diff --git a/src/sync/impl/__tests__/synchronize.test.js b/src/sync/impl/__tests__/synchronize.test.js
@@ -384,6 +384,66 @@ describe('synchronize', () => {
     expect(log.phase).toBe('ready to pull')
     expect(log.error).toBe(error)
   })
+  it('streams pullChanges chunks from an async generator', async () => {
+    const { database, projects } = makeDatabase()
+    const onWillApplyRemoteChanges = jest.fn()
+
+    async function* chunkedPullChanges() {
+      yield {
+        changes: makeChangeSet({
+          mock_projects: {
+            created: [{ id: 'stream_project', name: 'remote-initial' }],
+          },
+        }),
+        timestamp: 1500,
+      }
+      await delay(1)
+      yield {
+        changes: makeChangeSet({
+          mock_projects: {
+            updated: [{ id: 'stream_project', name: 'remote-updated' }],
+          },
+        }),
+        timestamp: 2500,
+      }
+    }
+
+    await synchronize({
+      database,
+      unsafePullChangesAsyncGenerator: chunkedPullChanges,
+      pushChanges: jest.fn(),
+      onWillApplyRemoteChanges,
+    })
+
+    expect(onWillApplyRemoteChanges).toHaveBeenCalledTimes(2)
+    await expectSyncedAndMatches(projects, 'stream_project', { name: 'remote-updated' })
+    expect(await getLastPulledAt(database)).toBe(2500)
+  })
+  it('rejects when chunked async generator yields no results', async () => {
+    const { database } = makeDatabase()
+
+    async function* emptyGenerator() {}
+
+    await expectToRejectWithMessage(
+      synchronize({
+        database,
+        unsafePullChangesAsyncGenerator: emptyGenerator,
+        pushChanges: jest.fn(),
+      }),
+      /A pullChanges\(\) function must yield at least one result/,
+    )
+  })
+  it('rejects when pullChanges and unsafePullChangesAsyncGenerator not provided', async () => {
+    const { database } = makeDatabase()
+
+    await expectToRejectWithMessage(
+      synchronize({
+        database,
+        pushChanges: jest.fn(),
+      }),
+      /Either pullChanges or unsafePullChangesAsyncGenerator must be provided/,
+    )
+  })
   it('can recover from push failure', async () => {
     const { database, projects } = makeDatabase()
 
diff --git a/src/sync/impl/synchronize.d.ts b/src/sync/impl/synchronize.d.ts
@@ -1,14 +1,3 @@
 import type { SyncArgs } from '../index'
 
-export default function synchronize({
-  database,
-  pullChanges,
-  onDidPullChanges,
-  pushChanges,
-  sendCreatedAsUpdated,
-  migrationsEnabledAtVersion,
-  log,
-  conflictResolver,
-  _unsafeBatchPerCollection,
-  unsafeTurbo,
-}: SyncArgs): Promise<void>
+export default function synchronize(params : SyncArgs): Promise<void>
diff --git a/src/sync/impl/synchronize.js b/src/sync/impl/synchronize.js
@@ -13,10 +13,16 @@ import {
 } from './index'
 import { ensureSameDatabase, isChangeSetEmpty, changeSetCount } from './helpers'
 import type { SyncArgs, Timestamp, SyncPullStrategy } from '../index'
+import type { SyncPullResult } from "..";
+
+async function* liftToAsyncGenerator<T>(promise: Promise<T>): AsyncGenerator<T, void, void> {
+  yield await promise
+}
 
 export default async function synchronize({
   database,
   pullChanges,
+  unsafePullChangesAsyncGenerator,
   onWillApplyRemoteChanges,
   onDidPullChanges,
   pushChanges,
@@ -27,14 +33,15 @@ export default async function synchronize({
   _unsafeBatchPerCollection,
   unsafeTurbo,
 }: SyncArgs): Promise<void> {
+
   const resetCount = database._resetCount
   log && (log.startedAt = new Date())
   log && (log.phase = 'starting')
 
   // TODO: Wrap the three computionally intensive phases in `requestIdleCallback`
 
   // pull phase
-  const lastPulledAt = await getLastPulledAt(database)
+  let lastPulledAt = await getLastPulledAt(database)
   log && (log.lastPulledAt = lastPulledAt)
 
   const { schemaVersion, migration, shouldSaveSchemaVersion } = await getMigrationInfo(
@@ -45,79 +52,101 @@ export default async function synchronize({
   )
   log && (log.phase = 'ready to pull')
 
-  // $FlowFixMe
-  const pullResult = await pullChanges({
-    lastPulledAt,
-    schemaVersion,
-    migration,
-  })
-  log && (log.phase = 'pulled')
-
-  let newLastPulledAt: Timestamp = (pullResult: any).timestamp
-  const remoteChangeCount = pullResult.changes ? changeSetCount(pullResult.changes) : NaN
+  let pullChunks: AsyncGenerator<SyncPullResult, void, void>
 
-  if (onWillApplyRemoteChanges) {
-    await onWillApplyRemoteChanges({ remoteChangeCount })
+  if(unsafePullChangesAsyncGenerator) {
+    // $FlowFixMe
+    pullChunks = unsafePullChangesAsyncGenerator({ lastPulledAt, schemaVersion, migration })
+  } else if(pullChanges) {
+    pullChunks = liftToAsyncGenerator(pullChanges({ lastPulledAt, schemaVersion, migration }))
+  } else {
+    invariant(false, 'Either pullChanges or unsafePullChangesAsyncGenerator must be provided')
   }
 
-  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.',
-    )
+  let newLastPulledAt: Timestamp | null = null
+  let result = await pullChunks.next()
+  log && (log.phase = 'pulled')
+
+  // To answer your question - yes it would be nice to use a for await loop here
+  // however, because of the use of 'fast-async' (see babel config), this syntax is *not* supported and will cause the code to hang...
+  while(!result.done) {
+    const pullResult = result.value
+    let chunkNewLastPulledAt: Timestamp = (pullResult: any).timestamp
+    newLastPulledAt = chunkNewLastPulledAt
+    const remoteChangeCount = pullResult.changes ? changeSetCount(pullResult.changes) : NaN
 
-    if (unsafeTurbo) {
+    if (onWillApplyRemoteChanges) {
+      await onWillApplyRemoteChanges({ remoteChangeCount })
+    }
+
+    await database.write(async () => {
+      ensureSameDatabase(database, resetCount)
       invariant(
-        !_unsafeBatchPerCollection,
-        'unsafeTurbo must not be used with _unsafeBatchPerCollection',
+        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 (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)
+        chunkNewLastPulledAt = resultRest.timestamp
+        onDidPullChanges && onDidPullChanges(resultRest)
+      }
+
+      log && (log.newLastPulledAt = chunkNewLastPulledAt)
       invariant(
-        'syncJson' in pullResult || 'syncJsonId' in pullResult,
-        'missing syncJson/syncJsonId',
+        typeof chunkNewLastPulledAt === 'number' && chunkNewLastPulledAt > 0,
+        `pullChanges() returned invalid timestamp ${chunkNewLastPulledAt}. timestamp must be a non-zero number`,
       )
-      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) {
+        // $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)
       }
 
-      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`,
-    )
-
-    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, chunkNewLastPulledAt)
 
-    log && (log.phase = 'applied remote changes')
-    await setLastPulledAt(database, newLastPulledAt)
+      if (shouldSaveSchemaVersion) {
+        await setLastPulledSchemaVersion(database, schemaVersion)
+      }
+    }, 'sync-synchronize-apply')
+    lastPulledAt = await getLastPulledAt(database)
+    result = await pullChunks.next()
+  }
 
-    if (shouldSaveSchemaVersion) {
-      await setLastPulledSchemaVersion(database, schemaVersion)
-    }
-  }, 'sync-synchronize-apply')
+  if (newLastPulledAt === null) {
+      invariant(
+        typeof newLastPulledAt === 'number',
+        `A pullChanges() function must yield at least one result`,
+      )
+  }
 
   // push phase
   if (pushChanges) {
diff --git a/src/sync/index.d.ts b/src/sync/index.d.ts
@@ -56,9 +56,8 @@ export type SyncConflictResolver = (
   resolved: DirtyRaw,
 ) => DirtyRaw
 
-export type SyncArgs = $Exact<{
+type BaseSyncArgs = $Exact<{
   database: Database
-  pullChanges: (_: SyncPullArgs) => Promise<SyncPullResult>
   pushChanges?: (_: SyncPushArgs) => Promise<SyncPushResult | undefined | void>
   // version at which support for migration syncs was added - the version BEFORE first syncable migration
   migrationsEnabledAtVersion?: SchemaVersion
@@ -88,6 +87,16 @@ export type SyncArgs = $Exact<{
   onWillApplyRemoteChanges?: (info: $Exact<{ remoteChangeCount: number }>) => Promise<void>
 }>
 
+type SimpleSyncArgs = $Exact<BaseSyncArgs & {
+  pullChanges: (_: SyncPullArgs) => Promise<SyncPullResult>
+}>
+
+type ChunkedSyncArgs = $Exact<BaseSyncArgs & {
+  unsafePullChangesAsyncGenerator: (_: SyncPullArgs) => AsyncGenerator<SyncPullResult, void, void>
+}>
+
+export type SyncArgs = SimpleSyncArgs | ChunkedSyncArgs
+
 export function synchronize(args: SyncArgs): Promise<void>
 
 export function hasUnsyncedChanges({ database }: $Exact<{ database: Database }>): Promise<boolean>
diff --git a/src/sync/index.js b/src/sync/index.js
@@ -85,7 +85,6 @@ export type SyncConflictResolver = (
 // TODO: JSDoc'ify this
 export type SyncArgs = $Exact<{
   database: Database,
-  pullChanges: (SyncPullArgs) => Promise<SyncPullResult>,
   pushChanges?: (SyncPushArgs) => Promise<?SyncPushResult>,
   // version at which support for migration syncs was added - the version BEFORE first syncable migration
   migrationsEnabledAtVersion?: SchemaVersion,
@@ -113,6 +112,8 @@ export type SyncArgs = $Exact<{
   // when processing a very large sync (could be useful for replacement syncs). Note that remote change count
   // is NaN in turbo mode.
   onWillApplyRemoteChanges?: (info: $Exact<{ remoteChangeCount: number }>) => Promise<void>,
+  pullChanges?: (_: SyncPullArgs) => Promise<SyncPullResult>,
+  unsafePullChangesAsyncGenerator?: (_: SyncPullArgs) => AsyncGenerator<SyncPullResult, void, void>,
 }>
 
 /**
diff --git a/yarn.lock b/yarn.lock
@@ -1231,6 +1231,15 @@
   dependencies:
     "@babel/helper-plugin-utils" "^7.25.9"
 
+"@babel/plugin-transform-async-generator-functions@^7.25.9":
+  version "7.25.9"
+  resolved "https://registry.yarnpkg.com/@babel/plugin-transform-async-generator-functions/-/plugin-transform-async-generator-functions-7.25.9.tgz#1b18530b077d18a407c494eb3d1d72da505283a2"
+  integrity sha512-RXV6QAzTBbhDMO9fWwOmwwTuYaiPbggWQ9INdZqAYeSHyG7FzQ+nOZaUUjNwKv9pV3aE4WFqFm1Hnbci5tBCAw==
+  dependencies:
+    "@babel/helper-plugin-utils" "^7.25.9"
+    "@babel/helper-remap-async-to-generator" "^7.25.9"
+    "@babel/traverse" "^7.25.9"
+
 "@babel/plugin-transform-async-to-generator@^7.20.0":
   version "7.24.7"
   resolved "https://registry.yarnpkg.com/@babel/plugin-transform-async-to-generator/-/plugin-transform-async-to-generator-7.24.7.tgz#72a3af6c451d575842a7e9b5a02863414355bdcc"

Original file line number	Diff line number	Diff line change
`@@ -41,6 +41,7 @@ const plugins = [`
`41`	`41`	`'@babel/plugin-proposal-nullish-coalescing-operator',`
`42`	`42`	`'@babel/plugin-transform-shorthand-properties',`
`43`	`43`	`'@babel/plugin-transform-spread',`
	`44`	`+ '@babel/plugin-transform-async-generator-functions',`
`44`	`45`	`[`
`45`	`46`	`'@babel/plugin-proposal-object-rest-spread',`
`46`	`47`	`{`