[Invalidaton] Reduce cache size for unacked invalidations

All InvalidationHandler instances are looking at the latest invalidation
per topic only in OnIncomingInvalidation:

This CL therefore reduces the cache size for unacknowledged
invalidations to at most one invalidation per topic.

Bug: b/305177486
Change-Id: I71746f070d1ce42e5c76ee08877121a9cddc62f3
Tests: Adjusted unit tests
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/4941730
Commit-Queue: Roland Bock <rbock@google.com>
Reviewed-by: Artem Sumaneev <asumaneev@google.com>
Cr-Commit-Position: refs/heads/main@{#1210121}

components/invalidation/impl/BUILD.gn

@@ -47,8 +47,6 @@ static_library("impl") {
     "profile_invalidation_provider.h",
     "status.cc",
     "status.h",
-    "unacked_invalidation_set.cc",
-    "unacked_invalidation_set.h",
   ]
 
   public_deps = [
@@ -107,7 +105,6 @@ source_set("unit_tests") {
     sources += [
       "single_topic_invalidation_set_unittest.cc",
       "topic_invalidation_map_unittest.cc",
-      "unacked_invalidation_set_unittest.cc",
     ]
   }
 }

components/invalidation/impl/fcm_invalidation_listener.cc

@@ -5,15 +5,36 @@
 #include "components/invalidation/impl/fcm_invalidation_listener.h"
 
 #include "base/functional/bind.h"
-#include "base/functional/callback.h"
 #include "base/logging.h"
 #include "base/task/single_thread_task_runner.h"
+#include "components/invalidation/public/invalidation.h"
 #include "components/invalidation/public/invalidation_util.h"
 #include "components/invalidation/public/topic_invalidation_map.h"
-#include "components/prefs/pref_service.h"
 
 namespace invalidation {
 
+namespace {
+
+// Insert or update the invalidation in the map at `invalidation.topic()`.
+// If `map` does not have an invalidation for that topic, a copy of `inv` will
+// be inserted.
+// Otherwise, the existing invalidation for the topic will be replaced by `inv`
+// if and only if `inv` has a higher version than `map.at(inv.topic())`.
+void Upsert(std::map<Topic, Invalidation>& map,
+            const Invalidation& invalidation) {
+  auto it = map.find(invalidation.topic());
+  if (it == map.end()) {
+    map.emplace(invalidation.topic(), invalidation);
+    return;
+  }
+  if (it->second.version() < invalidation.version()) {
+    it->second = invalidation;
+    return;
+  }
+}
+
+}  // namespace
+
 FCMInvalidationListener::FCMInvalidationListener(
     std::unique_ptr<FCMSyncNetworkChannel> network_channel)
     : network_channel_(std::move(network_channel)) {
@@ -76,38 +97,26 @@ void FCMInvalidationListener::InvalidationReceived(
              << expected_public_topic.value_or("<None>");
     return;
   }
-  TopicInvalidationMap invalidations;
   Invalidation inv =
       Invalidation::Init(*expected_public_topic, version, payload);
   inv.SetAckHandler(weak_factory_.GetWeakPtr(),
                     base::SingleThreadTaskRunner::GetCurrentDefault());
   DVLOG(1) << "Received invalidation with version " << inv.version() << " for "
            << *expected_public_topic;
 
-  invalidations.Insert(inv);
-  DispatchInvalidations(invalidations);
+  DispatchInvalidation(inv);
 }
 
-void FCMInvalidationListener::DispatchInvalidations(
-    const TopicInvalidationMap& invalidations) {
-  TopicInvalidationMap to_save = invalidations;
-  TopicInvalidationMap to_emit =
-      invalidations.GetSubsetWithTopics(interested_topics_);
+void FCMInvalidationListener::DispatchInvalidation(
+    const Invalidation& invalidation) {
+  // Cache invalidation
+  Upsert(unacked_invalidations_map_, invalidation);
 
-  SaveInvalidations(to_save);
-  EmitSavedInvalidations(to_emit);
-}
-
-void FCMInvalidationListener::SaveInvalidations(
-    const TopicInvalidationMap& to_save) {
-  for (const Topic& topic : to_save.GetTopics()) {
-    auto lookup = unacked_invalidations_map_.find(topic);
-    if (lookup == unacked_invalidations_map_.end()) {
-      lookup = unacked_invalidations_map_
-                   .emplace(topic, UnackedInvalidationSet(topic))
-                   .first;
-    }
-    lookup->second.AddSet(to_save.ForTopic(topic));
+  // Emit invalidation to registered handlers (if any).
+  if (interested_topics_.contains(invalidation.topic())) {
+    TopicInvalidationMap topic_invalidation_map;
+    topic_invalidation_map.Insert(invalidation);
+    EmitSavedInvalidations(topic_invalidation_map);
   }
 }
 
@@ -135,7 +144,11 @@ void FCMInvalidationListener::Acknowledge(const Topic& topic,
     DLOG(WARNING) << "Received acknowledgement for untracked topic";
     return;
   }
-  lookup->second.Acknowledge(handle);
+  if (lookup->second.ack_handle().Equals(handle)) {
+    unacked_invalidations_map_.erase(topic);
+    return;
+  }
+  DLOG(WARNING) << "Unrecognized to ack for topic " << topic;
 }
 
 void FCMInvalidationListener::DoSubscriptionUpdate() {
@@ -152,15 +165,11 @@ void FCMInvalidationListener::DoSubscriptionUpdate() {
   // already dispatched but not acked yet.
   // TODO(melandory): remove unacked invalidations for unregistered topics.
   TopicInvalidationMap topic_invalidation_map;
-  for (const auto& unacked : unacked_invalidations_map_) {
-    if (interested_topics_.find(unacked.first) == interested_topics_.end()) {
+  for (const auto& [topic, invalidation] : unacked_invalidations_map_) {
+    if (interested_topics_.find(topic) == interested_topics_.end()) {
       continue;
     }
-
-    unacked.second.ExportInvalidations(
-        weak_factory_.GetWeakPtr(),
-        base::SingleThreadTaskRunner::GetCurrentDefault(),
-        &topic_invalidation_map);
+    topic_invalidation_map.Insert(invalidation);
   }
 
   // There's no need to run these through DispatchInvalidations(); they've

components/invalidation/impl/fcm_invalidation_listener.h

@@ -5,16 +5,14 @@
 #ifndef COMPONENTS_INVALIDATION_IMPL_FCM_INVALIDATION_LISTENER_H_
 #define COMPONENTS_INVALIDATION_IMPL_FCM_INVALIDATION_LISTENER_H_
 
+#include <map>
 #include <memory>
 
-#include "base/functional/callback_forward.h"
 #include "base/memory/raw_ptr.h"
 #include "base/memory/weak_ptr.h"
-#include "base/values.h"
 #include "components/invalidation/impl/channels_states.h"
 #include "components/invalidation/impl/fcm_sync_network_channel.h"
 #include "components/invalidation/impl/per_user_topic_subscription_manager.h"
-#include "components/invalidation/impl/unacked_invalidation_set.h"
 #include "components/invalidation/public/ack_handler.h"
 #include "components/invalidation/public/invalidation_util.h"
 #include "components/invalidation/public/invalidator_state.h"
@@ -102,26 +100,14 @@ class FCMInvalidationListener
 
   void EmitStateChange();
 
-  // Sends invalidations to their appropriate destination.
-  //
-  // If there are no observers registered for them, they will be saved for
-  // later.
-  //
-  // If there are observers registered, they will be saved (to make sure we
-  // don't drop them until they've been acted on) and emitted to the observers.
-  void DispatchInvalidations(const TopicInvalidationMap& invalidations);
-
-  // Saves invalidations.
-  //
-  // This call isn't synchronous so we can't guarantee these invalidations will
-  // be safely on disk by the end of the call, but it should ensure that the
-  // data makes it to disk eventually.
-  void SaveInvalidations(const TopicInvalidationMap& to_save);
+  // Cache `invalidation` and emit it to registered handlers (if any).
+  void DispatchInvalidation(const Invalidation& invalidation);
+
   // Emits previously saved invalidations to their registered observers.
   void EmitSavedInvalidations(const TopicInvalidationMap& to_emit);
 
   std::unique_ptr<FCMSyncNetworkChannel> network_channel_;
-  UnackedInvalidationsMap unacked_invalidations_map_;
+  std::map<Topic, Invalidation> unacked_invalidations_map_;
   raw_ptr<Delegate> delegate_ = nullptr;
 
   // The set of topics for which we want to receive invalidations. We'll pass

components/invalidation/impl/fcm_invalidation_listener_unittest.cc

@@ -278,28 +278,34 @@ TEST_F(FCMInvalidationListenerTest, InvalidateBeforeRegistration_Simple) {
   EXPECT_EQ(kPayload1, GetPayload(topic));
 }
 
-// Fire ten invalidations before an topics registers.  Some invalidations will
-// be dropped an replaced with an unknown version invalidation.
+// Fire a couple of invalidations before any topic registers. For each topic,
+// all but the invalidation with the highest version number will be dropped.
 TEST_F(FCMInvalidationListenerTest, InvalidateBeforeRegistration_Drop) {
-  const int kRepeatCount =
-      UnackedInvalidationSet::kMaxBufferedInvalidations + 1;
-  const Topic kUnregisteredId("unregistered");
-  const Topic& topic = kUnregisteredId;
+  const int kRepeatCount = 10;
+  const Topic kTopicA = "unregistered topic a";
+  const Topic kTopicB = "unregistered topic b";
   Topics topics;
-  topics.emplace(topic, TopicMetadata{false});
+  topics.emplace(kTopicA, TopicMetadata{false});
+  topics.emplace(kTopicB, TopicMetadata{false});
 
-  EXPECT_EQ(0U, GetInvalidationCount(topic));
+  EXPECT_EQ(0U, GetInvalidationCount(kTopicA));
+  EXPECT_EQ(0U, GetInvalidationCount(kTopicB));
 
-  int64_t initial_version = kVersion1;
-  for (int64_t i = initial_version; i < initial_version + kRepeatCount; ++i) {
-    FireInvalidate(topic, i, kPayload1);
+  const int64_t initial_version = kVersion1;
+  const int64_t max_version = initial_version + kRepeatCount;
+  for (int64_t i = initial_version; i <= initial_version + kRepeatCount; ++i) {
+    FireInvalidate(kTopicA, i, kPayload1);
+    FireInvalidate(kTopicB, i, kPayload1);
   }
 
   EnableNotifications();
   listener_.UpdateInterestedTopics(topics);
 
-  ASSERT_EQ(UnackedInvalidationSet::kMaxBufferedInvalidations,
-            GetInvalidationCount(topic));
+  EXPECT_EQ(1U, GetInvalidationCount(kTopicA));
+  EXPECT_EQ(max_version, GetVersion(kTopicA));
+
+  EXPECT_EQ(1U, GetInvalidationCount(kTopicB));
+  EXPECT_EQ(max_version, GetVersion(kTopicB));
 }
 
 // Fire an invalidation, then fire another one with a lower version.  Both