sync: Implement per-type update processing

Introduces a new class that represents a syncable::Directory's udpate
requesting and processing capabilities.  The intention is that this will
eventually be expanded to cover commits as well.  Eventually, this will
be used as the basis for an interface between sync and 

Some update logic has been moved into this SyncDirectoryUpdateHandler
class or download.cc.  The rest of it can be found in
process_update_util.cc, the successor to the old ProcessUpdatesCommand.
The StoreTimestampsCommand has been entirely removed.  The unit tests
associated with these SyncerCommands have been ported to
sync_directory_update_handler_unittest.cc.

Except for a few error scenarios that are now handled differently, the
observable behavior of the client should not be changed by this CL.

BUG=278484

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

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

15 files changed, 597 insertions, 541 deletions

diff --git a/sync/engine/download.cc b/sync/engine/download.cc
index acda834..9b74417 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,8 @@ using std::string;
 
 namespace {
 
+typedef std::map<ModelType, size_t> TypeToIndexMap;
+
 SyncerError HandleGetEncryptionKeyResponse(
     const sync_pb::ClientToServerResponse& update_response,
     syncable::Directory* dir) {
@@ -98,7 +100,7 @@ void InitDownloadUpdatesRequest(
     SyncSession* session,
     bool create_mobile_bookmarks_folder,
     sync_pb::ClientToServerMessage* message,
-    ModelTypeSet request_types) {
+    ModelTypeSet proto_request_types) {
   message->set_share(session->context()->account_name());
   message->set_message_contents(sync_pb::ClientToServerMessage::GET_UPDATES);
 
@@ -122,19 +124,93 @@ void InitDownloadUpdatesRequest(
       session->context()->notifications_enabled());
 
   StatusController* status = session->mutable_status_controller();
-  status->set_updates_request_types(request_types);
+  status->set_updates_request_types(proto_request_types);
+
+  UpdateHandlerMap* handler_map = session->context()->update_handler_map();
 
-  syncable::Directory* dir = session->context()->directory();
-  for (ModelTypeSet::Iterator it = request_types.First();
+  for (ModelTypeSet::Iterator it = proto_request_types.First();
        it.Good(); it.Inc()) {
-    if (ProxyTypes().Has(it.Get()))
-      continue;
+    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();
-    dir->GetDownloadProgress(it.Get(), 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 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());
+
+  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(
@@ -147,7 +223,7 @@ void BuildNormalDownloadUpdates(
       session,
       create_mobile_bookmarks_folder,
       client_to_server_message,
-      request_types);
+      Intersection(request_types, ProtocolTypes()));
   sync_pb::GetUpdatesMessage* get_updates =
       client_to_server_message->mutable_get_updates();
 
@@ -190,7 +266,7 @@ void BuildDownloadUpdatesForConfigure(
       session,
       create_mobile_bookmarks_folder,
       client_to_server_message,
-      request_types);
+      Intersection(request_types, ProtocolTypes()));
   sync_pb::GetUpdatesMessage* get_updates =
       client_to_server_message->mutable_get_updates();
 
@@ -217,7 +293,7 @@ void BuildDownloadUpdatesForPoll(
       session,
       create_mobile_bookmarks_folder,
       client_to_server_message,
-      request_types);
+      Intersection(request_types, ProtocolTypes()));
   sync_pb::GetUpdatesMessage* get_updates =
       client_to_server_message->mutable_get_updates();
 
@@ -234,6 +310,7 @@ void BuildDownloadUpdatesForPoll(
 }
 
 SyncerError ExecuteDownloadUpdates(
+    ModelTypeSet request_types,
     SyncSession* session,
     sync_pb::ClientToServerMessage* msg) {
   sync_pb::ClientToServerResponse update_response;
@@ -251,30 +328,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;
   }
 
-  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
diff --git a/sync/engine/download.h b/sync/engine/download.h
index ddae79b..7c5c582 100644
--- a/sync/engine/download.h
+++ b/sync/engine/download.h
@@ -55,7 +55,8 @@ SYNC_EXPORT_PRIVATE void BuildDownloadUpdatesForPoll(
 // Sends the specified message to the server and stores the response in a member
 // of the |session|'s StatusController.
 SYNC_EXPORT_PRIVATE SyncerError
-    ExecuteDownloadUpdates(sessions::SyncSession* session,
+    ExecuteDownloadUpdates(ModelTypeSet request_types,
+                           sessions::SyncSession* session,
                            sync_pb::ClientToServerMessage* msg);
 
 }  // namespace syncer
diff --git a/sync/engine/process_updates_command.h b/sync/engine/process_updates_command.h
deleted file mode 100644
index 4d7d287..0000000
--- a/sync/engine/process_updates_command.h
+++ /dev/null
@@ -1,67 +0,0 @@
-// 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.
-
-#ifndef SYNC_ENGINE_PROCESS_UPDATES_COMMAND_H_
-#define SYNC_ENGINE_PROCESS_UPDATES_COMMAND_H_
-
-#include "base/compiler_specific.h"
-#include "sync/base/sync_export.h"
-#include "sync/engine/model_changing_syncer_command.h"
-#include "sync/engine/syncer_types.h"
-
-namespace sync_pb {
-class GetUpdatesResponse;
-class SyncEntity;
-}
-
-namespace syncer {
-
-namespace syncable {
-class ModelNeutralWriteTransaction;
-}
-
-class Cryptographer;
-
-// A syncer command for verifying and processing updates.
-//
-// Preconditions - Updates in the SyncerSesssion have been downloaded.
-//
-// Postconditions - All of the verified SyncEntity data will be copied to
-//                  the server fields of the corresponding syncable entries.
-class SYNC_EXPORT_PRIVATE ProcessUpdatesCommand : public SyncerCommand {
- public:
-  ProcessUpdatesCommand();
-  virtual ~ProcessUpdatesCommand();
-
- protected:
-  // SyncerCommand implementation.
-  virtual SyncerError ExecuteImpl(sessions::SyncSession* session) OVERRIDE;
-
- private:
-  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.  ModelTypes that had no items in the update response
-  // will not be included in this map.
-  static void PartitionUpdatesByType(
-      const sync_pb::GetUpdatesResponse& updates,
-      TypeSyncEntityMap* updates_by_type);
-
-  VerifyResult VerifyUpdate(
-      syncable::ModelNeutralWriteTransaction* trans,
-      const sync_pb::SyncEntity& entry,
-      ModelTypeSet requested_types,
-      const ModelSafeRoutingInfo& routes);
-  void ProcessUpdate(
-      const sync_pb::SyncEntity& proto_update,
-      const Cryptographer* cryptographer,
-      syncable::ModelNeutralWriteTransaction* const trans);
-  DISALLOW_COPY_AND_ASSIGN(ProcessUpdatesCommand);
-};
-
-}  // namespace syncer
-
-#endif  // SYNC_ENGINE_PROCESS_UPDATES_COMMAND_H_
diff --git a/sync/engine/process_updates_command_unittest.cc b/sync/engine/process_updates_command_unittest.cc
deleted file mode 100644
index 44b308a..0000000
--- a/sync/engine/process_updates_command_unittest.cc
+++ /dev/null
@@ -1,173 +0,0 @@
-// 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 "base/basictypes.h"
-#include "sync/engine/process_updates_command.h"
-#include "sync/engine/syncer_proto_util.h"
-#include "sync/internal_api/public/base/model_type.h"
-#include "sync/internal_api/public/test/test_entry_factory.h"
-#include "sync/sessions/sync_session.h"
-#include "sync/syncable/mutable_entry.h"
-#include "sync/syncable/syncable_id.h"
-#include "sync/syncable/syncable_proto_util.h"
-#include "sync/syncable/syncable_read_transaction.h"
-#include "sync/syncable/syncable_write_transaction.h"
-#include "sync/test/engine/fake_model_worker.h"
-#include "sync/test/engine/syncer_command_test.h"
-#include "testing/gtest/include/gtest/gtest.h"
-
-namespace syncer {
-
-using sync_pb::SyncEntity;
-using syncable::Id;
-using syncable::MutableEntry;
-using syncable::UNITTEST;
-using syncable::WriteTransaction;
-
-namespace {
-
-class ProcessUpdatesCommandTest : public SyncerCommandTest {
- protected:
-  ProcessUpdatesCommandTest() {}
-  virtual ~ProcessUpdatesCommandTest() {}
-
-  virtual void SetUp() {
-    workers()->push_back(
-        make_scoped_refptr(new FakeModelWorker(GROUP_UI)));
-    workers()->push_back(
-        make_scoped_refptr(new FakeModelWorker(GROUP_DB)));
-    (*mutable_routing_info())[PREFERENCES] = GROUP_UI;
-    (*mutable_routing_info())[BOOKMARKS] = GROUP_UI;
-    (*mutable_routing_info())[AUTOFILL] = GROUP_DB;
-    SyncerCommandTest::SetUp();
-    test_entry_factory_.reset(new TestEntryFactory(directory()));
-  }
-
-  void CreateLocalItem(const std::string& item_id,
-                       const std::string& parent_id,
-                       const ModelType& type) {
-    WriteTransaction trans(FROM_HERE, UNITTEST, directory());
-    MutableEntry entry(&trans, syncable::CREATE_NEW_UPDATE_ITEM,
-        Id::CreateFromServerId(item_id));
-    ASSERT_TRUE(entry.good());
-
-    entry.PutBaseVersion(1);
-    entry.PutServerVersion(1);
-    entry.PutNonUniqueName(item_id);
-    entry.PutParentId(Id::CreateFromServerId(parent_id));
-    sync_pb::EntitySpecifics default_specifics;
-    AddDefaultFieldValue(type, &default_specifics);
-    entry.PutServerSpecifics(default_specifics);
-  }
-
-  SyncEntity* AddUpdate(sync_pb::GetUpdatesResponse* updates,
-      const std::string& id, const std::string& parent,
-      const ModelType& type) {
-    sync_pb::SyncEntity* e = updates->add_entries();
-    e->set_id_string(id);
-    e->set_parent_id_string(parent);
-    e->set_non_unique_name(id);
-    e->set_name(id);
-    e->set_version(1000);
-    AddDefaultFieldValue(type, e->mutable_specifics());
-    return e;
-  }
-
-  ProcessUpdatesCommand command_;
-  scoped_ptr<TestEntryFactory> test_entry_factory_;
-
- private:
-  DISALLOW_COPY_AND_ASSIGN(ProcessUpdatesCommandTest);
-};
-
-static const char kCacheGuid[] = "IrcjZ2jyzHDV9Io4+zKcXQ==";
-
-// Test that the bookmark tag is set on newly downloaded items.
-TEST_F(ProcessUpdatesCommandTest, NewBookmarkTag) {
-  std::string root = syncable::GetNullId().GetServerId();
-  sync_pb::GetUpdatesResponse* updates =
-      session()->mutable_status_controller()->
-      mutable_updates_response()->mutable_get_updates();
-  Id server_id = Id::CreateFromServerId("b1");
-  SyncEntity* e =
-      AddUpdate(updates, SyncableIdToProto(server_id), root, BOOKMARKS);
-
-  e->set_originator_cache_guid(
-      std::string(kCacheGuid, arraysize(kCacheGuid)-1));
-  Id client_id = Id::CreateFromClientString("-2");
-  e->set_originator_client_item_id(client_id.GetServerId());
-  e->set_position_in_parent(0);
-
-  command_.Execute(session());
-
-  syncable::ReadTransaction trans(FROM_HERE, directory());
-  syncable::Entry entry(&trans, syncable::GET_BY_ID, server_id);
-  ASSERT_TRUE(entry.good());
-  EXPECT_TRUE(
-      UniquePosition::IsValidSuffix(entry.GetUniqueBookmarkTag()));
-  EXPECT_TRUE(entry.GetServerUniquePosition().IsValid());
-
-  // If this assertion fails, that might indicate that the algorithm used to
-  // generate bookmark tags has been modified.  This could have implications for
-  // bookmark ordering.  Please make sure you know what you're doing if you
-  // intend to make such a change.
-  EXPECT_EQ("6wHRAb3kbnXV5GHrejp4/c1y5tw=",
-            entry.GetUniqueBookmarkTag());
-}
-
-TEST_F(ProcessUpdatesCommandTest, ReceiveServerCreatedBookmarkFolders) {
-  Id server_id = Id::CreateFromServerId("xyz");
-  std::string root = syncable::GetNullId().GetServerId();
-  sync_pb::GetUpdatesResponse* updates =
-      session()->mutable_status_controller()->
-      mutable_updates_response()->mutable_get_updates();
-
-  // Create an update that mimics the bookmark root.
-  SyncEntity* e =
-      AddUpdate(updates, SyncableIdToProto(server_id), root, BOOKMARKS);
-  e->set_server_defined_unique_tag("google_chrome_bookmarks");
-  e->set_folder(true);
-
-  EXPECT_FALSE(SyncerProtoUtil::ShouldMaintainPosition(*e));
-
-  command_.Execute(session());
-
-  syncable::ReadTransaction trans(FROM_HERE, directory());
-  syncable::Entry entry(&trans, syncable::GET_BY_ID, server_id);
-  ASSERT_TRUE(entry.good());
-
-  EXPECT_FALSE(entry.ShouldMaintainPosition());
-  EXPECT_FALSE(entry.GetUniquePosition().IsValid());
-  EXPECT_FALSE(entry.GetServerUniquePosition().IsValid());
-  EXPECT_TRUE(entry.GetUniqueBookmarkTag().empty());
-}
-
-TEST_F(ProcessUpdatesCommandTest, ReceiveNonBookmarkItem) {
-  Id server_id = Id::CreateFromServerId("xyz");
-  std::string root = syncable::GetNullId().GetServerId();
-  sync_pb::GetUpdatesResponse* updates =
-      session()->mutable_status_controller()->
-      mutable_updates_response()->mutable_get_updates();
-
-  SyncEntity* e =
-      AddUpdate(updates, SyncableIdToProto(server_id), root, AUTOFILL);
-  e->set_server_defined_unique_tag("9PGRuKdX5sHyGMB17CvYTXuC43I=");
-
-  EXPECT_FALSE(SyncerProtoUtil::ShouldMaintainPosition(*e));
-
-  command_.Execute(session());
-
-  syncable::ReadTransaction trans(FROM_HERE, directory());
-  syncable::Entry entry(&trans, syncable::GET_BY_ID, server_id);
-  ASSERT_TRUE(entry.good());
-
-  EXPECT_FALSE(entry.ShouldMaintainPosition());
-  EXPECT_FALSE(entry.GetUniquePosition().IsValid());
-  EXPECT_FALSE(entry.GetServerUniquePosition().IsValid());
-  EXPECT_TRUE(entry.GetUniqueBookmarkTag().empty());
-}
-
-}  // namespace
-
-}  // namespace syncer
diff --git a/sync/engine/process_updates_command.cc b/sync/engine/process_updates_util.cc
index d69c109..49e40b3 100644
--- a/sync/engine/process_updates_command.cc
+++ b/sync/engine/process_updates_util.cc
@@ -1,17 +1,12 @@
-// Copyright 2012 The Chromium Authors. All rights reserved.
+// Copyright 2013 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"
@@ -19,18 +14,12 @@
 #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
@@ -87,57 +76,51 @@ bool UpdateContainsNewVersion(syncable::BaseTransaction *trans,
 
 }  // 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);
+      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);
   }
 }
 
@@ -160,11 +143,10 @@ VerifyResult VerifyTagConsistency(
 
 }  // 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;
 
@@ -198,8 +180,7 @@ VerifyResult ProcessUpdatesCommand::VerifyUpdate(
     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;
@@ -240,7 +221,7 @@ bool ReverifyEntry(syncable::ModelNeutralWriteTransaction* trans,
 }  // 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) {
diff --git a/sync/engine/process_updates_util.h b/sync/engine/process_updates_util.h
new file mode 100644
index 0000000..6e8bc71
--- /dev/null
+++ b/sync/engine/process_updates_util.h
@@ -0,0 +1,73 @@
+// Copyright 2013 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_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;
+}
+
+namespace syncer {
+
+namespace sessions {
+class StatusController;
+}
+
+namespace syncable {
+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(
+    syncable::Directory* dir,
+    syncable::ModelNeutralWriteTransaction* trans,
+    ModelType type,
+    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/store_timestamps_command.cc b/sync/engine/store_timestamps_command.cc
deleted file mode 100644
index cbff68f..0000000
--- a/sync/engine/store_timestamps_command.cc
+++ /dev/null
@@ -1,63 +0,0 @@
-// 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/store_timestamps_command.h"
-
-#include "base/logging.h"
-#include "sync/sessions/status_controller.h"
-#include "sync/sessions/sync_session.h"
-#include "sync/syncable/directory.h"
-
-namespace syncer {
-
-ModelTypeSet ProcessNewProgressMarkers(
-    const sync_pb::GetUpdatesResponse& response,
-    syncable::Directory* dir) {
-  ModelTypeSet forward_progress_types;
-  // If a marker was omitted for any one type, that indicates no
-  // change from the previous state.
-  for (int i = 0; i < response.new_progress_marker_size(); ++i) {
-    int field_number = 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;
-    }
-    forward_progress_types.Put(model_type);
-    dir->SetDownloadProgress(model_type, response.new_progress_marker(i));
-  }
-  return forward_progress_types;
-}
-
-StoreTimestampsCommand::StoreTimestampsCommand() {}
-StoreTimestampsCommand::~StoreTimestampsCommand() {}
-
-SyncerError StoreTimestampsCommand::ExecuteImpl(
-    sessions::SyncSession* session) {
-  const sync_pb::GetUpdatesResponse& updates =
-      session->status_controller().updates_response().get_updates();
-
-  sessions::StatusController* status = session->mutable_status_controller();
-
-  ModelTypeSet forward_progress_types =
-      ProcessNewProgressMarkers(updates, session->context()->directory());
-  DCHECK(!forward_progress_types.Empty() ||
-         updates.changes_remaining() == 0);
-  if (VLOG_IS_ON(1)) {
-    DVLOG_IF(1, !forward_progress_types.Empty())
-        << "Get Updates got new progress marker for types: "
-        << ModelTypeSetToString(forward_progress_types)
-        << " out of possible: "
-        << ModelTypeSetToString(status->updates_request_types());
-  }
-  if (updates.has_changes_remaining()) {
-    int64 changes_left = updates.changes_remaining();
-    DVLOG(1) << "Changes remaining: " << changes_left;
-    status->set_num_server_changes_remaining(changes_left);
-  }
-
-  return SYNCER_OK;
-}
-
-}  // namespace syncer
diff --git a/sync/engine/store_timestamps_command.h b/sync/engine/store_timestamps_command.h
deleted file mode 100644
index a6426a8..0000000
--- a/sync/engine/store_timestamps_command.h
+++ /dev/null
@@ -1,56 +0,0 @@
-// 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.
-
-#ifndef SYNC_ENGINE_STORE_TIMESTAMPS_COMMAND_H_
-#define SYNC_ENGINE_STORE_TIMESTAMPS_COMMAND_H_
-
-#include "base/compiler_specific.h"
-#include "sync/base/sync_export.h"
-#include "sync/engine/syncer_command.h"
-#include "sync/engine/syncer_types.h"
-#include "sync/internal_api/public/base/model_type.h"
-
-namespace sync_pb {
-class GetUpdatesResponse;
-}  // namespace sync_pb
-
-namespace syncer {
-
-namespace syncable {
-class Directory;
-}  // namespace syncable
-
-// Sets |dir|'s progress markers from the data in |response|.  Returns
-// the set of model types with new progress markers.
-SYNC_EXPORT_PRIVATE ModelTypeSet ProcessNewProgressMarkers(
-    const sync_pb::GetUpdatesResponse& response,
-    syncable::Directory* dir);
-
-// A syncer command that extracts the changelog timestamp information from
-// a GetUpdatesResponse (fetched in DownloadUpdatesCommand) and stores
-// it in the directory.  This is meant to run immediately after
-// ProcessUpdatesCommand.
-//
-// Preconditions - all updates in the SyncerSesssion have been stored in the
-//                 database, meaning it is safe to update the persisted
-//                 timestamps.
-//
-// Postconditions - The next_timestamp returned by the server will be
-//                  saved into the directory (where it will be used
-//                  the next time that DownloadUpdatesCommand runs).
-class StoreTimestampsCommand : public SyncerCommand {
- public:
-  StoreTimestampsCommand();
-  virtual ~StoreTimestampsCommand();
-
-  // SyncerCommand implementation.
-  virtual SyncerError ExecuteImpl(sessions::SyncSession* session) OVERRIDE;
-
- private:
-  DISALLOW_COPY_AND_ASSIGN(StoreTimestampsCommand);
-};
-
-}  // namespace syncer
-
-#endif  // SYNC_ENGINE_STORE_TIMESTAMPS_COMMAND_H_
diff --git a/sync/engine/store_timestamps_command_unittest.cc b/sync/engine/store_timestamps_command_unittest.cc
deleted file mode 100644
index e49eb31..0000000
--- a/sync/engine/store_timestamps_command_unittest.cc
+++ /dev/null
@@ -1,83 +0,0 @@
-// 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 "base/basictypes.h"
-#include "sync/engine/store_timestamps_command.h"
-#include "sync/internal_api/public/base/model_type.h"
-#include "sync/protocol/sync.pb.h"
-#include "sync/test/engine/syncer_command_test.h"
-#include "testing/gtest/include/gtest/gtest.h"
-
-namespace syncer {
-
-namespace {
-
-// Adds a progress marker to |response| for the given field number and
-// token.
-void AddProgressMarkerForFieldNumber(
-    sync_pb::GetUpdatesResponse* response,
-    int field_number, const std::string& token) {
-  sync_pb::DataTypeProgressMarker* marker =
-      response->add_new_progress_marker();
-  marker->set_data_type_id(field_number);
-  marker->set_token(token);
-}
-
-// Adds a progress marker to |response| for the given model type and
-// token.
-void AddProgressMarkerForModelType(
-    sync_pb::GetUpdatesResponse* response,
-    ModelType model_type, const std::string& token) {
-  AddProgressMarkerForFieldNumber(
-      response, GetSpecificsFieldNumberFromModelType(model_type), token);
-}
-
-class StoreTimestampsCommandTest : public SyncerCommandTest {
- protected:
-  // Gets the directory's progress marker's token for the given model
-  // type.
-  std::string GetProgessMarkerToken(ModelType model_type) {
-    sync_pb::DataTypeProgressMarker progress_marker;
-    session()->context()->directory()->GetDownloadProgress(
-        model_type, &progress_marker);
-    EXPECT_EQ(
-        GetSpecificsFieldNumberFromModelType(model_type),
-        progress_marker.data_type_id());
-    return progress_marker.token();
-  }
-};
-
-// Builds a GetUpdatesResponse with some progress markers, including
-// invalid ones.  ProcessNewProgressMarkers() should return the model
-// types for the valid progress markers and fill in the progress
-// markers in the directory.
-TEST_F(StoreTimestampsCommandTest, ProcessNewProgressMarkers) {
-  sync_pb::GetUpdatesResponse response;
-  AddProgressMarkerForModelType(&response, BOOKMARKS, "token1");
-  AddProgressMarkerForModelType(&response,
-                                HISTORY_DELETE_DIRECTIVES, "token2");
-  AddProgressMarkerForFieldNumber(&response, -1, "bad token");
-
-  ModelTypeSet forward_progress_types =
-      ProcessNewProgressMarkers(
-          response, session()->context()->directory());
-
-  EXPECT_TRUE(
-      forward_progress_types.Equals(
-          ModelTypeSet(BOOKMARKS, HISTORY_DELETE_DIRECTIVES)));
-
-  EXPECT_EQ("token1", GetProgessMarkerToken(BOOKMARKS));
-  EXPECT_EQ("token2", GetProgessMarkerToken(HISTORY_DELETE_DIRECTIVES));
-
-  ModelTypeSet non_forward_progress_types =
-      Difference(ProtocolTypes(), forward_progress_types);
-  for (ModelTypeSet::Iterator it = non_forward_progress_types.First();
-       it.Good(); it.Inc()) {
-    EXPECT_TRUE(GetProgessMarkerToken(it.Get()).empty());
-  }
-}
-
-}  // namespace
-
-}  // namespace syncer
diff --git a/sync/engine/sync_directory_update_handler.cc b/sync/engine/sync_directory_update_handler.cc
new file mode 100644
index 0000000..0d7953b
--- /dev/null
+++ b/sync/engine/sync_directory_update_handler.cc
@@ -0,0 +1,57 @@
+// Copyright 2013 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/sync_directory_update_handler.h"
+
+#include "sync/engine/process_updates_util.h"
+#include "sync/sessions/status_controller.h"
+#include "sync/syncable/directory.h"
+#include "sync/syncable/syncable_model_neutral_write_transaction.h"
+
+namespace syncer {
+
+using syncable::SYNCER;
+
+SyncDirectoryUpdateHandler::SyncDirectoryUpdateHandler(
+    syncable::Directory* dir, ModelType type)
+  : dir_(dir),
+    type_(type) {}
+
+SyncDirectoryUpdateHandler::~SyncDirectoryUpdateHandler() {}
+
+void SyncDirectoryUpdateHandler::GetDownloadProgress(
+    sync_pb::DataTypeProgressMarker* progress_marker) const {
+  dir_->GetDownloadProgress(type_, progress_marker);
+}
+
+void SyncDirectoryUpdateHandler::ProcessGetUpdatesResponse(
+    const sync_pb::DataTypeProgressMarker& progress_marker,
+    const SyncEntityList& applicable_updates,
+    sessions::StatusController* status) {
+  syncable::ModelNeutralWriteTransaction trans(FROM_HERE, SYNCER, dir_);
+  UpdateSyncEntities(&trans, applicable_updates, status);
+  UpdateProgressMarker(progress_marker);
+}
+
+void SyncDirectoryUpdateHandler::UpdateSyncEntities(
+    syncable::ModelNeutralWriteTransaction* trans,
+    const SyncEntityList& applicable_updates,
+    sessions::StatusController* status) {
+  ProcessDownloadedUpdates(dir_, trans, type_, applicable_updates, status);
+}
+
+void SyncDirectoryUpdateHandler::UpdateProgressMarker(
+    const sync_pb::DataTypeProgressMarker& progress_marker) {
+  int field_number = progress_marker.data_type_id();
+  ModelType model_type = GetModelTypeFromSpecificsFieldNumber(field_number);
+  if (!IsRealDataType(model_type) || type_ != model_type) {
+    NOTREACHED()
+        << "Update handler of type " << ModelTypeToString(type_)
+        << " asked to process progress marker with invalid type "
+        << field_number;
+  }
+  dir_->SetDownloadProgress(type_, progress_marker);
+}
+
+}  // namespace syncer
diff --git a/sync/engine/sync_directory_update_handler.h b/sync/engine/sync_directory_update_handler.h
new file mode 100644
index 0000000..38256a4
--- /dev/null
+++ b/sync/engine/sync_directory_update_handler.h
@@ -0,0 +1,82 @@
+// Copyright 2013 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_SYNC_DIRECTORY_UPDATE_HANDLER_H_
+#define SYNC_ENGINE_SYNC_DIRECTORY_UPDATE_HANDLER_H_
+
+#include <map>
+
+#include "base/basictypes.h"
+#include "sync/base/sync_export.h"
+#include "sync/engine/process_updates_util.h"
+#include "sync/internal_api/public/base/model_type.h"
+
+namespace sync_pb {
+class DataTypeProgressMarker;
+class GetUpdatesResponse;
+}
+
+namespace syncer {
+
+namespace sessions {
+class StatusController;
+}
+
+namespace syncable {
+class Directory;
+}
+
+// This class represents the syncable::Directory's processes for requesting and
+// processing updates from the sync server.
+//
+// Each instance of this class represents a particular type in the
+// syncable::Directory.  It can store and retreive that type's progress markers.
+// It can also process a set of received SyncEntities and store their data.
+class SYNC_EXPORT_PRIVATE SyncDirectoryUpdateHandler {
+ public:
+  SyncDirectoryUpdateHandler(syncable::Directory* dir, ModelType type);
+  ~SyncDirectoryUpdateHandler();
+
+  // Fills the given parameter with the stored progress marker for this type.
+  void GetDownloadProgress(
+      sync_pb::DataTypeProgressMarker* progress_marker) const;
+
+  // Processes the contents of a GetUpdates response message.
+  //
+  // Should be invoked with the progress marker and set of SyncEntities from a
+  // single GetUpdates response message.  The progress marker's type must match
+  // this update handler's type, and the set of SyncEntities must include all
+  // entities of this type found in the response message.
+  void ProcessGetUpdatesResponse(
+      const sync_pb::DataTypeProgressMarker& progress_marker,
+      const SyncEntityList& applicable_updates,
+      sessions::StatusController* status);
+
+ private:
+  friend class SyncDirectoryUpdateHandlerTest;
+
+  // Processes the given SyncEntities and stores their data in the directory.
+  // Their types must match this update handler's type.
+  void UpdateSyncEntities(
+      syncable::ModelNeutralWriteTransaction* trans,
+      const SyncEntityList& applicable_updates,
+      sessions::StatusController* status);
+
+  // Stores the given progress marker in the directory.
+  // Its type must match this update handler's type.
+  void UpdateProgressMarker(
+      const sync_pb::DataTypeProgressMarker& progress_marker);
+
+  syncable::Directory* dir_;
+  ModelType type_;
+
+  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
new file mode 100644
index 0000000..53589b8
--- /dev/null
+++ b/sync/engine/sync_directory_update_handler_unittest.cc
@@ -0,0 +1,211 @@
+// Copyright 2013 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/sync_directory_update_handler.h"
+
+#include "base/compiler_specific.h"
+#include "base/message_loop/message_loop.h"
+#include "sync/engine/syncer_proto_util.h"
+#include "sync/internal_api/public/base/model_type.h"
+#include "sync/protocol/sync.pb.h"
+#include "sync/sessions/status_controller.h"
+#include "sync/syncable/directory.h"
+#include "sync/syncable/entry.h"
+#include "sync/syncable/syncable_model_neutral_write_transaction.h"
+#include "sync/syncable/syncable_proto_util.h"
+#include "sync/syncable/syncable_read_transaction.h"
+#include "sync/test/engine/test_directory_setter_upper.h"
+#include "sync/test/engine/test_syncable_utils.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace syncer {
+
+using syncable::UNITTEST;
+
+class SyncDirectoryUpdateHandlerTest : public ::testing::Test {
+ public:
+  virtual void SetUp() OVERRIDE {
+    dir_maker_.SetUp();
+  }
+
+  virtual void TearDown() OVERRIDE {
+    dir_maker_.TearDown();
+  }
+
+  syncable::Directory* dir() {
+    return dir_maker_.directory();
+  }
+ protected:
+  scoped_ptr<sync_pb::SyncEntity> CreateUpdate(
+      const std::string& id,
+      const std::string& parent,
+      const ModelType& type);
+
+  // This exists mostly to give tests access to the protected member function.
+  // Warning: This takes the syncable directory lock.
+  void UpdateSyncEntities(
+      SyncDirectoryUpdateHandler* handler,
+      const SyncEntityList& applicable_updates,
+      sessions::StatusController* status);
+
+  // Another function to access private member functions.
+  void UpdateProgressMarkers(
+      SyncDirectoryUpdateHandler* handler,
+      const sync_pb::DataTypeProgressMarker& progress);
+
+ private:
+  base::MessageLoop loop_;  // Needed to initialize the directory.
+  TestDirectorySetterUpper dir_maker_;
+};
+
+scoped_ptr<sync_pb::SyncEntity> SyncDirectoryUpdateHandlerTest::CreateUpdate(
+    const std::string& id,
+    const std::string& parent,
+    const ModelType& type) {
+  scoped_ptr<sync_pb::SyncEntity> e(new sync_pb::SyncEntity());
+  e->set_id_string(id);
+  e->set_parent_id_string(parent);
+  e->set_non_unique_name(id);
+  e->set_name(id);
+  e->set_version(1000);
+  AddDefaultFieldValue(type, e->mutable_specifics());
+  return e.Pass();
+}
+
+void SyncDirectoryUpdateHandlerTest::UpdateSyncEntities(
+    SyncDirectoryUpdateHandler* handler,
+    const SyncEntityList& applicable_updates,
+    sessions::StatusController* status) {
+  syncable::ModelNeutralWriteTransaction trans(FROM_HERE, UNITTEST, dir());
+  handler->UpdateSyncEntities(&trans, applicable_updates, status);
+}
+
+void SyncDirectoryUpdateHandlerTest::UpdateProgressMarkers(
+    SyncDirectoryUpdateHandler* handler,
+    const sync_pb::DataTypeProgressMarker& progress) {
+  handler->UpdateProgressMarker(progress);
+}
+
+static const char kCacheGuid[] = "IrcjZ2jyzHDV9Io4+zKcXQ==";
+
+// Test that the bookmark tag is set on newly downloaded items.
+TEST_F(SyncDirectoryUpdateHandlerTest, NewBookmarkTag) {
+  SyncDirectoryUpdateHandler handler(dir(), BOOKMARKS);
+  sync_pb::GetUpdatesResponse gu_response;
+  sessions::StatusController status;
+
+  // Add a bookmark item to the update message.
+  std::string root = syncable::GetNullId().GetServerId();
+  syncable::Id server_id = syncable::Id::CreateFromServerId("b1");
+  scoped_ptr<sync_pb::SyncEntity> e =
+      CreateUpdate(SyncableIdToProto(server_id), root, BOOKMARKS);
+  e->set_originator_cache_guid(
+      std::string(kCacheGuid, arraysize(kCacheGuid)-1));
+  syncable::Id client_id = syncable::Id::CreateFromClientString("-2");
+  e->set_originator_client_item_id(client_id.GetServerId());
+  e->set_position_in_parent(0);
+
+  // Add it to the applicable updates list.
+  SyncEntityList bookmark_updates;
+  bookmark_updates.push_back(e.get());
+
+  // Process the update.
+  UpdateSyncEntities(&handler, bookmark_updates, &status);
+
+  syncable::ReadTransaction trans(FROM_HERE, dir());
+  syncable::Entry entry(&trans, syncable::GET_BY_ID, server_id);
+  ASSERT_TRUE(entry.good());
+  EXPECT_TRUE(UniquePosition::IsValidSuffix(entry.GetUniqueBookmarkTag()));
+  EXPECT_TRUE(entry.GetServerUniquePosition().IsValid());
+
+  // If this assertion fails, that might indicate that the algorithm used to
+  // generate bookmark tags has been modified.  This could have implications for
+  // bookmark ordering.  Please make sure you know what you're doing if you
+  // intend to make such a change.
+  EXPECT_EQ("6wHRAb3kbnXV5GHrejp4/c1y5tw=", entry.GetUniqueBookmarkTag());
+}
+
+// Test the receipt of a type root node.
+TEST_F(SyncDirectoryUpdateHandlerTest, ReceiveServerCreatedBookmarkFolders) {
+  SyncDirectoryUpdateHandler handler(dir(), BOOKMARKS);
+  sync_pb::GetUpdatesResponse gu_response;
+  sessions::StatusController status;
+
+  // Create an update that mimics the bookmark root.
+  syncable::Id server_id = syncable::Id::CreateFromServerId("xyz");
+  std::string root = syncable::GetNullId().GetServerId();
+  scoped_ptr<sync_pb::SyncEntity> e =
+      CreateUpdate(SyncableIdToProto(server_id), root, BOOKMARKS);
+  e->set_server_defined_unique_tag("google_chrome_bookmarks");
+  e->set_folder(true);
+
+  // Add it to the applicable updates list.
+  SyncEntityList bookmark_updates;
+  bookmark_updates.push_back(e.get());
+
+  EXPECT_FALSE(SyncerProtoUtil::ShouldMaintainPosition(*e));
+
+  // Process it.
+  UpdateSyncEntities(&handler, bookmark_updates, &status);
+
+  // Verify the results.
+  syncable::ReadTransaction trans(FROM_HERE, dir());
+  syncable::Entry entry(&trans, syncable::GET_BY_ID, server_id);
+  ASSERT_TRUE(entry.good());
+
+  EXPECT_FALSE(entry.ShouldMaintainPosition());
+  EXPECT_FALSE(entry.GetUniquePosition().IsValid());
+  EXPECT_FALSE(entry.GetServerUniquePosition().IsValid());
+  EXPECT_TRUE(entry.GetUniqueBookmarkTag().empty());
+}
+
+// Test the receipt of a non-bookmark item.
+TEST_F(SyncDirectoryUpdateHandlerTest, ReceiveNonBookmarkItem) {
+  SyncDirectoryUpdateHandler handler(dir(), AUTOFILL);
+  sync_pb::GetUpdatesResponse gu_response;
+  sessions::StatusController status;
+
+  std::string root = syncable::GetNullId().GetServerId();
+  syncable::Id server_id = syncable::Id::CreateFromServerId("xyz");
+  scoped_ptr<sync_pb::SyncEntity> e =
+      CreateUpdate(SyncableIdToProto(server_id), root, AUTOFILL);
+  e->set_server_defined_unique_tag("9PGRuKdX5sHyGMB17CvYTXuC43I=");
+
+  // Add it to the applicable updates list.
+  SyncEntityList autofill_updates;
+  autofill_updates.push_back(e.get());
+
+  EXPECT_FALSE(SyncerProtoUtil::ShouldMaintainPosition(*e));
+
+  // Process it.
+  UpdateSyncEntities(&handler, autofill_updates, &status);
+
+  syncable::ReadTransaction trans(FROM_HERE, dir());
+  syncable::Entry entry(&trans, syncable::GET_BY_ID, server_id);
+  ASSERT_TRUE(entry.good());
+
+  EXPECT_FALSE(entry.ShouldMaintainPosition());
+  EXPECT_FALSE(entry.GetUniquePosition().IsValid());
+  EXPECT_FALSE(entry.GetServerUniquePosition().IsValid());
+  EXPECT_TRUE(entry.GetUniqueBookmarkTag().empty());
+}
+
+// Tests the setting of progress markers.
+TEST_F(SyncDirectoryUpdateHandlerTest, ProcessNewProgressMarkers) {
+  SyncDirectoryUpdateHandler handler(dir(), BOOKMARKS);
+
+  sync_pb::DataTypeProgressMarker progress;
+  progress.set_data_type_id(GetSpecificsFieldNumberFromModelType(BOOKMARKS));
+  progress.set_token("token");
+
+  UpdateProgressMarkers(&handler, progress);
+
+  sync_pb::DataTypeProgressMarker saved;
+  dir()->GetDownloadProgress(BOOKMARKS, &saved);
+
+  EXPECT_EQ(progress.token(), saved.token());
+  EXPECT_EQ(progress.data_type_id(), saved.data_type_id());
+}
+
+}  // namespace syncer
diff --git a/sync/engine/syncer.cc b/sync/engine/syncer.cc
index 9f2c378..50cc167 100644
--- a/sync/engine/syncer.cc
+++ b/sync/engine/syncer.cc
@@ -61,6 +61,7 @@ bool Syncer::NormalSyncShare(ModelTypeSet request_types,
   if (nudge_tracker.IsGetUpdatesRequired() ||
       session->context()->ShouldFetchUpdatesBeforeCommit()) {
     if (!DownloadAndApplyUpdates(
+            request_types,
             session,
             base::Bind(&BuildNormalDownloadUpdates,
                        session,
@@ -85,6 +86,7 @@ bool Syncer::ConfigureSyncShare(
   HandleCycleBegin(session);
   VLOG(1) << "Configuring types " << ModelTypeSetToString(request_types);
   DownloadAndApplyUpdates(
+      request_types,
       session,
       base::Bind(&BuildDownloadUpdatesForConfigure,
                  session,
@@ -99,6 +101,7 @@ bool Syncer::PollSyncShare(ModelTypeSet request_types,
   HandleCycleBegin(session);
   VLOG(1) << "Polling types " << ModelTypeSetToString(request_types);
   DownloadAndApplyUpdates(
+      request_types,
       session,
       base::Bind(&BuildDownloadUpdatesForPoll,
                  session,
@@ -122,13 +125,15 @@ void Syncer::ApplyUpdates(SyncSession* session) {
 }
 
 bool Syncer::DownloadAndApplyUpdates(
+    ModelTypeSet request_types,
     SyncSession* session,
     base::Callback<void(sync_pb::ClientToServerMessage*)> build_fn) {
   while (!session->status_controller().ServerSaysNothingMoreToDownload()) {
     TRACE_EVENT0("sync", "DownloadUpdates");
     sync_pb::ClientToServerMessage msg;
     build_fn.Run(&msg);
-    SyncerError download_result = ExecuteDownloadUpdates(session, &msg);
+    SyncerError download_result =
+        ExecuteDownloadUpdates(request_types, session, &msg);
     session->mutable_status_controller()->set_last_download_updates_result(
         download_result);
     if (download_result != SYNCER_OK) {
diff --git a/sync/engine/syncer.h b/sync/engine/syncer.h
index d943e5f..5661fff 100644
--- a/sync/engine/syncer.h
+++ b/sync/engine/syncer.h
@@ -70,6 +70,7 @@ class SYNC_EXPORT_PRIVATE Syncer {
  private:
   void ApplyUpdates(sessions::SyncSession* session);
   bool DownloadAndApplyUpdates(
+      ModelTypeSet request_types,
       sessions::SyncSession* session,
       base::Callback<void(sync_pb::ClientToServerMessage*)> build_fn);
 
diff --git a/sync/engine/syncer_unittest.cc b/sync/engine/syncer_unittest.cc
index a7825dc..b9c7780 100644
--- a/sync/engine/syncer_unittest.cc
+++ b/sync/engine/syncer_unittest.cc
@@ -25,7 +25,6 @@
 #include "build/build_config.h"
 #include "sync/engine/get_commit_ids.h"
 #include "sync/engine/net/server_connection_manager.h"
-#include "sync/engine/process_updates_command.h"
 #include "sync/engine/sync_scheduler_impl.h"
 #include "sync/engine/syncer.h"
 #include "sync/engine/syncer_proto_util.h"