[Sync] Add support for retrying a getupdates due to a context change

We detect a context conflict using the version value, and if detected force the
getupdates to retry.

BUG=360280

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

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

10 files changed, 139 insertions, 31 deletions

diff --git a/sync/engine/backoff_delay_provider.cc b/sync/engine/backoff_delay_provider.cc
index f8e2750..bff3fd6 100644
--- a/sync/engine/backoff_delay_provider.cc
+++ b/sync/engine/backoff_delay_provider.cc
@@ -97,6 +97,11 @@ TimeDelta BackoffDelayProvider::GetInitialDelay(
     return short_initial_backoff_;
   }
 
+  // If a datatype decides the GetUpdates must be retried (e.g. because the
+  // context has been updated since the request), use the short delay.
+  if (state.last_download_updates_result == DATATYPE_TRIGGERED_RETRY)
+    return short_initial_backoff_;
+
   // When the server tells us we have a conflict, then we should download the
   // latest updates so we can see the conflict ourselves, resolve it locally,
   // then try again to commit.  Running another sync cycle will do all these
@@ -105,9 +110,8 @@ TimeDelta BackoffDelayProvider::GetInitialDelay(
   // TODO(sync): We shouldn't need to handle this in BackoffDelayProvider.
   // There should be a way to deal with protocol errors before we get to this
   // point.
-  if (state.commit_result == SERVER_RETURN_CONFLICT) {
+  if (state.commit_result == SERVER_RETURN_CONFLICT)
     return short_initial_backoff_;
-  }
 
   return default_initial_backoff_;
 }
diff --git a/sync/engine/backoff_delay_provider_unittest.cc b/sync/engine/backoff_delay_provider_unittest.cc
index 1bf17cb..2e58186 100644
--- a/sync/engine/backoff_delay_provider_unittest.cc
+++ b/sync/engine/backoff_delay_provider_unittest.cc
@@ -57,6 +57,10 @@ TEST_F(BackoffDelayProviderTest, GetInitialDelay) {
   EXPECT_EQ(kInitialBackoffRetrySeconds,
             delay->GetInitialDelay(state).InSeconds());
 
+  state.last_download_updates_result = DATATYPE_TRIGGERED_RETRY;
+  EXPECT_EQ(kInitialBackoffImmediateRetrySeconds,
+            delay->GetInitialDelay(state).InSeconds());
+
   state.last_download_updates_result = SYNCER_OK;
   // Note that updating credentials triggers a canary job, trumping
   // the initial delay, but in theory we still expect this function to treat
@@ -99,6 +103,10 @@ TEST_F(BackoffDelayProviderTest, GetInitialDelayWithOverride) {
   EXPECT_EQ(kInitialBackoffShortRetrySeconds,
             delay->GetInitialDelay(state).InSeconds());
 
+  state.last_download_updates_result = DATATYPE_TRIGGERED_RETRY;
+  EXPECT_EQ(kInitialBackoffImmediateRetrySeconds,
+            delay->GetInitialDelay(state).InSeconds());
+
   state.last_download_updates_result = SYNCER_OK;
   // Note that updating credentials triggers a canary job, trumping
   // the initial delay, but in theory we still expect this function to treat
diff --git a/sync/engine/directory_update_handler.cc b/sync/engine/directory_update_handler.cc
index 5509520..72cdf12 100644
--- a/sync/engine/directory_update_handler.cc
+++ b/sync/engine/directory_update_handler.cc
@@ -37,19 +37,12 @@ void DirectoryUpdateHandler::GetDataTypeContext(
   dir_->GetDataTypeContext(&trans, type_, context);
 }
 
-void DirectoryUpdateHandler::ProcessGetUpdatesResponse(
+SyncerError 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_);
-  UpdateSyncEntities(&trans, applicable_updates, status);
-
-  if (IsValidProgressMarker(progress_marker)) {
-    ExpireEntriesIfNeeded(&trans, progress_marker);
-    UpdateProgressMarker(progress_marker);
-  }
-
   if (mutated_context.has_context()) {
     sync_pb::DataTypeContext local_context;
     dir_->GetDataTypeContext(&trans, type_, &local_context);
@@ -61,8 +54,21 @@ void DirectoryUpdateHandler::ProcessGetUpdatesResponse(
         local_context.context() != mutated_context.context()) {
       dir_->SetDataTypeContext(&trans, type_, mutated_context);
       // TODO(zea): trigger the datatype's UpdateDataTypeContext method.
+    } else if (mutated_context.version() < local_context.version()) {
+      // A GetUpdates using the old context was in progress when the context was
+      // set. Fail this get updates cycle, to force a retry.
+      DVLOG(1) << "GU Context conflict detected, forcing GU retry.";
+      return DATATYPE_TRIGGERED_RETRY;
     }
   }
+
+  UpdateSyncEntities(&trans, applicable_updates, status);
+
+  if (IsValidProgressMarker(progress_marker)) {
+    ExpireEntriesIfNeeded(&trans, progress_marker);
+    UpdateProgressMarker(progress_marker);
+  }
+  return SYNCER_OK;
 }
 
 void DirectoryUpdateHandler::ApplyUpdates(sessions::StatusController* status) {
diff --git a/sync/engine/directory_update_handler.h b/sync/engine/directory_update_handler.h
index 74c7046..ef5052de 100644
--- a/sync/engine/directory_update_handler.h
+++ b/sync/engine/directory_update_handler.h
@@ -52,7 +52,7 @@ class SYNC_EXPORT_PRIVATE DirectoryUpdateHandler : public UpdateHandler {
       sync_pb::DataTypeProgressMarker* progress_marker) const OVERRIDE;
   virtual void GetDataTypeContext(sync_pb::DataTypeContext* context) const
       OVERRIDE;
-  virtual void ProcessGetUpdatesResponse(
+  virtual SyncerError ProcessGetUpdatesResponse(
       const sync_pb::DataTypeProgressMarker& progress_marker,
       const sync_pb::DataTypeContext& mutated_context,
       const SyncEntityList& applicable_updates,
diff --git a/sync/engine/directory_update_handler_unittest.cc b/sync/engine/directory_update_handler_unittest.cc
index 6d1b35a..ec8cc35 100644
--- a/sync/engine/directory_update_handler_unittest.cc
+++ b/sync/engine/directory_update_handler_unittest.cc
@@ -195,14 +195,14 @@ TEST_F(DirectoryUpdateHandlerProcessUpdateTest,
 
 // Test the receipt of a non-bookmark item.
 TEST_F(DirectoryUpdateHandlerProcessUpdateTest, ReceiveNonBookmarkItem) {
-  DirectoryUpdateHandler handler(dir(), PREFERENCES, ui_worker());
+  DirectoryUpdateHandler handler(dir(), AUTOFILL, ui_worker());
   sync_pb::GetUpdatesResponse gu_response;
   sessions::StatusController status;
 
   std::string root = syncable::GetNullId().GetServerId();
   syncable::Id server_id = syncable::Id::CreateFromServerId("xyz");
   scoped_ptr<sync_pb::SyncEntity> e =
-      CreateUpdate(SyncableIdToProto(server_id), root, PREFERENCES);
+      CreateUpdate(SyncableIdToProto(server_id), root, AUTOFILL);
   e->set_server_defined_unique_tag("9PGRuKdX5sHyGMB17CvYTXuC43I=");
 
   // Add it to the applicable updates list.
@@ -283,7 +283,9 @@ TEST_F(DirectoryUpdateHandlerProcessUpdateTest, GarbageCollectionByVersion) {
   updates.push_back(e2.get());
 
   // Process and apply updates.
-  handler.ProcessGetUpdatesResponse(progress, context, updates, &status);
+  EXPECT_EQ(
+      SYNCER_OK,
+      handler.ProcessGetUpdatesResponse(progress, context, updates, &status));
   handler.ApplyUpdates(&status);
 
   // Verify none is deleted because they are unapplied during GC.
@@ -293,14 +295,93 @@ 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, context, SyncEntityList(), &status);
+  EXPECT_EQ(SYNCER_OK,
+            handler.ProcessGetUpdatesResponse(
+                progress, context, SyncEntityList(), &status));
   handler.ApplyUpdates(&status);
   EXPECT_TRUE(EntryExists(type_root->id_string()));
   EXPECT_FALSE(EntryExists(e1->id_string()));
   EXPECT_TRUE(EntryExists(e2->id_string()));
 }
 
+TEST_F(DirectoryUpdateHandlerProcessUpdateTest, ContextVersion) {
+  DirectoryUpdateHandler handler(dir(), SYNCED_NOTIFICATIONS, ui_worker());
+  sessions::StatusController status;
+  int field_number = GetSpecificsFieldNumberFromModelType(SYNCED_NOTIFICATIONS);
+
+  sync_pb::DataTypeProgressMarker progress;
+  progress.set_data_type_id(
+      GetSpecificsFieldNumberFromModelType(SYNCED_NOTIFICATIONS));
+  progress.set_token("token");
+
+  sync_pb::DataTypeContext old_context;
+  old_context.set_version(1);
+  old_context.set_context("data");
+  old_context.set_data_type_id(field_number);
+
+  scoped_ptr<sync_pb::SyncEntity> type_root =
+      CreateUpdate(SyncableIdToProto(syncable::Id::CreateFromServerId("root")),
+                   syncable::GetNullId().GetServerId(),
+                   SYNCED_NOTIFICATIONS);
+  type_root->set_server_defined_unique_tag(
+      ModelTypeToRootTag(SYNCED_NOTIFICATIONS));
+  type_root->set_folder(true);
+  scoped_ptr<sync_pb::SyncEntity> e1 =
+      CreateUpdate(SyncableIdToProto(syncable::Id::CreateFromServerId("e1")),
+                   type_root->id_string(),
+                   SYNCED_NOTIFICATIONS);
+
+  SyncEntityList updates;
+  updates.push_back(type_root.get());
+  updates.push_back(e1.get());
+
+  // The first response should be processed fine.
+  EXPECT_EQ(SYNCER_OK,
+            handler.ProcessGetUpdatesResponse(
+                progress, old_context, updates, &status));
+  handler.ApplyUpdates(&status);
+
+  EXPECT_TRUE(EntryExists(type_root->id_string()));
+  EXPECT_TRUE(EntryExists(e1->id_string()));
+
+  {
+    sync_pb::DataTypeContext dir_context;
+    syncable::ReadTransaction trans(FROM_HERE, dir());
+    trans.directory()->GetDataTypeContext(
+        &trans, SYNCED_NOTIFICATIONS, &dir_context);
+    EXPECT_EQ(old_context.SerializeAsString(), dir_context.SerializeAsString());
+  }
+
+  sync_pb::DataTypeContext new_context;
+  new_context.set_version(0);
+  new_context.set_context("old");
+  new_context.set_data_type_id(field_number);
+
+  scoped_ptr<sync_pb::SyncEntity> e2 =
+      CreateUpdate(SyncableIdToProto(syncable::Id::CreateFromServerId("e2")),
+                   type_root->id_string(),
+                   SYNCED_NOTIFICATIONS);
+  updates.clear();
+  updates.push_back(e2.get());
+
+  // The second response, with an old context version, should result in an
+  // error and the updates should be dropped.
+  EXPECT_EQ(DATATYPE_TRIGGERED_RETRY,
+            handler.ProcessGetUpdatesResponse(
+                progress, new_context, updates, &status));
+  handler.ApplyUpdates(&status);
+
+  EXPECT_FALSE(EntryExists(e2->id_string()));
+
+  {
+    sync_pb::DataTypeContext dir_context;
+    syncable::ReadTransaction trans(FROM_HERE, dir());
+    trans.directory()->GetDataTypeContext(
+        &trans, SYNCED_NOTIFICATIONS, &dir_context);
+    EXPECT_EQ(old_context.SerializeAsString(), dir_context.SerializeAsString());
+  }
+}
+
 // A test harness for tests that focus on applying updates.
 //
 // Update application is performed when we want to take updates that were
diff --git a/sync/engine/get_updates_processor.cc b/sync/engine/get_updates_processor.cc
index d0aa9b9..8666e0a 100644
--- a/sync/engine/get_updates_processor.cc
+++ b/sync/engine/get_updates_processor.cc
@@ -284,9 +284,10 @@ SyncerError GetUpdatesProcessor::ProcessResponse(
   }
   status->set_num_server_changes_remaining(gu_response.changes_remaining());
 
-  if (!ProcessGetUpdatesResponse(request_types, gu_response, status)) {
-    return SERVER_RESPONSE_VALIDATION_FAILED;
-  }
+  syncer::SyncerError result =
+      ProcessGetUpdatesResponse(request_types, gu_response, status);
+  if (result != syncer::SYNCER_OK)
+    return result;
 
   if (gu_response.changes_remaining() == 0) {
     return SYNCER_OK;
@@ -295,7 +296,7 @@ SyncerError GetUpdatesProcessor::ProcessResponse(
   }
 }
 
-bool GetUpdatesProcessor::ProcessGetUpdatesResponse(
+syncer::SyncerError GetUpdatesProcessor::ProcessGetUpdatesResponse(
     ModelTypeSet gu_types,
     const sync_pb::GetUpdatesResponse& gu_response,
     sessions::StatusController* status_controller) {
@@ -309,7 +310,7 @@ bool GetUpdatesProcessor::ProcessGetUpdatesResponse(
                                  &progress_index_by_type);
   if (gu_types.Size() != progress_index_by_type.size()) {
     NOTREACHED() << "Missing progress markers in GetUpdates response.";
-    return false;
+    return syncer::SERVER_RESPONSE_VALIDATION_FAILED;
   }
 
   TypeToIndexMap context_by_type;
@@ -334,11 +335,14 @@ bool GetUpdatesProcessor::ProcessGetUpdatesResponse(
       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);
+      syncer::SyncerError result =
+          update_handler_iter->second->ProcessGetUpdatesResponse(
+              gu_response.new_progress_marker(progress_marker_iter->second),
+              context,
+              updates_iter->second,
+              status_controller);
+      if (result != syncer::SYNCER_OK)
+        return result;
     } else {
       DLOG(WARNING)
           << "Ignoring received updates of a type we can't handle.  "
@@ -349,7 +353,7 @@ bool GetUpdatesProcessor::ProcessGetUpdatesResponse(
   DCHECK(progress_marker_iter == progress_index_by_type.end() &&
          updates_iter == updates_by_type.end());
 
-  return true;
+  return syncer::SYNCER_OK;
 }
 
 void GetUpdatesProcessor::ApplyUpdates(
diff --git a/sync/engine/get_updates_processor.h b/sync/engine/get_updates_processor.h
index 40a5c3a..e66d100 100644
--- a/sync/engine/get_updates_processor.h
+++ b/sync/engine/get_updates_processor.h
@@ -82,7 +82,7 @@ class SYNC_EXPORT_PRIVATE GetUpdatesProcessor {
                               sessions::StatusController* status);
 
   // Processes a GetUpdates responses for each type.
-  bool ProcessGetUpdatesResponse(
+  syncer::SyncerError ProcessGetUpdatesResponse(
       ModelTypeSet gu_types,
       const sync_pb::GetUpdatesResponse& gu_response,
       sessions::StatusController* status_controller);
diff --git a/sync/engine/non_blocking_type_processor_core.cc b/sync/engine/non_blocking_type_processor_core.cc
index c1bc60a..ea0f918 100644
--- a/sync/engine/non_blocking_type_processor_core.cc
+++ b/sync/engine/non_blocking_type_processor_core.cc
@@ -44,7 +44,7 @@ void NonBlockingTypeProcessorCore::GetDataTypeContext(
   context->Clear();
 }
 
-void NonBlockingTypeProcessorCore::ProcessGetUpdatesResponse(
+SyncerError NonBlockingTypeProcessorCore::ProcessGetUpdatesResponse(
     const sync_pb::DataTypeProgressMarker& progress_marker,
     const sync_pb::DataTypeContext& mutated_context,
     const SyncEntityList& applicable_updates,
@@ -53,6 +53,7 @@ void NonBlockingTypeProcessorCore::ProcessGetUpdatesResponse(
   // TODO(rlarocque): Implement this properly.  crbug.com/351005.
   DVLOG(1) << "Processing updates response for: " << ModelTypeToString(type_);
   progress_marker_ = progress_marker;
+  return SYNCER_OK;
 }
 
 void NonBlockingTypeProcessorCore::ApplyUpdates(
diff --git a/sync/engine/non_blocking_type_processor_core.h b/sync/engine/non_blocking_type_processor_core.h
index 0c1cc1c..c9cec88 100644
--- a/sync/engine/non_blocking_type_processor_core.h
+++ b/sync/engine/non_blocking_type_processor_core.h
@@ -59,7 +59,7 @@ class SYNC_EXPORT NonBlockingTypeProcessorCore
       sync_pb::DataTypeProgressMarker* progress_marker) const OVERRIDE;
   virtual void GetDataTypeContext(sync_pb::DataTypeContext* context) const
       OVERRIDE;
-  virtual void ProcessGetUpdatesResponse(
+  virtual SyncerError ProcessGetUpdatesResponse(
       const sync_pb::DataTypeProgressMarker& progress_marker,
       const sync_pb::DataTypeContext& mutated_context,
       const SyncEntityList& applicable_updates,
diff --git a/sync/engine/update_handler.h b/sync/engine/update_handler.h
index 29bdbc9..e4c9abc 100644
--- a/sync/engine/update_handler.h
+++ b/sync/engine/update_handler.h
@@ -8,6 +8,7 @@
 #include <vector>
 
 #include "sync/base/sync_export.h"
+#include "sync/internal_api/public/util/syncer_error.h"
 
 namespace sync_pb {
 class DataTypeContext;
@@ -49,7 +50,10 @@ class SYNC_EXPORT_PRIVATE UpdateHandler {
   //
   // In this context, "applicable_updates" means the set of updates belonging to
   // this type.
-  virtual void ProcessGetUpdatesResponse(
+  //
+  // Returns SYNCER_OK if the all data was processed successfully, a syncer
+  // error otherwise.
+  virtual SyncerError ProcessGetUpdatesResponse(
       const sync_pb::DataTypeProgressMarker& progress_marker,
       const sync_pb::DataTypeContext& mutated_context,
       const SyncEntityList& applicable_updates,