Support waitForPendingWrites in secondary tabs

Author: schmidt-sebastian

packages/firestore/src/core/sync_engine.ts

@@ -1185,6 +1185,7 @@ export async function applyBatchState(
     // NOTE: Both these methods are no-ops for batches that originated from
     // other clients.
     processUserCallback(syncEngineImpl, batchId, error ? error : null);
+    triggerPendingWritesCallbacks(syncEngineImpl, batchId);
     removeCachedMutationBatchMetadata(syncEngineImpl.localStore, batchId);
   } else {
     fail(`Unknown batchState: ${batchState}`);

packages/firestore/test/unit/specs/spec_builder.ts

@@ -793,6 +793,14 @@ export class SpecBuilder {
     return this;
   }
 
+  waitForPendingWrites() {
+    this.nextStep();
+    this.currentStep = {
+      waitForPendingWrites: true
+    };
+    return this;
+  }
+
   expectUserCallbacks(docs: {
     acknowledged?: string[];
     rejected?: string[];
@@ -944,6 +952,13 @@ export class SpecBuilder {
     return this;
   }
 
+  expectWaitForPendingWritesEvent(count = 1): this {
+    this.assertStep('Expectations require previous step');
+    const currentStep = this.currentStep!;
+    currentStep.expectedWaitForPendingWritesEvents = count;
+    return this;
+  }
+
   private static queryToSpec(query: Query): SpecQuery {
     // TODO(dimond): full query support
     const spec: SpecQuery = { path: query.path.canonicalString() };

packages/firestore/test/unit/specs/spec_test_runner.ts

@@ -44,6 +44,7 @@ import { SnapshotVersion } from '../../../src/core/snapshot_version';
 import {
   activeLimboDocumentResolutions,
   enqueuedLimboDocumentResolutions,
+  registerPendingWritesCallback,
   SyncEngine,
   syncEngineListen,
   syncEngineUnlisten,
@@ -151,6 +152,8 @@ import {
   SharedFakeWebStorage,
   testWindow
 } from '../../util/test_platform';
+import { waitForPendingWrites } from '../../../exp/src/api/database';
+import { getSyncEngine } from '../../../exp/src/api/components';
 
 const ARBITRARY_SEQUENCE_NUMBER = 2;
 
@@ -195,6 +198,7 @@ abstract class TestRunner {
   private eventList: QueryEvent[] = [];
   private acknowledgedDocs: string[];
   private rejectedDocs: string[];
+  private waitForPendingWritesEvents = 0;
   private snapshotsInSyncListeners: Array<Observer<void>>;
   private snapshotsInSyncEvents = 0;
 
@@ -342,6 +346,9 @@ abstract class TestRunner {
     this.validateExpectedSnapshotEvents(step.expectedSnapshotEvents!);
     await this.validateExpectedState(step.expectedState!);
     this.validateSnapshotsInSyncEvents(step.expectedSnapshotsInSyncEvents);
+    this.validateWaitForPendingWritesEvents(
+      step.expectedWaitForPendingWritesEvents
+    );
     this.eventList = [];
     this.rejectedDocs = [];
     this.acknowledgedDocs = [];
@@ -382,6 +389,8 @@ abstract class TestRunner {
       return this.doWriteAck(step.writeAck!);
     } else if ('failWrite' in step) {
       return this.doFailWrite(step.failWrite!);
+    } else if ('waitForPendingWrites' in step) {
+      return this.doWaitForPendingWrites();
     } else if ('runTimer' in step) {
       return this.doRunTimer(step.runTimer!);
     } else if ('drainQueue' in step) {
@@ -716,6 +725,18 @@ abstract class TestRunner {
     });
   }
 
+  private async doWaitForPendingWrites(): Promise<void> {
+    const deferred = new Deferred();
+    deferred.promise.then(() => {
+      ++this.waitForPendingWritesEvents;
+      console.log(this.waitForPendingWritesEvents);
+    });
+
+    return this.queue.enqueue(() =>
+      registerPendingWritesCallback(this.syncEngine, deferred)
+    );
+  }
+
   private async doRunTimer(timer: string): Promise<void> {
     // We assume the timer string is a valid TimerID enum value, but if it's
     // not, then there won't be a matching item on the queue and
@@ -912,10 +933,23 @@ abstract class TestRunner {
     }
   }
 
+  private validateWaitForPendingWritesEvents(
+    expectedCount: number | undefined
+  ): void {
+    expect(this.waitForPendingWritesEvents).to.eq(
+      expectedCount || 0,
+      'for waitForPendingWritesEvents'
+    );
+    this.waitForPendingWritesEvents = 0;
+  }
+
   private validateSnapshotsInSyncEvents(
     expectedCount: number | undefined
   ): void {
-    expect(this.snapshotsInSyncEvents).to.eq(expectedCount || 0);
+    expect(this.snapshotsInSyncEvents).to.eq(
+      expectedCount || 0,
+      'for snapshotsInSyncEvents'
+    );
     this.snapshotsInSyncEvents = 0;
   }
 
@@ -1334,6 +1368,8 @@ export interface SpecStep {
   writeAck?: SpecWriteAck;
   /** Fail a write */
   failWrite?: SpecWriteFailure;
+  /** Add a new `waitForPendingWrites` listener. */
+  waitForPendingWrites?: true;
 
   /** Fails the listed database actions. */
   failDatabase?: false | PersistenceAction[];
@@ -1387,6 +1423,12 @@ export interface SpecStep {
    * If not provided, the test will fail if the step causes events to be raised.
    */
   expectedSnapshotsInSyncEvents?: number;
+
+  /**
+   * Optional expected number of waitForPendingWrite callbacks to be called.
+   * If not provided, the test will fail if the step causes events to be raised.
+   */
+  expectedWaitForPendingWritesEvents?: number;
 }
 
 /** [<target-id>, <query-path>] */

packages/firestore/test/unit/specs/write_spec.test.ts

@@ -17,7 +17,7 @@
 
 import { Document } from '../../../src/model/document';
 import { TimerId } from '../../../src/util/async_queue';
-import { Code } from '../../../src/util/error';
+import { Code, FirestoreError } from '../../../src/util/error';
 import { doc, query } from '../../util/helpers';
 
 import { describeSpec, specTest } from './describe_spec';
@@ -1490,4 +1490,117 @@ describeSpec('Writes:', [], () => {
         .expectEvents(query1, { added: [docA, docB], fromCache: true });
     }
   );
+
+  specTest(
+    'Wait for pending writes resolves after write acknowledgment',
+    [],
+    () => {
+      return spec()
+        .userSets('collection/a', { k: 'a' })
+        .userSets('collection/b', { k: 'b' })
+        .waitForPendingWrites()
+        .writeAcks('collection/a', 1001)
+        .failWrite(
+          'collection/b',
+          new RpcError(Code.FAILED_PRECONDITION, 'Write error')
+        )
+        .expectWaitForPendingWritesEvent();
+    }
+  );
+
+  specTest('Wait for pending writes resolves with no writes', [], () => {
+    return spec().waitForPendingWrites().expectWaitForPendingWritesEvent();
+  });
+
+  specTest('Wait for pending writes resolves multiple times', [], () => {
+    return spec()
+      .userSets('collection/a', { k: 'a' })
+      .waitForPendingWrites()
+      .waitForPendingWrites()
+      .writeAcks('collection/a', 1001)
+      .expectWaitForPendingWritesEvent(2);
+  });
+
+  specTest(
+    'Wait for pending writes resolves if another write is issued',
+    [],
+    () => {
+      return spec()
+        .userSets('collection/a', { k: 'a' })
+        .waitForPendingWrites()
+        .userSets('collection/b', { k: 'b' })
+        .writeAcks('collection/a', 1001)
+        .expectWaitForPendingWritesEvent()
+        .writeAcks('collection/b', 1002);
+    }
+  );
+
+  specTest(
+    'Wait for pending writes waits after restart',
+    ['durable-persistence'],
+    () => {
+      return spec()
+        .userSets('collection/a', { k: 'a' })
+        .restart()
+        .waitForPendingWrites()
+        .writeAcks('collection/a', 1001, { expectUserCallback: false })
+        .expectWaitForPendingWritesEvent();
+    }
+  );
+
+  specTest(
+    'Wait for pending writes resolves for write in secondary tab',
+    ['multi-client'],
+    () => {
+      return client(0)
+        .expectPrimaryState(true)
+        .client(1)
+        .userSets('collection/a', { k: 'a' })
+        .waitForPendingWrites()
+        .client(0)
+        .writeAcks('collection/a', 1001, { expectUserCallback: false })
+        .client(1)
+        .expectUserCallbacks({ acknowledged: ['collection/a'] })
+        .expectWaitForPendingWritesEvent();
+    }
+  );
+
+  specTest(
+    'Wait for pending writes resolve independently for different tabs',
+    ['multi-client'],
+    () => {
+      return client(0)
+        .userSets('collection/a', { k: 'a' })
+        .waitForPendingWrites()
+        .client(1)
+        .userSets('collection/b', { k: 'b' })
+        .waitForPendingWrites()
+        .client(2)
+        .userSets('collection/c', { k: 'c' })
+        .waitForPendingWrites()
+        .client(0)
+        .writeAcks('collection/a', 1001)
+        .expectWaitForPendingWritesEvent(/* count= */ 1)
+        .client(1)
+        .expectWaitForPendingWritesEvent(/* count= */ 0)
+        .client(2)
+        .expectWaitForPendingWritesEvent(/* count= */ 0)
+        .client(0)
+        .writeAcks('collection/b', 1002, { expectUserCallback: false })
+        .expectWaitForPendingWritesEvent(/* count= */ 0)
+        .client(1)
+        .expectUserCallbacks({ acknowledged: ['collection/b'] })
+        .expectWaitForPendingWritesEvent(/* count= */ 1)
+        .client(2)
+        .expectWaitForPendingWritesEvent(/* count= */ 0)
+        .client(0)
+        .writeAcks('collection/c', 1003, { expectUserCallback: false })
+        .expectWaitForPendingWritesEvent(/* count= */ 0)
+        .client(1)
+        .expectWaitForPendingWritesEvent(/* count= */ 0)
+        .client(2)
+        .expectUserCallbacks({ acknowledged: ['collection/c'] })
+        .expectWaitForPendingWritesEvent(/* count= */ 1);
+    }
+  );
 });