sync: Loop committing items without downloading updates

sync/engine/process_commit_response_command.cc

 // Copyright (c) 2012 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.
 
 #include "sync/engine/process_commit_response_command.h"
 
 #include <cstddef>
 #include <set>
 #include <string>
 #include <vector>
@@ skipping 30 matching lines @@
 using syncable::SYNCER;
 using syncable::SYNCING;
 
 namespace browser_sync {
 
 using sessions::OrderedCommitSet;
 using sessions::StatusController;
 using sessions::SyncSession;
 using sessions::ConflictProgress;
 
-ProcessCommitResponseCommand::ProcessCommitResponseCommand() {}
+ProcessCommitResponseCommand::ProcessCommitResponseCommand(
+      const sessions::OrderedCommitSet& commit_set,
+      const ClientToServerMessage& commit_message,
+      const ClientToServerResponse& commit_response)
+  : commit_set_(commit_set),
+    commit_message_(commit_message),
+    commit_response_(commit_response) {
+}
+
 ProcessCommitResponseCommand::~ProcessCommitResponseCommand() {}
 
 std::set<ModelSafeGroup> ProcessCommitResponseCommand::GetGroupsToChange(
     const sessions::SyncSession& session) const {
   std::set<ModelSafeGroup> groups_with_commits;
 
   syncable::Directory* dir = session.context()->directory();
   syncable::ReadTransaction trans(FROM_HERE, dir);
-  const StatusController& status = session.status_controller();
-  for (size_t i = 0; i < status.commit_ids().size(); ++i) {
+  for (size_t i = 0; i < commit_set_.Size(); ++i) {
     groups_with_commits.insert(
-        GetGroupForModelType(status.GetUnrestrictedCommitModelTypeAt(i),
+        GetGroupForModelType(commit_set_.GetModelTypeAt(i),
                              session.routing_info()));
   }
 
   return groups_with_commits;
 }
 
 SyncerError ProcessCommitResponseCommand::ModelNeutralExecuteImpl(
-    sessions::SyncSession* session) {
-  const StatusController& status = session->status_controller();
-  const ClientToServerResponse& response(status.commit_response());
-  const vector<syncable::Id>& commit_ids(status.commit_ids());
+    SyncSession* session) {
+  syncable::Directory* dir = session->context()->directory();
+  const vector<syncable::Id>& commit_ids = commit_set_.GetAllCommitIds();
 
-  if (!response.has_commit()) {
-    // TODO(sync): What if we didn't try to commit anything?
+  if (!commit_response_.has_commit()) {
     LOG(WARNING) << "Commit response has no commit body!";
+    ClearSyncingBits(dir, commit_ids);
     return SERVER_RESPONSE_VALIDATION_FAILED;
   }
 
-  const CommitResponse& cr = response.commit();
-  int commit_count = commit_ids.size();
+  const CommitResponse& cr = commit_response_.commit();
+  int commit_count = commit_set_.Size();
   if (cr.entryresponse_size() != commit_count) {
     LOG(ERROR) << "Commit response has wrong number of entries! Expected:" <<
                commit_count << " Got:" << cr.entryresponse_size();
     for (int i = 0 ; i < cr.entryresponse_size() ; i++) {
       LOG(ERROR) << "Response #" << i << " Value: " <<
                  cr.entryresponse(i).response_type();
       if (cr.entryresponse(i).has_error_message())
         LOG(ERROR) << "  " << cr.entryresponse(i).error_message();
     }
+    ClearSyncingBits(dir, commit_ids);
     return SERVER_RESPONSE_VALIDATION_FAILED;
   }
   return SYNCER_OK;
 }
 
 SyncerError ProcessCommitResponseCommand::ModelChangingExecuteImpl(
     SyncSession* session) {
   SyncerError result = ProcessCommitResponse(session);
   ExtensionsActivityMonitor* monitor = session->context()->extensions_monitor();
-  if (session->status_controller().HasBookmarkCommitActivity() &&
-      session->status_controller().syncer_status()
+
+  // This is to be run on one model only: the bookmark model.
+  if (session->status_controller()
+          .ActiveGroupRestrictionIncludesModel(syncable::BOOKMARKS)) {
+    // If the commit failed, return the data to the ExtensionsActivityMonitor.
+    if (session->status_controller().syncer_status()
           .num_successful_bookmark_commits == 0) {
-    monitor->PutRecords(session->extensions_activity());
+      monitor->PutRecords(session->extensions_activity());
+    }
+    // Clear our cached data in either case.
     session->mutable_extensions_activity()->clear();
   }
+
   return result;
 }
 
 SyncerError ProcessCommitResponseCommand::ProcessCommitResponse(
     SyncSession* session) {
   syncable::Directory* dir = session->context()->directory();
+  StatusController* status = session->mutable_status_controller();
+  const CommitResponse& cr = commit_response_.commit();
+  const sync_pb::CommitMessage& commit_message = commit_message_.commit();
 
-  StatusController* status = session->mutable_status_controller();
-  const ClientToServerResponse& response(status->commit_response());
-  const CommitResponse& cr = response.commit();
-  const sync_pb::CommitMessage& commit_message =
-      status->commit_message().commit();
-
-  // If we try to commit a parent and child together and the parent conflicts
-  // the child will have a bad parent causing an error. As this is not a
-  // critical error, we trap it and don't LOG(ERROR). To enable this we keep
-  // a map of conflicting new folders.
   int transient_error_commits = 0;
   int conflicting_commits = 0;
   int error_commits = 0;
   int successes = 0;
-  set<syncable::Id> conflicting_new_folder_ids;
+
   set<syncable::Id> deleted_folders;
   ConflictProgress* conflict_progress = status->mutable_conflict_progress();
-  OrderedCommitSet::Projection proj = status->commit_id_projection();
+  OrderedCommitSet::Projection proj = status->commit_id_projection(
+      commit_set_);
+
   if (!proj.empty()) { // Scope for WriteTransaction.
     WriteTransaction trans(FROM_HERE, SYNCER, dir);
     for (size_t i = 0; i < proj.size(); i++) {
-      CommitResponse::ResponseType response_type =
-          ProcessSingleCommitResponse(&trans, cr.entryresponse(proj[i]),
-                                      commit_message.entries(proj[i]),
-                                      status->GetCommitIdAt(proj[i]),
-                                      &conflicting_new_folder_ids,
-                                      &deleted_folders);
+      CommitResponse::ResponseType response_type = ProcessSingleCommitResponse(
+          &trans,
+          cr.entryresponse(proj[i]),
+          commit_message.entries(proj[i]),
+          commit_set_.GetCommitIdAt(proj[i]),
+          &deleted_folders);
       switch (response_type) {
         case CommitResponse::INVALID_MESSAGE:
           ++error_commits;
           break;
         case CommitResponse::CONFLICT:
           ++conflicting_commits;
-          // Only server CONFLICT responses will activate conflict resolution.
           conflict_progress->AddServerConflictingItemById(
-              status->GetCommitIdAt(proj[i]));
+              commit_set_.GetCommitIdAt(proj[i]));
           break;
         case CommitResponse::SUCCESS:
           // TODO(sync): worry about sync_rate_ rate calc?
           ++successes;
-          if (status->GetCommitModelTypeAt(proj[i]) == syncable::BOOKMARKS)
+          if (commit_set_.GetModelTypeAt(proj[i]) == syncable::BOOKMARKS)
             status->increment_num_successful_bookmark_commits();
           status->increment_num_successful_commits();
           break;
         case CommitResponse::OVER_QUOTA:
           // We handle over quota like a retry, which is same as transient.
         case CommitResponse::RETRY:
         case CommitResponse::TRANSIENT_ERROR:
           ++transient_error_commits;
           break;
         default:
           LOG(FATAL) << "Bad return from ProcessSingleCommitResponse";
       }
     }
   }
 
   SyncerUtil::MarkDeletedChildrenSynced(dir, &deleted_folders);
 
   int commit_count = static_cast<int>(proj.size());
-  if (commit_count == (successes + conflicting_commits)) {
-    // We consider conflicting commits as a success because things will work out
-    // on their own when we receive them.  Flags will be set so that
-    // HasMoreToSync() will cause SyncScheduler to enter another sync cycle to
-    // handle this condition.
+  if (commit_count == successes) {
     return SYNCER_OK;
   } else if (error_commits > 0) {
     return SERVER_RETURN_UNKNOWN_ERROR;
   } else if (transient_error_commits > 0) {
     return SERVER_RETURN_TRANSIENT_ERROR;
+  } else if (conflicting_commits > 0) {
+    // This means that the server already has an item with this version, but
+    // we haven't seen that update yet.  The correct response is to go back
+    // to the download upates phase and try to fetch the conflicting update
+    // so we can resolve it.
+    //
+    // The easiest and safest way to handle this is to treat it as a transient
+    // error.  This will cause us to abort this sync cycle, wait a little while,
+    // then try again starting with a GetUpdate.
+    //
+    // TODO: Remove this option when the CONFLICT return value is fully
+    // deprecated.
+    return SERVER_RETURN_TRANSIENT_ERROR;

[rlarocque, 2012/04/13 02:02:22]

You could argue that there should be some other return value to explicitly
support going back to GetUpdates.  There's no need to add a delay here.

I would counter-argue that this code path is deprecated, so any hack that
doesn't send us into an infinite loop is good enough.  Adding more code to
handle this case correctly has a non-zero risk; I don't think it's worth it.

   } else {
     LOG(FATAL) << "Inconsistent counts when processing commit response";
     return SYNCER_OK;
   }
 }
 
 void LogServerError(const CommitResponse_EntryResponse& res) {
   if (res.has_error_message())
     LOG(WARNING) << "  " << res.error_message();
   else
     LOG(WARNING) << "  No detailed error message returned from server";
 }
 
 CommitResponse::ResponseType
 ProcessCommitResponseCommand::ProcessSingleCommitResponse(
     syncable::WriteTransaction* trans,
     const sync_pb::CommitResponse_EntryResponse& pb_server_entry,
     const sync_pb::SyncEntity& commit_request_entry,
     const syncable::Id& pre_commit_id,
-    std::set<syncable::Id>* conflicting_new_folder_ids,
     set<syncable::Id>* deleted_folders) {
 
   const CommitResponse_EntryResponse& server_entry =
       *static_cast<const CommitResponse_EntryResponse*>(&pb_server_entry);
   MutableEntry local_entry(trans, GET_BY_ID, pre_commit_id);
   CHECK(local_entry.good());
   bool syncing_was_set = local_entry.Get(SYNCING);
   local_entry.Put(SYNCING, false);
 
   CommitResponse::ResponseType response = (CommitResponse::ResponseType)
       server_entry.response_type();
   if (!CommitResponse::ResponseType_IsValid(response)) {
     LOG(ERROR) << "Commit response has unknown response type! Possibly out "
                "of date client?";
     return CommitResponse::INVALID_MESSAGE;
   }
   if (CommitResponse::TRANSIENT_ERROR == response) {
     DVLOG(1) << "Transient Error Committing: " << local_entry;
     LogServerError(server_entry);
     return CommitResponse::TRANSIENT_ERROR;
   }
   if (CommitResponse::INVALID_MESSAGE == response) {
     LOG(ERROR) << "Error Commiting: " << local_entry;
     LogServerError(server_entry);
     return response;
   }
   if (CommitResponse::CONFLICT == response) {
     DVLOG(1) << "Conflict Committing: " << local_entry;
-    // TODO(nick): conflicting_new_folder_ids is a purposeless anachronism.
-    if (!pre_commit_id.ServerKnows() && local_entry.Get(IS_DIR)) {
-      conflicting_new_folder_ids->insert(pre_commit_id);
-    }
     return response;
   }
   if (CommitResponse::RETRY == response) {
     DVLOG(1) << "Retry Committing: " << local_entry;
     return response;
   }
   if (CommitResponse::OVER_QUOTA == response) {
     LOG(WARNING) << "Hit deprecated OVER_QUOTA Committing: " << local_entry;
     return response;
   }
@@ skipping 226 matching lines @@
   // Make a note of any deleted folders, whose children would have
   // been recursively deleted.
   // TODO(nick): Here, commit_message.deleted() would be more correct than
   // local_entry->Get(IS_DEL).  For example, an item could be renamed, and then
   // deleted during the commit of the rename.  Unit test & fix.
   if (local_entry->Get(IS_DIR) && local_entry->Get(IS_DEL)) {
     deleted_folders->insert(local_entry->Get(ID));
   }
 }
 
+void ProcessCommitResponseCommand::ClearSyncingBits(
+    syncable::Directory *dir,
+    const vector<syncable::Id>& commit_ids) {
+  // This is part of the cleanup in the case of a failed commit.  Normally we
+  // would unset the SYNCING bit when processing the commit response.  In the
+  // failure case we don't process the response, so we need to clear those bits
+  // here.
+  syncable::WriteTransaction trans(FROM_HERE, syncable::SYNCER, dir);
+  for (size_t i = 0; i < commit_ids.size(); i++) {
+    syncable::MutableEntry entry(&trans, syncable::GET_BY_ID, commit_ids[i]);
+    if (entry.good()) {
+      entry.Put(syncable::SYNCING, false);
+    } else {
+      LOG(WARNING) << "Id: " << commit_ids[i] << " disappeared";
+    }
+  }
+}
+
 }  // namespace browser_sync