sync: Add non-blocking type encryption (retry)

Fixes some memory leak issues that were present in the first instance
of this CL.  Original description follows:

Introduces the framework for dealing with sync encryption in
non-blocking types.  Unlike directory sync types, non-blocking type
encryption only encrypts data before it is sent to the server.
Encrypting the data on-disk is a separate problem.

Adds code to the ModelTypeSyncWorker so it can access the directory's
cryptographer (through a CryptographerProvider interface) and use it to
encrypt entities before it sends them to the server.  If the
cryptographer is unable to encrypt with the desired key, the worker will
not commit until the cryptographer returns to a good state.

Adds the concept of a "desired encryption key" to the data type state.
When the cryptographer key to be used to encrypt a type changes, this
will be reflected in the data type state.  The ModelTypeSyncProxy is
responsible for ensuring that all items which have not yet been
encrypted with this desired key are enqueued for commit.

Makes the ModelTypeSyncWorker, EntityTracker, and ModelTypeSyncProxy
collaborate on the management of undecryptable (inapplicable) updates.
The EntityTracker keeps track of their version numbers and content, and
prevents the committing of new items to the server until the
inapplicable update has been dealt with.  The ModelTypeSyncProxy is
responsible for saving inapplicable updates across restarts.

This CL alone is not enough to enable encryption support for
non-blocking types.  It requires additional code to hook up the
ModelTypeSyncWorkers to receive cryptographer events.  This will be
added in a future commit.  In the meantime, this CL includes plenty
of unit tests to verify the functionality that's being added.

BUG=351005

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

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

32 files changed, 1250 insertions, 87 deletions

diff --git a/components/sync_driver/non_blocking_data_type_controller_unittest.cc b/components/sync_driver/non_blocking_data_type_controller_unittest.cc
index 9cf9489a..10987bd 100644
--- a/components/sync_driver/non_blocking_data_type_controller_unittest.cc
+++ b/components/sync_driver/non_blocking_data_type_controller_unittest.cc
@@ -85,6 +85,7 @@ class MockSyncContextProxy : public syncer::SyncContextProxy {
   virtual void ConnectTypeToSync(
       syncer::ModelType type,
       const syncer::DataTypeState& data_type_state,
+      const syncer::UpdateResponseDataList& saved_pending_updates,
       const base::WeakPtr<syncer::ModelTypeSyncProxyImpl>& type_proxy)
       OVERRIDE {
     // Normally we'd use MessageLoopProxy::current() as the TaskRunner argument
diff --git a/sync/engine/entity_tracker.cc b/sync/engine/entity_tracker.cc
index 0f10906..bad76f9 100644
--- a/sync/engine/entity_tracker.cc
+++ b/sync/engine/entity_tracker.cc
@@ -193,8 +193,14 @@ void EntityTracker::ReceiveCommitResponse(const std::string& response_id,
 }
 
 void EntityTracker::ReceiveUpdate(int64 version) {
-  highest_gu_response_version_ =
-      std::max(highest_gu_response_version_, version);
+  if (version <= highest_gu_response_version_)
+    return;
+
+  highest_gu_response_version_ = version;
+
+  // Got an applicable update newer than any pending updates.  It must be safe
+  // to discard the old pending update, if there was one.
+  ClearPendingUpdate();
 
   if (IsInConflict()) {
     // Incoming update clobbers the pending commit on the sync thread.
@@ -203,10 +209,35 @@ void EntityTracker::ReceiveUpdate(int64 version) {
   }
 }
 
+bool EntityTracker::ReceivePendingUpdate(const UpdateResponseData& data) {
+  if (data.response_version < highest_gu_response_version_)
+    return false;
+
+  highest_gu_response_version_ = data.response_version;
+  pending_update_.reset(new UpdateResponseData(data));
+  ClearPendingCommit();
+  return true;
+}
+
+bool EntityTracker::HasPendingUpdate() const {
+  return !!pending_update_;
+}
+
+UpdateResponseData EntityTracker::GetPendingUpdate() const {
+  return *pending_update_;
+}
+
+void EntityTracker::ClearPendingUpdate() {
+  pending_update_.reset();
+}
+
 bool EntityTracker::IsInConflict() const {
   if (!is_commit_pending_)
     return false;
 
+  if (HasPendingUpdate())
+    return true;
+
   if (highest_gu_response_version_ <= highest_commit_response_version_) {
     // The most recent server state was created in a commit made by this
     // client.  We're fully up to date, and therefore not in conflict.
diff --git a/sync/engine/entity_tracker.h b/sync/engine/entity_tracker.h
index db2d09b..e969ff9 100644
--- a/sync/engine/entity_tracker.h
+++ b/sync/engine/entity_tracker.h
@@ -8,8 +8,10 @@
 #include <string>
 
 #include "base/basictypes.h"
+#include "base/memory/scoped_ptr.h"
 #include "base/time/time.h"
 #include "sync/base/sync_export.h"
+#include "sync/internal_api/public/non_blocking_sync_common.h"
 #include "sync/protocol/sync.pb.h"
 
 namespace syncer {
@@ -81,6 +83,20 @@ class SYNC_EXPORT EntityTracker {
   // Handles receipt of an update from the server.
   void ReceiveUpdate(int64 version);
 
+  // Handles the receipt of an pending update from the server.
+  //
+  // Returns true if the tracker decides this item is worth keeping.  Returns
+  // false if the item is discarded, which could happen if the version number
+  // is out of date.
+  bool ReceivePendingUpdate(const UpdateResponseData& data);
+
+  // Functions to fetch the latest pending update.
+  bool HasPendingUpdate() const;
+  UpdateResponseData GetPendingUpdate() const;
+
+  // Clears the pending update.  Allows us to resume regular commit behavior.
+  void ClearPendingUpdate();
+
  private:
   // Initializes received update state.  Does not initialize state related to
   // pending commits and sets |is_commit_pending_| to false.
@@ -146,6 +162,11 @@ class SYNC_EXPORT EntityTracker {
   bool deleted_;
   sync_pb::EntitySpecifics specifics_;
 
+  // An update for this item which can't be applied right now.  The presence of
+  // an pending update prevents commits.  As of this writing, the only source
+  // of pending updates is updates we can't decrypt right now.
+  scoped_ptr<UpdateResponseData> pending_update_;
+
   DISALLOW_COPY_AND_ASSIGN(EntityTracker);
 };
 
diff --git a/sync/engine/model_type_entity.cc b/sync/engine/model_type_entity.cc
index 5acf5db..2f758a8 100644
--- a/sync/engine/model_type_entity.cc
+++ b/sync/engine/model_type_entity.cc
@@ -24,7 +24,8 @@ scoped_ptr<ModelTypeEntity> ModelTypeEntity::NewLocalItem(
       specifics,
       false,
       now,
-      now));
+      now,
+      std::string()));
 }
 
 scoped_ptr<ModelTypeEntity> ModelTypeEntity::FromServerUpdate(
@@ -35,7 +36,8 @@ scoped_ptr<ModelTypeEntity> ModelTypeEntity::FromServerUpdate(
     const sync_pb::EntitySpecifics& specifics,
     bool deleted,
     base::Time ctime,
-    base::Time mtime) {
+    base::Time mtime,
+    const std::string& encryption_key_name) {
   return scoped_ptr<ModelTypeEntity>(new ModelTypeEntity(0,
                                                          0,
                                                          0,
@@ -47,7 +49,8 @@ scoped_ptr<ModelTypeEntity> ModelTypeEntity::FromServerUpdate(
                                                          specifics,
                                                          deleted,
                                                          ctime,
-                                                         mtime));
+                                                         mtime,
+                                                         encryption_key_name));
 }
 
 ModelTypeEntity::ModelTypeEntity(int64 sequence_number,
@@ -61,7 +64,8 @@ ModelTypeEntity::ModelTypeEntity(int64 sequence_number,
                                  const sync_pb::EntitySpecifics& specifics,
                                  bool deleted,
                                  base::Time ctime,
-                                 base::Time mtime)
+                                 base::Time mtime,
+                                 const std::string& encryption_key_name)
     : sequence_number_(sequence_number),
       commit_requested_sequence_number_(commit_requested_sequence_number),
       acked_sequence_number_(acked_sequence_number),
@@ -73,7 +77,8 @@ ModelTypeEntity::ModelTypeEntity(int64 sequence_number,
       specifics_(specifics),
       deleted_(deleted),
       ctime_(ctime),
-      mtime_(mtime) {
+      mtime_(mtime),
+      encryption_key_name_(encryption_key_name) {
 }
 
 ModelTypeEntity::~ModelTypeEntity() {
@@ -103,7 +108,8 @@ void ModelTypeEntity::ApplyUpdateFromServer(
     int64 update_version,
     bool deleted,
     const sync_pb::EntitySpecifics& specifics,
-    base::Time mtime) {
+    base::Time mtime,
+    const std::string& encryption_key_name) {
   // There was a conflict and the server just won it.
   // This implicitly acks all outstanding commits because a received update
   // will clobber any pending commits on the sync thread.
@@ -121,6 +127,15 @@ void ModelTypeEntity::MakeLocalChange(
   specifics_ = specifics;
 }
 
+void ModelTypeEntity::UpdateDesiredEncryptionKey(const std::string& name) {
+  if (encryption_key_name_ == name)
+    return;
+
+  // Schedule commit with the expectation that the worker will re-encrypt with
+  // the latest encryption key as it does.
+  sequence_number_++;
+}
+
 void ModelTypeEntity::Delete() {
   sequence_number_++;
   specifics_.Clear();
@@ -144,12 +159,15 @@ void ModelTypeEntity::SetCommitRequestInProgress() {
   commit_requested_sequence_number_ = sequence_number_;
 }
 
-void ModelTypeEntity::ReceiveCommitResponse(const std::string& id,
-                                            int64 sequence_number,
-                                            int64 response_version) {
+void ModelTypeEntity::ReceiveCommitResponse(
+    const std::string& id,
+    int64 sequence_number,
+    int64 response_version,
+    const std::string& encryption_key_name) {
   id_ = id;  // The server can assign us a new ID in a commit response.
   acked_sequence_number_ = sequence_number;
   base_version_ = response_version;
+  encryption_key_name_ = encryption_key_name;
 }
 
 void ModelTypeEntity::ClearTransientSyncState() {
diff --git a/sync/engine/model_type_entity.h b/sync/engine/model_type_entity.h
index abc67a9..e05d469 100644
--- a/sync/engine/model_type_entity.h
+++ b/sync/engine/model_type_entity.h
@@ -46,7 +46,8 @@ class SYNC_EXPORT_PRIVATE ModelTypeEntity {
       const sync_pb::EntitySpecifics& specifics,
       bool deleted,
       base::Time ctime,
-      base::Time mtime);
+      base::Time mtime,
+      const std::string& encryption_key_name);
 
   // TODO(rlarocque): Implement FromDisk constructor when we implement storage.
 
@@ -79,11 +80,17 @@ class SYNC_EXPORT_PRIVATE ModelTypeEntity {
   void ApplyUpdateFromServer(int64 update_version,
                              bool deleted,
                              const sync_pb::EntitySpecifics& specifics,
-                             base::Time mtime);
+                             base::Time mtime,
+                             const std::string& encryption_key_name);
 
   // Applies a local change to this item.
   void MakeLocalChange(const sync_pb::EntitySpecifics& specifics);
 
+  // Schedule a commit if the |name| does not match this item's last known
+  // encryption key.  The worker that performs the commit is expected to
+  // encrypt the item using the latest available key.
+  void UpdateDesiredEncryptionKey(const std::string& name);
+
   // Applies a local deletion to this item.
   void Delete();
 
@@ -104,7 +111,8 @@ class SYNC_EXPORT_PRIVATE ModelTypeEntity {
   // reached the server.
   void ReceiveCommitResponse(const std::string& id,
                              int64 sequence_number,
-                             int64 response_version);
+                             int64 response_version,
+                             const std::string& encryption_key_name);
 
   // Clears any in-memory sync state associated with outstanding commits.
   void ClearTransientSyncState();
@@ -124,7 +132,8 @@ class SYNC_EXPORT_PRIVATE ModelTypeEntity {
                   const sync_pb::EntitySpecifics& specifics,
                   bool deleted,
                   base::Time ctime,
-                  base::Time mtime);
+                  base::Time mtime,
+                  const std::string& encryption_key_name);
 
   // A sequence number used to track in-progress commits.  Each local change
   // increments this number.
@@ -185,6 +194,10 @@ class SYNC_EXPORT_PRIVATE ModelTypeEntity {
   // doesn't bother to inspect their values.
   base::Time ctime_;
   base::Time mtime_;
+
+  // The name of the encryption key used to encrypt this item on the server.
+  // Empty when no encryption is in use.
+  std::string encryption_key_name_;
 };
 
 }  // namespace syncer
diff --git a/sync/engine/model_type_entity_unittest.cc b/sync/engine/model_type_entity_unittest.cc
index c605732..817bc84 100644
--- a/sync/engine/model_type_entity_unittest.cc
+++ b/sync/engine/model_type_entity_unittest.cc
@@ -64,7 +64,8 @@ TEST_F(ModelTypeEntityTest, FromServerUpdate) {
                                         specifics,
                                         false,
                                         kCtime,
-                                        kMtime));
+                                        kMtime,
+                                        std::string()));
 
   EXPECT_TRUE(entity->IsWriteRequired());
   EXPECT_FALSE(entity->IsUnsynced());
@@ -87,7 +88,8 @@ TEST_F(ModelTypeEntityTest, TombstoneUpdate) {
                                         sync_pb::EntitySpecifics(),
                                         true,
                                         kCtime,
-                                        kMtime));
+                                        kMtime,
+                                        std::string()));
 
   EXPECT_TRUE(entity->IsWriteRequired());
   EXPECT_FALSE(entity->IsUnsynced());
@@ -107,13 +109,15 @@ TEST_F(ModelTypeEntityTest, ApplyUpdate) {
                                         specifics,
                                         false,
                                         kCtime,
-                                        kMtime));
+                                        kMtime,
+                                        std::string()));
 
   // A deletion update one version later.
   entity->ApplyUpdateFromServer(11,
                                 true,
                                 sync_pb::EntitySpecifics(),
-                                kMtime + base::TimeDelta::FromSeconds(10));
+                                kMtime + base::TimeDelta::FromSeconds(10),
+                                std::string());
 
   EXPECT_TRUE(entity->IsWriteRequired());
   EXPECT_FALSE(entity->IsUnsynced());
@@ -130,7 +134,8 @@ TEST_F(ModelTypeEntityTest, LocalChange) {
                                         specifics,
                                         false,
                                         kCtime,
-                                        kMtime));
+                                        kMtime,
+                                        std::string()));
 
   sync_pb::EntitySpecifics specifics2;
   specifics2.CopyFrom(specifics);
@@ -156,7 +161,8 @@ TEST_F(ModelTypeEntityTest, LocalDeletion) {
                                         specifics,
                                         false,
                                         kCtime,
-                                        kMtime));
+                                        kMtime,
+                                        std::string()));
 
   entity->Delete();
 
diff --git a/sync/engine/model_type_sync_proxy.h b/sync/engine/model_type_sync_proxy.h
index 706fbc7..3c6e079 100644
--- a/sync/engine/model_type_sync_proxy.h
+++ b/sync/engine/model_type_sync_proxy.h
@@ -21,7 +21,8 @@ class SYNC_EXPORT_PRIVATE ModelTypeSyncProxy {
       const CommitResponseDataList& response_list) = 0;
   virtual void OnUpdateReceived(
       const DataTypeState& type_state,
-      const UpdateResponseDataList& response_list) = 0;
+      const UpdateResponseDataList& response_list,
+      const UpdateResponseDataList& pending_updates) = 0;
 };
 
 }  // namespace syncer
diff --git a/sync/engine/model_type_sync_proxy_impl.cc b/sync/engine/model_type_sync_proxy_impl.cc
index 7b2148a..7ea5d52 100644
--- a/sync/engine/model_type_sync_proxy_impl.cc
+++ b/sync/engine/model_type_sync_proxy_impl.cc
@@ -18,6 +18,7 @@ ModelTypeSyncProxyImpl::ModelTypeSyncProxyImpl(ModelType type)
       is_preferred_(false),
       is_connected_(false),
       entities_deleter_(&entities_),
+      pending_updates_map_deleter_(&pending_updates_map_),
       weak_ptr_factory_for_ui_(this),
       weak_ptr_factory_for_sync_(this) {
 }
@@ -51,10 +52,12 @@ void ModelTypeSyncProxyImpl::Enable(
   data_type_state_.progress_marker.set_data_type_id(
       GetSpecificsFieldNumberFromModelType(type_));
 
+  UpdateResponseDataList saved_pending_updates = GetPendingUpdates();
   sync_context_proxy_ = sync_context_proxy.Pass();
   sync_context_proxy_->ConnectTypeToSync(
       GetModelType(),
       data_type_state_,
+      saved_pending_updates,
       weak_ptr_factory_for_sync_.GetWeakPtr());
 }
 
@@ -180,16 +183,18 @@ void ModelTypeSyncProxyImpl::OnCommitCompleted(
     } else {
       it->second->ReceiveCommitResponse(response_data.id,
                                         response_data.sequence_number,
-                                        response_data.response_version);
+                                        response_data.response_version,
+                                        data_type_state_.encryption_key_name);
     }
   }
 }
 
 void ModelTypeSyncProxyImpl::OnUpdateReceived(
     const DataTypeState& data_type_state,
-    const UpdateResponseDataList& response_list) {
-  bool initial_sync_just_finished =
-      !data_type_state_.initial_sync_done && data_type_state.initial_sync_done;
+    const UpdateResponseDataList& response_list,
+    const UpdateResponseDataList& pending_updates) {
+  bool got_new_encryption_requirements = data_type_state_.encryption_key_name !=
+                                         data_type_state.encryption_key_name;
 
   data_type_state_ = data_type_state;
 
@@ -199,6 +204,14 @@ void ModelTypeSyncProxyImpl::OnUpdateReceived(
     const UpdateResponseData& response_data = *list_it;
     const std::string& client_tag_hash = response_data.client_tag_hash;
 
+    UpdateMap::iterator old_it = pending_updates_map_.find(client_tag_hash);
+    if (old_it != pending_updates_map_.end()) {
+      // If we're being asked to apply an update to this entity, this overrides
+      // the previous pending updates.
+      delete old_it->second;
+      pending_updates_map_.erase(old_it);
+    }
+
     EntityMap::iterator it = entities_.find(client_tag_hash);
     if (it == entities_.end()) {
       scoped_ptr<ModelTypeEntity> entity =
@@ -209,22 +222,74 @@ void ModelTypeSyncProxyImpl::OnUpdateReceived(
                                             response_data.specifics,
                                             response_data.deleted,
                                             response_data.ctime,
-                                            response_data.mtime);
+                                            response_data.mtime,
+                                            response_data.encryption_key_name);
       entities_.insert(std::make_pair(client_tag_hash, entity.release()));
     } else {
       ModelTypeEntity* entity = it->second;
       entity->ApplyUpdateFromServer(response_data.response_version,
                                     response_data.deleted,
                                     response_data.specifics,
-                                    response_data.mtime);
+                                    response_data.mtime,
+                                    response_data.encryption_key_name);
+
       // TODO: Do something special when conflicts are detected.
     }
+
+    // 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) {
+      EntityMap::iterator it2 = entities_.find(client_tag_hash);
+      it2->second->UpdateDesiredEncryptionKey(
+          data_type_state_.encryption_key_name);
+    }
   }
 
-  if (initial_sync_just_finished)
-    FlushPendingCommitRequests();
+  // Save pending updates in the appropriate data structure.
+  for (UpdateResponseDataList::const_iterator list_it = pending_updates.begin();
+       list_it != pending_updates.end();
+       ++list_it) {
+    const UpdateResponseData& update = *list_it;
+    const std::string& client_tag_hash = update.client_tag_hash;
+
+    UpdateMap::iterator lookup_it = pending_updates_map_.find(client_tag_hash);
+    if (lookup_it == pending_updates_map_.end()) {
+      pending_updates_map_.insert(
+          std::make_pair(client_tag_hash, new UpdateResponseData(update)));
+    } else if (lookup_it->second->response_version <= update.response_version) {
+      delete lookup_it->second;
+      pending_updates_map_.erase(lookup_it);
+      pending_updates_map_.insert(
+          std::make_pair(client_tag_hash, new UpdateResponseData(update)));
+    } else {
+      // Received update is stale, do not overwrite existing.
+    }
+  }
+
+  if (got_new_encryption_requirements) {
+    for (EntityMap::iterator it = entities_.begin(); it != entities_.end();
+         ++it) {
+      it->second->UpdateDesiredEncryptionKey(
+          data_type_state_.encryption_key_name);
+    }
+  }
+
+  // We may have new reasons to commit by the time this function is done.
+  FlushPendingCommitRequests();
 
   // TODO: Inform the model of the new or updated data.
+  // TODO: Persist the new data on disk.
+}
+
+UpdateResponseDataList ModelTypeSyncProxyImpl::GetPendingUpdates() {
+  UpdateResponseDataList pending_updates_list;
+  for (UpdateMap::const_iterator it = pending_updates_map_.begin();
+       it != pending_updates_map_.end();
+       ++it) {
+    pending_updates_list.push_back(*it->second);
+  }
+  return pending_updates_list;
 }
 
 void ModelTypeSyncProxyImpl::ClearTransientSyncState() {
@@ -239,7 +304,7 @@ void ModelTypeSyncProxyImpl::ClearSyncState() {
        ++it) {
     it->second->ClearSyncState();
   }
-
+  STLDeleteValues(&pending_updates_map_);
   data_type_state_ = DataTypeState();
 }
 
diff --git a/sync/engine/model_type_sync_proxy_impl.h b/sync/engine/model_type_sync_proxy_impl.h
index 2390eda..f160669 100644
--- a/sync/engine/model_type_sync_proxy_impl.h
+++ b/sync/engine/model_type_sync_proxy_impl.h
@@ -73,7 +73,16 @@ class SYNC_EXPORT_PRIVATE ModelTypeSyncProxyImpl : base::NonThreadSafe {
   // Informs this object that there are some incoming updates is should
   // handle.
   void OnUpdateReceived(const DataTypeState& type_state,
-                        const UpdateResponseDataList& response_list);
+                        const UpdateResponseDataList& response_list,
+                        const UpdateResponseDataList& pending_updates);
+
+  // Returns the list of pending updates.
+  //
+  // This is used as a helper function, but it's public mainly for testing.
+  // The current test harness setup doesn't allow us to test the data that the
+  // proxy sends to the worker during initialization, so we use this to inspect
+  // its state instead.
+  UpdateResponseDataList GetPendingUpdates();
 
   // Returns the long-lived WeakPtr that is intended to be registered with the
   // ProfileSyncService.
@@ -81,6 +90,7 @@ class SYNC_EXPORT_PRIVATE ModelTypeSyncProxyImpl : base::NonThreadSafe {
 
  private:
   typedef std::map<std::string, ModelTypeEntity*> EntityMap;
+  typedef std::map<std::string, UpdateResponseData*> UpdateMap;
 
   // Sends all commit requests that are due to be sent to the sync thread.
   void FlushPendingCommitRequests();
@@ -123,6 +133,12 @@ class SYNC_EXPORT_PRIVATE ModelTypeSyncProxyImpl : base::NonThreadSafe {
   EntityMap entities_;
   STLValueDeleter<EntityMap> entities_deleter_;
 
+  // A set of updates that can not be applied at this time.  These are never
+  // used by the model.  They are kept here only so we can save and restore
+  // them across restarts, and keep them in sync with our progress markers.
+  UpdateMap pending_updates_map_;
+  STLValueDeleter<UpdateMap> pending_updates_map_deleter_;
+
   // We use two different WeakPtrFactories because we want the pointers they
   // issue to have different lifetimes.  When asked to disconnect from the sync
   // thread, we want to make sure that no tasks generated as part of the
diff --git a/sync/engine/model_type_sync_proxy_impl_unittest.cc b/sync/engine/model_type_sync_proxy_impl_unittest.cc
index 0597657..860cc70 100644
--- a/sync/engine/model_type_sync_proxy_impl_unittest.cc
+++ b/sync/engine/model_type_sync_proxy_impl_unittest.cc
@@ -77,6 +77,22 @@ class ModelTypeSyncProxyImplTest : public ::testing::Test {
                         const std::string& value);
   void TombstoneFromServer(int64 version_offset, const std::string& tag);
 
+  // Emulate the receipt of pending updates from the server.
+  // Pending updates are usually caused by a temporary decryption failure.
+  void PendingUpdateFromServer(int64 version_offset,
+                               const std::string& tag,
+                               const std::string& value,
+                               const std::string& key_name);
+
+  // Returns true if the proxy has an pending update with specified tag.
+  bool HasPendingUpdate(const std::string& tag) const;
+
+  // Returns the pending update with the specified tag.
+  UpdateResponseData GetPendingUpdate(const std::string& tag) const;
+
+  // Returns the number of pending updates.
+  size_t GetNumPendingUpdates() const;
+
   // Read emitted commit requests as batches.
   size_t GetNumCommitRequestLists();
   CommitRequestDataList GetNthCommitRequestList(size_t n);
@@ -88,10 +104,22 @@ class ModelTypeSyncProxyImplTest : public ::testing::Test {
   // Sends the type sync proxy a successful commit response.
   void SuccessfulCommitResponse(const CommitRequestData& request_data);
 
+  // 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);
+
+  // Sets the key_name that the mock ModelTypeSyncWorker will claim is in use
+  // when receiving items.
+  void SetServerEncryptionKey(const std::string& key_name);
+
  private:
   static std::string GenerateTagHash(const std::string& tag);
   static sync_pb::EntitySpecifics GenerateSpecifics(const std::string& tag,
                                                     const std::string& value);
+  static sync_pb::EntitySpecifics GenerateEncryptedSpecifics(
+      const std::string& tag,
+      const std::string& value,
+      const std::string& key_name);
 
   int64 GetServerVersion(const std::string& tag);
   void SetServerVersion(const std::string& tag, int64 version);
@@ -141,7 +169,7 @@ void ModelTypeSyncProxyImplTest::Disable() {
 void ModelTypeSyncProxyImplTest::ReEnable() {
   DCHECK(!type_sync_proxy_->IsConnected());
 
-  // Prepare a new NonBlockingTypeProcesorCore instance, just as we would
+  // Prepare a new MockModelTypeSyncWorker instance, just as we would
   // if this happened in the real world.
   mock_worker_ = new MockModelTypeSyncWorker();
   injectable_sync_context_proxy_.reset(
@@ -165,7 +193,8 @@ void ModelTypeSyncProxyImplTest::OnInitialSyncDone() {
   data_type_state_.initial_sync_done = true;
   UpdateResponseDataList empty_update_list;
 
-  type_sync_proxy_->OnUpdateReceived(data_type_state_, empty_update_list);
+  type_sync_proxy_->OnUpdateReceived(
+      data_type_state_, empty_update_list, empty_update_list);
 }
 
 void ModelTypeSyncProxyImplTest::UpdateFromServer(int64 version_offset,
@@ -177,7 +206,25 @@ void ModelTypeSyncProxyImplTest::UpdateFromServer(int64 version_offset,
 
   UpdateResponseDataList list;
   list.push_back(data);
-  type_sync_proxy_->OnUpdateReceived(data_type_state_, list);
+  type_sync_proxy_->OnUpdateReceived(
+      data_type_state_, list, UpdateResponseDataList());
+}
+
+void ModelTypeSyncProxyImplTest::PendingUpdateFromServer(
+    int64 version_offset,
+    const std::string& tag,
+    const std::string& value,
+    const std::string& key_name) {
+  const std::string tag_hash = GenerateTagHash(tag);
+  UpdateResponseData data = mock_worker_->UpdateFromServer(
+      version_offset,
+      tag_hash,
+      GenerateEncryptedSpecifics(tag, value, key_name));
+
+  UpdateResponseDataList list;
+  list.push_back(data);
+  type_sync_proxy_->OnUpdateReceived(
+      data_type_state_, UpdateResponseDataList(), list);
 }
 
 void ModelTypeSyncProxyImplTest::TombstoneFromServer(int64 version_offset,
@@ -190,7 +237,40 @@ void ModelTypeSyncProxyImplTest::TombstoneFromServer(int64 version_offset,
 
   UpdateResponseDataList list;
   list.push_back(data);
-  type_sync_proxy_->OnUpdateReceived(data_type_state_, list);
+  type_sync_proxy_->OnUpdateReceived(
+      data_type_state_, list, UpdateResponseDataList());
+}
+
+bool ModelTypeSyncProxyImplTest::HasPendingUpdate(
+    const std::string& tag) const {
+  const std::string client_tag_hash = GenerateTagHash(tag);
+  const UpdateResponseDataList list = type_sync_proxy_->GetPendingUpdates();
+  for (UpdateResponseDataList::const_iterator it = list.begin();
+       it != list.end();
+       ++it) {
+    if (it->client_tag_hash == client_tag_hash)
+      return true;
+  }
+  return false;
+}
+
+UpdateResponseData ModelTypeSyncProxyImplTest::GetPendingUpdate(
+    const std::string& tag) const {
+  DCHECK(HasPendingUpdate(tag));
+  const std::string client_tag_hash = GenerateTagHash(tag);
+  const UpdateResponseDataList list = type_sync_proxy_->GetPendingUpdates();
+  for (UpdateResponseDataList::const_iterator it = list.begin();
+       it != list.end();
+       ++it) {
+    if (it->client_tag_hash == client_tag_hash)
+      return *it;
+  }
+  NOTREACHED();
+  return UpdateResponseData();
+}
+
+size_t ModelTypeSyncProxyImplTest::GetNumPendingUpdates() const {
+  return type_sync_proxy_->GetPendingUpdates().size();
 }
 
 void ModelTypeSyncProxyImplTest::SuccessfulCommitResponse(
@@ -200,6 +280,18 @@ void ModelTypeSyncProxyImplTest::SuccessfulCommitResponse(
   type_sync_proxy_->OnCommitCompleted(data_type_state_, list);
 }
 
+void ModelTypeSyncProxyImplTest::UpdateDesiredEncryptionKey(
+    const std::string& key_name) {
+  data_type_state_.encryption_key_name = key_name;
+  type_sync_proxy_->OnUpdateReceived(
+      data_type_state_, UpdateResponseDataList(), UpdateResponseDataList());
+}
+
+void ModelTypeSyncProxyImplTest::SetServerEncryptionKey(
+    const std::string& key_name) {
+  mock_worker_->SetServerEncryptionKey(key_name);
+}
+
 std::string ModelTypeSyncProxyImplTest::GenerateTagHash(
     const std::string& tag) {
   return syncable::GenerateSyncableHash(kModelType, tag);
@@ -214,6 +306,19 @@ sync_pb::EntitySpecifics ModelTypeSyncProxyImplTest::GenerateSpecifics(
   return specifics;
 }
 
+// These tests never decrypt anything, so we can get away with faking the
+// encryption for now.
+sync_pb::EntitySpecifics ModelTypeSyncProxyImplTest::GenerateEncryptedSpecifics(
+    const std::string& tag,
+    const std::string& value,
+    const std::string& key_name) {
+  sync_pb::EntitySpecifics specifics;
+  AddDefaultFieldValue(kModelType, &specifics);
+  specifics.mutable_encrypted()->set_key_name(key_name);
+  specifics.mutable_encrypted()->set_blob("BLOB" + key_name);
+  return specifics;
+}
+
 size_t ModelTypeSyncProxyImplTest::GetNumCommitRequestLists() {
   return mock_worker_->GetNumCommitRequestLists();
 }
@@ -460,4 +565,131 @@ TEST_F(ModelTypeSyncProxyImplTest, Disable) {
   EXPECT_TRUE(HasCommitRequestForTag("tag3"));
 }
 
+// Test receipt of pending updates.
+TEST_F(ModelTypeSyncProxyImplTest, ReceivePendingUpdates) {
+  InitializeToReadyState();
+
+  EXPECT_FALSE(HasPendingUpdate("tag1"));
+  EXPECT_EQ(0U, GetNumPendingUpdates());
+
+  // Receive a pending update.
+  PendingUpdateFromServer(5, "tag1", "value1", "key1");
+  EXPECT_EQ(1U, GetNumPendingUpdates());
+  ASSERT_TRUE(HasPendingUpdate("tag1"));
+  UpdateResponseData data1 = GetPendingUpdate("tag1");
+  EXPECT_EQ(5, data1.response_version);
+
+  // Receive an updated version of a pending update.
+  // It should overwrite the existing item.
+  PendingUpdateFromServer(10, "tag1", "value15", "key1");
+  EXPECT_EQ(1U, GetNumPendingUpdates());
+  ASSERT_TRUE(HasPendingUpdate("tag1"));
+  UpdateResponseData data2 = GetPendingUpdate("tag1");
+  EXPECT_EQ(15, data2.response_version);
+
+  // Receive a stale version of a pending update.
+  // It should have no effect.
+  PendingUpdateFromServer(-3, "tag1", "value12", "key1");
+  EXPECT_EQ(1U, GetNumPendingUpdates());
+  ASSERT_TRUE(HasPendingUpdate("tag1"));
+  UpdateResponseData data3 = GetPendingUpdate("tag1");
+  EXPECT_EQ(15, data3.response_version);
+}
+
+// Test that Disable clears pending update state.
+TEST_F(ModelTypeSyncProxyImplTest, DisableWithPendingUpdates) {
+  InitializeToReadyState();
+
+  PendingUpdateFromServer(5, "tag1", "value1", "key1");
+  EXPECT_EQ(1U, GetNumPendingUpdates());
+  ASSERT_TRUE(HasPendingUpdate("tag1"));
+
+  Disable();
+  ReEnable();
+
+  EXPECT_EQ(0U, GetNumPendingUpdates());
+  EXPECT_FALSE(HasPendingUpdate("tag1"));
+}
+
+// Test that Disconnect does not clear pending update state.
+TEST_F(ModelTypeSyncProxyImplTest, DisconnectWithPendingUpdates) {
+  InitializeToReadyState();
+
+  PendingUpdateFromServer(5, "tag1", "value1", "key1");
+  EXPECT_EQ(1U, GetNumPendingUpdates());
+  ASSERT_TRUE(HasPendingUpdate("tag1"));
+
+  Disconnect();
+  ReEnable();
+
+  EXPECT_EQ(1U, GetNumPendingUpdates());
+  EXPECT_TRUE(HasPendingUpdate("tag1"));
+}
+
+// Test re-encrypt everything when desired encryption key changes.
+TEST_F(ModelTypeSyncProxyImplTest, ReEncryptCommitsWithNewKey) {
+  InitializeToReadyState();
+
+  // Commit an item.
+  WriteItem("tag1", "value1");
+  ASSERT_TRUE(HasCommitRequestForTag("tag1"));
+  const CommitRequestData& tag1_v1_data = GetLatestCommitRequestForTag("tag1");
+  SuccessfulCommitResponse(tag1_v1_data);
+
+  // Create another item and don't wait for its commit response.
+  WriteItem("tag2", "value2");
+
+  ASSERT_EQ(2U, GetNumCommitRequestLists());
+
+  // Receive notice that the account's desired encryption key has changed.
+  UpdateDesiredEncryptionKey("k1");
+
+  // That should trigger a new commit request.
+  ASSERT_EQ(3U, GetNumCommitRequestLists());
+  EXPECT_EQ(2U, GetNthCommitRequestList(2).size());
+
+  const CommitRequestData& tag1_enc = GetLatestCommitRequestForTag("tag1");
+  const CommitRequestData& tag2_enc = GetLatestCommitRequestForTag("tag2");
+
+  SuccessfulCommitResponse(tag1_enc);
+  SuccessfulCommitResponse(tag2_enc);
+
+  // And that should be the end of it.
+  ASSERT_EQ(3U, GetNumCommitRequestLists());
+}
+
+// Test receipt of updates with new and old keys.
+TEST_F(ModelTypeSyncProxyImplTest, ReEncryptUpdatesWithNewKey) {
+  InitializeToReadyState();
+
+  // Receive an unencrpted update.
+  UpdateFromServer(5, "no_enc", "value1");
+
+  ASSERT_EQ(0U, GetNumCommitRequestLists());
+
+  // Set desired encryption key to k2 to force updates to some items.
+  UpdateDesiredEncryptionKey("k2");
+
+  ASSERT_EQ(1U, GetNumCommitRequestLists());
+  EXPECT_EQ(1U, GetNthCommitRequestList(0).size());
+  EXPECT_TRUE(HasCommitRequestForTag("no_enc"));
+
+  // Receive an update that was encrypted with key k1.
+  SetServerEncryptionKey("k1");
+  UpdateFromServer(10, "enc_k1", "value1");
+
+  // Receipt of updates encrypted with old key also forces a re-encrypt commit.
+  ASSERT_EQ(2U, GetNumCommitRequestLists());
+  EXPECT_EQ(1U, GetNthCommitRequestList(1).size());
+  EXPECT_TRUE(HasCommitRequestForTag("enc_k1"));
+
+  // Receive an update that was encrypted with key k2.
+  SetServerEncryptionKey("k2");
+  UpdateFromServer(15, "enc_k2", "value1");
+
+  // That was the correct key, so no re-encryption is required.
+  EXPECT_EQ(2U, GetNumCommitRequestLists());
+  EXPECT_FALSE(HasCommitRequestForTag("enc_k2"));
+}
+
 }  // namespace syncer
diff --git a/sync/engine/model_type_sync_worker_impl.cc b/sync/engine/model_type_sync_worker_impl.cc
index a5402c2..a58400f4 100644
--- a/sync/engine/model_type_sync_worker_impl.cc
+++ b/sync/engine/model_type_sync_worker_impl.cc
@@ -13,6 +13,7 @@
 #include "sync/engine/model_type_sync_proxy.h"
 #include "sync/engine/non_blocking_type_commit_contribution.h"
 #include "sync/syncable/syncable_util.h"
+#include "sync/util/cryptographer.h"
 #include "sync/util/time.h"
 
 namespace syncer {
@@ -20,11 +21,14 @@ namespace syncer {
 ModelTypeSyncWorkerImpl::ModelTypeSyncWorkerImpl(
     ModelType type,
     const DataTypeState& initial_state,
+    const UpdateResponseDataList& saved_pending_updates,
+    CryptographerProvider* cryptographer_provider,
     NudgeHandler* nudge_handler,
     scoped_ptr<ModelTypeSyncProxy> type_sync_proxy)
     : type_(type),
       data_type_state_(initial_state),
       type_sync_proxy_(type_sync_proxy.Pass()),
+      cryptographer_provider_(cryptographer_provider),
       nudge_handler_(nudge_handler),
       entities_deleter_(&entities_),
       weak_ptr_factory_(this) {
@@ -32,6 +36,18 @@ ModelTypeSyncWorkerImpl::ModelTypeSyncWorkerImpl(
   if (!data_type_state_.initial_sync_done) {
     nudge_handler_->NudgeForInitialDownload(type_);
   }
+
+  for (UpdateResponseDataList::const_iterator it =
+           saved_pending_updates.begin();
+       it != saved_pending_updates.end();
+       ++it) {
+    EntityTracker* entity_tracker = EntityTracker::FromServerUpdate(
+        it->id, it->client_tag_hash, it->response_version);
+    entity_tracker->ReceivePendingUpdate(*it);
+    entities_.insert(std::make_pair(it->client_tag_hash, entity_tracker));
+  }
+
+  TryDecryptPendingUpdates();
 }
 
 ModelTypeSyncWorkerImpl::~ModelTypeSyncWorkerImpl() {
@@ -42,6 +58,33 @@ ModelType ModelTypeSyncWorkerImpl::GetModelType() const {
   return type_;
 }
 
+bool ModelTypeSyncWorkerImpl::IsEncryptionRequired() const {
+  return !data_type_state_.encryption_key_name.empty();
+}
+
+void ModelTypeSyncWorkerImpl::SetEncryptionKeyName(const std::string& name) {
+  if (data_type_state_.encryption_key_name == name)
+    return;
+
+  data_type_state_.encryption_key_name = name;
+
+  // Pretend to send an update.  This will cause the TypeSyncProxy to notice
+  // the new encryption key and take appropriate action.
+  type_sync_proxy_->OnUpdateReceived(
+      data_type_state_, UpdateResponseDataList(), UpdateResponseDataList());
+}
+
+void ModelTypeSyncWorkerImpl::OnCryptographerStateChanged() {
+  TryDecryptPendingUpdates();
+
+  ScopedCryptographerRef scoped_cryptographer_ref;
+  cryptographer_provider_->InitScopedCryptographerRef(
+      &scoped_cryptographer_ref);
+  Cryptographer* cryptographer = scoped_cryptographer_ref.Get();
+  if (CanCommitItems(cryptographer))
+    nudge_handler_->NudgeForCommit(type_);
+}
+
 // UpdateHandler implementation.
 void ModelTypeSyncWorkerImpl::GetDownloadProgress(
     sync_pb::DataTypeProgressMarker* progress_marker) const {
@@ -66,7 +109,14 @@ SyncerError ModelTypeSyncWorkerImpl::ProcessGetUpdatesResponse(
   data_type_state_.type_context = mutated_context;
   data_type_state_.progress_marker = progress_marker;
 
+  ScopedCryptographerRef scoped_cryptographer_ref;
+  cryptographer_provider_->InitScopedCryptographerRef(
+      &scoped_cryptographer_ref);
+  Cryptographer* cryptographer = scoped_cryptographer_ref.Get();
+  DCHECK(cryptographer);
+
   UpdateResponseDataList response_datas;
+  UpdateResponseDataList pending_updates;
 
   for (SyncEntityList::const_iterator update_it = applicable_updates.begin();
        update_it != applicable_updates.end();
@@ -86,16 +136,17 @@ SyncerError ModelTypeSyncWorkerImpl::ProcessGetUpdatesResponse(
       const std::string& client_tag_hash =
           update_entity->client_defined_unique_tag();
       DCHECK(!client_tag_hash.empty());
+
+      EntityTracker* entity_tracker = NULL;
       EntityMap::const_iterator map_it = entities_.find(client_tag_hash);
       if (map_it == entities_.end()) {
-        EntityTracker* entity =
+        entity_tracker =
             EntityTracker::FromServerUpdate(update_entity->id_string(),
                                             client_tag_hash,
                                             update_entity->version());
-        entities_.insert(std::make_pair(client_tag_hash, entity));
+        entities_.insert(std::make_pair(client_tag_hash, entity_tracker));
       } else {
-        EntityTracker* entity = map_it->second;
-        entity->ReceiveUpdate(update_entity->version());
+        entity_tracker = map_it->second;
       }
 
       // Prepare the message for the model thread.
@@ -107,14 +158,39 @@ SyncerError ModelTypeSyncWorkerImpl::ProcessGetUpdatesResponse(
       response_data.mtime = ProtoTimeToTime(update_entity->mtime());
       response_data.non_unique_name = update_entity->name();
       response_data.deleted = update_entity->deleted();
-      response_data.specifics = update_entity->specifics();
 
-      response_datas.push_back(response_data);
+      const sync_pb::EntitySpecifics& specifics = update_entity->specifics();
+
+      if (!specifics.has_encrypted()) {
+        // No encryption.
+        entity_tracker->ReceiveUpdate(update_entity->version());
+        response_data.specifics = specifics;
+        response_datas.push_back(response_data);
+      } else if (specifics.has_encrypted() &&
+                 cryptographer->CanDecrypt(specifics.encrypted())) {
+        // Encrypted, but we know the key.
+        if (DecryptSpecifics(
+                cryptographer, specifics, &response_data.specifics)) {
+          entity_tracker->ReceiveUpdate(update_entity->version());
+          response_data.encryption_key_name = specifics.encrypted().key_name();
+          response_datas.push_back(response_data);
+        }
+      } else if (specifics.has_encrypted() &&
+                 !cryptographer->CanDecrypt(specifics.encrypted())) {
+        // Can't decrypt right now.  Ask the entity tracker to handle it.
+        response_data.specifics = specifics;
+        if (entity_tracker->ReceivePendingUpdate(response_data)) {
+          // Send to the model thread for safe-keeping across restarts if the
+          // tracker decides the update is worth keeping.
+          pending_updates.push_back(response_data);
+        }
+      }
     }
   }
 
   // Forward these updates to the model thread so it can do the rest.
-  type_sync_proxy_->OnUpdateReceived(data_type_state_, response_datas);
+  type_sync_proxy_->OnUpdateReceived(
+      data_type_state_, response_datas, pending_updates);
 
   return SYNCER_OK;
 }
@@ -128,8 +204,8 @@ void ModelTypeSyncWorkerImpl::ApplyUpdates(sessions::StatusController* status) {
   if (!data_type_state_.initial_sync_done) {
     data_type_state_.initial_sync_done = true;
 
-    UpdateResponseDataList empty_update_list;
-    type_sync_proxy_->OnUpdateReceived(data_type_state_, empty_update_list);
+    type_sync_proxy_->OnUpdateReceived(
+        data_type_state_, UpdateResponseDataList(), UpdateResponseDataList());
   }
 }
 
@@ -144,7 +220,7 @@ void ModelTypeSyncWorkerImpl::EnqueueForCommit(
     const CommitRequestDataList& list) {
   DCHECK(CalledOnValidThread());
 
-  DCHECK(CanCommitItems())
+  DCHECK(IsTypeInitialized())
       << "Asked to commit items before type was initialized.  "
       << "ModelType is: " << ModelTypeToString(type_);
 
@@ -153,6 +229,13 @@ void ModelTypeSyncWorkerImpl::EnqueueForCommit(
        ++it) {
     StorePendingCommit(*it);
   }
+
+  ScopedCryptographerRef scoped_cryptographer_ref;
+  cryptographer_provider_->InitScopedCryptographerRef(
+      &scoped_cryptographer_ref);
+  Cryptographer* cryptographer = scoped_cryptographer_ref.Get();
+  if (CanCommitItems(cryptographer))
+    nudge_handler_->NudgeForCommit(type_);
 }
 
 // CommitContributor implementation.
@@ -164,7 +247,12 @@ scoped_ptr<CommitContribution> ModelTypeSyncWorkerImpl::GetContribution(
   std::vector<int64> sequence_numbers;
   google::protobuf::RepeatedPtrField<sync_pb::SyncEntity> commit_entities;
 
-  if (!CanCommitItems())
+  ScopedCryptographerRef scoped_cryptographer_ref;
+  cryptographer_provider_->InitScopedCryptographerRef(
+      &scoped_cryptographer_ref);
+  Cryptographer* cryptographer = scoped_cryptographer_ref.Get();
+
+  if (!CanCommitItems(cryptographer))
     return scoped_ptr<CommitContribution>();
 
   // TODO(rlarocque): Avoid iterating here.
@@ -177,7 +265,7 @@ scoped_ptr<CommitContribution> ModelTypeSyncWorkerImpl::GetContribution(
       int64 sequence_number = -1;
 
       entity->PrepareCommitProto(commit_entity, &sequence_number);
-      HelpInitializeCommitEntity(commit_entity);
+      HelpInitializeCommitEntity(cryptographer, commit_entity);
       sequence_numbers.push_back(sequence_number);
 
       space_remaining--;
@@ -222,9 +310,6 @@ void ModelTypeSyncWorkerImpl::StorePendingCommit(
                           request.deleted,
                           request.specifics);
   }
-
-  if (CanCommitItems())
-    nudge_handler_->NudgeForCommit(type_);
 }
 
 void ModelTypeSyncWorkerImpl::OnCommitResponse(
@@ -260,14 +345,30 @@ base::WeakPtr<ModelTypeSyncWorkerImpl> ModelTypeSyncWorkerImpl::AsWeakPtr() {
   return weak_ptr_factory_.GetWeakPtr();
 }
 
-bool ModelTypeSyncWorkerImpl::CanCommitItems() const {
-  // We can't commit anything until we know the type's parent node.
-  // We'll get it in the first update response.
+bool ModelTypeSyncWorkerImpl::IsTypeInitialized() const {
   return !data_type_state_.type_root_id.empty() &&
          data_type_state_.initial_sync_done;
 }
 
+bool ModelTypeSyncWorkerImpl::CanCommitItems(
+    Cryptographer* cryptographer) const {
+  // We can't commit anything until we know the type's parent node.
+  // We'll get it in the first update response.
+  if (!IsTypeInitialized())
+    return false;
+
+  // Don't commit if we should be encrypting but don't have the required keys.
+  if (IsEncryptionRequired() && (!cryptographer || !cryptographer->is_ready() ||
+                                 cryptographer->GetDefaultNigoriKeyName() !=
+                                     data_type_state_.encryption_key_name)) {
+    return false;
+  }
+
+  return true;
+}
+
 void ModelTypeSyncWorkerImpl::HelpInitializeCommitEntity(
+    Cryptographer* cryptographer,
     sync_pb::SyncEntity* sync_entity) {
   // Initial commits need our help to generate a client ID.
   if (!sync_entity->has_id_string()) {
@@ -277,14 +378,80 @@ void ModelTypeSyncWorkerImpl::HelpInitializeCommitEntity(
         base::StringPrintf("%s-%" PRId64, ModelTypeToString(type_), id));
   }
 
+  // Encrypt the specifics and hide the title if necessary.
+  if (IsEncryptionRequired()) {
+    sync_pb::EntitySpecifics encrypted_specifics;
+    cryptographer->Encrypt(sync_entity->specifics(),
+                           encrypted_specifics.mutable_encrypted());
+    sync_entity->mutable_specifics()->CopyFrom(encrypted_specifics);
+    sync_entity->set_name("encrypted");
+  }
+
   // Always include enough specifics to identify the type.  Do this even in
   // deletion requests, where the specifics are otherwise invalid.
-  if (!sync_entity->has_specifics()) {
-    AddDefaultFieldValue(type_, sync_entity->mutable_specifics());
-  }
+  AddDefaultFieldValue(type_, sync_entity->mutable_specifics());
 
   // We're always responsible for the parent ID.
   sync_entity->set_parent_id_string(data_type_state_.type_root_id);
 }
 
+void ModelTypeSyncWorkerImpl::TryDecryptPendingUpdates() {
+  UpdateResponseDataList response_datas;
+
+  ScopedCryptographerRef scoped_cryptographer_ref;
+  cryptographer_provider_->InitScopedCryptographerRef(
+      &scoped_cryptographer_ref);
+  Cryptographer* cryptographer = scoped_cryptographer_ref.Get();
+  DCHECK(cryptographer);
+
+  for (EntityMap::const_iterator it = entities_.begin(); it != entities_.end();
+       ++it) {
+    if (it->second->HasPendingUpdate()) {
+      const UpdateResponseData& saved_pending = it->second->GetPendingUpdate();
+
+      // We assume all pending updates are encrypted items for which we
+      // don't have the key.
+      DCHECK(saved_pending.specifics.has_encrypted());
+
+      if (cryptographer->CanDecrypt(saved_pending.specifics.encrypted())) {
+        UpdateResponseData decrypted_response = saved_pending;
+        if (DecryptSpecifics(cryptographer,
+                             saved_pending.specifics,
+                             &decrypted_response.specifics)) {
+          decrypted_response.encryption_key_name =
+              saved_pending.specifics.encrypted().key_name();
+          response_datas.push_back(decrypted_response);
+
+          it->second->ClearPendingUpdate();
+        }
+      }
+    }
+  }
+
+  if (!response_datas.empty()) {
+    type_sync_proxy_->OnUpdateReceived(
+        data_type_state_, response_datas, UpdateResponseDataList());
+  }
+}
+
+bool ModelTypeSyncWorkerImpl::DecryptSpecifics(
+    Cryptographer* cryptographer,
+    const sync_pb::EntitySpecifics& in,
+    sync_pb::EntitySpecifics* out) {
+  DCHECK(in.has_encrypted());
+  DCHECK(cryptographer->CanDecrypt(in.encrypted()));
+
+  std::string plaintext;
+  plaintext = cryptographer->DecryptToString(in.encrypted());
+  if (plaintext.empty()) {
+    LOG(ERROR) << "Failed to decrypt a decryptable entity";
+    return false;
+  }
+  if (!out->ParseFromString(plaintext)) {
+    LOG(ERROR) << "Failed to parse decrypted entity";
+    return false;
+  }
+  return true;
+}
+
 }  // namespace syncer
diff --git a/sync/engine/model_type_sync_worker_impl.h b/sync/engine/model_type_sync_worker_impl.h
index 6905c50..f2df976 100644
--- a/sync/engine/model_type_sync_worker_impl.h
+++ b/sync/engine/model_type_sync_worker_impl.h
@@ -10,11 +10,13 @@
 #include "base/threading/non_thread_safe.h"
 #include "sync/base/sync_export.h"
 #include "sync/engine/commit_contributor.h"
+#include "sync/engine/cryptographer_provider.h"
 #include "sync/engine/model_type_sync_worker.h"
 #include "sync/engine/nudge_handler.h"
 #include "sync/engine/update_handler.h"
 #include "sync/internal_api/public/base/model_type.h"
 #include "sync/internal_api/public/non_blocking_sync_common.h"
+#include "sync/internal_api/public/sync_encryption_handler.h"
 #include "sync/protocol/sync.pb.h"
 
 namespace base {
@@ -53,12 +55,19 @@ class SYNC_EXPORT ModelTypeSyncWorkerImpl : public UpdateHandler,
  public:
   ModelTypeSyncWorkerImpl(ModelType type,
                           const DataTypeState& initial_state,
+                          const UpdateResponseDataList& saved_pending_updates,
+                          CryptographerProvider* cryptographer_provider,
                           NudgeHandler* nudge_handler,
                           scoped_ptr<ModelTypeSyncProxy> type_sync_proxy);
   virtual ~ModelTypeSyncWorkerImpl();
 
   ModelType GetModelType() const;
 
+  bool IsEncryptionRequired() const;
+  void SetEncryptionKeyName(const std::string& name);
+
+  void OnCryptographerStateChanged();
+
   // UpdateHandler implementation.
   virtual void GetDownloadProgress(
       sync_pb::DataTypeProgressMarker* progress_marker) const OVERRIDE;
@@ -87,20 +96,45 @@ class SYNC_EXPORT ModelTypeSyncWorkerImpl : public UpdateHandler,
 
  private:
   typedef std::map<std::string, EntityTracker*> EntityMap;
+  typedef std::map<std::string, UpdateResponseData*> UpdateMap;
 
   // Stores a single commit request in this object's internal state.
   void StorePendingCommit(const CommitRequestData& request);
 
-  // Returns true if all data type state required for commits is available.  In
-  // practice, this means that it returns true from the time this object first
-  // receives notice of a successful update fetch from the server.
-  bool CanCommitItems() const;
+  // Returns true if this type has successfully fetched all available updates
+  // from the server at least once.  Our state may or may not be stale, but at
+  // least we know that it was valid at some point in the past.
+  bool IsTypeInitialized() const;
+
+  // Returns true if this type is prepared to commit items.  Currently, this
+  // depends on having downloaded the initial data and having the encryption
+  // settings in a good state.
+  bool CanCommitItems(Cryptographer* cryptographer) const;
 
   // Initializes the parts of a commit entity that are the responsibility of
   // this class, and not the EntityTracker.  Some fields, like the
   // client-assigned ID, can only be set by an entity with knowledge of the
   // entire data type's state.
-  void HelpInitializeCommitEntity(sync_pb::SyncEntity* commit_entity);
+  void HelpInitializeCommitEntity(Cryptographer* cryptographer,
+                                  sync_pb::SyncEntity* commit_entity);
+
+  // Attempts to decrypt pending updates stored in the EntityMap.  If
+  // successful, will remove the update from the its EntityTracker and forward
+  // it to the proxy thread for application.
+  void TryDecryptPendingUpdates();
+
+  // Attempts to decrypt the given specifics and return them in the |out|
+  // parameter.  Assumes cryptographer->CanDecrypt(specifics) returned true.
+  //
+  // Returns false if the decryption failed.  There are no guarantees about the
+  // contents of |out| when that happens.
+  //
+  // In theory, this should never fail.  Only corrupt or invalid entries could
+  // cause this to fail, and no clients are known to create such entries.  The
+  // failure case is an attempt to be defensive against bad input.
+  static bool DecryptSpecifics(Cryptographer* cryptographer,
+                               const sync_pb::EntitySpecifics& in,
+                               sync_pb::EntitySpecifics* out);
 
   ModelType type_;
 
@@ -111,6 +145,10 @@ class SYNC_EXPORT ModelTypeSyncWorkerImpl : public UpdateHandler,
   // This is NULL when no proxy is connected..
   scoped_ptr<ModelTypeSyncProxy> type_sync_proxy_;
 
+  // A helper to provide access to the syncable::Directory's cryptographer.
+  // Not owned.
+  CryptographerProvider* cryptographer_provider_;
+
   // Interface used to access and send nudges to the sync scheduler.  Not owned.
   NudgeHandler* nudge_handler_;
 
diff --git a/sync/engine/model_type_sync_worker_impl_unittest.cc b/sync/engine/model_type_sync_worker_impl_unittest.cc
index 0c1fa13..df2ed66 100644
--- a/sync/engine/model_type_sync_worker_impl_unittest.cc
+++ b/sync/engine/model_type_sync_worker_impl_unittest.cc
@@ -4,6 +4,7 @@
 
 #include "sync/engine/model_type_sync_worker_impl.h"
 
+#include "base/strings/stringprintf.h"
 #include "sync/engine/commit_contribution.h"
 #include "sync/engine/model_type_sync_proxy.h"
 #include "sync/internal_api/public/base/model_type.h"
@@ -13,13 +14,18 @@
 #include "sync/syncable/syncable_util.h"
 #include "sync/test/engine/mock_model_type_sync_proxy.h"
 #include "sync/test/engine/mock_nudge_handler.h"
+#include "sync/test/engine/simple_cryptographer_provider.h"
 #include "sync/test/engine/single_type_mock_server.h"
+#include "sync/test/fake_encryptor.h"
 
 #include "testing/gtest/include/gtest/gtest.h"
 
 static const std::string kTypeParentId = "PrefsRootNodeID";
 static const syncer::ModelType kModelType = syncer::PREFERENCES;
 
+// Special constant value taken from cryptographer.cc.
+const char kNigoriKeyName[] = "nigori-key";
+
 namespace syncer {
 
 // Tests the ModelTypeSyncWorkerImpl.
@@ -65,8 +71,23 @@ class ModelTypeSyncWorkerImplTest : public ::testing::Test {
   // committing items right away.
   void NormalInitialize();
 
-  // Initialize with a custom initial DataTypeState.
-  void InitializeWithState(const DataTypeState& state);
+  // Initialize with some saved pending updates from the model thread.
+  void InitializeWithPendingUpdates(
+      const UpdateResponseDataList& initial_pending_updates);
+
+  // Initialize with a custom initial DataTypeState and pending updates.
+  void InitializeWithState(const DataTypeState& state,
+                           const UpdateResponseDataList& pending_updates);
+
+  // Introduce a new key that the local cryptographer can't decrypt.
+  void NewForeignEncryptionKey();
+
+  // Update the local cryptographer with all relevant keys.
+  void UpdateLocalCryptographer();
+
+  // Use the Nth nigori instance to encrypt incoming updates.
+  // The default value, zero, indicates no encryption.
+  void SetUpdateEncryptionFilter(int n);
 
   // Modifications on the model thread that get sent to the worker under test.
   void CommitRequest(const std::string& tag, const std::string& value);
@@ -79,6 +100,13 @@ class ModelTypeSyncWorkerImplTest : public ::testing::Test {
                                const std::string& value);
   void TriggerTombstoneFromServer(int64 version_offset, const std::string& tag);
 
+  // Delivers specified protos as updates.
+  //
+  // Does not update mock server state.  Should be used as a last resort when
+  // writing test cases that require entities that don't fit the normal sync
+  // protocol.  Try to use the other, higher level methods if possible.
+  void DeliverRawUpdates(const SyncEntityList& update_list);
+
   // By default, this harness behaves as if all tasks posted to the model
   // thread are executed immediately.  However, this is not necessarily true.
   // The model's TaskRunner has a queue, and the tasks we post to it could
@@ -110,6 +138,7 @@ class ModelTypeSyncWorkerImplTest : public ::testing::Test {
   // be updated until the response is actually processed by the model thread.
   size_t GetNumModelThreadUpdateResponses() const;
   UpdateResponseDataList GetNthModelThreadUpdateResponse(size_t n) const;
+  UpdateResponseDataList GetNthModelThreadPendingUpdates(size_t n) const;
   DataTypeState GetNthModelThreadUpdateState(size_t n) const;
 
   // Reads the latest update response datas on the model thread.
@@ -144,7 +173,39 @@ class ModelTypeSyncWorkerImplTest : public ::testing::Test {
   static sync_pb::EntitySpecifics GenerateSpecifics(const std::string& tag,
                                                     const std::string& value);
 
+  // Returns a set of KeyParams for the cryptographer.  Each input 'n' value
+  // results in a different set of parameters.
+  static KeyParams GetNthKeyParams(int n);
+
+  // Returns the name for the given Nigori.
+  //
+  // Uses some 'white-box' knowledge to mimic the names that a real sync client
+  // would generate.  It's probably not necessary to do so, but it can't hurt.
+  static std::string GetNigoriName(const Nigori& nigori);
+
+  // Modifies the input/output parameter |specifics| by encrypting it with
+  // a Nigori intialized with the specified KeyParams.
+  static void EncryptUpdate(const KeyParams& params,
+                            sync_pb::EntitySpecifics* specifics);
+
  private:
+  // An encryptor for our cryptographer.
+  FakeEncryptor fake_encryptor_;
+
+  // The cryptographer itself.
+  Cryptographer cryptographer_;
+
+  // A CryptographerProvider for the ModelTypeSyncWorkerImpl.
+  SimpleCryptographerProvider cryptographer_provider_;
+
+  // The number of the most recent foreign encryption key known to our
+  // cryptographer.  Note that not all of these will be decryptable.
+  int foreign_encryption_key_index_;
+
+  // The number of the encryption key used to encrypt incoming updates.  A zero
+  // value implies no encryption.
+  int update_encryption_filter_index_;
+
   // The ModelTypeSyncWorkerImpl being tested.
   scoped_ptr<ModelTypeSyncWorkerImpl> worker_;
 
@@ -163,7 +224,12 @@ class ModelTypeSyncWorkerImplTest : public ::testing::Test {
 };
 
 ModelTypeSyncWorkerImplTest::ModelTypeSyncWorkerImplTest()
-    : mock_type_sync_proxy_(NULL), mock_server_(kModelType) {
+    : cryptographer_(&fake_encryptor_),
+      cryptographer_provider_(&cryptographer_),
+      foreign_encryption_key_index_(0),
+      update_encryption_filter_index_(0),
+      mock_type_sync_proxy_(NULL),
+      mock_server_(kModelType) {
 }
 
 ModelTypeSyncWorkerImplTest::~ModelTypeSyncWorkerImplTest() {
@@ -175,10 +241,15 @@ void ModelTypeSyncWorkerImplTest::FirstInitialize() {
       GetSpecificsFieldNumberFromModelType(kModelType));
   initial_state.next_client_id = 0;
 
-  InitializeWithState(initial_state);
+  InitializeWithState(initial_state, UpdateResponseDataList());
 }
 
 void ModelTypeSyncWorkerImplTest::NormalInitialize() {
+  InitializeWithPendingUpdates(UpdateResponseDataList());
+}
+
+void ModelTypeSyncWorkerImplTest::InitializeWithPendingUpdates(
+    const UpdateResponseDataList& initial_pending_updates) {
   DataTypeState initial_state;
   initial_state.progress_marker.set_data_type_id(
       GetSpecificsFieldNumberFromModelType(kModelType));
@@ -188,21 +259,79 @@ void ModelTypeSyncWorkerImplTest::NormalInitialize() {
   initial_state.type_root_id = kTypeParentId;
   initial_state.initial_sync_done = true;
 
-  InitializeWithState(initial_state);
+  InitializeWithState(initial_state, initial_pending_updates);
 
   mock_nudge_handler_.ClearCounters();
 }
 
 void ModelTypeSyncWorkerImplTest::InitializeWithState(
-    const DataTypeState& state) {
+    const DataTypeState& state,
+    const UpdateResponseDataList& initial_pending_updates) {
   DCHECK(!worker_);
 
   // We don't get to own this object.  The |worker_| keeps a scoped_ptr to it.
   mock_type_sync_proxy_ = new MockModelTypeSyncProxy();
   scoped_ptr<ModelTypeSyncProxy> proxy(mock_type_sync_proxy_);
 
-  worker_.reset(new ModelTypeSyncWorkerImpl(
-      kModelType, state, &mock_nudge_handler_, proxy.Pass()));
+  worker_.reset(new ModelTypeSyncWorkerImpl(kModelType,
+                                            state,
+                                            initial_pending_updates,
+                                            &cryptographer_provider_,
+                                            &mock_nudge_handler_,
+                                            proxy.Pass()));
+}
+
+void ModelTypeSyncWorkerImplTest::NewForeignEncryptionKey() {
+  foreign_encryption_key_index_++;
+
+  sync_pb::NigoriKeyBag bag;
+
+  for (int i = 0; i <= foreign_encryption_key_index_; ++i) {
+    Nigori nigori;
+    KeyParams params = GetNthKeyParams(i);
+    nigori.InitByDerivation(params.hostname, params.username, params.password);
+
+    sync_pb::NigoriKey* key = bag.add_key();
+
+    key->set_name(GetNigoriName(nigori));
+    nigori.ExportKeys(key->mutable_user_key(),
+                      key->mutable_encryption_key(),
+                      key->mutable_mac_key());
+  }
+
+  // Re-create the last nigori from that loop.
+  Nigori last_nigori;
+  KeyParams params = GetNthKeyParams(foreign_encryption_key_index_);
+  last_nigori.InitByDerivation(
+      params.hostname, params.username, params.password);
+
+  // Serialize and encrypt the bag with the last nigori.
+  std::string serialized_bag;
+  bag.SerializeToString(&serialized_bag);
+
+  sync_pb::EncryptedData encrypted;
+  encrypted.set_key_name(GetNigoriName(last_nigori));
+  last_nigori.Encrypt(serialized_bag, encrypted.mutable_blob());
+
+  // Update the cryptographer with new pending keys.
+  cryptographer_.SetPendingKeys(encrypted);
+
+  // Update the worker with the latest encryption key name.
+  if (worker_)
+    worker_->SetEncryptionKeyName(encrypted.key_name());
+}
+
+void ModelTypeSyncWorkerImplTest::UpdateLocalCryptographer() {
+  KeyParams params = GetNthKeyParams(foreign_encryption_key_index_);
+  bool success = cryptographer_.DecryptPendingKeys(params);
+  DCHECK(success);
+
+  if (worker_)
+    worker_->OnCryptographerStateChanged();
+}
+
+void ModelTypeSyncWorkerImplTest::SetUpdateEncryptionFilter(int n) {
+  update_encryption_filter_index_ = n;
 }
 
 void ModelTypeSyncWorkerImplTest::CommitRequest(const std::string& name,
@@ -243,6 +372,12 @@ void ModelTypeSyncWorkerImplTest::TriggerUpdateFromServer(
     const std::string& value) {
   sync_pb::SyncEntity entity = mock_server_.UpdateFromServer(
       version_offset, GenerateTagHash(tag), GenerateSpecifics(tag, value));
+
+  if (update_encryption_filter_index_ != 0) {
+    EncryptUpdate(GetNthKeyParams(update_encryption_filter_index_),
+                  entity.mutable_specifics());
+  }
+
   SyncEntityList entity_list;
   entity_list.push_back(&entity);
 
@@ -255,11 +390,27 @@ void ModelTypeSyncWorkerImplTest::TriggerUpdateFromServer(
   worker_->ApplyUpdates(&dummy_status);
 }
 
+void ModelTypeSyncWorkerImplTest::DeliverRawUpdates(
+    const SyncEntityList& list) {
+  sessions::StatusController dummy_status;
+  worker_->ProcessGetUpdatesResponse(mock_server_.GetProgress(),
+                                     mock_server_.GetContext(),
+                                     list,
+                                     &dummy_status);
+  worker_->ApplyUpdates(&dummy_status);
+}
+
 void ModelTypeSyncWorkerImplTest::TriggerTombstoneFromServer(
     int64 version_offset,
     const std::string& tag) {
   sync_pb::SyncEntity entity =
       mock_server_.TombstoneFromServer(version_offset, GenerateTagHash(tag));
+
+  if (update_encryption_filter_index_ != 0) {
+    EncryptUpdate(GetNthKeyParams(update_encryption_filter_index_),
+                  entity.mutable_specifics());
+  }
+
   SyncEntityList entity_list;
   entity_list.push_back(&entity);
 
@@ -346,6 +497,12 @@ ModelTypeSyncWorkerImplTest::GetNthModelThreadUpdateResponse(size_t n) const {
   return mock_type_sync_proxy_->GetNthUpdateResponse(n);
 }
 
+UpdateResponseDataList
+ModelTypeSyncWorkerImplTest::GetNthModelThreadPendingUpdates(size_t n) const {
+  DCHECK_LT(n, GetNumModelThreadUpdateResponses());
+  return mock_type_sync_proxy_->GetNthPendingUpdates(n);
+}
+
 DataTypeState ModelTypeSyncWorkerImplTest::GetNthModelThreadUpdateState(
     size_t n) const {
   DCHECK_LT(n, GetNumModelThreadUpdateResponses());
@@ -401,6 +558,7 @@ int ModelTypeSyncWorkerImplTest::GetNumInitialDownloadNudges() const {
   return mock_nudge_handler_.GetNumInitialDownloadNudges();
 }
 
+// static.
 std::string ModelTypeSyncWorkerImplTest::GenerateTagHash(
     const std::string& tag) {
   const std::string& client_tag_hash =
@@ -408,6 +566,7 @@ std::string ModelTypeSyncWorkerImplTest::GenerateTagHash(
   return client_tag_hash;
 }
 
+// static.
 sync_pb::EntitySpecifics ModelTypeSyncWorkerImplTest::GenerateSpecifics(
     const std::string& tag,
     const std::string& value) {
@@ -417,6 +576,46 @@ sync_pb::EntitySpecifics ModelTypeSyncWorkerImplTest::GenerateSpecifics(
   return specifics;
 }
 
+// static.
+std::string ModelTypeSyncWorkerImplTest::GetNigoriName(const Nigori& nigori) {
+  std::string name;
+  if (!nigori.Permute(Nigori::Password, kNigoriKeyName, &name)) {
+    NOTREACHED();
+    return std::string();
+  }
+
+  return name;
+}
+
+// static.
+KeyParams ModelTypeSyncWorkerImplTest::GetNthKeyParams(int n) {
+  KeyParams params;
+  params.hostname = std::string("localhost");
+  params.username = std::string("userX");
+  params.password = base::StringPrintf("pw%02d", n);
+  return params;
+}
+
+// static.
+void ModelTypeSyncWorkerImplTest::EncryptUpdate(
+    const KeyParams& params,
+    sync_pb::EntitySpecifics* specifics) {
+  Nigori nigori;
+  nigori.InitByDerivation(params.hostname, params.username, params.password);
+
+  sync_pb::EntitySpecifics original_specifics = *specifics;
+  std::string plaintext;
+  original_specifics.SerializeToString(&plaintext);
+
+  std::string encrypted;
+  nigori.Encrypt(plaintext, &encrypted);
+
+  specifics->Clear();
+  AddDefaultFieldValue(kModelType, specifics);
+  specifics->mutable_encrypted()->set_key_name(GetNigoriName(nigori));
+  specifics->mutable_encrypted()->set_blob(encrypted);
+}
+
 // Requests a commit and verifies the messages sent to the client and server as
 // a result.
 //
@@ -600,6 +799,7 @@ TEST_F(ModelTypeSyncWorkerImplTest, TwoNewItemsCommittedSeparately) {
   EXPECT_EQ(2U, GetNumModelThreadCommitResponses());
 }
 
+// Test normal update receipt code path.
 TEST_F(ModelTypeSyncWorkerImplTest, ReceiveUpdates) {
   NormalInitialize();
 
@@ -625,4 +825,281 @@ TEST_F(ModelTypeSyncWorkerImplTest, ReceiveUpdates) {
   EXPECT_EQ("value1", update.specifics.preference().value());
 }
 
+// Test commit of encrypted updates.
+TEST_F(ModelTypeSyncWorkerImplTest, EncryptedCommit) {
+  NormalInitialize();
+
+  NewForeignEncryptionKey();
+  UpdateLocalCryptographer();
+
+  // Normal commit request stuff.
+  CommitRequest("tag1", "value1");
+  DoSuccessfulCommit();
+  ASSERT_EQ(1U, GetNumCommitMessagesOnServer());
+  EXPECT_EQ(1, GetNthCommitMessageOnServer(0).commit().entries_size());
+  ASSERT_TRUE(HasCommitEntityOnServer("tag1"));
+  const sync_pb::SyncEntity& tag1_entity =
+      GetLatestCommitEntityOnServer("tag1");
+
+  EXPECT_TRUE(tag1_entity.specifics().has_encrypted());
+
+  // The title should be overwritten.
+  EXPECT_EQ(tag1_entity.name(), "encrypted");
+
+  // The type should be set, but there should be no non-encrypted contents.
+  EXPECT_TRUE(tag1_entity.specifics().has_preference());
+  EXPECT_FALSE(tag1_entity.specifics().preference().has_name());
+  EXPECT_FALSE(tag1_entity.specifics().preference().has_value());
+}
+
+// Test items are not committed when encryption is required but unavailable.
+TEST_F(ModelTypeSyncWorkerImplTest, EncryptionBlocksCommits) {
+  NormalInitialize();
+
+  CommitRequest("tag1", "value1");
+  EXPECT_TRUE(WillCommit());
+
+  // We know encryption is in use on this account, but don't have the necessary
+  // encryption keys.  The worker should refuse to commit.
+  NewForeignEncryptionKey();
+  EXPECT_FALSE(WillCommit());
+
+  // Once the cryptographer is returned to a normal state, we should be able to
+  // commit again.
+  EXPECT_EQ(1, GetNumCommitNudges());
+  UpdateLocalCryptographer();
+  EXPECT_EQ(2, GetNumCommitNudges());
+  EXPECT_TRUE(WillCommit());
+
+  // Verify the committed entity was properly encrypted.
+  DoSuccessfulCommit();
+  ASSERT_EQ(1U, GetNumCommitMessagesOnServer());
+  EXPECT_EQ(1, GetNthCommitMessageOnServer(0).commit().entries_size());
+  ASSERT_TRUE(HasCommitEntityOnServer("tag1"));
+  const sync_pb::SyncEntity& tag1_entity =
+      GetLatestCommitEntityOnServer("tag1");
+  EXPECT_TRUE(tag1_entity.specifics().has_encrypted());
+  EXPECT_EQ(tag1_entity.name(), "encrypted");
+  EXPECT_TRUE(tag1_entity.specifics().has_preference());
+  EXPECT_FALSE(tag1_entity.specifics().preference().has_name());
+  EXPECT_FALSE(tag1_entity.specifics().preference().has_value());
+}
+
+// Test the receipt of decryptable entities.
+TEST_F(ModelTypeSyncWorkerImplTest, ReceiveDecryptableEntities) {
+  NormalInitialize();
+
+  // Create a new Nigori and allow the cryptographer to decrypt it.
+  NewForeignEncryptionKey();
+  UpdateLocalCryptographer();
+
+  // First, receive an unencrypted entry.
+  TriggerUpdateFromServer(10, "tag1", "value1");
+
+  // Test some basic properties regarding the update.
+  ASSERT_TRUE(HasUpdateResponseOnModelThread("tag1"));
+  UpdateResponseData update1 = GetUpdateResponseOnModelThread("tag1");
+  EXPECT_EQ("tag1", update1.specifics.preference().name());
+  EXPECT_EQ("value1", update1.specifics.preference().value());
+  EXPECT_TRUE(update1.encryption_key_name.empty());
+
+  // Set received updates to be encrypted using the new nigori.
+  SetUpdateEncryptionFilter(1);
+
+  // This next update will be encrypted.
+  TriggerUpdateFromServer(10, "tag2", "value2");
+
+  // Test its basic features and the value of encryption_key_name.
+  ASSERT_TRUE(HasUpdateResponseOnModelThread("tag2"));
+  UpdateResponseData update2 = GetUpdateResponseOnModelThread("tag2");
+  EXPECT_EQ("tag2", update2.specifics.preference().name());
+  EXPECT_EQ("value2", update2.specifics.preference().value());
+  EXPECT_FALSE(update2.encryption_key_name.empty());
+}
+
+// Receive updates that are initially undecryptable, then ensure they get
+// delivered to the model thread when decryption becomes possible.
+TEST_F(ModelTypeSyncWorkerImplTest, ReceiveUndecryptableEntries) {
+  NormalInitialize();
+
+  // Set a new encryption key.  The model thread will be notified of the new
+  // encryption key through a faked update response.
+  NewForeignEncryptionKey();
+  EXPECT_EQ(1U, GetNumModelThreadUpdateResponses());
+
+  // Send an update using that new key.
+  SetUpdateEncryptionFilter(1);
+  TriggerUpdateFromServer(10, "tag1", "value1");
+
+  // At this point, the cryptographer does not have access to the key, so the
+  // updates will be undecryptable.  They'll be transfered to the model thread
+  // for safe-keeping as pending updates.
+  ASSERT_EQ(2U, GetNumModelThreadUpdateResponses());
+  UpdateResponseDataList updates_list = GetNthModelThreadUpdateResponse(1);
+  EXPECT_EQ(0U, updates_list.size());
+  UpdateResponseDataList pending_updates = GetNthModelThreadPendingUpdates(1);
+  EXPECT_EQ(1U, pending_updates.size());
+
+  // The update will be delivered as soon as decryption becomes possible.
+  UpdateLocalCryptographer();
+  ASSERT_TRUE(HasUpdateResponseOnModelThread("tag1"));
+  UpdateResponseData update = GetUpdateResponseOnModelThread("tag1");
+  EXPECT_EQ("tag1", update.specifics.preference().name());
+  EXPECT_EQ("value1", update.specifics.preference().value());
+  EXPECT_FALSE(update.encryption_key_name.empty());
+}
+
+// Ensure that even encrypted updates can cause conflicts.
+TEST_F(ModelTypeSyncWorkerImplTest, EncryptedUpdateOverridesPendingCommit) {
+  NormalInitialize();
+
+  // Prepeare to commit an item.
+  CommitRequest("tag1", "value1");
+  EXPECT_TRUE(WillCommit());
+
+  // Receive an encrypted update for that item.
+  SetUpdateEncryptionFilter(1);
+  TriggerUpdateFromServer(10, "tag1", "value1");
+
+  // The pending commit state should be cleared.
+  EXPECT_FALSE(WillCommit());
+
+  // The encrypted update will be delivered to the model thread.
+  ASSERT_EQ(1U, GetNumModelThreadUpdateResponses());
+  UpdateResponseDataList updates_list = GetNthModelThreadUpdateResponse(0);
+  EXPECT_EQ(0U, updates_list.size());
+  UpdateResponseDataList pending_updates = GetNthModelThreadPendingUpdates(0);
+  EXPECT_EQ(1U, pending_updates.size());
+}
+
+// Test decryption of pending updates saved across a restart.
+TEST_F(ModelTypeSyncWorkerImplTest, RestorePendingEntries) {
+  // Create a fake pending update.
+  UpdateResponseData update;
+
+  update.client_tag_hash = GenerateTagHash("tag1");
+  update.id = "SomeID";
+  update.response_version = 100;
+  update.ctime = base::Time::UnixEpoch() + base::TimeDelta::FromSeconds(10);
+  update.mtime = base::Time::UnixEpoch() + base::TimeDelta::FromSeconds(11);
+  update.non_unique_name = "encrypted";
+  update.deleted = false;
+
+  update.specifics = GenerateSpecifics("tag1", "value1");
+  EncryptUpdate(GetNthKeyParams(1), &(update.specifics));
+
+  // Inject the update during ModelTypeSyncWorker initialization.
+  UpdateResponseDataList saved_pending_updates;
+  saved_pending_updates.push_back(update);
+  InitializeWithPendingUpdates(saved_pending_updates);
+
+  // Update will be undecryptable at first.
+  EXPECT_EQ(0U, GetNumModelThreadUpdateResponses());
+  ASSERT_FALSE(HasUpdateResponseOnModelThread("tag1"));
+
+  // Update the cryptographer so it can decrypt that update.
+  NewForeignEncryptionKey();
+  UpdateLocalCryptographer();
+
+  // Verify the item gets decrypted and sent back to the model thread.
+  ASSERT_TRUE(HasUpdateResponseOnModelThread("tag1"));
+}
+
+// Test decryption of pending updates saved across a restart.  This test
+// differs from the previous one in that the restored updates can be decrypted
+// immediately after the ModelTypeSyncWorker is constructed.
+TEST_F(ModelTypeSyncWorkerImplTest, RestoreApplicableEntries) {
+  // Update the cryptographer so it can decrypt that update.
+  NewForeignEncryptionKey();
+  UpdateLocalCryptographer();
+
+  // Create a fake pending update.
+  UpdateResponseData update;
+  update.client_tag_hash = GenerateTagHash("tag1");
+  update.id = "SomeID";
+  update.response_version = 100;
+  update.ctime = base::Time::UnixEpoch() + base::TimeDelta::FromSeconds(10);
+  update.mtime = base::Time::UnixEpoch() + base::TimeDelta::FromSeconds(11);
+  update.non_unique_name = "encrypted";
+  update.deleted = false;
+
+  update.specifics = GenerateSpecifics("tag1", "value1");
+  EncryptUpdate(GetNthKeyParams(1), &(update.specifics));
+
+  // Inject the update during ModelTypeSyncWorker initialization.
+  UpdateResponseDataList saved_pending_updates;
+  saved_pending_updates.push_back(update);
+  InitializeWithPendingUpdates(saved_pending_updates);
+
+  // Verify the item gets decrypted and sent back to the model thread.
+  ASSERT_TRUE(HasUpdateResponseOnModelThread("tag1"));
+}
+
+// Test that undecryptable updates provide sufficient reason to not commit.
+//
+// This should be rare in practice.  Usually the cryptographer will be in an
+// unusable state when we receive undecryptable updates, and that alone will be
+// enough to prevent all commits.
+TEST_F(ModelTypeSyncWorkerImplTest, CommitBlockedByPending) {
+  NormalInitialize();
+
+  // Prepeare to commit an item.
+  CommitRequest("tag1", "value1");
+  EXPECT_TRUE(WillCommit());
+
+  // Receive an encrypted update for that item.
+  SetUpdateEncryptionFilter(1);
+  TriggerUpdateFromServer(10, "tag1", "value1");
+
+  // The pending commit state should be cleared.
+  EXPECT_FALSE(WillCommit());
+
+  // The pending update will be delivered to the model thread.
+  HasUpdateResponseOnModelThread("tag1");
+
+  // Pretend the update arrived too late to prevent another commit request.
+  CommitRequest("tag1", "value2");
+
+  EXPECT_FALSE(WillCommit());
+}
+
+// Verify that corrupted encrypted updates don't cause crashes.
+TEST_F(ModelTypeSyncWorkerImplTest, ReceiveCorruptEncryption) {
+  // Initialize the worker with basic encryption state.
+  NormalInitialize();
+  NewForeignEncryptionKey();
+  UpdateLocalCryptographer();
+
+  // Manually create an update.
+  sync_pb::SyncEntity entity;
+  entity.set_client_defined_unique_tag(GenerateTagHash("tag1"));
+  entity.set_id_string("SomeID");
+  entity.set_version(1);
+  entity.set_ctime(1000);
+  entity.set_mtime(1001);
+  entity.set_name("encrypted");
+  entity.set_deleted(false);
+
+  // Encrypt it.
+  entity.mutable_specifics()->CopyFrom(GenerateSpecifics("tag1", "value1"));
+  EncryptUpdate(GetNthKeyParams(1), entity.mutable_specifics());
+
+  // Replace a few bytes to corrupt it.
+  entity.mutable_specifics()->mutable_encrypted()->mutable_blob()->replace(
+      0, 4, "xyz!");
+
+  SyncEntityList entity_list;
+  entity_list.push_back(&entity);
+
+  // If a corrupt update could trigger a crash, this is where it would happen.
+  DeliverRawUpdates(entity_list);
+
+  EXPECT_FALSE(HasUpdateResponseOnModelThread("tag1"));
+
+  // Deliver a non-corrupt update to see if the everything still works.
+  SetUpdateEncryptionFilter(1);
+  TriggerUpdateFromServer(10, "tag1", "value1");
+  EXPECT_TRUE(HasUpdateResponseOnModelThread("tag1"));
+}
+
 }  // namespace syncer
diff --git a/sync/internal_api/public/non_blocking_sync_common.h b/sync/internal_api/public/non_blocking_sync_common.h
index 7c7580b..b53e6a6 100644
--- a/sync/internal_api/public/non_blocking_sync_common.h
+++ b/sync/internal_api/public/non_blocking_sync_common.h
@@ -38,6 +38,11 @@ struct SYNC_EXPORT_PRIVATE DataTypeState {
   // until the first download cycle has completed.
   std::string type_root_id;
 
+  // This value is set if this type's data should be encrypted on the server.
+  // If this key changes, the client will need to re-commit all of its local
+  // data to the server using the new encryption key.
+  std::string encryption_key_name;
+
   // A strictly increasing counter used to generate unique values for the
   // client-assigned IDs.  The incrementing and ID assignment happens on the
   // sync thread, but we store the value here so we can pass it back to the
@@ -51,7 +56,6 @@ struct SYNC_EXPORT_PRIVATE DataTypeState {
   // flag is set.
   bool initial_sync_done;
 };
-
 struct SYNC_EXPORT_PRIVATE CommitRequestData {
   CommitRequestData();
   ~CommitRequestData();
@@ -94,6 +98,7 @@ struct SYNC_EXPORT_PRIVATE UpdateResponseData {
   std::string non_unique_name;
   bool deleted;
   sync_pb::EntitySpecifics specifics;
+  std::string encryption_key_name;
 };
 
 typedef std::vector<CommitRequestData> CommitRequestDataList;
diff --git a/sync/internal_api/public/sync_context.h b/sync/internal_api/public/sync_context.h
index 67b1a41..12598cd 100644
--- a/sync/internal_api/public/sync_context.h
+++ b/sync/internal_api/public/sync_context.h
@@ -10,11 +10,11 @@
 #include "base/sequenced_task_runner.h"
 #include "sync/base/sync_export.h"
 #include "sync/internal_api/public/base/model_type.h"
+#include "sync/internal_api/public/non_blocking_sync_common.h"
 
 namespace syncer {
 
 class ModelTypeSyncProxyImpl;
-struct DataTypeState;
 
 // An interface of the core parts of sync.
 //
@@ -35,6 +35,7 @@ class SYNC_EXPORT_PRIVATE SyncContext {
   virtual void ConnectSyncTypeToWorker(
       syncer::ModelType type,
       const DataTypeState& data_type_state,
+      const syncer::UpdateResponseDataList& saved_pending_updates,
       const scoped_refptr<base::SequencedTaskRunner>& datatype_task_runner,
       const base::WeakPtr<ModelTypeSyncProxyImpl>& type_sync_proxy) = 0;
 
diff --git a/sync/internal_api/public/sync_context_proxy.h b/sync/internal_api/public/sync_context_proxy.h
index 14bf14f..230a42a 100644
--- a/sync/internal_api/public/sync_context_proxy.h
+++ b/sync/internal_api/public/sync_context_proxy.h
@@ -7,6 +7,7 @@
 
 #include "base/memory/weak_ptr.h"
 #include "sync/internal_api/public/base/model_type.h"
+#include "sync/internal_api/public/non_blocking_sync_common.h"
 
 namespace syncer {
 
@@ -27,6 +28,7 @@ class SYNC_EXPORT_PRIVATE SyncContextProxy {
   virtual void ConnectTypeToSync(
       syncer::ModelType type,
       const DataTypeState& data_type_state,
+      const UpdateResponseDataList& saved_pending_updates,
       const base::WeakPtr<ModelTypeSyncProxyImpl>& type_sync_proxy) = 0;
 
   // Tells the syncer that we're no longer interested in syncing this type.
diff --git a/sync/internal_api/public/test/null_sync_context_proxy.h b/sync/internal_api/public/test/null_sync_context_proxy.h
index 808f664..0ca080d 100644
--- a/sync/internal_api/public/test/null_sync_context_proxy.h
+++ b/sync/internal_api/public/test/null_sync_context_proxy.h
@@ -6,6 +6,7 @@
 #define SYNC_INTERNAL_API_PUBLIC_TEST_NULL_SYNC_CONTEXT_PROXY_H_
 
 #include "base/memory/weak_ptr.h"
+#include "sync/internal_api/public/non_blocking_sync_common.h"
 #include "sync/internal_api/public/sync_context_proxy.h"
 
 namespace syncer {
@@ -23,6 +24,7 @@ class NullSyncContextProxy : public SyncContextProxy {
   virtual void ConnectTypeToSync(
       syncer::ModelType type,
       const DataTypeState& data_type_state,
+      const UpdateResponseDataList& saved_pending_updates,
       const base::WeakPtr<ModelTypeSyncProxyImpl>& type_sync_proxy) OVERRIDE;
   virtual void Disconnect(syncer::ModelType type) OVERRIDE;
   virtual scoped_ptr<SyncContextProxy> Clone() const OVERRIDE;
diff --git a/sync/internal_api/sync_context_proxy_impl.cc b/sync/internal_api/sync_context_proxy_impl.cc
index 7d6a817..ee17c72 100644
--- a/sync/internal_api/sync_context_proxy_impl.cc
+++ b/sync/internal_api/sync_context_proxy_impl.cc
@@ -25,6 +25,7 @@ SyncContextProxyImpl::~SyncContextProxyImpl() {
 void SyncContextProxyImpl::ConnectTypeToSync(
     ModelType type,
     const DataTypeState& data_type_state,
+    const UpdateResponseDataList& saved_pending_updates,
     const base::WeakPtr<ModelTypeSyncProxyImpl>& type_sync_proxy) {
   VLOG(1) << "ConnectTypeToSync: " << ModelTypeToString(type);
   sync_task_runner_->PostTask(FROM_HERE,
@@ -32,6 +33,7 @@ void SyncContextProxyImpl::ConnectTypeToSync(
                                          sync_context_,
                                          type,
                                          data_type_state,
+                                         saved_pending_updates,
                                          base::ThreadTaskRunnerHandle::Get(),
                                          type_sync_proxy));
 }
diff --git a/sync/internal_api/sync_context_proxy_impl.h b/sync/internal_api/sync_context_proxy_impl.h
index fb7cd4f..7f121a3 100644
--- a/sync/internal_api/sync_context_proxy_impl.h
+++ b/sync/internal_api/sync_context_proxy_impl.h
@@ -40,6 +40,7 @@ class SYNC_EXPORT_PRIVATE SyncContextProxyImpl : public SyncContextProxy {
   virtual void ConnectTypeToSync(
       syncer::ModelType type,
       const DataTypeState& data_type_state,
+      const UpdateResponseDataList& pending_updates,
       const base::WeakPtr<ModelTypeSyncProxyImpl>& sync_proxy_impl) OVERRIDE;
 
   // Disables syncing for the given type on the sync thread.
diff --git a/sync/internal_api/sync_encryption_handler_impl.cc b/sync/internal_api/sync_encryption_handler_impl.cc
index 3a88f53..aea66e5 100644
--- a/sync/internal_api/sync_encryption_handler_impl.cc
+++ b/sync/internal_api/sync_encryption_handler_impl.cc
@@ -1519,7 +1519,7 @@ bool SyncEncryptionHandlerImpl::GetKeystoreDecryptor(
   DCHECK(!keystore_key.empty());
   DCHECK(cryptographer.is_ready());
   std::string serialized_nigori;
-  serialized_nigori = cryptographer.GetDefaultNigoriKey();
+  serialized_nigori = cryptographer.GetDefaultNigoriKeyData();
   if (serialized_nigori.empty()) {
     LOG(ERROR) << "Failed to get cryptographer bootstrap token.";
     return false;
diff --git a/sync/internal_api/test/null_sync_context_proxy.cc b/sync/internal_api/test/null_sync_context_proxy.cc
index e209c21..2fc8bbf 100644
--- a/sync/internal_api/test/null_sync_context_proxy.cc
+++ b/sync/internal_api/test/null_sync_context_proxy.cc
@@ -15,6 +15,7 @@ NullSyncContextProxy::~NullSyncContextProxy() {
 void NullSyncContextProxy::ConnectTypeToSync(
     syncer::ModelType type,
     const DataTypeState& data_type_state,
+    const UpdateResponseDataList& saved_pending_updates,
     const base::WeakPtr<ModelTypeSyncProxyImpl>& type_sync_proxy) {
   NOTREACHED() << "NullSyncContextProxy is not meant to be used";
 }
diff --git a/sync/sessions/model_type_registry.cc b/sync/sessions/model_type_registry.cc
index 5222202..c2ba5f4 100644
--- a/sync/sessions/model_type_registry.cc
+++ b/sync/sessions/model_type_registry.cc
@@ -15,6 +15,7 @@
 #include "sync/engine/model_type_sync_worker_impl.h"
 #include "sync/internal_api/public/non_blocking_sync_common.h"
 #include "sync/sessions/directory_type_debug_info_emitter.h"
+#include "sync/util/cryptographer.h"
 
 namespace syncer {
 
@@ -32,7 +33,8 @@ class ModelTypeSyncProxyWrapper : public ModelTypeSyncProxy {
       const CommitResponseDataList& response_list) OVERRIDE;
   virtual void OnUpdateReceived(
       const DataTypeState& type_state,
-      const UpdateResponseDataList& response_list) OVERRIDE;
+      const UpdateResponseDataList& response_list,
+      const UpdateResponseDataList& pending_updates) OVERRIDE;
 
  private:
   base::WeakPtr<ModelTypeSyncProxyImpl> processor_;
@@ -61,13 +63,15 @@ void ModelTypeSyncProxyWrapper::OnCommitCompleted(
 
 void ModelTypeSyncProxyWrapper::OnUpdateReceived(
     const DataTypeState& type_state,
-    const UpdateResponseDataList& response_list) {
+    const UpdateResponseDataList& response_list,
+    const UpdateResponseDataList& pending_updates) {
   processor_task_runner_->PostTask(
       FROM_HERE,
       base::Bind(&ModelTypeSyncProxyImpl::OnUpdateReceived,
                  processor_,
                  type_state,
-                 response_list));
+                 response_list,
+                 pending_updates));
 }
 
 class ModelTypeSyncWorkerWrapper : public ModelTypeSyncWorker {
@@ -107,6 +111,7 @@ ModelTypeRegistry::ModelTypeRegistry(
     syncable::Directory* directory,
     NudgeHandler* nudge_handler)
     : directory_(directory),
+      cryptographer_provider_(directory_),
       nudge_handler_(nudge_handler),
       weak_ptr_factory_(this) {
   for (size_t i = 0u; i < workers.size(); ++i) {
@@ -185,6 +190,7 @@ void ModelTypeRegistry::SetEnabledDirectoryTypes(
 void ModelTypeRegistry::ConnectSyncTypeToWorker(
     ModelType type,
     const DataTypeState& data_type_state,
+    const UpdateResponseDataList& saved_pending_updates,
     const scoped_refptr<base::SequencedTaskRunner>& type_task_runner,
     const base::WeakPtr<ModelTypeSyncProxyImpl>& proxy_impl) {
   DVLOG(1) << "Enabling an off-thread sync type: " << ModelTypeToString(type);
@@ -192,8 +198,13 @@ void ModelTypeRegistry::ConnectSyncTypeToWorker(
   // Initialize Worker -> Proxy communication channel.
   scoped_ptr<ModelTypeSyncProxy> proxy(
       new ModelTypeSyncProxyWrapper(proxy_impl, type_task_runner));
-  scoped_ptr<ModelTypeSyncWorkerImpl> worker(new ModelTypeSyncWorkerImpl(
-      type, data_type_state, nudge_handler_, proxy.Pass()));
+  scoped_ptr<ModelTypeSyncWorkerImpl> worker(
+      new ModelTypeSyncWorkerImpl(type,
+                                  data_type_state,
+                                  saved_pending_updates,
+                                  &cryptographer_provider_,
+                                  nudge_handler_,
+                                  proxy.Pass()));
 
   // Initialize Proxy -> Worker communication channel.
   scoped_ptr<ModelTypeSyncWorker> wrapped_worker(
diff --git a/sync/sessions/model_type_registry.h b/sync/sessions/model_type_registry.h
index e18ac9c..2399f22 100644
--- a/sync/sessions/model_type_registry.h
+++ b/sync/sessions/model_type_registry.h
@@ -12,9 +12,11 @@
 #include "base/memory/scoped_vector.h"
 #include "base/memory/weak_ptr.h"
 #include "sync/base/sync_export.h"
+#include "sync/engine/directory_cryptographer_provider.h"
 #include "sync/engine/nudge_handler.h"
 #include "sync/internal_api/public/base/model_type.h"
 #include "sync/internal_api/public/engine/model_safe_worker.h"
+#include "sync/internal_api/public/non_blocking_sync_common.h"
 #include "sync/internal_api/public/sessions/type_debug_info_observer.h"
 #include "sync/internal_api/public/sync_context.h"
 
@@ -31,7 +33,6 @@ class DirectoryTypeDebugInfoEmitter;
 class ModelTypeSyncWorkerImpl;
 class ModelTypeSyncProxyImpl;
 class UpdateHandler;
-struct DataTypeState;
 
 typedef std::map<ModelType, UpdateHandler*> UpdateHandlerMap;
 typedef std::map<ModelType, CommitContributor*> CommitContributorMap;
@@ -57,6 +58,7 @@ class SYNC_EXPORT_PRIVATE ModelTypeRegistry : public SyncContext {
   virtual void ConnectSyncTypeToWorker(
       syncer::ModelType type,
       const DataTypeState& data_type_state,
+      const syncer::UpdateResponseDataList& saved_pending_updates,
       const scoped_refptr<base::SequencedTaskRunner>& type_task_runner,
       const base::WeakPtr<ModelTypeSyncProxyImpl>& proxy) OVERRIDE;
 
@@ -112,6 +114,9 @@ class SYNC_EXPORT_PRIVATE ModelTypeRegistry : public SyncContext {
   // The directory.  Not owned.
   syncable::Directory* directory_;
 
+  // Provides access to the Directory's cryptographer.
+  DirectoryCryptographerProvider cryptographer_provider_;
+
   // The NudgeHandler.  Not owned.
   NudgeHandler* nudge_handler_;
 
diff --git a/sync/sessions/model_type_registry_unittest.cc b/sync/sessions/model_type_registry_unittest.cc
index b1f1a49..96d6294 100644
--- a/sync/sessions/model_type_registry_unittest.cc
+++ b/sync/sessions/model_type_registry_unittest.cc
@@ -154,6 +154,7 @@ TEST_F(ModelTypeRegistryTest, NonBlockingTypes) {
 
   registry()->ConnectSyncTypeToWorker(syncer::THEMES,
                                       MakeInitialDataTypeState(THEMES),
+                                      UpdateResponseDataList(),
                                       task_runner,
                                       themes_sync_proxy.AsWeakPtrForUI());
   EXPECT_TRUE(registry()->GetEnabledTypes().Equals(
@@ -161,6 +162,7 @@ TEST_F(ModelTypeRegistryTest, NonBlockingTypes) {
 
   registry()->ConnectSyncTypeToWorker(syncer::SESSIONS,
                                       MakeInitialDataTypeState(SESSIONS),
+                                      UpdateResponseDataList(),
                                       task_runner,
                                       sessions_sync_proxy.AsWeakPtrForUI());
   EXPECT_TRUE(registry()->GetEnabledTypes().Equals(
@@ -192,6 +194,7 @@ TEST_F(ModelTypeRegistryTest, NonBlockingTypesWithDirectoryTypes) {
   // Add the themes non-blocking type.
   registry()->ConnectSyncTypeToWorker(syncer::THEMES,
                                       MakeInitialDataTypeState(THEMES),
+                                      UpdateResponseDataList(),
                                       task_runner,
                                       themes_sync_proxy.AsWeakPtrForUI());
   current_types.Put(syncer::THEMES);
@@ -205,6 +208,7 @@ TEST_F(ModelTypeRegistryTest, NonBlockingTypesWithDirectoryTypes) {
   // Add sessions non-blocking type.
   registry()->ConnectSyncTypeToWorker(syncer::SESSIONS,
                                       MakeInitialDataTypeState(SESSIONS),
+                                      UpdateResponseDataList(),
                                       task_runner,
                                       sessions_sync_proxy.AsWeakPtrForUI());
   current_types.Put(syncer::SESSIONS);
@@ -235,10 +239,12 @@ TEST_F(ModelTypeRegistryTest, DeletionOrdering) {
 
   registry()->ConnectSyncTypeToWorker(syncer::THEMES,
                                       MakeInitialDataTypeState(THEMES),
+                                      UpdateResponseDataList(),
                                       task_runner,
                                       themes_sync_proxy->AsWeakPtrForUI());
   registry()->ConnectSyncTypeToWorker(syncer::SESSIONS,
                                       MakeInitialDataTypeState(SESSIONS),
+                                      UpdateResponseDataList(),
                                       task_runner,
                                       sessions_sync_proxy->AsWeakPtrForUI());
   EXPECT_TRUE(registry()->GetEnabledTypes().Equals(
diff --git a/sync/test/engine/injectable_sync_context_proxy.cc b/sync/test/engine/injectable_sync_context_proxy.cc
index 27de880..310a198 100644
--- a/sync/test/engine/injectable_sync_context_proxy.cc
+++ b/sync/test/engine/injectable_sync_context_proxy.cc
@@ -20,6 +20,7 @@ InjectableSyncContextProxy::~InjectableSyncContextProxy() {
 void InjectableSyncContextProxy::ConnectTypeToSync(
     syncer::ModelType type,
     const DataTypeState& data_type_state,
+    const UpdateResponseDataList& response_list,
     const base::WeakPtr<syncer::ModelTypeSyncProxyImpl>& type_sync_proxy) {
   // This class is allowed to participate in only one connection.
   DCHECK(!is_worker_connected_);
diff --git a/sync/test/engine/injectable_sync_context_proxy.h b/sync/test/engine/injectable_sync_context_proxy.h
index 7c9f48d..a6303f1 100644
--- a/sync/test/engine/injectable_sync_context_proxy.h
+++ b/sync/test/engine/injectable_sync_context_proxy.h
@@ -6,6 +6,7 @@
 #define SYNC_TEST_ENGINE_INJECTABLE_SYNC_CONTEXT_PROXY_H_
 
 #include "sync/internal_api/public/base/model_type.h"
+#include "sync/internal_api/public/non_blocking_sync_common.h"
 #include "sync/internal_api/public/sync_context_proxy.h"
 
 namespace syncer {
@@ -24,6 +25,7 @@ class InjectableSyncContextProxy : public syncer::SyncContextProxy {
   virtual void ConnectTypeToSync(
       syncer::ModelType type,
       const DataTypeState& data_type_state,
+      const UpdateResponseDataList& pending_updates,
       const base::WeakPtr<syncer::ModelTypeSyncProxyImpl>& type_sync_proxy)
       OVERRIDE;
   virtual void Disconnect(syncer::ModelType type) OVERRIDE;
diff --git a/sync/test/engine/mock_model_type_sync_proxy.cc b/sync/test/engine/mock_model_type_sync_proxy.cc
index 56f53fb1..423a37a 100644
--- a/sync/test/engine/mock_model_type_sync_proxy.cc
+++ b/sync/test/engine/mock_model_type_sync_proxy.cc
@@ -29,11 +29,13 @@ void MockModelTypeSyncProxy::OnCommitCompleted(
 
 void MockModelTypeSyncProxy::OnUpdateReceived(
     const DataTypeState& type_state,
-    const UpdateResponseDataList& response_list) {
+    const UpdateResponseDataList& response_list,
+    const UpdateResponseDataList& pending_updates) {
   base::Closure task = base::Bind(&MockModelTypeSyncProxy::OnUpdateReceivedImpl,
                                   base::Unretained(this),
                                   type_state,
-                                  response_list);
+                                  response_list,
+                                  pending_updates);
   pending_tasks_.push_back(task);
   if (is_synchronous_)
     RunQueuedTasks();
@@ -111,6 +113,12 @@ UpdateResponseDataList MockModelTypeSyncProxy::GetNthUpdateResponse(
   return received_update_responses_[n];
 }
 
+UpdateResponseDataList MockModelTypeSyncProxy::GetNthPendingUpdates(
+    size_t n) const {
+  DCHECK_LT(n, GetNumUpdateResponses());
+  return received_pending_updates_[n];
+}
+
 DataTypeState MockModelTypeSyncProxy::GetNthTypeStateReceivedInUpdateResponse(
     size_t n) const {
   DCHECK_LT(n, GetNumUpdateResponses());
@@ -181,8 +189,10 @@ void MockModelTypeSyncProxy::OnCommitCompletedImpl(
 
 void MockModelTypeSyncProxy::OnUpdateReceivedImpl(
     const DataTypeState& type_state,
-    const UpdateResponseDataList& response_list) {
+    const UpdateResponseDataList& response_list,
+    const UpdateResponseDataList& pending_updates) {
   received_update_responses_.push_back(response_list);
+  received_pending_updates_.push_back(pending_updates);
   type_states_received_on_update_.push_back(type_state);
   for (UpdateResponseDataList::const_iterator it = response_list.begin();
        it != response_list.end();
diff --git a/sync/test/engine/mock_model_type_sync_proxy.h b/sync/test/engine/mock_model_type_sync_proxy.h
index 271b59f..140904b 100644
--- a/sync/test/engine/mock_model_type_sync_proxy.h
+++ b/sync/test/engine/mock_model_type_sync_proxy.h
@@ -36,7 +36,8 @@ class MockModelTypeSyncProxy : public ModelTypeSyncProxy {
       const CommitResponseDataList& response_list) OVERRIDE;
   virtual void OnUpdateReceived(
       const DataTypeState& type_state,
-      const UpdateResponseDataList& response_list) OVERRIDE;
+      const UpdateResponseDataList& response_list,
+      const UpdateResponseDataList& pending_updates) OVERRIDE;
 
   // By default, this object behaves as if all messages are processed
   // immediately.  Sometimes it is useful to defer work until later, as might
@@ -65,6 +66,7 @@ class MockModelTypeSyncProxy : public ModelTypeSyncProxy {
   // Does not includes repsonses that are in pending tasks.
   size_t GetNumUpdateResponses() const;
   UpdateResponseDataList GetNthUpdateResponse(size_t n) const;
+  UpdateResponseDataList GetNthPendingUpdates(size_t n) const;
   DataTypeState GetNthTypeStateReceivedInUpdateResponse(size_t n) const;
 
   // Getters to access the log of received commit responses.
@@ -93,7 +95,8 @@ class MockModelTypeSyncProxy : public ModelTypeSyncProxy {
   //
   // Implemented as an Impl method so we can defer its execution in some cases.
   void OnUpdateReceivedImpl(const DataTypeState& type_state,
-                            const UpdateResponseDataList& response_list);
+                            const UpdateResponseDataList& response_list,
+                            const UpdateResponseDataList& pending_updates);
 
   // Getter and setter for per-item sequence number tracking.
   int64 GetCurrentSequenceNumber(const std::string& tag_hash) const;
@@ -116,6 +119,7 @@ class MockModelTypeSyncProxy : public ModelTypeSyncProxy {
   // A log of messages received by this object.
   std::vector<CommitResponseDataList> received_commit_responses_;
   std::vector<UpdateResponseDataList> received_update_responses_;
+  std::vector<UpdateResponseDataList> received_pending_updates_;
   std::vector<DataTypeState> type_states_received_on_update_;
   std::vector<DataTypeState> type_states_received_on_commit_;
 
diff --git a/sync/test/engine/mock_model_type_sync_worker.cc b/sync/test/engine/mock_model_type_sync_worker.cc
index 9c90232..8b6b523 100644
--- a/sync/test/engine/mock_model_type_sync_worker.cc
+++ b/sync/test/engine/mock_model_type_sync_worker.cc
@@ -94,6 +94,8 @@ UpdateResponseData MockModelTypeSyncWorker::UpdateFromServer(
   data.mtime = data.ctime + base::TimeDelta::FromSeconds(version);
   data.non_unique_name = specifics.preference().name();
 
+  data.encryption_key_name = server_encryption_key_name_;
+
   return data;
 }
 
@@ -118,6 +120,8 @@ UpdateResponseData MockModelTypeSyncWorker::TombstoneFromServer(
   data.mtime = data.ctime + base::TimeDelta::FromSeconds(version);
   data.non_unique_name = "Name Non Unique";
 
+  data.encryption_key_name = server_encryption_key_name_;
+
   return data;
 }
 
@@ -149,6 +153,11 @@ CommitResponseData MockModelTypeSyncWorker::SuccessfulCommitResponse(
   return response_data;
 }
 
+void MockModelTypeSyncWorker::SetServerEncryptionKey(
+    const std::string& key_name) {
+  server_encryption_key_name_ = key_name;
+}
+
 std::string MockModelTypeSyncWorker::GenerateId(const std::string& tag_hash) {
   return "FakeId:" + tag_hash;
 }
diff --git a/sync/test/engine/mock_model_type_sync_worker.h b/sync/test/engine/mock_model_type_sync_worker.h
index 498e9ee..b33e548 100644
--- a/sync/test/engine/mock_model_type_sync_worker.h
+++ b/sync/test/engine/mock_model_type_sync_worker.h
@@ -57,6 +57,11 @@ class MockModelTypeSyncWorker : public ModelTypeSyncWorker {
   CommitResponseData SuccessfulCommitResponse(
       const CommitRequestData& request_data);
 
+  // Sets the encryption key name used for updates from the server.
+  // (ie. the key other clients are using to encrypt their commits.)
+  // The default value is an empty string, which indicates no encryption.
+  void SetServerEncryptionKey(const std::string& key_name);
+
  private:
   // Generate an ID string.
   static std::string GenerateId(const std::string& tag_hash);
@@ -72,6 +77,9 @@ class MockModelTypeSyncWorker : public ModelTypeSyncWorker {
   // This is an essential part of the mocked server state.
   std::map<const std::string, int64> server_versions_;
 
+  // Name of the encryption key in use on other clients.
+  std::string server_encryption_key_name_;
+
   DISALLOW_COPY_AND_ASSIGN(MockModelTypeSyncWorker);
 };
 
diff --git a/sync/util/cryptographer.cc b/sync/util/cryptographer.cc
index 29f3781..cb155b5 100644
--- a/sync/util/cryptographer.cc
+++ b/sync/util/cryptographer.cc
@@ -251,7 +251,7 @@ bool Cryptographer::DecryptPendingKeys(const KeyParams& params) {
 
 bool Cryptographer::GetBootstrapToken(std::string* token) const {
   DCHECK(token);
-  std::string unencrypted_token = GetDefaultNigoriKey();
+  std::string unencrypted_token = GetDefaultNigoriKeyData();
   if (unencrypted_token.empty())
     return false;
 
@@ -324,7 +324,11 @@ bool Cryptographer::KeybagIsStale(
   return false;
 }
 
-std::string Cryptographer::GetDefaultNigoriKey() const {
+std::string Cryptographer::GetDefaultNigoriKeyName() const {
+  return default_nigori_name_;
+}
+
+std::string Cryptographer::GetDefaultNigoriKeyData() const {
   if (!is_initialized())
     return std::string();
   NigoriMap::const_iterator iter = nigoris_.find(default_nigori_name_);
diff --git a/sync/util/cryptographer.h b/sync/util/cryptographer.h
index 2dfdedc..9876f83 100644
--- a/sync/util/cryptographer.h
+++ b/sync/util/cryptographer.h
@@ -176,9 +176,12 @@ class SYNC_EXPORT Cryptographer {
   // and/or has a different default key.
   bool KeybagIsStale(const sync_pb::EncryptedData& keybag) const;
 
+  // Returns the name of the Nigori key currently used for encryption.
+  std::string GetDefaultNigoriKeyName() const;
+
   // Returns a serialized sync_pb::NigoriKey version of current default
   // encryption key.
-  std::string GetDefaultNigoriKey() const;
+  std::string GetDefaultNigoriKeyData() const;
 
   // Generates a new Nigori from |serialized_nigori_key|, and if successful
   // installs the new nigori as the default key.