b/68276665: Raise isFromCache=true events when offline (#358)

* Plumbs OnlineState changes through to views.
* View sets this.current to false on OnlineState.Failed, triggering
  isFromCache=true events. It will automatically be returned to true
  once the listen is reestablished and we get a new CURRENT message.
* Updated tests (and added one new one) to verify behavior.
* Unifies setOnlineStateToUnknown(), setOnlineStateToHealthy(), and
  updateAndBroadcastOnlineState() into a single updateOnlineState()
  method.
* Split disableNetwork() into `public disableNetwork()` and
  `private disableNetworkInternal(targetOnlineState: OnlineState)`.
* Some miscellaneous comment cleanup.

Author: mikelehen

packages/firestore/CHANGELOG.md

@@ -1,4 +1,11 @@
-# Unreleased
+# Unreleased (firestore-api-changes)
+- [changed] Snapshot listeners (with the `includeMetadataChanges` option
+  enabled) now receive an event with `snapshot.metadata.fromCache` set to
+  `true` if the SDK loses its connection to the backend. A new event with
+  `snapshot.metadata.fromCache` set to false will be raised once the
+  connection is restored and the query is in sync with the backend again.
+
+# v0.2.0
 - [feature] Added Node.js support for Cloud Firestore (with the exception of
   the offline persistence feature).
 - [changed] Webchannel requests use $httpHeaders URL parameter rather than

packages/firestore/src/core/firestore_client.ts

@@ -282,6 +282,7 @@ export class FirestoreClient {
         );
 
         const onlineStateChangedHandler = (onlineState: OnlineState) => {
+          this.syncEngine.applyOnlineStateChange(onlineState);
           this.eventMgr.onOnlineStateChanged(onlineState);
         };
 

packages/firestore/src/core/sync_engine.ts

@@ -42,7 +42,7 @@ import { Query } from './query';
 import { SnapshotVersion } from './snapshot_version';
 import { TargetIdGenerator } from './target_id_generator';
 import { Transaction } from './transaction';
-import { BatchId, ProtoByteString, TargetId } from './types';
+import { BatchId, OnlineState, ProtoByteString, TargetId } from './types';
 import {
   AddedLimboDocument,
   LimboDocumentChange,
@@ -343,6 +343,25 @@ export class SyncEngine implements RemoteSyncer {
     });
   }
 
+  /**
+   * Applies an OnlineState change to the sync engine and notifies any views of
+   * the change.
+   */
+  applyOnlineStateChange(onlineState: OnlineState) {
+    const newViewSnapshots = [] as ViewSnapshot[];
+    this.queryViewsByQuery.forEach((query, queryView) => {
+      const viewChange = queryView.view.applyOnlineStateChange(onlineState);
+      assert(
+        viewChange.limboChanges.length === 0,
+        'OnlineState should not affect limbo documents.'
+      );
+      if (viewChange.snapshot) {
+        newViewSnapshots.push(viewChange.snapshot);
+      }
+    });
+    this.viewHandler(newViewSnapshots);
+  }
+
   rejectListen(targetId: TargetId, err: FirestoreError): Promise<void> {
     this.assertSubscribed('rejectListens()');
     const limboKey = this.limboKeysByTarget[targetId];

packages/firestore/src/core/types.ts

@@ -34,22 +34,22 @@ export type ProtoByteString = Uint8Array | string;
 export enum OnlineState {
   /**
    * The Firestore client is in an unknown online state. This means the client
-   * is either not actively trying to establish a connection or it was
-   * previously in an unknown state and is trying to establish a connection.
+   * is either not actively trying to establish a connection or it is currently
+   * trying to establish a connection, but it has not succeeded or failed yet.
    */
   Unknown,
 
   /**
    * The client is connected and the connections are healthy. This state is
    * reached after a successful connection and there has been at least one
-   * succesful message received from the backends.
+   * successful message received from the backends.
    */
   Healthy,
 
   /**
-   * The client has tried to establish a connection but has failed.
-   * This state is reached after either a connection attempt failed or a
-   * healthy stream was closed for unexpected reasons.
+   * The client considers itself offline. It is either trying to establish a
+   * connection but failing, or it has been explicitly marked offline via a call
+   * to disableNetwork().
    */
   Failed
 }

packages/firestore/src/core/view.ts

@@ -31,6 +31,7 @@ import {
 import { assert, fail } from '../util/assert';
 
 import { Query } from './query';
+import { OnlineState } from './types';
 import {
   ChangeType,
   DocumentChangeSet,
@@ -50,7 +51,7 @@ export class RemovedLimboDocument {
 export interface ViewDocumentChanges {
   /** The new set of docs that should be in the view. */
   documentSet: DocumentSet;
-  /** The diff of this these docs with the previous set of docs. */
+  /** The diff of these docs with the previous set of docs. */
   changeSet: DocumentChangeSet;
   /**
    * Whether the set of documents passed in was not sufficient to calculate the
@@ -268,6 +269,29 @@ export class View {
     }
   }
 
+  /**
+   * Applies an OnlineState change to the view, potentially generating a
+   * ViewChange if the view's syncState changes as a result.
+   */
+  applyOnlineStateChange(onlineState: OnlineState): ViewChange {
+    if (this.current && onlineState === OnlineState.Failed) {
+      // If we're offline, set `current` to false and then call applyChanges()
+      // to refresh our syncState and generate a ViewChange as appropriate. We
+      // are guaranteed to get a new TargetChange that sets `current` back to
+      // true once the client is back online.
+      this.current = false;
+      return this.applyChanges({
+        documentSet: this.documentSet,
+        changeSet: new DocumentChangeSet(),
+        mutatedKeys: this.mutatedKeys,
+        needsRefill: false
+      });
+    } else {
+      // No effect, just return a no-op ViewChange.
+      return { limboChanges: [] };
+    }
+  }
+
   /**
    * Returns whether the doc for the given key should be in limbo.
    */

packages/firestore/src/remote/remote_store.ts

@@ -153,48 +153,53 @@ export class RemoteStore {
     return this.enableNetwork();
   }
 
-  private setOnlineStateToHealthy(): void {
-    this.shouldWarnOffline = false;
-    this.updateAndBroadcastOnlineState(OnlineState.Healthy);
-  }
+  /**
+   * Updates our OnlineState to the new state, updating local state
+   * and notifying the onlineStateHandler as appropriate.
+   */
+  private updateOnlineState(newState: OnlineState) {
+    if (newState === OnlineState.Healthy) {
+      // We've connected to watch at least once. Don't warn the developer about
+      // being offline going forward.
+      this.shouldWarnOffline = false;
+    } else if (newState === OnlineState.Unknown) {
+      // The state is set to unknown when a healthy stream is closed (e.g. due to
+      // a token timeout) or when we have no active listens and therefore there's
+      // no need to start the stream. Assuming there is (possibly in the future)
+      // an active listen, then we will eventually move to state Online or Failed,
+      // but we always want to make at least ONLINE_ATTEMPTS_BEFORE_FAILURE
+      // attempts before failing, so we reset the count here.
+      this.watchStreamFailures = 0;
+    }
 
-  private setOnlineStateToUnknown(): void {
-    // The state is set to unknown when a healthy stream is closed (e.g. due to
-    // a token timeout) or when we have no active listens and therefore there's
-    // no need to start the stream. Assuming there is (possibly in the future)
-    // an active listen, then we will eventually move to state Online or Failed,
-    // but we always want to make at least ONLINE_ATTEMPTS_BEFORE_FAILURE
-    // attempts before failing, so we reset the count here.
-    this.watchStreamFailures = 0;
-    this.updateAndBroadcastOnlineState(OnlineState.Unknown);
+    // Update and broadcast the new state.
+    if (newState !== this.watchStreamOnlineState) {
+      this.watchStreamOnlineState = newState;
+      this.onlineStateHandler(newState);
+    }
   }
 
+  /**
+   * Updates our OnlineState as appropriate after the watch stream reports a
+   * failure. The first failure moves us to the 'Unknown' state. We then may
+   * allow multiple failures (based on ONLINE_ATTEMPTS_BEFORE_FAILURE) before we
+   * actually transition to OnlineState.Failed.
+   */
   private updateOnlineStateAfterFailure(): void {
-    // The first failure after we are successfully connected moves us to the
-    // 'Unknown' state. We then may make multiple attempts (based on
-    // ONLINE_ATTEMPTS_BEFORE_FAILURE) before we actually report failure.
     if (this.watchStreamOnlineState === OnlineState.Healthy) {
-      this.setOnlineStateToUnknown();
+      this.updateOnlineState(OnlineState.Unknown);
     } else {
       this.watchStreamFailures++;
       if (this.watchStreamFailures >= ONLINE_ATTEMPTS_BEFORE_FAILURE) {
         if (this.shouldWarnOffline) {
           log.debug(LOG_TAG, 'Could not reach Firestore backend.');
           this.shouldWarnOffline = false;
         }
-        this.updateAndBroadcastOnlineState(OnlineState.Failed);
+        this.updateOnlineState(OnlineState.Failed);
       }
     }
   }
 
-  private updateAndBroadcastOnlineState(onlineState: OnlineState): void {
-    const didChange = this.watchStreamOnlineState !== onlineState;
-    this.watchStreamOnlineState = onlineState;
-    if (didChange) {
-      this.onlineStateHandler(onlineState);
-    }
-  }
-
   private isNetworkEnabled(): boolean {
     assert(
       (this.watchStream == null) == (this.writeStream == null),
@@ -226,16 +231,28 @@ export class RemoteStore {
         this.startWatchStream();
       }
 
-      this.updateAndBroadcastOnlineState(OnlineState.Unknown);
+      this.updateOnlineState(OnlineState.Unknown);
 
       return this.fillWritePipeline(); // This may start the writeStream.
     });
   }
 
-  /** Temporarily disables the network. The network can be re-enabled using enableNetwork(). */
+  /**
+   * Temporarily disables the network. The network can be re-enabled using
+   * enableNetwork().
+   */
   disableNetwork(): Promise<void> {
-    this.updateAndBroadcastOnlineState(OnlineState.Failed);
+    // Set the OnlineState to failed so get()'s return from cache, etc.
+    return this.disableNetworkInternal(OnlineState.Failed);
+  }
 
+  /**
+   * Disables the network, setting the OnlineState to the specified
+   * targetOnlineState.
+   */
+  private disableNetworkInternal(
+    targetOnlineState: OnlineState
+  ): Promise<void> {
     // NOTE: We're guaranteed not to get any further events from these streams (not even a close
     // event).
     this.watchStream.stop();
@@ -247,12 +264,16 @@ export class RemoteStore {
     this.writeStream = null;
     this.watchStream = null;
 
+    this.updateOnlineState(targetOnlineState);
+
     return Promise.resolve();
   }
 
   shutdown(): Promise<void> {
     log.debug(LOG_TAG, 'RemoteStore shutting down.');
-    this.disableNetwork();
+    // Set the OnlineState to Unknown (rather than Failed) to avoid potentially
+    // triggering spurious listener events with cached data, etc.
+    this.disableNetworkInternal(OnlineState.Unknown);
     return Promise.resolve(undefined);
   }
 
@@ -376,7 +397,7 @@ export class RemoteStore {
       // No need to restart watch stream because there are no active targets.
       // The online state is set to unknown because there is no active attempt
       // at establishing a connection
-      this.setOnlineStateToUnknown();
+      this.updateOnlineState(OnlineState.Unknown);
     }
     return Promise.resolve();
   }
@@ -386,7 +407,7 @@ export class RemoteStore {
     snapshotVersion: SnapshotVersion
   ): Promise<void> {
     // Mark the connection as healthy because we got a message from the server
-    this.setOnlineStateToHealthy();
+    this.updateOnlineState(OnlineState.Healthy);
 
     if (
       watchChange instanceof WatchTargetChange &&
@@ -799,7 +820,7 @@ export class RemoteStore {
     // Tear down and re-create our network streams. This will ensure we get a fresh auth token
     // for the new user and re-fill the write pipeline with new mutations from the LocalStore
     // (since mutations are per-user).
-    this.disableNetwork();
+    this.disableNetworkInternal(OnlineState.Unknown);
     return this.enableNetwork();
   }
 }

packages/firestore/test/integration/api/query.test.ts

@@ -458,4 +458,41 @@ apiDescribe('Queries', persistence => {
       return deferred.promise.then(unregister);
     });
   });
+
+  it('Queries trigger with isFromCache=true when offline', () => {
+    return withTestCollection(persistence, { a: { foo: 1 } }, coll => {
+      // TODO(mikelehen): Find better way to expose this to tests.
+      // tslint:disable-next-line:no-any enableNetwork isn't exposed via d.ts
+      const firestoreInternal = coll.firestore.INTERNAL as any;
+
+      const accum = new EventsAccumulator<firestore.QuerySnapshot>();
+      const unregister = coll.onSnapshot(
+        { includeQueryMetadataChanges: true },
+        accum.storeEvent
+      );
+
+      return accum
+        .awaitEvent()
+        .then(querySnap => {
+          // initial event
+          expect(querySnap.docs.map(doc => doc.data())).to.deep.equal([
+            { foo: 1 }
+          ]);
+          expect(querySnap.metadata.fromCache).to.be.false;
+        })
+        .then(() => firestoreInternal.disableNetwork())
+        .then(() => accum.awaitEvent())
+        .then(querySnap => {
+          // offline event with fromCache = true
+          expect(querySnap.metadata.fromCache).to.be.true;
+        })
+        .then(() => firestoreInternal.enableNetwork())
+        .then(() => accum.awaitEvent())
+        .then(querySnap => {
+          // back online event with fromCache = false
+          expect(querySnap.metadata.fromCache).to.be.false;
+          unregister();
+        });
+    });
+  });
 });

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

@@ -253,6 +253,7 @@ describeSpec('Listens:', [], () => {
       .watchAcksFull(query, 1000, docA)
       .expectEvents(query, { added: [docA] })
       .disableNetwork()
+      .expectEvents(query, { fromCache: true })
       .enableNetwork()
       .restoreListen(query, 'resume-token-1000')
       .expectWatchStreamRequestCount(

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

@@ -17,7 +17,7 @@
 import { expect } from 'chai';
 import { Query } from '../../../src/core/query';
 import { Code } from '../../../src/util/error';
-import { path } from '../../util/helpers';
+import { doc, path } from '../../util/helpers';
 
 import { describeSpec, specTest } from './describe_spec';
 import { spec } from './spec_builder';
@@ -94,4 +94,25 @@ describeSpec('Offline:', [], () => {
       );
     }
   );
+
+  specTest('Queries revert to fromCache=true when offline.', [], () => {
+    const query = Query.atPath(path('collection'));
+    const docA = doc('collection/a', 1000, { key: 'a' });
+    return (
+      spec()
+        .userListens(query)
+        .watchAcksFull(query, 1000, docA)
+        .expectEvents(query, { added: [docA] })
+        // first error triggers unknown state
+        .watchStreamCloses(Code.UNAVAILABLE)
+        .restoreListen(query, 'resume-token-1000')
+        // getting two more errors triggers offline state and fromCache: true
+        .watchStreamCloses(Code.UNAVAILABLE)
+        .watchStreamCloses(Code.UNAVAILABLE)
+        .expectEvents(query, { fromCache: true })
+        // Going online and getting a CURRENT message triggers fromCache: false
+        .watchAcksFull(query, 1000)
+        .expectEvents(query, { fromCache: false })
+    );
+  });
 });

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

@@ -384,6 +384,7 @@ abstract class TestRunner {
       initialBackoffDelay
     );
     const onlineStateChangedHandler = (onlineState: OnlineState) => {
+      this.syncEngine.applyOnlineStateChange(onlineState);
       this.eventManager.onOnlineStateChanged(onlineState);
     };
     this.remoteStore = new RemoteStore(
@@ -498,7 +499,10 @@ abstract class TestRunner {
 
     await this.queue.schedule(async () => {
       const targetId = await this.eventManager.listen(queryListener);
-      expect(targetId).to.equal(expectedTargetId);
+      expect(targetId).to.equal(
+        expectedTargetId,
+        'targetId assigned to listen'
+      );
     });
     // Open should always have happened after a listen
     await this.connection.waitForWatchOpen();