[Sync] USS: Load data for pending commits on startup.

This change also includes fairly massive changes to how the SMTP test
file works. Specifically, there is now a SimpleStore for data and
metadata, and the tests have been modified to assert against the stores.
This was done to allow for more intuitive, flexible testing.

Other minor changes:

- Cleaned up some unnecessary functions in SMTP and ModelTypeEntity.
- Added helper functions for accessing ModelTypeEntity objects.
- Removed FakeMetadataChangeList. Simple* is sufficient for testing.
- Switch to using map[k] = v over map.insert(make_pair(k, v)) as the
  former looks cleaner and does not silently fail if a value for k
  already exists in the map.

BUG=569642

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

Cr-Commit-Position: refs/heads/master@{#377047}

11 files changed, 774 insertions, 574 deletions

diff --git a/sync/BUILD.gn b/sync/BUILD.gn
index 9cd103d..dfa0fae 100644
--- a/sync/BUILD.gn
+++ b/sync/BUILD.gn
@@ -554,7 +554,6 @@ static_library("test_support_sync_core") {
 static_library("test_support_sync_internal_api") {
   testonly = true
   sources = [
-    "internal_api/public/test/fake_metadata_change_list.h",
     "internal_api/public/test/fake_model_type_service.h",
     "internal_api/public/test/fake_sync_manager.h",
     "internal_api/public/test/model_type_store_test_util.h",
@@ -563,7 +562,6 @@ static_library("test_support_sync_internal_api") {
     "internal_api/public/test/test_entry_factory.h",
     "internal_api/public/test/test_internal_components_factory.h",
     "internal_api/public/test/test_user_share.h",
-    "internal_api/test/fake_metadata_change_list.cc",
     "internal_api/test/fake_model_type_service.cc",
     "internal_api/test/fake_sync_manager.cc",
     "internal_api/test/model_type_store_test_util.cc",
diff --git a/sync/internal_api/model_type_entity.cc b/sync/internal_api/model_type_entity.cc
index 99e1062..19be3e6 100644
--- a/sync/internal_api/model_type_entity.cc
+++ b/sync/internal_api/model_type_entity.cc
@@ -51,7 +51,12 @@ ModelTypeEntity::ModelTypeEntity(const std::string& client_tag,
 ModelTypeEntity::~ModelTypeEntity() {}
 
 void ModelTypeEntity::CacheCommitData(EntityData* data) {
+  DCHECK(RequiresCommitRequest());
+  if (data->client_tag_hash.empty()) {
+    data->client_tag_hash = metadata_.client_tag_hash();
+  }
   commit_data_ = data->PassToPtr();
+  DCHECK(HasCommitData());
 }
 
 bool ModelTypeEntity::HasCommitData() const {
@@ -66,6 +71,10 @@ bool ModelTypeEntity::RequiresCommitRequest() const {
   return metadata_.sequence_number() > commit_requested_sequence_number_;
 }
 
+bool ModelTypeEntity::RequiresCommitData() const {
+  return RequiresCommitRequest() && !HasCommitData() && !metadata_.is_deleted();
+}
+
 bool ModelTypeEntity::UpdateIsReflection(int64_t update_version) const {
   return metadata_.server_version() >= update_version;
 }
@@ -137,26 +146,27 @@ void ModelTypeEntity::Delete() {
   IncrementSequenceNumber();
   metadata_.set_is_deleted(true);
   metadata_.clear_specifics_hash();
-
-  EntityData data;
-  data.client_tag_hash = metadata_.client_tag_hash();
-  data.id = metadata_.server_id();
-  data.creation_time = syncer::ProtoTimeToTime(metadata_.creation_time());
-
-  CacheCommitData(&data);
+  // Clear any cached pending commit data.
+  if (HasCommitData()) commit_data_.reset();
 }
 
-void ModelTypeEntity::InitializeCommitRequestData(
-    CommitRequestData* request) const {
-  DCHECK(HasCommitData());
-  DCHECK_EQ(commit_data_->client_tag_hash, metadata_.client_tag_hash());
+void ModelTypeEntity::InitializeCommitRequestData(CommitRequestData* request) {
+  if (!metadata_.is_deleted()) {
+    DCHECK(HasCommitData());
+    DCHECK_EQ(commit_data_->client_tag_hash, metadata_.client_tag_hash());
+    request->entity = commit_data_;
+  } else {
+    // Make an EntityData with empty specifics to indicate deletion. This is
+    // done lazily here to simplify loading a pending deletion on startup.
+    EntityData data;
+    data.client_tag_hash = metadata_.client_tag_hash();
+    data.id = metadata_.server_id();
+    data.creation_time = syncer::ProtoTimeToTime(metadata_.creation_time());
+    request->entity = data.PassToPtr();
+  }
 
   request->sequence_number = metadata_.sequence_number();
   request->base_version = metadata_.server_version();
-  request->entity = commit_data_;
-}
-
-void ModelTypeEntity::SetCommitRequestInProgress() {
   commit_requested_sequence_number_ = metadata_.sequence_number();
 }
 
@@ -178,19 +188,6 @@ void ModelTypeEntity::ClearTransientSyncState() {
   commit_requested_sequence_number_ = metadata_.acked_sequence_number();
 }
 
-void ModelTypeEntity::ClearSyncState() {
-  // TODO(stanisc): crbug/561830: Need to review this entire method. It looks
-  // like the tests expect this to reset some metadata state but not the data.
-  // We should be able to reimplement this once we have the code fore
-  // fetching the data from the service.
-  metadata_.set_server_version(kUncommittedVersion);
-  // TODO(stanisc): Why is this 1 and not 0? This leaves the item unsynced.
-  metadata_.set_sequence_number(1);
-  metadata_.set_acked_sequence_number(0);
-  metadata_.clear_server_id();
-  commit_requested_sequence_number_ = 0;
-}
-
 void ModelTypeEntity::IncrementSequenceNumber() {
   DCHECK(metadata_.has_sequence_number());
   metadata_.set_sequence_number(metadata_.sequence_number() + 1);
diff --git a/sync/internal_api/model_type_entity_unittest.cc b/sync/internal_api/model_type_entity_unittest.cc
index 0f539e5..c1d0035 100644
--- a/sync/internal_api/model_type_entity_unittest.cc
+++ b/sync/internal_api/model_type_entity_unittest.cc
@@ -228,7 +228,8 @@ TEST_F(ModelTypeEntityTest, LocalDeletion) {
   entity->Delete();
   EXPECT_FALSE(HasSpecificsHash(entity));
 
-  EXPECT_TRUE(entity->HasCommitData());
+  EXPECT_FALSE(entity->HasCommitData());
+  EXPECT_FALSE(entity->RequiresCommitData());
   EXPECT_TRUE(entity->IsUnsynced());
 
   EXPECT_TRUE(entity->UpdateIsReflection(10));
diff --git a/sync/internal_api/public/model_type_entity.h b/sync/internal_api/public/model_type_entity.h
index 23a32ac..f8204a2 100644
--- a/sync/internal_api/public/model_type_entity.h
+++ b/sync/internal_api/public/model_type_entity.h
@@ -60,6 +60,10 @@ class SYNC_EXPORT ModelTypeEntity {
   // this item.
   bool RequiresCommitRequest() const;
 
+  // Whether commit data is needed to be cached before a commit request can be
+  // created. Note that deletions do not require cached data.
+  bool RequiresCommitData() const;
+
   // Returns true if the specified update version does not contain new data.
   bool UpdateIsReflection(int64_t update_version) const;
 
@@ -85,11 +89,8 @@ class SYNC_EXPORT ModelTypeEntity {
   void Delete();
 
   // Initializes a message representing this item's uncommitted state
-  // to be forwarded to the sync server for committing.
-  void InitializeCommitRequestData(CommitRequestData* request) const;
-
-  // Notes that the current version of this item has been queued for commit.
-  void SetCommitRequestInProgress();
+  // and assumes that it is forwarded to the sync engine for commiting.
+  void InitializeCommitRequestData(CommitRequestData* request);
 
   // Receives a successful commit response.
   //
@@ -107,9 +108,6 @@ class SYNC_EXPORT ModelTypeEntity {
   // Clears any in-memory sync state associated with outstanding commits.
   void ClearTransientSyncState();
 
-  // Clears all sync state.  Invoked when a user signs out.
-  void ClearSyncState();
-
   // Takes the passed commit data and caches it in the instance.
   // The data is swapped from the input struct without copying.
   void CacheCommitData(EntityData* data);
diff --git a/sync/internal_api/public/model_type_processor.h b/sync/internal_api/public/model_type_processor.h
index 4ab4946..2f1c724 100644
--- a/sync/internal_api/public/model_type_processor.h
+++ b/sync/internal_api/public/model_type_processor.h
@@ -35,7 +35,7 @@ class SYNC_EXPORT ModelTypeProcessor {
   // handle.
   virtual void OnUpdateReceived(
       const sync_pb::DataTypeState& type_state,
-      const UpdateResponseDataList& response_list,
+      const UpdateResponseDataList& updates,
       const UpdateResponseDataList& pending_updates) = 0;
 };
 
diff --git a/sync/internal_api/public/shared_model_type_processor.h b/sync/internal_api/public/shared_model_type_processor.h
index 606a22b..782cf8c 100644
--- a/sync/internal_api/public/shared_model_type_processor.h
+++ b/sync/internal_api/public/shared_model_type_processor.h
@@ -11,6 +11,7 @@
 #include "base/memory/scoped_ptr.h"
 #include "base/memory/weak_ptr.h"
 #include "base/threading/non_thread_safe.h"
+#include "sync/api/data_batch.h"
 #include "sync/api/metadata_batch.h"
 #include "sync/api/metadata_change_list.h"
 #include "sync/api/model_type_change_processor.h"
@@ -89,7 +90,7 @@ class SYNC_EXPORT SharedModelTypeProcessor : public ModelTypeProcessor,
   void OnCommitCompleted(const sync_pb::DataTypeState& type_state,
                          const CommitResponseDataList& response_list) override;
   void OnUpdateReceived(const sync_pb::DataTypeState& type_state,
-                        const UpdateResponseDataList& response_list,
+                        const UpdateResponseDataList& updates,
                         const UpdateResponseDataList& pending_updates) override;
 
  private:
@@ -98,21 +99,29 @@ class SYNC_EXPORT SharedModelTypeProcessor : public ModelTypeProcessor,
   using EntityMap = std::map<std::string, scoped_ptr<ModelTypeEntity>>;
   using UpdateMap = std::map<std::string, scoped_ptr<UpdateResponseData>>;
 
-  // Complete the start process.
-  void ReadyToConnect();
+  // Callback for ModelTypeService::GetData(). Used when we need to load data
+  // for pending commits during the initialization process.
+  void OnDataLoaded(syncer::SyncError error, scoped_ptr<DataBatch> data_batch);
+
+  // Check conditions, and if met inform sync that we are ready to connect.
+  void ConnectIfReady();
 
   // Handle the first update received from the server after being enabled.
   void OnInitialUpdateReceived(const sync_pb::DataTypeState& type_state,
-                               const UpdateResponseDataList& response_list,
+                               const UpdateResponseDataList& updates,
                                const UpdateResponseDataList& pending_updates);
 
   // Sends all commit requests that are due to be sent to the sync thread.
   void FlushPendingCommitRequests();
 
-  // Clears any state related to outstanding communications with the
-  // CommitQueue.  Used when we want to disconnect from
-  // the current worker.
-  void ClearTransientSyncState();
+  // Computes the client tag hash for the given client |tag|.
+  std::string GetHashForTag(const std::string& tag);
+
+  // Gets the entity for the given tag, which must exist.
+  ModelTypeEntity* GetEntityForTag(const std::string& tag);
+
+  // Gets the entity for the given tag hash, which must exist.
+  ModelTypeEntity* GetEntityForTagHash(const std::string& tag_hash);
 
   syncer::ModelType type_;
   sync_pb::DataTypeState data_type_state_;
@@ -123,6 +132,9 @@ class SYNC_EXPORT SharedModelTypeProcessor : public ModelTypeProcessor,
   // Indicates whether the metadata has finished loading.
   bool is_metadata_loaded_;
 
+  // Indicates whether data for pending commits has finished loading.
+  bool is_pending_commit_data_loaded_;
+
   // Reference to the CommitQueue.
   //
   // The interface hides the posting of tasks across threads as well as the
@@ -148,7 +160,7 @@ class SYNC_EXPORT SharedModelTypeProcessor : public ModelTypeProcessor,
   // thread, we want to make sure that no tasks generated as part of the
   // now-obsolete connection to affect us.  But we also want the WeakPtr we
   // sent to the UI thread to remain valid.
-  base::WeakPtrFactory<SharedModelTypeProcessor> weak_ptr_factory_for_ui_;
+  base::WeakPtrFactory<SharedModelTypeProcessor> weak_ptr_factory_;
   base::WeakPtrFactory<SharedModelTypeProcessor> weak_ptr_factory_for_sync_;
 };
 
diff --git a/sync/internal_api/public/test/fake_metadata_change_list.h b/sync/internal_api/public/test/fake_metadata_change_list.h
deleted file mode 100644
index 3941f0a..0000000
--- a/sync/internal_api/public/test/fake_metadata_change_list.h
+++ /dev/null
@@ -1,60 +0,0 @@
-// Copyright 2016 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_INTERNAL_API_PUBLIC_TEST_FAKE_METADATA_CHANGE_LIST_H_
-#define SYNC_INTERNAL_API_PUBLIC_TEST_FAKE_METADATA_CHANGE_LIST_H_
-
-#include <string>
-#include <vector>
-
-#include "sync/api/metadata_change_list.h"
-#include "sync/internal_api/public/non_blocking_sync_common.h"
-#include "sync/protocol/data_type_state.pb.h"
-#include "sync/protocol/entity_metadata.pb.h"
-
-namespace syncer_v2 {
-
-// A non-functional implementation of MetadataChangeList for
-// testing purposes.
-// This class simply records all calls with all arguments for further
-// analysis by the test code.
-class FakeMetadataChangeList : public MetadataChangeList {
- public:
-  FakeMetadataChangeList();
-  ~FakeMetadataChangeList() override;
-
-  void UpdateDataTypeState(
-      const sync_pb::DataTypeState& data_type_state) override;
-  void ClearDataTypeState() override;
-  void UpdateMetadata(const std::string& client_tag,
-                      const sync_pb::EntityMetadata& metadata) override;
-  void ClearMetadata(const std::string& client_tag) override;
-
-  enum Action {
-    UPDATE_DATA_TYPE_STATE,
-    CLEAR_DATA_TYPE_STATE,
-    UPDATE_METADATA,
-    CLEAR_METADATA
-  };
-
-  struct Record {
-    Record();
-    virtual ~Record();
-
-    Action action;
-    std::string tag;
-    sync_pb::DataTypeState data_type_state;
-    sync_pb::EntityMetadata metadata;
-  };
-
-  size_t GetNumRecords() const;
-  const Record& GetNthRecord(size_t n) const;
-
- private:
-  std::vector<Record> records_;
-};
-
-}  // namespace syncer_v2
-
-#endif  // SYNC_INTERNAL_API_PUBLIC_TEST_FAKE_METADATA_CHANGE_LIST_H_
diff --git a/sync/internal_api/shared_model_type_processor.cc b/sync/internal_api/shared_model_type_processor.cc
index 83a6a51..65e9924 100644
--- a/sync/internal_api/shared_model_type_processor.cc
+++ b/sync/internal_api/shared_model_type_processor.cc
@@ -5,6 +5,7 @@
 #include "sync/internal_api/public/shared_model_type_processor.h"
 
 #include <utility>
+#include <vector>
 
 #include "base/bind.h"
 #include "base/location.h"
@@ -29,7 +30,7 @@ class ModelTypeProcessorProxy : public ModelTypeProcessor {
   void OnCommitCompleted(const sync_pb::DataTypeState& type_state,
                          const CommitResponseDataList& response_list) override;
   void OnUpdateReceived(const sync_pb::DataTypeState& type_state,
-                        const UpdateResponseDataList& response_list,
+                        const UpdateResponseDataList& updates,
                         const UpdateResponseDataList& pending_updates) override;
 
  private:
@@ -60,11 +61,11 @@ void ModelTypeProcessorProxy::OnCommitCompleted(
 
 void ModelTypeProcessorProxy::OnUpdateReceived(
     const sync_pb::DataTypeState& type_state,
-    const UpdateResponseDataList& response_list,
+    const UpdateResponseDataList& updates,
     const UpdateResponseDataList& pending_updates) {
   processor_task_runner_->PostTask(
       FROM_HERE, base::Bind(&ModelTypeProcessor::OnUpdateReceived, processor_,
-                            type_state, response_list, pending_updates));
+                            type_state, updates, pending_updates));
 }
 
 }  // namespace
@@ -74,7 +75,7 @@ SharedModelTypeProcessor::SharedModelTypeProcessor(syncer::ModelType type,
     : type_(type),
       is_metadata_loaded_(false),
       service_(service),
-      weak_ptr_factory_for_ui_(this),
+      weak_ptr_factory_(this),
       weak_ptr_factory_for_sync_(this) {
   DCHECK(service);
 }
@@ -88,30 +89,38 @@ void SharedModelTypeProcessor::OnSyncStarting(StartCallback start_callback) {
   DVLOG(1) << "Sync is starting for " << ModelTypeToString(type_);
 
   start_callback_ = start_callback;
-
-  if (is_metadata_loaded_) {
-    // The metadata was already loaded, so we are ready to connect.
-    ReadyToConnect();
-  }
+  ConnectIfReady();
 }
 
 void SharedModelTypeProcessor::OnMetadataLoaded(
     scoped_ptr<MetadataBatch> batch) {
   DCHECK(CalledOnValidThread());
   DCHECK(entities_.empty());
+  DCHECK(!is_metadata_loaded_);
   DCHECK(!IsConnected());
 
+  // Flip this flag here to cover all cases where we don't need to load data.
+  is_pending_commit_data_loaded_ = true;
+
   if (batch->GetDataTypeState().initial_sync_done()) {
     EntityMetadataMap metadata_map(batch->TakeAllMetadata());
+    std::vector<std::string> entities_to_commit;
+
     for (auto it = metadata_map.begin(); it != metadata_map.end(); it++) {
-      entities_.insert(std::make_pair(
-          it->second.client_tag_hash(),
-          ModelTypeEntity::CreateFromMetadata(it->first, &it->second)));
+      scoped_ptr<ModelTypeEntity> entity =
+          ModelTypeEntity::CreateFromMetadata(it->first, &it->second);
+      if (entity->RequiresCommitData()) {
+        entities_to_commit.push_back(entity->client_tag());
+      }
+      entities_[entity->metadata().client_tag_hash()] = std::move(entity);
     }
     data_type_state_ = batch->GetDataTypeState();
-    // TODO(maxbogue): crbug.com/569642: Load data for pending commits.
-    // TODO(maxbogue): crbug.com/569642: Check for inconsistent state where
-    // we have data but no metadata?
+    if (!entities_to_commit.empty()) {
+      is_pending_commit_data_loaded_ = false;
+      service_->GetData(entities_to_commit,
+                        base::Bind(&SharedModelTypeProcessor::OnDataLoaded,
+                                   weak_ptr_factory_.GetWeakPtr()));
+    }
   } else {
     // First time syncing; initialize metadata.
     data_type_state_.mutable_progress_marker()->set_data_type_id(
@@ -119,17 +128,28 @@ void SharedModelTypeProcessor::OnMetadataLoaded(
   }
 
   is_metadata_loaded_ = true;
+  ConnectIfReady();
+}
 
-  if (!start_callback_.is_null()) {
-    // If OnSyncStarting() was already called, we are now ready to connect.
-    ReadyToConnect();
+void SharedModelTypeProcessor::OnDataLoaded(syncer::SyncError error,
+                                            scoped_ptr<DataBatch> data_batch) {
+  while (data_batch->HasNext()) {
+    TagAndData data = data_batch->Next();
+    ModelTypeEntity* entity = GetEntityForTag(data.first);
+    // If the entity wasn't deleted or updated with new commit.
+    if (entity != nullptr && entity->RequiresCommitData()) {
+      entity->CacheCommitData(data.second.get());
+    }
   }
+  is_pending_commit_data_loaded_ = true;
+  FlushPendingCommitRequests();
 }
 
-void SharedModelTypeProcessor::ReadyToConnect() {
+void SharedModelTypeProcessor::ConnectIfReady() {
   DCHECK(CalledOnValidThread());
-  DCHECK(is_metadata_loaded_);
-  DCHECK(!start_callback_.is_null());
+  if (!is_metadata_loaded_ || start_callback_.is_null()) {
+    return;
+  }
 
   scoped_ptr<ActivationContext> activation_context =
       make_scoped_ptr(new ActivationContext);
@@ -176,13 +196,15 @@ void SharedModelTypeProcessor::DisconnectSync() {
   weak_ptr_factory_for_sync_.InvalidateWeakPtrs();
   worker_.reset();
 
-  ClearTransientSyncState();
+  for (auto it = entities_.begin(); it != entities_.end(); ++it) {
+    it->second->ClearTransientSyncState();
+  }
 }
 
 base::WeakPtr<SharedModelTypeProcessor>
 SharedModelTypeProcessor::AsWeakPtrForUI() {
   DCHECK(CalledOnValidThread());
-  return weak_ptr_factory_for_ui_.GetWeakPtr();
+  return weak_ptr_factory_.GetWeakPtr();
 }
 
 void SharedModelTypeProcessor::ConnectSync(scoped_ptr<CommitQueue> worker) {
@@ -214,20 +236,17 @@ void SharedModelTypeProcessor::Put(const std::string& client_tag,
 
   base::Time now = base::Time::Now();
 
-  ModelTypeEntity* entity = nullptr;
   // TODO(stanisc): crbug.com/561818: Search by client_tag rather than
   // client_tag_hash.
-  auto it = entities_.find(client_tag_hash);
-  if (it == entities_.end()) {
+  ModelTypeEntity* entity = GetEntityForTagHash(client_tag_hash);
+  if (entity == nullptr) {
     // The service is creating a new entity.
     scoped_ptr<ModelTypeEntity> scoped_entity = ModelTypeEntity::CreateNew(
         client_tag, client_tag_hash, entity_data->id, now);
     entity = scoped_entity.get();
-    entities_.insert(
-        std::make_pair(client_tag_hash, std::move(scoped_entity)));
+    entities_[client_tag_hash] = std::move(scoped_entity);
   } else {
     // The service is updating an existing entity.
-    entity = it->second.get();
     DCHECK_EQ(client_tag, entity->client_tag());
   }
 
@@ -249,8 +268,8 @@ void SharedModelTypeProcessor::Delete(
 
   // TODO(skym): crbug.com/561818: Search by client_tag rather than
   // client_tag_hash.
-  auto it = entities_.find(client_tag_hash);
-  if (it == entities_.end()) {
+  ModelTypeEntity* entity = GetEntityForTagHash(client_tag_hash);
+  if (entity == nullptr) {
     // That's unusual, but not necessarily a bad thing.
     // Missing is as good as deleted as far as the model is concerned.
     DLOG(WARNING) << "Attempted to delete missing item."
@@ -258,7 +277,6 @@ void SharedModelTypeProcessor::Delete(
     return;
   }
 
-  ModelTypeEntity* entity = it->second.get();
   entity->Delete();
 
   metadata_change_list->UpdateMetadata(client_tag, entity->metadata());
@@ -277,13 +295,18 @@ void SharedModelTypeProcessor::FlushPendingCommitRequests() {
   if (!data_type_state_.initial_sync_done())
     return;
 
+  // Dont send anything if the initial data load is incomplete.
+  if (!is_pending_commit_data_loaded_)
+    return;
+
   // TODO(rlarocque): Do something smarter than iterate here.
   for (auto it = entities_.begin(); it != entities_.end(); ++it) {
-    if (it->second->RequiresCommitRequest()) {
+    ModelTypeEntity* entity = it->second.get();
+    if (entity->RequiresCommitRequest()) {
+      DCHECK(!entity->RequiresCommitData());
       CommitRequestData request;
-      it->second->InitializeCommitRequestData(&request);
+      entity->InitializeCommitRequestData(&request);
       commit_requests.push_back(request);
-      it->second->SetCommitRequestInProgress();
     }
   }
 
@@ -300,41 +323,34 @@ void SharedModelTypeProcessor::OnCommitCompleted(
   data_type_state_ = type_state;
   change_list->UpdateDataTypeState(data_type_state_);
 
-  for (auto list_it = response_list.begin(); list_it != response_list.end();
-       ++list_it) {
-    const CommitResponseData& response_data = *list_it;
-    const std::string& client_tag_hash = response_data.client_tag_hash;
-
-    auto it = entities_.find(client_tag_hash);
-    if (it == entities_.end()) {
+  for (const CommitResponseData& data : response_list) {
+    ModelTypeEntity* entity = GetEntityForTagHash(data.client_tag_hash);
+    if (entity == nullptr) {
       NOTREACHED() << "Received commit response for missing item."
-                   << " type: " << type_ << " client_tag: " << client_tag_hash;
-      return;
-    } else {
-      it->second->ReceiveCommitResponse(response_data.id,
-                                        response_data.sequence_number,
-                                        response_data.response_version,
-                                        data_type_state_.encryption_key_name());
-      // TODO(stanisc): crbug.com/573333: Delete case.
-      // This might be the right place to clear a metadata entry that has
-      // been deleted locally and confirmed deleted by the server.
-      change_list->UpdateMetadata(it->second->client_tag(),
-                                  it->second->metadata());
+                   << " type: " << type_
+                   << " client_tag_hash: " << data.client_tag_hash;
+      continue;
     }
+
+    entity->ReceiveCommitResponse(data.id, data.sequence_number,
+                                  data.response_version,
+                                  data_type_state_.encryption_key_name());
+    // TODO(stanisc): crbug.com/573333: Delete case.
+    // This might be the right place to clear a metadata entry that has
+    // been deleted locally and confirmed deleted by the server.
+    change_list->UpdateMetadata(entity->client_tag(), entity->metadata());
   }
 
-  // TODO(stanisc): What is the right method to submit metadata changes to the
-  // service? Is using empty EntityChangeList OK?
   // TODO(stanisc): crbug.com/570085: Error handling.
   service_->ApplySyncChanges(std::move(change_list), EntityChangeList());
 }
 
 void SharedModelTypeProcessor::OnUpdateReceived(
     const sync_pb::DataTypeState& data_type_state,
-    const UpdateResponseDataList& response_list,
+    const UpdateResponseDataList& updates,
     const UpdateResponseDataList& pending_updates) {
   if (!data_type_state_.initial_sync_done()) {
-    OnInitialUpdateReceived(data_type_state, response_list, pending_updates);
+    OnInitialUpdateReceived(data_type_state, updates, pending_updates);
   }
 
   scoped_ptr<MetadataChangeList> metadata_changes =
@@ -347,19 +363,16 @@ void SharedModelTypeProcessor::OnUpdateReceived(
       data_type_state.encryption_key_name();
   data_type_state_ = data_type_state;
 
-  for (auto list_it = response_list.begin(); list_it != response_list.end();
-       ++list_it) {
-    const UpdateResponseData& response_data = *list_it;
-    const EntityData& data = response_data.entity.value();
+  for (const UpdateResponseData& update : updates) {
+    const EntityData& data = update.entity.value();
     const std::string& client_tag_hash = data.client_tag_hash;
 
     // If we're being asked to apply an update to this entity, this overrides
     // the previous pending updates.
     pending_updates_map_.erase(client_tag_hash);
 
-    ModelTypeEntity* entity = nullptr;
-    auto it = entities_.find(client_tag_hash);
-    if (it == entities_.end()) {
+    ModelTypeEntity* entity = GetEntityForTagHash(client_tag_hash);
+    if (entity == nullptr) {
       if (data.is_deleted()) {
         DLOG(WARNING) << "Received remote delete for a non-existing item."
                       << " client_tag_hash: " << client_tag_hash;
@@ -367,31 +380,29 @@ void SharedModelTypeProcessor::OnUpdateReceived(
       }
 
       // Let the service define |client_tag| based on the entity data.
-      std::string client_tag = service_->GetClientTag(data);
+      const std::string client_tag = service_->GetClientTag(data);
 
       scoped_ptr<ModelTypeEntity> scoped_entity = ModelTypeEntity::CreateNew(
           client_tag, client_tag_hash, data.id, data.creation_time);
       entity = scoped_entity.get();
-      entities_.insert(
-          std::make_pair(client_tag_hash, std::move(scoped_entity)));
+      entities_[client_tag_hash] = std::move(scoped_entity);
 
       entity_changes.push_back(
-          EntityChange::CreateAdd(client_tag, response_data.entity));
+          EntityChange::CreateAdd(client_tag, update.entity));
 
     } else {
-      entity = it->second.get();
       if (data.is_deleted()) {
         entity_changes.push_back(
             EntityChange::CreateDelete(entity->client_tag()));
       } else {
         // TODO(stanisc): crbug.com/561829: Avoid sending an update to the
         // service if there is no actual change.
-        entity_changes.push_back(EntityChange::CreateUpdate(
-            entity->client_tag(), response_data.entity));
+        entity_changes.push_back(
+            EntityChange::CreateUpdate(entity->client_tag(), update.entity));
       }
     }
 
-    entity->ApplyUpdateFromServer(response_data);
+    entity->ApplyUpdateFromServer(update);
     // TODO(stanisc): crbug.com/573333: Delete case.
     // This might be the right place to clear metadata entry instead of
     // updating it.
@@ -402,10 +413,9 @@ void SharedModelTypeProcessor::OnUpdateReceived(
 
     // If the received entity has out of date encryption, we schedule another
     // commit to fix it.
-    if (data_type_state_.encryption_key_name() !=
-        response_data.encryption_key_name) {
+    if (data_type_state_.encryption_key_name() != update.encryption_key_name) {
       DVLOG(2) << ModelTypeToString(type_) << ": Requesting re-encrypt commit "
-               << response_data.encryption_key_name << " -> "
+               << update.encryption_key_name << " -> "
                << data_type_state_.encryption_key_name();
       auto it2 = entities_.find(client_tag_hash);
       it2->second->UpdateDesiredEncryptionKey(
@@ -415,9 +425,7 @@ void SharedModelTypeProcessor::OnUpdateReceived(
 
   // TODO(stanisc): crbug.com/529498: stop saving pending updates.
   // Save pending updates in the appropriate data structure.
-  for (auto list_it = pending_updates.begin(); list_it != pending_updates.end();
-       ++list_it) {
-    const UpdateResponseData& update = *list_it;
+  for (const UpdateResponseData& update : pending_updates) {
     const std::string& client_tag_hash = update.entity->client_tag_hash;
 
     auto lookup_it = pending_updates_map_.find(client_tag_hash);
@@ -450,7 +458,7 @@ void SharedModelTypeProcessor::OnUpdateReceived(
 
 void SharedModelTypeProcessor::OnInitialUpdateReceived(
     const sync_pb::DataTypeState& data_type_state,
-    const UpdateResponseDataList& response_list,
+    const UpdateResponseDataList& updates,
     const UpdateResponseDataList& pending_updates) {
   // TODO(maxbogue): crbug.com/569675: Generate metadata for all entities.
   // TODO(maxbogue): crbug.com/569675: Call merge method on the service.
@@ -465,10 +473,19 @@ UpdateResponseDataList SharedModelTypeProcessor::GetPendingUpdates() {
   return pending_updates_list;
 }
 
-void SharedModelTypeProcessor::ClearTransientSyncState() {
-  for (auto it = entities_.begin(); it != entities_.end(); ++it) {
-    it->second->ClearTransientSyncState();
-  }
+std::string SharedModelTypeProcessor::GetHashForTag(const std::string& tag) {
+  return syncer::syncable::GenerateSyncableHash(type_, tag);
+}
+
+ModelTypeEntity* SharedModelTypeProcessor::GetEntityForTag(
+    const std::string& tag) {
+  return GetEntityForTagHash(GetHashForTag(tag));
+}
+
+ModelTypeEntity* SharedModelTypeProcessor::GetEntityForTagHash(
+    const std::string& tag_hash) {
+  auto it = entities_.find(tag_hash);
+  return it != entities_.end() ? it->second.get() : nullptr;
 }
 
 }  // namespace syncer_v2
diff --git a/sync/internal_api/shared_model_type_processor_unittest.cc b/sync/internal_api/shared_model_type_processor_unittest.cc
index 183ff65..a79e5a6 100644
--- a/sync/internal_api/shared_model_type_processor_unittest.cc
+++ b/sync/internal_api/shared_model_type_processor_unittest.cc
@@ -6,15 +6,18 @@
 
 #include <stddef.h>
 #include <stdint.h>
+#include <unordered_map>
 
 #include "base/bind.h"
+#include "base/callback.h"
 #include "base/message_loop/message_loop.h"
 #include "base/run_loop.h"
 #include "sync/engine/commit_queue.h"
 #include "sync/internal_api/public/activation_context.h"
 #include "sync/internal_api/public/base/model_type.h"
+#include "sync/internal_api/public/data_batch_impl.h"
 #include "sync/internal_api/public/non_blocking_sync_common.h"
-#include "sync/internal_api/public/test/fake_metadata_change_list.h"
+#include "sync/internal_api/public/simple_metadata_change_list.h"
 #include "sync/internal_api/public/test/fake_model_type_service.h"
 #include "sync/protocol/data_type_state.pb.h"
 #include "sync/protocol/sync.pb.h"
@@ -27,56 +30,142 @@ namespace syncer_v2 {
 
 static const syncer::ModelType kModelType = syncer::PREFERENCES;
 
-// Tests the sync engine parts of SharedModelTypeProcessor.
-//
-// The SharedModelTypeProcessor contains a non-trivial amount of code dedicated
-// to turning the sync engine on and off again.  That code is fairly well
-// tested in the NonBlockingDataTypeController unit tests and it doesn't need
-// to be re-tested here.
-//
-// These tests skip past initialization and focus on steady state sync engine
-// behavior.  This is where we test how the processor responds to the
-// model's requests to make changes to its data, the messages incoming from the
-// sync server, and what happens when the two conflict.
+namespace {
+
+// It is intentionally very difficult to copy an EntityData, as in normal code
+// we never want to. However, since we store the data as an EntityData for the
+// test code here, this function is needed to manually copy it.
+scoped_ptr<EntityData> CopyEntityData(const EntityData& old_data) {
+  scoped_ptr<EntityData> new_data(new EntityData());
+  new_data->id = old_data.id;
+  new_data->client_tag_hash = old_data.client_tag_hash;
+  new_data->non_unique_name = old_data.non_unique_name;
+  new_data->specifics = old_data.specifics;
+  new_data->creation_time = old_data.creation_time;
+  new_data->modification_time = old_data.modification_time;
+  return new_data;
+}
+
+// A basic in-memory storage mechanism for data and metadata. This makes it
+// easier to test more complex behaviors involving when entities are written,
+// committed, etc. Having a separate class helps keep the main one cleaner.
+class SimpleStore {
+ public:
+  void PutData(const std::string& tag, const EntityData& data) {
+    data_store_[tag] = CopyEntityData(data);
+  }
+
+  void PutMetadata(const std::string& tag,
+                   const sync_pb::EntityMetadata& metadata) {
+    metadata_store_[tag] = metadata;
+  }
+
+  void RemoveData(const std::string& tag) { data_store_.erase(tag); }
+  void RemoveMetadata(const std::string& tag) { metadata_store_.erase(tag); }
+
+  bool HasData(const std::string& tag) const {
+    return data_store_.find(tag) != data_store_.end();
+  }
+
+  bool HasMetadata(const std::string& tag) const {
+    return metadata_store_.find(tag) != metadata_store_.end();
+  }
+
+  const EntityData& GetData(const std::string& tag) const {
+    return *data_store_.find(tag)->second;
+  }
+
+  const sync_pb::EntityMetadata& GetMetadata(const std::string& tag) const {
+    return metadata_store_.find(tag)->second;
+  }
+
+  size_t DataCount() const { return data_store_.size(); }
+  size_t MetadataCount() const { return metadata_store_.size(); }
+
+  const sync_pb::DataTypeState& data_type_state() const {
+    return data_type_state_;
+  }
+
+  void set_data_type_state(const sync_pb::DataTypeState& data_type_state) {
+    data_type_state_ = data_type_state;
+  }
+
+  scoped_ptr<MetadataBatch> CreateMetadataBatch() const {
+    scoped_ptr<MetadataBatch> metadata_batch(new MetadataBatch());
+    metadata_batch->SetDataTypeState(data_type_state_);
+    for (auto it = metadata_store_.begin(); it != metadata_store_.end(); it++) {
+      metadata_batch->AddMetadata(it->first, it->second);
+    }
+    return metadata_batch;
+  }
+
+  void Reset() {
+    data_store_.clear();
+    metadata_store_.clear();
+    data_type_state_.Clear();
+  }
+
+ private:
+  std::unordered_map<std::string, scoped_ptr<EntityData>> data_store_;
+  std::unordered_map<std::string, sync_pb::EntityMetadata> metadata_store_;
+  sync_pb::DataTypeState data_type_state_;
+};
+
+}  // namespace
+
+// Tests the various functionality of SharedModelTypeProcessor.
 //
-// Inputs:
-// - Initial state from permanent storage. (TODO)
-// - Create, update or delete requests from the model.
-// - Update responses and commit responses from the server.
+// The processor sits between the service (implemented by this test class) and
+// the worker, which is represented as a commit queue (MockCommitQueue). This
+// test suite exercises the initialization flows (whether initial sync is done,
+// loading pending updates, performing the initial merge, etc) as well as normal
+// functionality:
 //
-// Outputs:
-// - Writes to permanent storage. (TODO)
-// - Callbacks into the model. (TODO)
-// - Requests to the sync thread.  Tested with MockCommitQueue.
+// - Initialization before the initial sync and merge correctly performs a merge
+//   and initializes the metadata in storage. TODO(maxbogue): crbug.com/569675.
+// - Initialization after the initial sync correctly loads metadata and queues
+//   any pending commits.
+// - Put and Delete calls from the service result in the correct metadata in
+//   storage and the correct commit requests on the worker side.
+// - Updates and commit responses from the worker correctly affect data and
+//   metadata in storage on the service side.
 class SharedModelTypeProcessorTest : public ::testing::Test,
                                      public FakeModelTypeService {
  public:
-  SharedModelTypeProcessorTest()
-      : mock_queue_(new MockCommitQueue()),
-        mock_queue_ptr_(mock_queue_),
-        metadata_batch_(new MetadataBatch()) {
+  SharedModelTypeProcessorTest() {}
+
+  ~SharedModelTypeProcessorTest() override {}
+
+  void CreateProcessor() {
+    ASSERT_FALSE(type_processor());
     set_change_processor(
         make_scoped_ptr(new SharedModelTypeProcessor(kModelType, this)));
   }
 
-  ~SharedModelTypeProcessorTest() override {}
+  void InitializeToMetadataLoaded() {
+    CreateProcessor();
+    sync_pb::DataTypeState data_type_state(db_.data_type_state());
+    data_type_state.set_initial_sync_done(true);
+    db_.set_data_type_state(data_type_state);
+    OnMetadataLoaded();
+  }
 
   // Initialize to a "ready-to-commit" state.
   void InitializeToReadyState() {
-    data_type_state_.set_initial_sync_done(true);
-    OnMetadataLoaded();
+    InitializeToMetadataLoaded();
+    OnDataLoaded();
     OnSyncStarting();
-    // TODO(maxbogue): crbug.com/569642: Remove this once entity data is loaded
-    // for the normal startup flow.
-    UpdateResponseDataList empty_update_list;
-    type_processor()->OnUpdateReceived(data_type_state_, empty_update_list,
-                                       empty_update_list);
   }
 
   void OnMetadataLoaded() {
-    metadata_batch_->SetDataTypeState(data_type_state_);
-    type_processor()->OnMetadataLoaded(std::move(metadata_batch_));
-    metadata_batch_.reset(new MetadataBatch());
+    type_processor()->OnMetadataLoaded(db_.CreateMetadataBatch());
+  }
+
+  void OnDataLoaded() {
+    if (!data_callback_.is_null()) {
+      data_callback_.Run();
+      data_callback_.Reset();
+    }
   }
 
   void OnSyncStarting() {
@@ -87,55 +176,83 @@ class SharedModelTypeProcessorTest : public ::testing::Test,
 
   void DisconnectSync() {
     type_processor()->DisconnectSync();
-    mock_queue_ = NULL;
-    mock_queue_ptr_.reset();
+    mock_queue_ = nullptr;
   }
 
   // Disable sync for this SharedModelTypeProcessor.  Should cause sync state to
   // be discarded.
   void Disable() {
     type_processor()->Disable();
-    mock_queue_ = NULL;
-    mock_queue_ptr_.reset();
+    mock_queue_ = nullptr;
     EXPECT_FALSE(type_processor());
   }
 
-  // Restart sync after DisconnectSync() or Disable().
-  void Restart() {
-    if (!type_processor()) {
-      set_change_processor(
-          make_scoped_ptr(new SharedModelTypeProcessor(kModelType, this)));
-    }
-    // Prepare a new MockCommitQueue instance, just as we would
-    // if this happened in the real world.
-    mock_queue_ptr_.reset(new MockCommitQueue());
-    mock_queue_ = mock_queue_ptr_.get();
-    // Restart sync with the new CommitQueue.
-    OnSyncStarting();
-  }
-
   // Local data modification.  Emulates signals from the model thread.
-  void WriteItem(const std::string& tag,
-                 const std::string& value,
-                 MetadataChangeList* change_list) {
+  void WriteItem(const std::string& tag, const std::string& value) {
     scoped_ptr<EntityData> entity_data = make_scoped_ptr(new EntityData());
     entity_data->specifics = GenerateSpecifics(tag, value);
     entity_data->non_unique_name = tag;
-    type_processor()->Put(tag, std::move(entity_data), change_list);
+    WriteItem(tag, std::move(entity_data));
+  }
+
+  // Overloaded form to allow passing of custom entity data.
+  void WriteItem(const std::string& tag, scoped_ptr<EntityData> entity_data) {
+    db_.PutData(tag, *entity_data);
+    if (type_processor()) {
+      scoped_ptr<MetadataChangeList> change_list(
+          new SimpleMetadataChangeList());
+      type_processor()->Put(tag, std::move(entity_data), change_list.get());
+      ApplyMetadataChangeList(std::move(change_list));
+    }
+  }
+
+  // Writes data for |tag| and simulates a commit response for it.
+  void WriteItemAndAck(const std::string& tag, const std::string& value) {
+    WriteItem(tag, value);
+    ASSERT_TRUE(HasCommitRequestForTag(tag));
+    SuccessfulCommitResponse(GetLatestCommitRequestForTag(tag));
+  }
+
+  void DeleteItem(const std::string& tag) {
+    db_.RemoveData(tag);
+    if (type_processor()) {
+      scoped_ptr<MetadataChangeList> change_list(
+          new SimpleMetadataChangeList());
+      type_processor()->Delete(tag, change_list.get());
+      ApplyMetadataChangeList(std::move(change_list));
+    }
+  }
+
+  // Wipes existing DB and simulates one uncommited item.
+  void ResetStateWriteItem(const std::string& tag, const std::string& value) {
+    clear_change_processor();
+    db_.Reset();
+    InitializeToReadyState();
+    EXPECT_EQ(0U, ProcessorEntityCount());
+    WriteItem(tag, value);
+    EXPECT_EQ(1U, ProcessorEntityCount());
+    clear_change_processor();
   }
 
-  void DeleteItem(const std::string& tag, MetadataChangeList* change_list) {
-    type_processor()->Delete(tag, change_list);
+  // Wipes existing DB and simulates one uncommited deletion.
+  void ResetStateDeleteItem(const std::string& tag, const std::string& value) {
+    clear_change_processor();
+    db_.Reset();
+    InitializeToReadyState();
+    EXPECT_EQ(0U, ProcessorEntityCount());
+    WriteItemAndAck(tag, value);
+    EXPECT_EQ(1U, ProcessorEntityCount());
+    DeleteItem(tag);
+    EXPECT_EQ(1U, ProcessorEntityCount());
+    clear_change_processor();
   }
 
   // Emulates an "initial sync done" message from the CommitQueue.
   void OnInitialSyncDone() {
-    data_type_state_.set_initial_sync_done(true);
+    sync_pb::DataTypeState data_type_state(db_.data_type_state());
+    data_type_state.set_initial_sync_done(true);
     UpdateResponseDataList empty_update_list;
-
-    // TODO(stanisc): crbug/569645: replace this with loading the initial state
-    // via LoadMetadata callback.
-    type_processor()->OnUpdateReceived(data_type_state_, empty_update_list,
+    type_processor()->OnUpdateReceived(data_type_state, empty_update_list,
                                        empty_update_list);
   }
 
@@ -150,7 +267,7 @@ class SharedModelTypeProcessorTest : public ::testing::Test,
 
     UpdateResponseDataList list;
     list.push_back(data);
-    type_processor()->OnUpdateReceived(data_type_state_, list,
+    type_processor()->OnUpdateReceived(db_.data_type_state(), list,
                                        UpdateResponseDataList());
   }
 
@@ -163,7 +280,7 @@ class SharedModelTypeProcessorTest : public ::testing::Test,
 
     UpdateResponseDataList list;
     list.push_back(data);
-    type_processor()->OnUpdateReceived(data_type_state_, list,
+    type_processor()->OnUpdateReceived(db_.data_type_state(), list,
                                        UpdateResponseDataList());
   }
 
@@ -180,7 +297,7 @@ class SharedModelTypeProcessorTest : public ::testing::Test,
 
     UpdateResponseDataList list;
     list.push_back(data);
-    type_processor()->OnUpdateReceived(data_type_state_,
+    type_processor()->OnUpdateReceived(db_.data_type_state(),
                                        UpdateResponseDataList(), list);
   }
 
@@ -188,9 +305,8 @@ class SharedModelTypeProcessorTest : public ::testing::Test,
   bool HasPendingUpdate(const std::string& tag) const {
     const std::string client_tag_hash = GenerateTagHash(tag);
     const UpdateResponseDataList list = type_processor()->GetPendingUpdates();
-    for (UpdateResponseDataList::const_iterator it = list.begin();
-         it != list.end(); ++it) {
-      if (it->entity->client_tag_hash == client_tag_hash)
+    for (const UpdateResponseData& update : list) {
+      if (update.entity->client_tag_hash == client_tag_hash)
         return true;
     }
     return false;
@@ -201,10 +317,9 @@ class SharedModelTypeProcessorTest : public ::testing::Test,
     DCHECK(HasPendingUpdate(tag));
     const std::string client_tag_hash = GenerateTagHash(tag);
     const UpdateResponseDataList list = type_processor()->GetPendingUpdates();
-    for (UpdateResponseDataList::const_iterator it = list.begin();
-         it != list.end(); ++it) {
-      if (it->entity->client_tag_hash == client_tag_hash)
-        return *it;
+    for (const UpdateResponseData& update : list) {
+      if (update.entity->client_tag_hash == client_tag_hash)
+        return update;
     }
     NOTREACHED();
     return UpdateResponseData();
@@ -217,6 +332,7 @@ class SharedModelTypeProcessorTest : public ::testing::Test,
 
   // Read emitted commit requests as batches.
   size_t GetNumCommitRequestLists() {
+    DCHECK(mock_queue_);
     return mock_queue_->GetNumCommitRequestLists();
   }
 
@@ -239,15 +355,16 @@ class SharedModelTypeProcessorTest : public ::testing::Test,
   void SuccessfulCommitResponse(const CommitRequestData& request_data) {
     CommitResponseDataList list;
     list.push_back(mock_queue_->SuccessfulCommitResponse(request_data));
-    type_processor()->OnCommitCompleted(data_type_state_, list);
+    type_processor()->OnCommitCompleted(db_.data_type_state(), list);
   }
 
   // Sends the type sync proxy an updated DataTypeState to let it know that
   // the desired encryption key has changed.
   void UpdateDesiredEncryptionKey(const std::string& key_name) {
-    data_type_state_.set_encryption_key_name(key_name);
+    sync_pb::DataTypeState data_type_state(db_.data_type_state());
+    data_type_state.set_encryption_key_name(key_name);
     type_processor()->OnUpdateReceived(
-        data_type_state_, UpdateResponseDataList(), UpdateResponseDataList());
+        data_type_state, UpdateResponseDataList(), UpdateResponseDataList());
   }
 
   // Sets the key_name that the mock CommitQueue will claim is in use
@@ -256,38 +373,32 @@ class SharedModelTypeProcessorTest : public ::testing::Test,
     mock_queue_->SetServerEncryptionKey(key_name);
   }
 
-  void AddMetadataToBatch(const std::string& tag) {
-    const std::string tag_hash = GenerateTagHash(tag);
-    base::Time creation_time = base::Time::Now();
-
-    sync_pb::EntityMetadata metadata;
-    metadata.set_client_tag_hash(tag_hash);
-    metadata.set_creation_time(syncer::TimeToProtoTime(creation_time));
-
-    metadata_batch()->AddMetadata(tag, metadata);
-  }
-
   // Return the number of entities the processor has metadata for.
   size_t ProcessorEntityCount() const {
+    DCHECK(type_processor());
     return type_processor()->entities_.size();
   }
 
+  // Expect that the |n|th commit request list has one commit request for |tag|
+  // with |value| set.
+  void ExpectNthCommitRequestList(size_t n,
+                                  const std::string& tag,
+                                  const std::string& value) {
+    const CommitRequestDataList& list = GetNthCommitRequestList(n);
+    ASSERT_EQ(1U, list.size());
+    const EntityData& data = list[0].entity.value();
+    EXPECT_EQ(GenerateTagHash(tag), data.client_tag_hash);
+    EXPECT_EQ(value, data.specifics.preference().value());
+  }
+
+  const SimpleStore& db() const { return db_; }
+
   MockCommitQueue* mock_queue() { return mock_queue_; }
 
   SharedModelTypeProcessor* type_processor() const {
     return static_cast<SharedModelTypeProcessor*>(change_processor());
   }
 
-  const EntityChangeList* entity_change_list() const {
-    return entity_change_list_.get();
-  }
-
-  const FakeMetadataChangeList* metadata_change_list() const {
-    return metadata_change_list_.get();
-  }
-
-  MetadataBatch* metadata_batch() { return metadata_batch_.get(); }
-
  private:
   static std::string GenerateTagHash(const std::string& tag) {
     return syncer::syncable::GenerateSyncableHash(kModelType, tag);
@@ -316,10 +427,10 @@ class SharedModelTypeProcessorTest : public ::testing::Test,
 
   void OnReadyToConnect(syncer::SyncError error,
                         scoped_ptr<ActivationContext> context) {
-    // Hand off ownership of |mock_queue_ptr_|, while keeping
-    // an unsafe pointer to it.  This is why we can only connect once.
-    DCHECK(mock_queue_ptr_);
-    context->type_processor->ConnectSync(std::move(mock_queue_ptr_));
+    scoped_ptr<MockCommitQueue> commit_queue(new MockCommitQueue());
+    // Keep an unsafe pointer to the commit queue the processor will use.
+    mock_queue_ = commit_queue.get();
+    context->type_processor->ConnectSync(std::move(commit_queue));
     // The context's type processor is a proxy; run the task it posted.
     sync_loop_.RunUntilIdle();
   }
@@ -332,50 +443,297 @@ class SharedModelTypeProcessorTest : public ::testing::Test,
   }
 
   scoped_ptr<MetadataChangeList> CreateMetadataChangeList() override {
-    // Reset the current first and return a new one.
-    metadata_change_list_.reset();
-    return scoped_ptr<MetadataChangeList>(new FakeMetadataChangeList());
+    return scoped_ptr<MetadataChangeList>(new SimpleMetadataChangeList());
   }
 
   syncer::SyncError ApplySyncChanges(
-      scoped_ptr<MetadataChangeList> metadata_change_list,
+      scoped_ptr<MetadataChangeList> metadata_changes,
       EntityChangeList entity_changes) override {
-    EXPECT_FALSE(metadata_change_list_);
-    // |metadata_change_list| is expected to be an instance of
-    // FakeMetadataChangeList - see above.
-    metadata_change_list_.reset(
-        static_cast<FakeMetadataChangeList*>(metadata_change_list.release()));
-    EXPECT_TRUE(metadata_change_list_);
-    entity_change_list_.reset(new EntityChangeList(entity_changes));
+    for (const EntityChange& change : entity_changes) {
+      switch (change.type()) {
+        case EntityChange::ACTION_ADD:
+          EXPECT_FALSE(db_.HasData(change.client_tag()));
+          db_.PutData(change.client_tag(), change.data());
+          break;
+        case EntityChange::ACTION_UPDATE:
+          EXPECT_TRUE(db_.HasData(change.client_tag()));
+          db_.PutData(change.client_tag(), change.data());
+          break;
+        case EntityChange::ACTION_DELETE:
+          EXPECT_TRUE(db_.HasData(change.client_tag()));
+          db_.RemoveData(change.client_tag());
+          break;
+      }
+    }
+    ApplyMetadataChangeList(std::move(metadata_changes));
     return syncer::SyncError();
   }
 
+  void ApplyMetadataChangeList(scoped_ptr<MetadataChangeList> change_list) {
+    DCHECK(change_list);
+    SimpleMetadataChangeList* changes =
+        static_cast<SimpleMetadataChangeList*>(change_list.get());
+    const auto& metadata_changes = changes->GetMetadataChanges();
+    for (auto it = metadata_changes.begin(); it != metadata_changes.end();
+         it++) {
+      switch (it->second.type) {
+        case SimpleMetadataChangeList::UPDATE:
+          db_.PutMetadata(it->first, it->second.metadata);
+          break;
+        case SimpleMetadataChangeList::CLEAR:
+          db_.RemoveMetadata(it->first);
+          break;
+      }
+    }
+    if (changes->HasDataTypeStateChange()) {
+      const SimpleMetadataChangeList::DataTypeStateChange& state_change =
+          changes->GetDataTypeStateChange();
+      switch (state_change.type) {
+        case SimpleMetadataChangeList::UPDATE:
+          db_.set_data_type_state(state_change.state);
+          break;
+        case SimpleMetadataChangeList::CLEAR:
+          db_.set_data_type_state(sync_pb::DataTypeState());
+          break;
+      }
+    }
+  }
+
+  void GetData(ClientTagList tags, DataCallback callback) override {
+    scoped_ptr<DataBatchImpl> batch(new DataBatchImpl());
+    for (const std::string& tag : tags) {
+      batch->Put(tag, CopyEntityData(db_.GetData(tag)));
+    }
+    data_callback_ =
+        base::Bind(callback, syncer::SyncError(), base::Passed(&batch));
+  }
+
   // This sets ThreadTaskRunnerHandle on the current thread, which the type
   // processor will pick up as the sync task runner.
   base::MessageLoop sync_loop_;
 
-  // The current mock queue which might be owned by either |mock_queue_ptr_| or
-  // |type_processor()|.
+  // The current mock queue, which is owned by |type_processor()|.
   MockCommitQueue* mock_queue_;
-  scoped_ptr<MockCommitQueue> mock_queue_ptr_;
 
-  sync_pb::DataTypeState data_type_state_;
+  // Stores the data callback between GetData() and OnDataLoaded().
+  base::Closure data_callback_;
 
-  // The last received EntityChangeList.
-  scoped_ptr<EntityChangeList> entity_change_list_;
-  // The last received MetadataChangeList.
-  scoped_ptr<FakeMetadataChangeList> metadata_change_list_;
-  scoped_ptr<MetadataBatch> metadata_batch_;
+  // Contains all of the data and metadata state for these tests.
+  SimpleStore db_;
 };
 
-TEST_F(SharedModelTypeProcessorTest, Initialize) {
-  // TODO(maxbogue): crbug.com/569642: Add data for tag1.
-  AddMetadataToBatch("tag1");
-  ASSERT_EQ(0U, ProcessorEntityCount());
+// This test covers race conditions during loading pending data. All cases
+// start with no processor and one item with a pending commit. There are three
+// different events that can occur in any order once metadata is loaded:
+//
+// - Pending commit data is loaded.
+// - Sync gets connected.
+// - Optionally, a put or delete happens to the item.
+//
+// This results in 2 + 12 = 14 orderings of the events.
+TEST_F(SharedModelTypeProcessorTest, LoadPendingCommit) {
+  // Data, connect.
+  ResetStateWriteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  OnDataLoaded();
+  OnSyncStarting();
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "value1");
+
+  // Connect, data.
+  ResetStateWriteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  OnSyncStarting();
+  EXPECT_EQ(0U, GetNumCommitRequestLists());
+  OnDataLoaded();
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "value1");
+
+  // Data, connect, put.
+  ResetStateWriteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  OnDataLoaded();
+  OnSyncStarting();
+  WriteItem("tag1", "value2");
+  EXPECT_EQ(2U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "value1");
+  ExpectNthCommitRequestList(1, "tag1", "value2");
+
+  // Data, put, connect.
+  ResetStateWriteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  OnDataLoaded();
+  WriteItem("tag1", "value2");
+  OnSyncStarting();
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "value2");
+
+  // Connect, data, put.
+  ResetStateWriteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  OnSyncStarting();
+  OnDataLoaded();
+  WriteItem("tag1", "value2");
+  EXPECT_EQ(2U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "value1");
+  ExpectNthCommitRequestList(1, "tag1", "value2");
+
+  // Connect, put, data.
+  ResetStateWriteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  OnSyncStarting();
+  WriteItem("tag1", "value2");
+  EXPECT_EQ(0U, GetNumCommitRequestLists());
+  OnDataLoaded();
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "value2");
+
+  // Put, data, connect.
+  ResetStateWriteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  WriteItem("tag1", "value2");
+  OnDataLoaded();
+  OnSyncStarting();
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "value2");
+
+  // Put, connect, data.
+  ResetStateWriteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  WriteItem("tag1", "value2");
+  OnSyncStarting();
+  EXPECT_EQ(0U, GetNumCommitRequestLists());
+  OnDataLoaded();
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "value2");
+
+  // Data, connect, delete.
+  ResetStateWriteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  OnDataLoaded();
+  OnSyncStarting();
+  DeleteItem("tag1");
+  EXPECT_EQ(2U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "value1");
+  ExpectNthCommitRequestList(1, "tag1", "");
+
+  // Data, delete, connect.
+  ResetStateWriteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  OnDataLoaded();
+  DeleteItem("tag1");
+  OnSyncStarting();
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "");
+
+  // Connect, data, delete.
+  ResetStateWriteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  OnSyncStarting();
+  OnDataLoaded();
+  DeleteItem("tag1");
+  EXPECT_EQ(2U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "value1");
+  ExpectNthCommitRequestList(1, "tag1", "");
+
+  // Connect, delete, data.
+  ResetStateWriteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  OnSyncStarting();
+  DeleteItem("tag1");
+  EXPECT_EQ(0U, GetNumCommitRequestLists());
+  OnDataLoaded();
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "");
+
+  // Delete, data, connect.
+  ResetStateWriteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  DeleteItem("tag1");
+  OnDataLoaded();
+  OnSyncStarting();
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "");
+
+  // Delete, connect, data.
+  ResetStateWriteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  DeleteItem("tag1");
+  OnSyncStarting();
+  EXPECT_EQ(0U, GetNumCommitRequestLists());
+  OnDataLoaded();
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "");
+}
+
+// This test covers race conditions during loading a pending delete. All cases
+// start with no processor and one item with a pending delete. There are two
+// different events that can occur in any order once metadata is loaded, since
+// for a deletion there is no data to load:
+//
+// - Sync gets connected.
+// - Optionally, a put or delete happens to the item (repeated deletes should be
+//   handled properly).
+//
+// This results in 1 + 4 = 5 orderings of the events.
+TEST_F(SharedModelTypeProcessorTest, LoadPendingDelete) {
+  // Connect.
+  ResetStateDeleteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  OnSyncStarting();
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "");
+
+  // Connect, put.
+  ResetStateDeleteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  OnSyncStarting();
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  WriteItem("tag1", "value2");
+  EXPECT_EQ(2U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "");
+  ExpectNthCommitRequestList(1, "tag1", "value2");
+
+  // Put, connect.
+  ResetStateDeleteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  WriteItem("tag1", "value2");
+  OnSyncStarting();
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "value2");
+
+  // Connect, delete.
+  ResetStateDeleteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  OnSyncStarting();
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  DeleteItem("tag1");
+  EXPECT_EQ(2U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "");
+  ExpectNthCommitRequestList(1, "tag1", "");
+
+  // Delete, connect.
+  ResetStateDeleteItem("tag1", "value1");
+  InitializeToMetadataLoaded();
+  DeleteItem("tag1");
+  OnSyncStarting();
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  ExpectNthCommitRequestList(0, "tag1", "");
+}
+
+// Test that loading a committed item does not queue another commit.
+TEST_F(SharedModelTypeProcessorTest, LoadCommited) {
   InitializeToReadyState();
-  ASSERT_EQ(1U, ProcessorEntityCount());
-  // TODO(maxbogue): crbug.com/569642: Verify that a commit is added to the
-  // queue.
+  WriteItem("tag1", "value1");
+  // Complete the commit.
+  EXPECT_TRUE(HasCommitRequestForTag("tag1"));
+  SuccessfulCommitResponse(GetLatestCommitRequestForTag("tag1"));
+  clear_change_processor();
+
+  // Test that a new processor loads the metadata without committing.
+  InitializeToReadyState();
+  EXPECT_EQ(1U, ProcessorEntityCount());
+  EXPECT_EQ(0U, GetNumCommitRequestLists());
 }
 
 // Creates a new item locally.
@@ -384,8 +742,7 @@ TEST_F(SharedModelTypeProcessorTest, CreateLocalItem) {
   InitializeToReadyState();
   EXPECT_EQ(0U, GetNumCommitRequestLists());
 
-  FakeMetadataChangeList change_list;
-  WriteItem("tag1", "value1", &change_list);
+  WriteItem("tag1", "value1");
 
   // Verify the commit request this operation has triggered.
   EXPECT_EQ(1U, GetNumCommitRequestLists());
@@ -403,20 +760,17 @@ TEST_F(SharedModelTypeProcessorTest, CreateLocalItem) {
   EXPECT_EQ("tag1", tag1_data.specifics.preference().name());
   EXPECT_EQ("value1", tag1_data.specifics.preference().value());
 
-  EXPECT_EQ(1U, change_list.GetNumRecords());
-
-  const FakeMetadataChangeList::Record& record = change_list.GetNthRecord(0);
-  EXPECT_EQ(FakeMetadataChangeList::UPDATE_METADATA, record.action);
-  EXPECT_EQ("tag1", record.tag);
-  EXPECT_TRUE(record.metadata.has_client_tag_hash());
-  EXPECT_FALSE(record.metadata.has_server_id());
-  EXPECT_FALSE(record.metadata.is_deleted());
-  EXPECT_EQ(1, record.metadata.sequence_number());
-  EXPECT_EQ(0, record.metadata.acked_sequence_number());
-  EXPECT_EQ(kUncommittedVersion, record.metadata.server_version());
-  EXPECT_TRUE(record.metadata.has_creation_time());
-  EXPECT_TRUE(record.metadata.has_modification_time());
-  EXPECT_TRUE(record.metadata.has_specifics_hash());
+  EXPECT_EQ(1U, db().MetadataCount());
+  const sync_pb::EntityMetadata metadata = db().GetMetadata("tag1");
+  EXPECT_TRUE(metadata.has_client_tag_hash());
+  EXPECT_FALSE(metadata.has_server_id());
+  EXPECT_FALSE(metadata.is_deleted());
+  EXPECT_EQ(1, metadata.sequence_number());
+  EXPECT_EQ(0, metadata.acked_sequence_number());
+  EXPECT_EQ(kUncommittedVersion, metadata.server_version());
+  EXPECT_TRUE(metadata.has_creation_time());
+  EXPECT_TRUE(metadata.has_modification_time());
+  EXPECT_TRUE(metadata.has_specifics_hash());
 }
 
 // The purpose of this test case is to test setting |client_tag_hash| and |id|
@@ -427,8 +781,6 @@ TEST_F(SharedModelTypeProcessorTest, CreateAndModifyWithOverrides) {
   InitializeToReadyState();
   EXPECT_EQ(0U, GetNumCommitRequestLists());
 
-  FakeMetadataChangeList change_list;
-
   scoped_ptr<EntityData> entity_data = make_scoped_ptr(new EntityData());
   entity_data->specifics.mutable_preference()->set_name("name1");
   entity_data->specifics.mutable_preference()->set_value("value1");
@@ -436,7 +788,7 @@ TEST_F(SharedModelTypeProcessorTest, CreateAndModifyWithOverrides) {
   entity_data->non_unique_name = "name1";
   entity_data->client_tag_hash = "hash";
   entity_data->id = "cid1";
-  type_processor()->Put("tag1", std::move(entity_data), &change_list);
+  WriteItem("tag1", std::move(entity_data));
 
   // Don't access through tag because we forced a specific hash.
   EXPECT_EQ(1U, GetNumCommitRequestLists());
@@ -448,7 +800,10 @@ TEST_F(SharedModelTypeProcessorTest, CreateAndModifyWithOverrides) {
   EXPECT_EQ("cid1", out_entity1.id);
   EXPECT_EQ("value1", out_entity1.specifics.preference().value());
 
-  EXPECT_EQ(1U, change_list.GetNumRecords());
+  EXPECT_EQ(1U, db().MetadataCount());
+  const sync_pb::EntityMetadata metadata_v1 = db().GetMetadata("tag1");
+  EXPECT_EQ("cid1", metadata_v1.server_id());
+  EXPECT_EQ("hash", metadata_v1.client_tag_hash());
 
   entity_data.reset(new EntityData());
   entity_data->specifics.mutable_preference()->set_name("name2");
@@ -459,7 +814,7 @@ TEST_F(SharedModelTypeProcessorTest, CreateAndModifyWithOverrides) {
   // client once established.
   entity_data->id = "cid2";
 
-  type_processor()->Put("tag1", std::move(entity_data), &change_list);
+  WriteItem("tag1", std::move(entity_data));
 
   EXPECT_EQ(2U, GetNumCommitRequestLists());
   ASSERT_TRUE(mock_queue()->HasCommitRequestForTagHash("hash"));
@@ -471,23 +826,13 @@ TEST_F(SharedModelTypeProcessorTest, CreateAndModifyWithOverrides) {
   EXPECT_EQ("cid1", out_entity2.id);
   EXPECT_EQ("value2", out_entity2.specifics.preference().value());
 
-  EXPECT_EQ(2U, change_list.GetNumRecords());
-
-  const FakeMetadataChangeList::Record& record1 = change_list.GetNthRecord(0);
-  EXPECT_EQ(FakeMetadataChangeList::UPDATE_METADATA, record1.action);
-  EXPECT_EQ("tag1", record1.tag);
-  EXPECT_EQ("cid1", record1.metadata.server_id());
-  EXPECT_EQ("hash", record1.metadata.client_tag_hash());
-
-  const FakeMetadataChangeList::Record& record2 = change_list.GetNthRecord(1);
-  EXPECT_EQ(FakeMetadataChangeList::UPDATE_METADATA, record2.action);
-  EXPECT_EQ("tag1", record2.tag);
+  EXPECT_EQ(1U, db().MetadataCount());
+  const sync_pb::EntityMetadata metadata_v2 = db().GetMetadata("tag1");
   // TODO(skym): Is this correct?
-  EXPECT_EQ("cid1", record2.metadata.server_id());
-  EXPECT_EQ("hash", record2.metadata.client_tag_hash());
+  EXPECT_EQ("cid1", metadata_v2.server_id());
+  EXPECT_EQ("hash", metadata_v2.client_tag_hash());
 
-  EXPECT_NE(record1.metadata.specifics_hash(),
-            record2.metadata.specifics_hash());
+  EXPECT_NE(metadata_v1.specifics_hash(), metadata_v2.specifics_hash());
 }
 
 // Creates a new local item then modifies it.
@@ -496,75 +841,62 @@ TEST_F(SharedModelTypeProcessorTest, CreateAndModifyLocalItem) {
   InitializeToReadyState();
   EXPECT_EQ(0U, GetNumCommitRequestLists());
 
-  FakeMetadataChangeList change_list;
-
-  WriteItem("tag1", "value1", &change_list);
+  WriteItem("tag1", "value1");
   EXPECT_EQ(1U, GetNumCommitRequestLists());
+  EXPECT_EQ(1U, db().MetadataCount());
   ASSERT_TRUE(HasCommitRequestForTag("tag1"));
-  const CommitRequestData& tag1_v1_request_data =
+  const CommitRequestData& request_data_v1 =
       GetLatestCommitRequestForTag("tag1");
-  const EntityData& tag1_v1_data = tag1_v1_request_data.entity.value();
+  const EntityData& data_v1 = request_data_v1.entity.value();
+  const sync_pb::EntityMetadata metadata_v1 = db().GetMetadata("tag1");
 
-  WriteItem("tag1", "value2", &change_list);
+  WriteItem("tag1", "value2");
   EXPECT_EQ(2U, GetNumCommitRequestLists());
+  EXPECT_EQ(1U, db().MetadataCount());
 
   ASSERT_TRUE(HasCommitRequestForTag("tag1"));
-  const CommitRequestData& tag1_v2_request_data =
+  const CommitRequestData& request_data_v2 =
       GetLatestCommitRequestForTag("tag1");
-  const EntityData& tag1_v2_data = tag1_v2_request_data.entity.value();
+  const EntityData& data_v2 = request_data_v2.entity.value();
+  const sync_pb::EntityMetadata metadata_v2 = db().GetMetadata("tag1");
 
   // Test some of the relations between old and new commit requests.
-  EXPECT_GT(tag1_v2_request_data.sequence_number,
-            tag1_v1_request_data.sequence_number);
-  EXPECT_EQ(tag1_v1_data.specifics.preference().value(), "value1");
+  EXPECT_GT(request_data_v2.sequence_number, request_data_v1.sequence_number);
+  EXPECT_EQ(data_v1.specifics.preference().value(), "value1");
 
   // Perform a thorough examination of the update-generated request.
-  EXPECT_EQ(kUncommittedVersion, tag1_v2_request_data.base_version);
-  EXPECT_TRUE(tag1_v2_data.id.empty());
-  EXPECT_FALSE(tag1_v2_data.creation_time.is_null());
-  EXPECT_FALSE(tag1_v2_data.modification_time.is_null());
-  EXPECT_EQ("tag1", tag1_v2_data.non_unique_name);
-  EXPECT_FALSE(tag1_v2_data.is_deleted());
-  EXPECT_EQ("tag1", tag1_v2_data.specifics.preference().name());
-  EXPECT_EQ("value2", tag1_v2_data.specifics.preference().value());
-
-  EXPECT_EQ(2U, change_list.GetNumRecords());
-
-  const FakeMetadataChangeList::Record& record1 = change_list.GetNthRecord(0);
-  EXPECT_EQ(FakeMetadataChangeList::UPDATE_METADATA, record1.action);
-  EXPECT_EQ("tag1", record1.tag);
-  EXPECT_FALSE(record1.metadata.has_server_id());
-  EXPECT_FALSE(record1.metadata.is_deleted());
-  EXPECT_EQ(1, record1.metadata.sequence_number());
-  EXPECT_EQ(0, record1.metadata.acked_sequence_number());
-  EXPECT_EQ(kUncommittedVersion, record1.metadata.server_version());
-
-  const FakeMetadataChangeList::Record& record2 = change_list.GetNthRecord(1);
-  EXPECT_EQ(FakeMetadataChangeList::UPDATE_METADATA, record1.action);
-  EXPECT_EQ("tag1", record2.tag);
-  EXPECT_FALSE(record2.metadata.has_server_id());
-  EXPECT_FALSE(record2.metadata.is_deleted());
-  EXPECT_EQ(2, record2.metadata.sequence_number());
-  EXPECT_EQ(0, record2.metadata.acked_sequence_number());
-  EXPECT_EQ(kUncommittedVersion, record2.metadata.server_version());
-
-  EXPECT_EQ(record1.metadata.client_tag_hash(),
-            record2.metadata.client_tag_hash());
-  EXPECT_NE(record1.metadata.specifics_hash(),
-            record2.metadata.specifics_hash());
+  EXPECT_EQ(kUncommittedVersion, request_data_v2.base_version);
+  EXPECT_TRUE(data_v2.id.empty());
+  EXPECT_FALSE(data_v2.creation_time.is_null());
+  EXPECT_FALSE(data_v2.modification_time.is_null());
+  EXPECT_EQ("tag1", data_v2.non_unique_name);
+  EXPECT_FALSE(data_v2.is_deleted());
+  EXPECT_EQ("tag1", data_v2.specifics.preference().name());
+  EXPECT_EQ("value2", data_v2.specifics.preference().value());
+
+  EXPECT_FALSE(metadata_v1.has_server_id());
+  EXPECT_FALSE(metadata_v1.is_deleted());
+  EXPECT_EQ(1, metadata_v1.sequence_number());
+  EXPECT_EQ(0, metadata_v1.acked_sequence_number());
+  EXPECT_EQ(kUncommittedVersion, metadata_v1.server_version());
+
+  EXPECT_FALSE(metadata_v2.has_server_id());
+  EXPECT_FALSE(metadata_v2.is_deleted());
+  EXPECT_EQ(2, metadata_v2.sequence_number());
+  EXPECT_EQ(0, metadata_v2.acked_sequence_number());
+  EXPECT_EQ(kUncommittedVersion, metadata_v2.server_version());
+
+  EXPECT_EQ(metadata_v1.client_tag_hash(), metadata_v2.client_tag_hash());
+  EXPECT_NE(metadata_v1.specifics_hash(), metadata_v2.specifics_hash());
 }
 
 // Deletes an item we've never seen before.
 // Should have no effect and not crash.
 TEST_F(SharedModelTypeProcessorTest, DeleteUnknown) {
   InitializeToReadyState();
-
-  FakeMetadataChangeList change_list;
-
-  DeleteItem("tag1", &change_list);
+  DeleteItem("tag1");
   EXPECT_EQ(0U, GetNumCommitRequestLists());
-
-  EXPECT_EQ(0U, change_list.GetNumRecords());
+  EXPECT_EQ(0U, db().MetadataCount());
 }
 
 // Creates an item locally then deletes it.
@@ -575,38 +907,35 @@ TEST_F(SharedModelTypeProcessorTest, DeleteUnknown) {
 TEST_F(SharedModelTypeProcessorTest, DeleteServerUnknown) {
   InitializeToReadyState();
 
-  FakeMetadataChangeList change_list;
-
   // TODO(stanisc): crbug.com/573333: Review this case. If the flush of
   // all locally modified items was scheduled to run on a separate task, than
   // the correct behavior would be to commit just the detele, or perhaps no
   // commit at all.
 
-  WriteItem("tag1", "value1", &change_list);
+  WriteItem("tag1", "value1");
   EXPECT_EQ(1U, GetNumCommitRequestLists());
+  EXPECT_EQ(1U, db().MetadataCount());
   ASSERT_TRUE(HasCommitRequestForTag("tag1"));
-  const CommitRequestData& tag1_v1_data = GetLatestCommitRequestForTag("tag1");
+  const CommitRequestData& data_v1 = GetLatestCommitRequestForTag("tag1");
+  const sync_pb::EntityMetadata metadata_v1 = db().GetMetadata("tag1");
 
-  DeleteItem("tag1", &change_list);
+  DeleteItem("tag1");
   EXPECT_EQ(2U, GetNumCommitRequestLists());
+  EXPECT_EQ(1U, db().MetadataCount());
   ASSERT_TRUE(HasCommitRequestForTag("tag1"));
-  const CommitRequestData& tag1_v2_data = GetLatestCommitRequestForTag("tag1");
-
-  EXPECT_GT(tag1_v2_data.sequence_number, tag1_v1_data.sequence_number);
+  const CommitRequestData& data_v2 = GetLatestCommitRequestForTag("tag1");
+  const sync_pb::EntityMetadata metadata_v2 = db().GetMetadata("tag1");
 
-  EXPECT_TRUE(tag1_v2_data.entity->id.empty());
-  EXPECT_EQ(kUncommittedVersion, tag1_v2_data.base_version);
-  EXPECT_TRUE(tag1_v2_data.entity->is_deleted());
+  EXPECT_GT(data_v2.sequence_number, data_v1.sequence_number);
 
-  EXPECT_EQ(2U, change_list.GetNumRecords());
+  EXPECT_TRUE(data_v2.entity->id.empty());
+  EXPECT_EQ(kUncommittedVersion, data_v2.base_version);
+  EXPECT_TRUE(data_v2.entity->is_deleted());
 
-  const FakeMetadataChangeList::Record& record1 = change_list.GetNthRecord(0);
-  EXPECT_EQ(FakeMetadataChangeList::UPDATE_METADATA, record1.action);
-  EXPECT_EQ("tag1", record1.tag);
-  EXPECT_FALSE(record1.metadata.is_deleted());
-  EXPECT_EQ(1, record1.metadata.sequence_number());
-  EXPECT_EQ(0, record1.metadata.acked_sequence_number());
-  EXPECT_EQ(kUncommittedVersion, record1.metadata.server_version());
+  EXPECT_FALSE(metadata_v1.is_deleted());
+  EXPECT_EQ(1, metadata_v1.sequence_number());
+  EXPECT_EQ(0, metadata_v1.acked_sequence_number());
+  EXPECT_EQ(kUncommittedVersion, metadata_v1.server_version());
 
   // TODO(stanisc): crbug.com/573333: Review this case. Depending on the
   // implementation the second action performed on metadata change list might
@@ -615,13 +944,10 @@ TEST_F(SharedModelTypeProcessorTest, DeleteServerUnknown) {
   // records at all - the first call would create an entry and the second would
   // remove it.
 
-  const FakeMetadataChangeList::Record& record2 = change_list.GetNthRecord(1);
-  EXPECT_EQ(FakeMetadataChangeList::UPDATE_METADATA, record1.action);
-  EXPECT_EQ("tag1", record2.tag);
-  EXPECT_TRUE(record2.metadata.is_deleted());
-  EXPECT_EQ(2, record2.metadata.sequence_number());
-  EXPECT_EQ(0, record2.metadata.acked_sequence_number());
-  EXPECT_EQ(kUncommittedVersion, record2.metadata.server_version());
+  EXPECT_TRUE(metadata_v2.is_deleted());
+  EXPECT_EQ(2, metadata_v2.sequence_number());
+  EXPECT_EQ(0, metadata_v2.acked_sequence_number());
+  EXPECT_EQ(kUncommittedVersion, metadata_v2.server_version());
 }
 
 // Creates an item locally then deletes it.
@@ -632,48 +958,40 @@ TEST_F(SharedModelTypeProcessorTest, DeleteServerUnknown) {
 TEST_F(SharedModelTypeProcessorTest, DeleteServerUnknown_RacyCommitResponse) {
   InitializeToReadyState();
 
-  FakeMetadataChangeList change_list;
-
-  WriteItem("tag1", "value1", &change_list);
+  WriteItem("tag1", "value1");
   EXPECT_EQ(1U, GetNumCommitRequestLists());
+  EXPECT_EQ(1U, db().DataCount());
+  EXPECT_EQ(1U, db().MetadataCount());
   ASSERT_TRUE(HasCommitRequestForTag("tag1"));
-  const CommitRequestData& tag1_v1_data = GetLatestCommitRequestForTag("tag1");
+  const CommitRequestData& data_v1 = GetLatestCommitRequestForTag("tag1");
+  EXPECT_FALSE(db().GetMetadata("tag1").is_deleted());
 
-  DeleteItem("tag1", &change_list);
+  DeleteItem("tag1");
   EXPECT_EQ(2U, GetNumCommitRequestLists());
+  EXPECT_EQ(0U, db().DataCount());
+  EXPECT_EQ(1U, db().MetadataCount());
   ASSERT_TRUE(HasCommitRequestForTag("tag1"));
+  EXPECT_TRUE(db().GetMetadata("tag1").is_deleted());
 
   // This commit happened while the deletion was in progress, but the commit
   // response didn't arrive on our thread until after the delete was issued to
   // the sync thread.  It will update some metadata, but won't do much else.
-  SuccessfulCommitResponse(tag1_v1_data);
+  SuccessfulCommitResponse(data_v1);
+  EXPECT_EQ(0U, db().DataCount());
+  EXPECT_EQ(1U, db().MetadataCount());
 
   // In reality the change list used to commit local changes should never
   // overlap with the changelist used to deliver commit confirmation. In this
   // test setup the two change lists are isolated - one is on the stack and
   // another is the class member.
 
-  // Local metadata changes.
-  EXPECT_EQ(2U, change_list.GetNumRecords());
-
-  // Metadata changes from commit response.
-  EXPECT_TRUE(metadata_change_list());
-  EXPECT_EQ(2U, metadata_change_list()->GetNumRecords());
-
-  const FakeMetadataChangeList::Record& record1 =
-      metadata_change_list()->GetNthRecord(0);
-  EXPECT_EQ(FakeMetadataChangeList::UPDATE_DATA_TYPE_STATE, record1.action);
-
-  const FakeMetadataChangeList::Record& record2 =
-      metadata_change_list()->GetNthRecord(1);
-  EXPECT_EQ(FakeMetadataChangeList::UPDATE_METADATA, record2.action);
-  EXPECT_EQ("tag1", record2.tag);
+  const sync_pb::EntityMetadata metadata_v2 = db().GetMetadata("tag1");
   // Deleted from the second local modification.
-  EXPECT_TRUE(record2.metadata.is_deleted());
+  EXPECT_TRUE(metadata_v2.is_deleted());
   // sequence_number = 2 from the second local modification.
-  EXPECT_EQ(2, record2.metadata.sequence_number());
+  EXPECT_EQ(2, metadata_v2.sequence_number());
   // acked_sequence_number = 1 from the first commit response.
-  EXPECT_EQ(1, record2.metadata.acked_sequence_number());
+  EXPECT_EQ(1, metadata_v2.acked_sequence_number());
 
   // TODO(rlarocque): Verify the state of the item is correct once we get
   // storage hooked up in these tests.  For example, verify the item is still
@@ -686,60 +1004,52 @@ TEST_F(SharedModelTypeProcessorTest, TwoIndependentItems) {
   InitializeToReadyState();
   EXPECT_EQ(0U, GetNumCommitRequestLists());
 
-  FakeMetadataChangeList change_list;
-
-  WriteItem("tag1", "value1", &change_list);
+  WriteItem("tag1", "value1");
+  EXPECT_EQ(1U, db().DataCount());
+  EXPECT_EQ(1U, db().MetadataCount());
+  const sync_pb::EntityMetadata metadata1 = db().GetMetadata("tag1");
 
   // There should be one commit request for this item only.
   ASSERT_EQ(1U, GetNumCommitRequestLists());
   EXPECT_EQ(1U, GetNthCommitRequestList(0).size());
   ASSERT_TRUE(HasCommitRequestForTag("tag1"));
 
-  WriteItem("tag2", "value2", &change_list);
+  WriteItem("tag2", "value2");
+  EXPECT_EQ(2U, db().DataCount());
+  EXPECT_EQ(2U, db().MetadataCount());
+  const sync_pb::EntityMetadata metadata2 = db().GetMetadata("tag2");
 
   // The second write should trigger another single-item commit request.
   ASSERT_EQ(2U, GetNumCommitRequestLists());
   EXPECT_EQ(1U, GetNthCommitRequestList(1).size());
   ASSERT_TRUE(HasCommitRequestForTag("tag2"));
 
-  EXPECT_EQ(2U, change_list.GetNumRecords());
-
-  const FakeMetadataChangeList::Record& record1 = change_list.GetNthRecord(0);
-  EXPECT_EQ(FakeMetadataChangeList::UPDATE_METADATA, record1.action);
-  EXPECT_EQ("tag1", record1.tag);
-  EXPECT_FALSE(record1.metadata.is_deleted());
-  EXPECT_EQ(1, record1.metadata.sequence_number());
-  EXPECT_EQ(0, record1.metadata.acked_sequence_number());
-  EXPECT_EQ(kUncommittedVersion, record1.metadata.server_version());
-
-  const FakeMetadataChangeList::Record& record2 = change_list.GetNthRecord(1);
-  EXPECT_EQ(FakeMetadataChangeList::UPDATE_METADATA, record2.action);
-  EXPECT_EQ("tag2", record2.tag);
-  EXPECT_FALSE(record2.metadata.is_deleted());
-  EXPECT_EQ(1, record2.metadata.sequence_number());
-  EXPECT_EQ(0, record2.metadata.acked_sequence_number());
-  EXPECT_EQ(kUncommittedVersion, record2.metadata.server_version());
+  EXPECT_FALSE(metadata1.is_deleted());
+  EXPECT_EQ(1, metadata1.sequence_number());
+  EXPECT_EQ(0, metadata1.acked_sequence_number());
+  EXPECT_EQ(kUncommittedVersion, metadata1.server_version());
+
+  EXPECT_FALSE(metadata2.is_deleted());
+  EXPECT_EQ(1, metadata2.sequence_number());
+  EXPECT_EQ(0, metadata2.acked_sequence_number());
+  EXPECT_EQ(kUncommittedVersion, metadata2.server_version());
 }
 
 // Starts the type sync proxy with no local state.
 // Verify that it waits until initial sync is complete before requesting
 // commits.
 TEST_F(SharedModelTypeProcessorTest, NoCommitsUntilInitialSyncDone) {
-  OnSyncStarting();
+  CreateProcessor();
   OnMetadataLoaded();
+  OnSyncStarting();
 
-  FakeMetadataChangeList change_list;
-
-  WriteItem("tag1", "value1", &change_list);
+  WriteItem("tag1", "value1");
   EXPECT_EQ(0U, GetNumCommitRequestLists());
 
   // Even though there the item hasn't been committed its metadata should have
   // already been updated and the sequence number changed.
-  EXPECT_EQ(1U, change_list.GetNumRecords());
-  const FakeMetadataChangeList::Record& record1 = change_list.GetNthRecord(0);
-  EXPECT_EQ(FakeMetadataChangeList::UPDATE_METADATA, record1.action);
-  EXPECT_EQ("tag1", record1.tag);
-  EXPECT_EQ(1, record1.metadata.sequence_number());
+  EXPECT_EQ(1U, db().MetadataCount());
+  EXPECT_EQ(1, db().GetMetadata("tag1").sequence_number());
 
   OnInitialSyncDone();
   EXPECT_EQ(1U, GetNumCommitRequestLists());
@@ -753,23 +1063,20 @@ TEST_F(SharedModelTypeProcessorTest, NoCommitsUntilInitialSyncDone) {
 TEST_F(SharedModelTypeProcessorTest, Disconnect) {
   InitializeToReadyState();
 
-  FakeMetadataChangeList change_list;
-
   // The first item is fully committed.
-  WriteItem("tag1", "value1", &change_list);
-  ASSERT_TRUE(HasCommitRequestForTag("tag1"));
-  SuccessfulCommitResponse(GetLatestCommitRequestForTag("tag1"));
+  WriteItemAndAck("tag1", "value1");
 
   // The second item has a commit request in progress.
-  WriteItem("tag2", "value2", &change_list);
+  WriteItem("tag2", "value2");
   EXPECT_TRUE(HasCommitRequestForTag("tag2"));
 
   DisconnectSync();
 
   // The third item is added after stopping.
-  WriteItem("tag3", "value3", &change_list);
+  WriteItem("tag3", "value3");
 
-  Restart();
+  // Reconnect.
+  OnSyncStarting();
 
   EXPECT_EQ(1U, GetNumCommitRequestLists());
   EXPECT_EQ(2U, GetNthCommitRequestList(0).size());
@@ -791,24 +1098,20 @@ TEST_F(SharedModelTypeProcessorTest, Disconnect) {
 TEST_F(SharedModelTypeProcessorTest, DISABLED_Disable) {
   InitializeToReadyState();
 
-  FakeMetadataChangeList change_list;
-
   // The first item is fully committed.
-  WriteItem("tag1", "value1", &change_list);
-  ASSERT_TRUE(HasCommitRequestForTag("tag1"));
-  SuccessfulCommitResponse(GetLatestCommitRequestForTag("tag1"));
+  WriteItemAndAck("tag1", "value1");
 
   // The second item has a commit request in progress.
-  WriteItem("tag2", "value2", &change_list);
+  WriteItem("tag2", "value2");
   EXPECT_TRUE(HasCommitRequestForTag("tag2"));
 
   Disable();
 
   // The third item is added after disable.
-  WriteItem("tag3", "value3", &change_list);
+  WriteItem("tag3", "value3");
 
   // Now we re-enable.
-  Restart();
+  InitializeToReadyState();
 
   // Once we're ready to commit, all three local items should consider
   // themselves uncommitted and pending for commit.
@@ -860,7 +1163,7 @@ TEST_F(SharedModelTypeProcessorTest, DisableWithPendingUpdates) {
   ASSERT_TRUE(HasPendingUpdate("tag1"));
 
   Disable();
-  Restart();
+  InitializeToReadyState();
 
   EXPECT_EQ(0U, GetNumPendingUpdates());
   EXPECT_FALSE(HasPendingUpdate("tag1"));
@@ -875,7 +1178,7 @@ TEST_F(SharedModelTypeProcessorTest, DisconnectWithPendingUpdates) {
   ASSERT_TRUE(HasPendingUpdate("tag1"));
 
   DisconnectSync();
-  Restart();
+  OnSyncStarting();
 
   EXPECT_EQ(1U, GetNumPendingUpdates());
   EXPECT_TRUE(HasPendingUpdate("tag1"));
@@ -887,16 +1190,11 @@ TEST_F(SharedModelTypeProcessorTest, DisconnectWithPendingUpdates) {
 TEST_F(SharedModelTypeProcessorTest, DISABLED_ReEncryptCommitsWithNewKey) {
   InitializeToReadyState();
 
-  FakeMetadataChangeList change_list;
-
   // Commit an item.
-  WriteItem("tag1", "value1", &change_list);
-  ASSERT_TRUE(HasCommitRequestForTag("tag1"));
-  const CommitRequestData& tag1_v1_data = GetLatestCommitRequestForTag("tag1");
-  SuccessfulCommitResponse(tag1_v1_data);
+  WriteItemAndAck("tag1", "value1");
 
   // Create another item and don't wait for its commit response.
-  WriteItem("tag2", "value2", &change_list);
+  WriteItem("tag2", "value2");
 
   ASSERT_EQ(2U, GetNumCommitRequestLists());
 
diff --git a/sync/internal_api/test/fake_metadata_change_list.cc b/sync/internal_api/test/fake_metadata_change_list.cc
deleted file mode 100644
index ceba2cb..0000000
--- a/sync/internal_api/test/fake_metadata_change_list.cc
+++ /dev/null
@@ -1,59 +0,0 @@
-// Copyright 2016 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 <string>
-
-#include "sync/internal_api/public/test/fake_metadata_change_list.h"
-
-namespace syncer_v2 {
-
-FakeMetadataChangeList::FakeMetadataChangeList() {}
-
-FakeMetadataChangeList::~FakeMetadataChangeList() {}
-
-FakeMetadataChangeList::Record::Record() {}
-
-FakeMetadataChangeList::Record::~Record() {}
-
-void FakeMetadataChangeList::UpdateDataTypeState(
-    const sync_pb::DataTypeState& data_type_state) {
-  Record record;
-  record.action = UPDATE_DATA_TYPE_STATE;
-  record.data_type_state = data_type_state;
-  records_.push_back(record);
-}
-
-void FakeMetadataChangeList::ClearDataTypeState() {
-  Record record;
-  record.action = CLEAR_DATA_TYPE_STATE;
-  records_.push_back(record);
-}
-
-void FakeMetadataChangeList::UpdateMetadata(
-    const std::string& client_tag,
-    const sync_pb::EntityMetadata& metadata) {
-  Record record;
-  record.action = UPDATE_METADATA;
-  record.tag = client_tag;
-  record.metadata.CopyFrom(metadata);
-  records_.push_back(record);
-}
-
-void FakeMetadataChangeList::ClearMetadata(const std::string& client_tag) {
-  Record record;
-  record.action = CLEAR_METADATA;
-  record.tag = client_tag;
-  records_.push_back(record);
-}
-
-size_t FakeMetadataChangeList::GetNumRecords() const {
-  return records_.size();
-}
-
-const FakeMetadataChangeList::Record& FakeMetadataChangeList::GetNthRecord(
-    size_t n) const {
-  return records_[n];
-}
-
-}  // namespace syncer_v2
diff --git a/sync/sync_tests.gypi b/sync/sync_tests.gypi
index 6f3231a..122641e 100644
--- a/sync/sync_tests.gypi
+++ b/sync/sync_tests.gypi
@@ -183,7 +183,6 @@
         'test_support_sync_core',
       ],
       'sources': [
-        'internal_api/public/test/fake_metadata_change_list.h',
         'internal_api/public/test/fake_model_type_service.h',
         'internal_api/public/test/fake_sync_manager.h',
         'internal_api/public/test/model_type_store_test_util.h',
@@ -192,7 +191,6 @@
         'internal_api/public/test/test_entry_factory.h',
         'internal_api/public/test/test_internal_components_factory.h',
         'internal_api/public/test/test_user_share.h',
-        'internal_api/test/fake_metadata_change_list.cc',
         'internal_api/test/fake_model_type_service.cc',
         'internal_api/test/fake_sync_manager.cc',
         'internal_api/test/model_type_store_test_util.cc',