sync: Merge {Verify,Process}UpdatesCommand

sync/engine/process_updates_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_updates_command.h"
 
 #include <vector>
 
 #include "base/basictypes.h"
 #include "base/location.h"
 #include "sync/engine/syncer.h"
 #include "sync/engine/syncer_proto_util.h"
 #include "sync/engine/syncer_util.h"
 #include "sync/sessions/sync_session.h"
 #include "sync/syncable/directory.h"
 #include "sync/syncable/mutable_entry.h"
 #include "sync/syncable/syncable_proto_util.h"
 #include "sync/syncable/syncable_util.h"
 #include "sync/syncable/write_transaction.h"
 #include "sync/util/cryptographer.h"
 
 using std::vector;
 
 namespace syncer {
 
 using sessions::SyncSession;
 using sessions::StatusController;
 using sessions::UpdateProgress;
 
+using syncable::GET_BY_ID;
+
 ProcessUpdatesCommand::ProcessUpdatesCommand() {}
 ProcessUpdatesCommand::~ProcessUpdatesCommand() {}
 
 std::set<ModelSafeGroup> ProcessUpdatesCommand::GetGroupsToChange(
     const sessions::SyncSession& session) const {
-  return session.GetEnabledGroupsWithVerifiedUpdates();
-}
+  std::set<ModelSafeGroup> groups_with_updates;
+
+  const sync_pb::GetUpdatesResponse& updates =
+      session.status_controller().updates_response().get_updates();
+  for (int i = 0; i < updates.entries().size(); i++) {
+    groups_with_updates.insert(
+        GetGroupForModelType(GetModelType(updates.entries(i)),
+                             session.routing_info()));
+  }
+
+  return groups_with_updates;
+}
+
+namespace {
+
+// This function attempts to determine whether or not this update is genuinely
+// new, or if it is a reflection of one of our own commits.
+//
+// There is a known inaccuracy in its implementation.  If this update ends up
+// being applied to a local item with a different ID, we will count the change
+// as being a non-reflection update.  Fortunately, the server usually updates
+// our IDs correctly in its commit response, so a new ID during GetUpdate should
+// be rare.
+//
+// The only secnarios I can think of where this might happen are:
+// - We commit a  new item to the server, but we don't persist the
+// server-returned new ID to the database before we shut down.  On the GetUpdate
+// following the next restart, we will receive an update from the server that
+// updates its local ID.
+// - When two attempts to create an item with identical UNIQUE_CLIENT_TAG values
+// collide at the server.  I have seen this in testing.  When it happens, the
+// test server will send one of the clients a response to upate its local ID so
+// that both clients will refer to the item using the same ID going forward.  In
+// this case, we're right to assume that the update is not a reflection.
+//
+// For more information, see FindLocalIdToUpdate().
+bool UpdateContainsNewVersion(syncable::BaseTransaction *trans,
+                              const sync_pb::SyncEntity &update) {
+  int64 existing_version = -1; // The server always sends positive versions.
+  syncable::Entry existing_entry(trans, GET_BY_ID,
+                                 SyncableIdFromProto(update.id_string()));
+  if (existing_entry.good())
+    existing_version = existing_entry.Get(syncable::BASE_VERSION);
+
+  if (!existing_entry.good() && update.deleted()) {
+    // There are several possible explanations for this.  The most common cases
+    // will be first time sync and the redelivery of deletions we've already
+    // synced, accepted, and purged from our database.  In either case, the
+    // update is useless to us.  Let's count them all as "not new", even though
+    // that may not always be entirely accurate.
+    return false;
+  }
+
+  if (existing_entry.good() &&
+      !existing_entry.Get(syncable::UNIQUE_CLIENT_TAG).empty() &&
+      existing_entry.Get(syncable::IS_DEL) &&
+      update.deleted()) {
+    // Unique client tags will have their version set to zero when they're
+    // deleted.  The usual version comparison logic won't be able to detect
+    // reflections of these items.  Instead, we assume any received tombstones
+    // are reflections.  That should be correct most of the time.
+    return false;
+  }
+
+  return existing_version < update.version();
+}
+
+}  // namespace
 
 SyncerError ProcessUpdatesCommand::ModelChangingExecuteImpl(
     SyncSession* session) {
   syncable::Directory* dir = session->context()->directory();
 
-  const sessions::UpdateProgress* progress =
-      session->status_controller().update_progress();
-  if (!progress)
-    return SYNCER_OK;  // Nothing to do.
-
   syncable::WriteTransaction trans(FROM_HERE, syncable::SYNCER, dir);
-  vector<sessions::VerifiedUpdate>::const_iterator it;
-  for (it = progress->VerifiedUpdatesBegin();
-       it != progress->VerifiedUpdatesEnd();
-       ++it) {
-    const sync_pb::SyncEntity& update = it->second;
-
-    if (it->first != VERIFY_SUCCESS && it->first != VERIFY_UNDELETE)
+
+  sessions::StatusController* status = session->mutable_status_controller();
+  const sync_pb::GetUpdatesResponse& updates =
+      status->updates_response().get_updates();
+  int update_count = updates.entries().size();
+
+  ModelTypeSet requested_types = GetRoutingInfoTypes(
+      session->routing_info());
+
+  DVLOG(1) << update_count << " entries to verify";
+  for (int i = 0; i < update_count; i++) {
+    const sync_pb::SyncEntity& update = updates.entries(i);
+    ModelSafeGroup g = GetGroupForModelType(GetModelType(update),
+                                            session->routing_info());
+    if (g != status->group_restriction())
       continue;
-    switch (ProcessUpdate(update,
-                          dir->GetCryptographer(&trans),
-                          &trans)) {
-      case SUCCESS_PROCESSED:
-      case SUCCESS_STORED:
-        break;
-      default:
-        NOTREACHED();
-        break;
+
+    VerifyResult verify_result = VerifyUpdate(&trans, update,
+                                              requested_types,
+                                              session->routing_info());
+    status->mutable_update_progress()->AddVerifyResult(verify_result, update);
+    status->increment_num_updates_downloaded_by(1);
+    if (!UpdateContainsNewVersion(&trans, update))
+      status->increment_num_reflected_updates_downloaded_by(1);
+    if (update.deleted())
+      status->increment_num_tombstone_updates_downloaded_by(1);
+
+    if (verify_result != VERIFY_SUCCESS && verify_result != VERIFY_UNDELETE)

[rlarocque, 2012/10/08 23:16:41]

This function up until this point is mostly copied and pasted from
VerifyUpdatesCommand::ModelChangingExecuteImpl().

At this point we transition over to ProcessUpdatesCommand.  However, instead of
retrieving the update verification status from the UpdateProgress struct, we
simply carry on from the current update verification result.  Process and Verify
now share the same iteration logic.

+      continue;
+
+    ServerUpdateProcessingResult process_result =
+       ProcessUpdate(update, dir->GetCryptographer(&trans), &trans);
+
+    DCHECK(process_result == SUCCESS_PROCESSED ||
+           process_result == SUCCESS_STORED);
+  }
+
+  return SYNCER_OK;
+}
+
+namespace {
+
+// In the event that IDs match, but tags differ AttemptReuniteClient tag
+// will have refused to unify the update.
+// We should not attempt to apply it at all since it violates consistency
+// rules.
+VerifyResult VerifyTagConsistency(const sync_pb::SyncEntity& entry,
+                                  const syncable::MutableEntry& same_id) {
+  if (entry.has_client_defined_unique_tag() &&
+      entry.client_defined_unique_tag() !=
+          same_id.Get(syncable::UNIQUE_CLIENT_TAG)) {
+    return VERIFY_FAIL;
+  }
+  return VERIFY_UNDECIDED;
+}
+
+}  // namespace
+
+VerifyResult ProcessUpdatesCommand::VerifyUpdate(
+    syncable::WriteTransaction* trans, const sync_pb::SyncEntity& entry,
+    ModelTypeSet requested_types,
+    const ModelSafeRoutingInfo& routes) {
+  syncable::Id id = SyncableIdFromProto(entry.id_string());
+  VerifyResult result = VERIFY_FAIL;
+
+  const bool deleted = entry.has_deleted() && entry.deleted();
+  const bool is_directory = IsFolder(entry);
+  const ModelType model_type = GetModelType(entry);
+
+  if (!id.ServerKnows()) {
+    LOG(ERROR) << "Illegal negative id in received updates";
+    return result;
+  }
+  {
+    const std::string name = SyncerProtoUtil::NameFromSyncEntity(entry);
+    if (name.empty() && !deleted) {
+      LOG(ERROR) << "Zero length name in non-deleted update";
+      return result;
     }
   }
 
-  StatusController* status = session->mutable_status_controller();
-  status->mutable_update_progress()->ClearVerifiedUpdates();
-  return SYNCER_OK;
+  syncable::MutableEntry same_id(trans, GET_BY_ID, id);
+  result = VerifyNewEntry(entry, &same_id, deleted);
+
+  ModelType placement_type = !deleted ? GetModelType(entry)
+      : same_id.good() ? same_id.GetModelType() : UNSPECIFIED;
+
+  if (VERIFY_UNDECIDED == result) {
+    result = VerifyTagConsistency(entry, same_id);
+  }
+
+  if (VERIFY_UNDECIDED == result) {
+    if (deleted) {
+      // For deletes the server could send tombostones for items that
+      // the client did not request. If so ignore those items.
+      if (IsRealDataType(placement_type) &&
+          !requested_types.Has(placement_type)) {
+        result = VERIFY_SKIP;
+      } else {
+        result = VERIFY_SUCCESS;
+      }
+    }
+  }
+
+  // If we have an existing entry, we check here for updates that break
+  // consistency rules.
+  if (VERIFY_UNDECIDED == result) {
+    result = VerifyUpdateConsistency(trans, entry, &same_id,
+                                     deleted, is_directory, model_type);
+  }
+
+  if (VERIFY_UNDECIDED == result)
+    result = VERIFY_SUCCESS;  // No news is good news.
+
+  return result;  // This might be VERIFY_SUCCESS as well
 }
 
 namespace {
 // Returns true if the entry is still ok to process.
 bool ReverifyEntry(syncable::WriteTransaction* trans,
                    const sync_pb::SyncEntity& entry,
                    syncable::MutableEntry* same_id) {
 
   const bool deleted = entry.has_deleted() && entry.deleted();
   const bool is_directory = IsFolder(entry);
@@ skipping 22 matching lines @@
 
   // FindLocalEntryToUpdate has veto power.
   if (local_id.IsNull()) {
     return SUCCESS_PROCESSED;  // The entry has become irrelevant.
   }
 
   CreateNewEntry(trans, local_id);
 
   // We take a two step approach. First we store the entries data in the
   // server fields of a local entry and then move the data to the local fields
-  syncable::MutableEntry target_entry(trans, syncable::GET_BY_ID, local_id);
+  syncable::MutableEntry target_entry(trans, GET_BY_ID, local_id);
 
   // We need to run the Verify checks again; the world could have changed
-  // since VerifyUpdatesCommand.
+  // since we last verified.

[rlarocque, 2012/10/08 23:16:41]

This is at least somewhat redundant.  I'll investigate removing it in future
patches.

   if (!ReverifyEntry(trans, update, &target_entry)) {
     return SUCCESS_PROCESSED;  // The entry has become irrelevant.
   }
 
   // If we're repurposing an existing local entry with a new server ID,
   // change the ID now, after we're sure that the update can succeed.
   if (local_id != server_id) {
     DCHECK(!update.deleted());
     ChangeEntryIDAndUpdateChildren(trans, &target_entry, server_id);
     // When IDs change, versions become irrelevant.  Forcing BASE_VERSION
@@ skipping 49 matching lines @@
     target_entry.Put(syncable::BASE_SERVER_SPECIFICS,
                      sync_pb::EntitySpecifics());
   }
 
   UpdateServerFieldsFromUpdate(&target_entry, update, name);
 
   return SUCCESS_PROCESSED;
 }
 
 }  // namespace syncer