sync: Introduce ModelTypeRegistry and helpers

Introduce the ModelTypeRegistry class and use it to manage the creation
of UpdateHandlers and CommitContributors.  The ModelTypeRegistry also
gets some help from the newly introduced UpdaterList and CommitterList
classes.

This lets us move the verbose iteration logic out of the code that's 
focused on building and executing commits and updates, which should make 
those functions easier to read. It gives us more freedom to experiment 
with other ways to manage the lists of commit contributors and update 
handlers, should we choose to do so. It prevents us from leaking the 
set of enabled types through the per-type maps. 

This patch is one of the last in the stack related to building a 
per-type abstraction into the sync engine, and also one of the first 
steps towards implementing run-time enable and disable logic for the 
new-style sync types. 

BUG=278484

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

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

19 files changed, 487 insertions, 296 deletions

diff --git a/sync/engine/commit.cc b/sync/engine/commit.cc
index a8db1a4..0a37656 100644
--- a/sync/engine/commit.cc
+++ b/sync/engine/commit.cc
@@ -5,6 +5,7 @@
 #include "sync/engine/commit.h"
 
 #include "base/debug/trace_event.h"
+#include "sync/engine/commit_processor.h"
 #include "sync/engine/commit_util.h"
 #include "sync/engine/sync_directory_commit_contribution.h"
 #include "sync/engine/syncer.h"
@@ -30,34 +31,18 @@ Commit::~Commit() {
 
 Commit* Commit::Init(
     ModelTypeSet requested_types,
+    ModelTypeSet enabled_types,
     size_t max_entries,
     const std::string& account_name,
     const std::string& cache_guid,
-    CommitContributorMap* contributor_map,
+    CommitProcessor* commit_processor,
     ExtensionsActivity* extensions_activity) {
   // Gather per-type contributions.
   ContributionMap contributions;
-  size_t num_entries = 0;
-  for (ModelTypeSet::Iterator it = requested_types.First();
-       it.Good(); it.Inc()) {
-    CommitContributorMap::iterator cm_it = contributor_map->find(it.Get());
-    if (cm_it == contributor_map->end()) {
-      NOTREACHED()
-          << "Could not find requested type " << ModelTypeToString(it.Get())
-          << " in contributor map.";
-      continue;
-    }
-    size_t spaces_remaining = max_entries - num_entries;
-    SyncDirectoryCommitContribution* contribution =
-        cm_it->second->GetContribution(spaces_remaining);
-    if (contribution) {
-      num_entries += contribution->GetNumEntries();
-      contributions.insert(std::make_pair(it.Get(), contribution));
-    }
-    if (num_entries == max_entries) {
-      break;  // No point in continuting to iterate in this case.
-    }
-  }
+  commit_processor->GatherCommitContributions(
+      requested_types,
+      max_entries,
+      &contributions);
 
   // Give up if no one had anything to commit.
   if (contributions.empty())
@@ -81,13 +66,9 @@ Commit* Commit::Init(
   }
 
   // Set the client config params.
-  ModelTypeSet enabled_types;
-  for (CommitContributorMap::iterator it = contributor_map->begin();
-       it != contributor_map->end(); ++it) {
-    enabled_types.Put(it->first);
-  }
-  commit_util::AddClientConfigParamsToMessage(enabled_types,
-                                                    commit_message);
+  commit_util::AddClientConfigParamsToMessage(
+      enabled_types,
+      commit_message);
 
   // Finally, serialize all our contributions.
   for (std::map<ModelType, SyncDirectoryCommitContribution*>::iterator it =
diff --git a/sync/engine/commit.h b/sync/engine/commit.h
index 4750971..236b514 100644
--- a/sync/engine/commit.h
+++ b/sync/engine/commit.h
@@ -10,7 +10,6 @@
 #include "base/stl_util.h"
 #include "sync/base/sync_export.h"
 #include "sync/engine/sync_directory_commit_contribution.h"
-#include "sync/engine/sync_directory_commit_contributor.h"
 #include "sync/internal_api/public/base/model_type.h"
 #include "sync/internal_api/public/engine/model_safe_worker.h"
 #include "sync/internal_api/public/util/syncer_error.h"
@@ -24,6 +23,7 @@ class StatusController;
 class SyncSession;
 }
 
+class CommitProcessor;
 class Syncer;
 
 // This class wraps the actions related to building and executing a single
@@ -47,10 +47,11 @@ class SYNC_EXPORT_PRIVATE Commit {
 
   static Commit* Init(
       ModelTypeSet requested_types,
+      ModelTypeSet enabled_types,
       size_t max_entries,
       const std::string& account_name,
       const std::string& cache_guid,
-      CommitContributorMap* contributor_map,
+      CommitProcessor* commit_processor,
       ExtensionsActivity* extensions_activity);
 
   SyncerError PostAndProcessResponse(
diff --git a/sync/engine/commit_processor.cc b/sync/engine/commit_processor.cc
new file mode 100644
index 0000000..c54217b
--- /dev/null
+++ b/sync/engine/commit_processor.cc
@@ -0,0 +1,52 @@
+// Copyright 2014 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/commit_processor.h"
+
+#include <map>
+
+#include "sync/engine/sync_directory_commit_contribution.h"
+#include "sync/engine/sync_directory_commit_contributor.h"
+#include "sync/protocol/sync.pb.h"
+
+namespace syncer {
+
+typedef std::map<ModelType, size_t> TypeToIndexMap;
+
+CommitProcessor::CommitProcessor(CommitContributorMap* commit_contributor_map)
+    : commit_contributor_map_(commit_contributor_map) {}
+
+CommitProcessor::~CommitProcessor() {}
+
+void CommitProcessor::GatherCommitContributions(
+    ModelTypeSet commit_types,
+    size_t max_entries,
+    ContributionMap* contributions) {
+  size_t num_entries = 0;
+  for (ModelTypeSet::Iterator it = commit_types.First();
+       it.Good(); it.Inc()) {
+    CommitContributorMap::iterator cm_it =
+        commit_contributor_map_->find(it.Get());
+    if (cm_it == commit_contributor_map_->end()) {
+      NOTREACHED()
+          << "Could not find requested type " << ModelTypeToString(it.Get())
+          << " in contributor map.";
+      continue;
+    }
+    size_t spaces_remaining = max_entries - num_entries;
+    SyncDirectoryCommitContribution* contribution =
+        cm_it->second->GetContribution(spaces_remaining);
+    if (contribution) {
+      num_entries += contribution->GetNumEntries();
+      contributions->insert(std::make_pair(it.Get(), contribution));
+    }
+    if (num_entries >= max_entries) {
+      DCHECK_EQ(num_entries, max_entries)
+          << "Number of commit entries exceeeds maximum";
+      break;
+    }
+  }
+}
+
+}  // namespace syncer
diff --git a/sync/engine/commit_processor.h b/sync/engine/commit_processor.h
new file mode 100644
index 0000000..9ab0957
--- /dev/null
+++ b/sync/engine/commit_processor.h
@@ -0,0 +1,54 @@
+// Copyright 2014 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.
+
+#ifndef SYNC_ENGINE_COMMIT_PROCESSOR_H_
+#define SYNC_ENGINE_COMMIT_PROCESSOR_H_
+
+#include <map>
+#include <vector>
+
+#include "base/basictypes.h"
+#include "sync/base/sync_export.h"
+#include "sync/internal_api/public/base/model_type.h"
+#include "sync/internal_api/public/engine/model_safe_worker.h"
+#include "sync/sessions/model_type_registry.h"
+
+namespace syncer {
+
+namespace syncable {
+class Directory;
+}  // namespace syncable
+
+class SyncDirectoryCommitContributor;
+class SyncDirectoryCommitContribution;
+
+// This class manages the set of per-type committer objects.
+//
+// It owns these types and hides the details of iterating over all of them.
+// Many methods allow the caller to specify a subset of types on which the
+// operation is to be applied.  It is a logic error if the supplied set of types
+// contains a type which was not previously registered.
+class SYNC_EXPORT_PRIVATE CommitProcessor {
+ public:
+  typedef std::map<ModelType, SyncDirectoryCommitContribution*> ContributionMap;
+
+  explicit CommitProcessor(CommitContributorMap* commit_contributor_map);
+  ~CommitProcessor();
+
+  // Gathers a set of contributions to be used to populate a commit message.
+  void GatherCommitContributions(
+      ModelTypeSet commit_types,
+      size_t max_entries,
+      ContributionMap* contributions);
+
+ private:
+  // A map of 'commit contributors', one for each enabled type.
+  CommitContributorMap* commit_contributor_map_;
+
+  DISALLOW_COPY_AND_ASSIGN(CommitProcessor);
+};
+
+}  // namespace syncer
+
+#endif  // SYNC_ENGINE_COMMIT_PROCESSOR_H_
diff --git a/sync/engine/download.cc b/sync/engine/download.cc
index 448481f..3f32cd1 100644
--- a/sync/engine/download.cc
+++ b/sync/engine/download.cc
@@ -7,8 +7,6 @@
 #include <string>
 
 #include "base/command_line.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"
@@ -104,97 +102,11 @@ void InitDownloadUpdatesContext(
       session->context()->notifications_enabled());
 }
 
-void InitDownloadUpdatesProgress(
-    ModelTypeSet proto_request_types,
-    UpdateHandlerMap* handler_map,
-    sync_pb::GetUpdatesMessage* get_updates) {
-  for (ModelTypeSet::Iterator it = proto_request_types.First();
-       it.Good(); it.Inc()) {
-    UpdateHandlerMap::iterator handler_it = handler_map->find(it.Get());
-    DCHECK(handler_it != handler_map->end());
-    sync_pb::DataTypeProgressMarker* progress_marker =
-        get_updates->add_from_progress_marker();
-    handler_it->second->GetDownloadProgress(progress_marker);
-  }
-}
-
-// Builds a map of ModelTypes to indices to progress markers in the given
-// |gu_response| message.  The map is returned in the |index_map| parameter.
-void PartitionProgressMarkersByType(
-    const sync_pb::GetUpdatesResponse& gu_response,
-    ModelTypeSet request_types,
-    TypeToIndexMap* index_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;
-      continue;
-    }
-    if (!request_types.Has(model_type)) {
-      DLOG(WARNING)
-          << "Skipping unexpected progress marker for non-enabled type "
-          << ModelTypeToString(model_type);
-      continue;
-    }
-    index_map->insert(std::make_pair(model_type, i));
-  }
-}
-
-// Examines the contents of the GetUpdates response message and forwards
-// relevant data to the UpdateHandlers for processing and persisting.
-bool ProcessUpdateResponseContents(
-    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());
-
-  TypeToIndexMap progress_index_by_type;
-  PartitionProgressMarkersByType(gu_response,
-                                 proto_request_types,
-                                 &progress_index_by_type);
-  if (proto_request_types.Size() != progress_index_by_type.size()) {
-    NOTREACHED() << "Missing progress markers in GetUpdates response.";
-    return false;
-  }
-
-  // Iterate over these maps in parallel, processing updates for each type.
-  TypeToIndexMap::iterator progress_marker_iter =
-      progress_index_by_type.begin();
-  TypeSyncEntityMap::iterator updates_iter = updates_by_type.begin();
-  for ( ; (progress_marker_iter != progress_index_by_type.end()
-           && updates_iter != updates_by_type.end());
-       ++progress_marker_iter, ++updates_iter) {
-    DCHECK_EQ(progress_marker_iter->first, updates_iter->first);
-    ModelType type = progress_marker_iter->first;
-
-    UpdateHandlerMap::iterator update_handler_iter = handler_map->find(type);
-
-    if (update_handler_iter != handler_map->end()) {
-      update_handler_iter->second->ProcessGetUpdatesResponse(
-          gu_response.new_progress_marker(progress_marker_iter->second),
-          updates_iter->second,
-          status);
-    } else {
-      DLOG(WARNING)
-          << "Ignoring received updates of a type we can't handle.  "
-          << "Type is: " << ModelTypeToString(type);
-      continue;
-    }
-  }
-  DCHECK(progress_marker_iter == progress_index_by_type.end()
-         && updates_iter == updates_by_type.end());
-
-  return true;
-}
-
 }  // namespace
 
 void BuildNormalDownloadUpdates(
     SyncSession* session,
+    GetUpdatesProcessor* get_updates_processor,
     bool create_mobile_bookmarks_folder,
     ModelTypeSet request_types,
     const sessions::NudgeTracker& nudge_tracker,
@@ -211,22 +123,20 @@ void BuildNormalDownloadUpdates(
 
   BuildNormalDownloadUpdatesImpl(
       Intersection(request_types, ProtocolTypes()),
-      session->context()->update_handler_map(),
+      get_updates_processor,
       nudge_tracker,
       client_to_server_message->mutable_get_updates());
 }
 
 void BuildNormalDownloadUpdatesImpl(
     ModelTypeSet proto_request_types,
-    UpdateHandlerMap* update_handler_map,
+    GetUpdatesProcessor* get_updates_processor,
     const sessions::NudgeTracker& nudge_tracker,
     sync_pb::GetUpdatesMessage* get_updates) {
   DCHECK(!proto_request_types.Empty());
 
-  InitDownloadUpdatesProgress(
-      proto_request_types,
-      update_handler_map,
-      get_updates);
+  // Get progress markers and other data for requested types.
+  get_updates_processor->PrepareGetUpdates(proto_request_types, get_updates);
 
   // Set legacy GetUpdatesMessage.GetUpdatesCallerInfo information.
   get_updates->mutable_caller_info()->set_source(
@@ -256,6 +166,7 @@ void BuildNormalDownloadUpdatesImpl(
 
 void BuildDownloadUpdatesForConfigure(
     SyncSession* session,
+    GetUpdatesProcessor* get_updates_processor,
     bool create_mobile_bookmarks_folder,
     sync_pb::GetUpdatesCallerInfo::GetUpdatesSource source,
     ModelTypeSet request_types,
@@ -270,22 +181,20 @@ void BuildDownloadUpdatesForConfigure(
       client_to_server_message);
   BuildDownloadUpdatesForConfigureImpl(
       Intersection(request_types, ProtocolTypes()),
-      session->context()->update_handler_map(),
+      get_updates_processor,
       source,
       client_to_server_message->mutable_get_updates());
 }
 
 void BuildDownloadUpdatesForConfigureImpl(
     ModelTypeSet proto_request_types,
-    UpdateHandlerMap* update_handler_map,
+    GetUpdatesProcessor* get_updates_processor,
     sync_pb::GetUpdatesCallerInfo::GetUpdatesSource source,
     sync_pb::GetUpdatesMessage* get_updates) {
   DCHECK(!proto_request_types.Empty());
 
-  InitDownloadUpdatesProgress(
-      proto_request_types,
-      update_handler_map,
-      get_updates);
+  // Get progress markers and other data for requested types.
+  get_updates_processor->PrepareGetUpdates(proto_request_types, get_updates);
 
   // Set legacy GetUpdatesMessage.GetUpdatesCallerInfo information.
   get_updates->mutable_caller_info()->set_source(source);
@@ -298,6 +207,7 @@ void BuildDownloadUpdatesForConfigureImpl(
 
 void BuildDownloadUpdatesForPoll(
     SyncSession* session,
+    GetUpdatesProcessor* get_updates_processor,
     bool create_mobile_bookmarks_folder,
     ModelTypeSet request_types,
     sync_pb::ClientToServerMessage* client_to_server_message) {
@@ -310,20 +220,18 @@ void BuildDownloadUpdatesForPoll(
       client_to_server_message);
   BuildDownloadUpdatesForPollImpl(
       Intersection(request_types, ProtocolTypes()),
-      session->context()->update_handler_map(),
+      get_updates_processor,
       client_to_server_message->mutable_get_updates());
 }
 
 void BuildDownloadUpdatesForPollImpl(
     ModelTypeSet proto_request_types,
-    UpdateHandlerMap* update_handler_map,
+    GetUpdatesProcessor* get_updates_processor,
     sync_pb::GetUpdatesMessage* get_updates) {
   DCHECK(!proto_request_types.Empty());
 
-  InitDownloadUpdatesProgress(
-      proto_request_types,
-      update_handler_map,
-      get_updates);
+  // Get progress markers and other data for requested types.
+  get_updates_processor->PrepareGetUpdates(proto_request_types, get_updates);
 
   // Set legacy GetUpdatesMessage.GetUpdatesCallerInfo information.
   get_updates->mutable_caller_info()->set_source(
@@ -335,6 +243,7 @@ void BuildDownloadUpdatesForPollImpl(
 
 void BuildDownloadUpdatesForRetry(
     SyncSession* session,
+    GetUpdatesProcessor* get_updates_processor,
     bool create_mobile_bookmarks_folder,
     ModelTypeSet request_types,
     sync_pb::ClientToServerMessage* client_to_server_message) {
@@ -347,20 +256,17 @@ void BuildDownloadUpdatesForRetry(
       client_to_server_message);
   BuildDownloadUpdatesForRetryImpl(
       Intersection(request_types, ProtocolTypes()),
-      session->context()->update_handler_map(),
+      get_updates_processor,
       client_to_server_message->mutable_get_updates());
 }
 
 void BuildDownloadUpdatesForRetryImpl(
     ModelTypeSet proto_request_types,
-    UpdateHandlerMap* update_handler_map,
+    GetUpdatesProcessor* get_updates_processor,
     sync_pb::GetUpdatesMessage* get_updates) {
   DCHECK(!proto_request_types.Empty());
 
-  InitDownloadUpdatesProgress(
-      proto_request_types,
-      update_handler_map,
-      get_updates);
+  get_updates_processor->PrepareGetUpdates(proto_request_types, get_updates);
 
   // Set legacy GetUpdatesMessage.GetUpdatesCallerInfo information.
   get_updates->mutable_caller_info()->set_source(
@@ -374,6 +280,7 @@ void BuildDownloadUpdatesForRetryImpl(
 SyncerError ExecuteDownloadUpdates(
     ModelTypeSet request_types,
     SyncSession* session,
+    GetUpdatesProcessor* get_updates_processor,
     sync_pb::ClientToServerMessage* msg) {
   sync_pb::ClientToServerResponse update_response;
   StatusController* status = session->mutable_status_controller();
@@ -421,14 +328,14 @@ SyncerError ExecuteDownloadUpdates(
 
   return ProcessResponse(update_response.get_updates(),
                          proto_request_types,
-                         session->context()->update_handler_map(),
+                         get_updates_processor,
                          status);
 }
 
 SyncerError ProcessResponse(
     const sync_pb::GetUpdatesResponse& gu_response,
     ModelTypeSet proto_request_types,
-    UpdateHandlerMap* handler_map,
+    GetUpdatesProcessor* get_updates_processor,
     StatusController* status) {
   status->increment_num_updates_downloaded_by(gu_response.entries_size());
 
@@ -440,10 +347,9 @@ SyncerError ProcessResponse(
   status->set_num_server_changes_remaining(gu_response.changes_remaining());
 
 
-  if (!ProcessUpdateResponseContents(gu_response,
-                                     proto_request_types,
-                                     handler_map,
-                                     status)) {
+  if (!get_updates_processor->ProcessGetUpdatesResponse(proto_request_types,
+                                                        gu_response,
+                                                        status)) {
     return SERVER_RESPONSE_VALIDATION_FAILED;
   }
 
diff --git a/sync/engine/download.h b/sync/engine/download.h
index e3230d8..22a3a42 100644
--- a/sync/engine/download.h
+++ b/sync/engine/download.h
@@ -6,7 +6,7 @@
 #define SYNC_ENGINE_DOWNLOAD_H_
 
 #include "sync/base/sync_export.h"
-#include "sync/engine/sync_directory_update_handler.h"
+#include "sync/engine/get_updates_processor.h"
 #include "sync/internal_api/public/base/model_type.h"
 #include "sync/internal_api/public/util/syncer_error.h"
 #include "sync/protocol/sync.pb.h"
@@ -31,6 +31,7 @@ namespace download {
 // keep types in sync when in normal mode.
 SYNC_EXPORT_PRIVATE void BuildNormalDownloadUpdates(
     sessions::SyncSession* session,
+    GetUpdatesProcessor* get_updates_processor,
     bool create_mobile_bookmarks_folder,
     ModelTypeSet request_types,
     const sessions::NudgeTracker& nudge_tracker,
@@ -39,7 +40,7 @@ SYNC_EXPORT_PRIVATE void BuildNormalDownloadUpdates(
 // Helper function.  Defined here for testing.
 SYNC_EXPORT_PRIVATE void BuildNormalDownloadUpdatesImpl(
     ModelTypeSet proto_request_types,
-    UpdateHandlerMap* update_handler_map,
+    GetUpdatesProcessor* get_updates_processor,
     const sessions::NudgeTracker& nudge_tracker,
     sync_pb::GetUpdatesMessage* get_updates);
 
@@ -48,6 +49,7 @@ SYNC_EXPORT_PRIVATE void BuildNormalDownloadUpdatesImpl(
 // initialize a type for the first time.
 SYNC_EXPORT_PRIVATE void BuildDownloadUpdatesForConfigure(
     sessions::SyncSession* session,
+    GetUpdatesProcessor* get_updates_processor,
     bool create_mobile_bookmarks_folder,
     sync_pb::GetUpdatesCallerInfo::GetUpdatesSource source,
     ModelTypeSet request_types,
@@ -56,7 +58,7 @@ SYNC_EXPORT_PRIVATE void BuildDownloadUpdatesForConfigure(
 // Helper function.  Defined here for testing.
 SYNC_EXPORT_PRIVATE void BuildDownloadUpdatesForConfigureImpl(
     ModelTypeSet proto_request_types,
-    UpdateHandlerMap* update_handler_map,
+    GetUpdatesProcessor* get_updates_processor,
     sync_pb::GetUpdatesCallerInfo::GetUpdatesSource source,
     sync_pb::GetUpdatesMessage* get_updates);
 
@@ -65,6 +67,7 @@ SYNC_EXPORT_PRIVATE void BuildDownloadUpdatesForConfigureImpl(
 // periodic polling.
 SYNC_EXPORT_PRIVATE void BuildDownloadUpdatesForPoll(
     sessions::SyncSession* session,
+    GetUpdatesProcessor* get_updates_processor,
     bool create_mobile_bookmarks_folder,
     ModelTypeSet request_types,
     sync_pb::ClientToServerMessage* client_to_server_message);
@@ -72,13 +75,14 @@ SYNC_EXPORT_PRIVATE void BuildDownloadUpdatesForPoll(
 // Helper function.  Defined here for testing.
 SYNC_EXPORT_PRIVATE void BuildDownloadUpdatesForPollImpl(
     ModelTypeSet proto_request_types,
-    UpdateHandlerMap* update_handler_map,
+    GetUpdatesProcessor* get_updates_processor,
     sync_pb::GetUpdatesMessage* get_updates);
 
 // Same as BuildDownloadUpdatesForPoll() except the update origin/source is
 // RETRY.
 SYNC_EXPORT_PRIVATE void BuildDownloadUpdatesForRetry(
     sessions::SyncSession* session,
+    GetUpdatesProcessor* get_updates_processor,
     bool create_mobile_bookmarks_folder,
     ModelTypeSet request_types,
     sync_pb::ClientToServerMessage* client_to_server_message);
@@ -87,7 +91,7 @@ SYNC_EXPORT_PRIVATE void BuildDownloadUpdatesForRetry(
 // RETRY.
 SYNC_EXPORT_PRIVATE void BuildDownloadUpdatesForRetryImpl(
     ModelTypeSet proto_request_types,
-    UpdateHandlerMap* update_handler_map,
+    GetUpdatesProcessor* get_updates_processor,
     sync_pb::GetUpdatesMessage* get_updates);
 
 // Sends the specified message to the server and stores the response in a member
@@ -95,6 +99,7 @@ SYNC_EXPORT_PRIVATE void BuildDownloadUpdatesForRetryImpl(
 SYNC_EXPORT_PRIVATE SyncerError
     ExecuteDownloadUpdates(ModelTypeSet request_types,
                            sessions::SyncSession* session,
+                           GetUpdatesProcessor* get_updates_processor,
                            sync_pb::ClientToServerMessage* msg);
 
 // Helper function for processing responses from the server.
@@ -102,7 +107,7 @@ SYNC_EXPORT_PRIVATE SyncerError
 SYNC_EXPORT_PRIVATE SyncerError ProcessResponse(
     const sync_pb::GetUpdatesResponse& gu_response,
     ModelTypeSet proto_request_types,
-    UpdateHandlerMap* handler_map,
+    GetUpdatesProcessor* get_updates_processor,
     sessions::StatusController* status);
 
 // Helper function to copy client debug info from debug_info_getter to
diff --git a/sync/engine/download_unittest.cc b/sync/engine/download_unittest.cc
index 134f441..14c9305 100644
--- a/sync/engine/download_unittest.cc
+++ b/sync/engine/download_unittest.cc
@@ -25,9 +25,7 @@ using sessions::MockDebugInfoGetter;
 // A test fixture for tests exercising download updates functions.
 class DownloadUpdatesTest : public ::testing::Test {
  protected:
-  DownloadUpdatesTest()
-      : update_handler_map_deleter_(&update_handler_map_) {
-  }
+  DownloadUpdatesTest() : update_handler_deleter_(&update_handler_map_) {}
 
   virtual void SetUp() {
     dir_maker_.SetUp();
@@ -42,20 +40,15 @@ class DownloadUpdatesTest : public ::testing::Test {
   }
 
   ModelTypeSet proto_request_types() {
-    ModelTypeSet types;
-    for (UpdateHandlerMap::iterator it = update_handler_map_.begin();
-         it != update_handler_map_.end(); ++it) {
-      types.Put(it->first);
-    }
-    return types;
+    return proto_request_types_;
   }
 
   syncable::Directory* directory() {
     return dir_maker_.directory();
   }
 
-  UpdateHandlerMap* update_handler_map() {
-    return &update_handler_map_;
+  GetUpdatesProcessor* get_updates_processor() {
+    return get_updates_processor_.get();
   }
 
   void InitFakeUpdateResponse(sync_pb::GetUpdatesResponse* response) {
@@ -74,17 +67,23 @@ class DownloadUpdatesTest : public ::testing::Test {
  private:
   void AddUpdateHandler(ModelType type, ModelSafeGroup group) {
     DCHECK(directory());
+
+    proto_request_types_.Put(type);
+
     scoped_refptr<ModelSafeWorker> worker = new FakeModelWorker(group);
     SyncDirectoryUpdateHandler* handler =
         new SyncDirectoryUpdateHandler(directory(), type, worker);
     update_handler_map_.insert(std::make_pair(type, handler));
+    get_updates_processor_.reset(new GetUpdatesProcessor(&update_handler_map_));
   }
 
   base::MessageLoop loop_;  // Needed for directory init.
   TestDirectorySetterUpper dir_maker_;
 
+  ModelTypeSet proto_request_types_;
   UpdateHandlerMap update_handler_map_;
-  STLValueDeleter<UpdateHandlerMap> update_handler_map_deleter_;
+  STLValueDeleter<UpdateHandlerMap> update_handler_deleter_;
+  scoped_ptr<GetUpdatesProcessor> get_updates_processor_;
 
   DISALLOW_COPY_AND_ASSIGN(DownloadUpdatesTest);
 };
@@ -96,7 +95,7 @@ TEST_F(DownloadUpdatesTest, BookmarkNudge) {
 
   sync_pb::ClientToServerMessage msg;
   download::BuildNormalDownloadUpdatesImpl(proto_request_types(),
-                                           update_handler_map(),
+                                           get_updates_processor(),
                                            nudge_tracker,
                                            msg.mutable_get_updates());
 
@@ -144,7 +143,7 @@ TEST_F(DownloadUpdatesTest, NotifyMany) {
 
   sync_pb::ClientToServerMessage msg;
   download::BuildNormalDownloadUpdatesImpl(proto_request_types(),
-                                           update_handler_map(),
+                                           get_updates_processor(),
                                            nudge_tracker,
                                            msg.mutable_get_updates());
 
@@ -178,7 +177,7 @@ TEST_F(DownloadUpdatesTest, ConfigureTest) {
   sync_pb::ClientToServerMessage msg;
   download::BuildDownloadUpdatesForConfigureImpl(
       proto_request_types(),
-      update_handler_map(),
+      get_updates_processor(),
       sync_pb::GetUpdatesCallerInfo::RECONFIGURATION,
       msg.mutable_get_updates());
 
@@ -201,7 +200,7 @@ TEST_F(DownloadUpdatesTest, PollTest) {
   sync_pb::ClientToServerMessage msg;
   download::BuildDownloadUpdatesForPollImpl(
       proto_request_types(),
-      update_handler_map(),
+      get_updates_processor(),
       msg.mutable_get_updates());
 
   const sync_pb::GetUpdatesMessage& gu_msg = msg.get_updates();
@@ -223,7 +222,7 @@ TEST_F(DownloadUpdatesTest, RetryTest) {
   sync_pb::ClientToServerMessage msg;
   download::BuildDownloadUpdatesForRetryImpl(
       proto_request_types(),
-      update_handler_map(),
+      get_updates_processor(),
       msg.mutable_get_updates());
 
   const sync_pb::GetUpdatesMessage& gu_msg = msg.get_updates();
@@ -250,7 +249,7 @@ TEST_F(DownloadUpdatesTest, NudgeWithRetryTest) {
 
   sync_pb::ClientToServerMessage msg;
   download::BuildNormalDownloadUpdatesImpl(proto_request_types(),
-                                           update_handler_map(),
+                                           get_updates_processor(),
                                            nudge_tracker,
                                            msg.mutable_get_updates());
   EXPECT_TRUE(msg.get_updates().is_retry());
@@ -268,7 +267,7 @@ TEST_F(DownloadUpdatesTest, InvalidResponse) {
   sessions::StatusController status;
   SyncerError error = download::ProcessResponse(gu_response,
                                                 proto_request_types(),
-                                                update_handler_map(),
+                                                get_updates_processor(),
                                                 &status);
   EXPECT_EQ(error, SERVER_RESPONSE_VALIDATION_FAILED);
 }
@@ -282,7 +281,7 @@ TEST_F(DownloadUpdatesTest, MoreToDownloadResponse) {
   sessions::StatusController status;
   SyncerError error = download::ProcessResponse(gu_response,
                                                 proto_request_types(),
-                                                update_handler_map(),
+                                                get_updates_processor(),
                                                 &status);
   EXPECT_EQ(error, SERVER_MORE_TO_DOWNLOAD);
 }
@@ -296,7 +295,7 @@ TEST_F(DownloadUpdatesTest, NormalResponseTest) {
   sessions::StatusController status;
   SyncerError error = download::ProcessResponse(gu_response,
                                                 proto_request_types(),
-                                                update_handler_map(),
+                                                get_updates_processor(),
                                                 &status);
   EXPECT_EQ(error, SYNCER_OK);
 }
diff --git a/sync/engine/get_updates_processor.cc b/sync/engine/get_updates_processor.cc
new file mode 100644
index 0000000..b85329a
--- /dev/null
+++ b/sync/engine/get_updates_processor.cc
@@ -0,0 +1,148 @@
+// Copyright 2014 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/get_updates_processor.h"
+
+#include <map>
+
+#include "sync/engine/sync_directory_update_handler.h"
+#include "sync/protocol/sync.pb.h"
+
+namespace syncer {
+
+typedef std::map<ModelType, size_t> TypeToIndexMap;
+
+namespace {
+
+// Given a GetUpdates response, iterates over all the returned items and
+// divides them according to their type.  Outputs a map from model types to
+// received SyncEntities.  The output map will have entries (possibly empty)
+// for all types in |requested_types|.
+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);
+    if (!IsRealDataType(type)) {
+      NOTREACHED() << "Received update with invalid type.";
+      continue;
+    }
+
+    TypeSyncEntityMap::iterator it = updates_by_type->find(type);
+    if (it == updates_by_type->end()) {
+      NOTREACHED() << "Received update for unexpected type "
+                   << ModelTypeToString(type);
+      continue;
+    }
+
+    it->second.push_back(&update);
+  }
+}
+
+// Builds a map of ModelTypes to indices to progress markers in the given
+// |gu_response| message.  The map is returned in the |index_map| parameter.
+void PartitionProgressMarkersByType(
+    const sync_pb::GetUpdatesResponse& gu_response,
+    ModelTypeSet request_types,
+    TypeToIndexMap* index_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;
+      continue;
+    }
+    if (!request_types.Has(model_type)) {
+      DLOG(WARNING)
+          << "Skipping unexpected progress marker for non-enabled type "
+          << ModelTypeToString(model_type);
+      continue;
+    }
+    index_map->insert(std::make_pair(model_type, i));
+  }
+}
+
+}  // namespace
+
+GetUpdatesProcessor::GetUpdatesProcessor(UpdateHandlerMap* update_handler_map)
+    : update_handler_map_(update_handler_map) {}
+
+GetUpdatesProcessor::~GetUpdatesProcessor() {}
+
+void GetUpdatesProcessor::PrepareGetUpdates(
+    ModelTypeSet gu_types,
+    sync_pb::GetUpdatesMessage* get_updates) {
+  for (ModelTypeSet::Iterator it = gu_types.First(); it.Good(); it.Inc()) {
+    UpdateHandlerMap::iterator handler_it = update_handler_map_->find(it.Get());
+    DCHECK(handler_it != update_handler_map_->end());
+    sync_pb::DataTypeProgressMarker* progress_marker =
+        get_updates->add_from_progress_marker();
+    handler_it->second->GetDownloadProgress(progress_marker);
+  }
+}
+
+bool GetUpdatesProcessor::ProcessGetUpdatesResponse(
+    ModelTypeSet gu_types,
+    const sync_pb::GetUpdatesResponse& gu_response,
+    sessions::StatusController* status_controller) {
+  TypeSyncEntityMap updates_by_type;
+  PartitionUpdatesByType(gu_response, gu_types, &updates_by_type);
+  DCHECK_EQ(gu_types.Size(), updates_by_type.size());
+
+  TypeToIndexMap progress_index_by_type;
+  PartitionProgressMarkersByType(gu_response,
+                                 gu_types,
+                                 &progress_index_by_type);
+  if (gu_types.Size() != progress_index_by_type.size()) {
+    NOTREACHED() << "Missing progress markers in GetUpdates response.";
+    return false;
+  }
+
+  // Iterate over these maps in parallel, processing updates for each type.
+  TypeToIndexMap::iterator progress_marker_iter =
+      progress_index_by_type.begin();
+  TypeSyncEntityMap::iterator updates_iter = updates_by_type.begin();
+  for (; (progress_marker_iter != progress_index_by_type.end()
+           && updates_iter != updates_by_type.end());
+       ++progress_marker_iter, ++updates_iter) {
+    DCHECK_EQ(progress_marker_iter->first, updates_iter->first);
+    ModelType type = progress_marker_iter->first;
+
+    UpdateHandlerMap::iterator update_handler_iter =
+        update_handler_map_->find(type);
+
+    if (update_handler_iter != update_handler_map_->end()) {
+      update_handler_iter->second->ProcessGetUpdatesResponse(
+          gu_response.new_progress_marker(progress_marker_iter->second),
+          updates_iter->second,
+          status_controller);
+    } else {
+      DLOG(WARNING)
+          << "Ignoring received updates of a type we can't handle.  "
+          << "Type is: " << ModelTypeToString(type);
+      continue;
+    }
+  }
+  DCHECK(progress_marker_iter == progress_index_by_type.end() &&
+         updates_iter == updates_by_type.end());
+
+  return true;
+}
+
+void GetUpdatesProcessor::ApplyUpdatesForAllTypes(
+    sessions::StatusController* status_controller) {
+  for (UpdateHandlerMap::iterator it = update_handler_map_->begin();
+       it != update_handler_map_->end(); ++it) {
+    it->second->ApplyUpdates(status_controller);
+  }
+}
+
+}  // namespace syncer
diff --git a/sync/engine/get_updates_processor.h b/sync/engine/get_updates_processor.h
new file mode 100644
index 0000000..82634b2
--- /dev/null
+++ b/sync/engine/get_updates_processor.h
@@ -0,0 +1,72 @@
+// Copyright 2014 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.
+
+#ifndef SYNC_ENGINE_GET_UPDATES_PROCESSOR_H
+#define SYNC_ENGINE_GET_UPDATES_PROCESSOR_H
+
+#include <map>
+#include <vector>
+
+#include "base/basictypes.h"
+#include "sync/base/sync_export.h"
+#include "sync/internal_api/public/base/model_type.h"
+#include "sync/internal_api/public/engine/model_safe_worker.h"
+#include "sync/sessions/model_type_registry.h"
+
+namespace sync_pb {
+class GetUpdatesMessage;
+class GetUpdatesResponse;
+}  // namespace sync_pb
+
+namespace syncer {
+
+namespace sessions {
+class StatusController;
+}  // namespace sessions
+
+namespace syncable {
+class Directory;
+}  // namespace syncable
+
+class SyncDirectoryUpdateHandler;
+
+typedef std::vector<const sync_pb::SyncEntity*> SyncEntityList;
+typedef std::map<ModelType, SyncEntityList> TypeSyncEntityMap;
+
+// This class manages the set of per-type syncer objects.
+//
+// It owns these types and hides the details of iterating over all of them.
+// Most methods allow the caller to specify a subset of types on which the
+// operation is to be applied.  It is a logic error if the supplied set of types
+// contains a type which was not previously registered with the manager.
+class SYNC_EXPORT_PRIVATE GetUpdatesProcessor {
+ public:
+  explicit GetUpdatesProcessor(UpdateHandlerMap* update_handler_map);
+  ~GetUpdatesProcessor();
+
+  // Populates a GetUpdates request message with per-type information.
+  void PrepareGetUpdates(ModelTypeSet gu_types,
+                         sync_pb::GetUpdatesMessage* get_updates);
+
+  // Processes a GetUpdates responses for each type.
+  bool ProcessGetUpdatesResponse(
+      ModelTypeSet gu_types,
+      const sync_pb::GetUpdatesResponse& gu_response,
+      sessions::StatusController* status_controller);
+
+  // Applies pending updates for all sync types known to the manager.
+  void ApplyUpdatesForAllTypes(sessions::StatusController* status_controller);
+
+ private:
+  // A map of 'update handlers', one for each enabled type.
+  // This must be kept in sync with the routing info.  Our temporary solution to
+  // that problem is to initialize this map in set_routing_info().
+  UpdateHandlerMap* update_handler_map_;
+
+  DISALLOW_COPY_AND_ASSIGN(GetUpdatesProcessor);
+};
+
+}  // namespace syncer
+
+#endif  // SYNC_ENGINE_GET_UPDATES_PROCESSOR_H_
diff --git a/sync/engine/process_updates_util.cc b/sync/engine/process_updates_util.cc
index 49e40b3..b2cab4a 100644
--- a/sync/engine/process_updates_util.cc
+++ b/sync/engine/process_updates_util.cc
@@ -6,6 +6,7 @@
 
 #include "base/location.h"
 #include "sync/engine/syncer_proto_util.h"
+#include "sync/engine/syncer_types.h"
 #include "sync/engine/syncer_util.h"
 #include "sync/syncable/directory.h"
 #include "sync/syncable/model_neutral_mutable_entry.h"
@@ -74,58 +75,6 @@ bool UpdateContainsNewVersion(syncable::BaseTransaction *trans,
   return existing_version < update.version();
 }
 
-}  // namespace
-
-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);
-    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);
-      continue;
-    }
-
-    it->second.push_back(&update);
-  }
-}
-
-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
@@ -141,8 +90,10 @@ VerifyResult VerifyTagConsistency(
   return VERIFY_UNDECIDED;
 }
 
-}  // namespace
-
+// Checks whether or not an update is fit for processing.
+//
+// The answer may be "no" if the update appears invalid, or it's not releveant
+// (ie. a delete for an item we've never heard of), or other reasons.
 VerifyResult VerifyUpdate(
     syncable::ModelNeutralWriteTransaction* trans,
     const sync_pb::SyncEntity& entry,
@@ -201,7 +152,6 @@ VerifyResult VerifyUpdate(
   return result;  // This might be VERIFY_SUCCESS as well
 }
 
-namespace {
 // Returns true if the entry is still ok to process.
 bool ReverifyEntry(syncable::ModelNeutralWriteTransaction* trans,
                    const sync_pb::SyncEntity& entry,
@@ -218,9 +168,11 @@ bool ReverifyEntry(syncable::ModelNeutralWriteTransaction* trans,
                                                    model_type,
                                                    same_id);
 }
-}  // namespace
 
 // Process a single update. Will avoid touching global state.
+//
+// If the update passes a series of checks, this function will copy
+// the SyncEntity's data into the SERVER side of the syncable::Directory.
 void ProcessUpdate(
     const sync_pb::SyncEntity& update,
     const Cryptographer* cryptographer,
@@ -326,4 +278,26 @@ void ProcessUpdate(
   return;
 }
 
+}  // namespace
+
+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 syncer
diff --git a/sync/engine/process_updates_util.h b/sync/engine/process_updates_util.h
index 6e8bc71..428ff57 100644
--- a/sync/engine/process_updates_util.h
+++ b/sync/engine/process_updates_util.h
@@ -5,16 +5,11 @@
 #ifndef SYNC_ENGINE_PROCESS_UPDATES_UTIL_H_
 #define SYNC_ENGINE_PROCESS_UPDATES_UTIL_H_
 
-#include <map>
 #include <vector>
 
-#include "base/compiler_specific.h"
-#include "sync/base/sync_export.h"
-#include "sync/engine/syncer_types.h"
 #include "sync/internal_api/public/base/model_type.h"
 
 namespace sync_pb {
-class GetUpdatesResponse;
 class SyncEntity;
 }
 
@@ -29,20 +24,7 @@ class ModelNeutralWriteTransaction;
 class Directory;
 }
 
-class Cryptographer;
-
-// TODO(rlarocque): Move these definitions somewhere else?
 typedef std::vector<const sync_pb::SyncEntity*> SyncEntityList;
-typedef std::map<ModelType, SyncEntityList> TypeSyncEntityMap;
-
-// Given a GetUpdates response, iterates over all the returned items and
-// divides them according to their type.  Outputs a map from model types to
-// received SyncEntities.  The output map will have entries (possibly empty)
-// for all types in |requested_types|.
-void PartitionUpdatesByType(
-    const sync_pb::GetUpdatesResponse& updates,
-    ModelTypeSet requested_types,
-    TypeSyncEntityMap* updates_by_type);
 
 // Processes all the updates associated with a single ModelType.
 void ProcessDownloadedUpdates(
@@ -52,22 +34,6 @@ void ProcessDownloadedUpdates(
     const SyncEntityList& applicable_updates,
     sessions::StatusController* status);
 
-// Checks whether or not an update is fit for processing.
-//
-// The answer may be "no" if the update appears invalid, or it's not releveant
-// (ie. a delete for an item we've never heard of), or other reasons.
-VerifyResult VerifyUpdate(
-    syncable::ModelNeutralWriteTransaction* trans,
-    const sync_pb::SyncEntity& entry,
-    ModelType requested_type);
-
-// If the update passes a series of checks, this function will copy
-// the SyncEntity's data into the SERVER side of the syncable::Directory.
-void ProcessUpdate(
-    const sync_pb::SyncEntity& proto_update,
-    const Cryptographer* cryptographer,
-    syncable::ModelNeutralWriteTransaction* const trans);
-
 }  // namespace syncer
 
 #endif  // SYNC_ENGINE_PROCESS_UPDATES_UTIL_H_
diff --git a/sync/engine/sync_directory_commit_contributor.h b/sync/engine/sync_directory_commit_contributor.h
index 6ffaeb7..4a90075 100644
--- a/sync/engine/sync_directory_commit_contributor.h
+++ b/sync/engine/sync_directory_commit_contributor.h
@@ -36,10 +36,6 @@ class SyncDirectoryCommitContributor {
   ModelType type_;
 };
 
-// TODO(rlarocque): Find a better place for this definition.
-typedef std::map<ModelType, SyncDirectoryCommitContributor*>
-    CommitContributorMap;
-
 }  // namespace
 
 #endif  // SYNC_ENGINE_SYNC_DIRECTORY_COMMIT_CONTRIBUTOR_H_
diff --git a/sync/engine/sync_directory_update_handler.h b/sync/engine/sync_directory_update_handler.h
index ea4d7914..9cfdaed 100644
--- a/sync/engine/sync_directory_update_handler.h
+++ b/sync/engine/sync_directory_update_handler.h
@@ -89,9 +89,6 @@ class SYNC_EXPORT_PRIVATE SyncDirectoryUpdateHandler {
   DISALLOW_COPY_AND_ASSIGN(SyncDirectoryUpdateHandler);
 };
 
-// TODO(rlarocque): Find a better place to define this.
-typedef std::map<ModelType, SyncDirectoryUpdateHandler*> UpdateHandlerMap;
-
 }  // namespace syncer
 
 #endif  // SYNC_ENGINE_SYNC_DIRECTORY_UPDATE_HANDLER_H_
diff --git a/sync/engine/sync_directory_update_handler_unittest.cc b/sync/engine/sync_directory_update_handler_unittest.cc
index 86d447e..3cc8e7d 100644
--- a/sync/engine/sync_directory_update_handler_unittest.cc
+++ b/sync/engine/sync_directory_update_handler_unittest.cc
@@ -291,6 +291,8 @@ class SyncDirectoryUpdateHandlerApplyUpdateTest : public ::testing::Test {
   }
 
  private:
+  typedef std::map<ModelType, SyncDirectoryUpdateHandler*> UpdateHandlerMap;
+
   base::MessageLoop loop_;  // Needed to initialize the directory.
   TestDirectorySetterUpper dir_maker_;
   scoped_ptr<TestEntryFactory> entry_factory_;
diff --git a/sync/engine/sync_scheduler_impl.cc b/sync/engine/sync_scheduler_impl.cc
index 6c784bc..ac169ec 100644
--- a/sync/engine/sync_scheduler_impl.cc
+++ b/sync/engine/sync_scheduler_impl.cc
@@ -294,7 +294,7 @@ void SyncSchedulerImpl::ScheduleConfiguration(
   BuildModelSafeParams(params.types_to_download,
                        params.routing_info,
                        &restricted_routes);
-  session_context_->set_routing_info(restricted_routes);
+  session_context_->SetRoutingInfo(restricted_routes);
 
   // Only reconfigure if we have types to download.
   if (!params.types_to_download.Empty()) {
diff --git a/sync/engine/sync_scheduler_unittest.cc b/sync/engine/sync_scheduler_unittest.cc
index 62aa2c9..7b0a0d3 100644
--- a/sync/engine/sync_scheduler_unittest.cc
+++ b/sync/engine/sync_scheduler_unittest.cc
@@ -130,14 +130,18 @@ class SyncSchedulerTest : public testing::Test {
     connection_.reset(new MockConnectionManager(directory(),
                                                 &cancelation_signal_));
     connection_->SetServerReachable();
+
+    model_type_registry_.reset(new ModelTypeRegistry(workers_, directory()));
+
     context_.reset(new SyncSessionContext(
-            connection_.get(), directory(), workers_,
+            connection_.get(), directory(),
             extensions_activity_.get(),
             std::vector<SyncEngineEventListener*>(), NULL, NULL,
+            model_type_registry_.get(),
             true,  // enable keystore encryption
             false,  // force enable pre-commit GU avoidance
             "fake_invalidator_client_id"));
-    context_->set_routing_info(routing_info_);
+    context_->SetRoutingInfo(routing_info_);
     context_->set_notifications_enabled(true);
     context_->set_account_name("Test");
     scheduler_.reset(
@@ -226,6 +230,7 @@ class SyncSchedulerTest : public testing::Test {
   TestDirectorySetterUpper dir_maker_;
   CancelationSignal cancelation_signal_;
   scoped_ptr<MockConnectionManager> connection_;
+  scoped_ptr<ModelTypeRegistry> model_type_registry_;
   scoped_ptr<SyncSessionContext> context_;
   scoped_ptr<SyncSchedulerImpl> scheduler_;
   MockSyncer* syncer_;
@@ -894,7 +899,7 @@ TEST_F(SyncSchedulerTest, ConfigurationMode) {
 
   // TODO(tim): Figure out how to remove this dangerous need to reset
   // routing info between mode switches.
-  context()->set_routing_info(routing_info());
+  context()->SetRoutingInfo(routing_info());
   StartSyncScheduler(SyncScheduler::NORMAL_MODE);
 
   RunLoop();
diff --git a/sync/engine/syncer.cc b/sync/engine/syncer.cc
index e75c1c6..339948d 100644
--- a/sync/engine/syncer.cc
+++ b/sync/engine/syncer.cc
@@ -12,8 +12,10 @@
 #include "build/build_config.h"
 #include "sync/engine/apply_control_data_updates.h"
 #include "sync/engine/commit.h"
+#include "sync/engine/commit_processor.h"
 #include "sync/engine/conflict_resolver.h"
 #include "sync/engine/download.h"
+#include "sync/engine/get_updates_processor.h"
 #include "sync/engine/net/server_connection_manager.h"
 #include "sync/engine/syncer_types.h"
 #include "sync/internal_api/public/base/cancelation_signal.h"
@@ -56,14 +58,18 @@ bool Syncer::NormalSyncShare(ModelTypeSet request_types,
                              const NudgeTracker& nudge_tracker,
                              SyncSession* session) {
   HandleCycleBegin(session);
+  GetUpdatesProcessor get_updates_processor(
+      session->context()->model_type_registry()->update_handler_map());
   VLOG(1) << "Downloading types " << ModelTypeSetToString(request_types);
   if (nudge_tracker.IsGetUpdatesRequired(base::TimeTicks::Now()) ||
       session->context()->ShouldFetchUpdatesBeforeCommit()) {
     if (!DownloadAndApplyUpdates(
             request_types,
             session,
+            &get_updates_processor,
             base::Bind(&download::BuildNormalDownloadUpdates,
                        session,
+                       &get_updates_processor,
                        kCreateMobileBookmarksFolder,
                        request_types,
                        base::ConstRef(nudge_tracker)))) {
@@ -72,7 +78,10 @@ bool Syncer::NormalSyncShare(ModelTypeSet request_types,
   }
 
   VLOG(1) << "Committing from types " << ModelTypeSetToString(request_types);
-  SyncerError commit_result = BuildAndPostCommits(request_types, session);
+  CommitProcessor commit_processor(
+      session->context()->model_type_registry()->commit_contributor_map());
+  SyncerError commit_result =
+      BuildAndPostCommits(request_types, session, &commit_processor);
   session->mutable_status_controller()->set_commit_result(commit_result);
 
   return HandleCycleEnd(session, nudge_tracker.updates_source());
@@ -83,12 +92,16 @@ bool Syncer::ConfigureSyncShare(
     sync_pb::GetUpdatesCallerInfo::GetUpdatesSource source,
     SyncSession* session) {
   HandleCycleBegin(session);
+  GetUpdatesProcessor get_updates_processor(
+      session->context()->model_type_registry()->update_handler_map());
   VLOG(1) << "Configuring types " << ModelTypeSetToString(request_types);
   DownloadAndApplyUpdates(
       request_types,
       session,
+      &get_updates_processor,
       base::Bind(&download::BuildDownloadUpdatesForConfigure,
                  session,
+                 &get_updates_processor,
                  kCreateMobileBookmarksFolder,
                  source,
                  request_types));
@@ -98,12 +111,16 @@ bool Syncer::ConfigureSyncShare(
 bool Syncer::PollSyncShare(ModelTypeSet request_types,
                            SyncSession* session) {
   HandleCycleBegin(session);
+  GetUpdatesProcessor get_updates_processor(
+      session->context()->model_type_registry()->update_handler_map());
   VLOG(1) << "Polling types " << ModelTypeSetToString(request_types);
   DownloadAndApplyUpdates(
       request_types,
       session,
+      &get_updates_processor,
       base::Bind(&download::BuildDownloadUpdatesForPoll,
                  session,
+                 &get_updates_processor,
                  kCreateMobileBookmarksFolder,
                  request_types));
   return HandleCycleEnd(session, sync_pb::GetUpdatesCallerInfo::PERIODIC);
@@ -112,27 +129,29 @@ bool Syncer::PollSyncShare(ModelTypeSet request_types,
 bool Syncer::RetrySyncShare(ModelTypeSet request_types,
                             SyncSession* session) {
   HandleCycleBegin(session);
+  GetUpdatesProcessor get_updates_processor(
+      session->context()->model_type_registry()->update_handler_map());
   VLOG(1) << "Retrying types " << ModelTypeSetToString(request_types);
   DownloadAndApplyUpdates(
       request_types,
       session,
+      &get_updates_processor,
       base::Bind(&download::BuildDownloadUpdatesForRetry,
                  session,
+                 &get_updates_processor,
                  kCreateMobileBookmarksFolder,
                  request_types));
   return HandleCycleEnd(session, sync_pb::GetUpdatesCallerInfo::RETRY);
 }
 
-void Syncer::ApplyUpdates(SyncSession* session) {
+void Syncer::ApplyUpdates(SyncSession* session,
+                          GetUpdatesProcessor* get_updates_processor) {
   TRACE_EVENT0("sync", "ApplyUpdates");
 
   ApplyControlDataUpdates(session->context()->directory());
 
-  UpdateHandlerMap* handler_map = session->context()->update_handler_map();
-  for (UpdateHandlerMap::iterator it = handler_map->begin();
-       it != handler_map->end(); ++it) {
-    it->second->ApplyUpdates(session->mutable_status_controller());
-  }
+  get_updates_processor->ApplyUpdatesForAllTypes(
+      session->mutable_status_controller());
 
   session->context()->set_hierarchy_conflict_detected(
       session->status_controller().num_hierarchy_conflicts() > 0);
@@ -143,14 +162,17 @@ void Syncer::ApplyUpdates(SyncSession* session) {
 bool Syncer::DownloadAndApplyUpdates(
     ModelTypeSet request_types,
     SyncSession* session,
+    GetUpdatesProcessor* get_updates_processor,
     base::Callback<void(sync_pb::ClientToServerMessage*)> build_fn) {
   SyncerError download_result = UNSET;
   do {
     TRACE_EVENT0("sync", "DownloadUpdates");
     sync_pb::ClientToServerMessage msg;
     build_fn.Run(&msg);
-    download_result =
-        download::ExecuteDownloadUpdates(request_types, session, &msg);
+    download_result = download::ExecuteDownloadUpdates(request_types,
+                                                       session,
+                                                       get_updates_processor,
+                                                       &msg);
     session->mutable_status_controller()->set_last_download_updates_result(
         download_result);
   } while (download_result == SERVER_MORE_TO_DOWNLOAD);
@@ -161,14 +183,15 @@ bool Syncer::DownloadAndApplyUpdates(
   if (ExitRequested())
     return false;
 
-  ApplyUpdates(session);
+  ApplyUpdates(session, get_updates_processor);
   if (ExitRequested())
     return false;
   return true;
 }
 
 SyncerError Syncer::BuildAndPostCommits(ModelTypeSet requested_types,
-                                        sessions::SyncSession* session) {
+                                        sessions::SyncSession* session,
+                                        CommitProcessor* commit_processor) {
   // The ExitRequested() check is unnecessary, since we should start getting
   // errors from the ServerConnectionManager if an exist has been requested.
   // However, it doesn't hurt to check it anyway.
@@ -176,10 +199,11 @@ SyncerError Syncer::BuildAndPostCommits(ModelTypeSet requested_types,
     scoped_ptr<Commit> commit(
         Commit::Init(
             requested_types,
+            session->context()->enabled_types(),
             session->context()->max_commit_batch_size(),
             session->context()->account_name(),
             session->context()->directory()->cache_guid(),
-            session->context()->commit_contributor_map(),
+            commit_processor,
             session->context()->extensions_activity()));
     if (!commit) {
       break;
diff --git a/sync/engine/syncer.h b/sync/engine/syncer.h
index 225b4e3..51c66aa 100644
--- a/sync/engine/syncer.h
+++ b/sync/engine/syncer.h
@@ -22,6 +22,8 @@
 namespace syncer {
 
 class CancelationSignal;
+class CommitProcessor;
+class GetUpdatesProcessor;
 
 // A Syncer provides a control interface for driving the sync cycle.  These
 // cycles consist of downloading updates, parsing the response (aka. process
@@ -69,10 +71,12 @@ class SYNC_EXPORT_PRIVATE Syncer {
                               sessions::SyncSession* session);
 
  private:
-  void ApplyUpdates(sessions::SyncSession* session);
+  void ApplyUpdates(sessions::SyncSession* session,
+                    GetUpdatesProcessor* get_updates_processor);
   bool DownloadAndApplyUpdates(
       ModelTypeSet request_types,
       sessions::SyncSession* session,
+      GetUpdatesProcessor* get_updates_processor,
       base::Callback<void(sync_pb::ClientToServerMessage*)> build_fn);
 
   // This function will commit batches of unsynced items to the server until the
@@ -81,7 +85,8 @@ class SYNC_EXPORT_PRIVATE Syncer {
   // abort any blocking operations.
   SyncerError BuildAndPostCommits(
       ModelTypeSet request_types,
-      sessions::SyncSession* session);
+      sessions::SyncSession* session,
+      CommitProcessor* commit_processor);
 
   void HandleCycleBegin(sessions::SyncSession* session);
   bool HandleCycleEnd(
diff --git a/sync/engine/syncer_unittest.cc b/sync/engine/syncer_unittest.cc
index 119deb1..a8e7c02 100644
--- a/sync/engine/syncer_unittest.cc
+++ b/sync/engine/syncer_unittest.cc
@@ -220,15 +220,18 @@ class SyncerTest : public testing::Test,
     ModelSafeRoutingInfo routing_info;
     GetModelSafeRoutingInfo(&routing_info);
 
+    model_type_registry_.reset(new ModelTypeRegistry(workers_, directory()));
+
     context_.reset(
         new SyncSessionContext(
-            mock_server_.get(), directory(), workers_,
+            mock_server_.get(), directory(),
             extensions_activity_,
             listeners, debug_info_getter_.get(), &traffic_recorder_,
+            model_type_registry_.get(),
             true,  // enable keystore encryption
             false,  // force enable pre-commit GU avoidance experiment
             "fake_invalidator_client_id"));
-    context_->set_routing_info(routing_info);
+    context_->SetRoutingInfo(routing_info);
     syncer_ = new Syncer(&cancelation_signal_);
 
     syncable::ReadTransaction trans(FROM_HERE, directory());
@@ -437,7 +440,7 @@ class SyncerTest : public testing::Test,
     GetModelSafeRoutingInfo(&routing_info);
 
     if (context_) {
-      context_->set_routing_info(routing_info);
+      context_->SetRoutingInfo(routing_info);
     }
 
     mock_server_->ExpectGetUpdatesRequestTypes(enabled_datatypes_);
@@ -450,7 +453,7 @@ class SyncerTest : public testing::Test,
     GetModelSafeRoutingInfo(&routing_info);
 
     if (context_) {
-      context_->set_routing_info(routing_info);
+      context_->SetRoutingInfo(routing_info);
     }
 
     mock_server_->ExpectGetUpdatesRequestTypes(enabled_datatypes_);
@@ -493,6 +496,7 @@ class SyncerTest : public testing::Test,
   Syncer* syncer_;
 
   scoped_ptr<SyncSession> session_;
+  scoped_ptr<ModelTypeRegistry> model_type_registry_;
   scoped_ptr<SyncSessionContext> context_;
   bool saw_syncer_event_;
   base::TimeDelta last_short_poll_interval_received_;