Introduce infrastructure for deferred local Nigori commits

Nigori is a special datatype because failing to commit local changes
(e.g. custom passphrase set) can lead to problematic scenarios (e.g.
user data exposed unencrypted) or in rare cases complicated
conflict-resolution scenarios (e.g. keystore rotation took place before
committing the custom passphrase) that worst-case cannot be resolved
synchronously (e.g. keystore keys are missing and hence keys are
pending for some time, until keystore keys are received).

It's very hard to universally reason about all side effects of local
commits and conflict resolution cases. Instead, this patch introduces
infrastructure that adopts the command design pattern to concisely
represent what the intended local change is.

Local changes may now fail to apply, in particular if there is a
conflict. But in such case the infrastructure allows dealing with the
failure, with custom logic that depends on the precise command type.

Similarly, success is only reported once the commit is acked by the
sync server, which is a desirable property and useful foundation for
future work. This may introduce additional latency in the existing UX,
e.g. while setting up a custom passphrase, but manual tests indicate
that it's not noticeable.

In this patch, only the infrastructure is introduced without actually
adopting it for the various local changes. Future patches will take
care of such adoption.

Bug: 922900
Change-Id: Ib55d3a401f1f68a1a910ee1e2b9119889d4f73b6
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1864770
Commit-Queue: Mikel Astiz <mastiz@chromium.org>
Reviewed-by: Maksim Moskvitin <mmoskvitin@google.com>
Cr-Commit-Position: refs/heads/master@{#706401}

components/sync/BUILD.gn

@@ -331,6 +331,7 @@ jumbo_static_library("rest_of_sync") {
     "nigori/nigori_sync_bridge.h",
     "nigori/nigori_sync_bridge_impl.cc",
     "nigori/nigori_sync_bridge_impl.h",
+    "nigori/pending_local_nigori_commit.h",
     "syncable/base_node.cc",
     "syncable/base_node.h",
     "syncable/base_transaction.cc",

components/sync/nigori/cryptographer_impl.cc

@@ -88,11 +88,15 @@ sync_pb::NigoriKey CryptographerImpl::ExportDefaultKey() const {
   return key_bag_.ExportKey(default_encryption_key_name_);
 }
 
-std::unique_ptr<Cryptographer> CryptographerImpl::Clone() const {
+std::unique_ptr<CryptographerImpl> CryptographerImpl::CloneImpl() const {
   return base::WrapUnique(
       new CryptographerImpl(key_bag_.Clone(), default_encryption_key_name_));
 }
 
+std::unique_ptr<Cryptographer> CryptographerImpl::Clone() const {
+  return CloneImpl();
+}
+
 bool CryptographerImpl::CanEncrypt() const {
   return !default_encryption_key_name_.empty();
 }

components/sync/nigori/cryptographer_impl.h

@@ -69,6 +69,9 @@ class CryptographerImpl : public Cryptographer {
   // have a default encryption key set, as reflected by CanEncrypt().
   sync_pb::NigoriKey ExportDefaultKey() const;
 
+  // Similar to Clone() but returns CryptographerImpl.
+  std::unique_ptr<CryptographerImpl> CloneImpl() const;
+
   // Cryptographer overrides.
   std::unique_ptr<Cryptographer> Clone() const override;
   bool CanEncrypt() const override;

components/sync/nigori/nigori_local_change_processor.h

@@ -50,6 +50,10 @@ class NigoriLocalChangeProcessor {
   // Informs the Nigori processor of a new or updated Nigori entity.
   virtual void Put(std::unique_ptr<EntityData> entity_data) = 0;
 
+  // Returns true the Nigori entity as tracked by the processor has local
+  // changes. A commit may or may not be in progress at this time.
+  virtual bool IsEntityUnsynced() = 0;
+
   // Returns both the entity metadata and model type state such that the Nigori
   // model takes care of persisting them.
   virtual NigoriMetadataBatch GetMetadata() = 0;

components/sync/nigori/nigori_model_type_processor.cc

@@ -335,6 +335,15 @@ void NigoriModelTypeProcessor::Put(std::unique_ptr<EntityData> entity_data) {
   NudgeForCommitIfNeeded();
 }
 
+bool NigoriModelTypeProcessor::IsEntityUnsynced() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  if (entity_ == nullptr) {
+    return false;
+  }
+
+  return entity_->IsUnsynced();
+}
+
 NigoriMetadataBatch NigoriModelTypeProcessor::GetMetadata() {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
   DCHECK(IsTrackingMetadata());

components/sync/nigori/nigori_model_type_processor.h

@@ -50,6 +50,7 @@ class NigoriModelTypeProcessor : public ModelTypeProcessor,
   void ModelReadyToSync(NigoriSyncBridge* bridge,
                         NigoriMetadataBatch nigori_metadata) override;
   void Put(std::unique_ptr<EntityData> entity_data) override;
+  bool IsEntityUnsynced() override;
   NigoriMetadataBatch GetMetadata() override;
   void ReportError(const ModelError& error) override;
   base::WeakPtr<ModelTypeControllerDelegate> GetControllerDelegate() override;

components/sync/nigori/nigori_state.cc

@@ -303,4 +303,19 @@ sync_pb::NigoriSpecifics NigoriState::ToSpecificsProto() const {
   return specifics;
 }
 
+NigoriState NigoriState::Clone() const {
+  NigoriState result;
+  result.cryptographer = cryptographer->CloneImpl();
+  result.pending_keys = pending_keys;
+  result.passphrase_type = passphrase_type;
+  result.keystore_migration_time = keystore_migration_time;
+  result.custom_passphrase_time = custom_passphrase_time;
+  result.custom_passphrase_key_derivation_params =
+      custom_passphrase_key_derivation_params;
+  result.encrypt_everything = encrypt_everything;
+  result.keystore_keys = keystore_keys;
+  result.pending_keystore_decryptor_token = pending_keystore_decryptor_token;
+  return result;
+}
+
 }  // namespace syncer

components/sync/nigori/nigori_state.h

@@ -52,6 +52,9 @@ struct NigoriState {
   // Serialization to proto as sent to the sync server.
   sync_pb::NigoriSpecifics ToSpecificsProto() const;
 
+  // Makes a deep copy of |this|.
+  NigoriState Clone() const;
+
   std::unique_ptr<CryptographerImpl> cryptographer;
 
   // Pending keys represent a remote update that contained a keybag that cannot

components/sync/nigori/nigori_sync_bridge_impl.cc

@@ -7,6 +7,7 @@
 #include <utility>
 
 #include "base/base64.h"
+#include "base/bind.h"
 #include "base/location.h"
 #include "base/metrics/histogram_macros.h"
 #include "components/sync/base/encryptor.h"
@@ -17,6 +18,7 @@
 #include "components/sync/model/entity_data.h"
 #include "components/sync/nigori/nigori.h"
 #include "components/sync/nigori/nigori_storage.h"
+#include "components/sync/nigori/pending_local_nigori_commit.h"
 #include "components/sync/protocol/encryption.pb.h"
 #include "components/sync/protocol/nigori_local_data.pb.h"
 #include "components/sync/protocol/nigori_specifics.pb.h"
@@ -330,7 +332,10 @@ ModelTypeSet GetEncryptedTypes(bool encrypt_everything) {
 class NigoriSyncBridgeImpl::BroadcastingObserver
     : public SyncEncryptionHandler::Observer {
  public:
-  BroadcastingObserver() = default;
+  explicit BroadcastingObserver(
+      const base::RepeatingClosure& post_passphrase_accepted_cb)
+      : post_passphrase_accepted_cb_(post_passphrase_accepted_cb) {}
+
   ~BroadcastingObserver() override = default;
 
   void AddObserver(SyncEncryptionHandler::Observer* observer) {
@@ -356,6 +361,7 @@ class NigoriSyncBridgeImpl::BroadcastingObserver
     for (auto& observer : observers_) {
       observer.OnPassphraseAccepted();
     }
+    post_passphrase_accepted_cb_.Run();
   }
 
   void OnTrustedVaultKeyRequired() override {
@@ -410,6 +416,8 @@ class NigoriSyncBridgeImpl::BroadcastingObserver
   // implementation to use checked ObserverList as well.
   base::ObserverList<SyncEncryptionHandler::Observer>::Unchecked observers_;
 
+  const base::RepeatingClosure post_passphrase_accepted_cb_;
+
   DISALLOW_COPY_AND_ASSIGN(BroadcastingObserver);
 };
 
@@ -426,7 +434,12 @@ NigoriSyncBridgeImpl::NigoriSyncBridgeImpl(
       explicit_passphrase_key_(
           UnpackExplicitPassphraseKey(*encryptor,
                                       packed_explicit_passphrase_key)),
-      broadcasting_observer_(std::make_unique<BroadcastingObserver>()) {
+      broadcasting_observer_(std::make_unique<BroadcastingObserver>(
+          // base::Unretained() legit because the observer gets destroyed
+          // together with |this|.
+          base::BindRepeating(
+              &NigoriSyncBridgeImpl::MaybeNotifyBootstrapTokenUpdated,
+              base::Unretained(this)))) {
   DCHECK(encryptor);
 
   // TODO(crbug.com/922900): we currently don't verify |deserialized_data|.
@@ -512,6 +525,7 @@ bool NigoriSyncBridgeImpl::Init() {
 void NigoriSyncBridgeImpl::SetEncryptionPassphrase(
     const std::string& passphrase) {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  // TODO(crbug.com/922900): Adopt QueuePendingLocalCommit().
   switch (state_.passphrase_type) {
     case NigoriSpecifics::UNKNOWN:
       NOTREACHED();
@@ -570,7 +584,6 @@ void NigoriSyncBridgeImpl::SetEncryptionPassphrase(
   broadcasting_observer_->OnEncryptedTypesChanged(EncryptableUserTypes(),
                                                   state_.encrypt_everything);
 
-  MaybeNotifyBootstrapTokenUpdated();
   UMA_HISTOGRAM_BOOLEAN("Sync.CustomEncryption", true);
   // OnLocalSetPassphraseEncryption() is intentionally not called here, because
   // it's needed only for the Directory implementation unit tests.
@@ -611,7 +624,6 @@ void NigoriSyncBridgeImpl::SetDecryptionPassphrase(
       state_.cryptographer.get(), state_.pending_keys.has_value());
   broadcasting_observer_->OnPassphraseAccepted();
 
-  MaybeNotifyBootstrapTokenUpdated();
   // TODO(crbug.com/922900): we may need to rewrite encryption_keybag in Nigori
   // node in case we have some keys in |cryptographer_| which is not stored in
   // encryption_keybag yet.
@@ -764,6 +776,7 @@ base::Optional<ModelError> NigoriSyncBridgeImpl::MergeSyncData(
   }
   // We received uninitialized Nigori and need to initialize it as default
   // keystore Nigori.
+  // TODO(crbug.com/922900): Adopt QueuePendingLocalCommit().
   NigoriSpecifics initialized_specifics =
       MakeDefaultKeystoreNigori(state_.keystore_keys);
   // In rare cases the crypto operations may fail.
@@ -779,14 +792,28 @@ base::Optional<ModelError> NigoriSyncBridgeImpl::ApplySyncChanges(
     base::Optional<EntityData> data) {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
   DCHECK_NE(state_.passphrase_type, NigoriSpecifics::UNKNOWN);
-  if (!data) {
-    // Receiving empty |data| means metadata-only change, we need to persist
-    // its state.
-    storage_->StoreData(SerializeAsNigoriLocalData());
-    return base::nullopt;
-  }
+
+  if (data) {
     DCHECK(data->specifics.has_nigori());
     return UpdateLocalState(data->specifics.nigori());
+  }
+
+  if (!pending_local_commit_queue_.empty() && !processor_->IsEntityUnsynced()) {
+    // Successfully committed first element in queue.
+    bool success = pending_local_commit_queue_.front()->TryApply(&state_);
+    DCHECK(success);
+    pending_local_commit_queue_.front()->OnSuccess(
+        state_, broadcasting_observer_.get());
+    pending_local_commit_queue_.pop_front();
+
+    // Advance until the next applicable local change if any and call Put().
+    PutNextApplicablePendingLocalCommit();
+  }
+
+  // Receiving empty |data| means metadata-only change (e.g. no remote updates,
+  // or local commit completion), so we need to persist its state.
+  storage_->StoreData(SerializeAsNigoriLocalData());
+  return base::nullopt;
 }
 
 base::Optional<ModelError> NigoriSyncBridgeImpl::UpdateLocalState(
@@ -856,7 +883,6 @@ base::Optional<ModelError> NigoriSyncBridgeImpl::UpdateLocalState(
       UpdateCryptographerFromNonKeystoreNigori(encryption_keybag);
   }
 
-  storage_->StoreData(SerializeAsNigoriLocalData());
   if (passphrase_type_changed) {
     broadcasting_observer_->OnPassphraseTypeChanged(
         *ProtoPassphraseInt32ToEnum(state_.passphrase_type),
@@ -873,6 +899,18 @@ base::Optional<ModelError> NigoriSyncBridgeImpl::UpdateLocalState(
       state_.cryptographer.get(), state_.pending_keys.has_value());
 
   MaybeNotifyOfPendingKeys();
+
+  // TODO(crbug.com/): Instead of issuing errors for local commits, call
+  // PutNextApplicablePendingLocalCommit() to verify pending local changes
+  // (relevant for the conflict case). Issue failures if they are no longer
+  // applicable, or call Put() otherwise.
+  for (const auto& pending_local_commit : pending_local_commit_queue_) {
+    pending_local_commit->OnFailure(broadcasting_observer_.get());
+  }
+  pending_local_commit_queue_.clear();
+
+  storage_->StoreData(SerializeAsNigoriLocalData());
+
   return base::nullopt;
 }
 
@@ -983,7 +1021,19 @@ bool NigoriSyncBridgeImpl::TryDecryptPendingKeys() {
 
 std::unique_ptr<EntityData> NigoriSyncBridgeImpl::GetData() {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
-  if (state_.passphrase_type == NigoriSpecifics::UNKNOWN) {
+
+  NigoriSpecifics specifics;
+  if (!pending_local_commit_queue_.empty()) {
+    NigoriState changed_state = state_.Clone();
+    bool success =
+        pending_local_commit_queue_.front()->TryApply(&changed_state);
+    DCHECK(success);
+    specifics = changed_state.ToSpecificsProto();
+  } else {
+    specifics = state_.ToSpecificsProto();
+  }
+
+  if (specifics.passphrase_type() == NigoriSpecifics::UNKNOWN) {
     // Bridge never received NigoriSpecifics from the server. This line should
     // be reachable only from processor's GetAllNodesForDebugging().
     DCHECK(!state_.cryptographer->CanEncrypt());
@@ -991,7 +1041,6 @@ std::unique_ptr<EntityData> NigoriSyncBridgeImpl::GetData() {
     return nullptr;
   }
 
-  const sync_pb::NigoriSpecifics specifics = state_.ToSpecificsProto();
   DCHECK(IsValidNigoriSpecifics(specifics));
 
   auto entity_data = std::make_unique<EntityData>();
@@ -1160,4 +1209,38 @@ sync_pb::NigoriLocalData NigoriSyncBridgeImpl::SerializeAsNigoriLocalData()
   return output;
 }
 
+void NigoriSyncBridgeImpl::QueuePendingLocalCommit(
+    std::unique_ptr<PendingLocalNigoriCommit> local_commit) {
+  NigoriState tmp_state = state_.Clone();
+  if (state_.passphrase_type == NigoriSpecifics::UNKNOWN) {
+    local_commit->OnFailure(broadcasting_observer_.get());
+    return;
+  }
+
+  pending_local_commit_queue_.push_back(std::move(local_commit));
+
+  if (pending_local_commit_queue_.size() == 1) {
+    // Verify that the newly-introduced commit (if first in the queue) applies
+    // and if so call Put(), or otherwise issue an immediate failure.
+    PutNextApplicablePendingLocalCommit();
+  }
+}
+
+void NigoriSyncBridgeImpl::PutNextApplicablePendingLocalCommit() {
+  while (!pending_local_commit_queue_.empty()) {
+    NigoriState tmp_state = state_.Clone();
+    bool success = pending_local_commit_queue_.front()->TryApply(&tmp_state);
+    if (success) {
+      // This particular commit applies cleanly.
+      processor_->Put(GetData());
+      break;
+    }
+
+    // The local change failed to apply.
+    pending_local_commit_queue_.front()->OnFailure(
+        broadcasting_observer_.get());
+    pending_local_commit_queue_.pop_front();
+  }
+}
+
 }  // namespace syncer

components/sync/nigori/nigori_sync_bridge_impl.h

@@ -5,6 +5,7 @@
 #ifndef COMPONENTS_SYNC_NIGORI_NIGORI_SYNC_BRIDGE_IMPL_H_
 #define COMPONENTS_SYNC_NIGORI_NIGORI_SYNC_BRIDGE_IMPL_H_
 
+#include <list>
 #include <memory>
 #include <string>
 #include <vector>
@@ -32,6 +33,7 @@ namespace syncer {
 
 class Encryptor;
 class NigoriStorage;
+class PendingLocalNigoriCommit;
 
 // USS implementation of SyncEncryptionHandler.
 // This class holds the current Nigori state and processes incoming changes and
@@ -130,6 +132,17 @@ class NigoriSyncBridgeImpl : public KeystoreKeysHandler,
   // Serializes state of the bridge and sync metadata into the proto.
   sync_pb::NigoriLocalData SerializeAsNigoriLocalData() const;
 
+  // Appends |local_commit| to |pending_local_commit_queue_| and if appropriate
+  // calls Put() to trigger the commit.
+  void QueuePendingLocalCommit(
+      std::unique_ptr<PendingLocalNigoriCommit> local_commit);
+
+  // Processes |pending_local_commit_queue_| FIFO such that all non-applicable
+  // pending commits issue a failure, until the first one that is applicable is
+  // found (if any). If such applicable commit is found, the corresponding Put()
+  // call is issued.
+  void PutNextApplicablePendingLocalCommit();
+
   const Encryptor* const encryptor_;
 
   const std::unique_ptr<NigoriLocalChangeProcessor> processor_;
@@ -146,6 +159,9 @@ class NigoriSyncBridgeImpl : public KeystoreKeysHandler,
 
   syncer::NigoriState state_;
 
+  std::list<std::unique_ptr<PendingLocalNigoriCommit>>
+      pending_local_commit_queue_;
+
   // Observer that owns the list of actual observers, and broadcasts
   // notifications to all observers in the list.
   class BroadcastingObserver;

components/sync/nigori/nigori_sync_bridge_impl_unittest.cc

@@ -292,6 +292,7 @@ class MockNigoriLocalChangeProcessor : public NigoriLocalChangeProcessor {
 
   MOCK_METHOD2(ModelReadyToSync, void(NigoriSyncBridge*, NigoriMetadataBatch));
   MOCK_METHOD1(Put, void(std::unique_ptr<EntityData>));
+  MOCK_METHOD0(IsEntityUnsynced, bool());
   MOCK_METHOD0(GetMetadata, NigoriMetadataBatch());
   MOCK_METHOD1(ReportError, void(const ModelError&));
   MOCK_METHOD0(GetControllerDelegate,

components/sync/nigori/pending_local_nigori_commit.h

+// Copyright 2019 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef COMPONENTS_SYNC_NIGORI_PENDING_LOCAL_NIGORI_COMMIT_H_
+#define COMPONENTS_SYNC_NIGORI_PENDING_LOCAL_NIGORI_COMMIT_H_
+
+#include <string>
+
+#include "base/macros.h"
+#include "components/sync/engine/sync_encryption_handler.h"
+
+namespace syncer {
+
+struct NigoriState;
+
+// Interface representing an intended local change to the Nigori state that
+// is pending a commit to the sync server.
+class PendingLocalNigoriCommit {
+ public:
+  PendingLocalNigoriCommit() = default;
+  virtual ~PendingLocalNigoriCommit() = default;
+
+  // Attempts to modify |*state| to reflect the intended commit. Returns true if
+  // the change was successfully applied (which may include the no-op case) or
+  // false if it no longer applies (leading to OnFailure()).
+  //
+  // |state| must not be null.
+  virtual bool TryApply(NigoriState* state) const = 0;
+
+  // Invoked when the commit has been successfully acked by the server.
+  // |observer| must not be null.
+  virtual void OnSuccess(const NigoriState& state,
+                         SyncEncryptionHandler::Observer* observer) = 0;
+
+  // Invoked when the change no longer applies or was aborted for a different
+  // reason (e.g. sync disabled). |observer| must not be null.
+  virtual void OnFailure(SyncEncryptionHandler::Observer* observer) = 0;
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(PendingLocalNigoriCommit);
+};
+
+}  // namespace syncer
+
+#endif  // COMPONENTS_SYNC_NIGORI_PENDING_LOCAL_NIGORI_COMMIT_H_