sync: Implement per-type update processing

sync/engine/download.cc

Index: sync/engine/download.cc
diff --git a/sync/engine/download.cc b/sync/engine/download.cc
index acda8349edcf2981458b0b680a973da5c4a32b3e..d590a60a6d076249aba0fd8b8723cc1882d6a6d5 100644
--- a/sync/engine/download.cc
+++ b/sync/engine/download.cc
@@ -7,8 +7,8 @@
 #include <string>
 
 #include "base/command_line.h"
-#include "sync/engine/process_updates_command.h"
-#include "sync/engine/store_timestamps_command.h"
+#include "sync/engine/process_updates_util.h"
+#include "sync/engine/sync_directory_update_handler.h"
 #include "sync/engine/syncer.h"
 #include "sync/engine/syncer_proto_util.h"
 #include "sync/sessions/nudge_tracker.h"
@@ -27,6 +27,9 @@ using std::string;
 
 namespace {
 
+typedef std::map<ModelType, const sync_pb::DataTypeProgressMarker*>
+    TypeProgressMarkerMap;
+
 SyncerError HandleGetEncryptionKeyResponse(
     const sync_pb::ClientToServerResponse& update_response,
     syncable::Directory* dir) {
@@ -135,6 +138,76 @@ void InitDownloadUpdatesRequest(
   }
 }
 
+// Builds a map of ModelTypes to pointers to progress markers in the given
+// update response.  The map is returned in the |marker_map| parameter.
+void PartitionProgressMarkersByType(
+    const sync_pb::GetUpdatesResponse& gu_response,
+    ModelTypeSet request_types,
+    TypeProgressMarkerMap* marker_map) {
+  for (int i = 0; i < gu_response.new_progress_marker_size(); ++i) {
+    int field_number = gu_response.new_progress_marker(i).data_type_id();
+    ModelType model_type = GetModelTypeFromSpecificsFieldNumber(field_number);
+    if (!IsRealDataType(model_type)) {
+      DLOG(WARNING) << "Unknown field number " << field_number;

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

do NOTREACHED instead? (here and below)

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

If we ever get here, I expect the program will end up crashing on the
NOTREACHED() at line 181.  Allowing this loop to continue might result in
slightly more informative messages.  For example, I'd like to know if there was
more than one unknown field number present.

Normally I wouldn't resort to this level of cleverness just for nicer debug
messages, but the NOTREACEHD at 181 had to be there anyway, so we'd might as
well take advantage of it.

+      continue;
+    }
+    if (!request_types.Has(model_type)) {
+      DLOG(WARNING)
+          << "Skipping unexpected progress marker for non-enabled type "
+          << ModelTypeToString(model_type);
+      continue;
+    }
+    marker_map->insert(
+        std::make_pair(model_type, &gu_response.new_progress_marker(i)));

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

It seems risky to take the pointer of a possibly temp variable with unknown
scope. The progress markers aren't themselves particularly large right? Why not
just pass by value? Or alternatively store the index within the gu_response

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

The progress markers are moderately large.  I checked on my regular browser and
found that they range in size from 177 to 349 bytes.  There ~20 of them.  The
cost of copying isn't huge, but it would be nice to avoid if possible.

I wanted to avoid giving the update processing functions direct access to the
protobuf message.  The less functions that have access to it, the better. 
Changing that protobuf is hard enough as is.

I guess I could use indices.

+  }
+}
+
+// Examines the contents of the GetUpdates response message and forwards
+// relevant data to the UpdateHandlers for processing and persisting.
+bool ProcessUpdateResponseMessage(
+    const sync_pb::GetUpdatesResponse& gu_response,
+    ModelTypeSet proto_request_types,
+    UpdateHandlerMap* handler_map,
+    StatusController* status) {
+  TypeSyncEntityMap updates_by_type;
+  PartitionUpdatesByType(gu_response, proto_request_types, &updates_by_type);
+  DCHECK_EQ(proto_request_types.Size(), updates_by_type.size());
+
+  TypeProgressMarkerMap progress_by_type;
+  PartitionProgressMarkersByType(gu_response,
+                                 proto_request_types,
+                                 &progress_by_type);
+  if (proto_request_types.Size() != progress_by_type.size()) {
+    NOTREACHED() << "Missing progress markers in GetUpdates response.";
+    return false;
+  }
+
+  // Iterate over these maps in parallel, processing updates for each type.
+  TypeProgressMarkerMap::iterator pm_it = progress_by_type.begin();

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

naming nit: pm_it, up_it, and uh_it are all fairly ambiguous names, perhaps
progress_marker_iter, update_iter, and update_handler_iter?

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

Done.

+  TypeSyncEntityMap::iterator up_it = updates_by_type.begin();
+  for ( ; pm_it != progress_by_type.end() && up_it != updates_by_type.end();
+       ++pm_it, ++up_it) {
+    DCHECK_EQ(pm_it->first, up_it->first);
+    ModelType type = pm_it->first;
+
+    UpdateHandlerMap::iterator uh_it = handler_map->find(type);
+
+    if (uh_it != handler_map->end()) {
+      uh_it->second->ProcessGetUpdatesResponse(*pm_it->second,
+                                               up_it->second,
+                                               status);
+    } else {
+      DLOG(WARNING)
+          << "Ignoring received updates of a type we can't handle.  "
+          << "Type is: " << ModelTypeToString(type);
+      continue;
+    }
+  }
+  DCHECK(pm_it == progress_by_type.end() && up_it == updates_by_type.end());
+
+  return true;
+}
+
 }  // namespace
 
 void BuildNormalDownloadUpdates(
@@ -234,6 +307,7 @@ void BuildDownloadUpdatesForPoll(
 }
 
 SyncerError ExecuteDownloadUpdates(
+    ModelTypeSet request_types,
     SyncSession* session,
     sync_pb::ClientToServerMessage* msg) {
   sync_pb::ClientToServerResponse update_response;
@@ -251,30 +325,41 @@ SyncerError ExecuteDownloadUpdates(
   if (result != SYNCER_OK) {
     status->mutable_updates_response()->Clear();
     LOG(ERROR) << "PostClientToServerMessage() failed during GetUpdates";
-  } else {
-    status->mutable_updates_response()->CopyFrom(update_response);
-
-    DVLOG(1) << "GetUpdates "
-             << " returned " << update_response.get_updates().entries_size()
-             << " updates and indicated "
-             << update_response.get_updates().changes_remaining()
-             << " updates left on server.";
-
-    if (need_encryption_key ||
-        update_response.get_updates().encryption_keys_size() > 0) {
-      syncable::Directory* dir = session->context()->directory();
-      status->set_last_get_key_result(
-          HandleGetEncryptionKeyResponse(update_response, dir));
-    }
+    return result;

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

interesting, we would attempt to process updates and store timestamps even in
cases where the result was not SYNCER_OK before? Did that accomplish anything?

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

AFAIK, no.  This behavior was in the state machine (I think) because we needed
to make it all the way to the APPLY_UPDATES before we exited the loop.  I tried
to keep it intact when I refactored the syncer, even though I was pretty sure it
had no purpose.

We clear the StatusController's updates response two lines above, so there would
be no progress markers to save and no updates to process when those
SyncerCommands were invoked.

   }
 
-  ProcessUpdatesCommand process_updates;
-  process_updates.Execute(session);
+  status->mutable_updates_response()->CopyFrom(update_response);
 
-  StoreTimestampsCommand store_timestamps;
-  store_timestamps.Execute(session);
+  DVLOG(1) << "GetUpdates "
+           << " returned " << update_response.get_updates().entries_size()
+           << " updates and indicated "
+           << update_response.get_updates().changes_remaining()
+           << " updates left on server.";
 
-  return result;
+  if (need_encryption_key ||
+      update_response.get_updates().encryption_keys_size() > 0) {
+    syncable::Directory* dir = session->context()->directory();
+    status->set_last_get_key_result(
+        HandleGetEncryptionKeyResponse(update_response, dir));
+  }
+
+  const sync_pb::GetUpdatesResponse& gu_response =
+      update_response.get_updates();
+  status->increment_num_updates_downloaded_by(gu_response.entries_size());
+  DCHECK(gu_response.has_changes_remaining());
+  status->set_num_server_changes_remaining(gu_response.changes_remaining());
+
+  const ModelTypeSet proto_request_types =
+      Intersection(request_types, ProtocolTypes());
+
+  if (!ProcessUpdateResponseMessage(gu_response,
+                                    proto_request_types,
+                                    session->context()->update_handler_map(),
+                                    status)) {
+    return SERVER_RESPONSE_VALIDATION_FAILED;
+  } else {
+    return result;
+  }
 }
 
 }  // namespace syncer