[Sync] Resend previously persisted consents when sync is reenabled.

This is a followup to
https://chromium-review.googlesource.com/c/chromium/src/+/986514. After
that CL, UserEventSyncBridge does not delete user consent events when
sync is disabled, instead they are persisted in the store. In this CL,
these user consent event are resubmitted to sync when sync is reenabled
(based on OnSyncStarting call). This is done to avoid losing consents,
especially in DICE where sync may be disabled frequently. On the other
hand, it is allowed to record the same consent multiple times and this
is heavily utilized in this CL.

One could potentially come up with something simpler, however, later we
will have to avoid syncing events emitted from a different account.
Due to this requirement, the approach in this CL could not be simplified
further.

Bug: 781765
Change-Id: Ia1278d716b0ace05084f1de56f0647e149fd0e5e
Reviewed-on: https://chromium-review.googlesource.com/1004896
Commit-Queue: vitaliii <vitaliii@chromium.org>
Reviewed-by: Mikel Astiz <mastiz@chromium.org>
Cr-Commit-Position: refs/heads/master@{#549888}

chrome/browser/sync/test/integration/single_client_user_events_sync_test.cc

@@ -339,4 +339,20 @@ IN_PROC_BROWSER_TEST_F(SingleClientUserEventsSyncTest, FieldTrial) {
       .Wait();
 }
 
+IN_PROC_BROWSER_TEST_F(SingleClientUserEventsSyncTest,
+                       PreserveConsentsOnDisableSync) {
+  const UserEventSpecifics testEvent1 = CreateTestEvent(1);
+  const UserEventSpecifics consent1 = CreateUserConsent(2);
+
+  ASSERT_TRUE(SetupSync());
+  syncer::UserEventService* event_service =
+      browser_sync::UserEventServiceFactory::GetForProfile(GetProfile(0));
+  event_service->RecordUserEvent(testEvent1);
+  event_service->RecordUserEvent(consent1);
+
+  ASSERT_TRUE(GetClient(0)->RestartSyncService());
+
+  EXPECT_TRUE(ExpectUserEvents({consent1}));
+}
+
 }  // namespace

components/sync/model/model_type_sync_bridge.h

@@ -127,8 +127,10 @@ class ModelTypeSyncBridge : public base::SupportsWeakPtr<ModelTypeSyncBridge> {
   // Called by the DataTypeController to gather additional information needed
   // before the processor can be connected to a sync worker. Once the
   // metadata has been loaded, the info is collected and given to |callback|.
-  void OnSyncStarting(const ModelErrorHandler& error_handler,
-                      const ModelTypeChangeProcessor::StartCallback& callback);
+  // When overriding, the bridge must notify the processor.
+  virtual void OnSyncStarting(
+      const ModelErrorHandler& error_handler,
+      const ModelTypeChangeProcessor::StartCallback& callback);
 
   // Indicates that we no longer want to do any sync-related things for this
   // data type. Severs all ties to the sync thread, deletes all local sync

components/sync/user_events/user_event_sync_bridge.cc

@@ -72,7 +72,8 @@ UserEventSyncBridge::UserEventSyncBridge(
     std::unique_ptr<ModelTypeChangeProcessor> change_processor,
     GlobalIdMapper* global_id_mapper)
     : ModelTypeSyncBridge(std::move(change_processor)),
-      global_id_mapper_(global_id_mapper) {
+      global_id_mapper_(global_id_mapper),
+      is_sync_starting_or_started_(false) {
   DCHECK(global_id_mapper_);
   std::move(store_factory)
       .Run(USER_EVENTS, base::BindOnce(&UserEventSyncBridge::OnStoreCreated,
@@ -148,10 +149,26 @@ std::string UserEventSyncBridge::GetStorageKey(const EntityData& entity_data) {
   return GetStorageKeyFromSpecifics(entity_data.specifics.user_event());
 }
 
+void UserEventSyncBridge::OnSyncStarting(
+    const ModelErrorHandler& error_handler,
+    const ModelTypeChangeProcessor::StartCallback& start_callback) {
+  change_processor()->OnSyncStarting(std::move(error_handler), start_callback);
+
+  bool was_sync_starting_or_started = is_sync_starting_or_started_;
+  is_sync_starting_or_started_ = true;
+  if (store_ && change_processor()->IsTrackingMetadata() &&
+      !was_sync_starting_or_started) {
+    ReadAllDataAndResubmit();
+  }
+}
+
 void UserEventSyncBridge::ApplyDisableSyncChanges(
     std::unique_ptr<MetadataChangeList> delete_metadata_change_list) {
   // Sync can only be disabled after initialization.
   DCHECK(deferred_user_events_while_initializing_.empty());
+
+  is_sync_starting_or_started_ = false;
+
   // Delete everything except user consents. With DICE the signout may happen
   // frequently. It is important to report all user consents, thus, they are
   // persisted for some time even after signout.
@@ -189,6 +206,38 @@ void UserEventSyncBridge::OnReadAllDataToDelete(
       base::BindOnce(&UserEventSyncBridge::OnCommit, base::AsWeakPtr(this)));
 }
 
+void UserEventSyncBridge::ReadAllDataAndResubmit() {
+  DCHECK(is_sync_starting_or_started_);
+  DCHECK(change_processor()->IsTrackingMetadata());
+  DCHECK(store_);
+  store_->ReadAllData(base::BindOnce(
+      &UserEventSyncBridge::OnReadAllDataToResubmit, base::AsWeakPtr(this)));
+}
+
+void UserEventSyncBridge::OnReadAllDataToResubmit(
+    const base::Optional<ModelError>& error,
+    std::unique_ptr<RecordList> data_records) {
+  if (!is_sync_starting_or_started_) {
+    // Meanwhile the sync has been disabled. We will try next time.
+    return;
+  }
+  DCHECK(change_processor()->IsTrackingMetadata());
+
+  if (error) {
+    change_processor()->ReportError(*error);
+    return;
+  }
+
+  for (const Record& r : *data_records) {
+    auto specifics = std::make_unique<UserEventSpecifics>();
+    if (specifics->ParseFromString(r.value) &&
+        specifics->event_case() ==
+            UserEventSpecifics::EventCase::kUserConsent) {
+      RecordUserEventImpl(std::move(specifics));
+    }
+  }
+}
+
 void UserEventSyncBridge::RecordUserEvent(
     std::unique_ptr<UserEventSpecifics> specifics) {
   if (change_processor()->IsTrackingMetadata()) {
@@ -249,8 +298,6 @@ void UserEventSyncBridge::OnStoreCreated(
     return;
   }
 
-  // TODO(vitaliii): Attempt to resubmit persistently stored user consents
-  // (probably not immediately on startup).
   // TODO(vitaliii): Garbage collect old user consents if sync is disabled.
 
   store_ = std::move(store);
@@ -266,6 +313,9 @@ void UserEventSyncBridge::OnReadAllMetadata(
   } else {
     change_processor()->ModelReadyToSync(this, std::move(metadata_batch));
     DCHECK(change_processor()->IsTrackingMetadata());
+    if (is_sync_starting_or_started_) {
+      ReadAllDataAndResubmit();
+    }
     ProcessQueuedEvents();
   }
 }

components/sync/user_events/user_event_sync_bridge.h

@@ -41,6 +41,9 @@ class UserEventSyncBridge : public ModelTypeSyncBridge {
   void GetAllData(DataCallback callback) override;
   std::string GetClientTag(const EntityData& entity_data) override;
   std::string GetStorageKey(const EntityData& entity_data) override;
+  void OnSyncStarting(
+      const ModelErrorHandler& error_handler,
+      const ModelTypeChangeProcessor::StartCallback& callback) override;
   void ApplyDisableSyncChanges(
       std::unique_ptr<MetadataChangeList> delete_metadata_change_list) override;
 
@@ -49,6 +52,7 @@ class UserEventSyncBridge : public ModelTypeSyncBridge {
  private:
   void RecordUserEventImpl(
       std::unique_ptr<sync_pb::UserEventSpecifics> specifics);
+  // Record events in the deferred queue and clear the queue.
   void ProcessQueuedEvents();
 
   void OnStoreCreated(const base::Optional<ModelError>& error,
@@ -68,6 +72,14 @@ class UserEventSyncBridge : public ModelTypeSyncBridge {
       const base::Optional<ModelError>& error,
       std::unique_ptr<ModelTypeStore::RecordList> data_records);
 
+  // Resubmit all the events persisted in the store to sync events, which were
+  // preserved when sync was disabled. This may resubmit entities that the
+  // processor already knows about (i.e. with metadata), but it is allowed.
+  void ReadAllDataAndResubmit();
+  void OnReadAllDataToResubmit(
+      const base::Optional<ModelError>& error,
+      std::unique_ptr<ModelTypeStore::RecordList> data_records);
+
   void HandleGlobalIdChange(int64_t old_global_id, int64_t new_global_id);
 
   // Persistent storage for in flight events. Should remain quite small, as we
@@ -85,6 +97,8 @@ class UserEventSyncBridge : public ModelTypeSyncBridge {
 
   GlobalIdMapper* global_id_mapper_;
 
+  bool is_sync_starting_or_started_;
+
   DISALLOW_COPY_AND_ASSIGN(UserEventSyncBridge);
 };
 

components/sync/user_events/user_event_sync_bridge_unittest.cc

@@ -9,6 +9,7 @@
 #include <utility>
 
 #include "base/bind.h"
+#include "base/bind_helpers.h"
 #include "base/message_loop/message_loop.h"
 #include "base/run_loop.h"
 #include "base/test/bind_test_util.h"
@@ -378,9 +379,12 @@ TEST_F(UserEventSyncBridgeTest, RecordWithLateInitializedStore) {
   EXPECT_CALL(*processor(), DoModelReadyToSync(&late_init_bridge, NotNull()));
   ON_CALL(*processor(), IsTrackingMetadata()).WillByDefault(Return(true));
   std::move(store_init_callback)
-      .Run(base::nullopt,
+      .Run(/*error=*/base::nullopt,
            ModelTypeStoreTestUtil::CreateInMemoryStoreForTest(store_init_type));
 
+  late_init_bridge.OnSyncStarting(/*error_handler=*/base::DoNothing(),
+                                  /*start_callback=*/base::DoNothing());
+
   // Record events after metadata is ready.
   late_init_bridge.RecordUserEvent(
       std::make_unique<UserEventSpecifics>(consent2));
@@ -396,6 +400,139 @@ TEST_F(UserEventSyncBridgeTest, RecordWithLateInitializedStore) {
           Pair(GetStorageKey(specifics2), MatchesUserEvent(specifics2))));
 }
 
+TEST_F(UserEventSyncBridgeTest,
+       ShouldReportPreviouslyPersistedConsentsWhenSyncIsReenabled) {
+  UserEventSpecifics consent = CreateSpecifics(1u, 1u, 1u);
+  consent.mutable_user_consent();
+
+  bridge()->RecordUserEvent(std::make_unique<UserEventSpecifics>(consent));
+
+  EXPECT_CALL(*processor(), DisableSync());
+  bridge()->DisableSync();
+  // The bridge may asynchronously query the store to choose what to delete.
+  base::RunLoop().RunUntilIdle();
+
+  ASSERT_THAT(GetAllData(), SizeIs(1));
+
+  // Reenable sync.
+  ON_CALL(*processor(), IsTrackingMetadata()).WillByDefault(Return(true));
+  std::string storage_key;
+  EXPECT_CALL(*processor(), DoPut(_, _, _)).WillOnce(SaveArg<0>(&storage_key));
+  bridge()->OnSyncStarting(/*error_handler=*/base::DoNothing(),
+                           /*start_callback=*/base::DoNothing());
+
+  // The bridge may asynchronously query the store to choose what to resubmit.
+  base::RunLoop().RunUntilIdle();
+
+  EXPECT_THAT(storage_key, GetStorageKey(consent));
+}
+
+TEST_F(UserEventSyncBridgeTest,
+       ShouldReportPersistedConsentsOnStartupEvenWithLateStoreInitialization) {
+  // Wait until bridge() is ready to avoid interference with processor() mock.
+  base::RunLoop().RunUntilIdle();
+
+  UserEventSpecifics consent = CreateSpecifics(1u, 1u, 1u);
+  consent.mutable_user_consent();
+
+  ON_CALL(*processor(), IsTrackingMetadata()).WillByDefault(Return(false));
+  ModelType store_init_type;
+  ModelTypeStore::InitCallback store_init_callback;
+  UserEventSyncBridge late_init_bridge(
+      base::BindLambdaForTesting(
+          [&](ModelType type, ModelTypeStore::InitCallback callback) {
+            store_init_type = type;
+            store_init_callback = std::move(callback);
+          }),
+      processor()->CreateForwardingProcessor(), mapper());
+
+  // Sync is active, but the store is not ready yet.
+  EXPECT_CALL(*processor(), DoModelReadyToSync(_, _)).Times(0);
+  late_init_bridge.OnSyncStarting(/*error_handler=*/base::DoNothing(),
+                                  /*start_callback=*/base::DoNothing());
+
+  // Initialize the store.
+  std::unique_ptr<ModelTypeStore> store =
+      ModelTypeStoreTestUtil::CreateInMemoryStoreForTest(store_init_type);
+
+  // TODO(vitaliii): Try to avoid putting the data directly into the store (e.g.
+  // by using a forwarding store), because this is an implementation detail.
+  // However, currently the bridge owns the store and there is no obvious way to
+  // preserve it.
+
+  // Put the consent manually to simulate a restart with disabled sync.
+  auto batch = store->CreateWriteBatch();
+  batch->WriteData(GetStorageKey(consent), consent.SerializeAsString());
+  store->CommitWriteBatch(std::move(batch), base::DoNothing());
+  base::RunLoop().RunUntilIdle();
+
+  EXPECT_CALL(*processor(), DoModelReadyToSync(&late_init_bridge, NotNull()));
+  ON_CALL(*processor(), IsTrackingMetadata()).WillByDefault(Return(true));
+  std::string storage_key;
+  EXPECT_CALL(*processor(), DoPut(_, _, _)).WillOnce(SaveArg<0>(&storage_key));
+  std::move(store_init_callback)
+      .Run(/*error=*/base::nullopt,
+           ModelTypeStoreTestUtil::CreateInMemoryStoreForTest(store_init_type));
+
+  // The bridge may asynchronously query the store to choose what to resubmit.
+  base::RunLoop().RunUntilIdle();
+
+  EXPECT_THAT(storage_key, GetStorageKey(consent));
+}
+
+TEST_F(UserEventSyncBridgeTest,
+       ShouldReportPersistedConsentsOnStartupEvenWithLateSyncInitialization) {
+  // Wait until bridge() is ready to avoid interference with processor() mock.
+  base::RunLoop().RunUntilIdle();
+
+  UserEventSpecifics consent = CreateSpecifics(1u, 1u, 1u);
+  consent.mutable_user_consent();
+
+  ON_CALL(*processor(), IsTrackingMetadata()).WillByDefault(Return(false));
+  ModelType store_init_type;
+  ModelTypeStore::InitCallback store_init_callback;
+  UserEventSyncBridge late_init_bridge(
+      base::BindLambdaForTesting(
+          [&](ModelType type, ModelTypeStore::InitCallback callback) {
+            store_init_type = type;
+            store_init_callback = std::move(callback);
+          }),
+      processor()->CreateForwardingProcessor(), mapper());
+
+  // Initialize the store.
+  std::unique_ptr<ModelTypeStore> store =
+      ModelTypeStoreTestUtil::CreateInMemoryStoreForTest(store_init_type);
+
+  // TODO(vitaliii): Try to avoid putting the data directly into the store (e.g.
+  // by using a forwarding store), because this is an implementation detail.
+  // However, currently the bridge owns the store and there is no obvious way to
+  // preserve it.
+
+  // Put the consent manually to simulate a restart with disabled sync.
+  auto batch = store->CreateWriteBatch();
+  batch->WriteData(GetStorageKey(consent), consent.SerializeAsString());
+  store->CommitWriteBatch(std::move(batch), base::DoNothing());
+  base::RunLoop().RunUntilIdle();
+
+  // The store has been initialized, but the sync is not active yet.
+  ON_CALL(*processor(), IsTrackingMetadata()).WillByDefault(Return(true));
+  EXPECT_CALL(*processor(), DoModelReadyToSync(&late_init_bridge, NotNull()));
+  std::move(store_init_callback)
+      .Run(/*error=*/base::nullopt,
+           ModelTypeStoreTestUtil::CreateInMemoryStoreForTest(store_init_type));
+  base::RunLoop().RunUntilIdle();
+
+  late_init_bridge.OnSyncStarting(/*error_handler=*/base::DoNothing(),
+                                  /*start_callback=*/base::DoNothing());
+
+  std::string storage_key;
+  EXPECT_CALL(*processor(), DoPut(_, _, _)).WillOnce(SaveArg<0>(&storage_key));
+  // The bridge may asynchronously query the store to choose what to resubmit.
+  base::RunLoop().RunUntilIdle();
+
+  EXPECT_THAT(storage_key, GetStorageKey(consent));
+}
+
 }  // namespace
 
 }  // namespace syncer