[Sync] Add plumbing of context from client to server

The datatype context is now written into the directory and plumbed up to the
server on every GetUpdates or Commit request. GetUpdatesResponses also now
overwrite the context if the version returned is the same or higher.

BUG=345420

Review URL: https://codereview.chromium.org/215973007

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@261873 0039d316-1c4b-4281-b951-d872f2087c98

14 files changed, 155 insertions, 37 deletions

diff --git a/sync/engine/directory_commit_contribution.cc b/sync/engine/directory_commit_contribution.cc
index 2fd7b30..a662c0e 100644
--- a/sync/engine/directory_commit_contribution.cc
+++ b/sync/engine/directory_commit_contribution.cc
@@ -41,8 +41,11 @@ scoped_ptr<DirectoryCommitContribution> DirectoryCommitContribution::Build(
     entry.PutSyncing(true);
   }
 
+  sync_pb::DataTypeContext context;
+  dir->GetDataTypeContext(&trans, type, &context);
+
   return scoped_ptr<DirectoryCommitContribution>(
-      new DirectoryCommitContribution(metahandles, entities, dir));
+      new DirectoryCommitContribution(metahandles, entities, context, dir));
 }
 
 void DirectoryCommitContribution::AddToCommitMessage(
@@ -53,6 +56,8 @@ void DirectoryCommitContribution::AddToCommitMessage(
   std::copy(entities_.begin(),
             entities_.end(),
             RepeatedPtrFieldBackInserter(commit_message->mutable_entries()));
+  if (!context_.context().empty())
+    commit_message->add_client_contexts()->Swap(&context_);
 }
 
 SyncerError DirectoryCommitContribution::ProcessCommitResponse(
@@ -144,13 +149,14 @@ size_t DirectoryCommitContribution::GetNumEntries() const {
 DirectoryCommitContribution::DirectoryCommitContribution(
     const std::vector<int64>& metahandles,
     const google::protobuf::RepeatedPtrField<sync_pb::SyncEntity>& entities,
+    const sync_pb::DataTypeContext& context,
     syncable::Directory* dir)
-  : dir_(dir),
-    metahandles_(metahandles),
-    entities_(entities),
-    entries_start_index_(0xDEADBEEF),
-    syncing_bits_set_(true) {
-}
+    : dir_(dir),
+      metahandles_(metahandles),
+      entities_(entities),
+      context_(context),
+      entries_start_index_(0xDEADBEEF),
+      syncing_bits_set_(true) {}
 
 void DirectoryCommitContribution::UnsetSyncingBits() {
   syncable::ModelNeutralWriteTransaction trans(FROM_HERE, SYNCER, dir_);
diff --git a/sync/engine/directory_commit_contribution.h b/sync/engine/directory_commit_contribution.h
index 508ac7c..83cefe9 100644
--- a/sync/engine/directory_commit_contribution.h
+++ b/sync/engine/directory_commit_contribution.h
@@ -66,7 +66,7 @@ class SYNC_EXPORT_PRIVATE DirectoryCommitContribution
       const sync_pb::ClientToServerResponse& response,
       sessions::StatusController* status) OVERRIDE;
 
-  // Cleans up any temproary state associated with the commit.  Must be called
+  // Cleans up any temporary state associated with the commit.  Must be called
   // before destruction.
   virtual void CleanUp() OVERRIDE;
 
@@ -82,6 +82,7 @@ class SYNC_EXPORT_PRIVATE DirectoryCommitContribution
   DirectoryCommitContribution(
       const std::vector<int64>& metahandles,
       const google::protobuf::RepeatedPtrField<sync_pb::SyncEntity>& entities,
+      const sync_pb::DataTypeContext& context,
       syncable::Directory* directory);
 
   void UnsetSyncingBits();
@@ -89,6 +90,7 @@ class SYNC_EXPORT_PRIVATE DirectoryCommitContribution
   syncable::Directory* dir_;
   const std::vector<int64> metahandles_;
   const google::protobuf::RepeatedPtrField<sync_pb::SyncEntity> entities_;
+  sync_pb::DataTypeContext context_;
   size_t entries_start_index_;
 
   // This flag is tracks whether or not the directory entries associated with
diff --git a/sync/engine/directory_update_handler.cc b/sync/engine/directory_update_handler.cc
index 9d97765..5509520 100644
--- a/sync/engine/directory_update_handler.cc
+++ b/sync/engine/directory_update_handler.cc
@@ -31,8 +31,15 @@ void DirectoryUpdateHandler::GetDownloadProgress(
   dir_->GetDownloadProgress(type_, progress_marker);
 }
 
+void DirectoryUpdateHandler::GetDataTypeContext(
+    sync_pb::DataTypeContext* context) const {
+  syncable::ModelNeutralWriteTransaction trans(FROM_HERE, SYNCER, dir_);
+  dir_->GetDataTypeContext(&trans, type_, context);
+}
+
 void DirectoryUpdateHandler::ProcessGetUpdatesResponse(
     const sync_pb::DataTypeProgressMarker& progress_marker,
+    const sync_pb::DataTypeContext& mutated_context,
     const SyncEntityList& applicable_updates,
     sessions::StatusController* status) {
   syncable::ModelNeutralWriteTransaction trans(FROM_HERE, SYNCER, dir_);
@@ -42,6 +49,20 @@ void DirectoryUpdateHandler::ProcessGetUpdatesResponse(
     ExpireEntriesIfNeeded(&trans, progress_marker);
     UpdateProgressMarker(progress_marker);
   }
+
+  if (mutated_context.has_context()) {
+    sync_pb::DataTypeContext local_context;
+    dir_->GetDataTypeContext(&trans, type_, &local_context);
+
+    // Only update the local context if it is still relevant. If the local
+    // version is higher, it means a local change happened while the mutation
+    // was in flight, and the local context takes priority.
+    if (mutated_context.version() >= local_context.version() &&
+        local_context.context() != mutated_context.context()) {
+      dir_->SetDataTypeContext(&trans, type_, mutated_context);
+      // TODO(zea): trigger the datatype's UpdateDataTypeContext method.
+    }
+  }
 }
 
 void DirectoryUpdateHandler::ApplyUpdates(sessions::StatusController* status) {
diff --git a/sync/engine/directory_update_handler.h b/sync/engine/directory_update_handler.h
index 5cc150d..74c7046 100644
--- a/sync/engine/directory_update_handler.h
+++ b/sync/engine/directory_update_handler.h
@@ -47,27 +47,17 @@ class SYNC_EXPORT_PRIVATE DirectoryUpdateHandler : public UpdateHandler {
                          scoped_refptr<ModelSafeWorker> worker);
   virtual ~DirectoryUpdateHandler();
 
-  // Fills the given parameter with the stored progress marker for this type.
+  // UpdateHandler implementation.
   virtual void GetDownloadProgress(
       sync_pb::DataTypeProgressMarker* progress_marker) const OVERRIDE;
-
-  // Processes the contents of a GetUpdates response message.
-  //
-  // Should be invoked with the progress marker and set of SyncEntities from a
-  // single GetUpdates response message.  The progress marker's type must match
-  // this update handler's type, and the set of SyncEntities must include all
-  // entities of this type found in the response message.
+  virtual void GetDataTypeContext(sync_pb::DataTypeContext* context) const
+      OVERRIDE;
   virtual void ProcessGetUpdatesResponse(
       const sync_pb::DataTypeProgressMarker& progress_marker,
+      const sync_pb::DataTypeContext& mutated_context,
       const SyncEntityList& applicable_updates,
       sessions::StatusController* status) OVERRIDE;
-
-  // If there are updates to apply, apply them on the proper thread.
-  // Delegates to ApplyUpdatesImpl().
   virtual void ApplyUpdates(sessions::StatusController* status) OVERRIDE;
-
-  // Apply updates on the sync thread.  This is for use during initial sync
-  // prior to model association.
   virtual void PassiveApplyUpdates(sessions::StatusController* status) OVERRIDE;
 
  private:
diff --git a/sync/engine/directory_update_handler_unittest.cc b/sync/engine/directory_update_handler_unittest.cc
index ada49e1..6d1b35a 100644
--- a/sync/engine/directory_update_handler_unittest.cc
+++ b/sync/engine/directory_update_handler_unittest.cc
@@ -251,6 +251,12 @@ TEST_F(DirectoryUpdateHandlerProcessUpdateTest, GarbageCollectionByVersion) {
   progress.set_token("token");
   progress.mutable_gc_directive()->set_version_watermark(kDefaultVersion + 10);
 
+  sync_pb::DataTypeContext context;
+  context.set_data_type_id(
+      GetSpecificsFieldNumberFromModelType(SYNCED_NOTIFICATIONS));
+  context.set_context("context");
+  context.set_version(1);
+
   scoped_ptr<sync_pb::SyncEntity> type_root =
       CreateUpdate(SyncableIdToProto(syncable::Id::CreateFromServerId("root")),
                    syncable::GetNullId().GetServerId(),
@@ -277,7 +283,7 @@ TEST_F(DirectoryUpdateHandlerProcessUpdateTest, GarbageCollectionByVersion) {
   updates.push_back(e2.get());
 
   // Process and apply updates.
-  handler.ProcessGetUpdatesResponse(progress, updates, &status);
+  handler.ProcessGetUpdatesResponse(progress, context, updates, &status);
   handler.ApplyUpdates(&status);
 
   // Verify none is deleted because they are unapplied during GC.
@@ -287,7 +293,8 @@ TEST_F(DirectoryUpdateHandlerProcessUpdateTest, GarbageCollectionByVersion) {
 
   // Process and apply again. Old entry is deleted but not root.
   progress.mutable_gc_directive()->set_version_watermark(kDefaultVersion + 20);
-  handler.ProcessGetUpdatesResponse(progress, SyncEntityList(), &status);
+  handler.ProcessGetUpdatesResponse(
+      progress, context, SyncEntityList(), &status);
   handler.ApplyUpdates(&status);
   EXPECT_TRUE(EntryExists(type_root->id_string()));
   EXPECT_FALSE(EntryExists(e1->id_string()));
diff --git a/sync/engine/get_updates_processor.cc b/sync/engine/get_updates_processor.cc
index eb1d7be..d0aa9b9 100644
--- a/sync/engine/get_updates_processor.cc
+++ b/sync/engine/get_updates_processor.cc
@@ -60,17 +60,16 @@ SyncerError HandleGetEncryptionKeyResponse(
 // divides them according to their type.  Outputs a map from model types to
 // received SyncEntities.  The output map will have entries (possibly empty)
 // for all types in |requested_types|.
-void PartitionUpdatesByType(
-    const sync_pb::GetUpdatesResponse& updates,
-    ModelTypeSet requested_types,
-    TypeSyncEntityMap* updates_by_type) {
-  int update_count = updates.entries().size();
+void PartitionUpdatesByType(const sync_pb::GetUpdatesResponse& gu_response,
+                            ModelTypeSet requested_types,
+                            TypeSyncEntityMap* updates_by_type) {
+  int update_count = gu_response.entries().size();
   for (ModelTypeSet::Iterator it = requested_types.First();
        it.Good(); it.Inc()) {
     updates_by_type->insert(std::make_pair(it.Get(), SyncEntityList()));
   }
   for (int i = 0; i < update_count; ++i) {
-    const sync_pb::SyncEntity& update = updates.entries(i);
+    const sync_pb::SyncEntity& update = gu_response.entries(i);
     ModelType type = GetModelType(update);
     if (!IsRealDataType(type)) {
       NOTREACHED() << "Received update with invalid type.";
@@ -111,6 +110,27 @@ void PartitionProgressMarkersByType(
   }
 }
 
+void PartitionContextMutationsByType(
+    const sync_pb::GetUpdatesResponse& gu_response,
+    ModelTypeSet request_types,
+    TypeToIndexMap* index_map) {
+  for (int i = 0; i < gu_response.context_mutations_size(); ++i) {
+    int field_number = gu_response.context_mutations(i).data_type_id();
+    ModelType model_type = GetModelTypeFromSpecificsFieldNumber(field_number);
+    if (!IsRealDataType(model_type)) {
+      DLOG(WARNING) << "Unknown field number " << field_number;
+      continue;
+    }
+    if (!request_types.Has(model_type)) {
+      DLOG(WARNING)
+          << "Skipping unexpected context mutation for non-enabled type "
+          << ModelTypeToString(model_type);
+      continue;
+    }
+    index_map->insert(std::make_pair(model_type, i));
+  }
+}
+
 // Initializes the parts of the GetUpdatesMessage that depend on shared state,
 // like the ShouldRequestEncryptionKey() status.  This is kept separate from the
 // other of the message-building functions to make the rest of the code easier
@@ -178,7 +198,13 @@ void GetUpdatesProcessor::PrepareGetUpdates(
         get_updates->add_from_progress_marker();
     handler_it->second->GetDownloadProgress(progress_marker);
     progress_marker->clear_gc_directive();
+
+    sync_pb::DataTypeContext context;
+    handler_it->second->GetDataTypeContext(&context);
+    if (!context.context().empty())
+      get_updates->add_client_contexts()->Swap(&context);
   }
+
   delegate_.HelpPopulateGuMessage(get_updates);
 }
 
@@ -286,6 +312,9 @@ bool GetUpdatesProcessor::ProcessGetUpdatesResponse(
     return false;
   }
 
+  TypeToIndexMap context_by_type;
+  PartitionContextMutationsByType(gu_response, gu_types, &context_by_type);
+
   // Iterate over these maps in parallel, processing updates for each type.
   TypeToIndexMap::iterator progress_marker_iter =
       progress_index_by_type.begin();
@@ -299,9 +328,15 @@ bool GetUpdatesProcessor::ProcessGetUpdatesResponse(
     UpdateHandlerMap::iterator update_handler_iter =
         update_handler_map_->find(type);
 
+    sync_pb::DataTypeContext context;
+    TypeToIndexMap::iterator context_iter = context_by_type.find(type);
+    if (context_iter != context_by_type.end())
+      context.CopyFrom(gu_response.context_mutations(context_iter->second));
+
     if (update_handler_iter != update_handler_map_->end()) {
       update_handler_iter->second->ProcessGetUpdatesResponse(
           gu_response.new_progress_marker(progress_marker_iter->second),
+          context,
           updates_iter->second,
           status_controller);
     } else {
diff --git a/sync/engine/get_updates_processor_unittest.cc b/sync/engine/get_updates_processor_unittest.cc
index b8ea889..7c87fed 100644
--- a/sync/engine/get_updates_processor_unittest.cc
+++ b/sync/engine/get_updates_processor_unittest.cc
@@ -53,6 +53,10 @@ class GetUpdatesProcessorTest : public ::testing::Test {
           response->add_new_progress_marker();
       marker->set_data_type_id(GetSpecificsFieldNumberFromModelType(it.Get()));
       marker->set_token("foobarbaz");
+      sync_pb::DataTypeContext* context = response->add_context_mutations();
+      context->set_data_type_id(GetSpecificsFieldNumberFromModelType(it.Get()));
+      context->set_version(1);
+      context->set_context("context");
     }
 
     response->set_changes_remaining(0);
diff --git a/sync/engine/non_blocking_type_processor_core.cc b/sync/engine/non_blocking_type_processor_core.cc
index fdb9e10..c1bc60a 100644
--- a/sync/engine/non_blocking_type_processor_core.cc
+++ b/sync/engine/non_blocking_type_processor_core.cc
@@ -33,17 +33,25 @@ void NonBlockingTypeProcessorCore::GetDownloadProgress(
     sync_pb::DataTypeProgressMarker* progress_marker) const {
   DCHECK(CalledOnValidThread());
   // TODO(rlarocque): Implement this properly.  crbug.com/351005.
-  VLOG(1) << "Getting progress for: " << ModelTypeToString(type_);
+  DVLOG(1) << "Getting progress for: " << ModelTypeToString(type_);
   *progress_marker = progress_marker_;
 }
 
+void NonBlockingTypeProcessorCore::GetDataTypeContext(
+    sync_pb::DataTypeContext* context) const {
+  // TODO(rlarocque): Implement this properly.  crbug.com/351005.
+  DVLOG(1) << "Getting context for: " << ModelTypeToString(type_);
+  context->Clear();
+}
+
 void NonBlockingTypeProcessorCore::ProcessGetUpdatesResponse(
     const sync_pb::DataTypeProgressMarker& progress_marker,
+    const sync_pb::DataTypeContext& mutated_context,
     const SyncEntityList& applicable_updates,
     sessions::StatusController* status) {
   DCHECK(CalledOnValidThread());
   // TODO(rlarocque): Implement this properly.  crbug.com/351005.
-  VLOG(1) << "Processing updates response for: " << ModelTypeToString(type_);
+  DVLOG(1) << "Processing updates response for: " << ModelTypeToString(type_);
   progress_marker_ = progress_marker;
 }
 
@@ -51,7 +59,7 @@ void NonBlockingTypeProcessorCore::ApplyUpdates(
     sessions::StatusController* status) {
   DCHECK(CalledOnValidThread());
   // TODO(rlarocque): Implement this properly.  crbug.com/351005.
-  VLOG(1) << "Applying updates for: " << ModelTypeToString(type_);
+  DVLOG(1) << "Applying updates for: " << ModelTypeToString(type_);
 }
 
 void NonBlockingTypeProcessorCore::PassiveApplyUpdates(
@@ -66,7 +74,7 @@ scoped_ptr<CommitContribution>
 NonBlockingTypeProcessorCore::GetContribution(size_t max_entries) {
   DCHECK(CalledOnValidThread());
   // TODO(rlarocque): Implement this properly.  crbug.com/351005.
-  VLOG(1) << "Getting commit contribution for: " << ModelTypeToString(type_);
+  DVLOG(1) << "Getting commit contribution for: " << ModelTypeToString(type_);
   return scoped_ptr<CommitContribution>();
 }
 
diff --git a/sync/engine/non_blocking_type_processor_core.h b/sync/engine/non_blocking_type_processor_core.h
index 412d2dd..0c1cc1c 100644
--- a/sync/engine/non_blocking_type_processor_core.h
+++ b/sync/engine/non_blocking_type_processor_core.h
@@ -57,8 +57,11 @@ class SYNC_EXPORT NonBlockingTypeProcessorCore
   // UpdateHandler implementation.
   virtual void GetDownloadProgress(
       sync_pb::DataTypeProgressMarker* progress_marker) const OVERRIDE;
+  virtual void GetDataTypeContext(sync_pb::DataTypeContext* context) const
+      OVERRIDE;
   virtual void ProcessGetUpdatesResponse(
       const sync_pb::DataTypeProgressMarker& progress_marker,
+      const sync_pb::DataTypeContext& mutated_context,
       const SyncEntityList& applicable_updates,
       sessions::StatusController* status) OVERRIDE;
   virtual void ApplyUpdates(sessions::StatusController* status) OVERRIDE;
diff --git a/sync/engine/update_handler.h b/sync/engine/update_handler.h
index eb1ee4d..29bdbc9 100644
--- a/sync/engine/update_handler.h
+++ b/sync/engine/update_handler.h
@@ -10,6 +10,7 @@
 #include "sync/base/sync_export.h"
 
 namespace sync_pb {
+class DataTypeContext;
 class DataTypeProgressMarker;
 class SyncEntity;
 }
@@ -35,6 +36,10 @@ class SYNC_EXPORT_PRIVATE UpdateHandler {
   virtual void GetDownloadProgress(
       sync_pb::DataTypeProgressMarker* progress_marker) const = 0;
 
+  // Fills |context| with the per-client datatype context, if one exists. Clears
+  // |context| otherwise.
+  virtual void GetDataTypeContext(sync_pb::DataTypeContext* context) const = 0;
+
   // Processes the contents of a GetUpdates response message.
   //
   // Should be invoked with the progress marker and set of SyncEntities from a
@@ -46,6 +51,7 @@ class SYNC_EXPORT_PRIVATE UpdateHandler {
   // this type.
   virtual void ProcessGetUpdatesResponse(
       const sync_pb::DataTypeProgressMarker& progress_marker,
+      const sync_pb::DataTypeContext& mutated_context,
       const SyncEntityList& applicable_updates,
       sessions::StatusController* status) = 0;
 
diff --git a/sync/internal_api/public/write_transaction.h b/sync/internal_api/public/write_transaction.h
index 9008b4f..8e797be 100644
--- a/sync/internal_api/public/write_transaction.h
+++ b/sync/internal_api/public/write_transaction.h
@@ -43,6 +43,9 @@ class SYNC_EXPORT WriteTransaction : public BaseTransaction {
   virtual syncable::BaseTransaction* GetWrappedTrans() const OVERRIDE;
   syncable::WriteTransaction* GetWrappedWriteTrans() { return transaction_; }
 
+  // Set's a |type|'s local context. Does not affect any individual entities.
+  void SetDataTypeContext(ModelType type, const std::string& context);
+
  protected:
   WriteTransaction() {}
 
diff --git a/sync/internal_api/write_transaction.cc b/sync/internal_api/write_transaction.cc
index d33093da..b0ed6a2 100644
--- a/sync/internal_api/write_transaction.cc
+++ b/sync/internal_api/write_transaction.cc
@@ -4,6 +4,7 @@
 
 #include "sync/internal_api/public/write_transaction.h"
 
+#include "sync/syncable/directory.h"
 #include "sync/syncable/syncable_write_transaction.h"
 
 namespace syncer {
@@ -36,4 +37,27 @@ syncable::BaseTransaction* WriteTransaction::GetWrappedTrans() const {
   return transaction_;
 }
 
+void WriteTransaction::SetDataTypeContext(ModelType type,
+                                          const std::string& context) {
+  sync_pb::DataTypeContext local_context;
+  GetDirectory()->GetDataTypeContext(transaction_,
+                                     type,
+                                     &local_context);
+  if (local_context.context() == context)
+    return;
+
+  if (!local_context.has_data_type_id()) {
+    local_context.set_data_type_id(
+        syncer::GetSpecificsFieldNumberFromModelType(type));
+  }
+  DCHECK_EQ(syncer::GetSpecificsFieldNumberFromModelType(type),
+            local_context.data_type_id());
+  DCHECK_GE(local_context.version(), 0);
+  local_context.set_version(local_context.version() + 1);
+  local_context.set_context(context);
+  GetDirectory()->SetDataTypeContext(transaction_,
+                                     type,
+                                     local_context);
+}
+
 }  // namespace syncer
diff --git a/sync/test/engine/mock_update_handler.cc b/sync/test/engine/mock_update_handler.cc
index 098086e..da2db88 100644
--- a/sync/test/engine/mock_update_handler.cc
+++ b/sync/test/engine/mock_update_handler.cc
@@ -24,10 +24,16 @@ void MockUpdateHandler::GetDownloadProgress(
   progress_marker->CopyFrom(progress_marker_);
 }
 
+void MockUpdateHandler::GetDataTypeContext(
+    sync_pb::DataTypeContext* context) const {
+  context->Clear();
+}
+
 void MockUpdateHandler::ProcessGetUpdatesResponse(
-      const sync_pb::DataTypeProgressMarker& progress_marker,
-      const SyncEntityList& applicable_updates,
-      sessions::StatusController* status) {
+    const sync_pb::DataTypeProgressMarker& progress_marker,
+    const sync_pb::DataTypeContext& mutated_context,
+    const SyncEntityList& applicable_updates,
+    sessions::StatusController* status) {
   progress_marker_.CopyFrom(progress_marker);
 }
 
diff --git a/sync/test/engine/mock_update_handler.h b/sync/test/engine/mock_update_handler.h
index 8d11f96..0cbeca3 100644
--- a/sync/test/engine/mock_update_handler.h
+++ b/sync/test/engine/mock_update_handler.h
@@ -20,8 +20,11 @@ class MockUpdateHandler : public UpdateHandler {
   // UpdateHandler implementation.
   virtual void GetDownloadProgress(
       sync_pb::DataTypeProgressMarker* progress_marker) const OVERRIDE;
+  virtual void GetDataTypeContext(sync_pb::DataTypeContext* context) const
+      OVERRIDE;
   virtual void ProcessGetUpdatesResponse(
       const sync_pb::DataTypeProgressMarker& progress_marker,
+      const sync_pb::DataTypeContext& mutated_context,
       const SyncEntityList& applicable_updates,
       sessions::StatusController* status) OVERRIDE;
   virtual void ApplyUpdates(sessions::StatusController* status) OVERRIDE;