Firestore: QueryTest.java: improve the test that resumes a query with existence filter to actually validate the existence filter.

firebase-firestore/src/androidTest/java/com/google/firebase/firestore/QueryTest.java

@@ -37,6 +37,7 @@
 import com.google.android.gms.tasks.Task;
 import com.google.common.collect.Lists;
 import com.google.firebase.firestore.Query.Direction;
+import com.google.firebase.firestore.remote.WatchChangeAggregatorTestingHooksAccessor;
 import com.google.firebase.firestore.testutil.EventAccumulator;
 import com.google.firebase.firestore.testutil.IntegrationTestUtil;
 import java.util.ArrayList;
@@ -1053,6 +1054,7 @@ public void resumingAQueryShouldUseExistenceFilterToDetectDeletes() throws Excep
         createdDocuments.add(documentSnapshot.getReference());
       }
     }
+    assertWithMessage("createdDocuments").that(createdDocuments).hasSize(100);
 
     // Delete 50 of the 100 documents. Do this in a transaction, rather than
     // DocumentReference.delete(), to avoid affecting the local cache.
@@ -1069,13 +1071,35 @@ public void resumingAQueryShouldUseExistenceFilterToDetectDeletes() throws Excep
                   }
                   return null;
                 }));
+    assertWithMessage("deletedDocumentIds").that(deletedDocumentIds).hasSize(50);
 
     // Wait for 10 seconds, during which Watch will stop tracking the query and will send an
     // existence filter rather than "delete" events when the query is resumed.
     Thread.sleep(10000);
 
-    // Resume the query and save the resulting snapshot for verification.
-    QuerySnapshot snapshot2 = waitFor(collection.get());
+    // Resume the query and save the resulting snapshot for verification. Use some internal testing
+    // hooks to "capture" the existence filter mismatches to verify them.
+    QuerySnapshot snapshot2;
+    WatchChangeAggregatorTestingHooksAccessor.ExistenceFilterMismatchInfo
+        existenceFilterMismatchInfo;
+    WatchChangeAggregatorTestingHooksAccessor.ExistenceFilterMismatchAccumulator
+        existenceFilterMismatchAccumulator =
+            new WatchChangeAggregatorTestingHooksAccessor.ExistenceFilterMismatchAccumulator();
+    existenceFilterMismatchAccumulator.register();
+    try {
+      snapshot2 = waitFor(collection.get());
+      // TODO(b/270731363): Remove the "if" condition below once the Firestore Emulator is fixed
+      //  to send an existence filter.
+      if (isRunningAgainstEmulator()) {
+        existenceFilterMismatchInfo = null;
+      } else {
+        existenceFilterMismatchInfo =
+            existenceFilterMismatchAccumulator.waitForExistenceFilterMismatch(
+                /*timeoutMillis=*/ 5000);
+      }
+    } finally {
+      existenceFilterMismatchAccumulator.unregister();
+    }
 
     // Verify that the snapshot from the resumed query contains the expected documents; that is,
     // that it contains the 50 documents that were _not_ deleted.
@@ -1098,6 +1122,26 @@ public void resumingAQueryShouldUseExistenceFilterToDetectDeletes() throws Excep
           .that(actualDocumentIds)
           .containsExactlyElementsIn(expectedDocumentIds);
     }
+
+    // Skip the verification of the existence filter mismatch when testing against the Firestore
+    // emulator because the Firestore emulator fails to to send an existence filter at all.
+    // TODO(b/270731363): Enable the verification of the existence filter mismatch once the
+    // Firestore emulator is fixed to send an existence filter.
+    if (isRunningAgainstEmulator()) {
+      return;
+    }
+
+    // Verify that Watch sent an existence filter with the correct counts when the query was
+    // resumed.
+    assertWithMessage("Watch should have sent an existence filter")
+        .that(existenceFilterMismatchInfo)
+        .isNotNull();
+    assertWithMessage("localCacheCount")
+        .that(existenceFilterMismatchInfo.localCacheCount())
+        .isEqualTo(100);
+    assertWithMessage("existenceFilterCount")
+        .that(existenceFilterMismatchInfo.existenceFilterCount())
+        .isEqualTo(50);
   }
 
   @Test

firebase-firestore/src/androidTest/java/com/google/firebase/firestore/remote/WatchChangeAggregatorTestingHooksAccessor.java

@@ -0,0 +1,173 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.firebase.firestore.remote;
+
+import static com.google.firebase.firestore.util.Preconditions.checkNotNull;
+
+import android.os.SystemClock;
+import androidx.annotation.AnyThread;
+import androidx.annotation.NonNull;
+import androidx.annotation.Nullable;
+import com.google.firebase.firestore.ListenerRegistration;
+import java.util.ArrayList;
+
+/**
+ * Provides access to the {@link WatchChangeAggregatorTestingHooks} class and its methods.
+ *
+ * <p>The {@link WatchChangeAggregatorTestingHooks} class has default visibility, and, therefore, is
+ * only visible to other classes declared in the same package. This class effectively "re-exports"
+ * the functionality from {@link WatchChangeAggregatorTestingHooks} in a class with {@code public}
+ * visibility so that tests written in other packages can access its functionality.
+ */
+public final class WatchChangeAggregatorTestingHooksAccessor {
+
+  private WatchChangeAggregatorTestingHooksAccessor() {}
+
+  /** @see WatchChangeAggregatorTestingHooks#addExistenceFilterMismatchListener */
+  public static ListenerRegistration addExistenceFilterMismatchListener(
+      @NonNull ExistenceFilterMismatchListener listener) {
+    checkNotNull(listener, "a null listener is not allowed");
+    return WatchChangeAggregatorTestingHooks.addExistenceFilterMismatchListener(
+        new ExistenceFilterMismatchListenerWrapper(listener));
+  }
+
+  /** @see WatchChangeAggregatorTestingHooks.ExistenceFilterMismatchListener */
+  public interface ExistenceFilterMismatchListener {
+    @AnyThread
+    void onExistenceFilterMismatch(ExistenceFilterMismatchInfo info);
+  }
+
+  /** @see WatchChangeAggregatorTestingHooks.ExistenceFilterMismatchInfo */
+  public interface ExistenceFilterMismatchInfo {
+    int localCacheCount();
+
+    int existenceFilterCount();
+  }
+
+  private static final class ExistenceFilterMismatchInfoImpl
+      implements ExistenceFilterMismatchInfo {
+
+    private final WatchChangeAggregatorTestingHooks.ExistenceFilterMismatchInfo info;
+
+    ExistenceFilterMismatchInfoImpl(
+        @NonNull WatchChangeAggregatorTestingHooks.ExistenceFilterMismatchInfo info) {
+      this.info = info;
+    }
+
+    @Override
+    public int localCacheCount() {
+      return info.localCacheCount();
+    }
+
+    @Override
+    public int existenceFilterCount() {
+      return info.existenceFilterCount();
+    }
+  }
+
+  private static final class ExistenceFilterMismatchListenerWrapper
+      implements WatchChangeAggregatorTestingHooks.ExistenceFilterMismatchListener {
+
+    private final ExistenceFilterMismatchListener wrappedListener;
+
+    ExistenceFilterMismatchListenerWrapper(
+        @NonNull ExistenceFilterMismatchListener listenerToWrap) {
+      this.wrappedListener = listenerToWrap;
+    }
+
+    @Override
+    public void onExistenceFilterMismatch(
+        WatchChangeAggregatorTestingHooks.ExistenceFilterMismatchInfo info) {
+      this.wrappedListener.onExistenceFilterMismatch(new ExistenceFilterMismatchInfoImpl(info));
+    }
+  }
+
+  public static final class ExistenceFilterMismatchAccumulator {
+
+    private ExistenceFilterMismatchListenerImpl listener;
+    private ListenerRegistration listenerRegistration = null;
+
+    /** Registers the accumulator to begin listening for existence filter mismatches. */
+    public synchronized void register() {
+      if (listener != null) {
+        throw new IllegalStateException("already registered");
+      }
+      listener = new ExistenceFilterMismatchListenerImpl();
+      listenerRegistration =
+          WatchChangeAggregatorTestingHooksAccessor.addExistenceFilterMismatchListener(listener);
+    }
+
+    /** Unregisters the accumulator from listening for existence filter mismatches. */
+    public synchronized void unregister() {
+      if (listener == null) {
+        return;
+      }
+      listenerRegistration.remove();
+      listenerRegistration = null;
+      listener = null;
+    }
+
+    @Nullable
+    public WatchChangeAggregatorTestingHooksAccessor.ExistenceFilterMismatchInfo
+        waitForExistenceFilterMismatch(long timeoutMillis) throws InterruptedException {
+      ExistenceFilterMismatchListenerImpl capturedListener;
+      synchronized (this) {
+        capturedListener = listener;
+      }
+      if (capturedListener == null) {
+        throw new IllegalStateException(
+            "must be registered before waiting for an existence filter mismatch");
+      }
+      return capturedListener.waitForExistenceFilterMismatch(timeoutMillis);
+    }
+
+    private static final class ExistenceFilterMismatchListenerImpl
+        implements WatchChangeAggregatorTestingHooksAccessor.ExistenceFilterMismatchListener {
+
+      private final ArrayList<ExistenceFilterMismatchInfo> existenceFilterMismatches =
+          new ArrayList<>();
+
+      @Override
+      public void onExistenceFilterMismatch(
+          WatchChangeAggregatorTestingHooksAccessor.ExistenceFilterMismatchInfo info) {
+        synchronized (existenceFilterMismatches) {
+          existenceFilterMismatches.add(info);
+          existenceFilterMismatches.notifyAll();
+        }
+      }
+
+      @Nullable
+      WatchChangeAggregatorTestingHooksAccessor.ExistenceFilterMismatchInfo
+          waitForExistenceFilterMismatch(long timeoutMillis) throws InterruptedException {
+        if (timeoutMillis <= 0) {
+          throw new IllegalArgumentException("invalid timeout: " + timeoutMillis);
+        }
+        synchronized (existenceFilterMismatches) {
+          long endTimeMillis = SystemClock.uptimeMillis() + timeoutMillis;
+          while (true) {
+            if (existenceFilterMismatches.size() > 0) {
+              return existenceFilterMismatches.remove(0);
+            }
+            long currentWaitMillis = endTimeMillis - SystemClock.uptimeMillis();
+            if (currentWaitMillis <= 0) {
+              return null;
+            }
+            existenceFilterMismatches.wait(currentWaitMillis);
+          }
+        }
+      }
+    }
+  }
+}

firebase-firestore/src/main/java/com/google/firebase/firestore/remote/WatchChangeAggregator.java

@@ -202,6 +202,10 @@ public void handleExistenceFilter(ExistenceFilterWatchChange watchChange) {
           // `isFromCache:true`.
           resetTarget(targetId);
           pendingTargetResets.add(targetId);
+
+          WatchChangeAggregatorTestingHooks.notifyOnExistenceFilterMismatch(
+              WatchChangeAggregatorTestingHooks.ExistenceFilterMismatchInfo.from(
+                  (int) currentSize, watchChange.getExistenceFilter()));
         }
       }
     }

firebase-firestore/src/main/java/com/google/firebase/firestore/remote/WatchChangeAggregatorTestingHooks.java

@@ -0,0 +1,110 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.firebase.firestore.remote;
+
+import static com.google.firebase.firestore.util.Preconditions.checkNotNull;
+
+import androidx.annotation.AnyThread;
+import androidx.annotation.NonNull;
+import androidx.annotation.VisibleForTesting;
+import com.google.auto.value.AutoValue;
+import com.google.firebase.firestore.ListenerRegistration;
+import com.google.firebase.firestore.util.Executors;
+import java.util.HashMap;
+import java.util.Map;
+
+final class WatchChangeAggregatorTestingHooks {
+
+  private WatchChangeAggregatorTestingHooks() {}
+
+  private static final Map<Object, ExistenceFilterMismatchListener>
+      existenceFilterMismatchListeners = new HashMap<>();
+
+  /**
+   * Notifies all registered {@link ExistenceFilterMismatchListener}` listeners registered via
+   * {@link #addExistenceFilterMismatchListener}.
+   *
+   * @param info Information about the existence filter mismatch to deliver to the listeners.
+   */
+  static void notifyOnExistenceFilterMismatch(ExistenceFilterMismatchInfo info) {
+    synchronized (existenceFilterMismatchListeners) {
+      for (ExistenceFilterMismatchListener listener : existenceFilterMismatchListeners.values()) {
+        Executors.BACKGROUND_EXECUTOR.execute(() -> listener.onExistenceFilterMismatch(info));
+      }
+    }
+  }
+
+  /**
+   * Registers a {@link ExistenceFilterMismatchListener} to be notified when an existence filter
+   * mismatch occurs in the Watch listen stream.
+   *
+   * <p>The relative order in which callbacks are notified is unspecified; do not rely on any
+   * particular ordering. If a given callback is registered multiple times then it will be notified
+   * multiple times, once per registration.
+   *
+   * <p>The thread on which the callback occurs is unspecified; listeners should perform their work
+   * as quickly as possible and return to avoid blocking any critical work. In particular, the
+   * listener callbacks should <em>not</em> block or perform long-running operations. Listener
+   * callbacks can occur concurrently with other callbacks on the same and other listeners.
+   *
+   * @param listener the listener to register.
+   * @return an object that unregisters the given listener via its {@link
+   *     ListenerRegistration#remove} method; only the first unregistration request does anything;
+   *     all subsequent requests do nothing.
+   */
+  @VisibleForTesting
+  static ListenerRegistration addExistenceFilterMismatchListener(
+      @NonNull ExistenceFilterMismatchListener listener) {
+    checkNotNull(listener, "a null listener is not allowed");
+
+    Object listenerId = new Object();
+    synchronized (existenceFilterMismatchListeners) {
+      existenceFilterMismatchListeners.put(listenerId, listener);
+    }
+
+    return () -> {
+      synchronized (existenceFilterMismatchListeners) {
+        existenceFilterMismatchListeners.remove(listenerId);
+      }
+    };
+  }
+
+  interface ExistenceFilterMismatchListener {
+    @AnyThread
+    void onExistenceFilterMismatch(ExistenceFilterMismatchInfo info);
+  }
+
+  @AutoValue
+  abstract static class ExistenceFilterMismatchInfo {
+
+    static ExistenceFilterMismatchInfo create(int localCacheCount, int existenceFilterCount) {
+      return new AutoValue_WatchChangeAggregatorTestingHooks_ExistenceFilterMismatchInfo(
+          localCacheCount, existenceFilterCount);
+    }
+
+    /** Returns the number of documents that matched the query in the local cache. */
+    abstract int localCacheCount();
+
+    /**
+     * Returns the number of documents that matched the query on the server, as specified in the
+     * ExistenceFilter message's `count` field.
+     */
+    abstract int existenceFilterCount();
+
+    static ExistenceFilterMismatchInfo from(int localCacheCount, ExistenceFilter existenceFilter) {
+      return create(localCacheCount, existenceFilter.getCount());
+    }
+  }
+}