sync: Implement per-type update processing

sync/engine/process_updates_util.cc

 // Copyright 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 "sync/engine/process_updates_util.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/model_neutral_mutable_entry.h"
 #include "sync/syncable/syncable_model_neutral_write_transaction.h"
 #include "sync/syncable/syncable_proto_util.h"
 #include "sync/syncable/syncable_util.h"
 #include "sync/util/cryptographer.h"
 
-using std::vector;
-
 namespace syncer {
 
-using sessions::SyncSession;
 using sessions::StatusController;
 
 using syncable::GET_BY_ID;
 
-ProcessUpdatesCommand::ProcessUpdatesCommand() {}
-ProcessUpdatesCommand::~ProcessUpdatesCommand() {}
-
 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.
@@ skipping 36 matching lines @@
     // 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::ExecuteImpl(SyncSession* session) {
-  syncable::Directory* dir = session->context()->directory();
-
-  syncable::ModelNeutralWriteTransaction trans(
-      FROM_HERE, syncable::SYNCER, dir);
-
-  sessions::StatusController* status = session->mutable_status_controller();
-  const sync_pb::GetUpdatesResponse& updates =
-      status->updates_response().get_updates();
-  ModelTypeSet requested_types = GetRoutingInfoTypes(
-      session->context()->routing_info());
-
-  TypeSyncEntityMap updates_by_type;
-
-  PartitionUpdatesByType(updates, &updates_by_type);
-
-  int update_count = updates.entries().size();
-  status->increment_num_updates_downloaded_by(update_count);
-
-  DVLOG(1) << update_count << " entries to verify";
-  for (TypeSyncEntityMap::iterator it = updates_by_type.begin();
-       it != updates_by_type.end(); ++it) {
-    for (SyncEntityList::iterator update_it = it->second.begin();
-         update_it != it->second.end(); ++update_it) {
-      if (!UpdateContainsNewVersion(&trans, **update_it))
-        status->increment_num_reflected_updates_downloaded_by(1);
-      if ((*update_it)->deleted())
-        status->increment_num_tombstone_updates_downloaded_by(1);
-      VerifyResult verify_result = VerifyUpdate(
-          &trans,
-          **update_it,
-          requested_types,
-          session->context()->routing_info());
-      if (verify_result != VERIFY_SUCCESS && verify_result != VERIFY_UNDELETE)
-        continue;
-      ProcessUpdate(**update_it, dir->GetCryptographer(&trans), &trans);
-    }
-  }
-
-  return SYNCER_OK;
-}
-
-void ProcessUpdatesCommand::PartitionUpdatesByType(
+void PartitionUpdatesByType(
     const sync_pb::GetUpdatesResponse& updates,
+    ModelTypeSet requested_types,
     TypeSyncEntityMap* updates_by_type) {
   int update_count = updates.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);
     ModelType type = GetModelType(update);
-    DCHECK(IsRealDataType(type));
-    (*updates_by_type)[type].push_back(&update);
+    if (!IsRealDataType(type)) {
+      NOTREACHED() << "Received update with invalid type.";
+      continue;
+    }
+
+    TypeSyncEntityMap::iterator it = updates_by_type->find(type);
+    if (it == updates_by_type->end()) {
+      DLOG(WARNING) << "Skipping update for unexpected type "
+          << ModelTypeToString(type);

[Nicolas Zea, 2013/10/25 21:04:41]

nit: align <<

[rlarocque, 2013/10/25 22:29:47]

Done.

+      continue;
+    }
+
+    it->second.push_back(&update);

[Nicolas Zea, 2013/10/25 21:04:41]

Same comment applies here. I think you can get the same benefit by just storing
the entry index within the updates list and therefore not run the risk that you
keep a pointer to an object that has been de-alloc'd.

[rlarocque, 2013/10/25 22:29:47]

The trade off here is not so clear.  We'd need to pass the callee a list of
indices and the entire vector.  Then the callee has the opportunity to make the
more subtle and dangerous error of processing items it's not supposed to.

   }
 }
 
+void ProcessDownloadedUpdates(
+    syncable::Directory* dir,
+    syncable::ModelNeutralWriteTransaction* trans,
+    ModelType type,
+    const SyncEntityList& applicable_updates,
+    sessions::StatusController* status) {
+  for (SyncEntityList::const_iterator update_it = applicable_updates.begin();
+       update_it != applicable_updates.end(); ++update_it) {
+    DCHECK_EQ(type, GetModelType(**update_it));
+    if (!UpdateContainsNewVersion(trans, **update_it))
+      status->increment_num_reflected_updates_downloaded_by(1);
+    if ((*update_it)->deleted())
+      status->increment_num_tombstone_updates_downloaded_by(1);
+    VerifyResult verify_result = VerifyUpdate(trans, **update_it, type);
+    if (verify_result != VERIFY_SUCCESS && verify_result != VERIFY_UNDELETE)
+      continue;
+    ProcessUpdate(**update_it, dir->GetCryptographer(trans), trans);
+  }
+}
+
 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::ModelNeutralMutableEntry& same_id) {
   if (entry.has_client_defined_unique_tag() &&
       entry.client_defined_unique_tag() !=
           same_id.GetUniqueClientTag()) {
     return VERIFY_FAIL;
   }
   return VERIFY_UNDECIDED;
 }
 
 }  // namespace
 
-VerifyResult ProcessUpdatesCommand::VerifyUpdate(
+VerifyResult VerifyUpdate(
     syncable::ModelNeutralWriteTransaction* trans,
     const sync_pb::SyncEntity& entry,
-    ModelTypeSet requested_types,
-    const ModelSafeRoutingInfo& routes) {
+    ModelType requested_type) {
   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;
@@ skipping 13 matching lines @@
       : 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)) {
+      if (IsRealDataType(placement_type) && requested_type != 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) {
@@ skipping 20 matching lines @@
   return VERIFY_SUCCESS == VerifyUpdateConsistency(trans,
                                                    entry,
                                                    deleted,
                                                    is_directory,
                                                    model_type,
                                                    same_id);
 }
 }  // namespace
 
 // Process a single update. Will avoid touching global state.
-void ProcessUpdatesCommand::ProcessUpdate(
+void ProcessUpdate(
     const sync_pb::SyncEntity& update,
     const Cryptographer* cryptographer,
     syncable::ModelNeutralWriteTransaction* const trans) {
   const syncable::Id& server_id = SyncableIdFromProto(update.id_string());
   const std::string name = SyncerProtoUtil::NameFromSyncEntity(update);
 
   // Look to see if there's a local item that should recieve this update,
   // maybe due to a duplicate client tag or a lost commit response.
   syncable::Id local_id = FindLocalIdToUpdate(trans, update);
 
@@ skipping 85 matching lines @@
     target_entry.PutBaseServerSpecifics(
                      sync_pb::EntitySpecifics());
   }
 
   UpdateServerFieldsFromUpdate(&target_entry, update, name);
 
   return;
 }
 
 }  // namespace syncer