[Sync] Refactor non-frontend DTC to handle new API properly.

We now support disconnecting the syncableservice from the syncer via
it's sync change processor. AutofillProfile has been modified to support
this. As a result of the refactor and this disconnect functionality, we don't
need to block sync shutdown on datatypes implemented the new API, even
if they don't run on the UI thread.

From here on, datatypes that are not on the UI thread should have their
controller inherit from NewNonFrontendDataTypeController, and should
have their syncable service take ownership of the sync change processor it
receives.

A remaining TODO is to apply these changes to UI thread based datatypes.
This involves having them all take ownership of their sync change processors
(this change has their datatype controllers take ownership instead)
and creating a new parent datatype controller class that is API aware,
allowing us to remove the SyncableServiceAdapater completely. See
crbug.com/100114.

BUG=96889
TEST=unit_tests, integration tests

Review URL: http://codereview.chromium.org/8065016

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

46 files changed, 1618 insertions, 193 deletions

diff --git a/chrome/browser/sync/api/sync_change_processor.cc b/chrome/browser/sync/api/sync_change_processor.cc
index b86da3c..588d17a 100644
--- a/chrome/browser/sync/api/sync_change_processor.cc
+++ b/chrome/browser/sync/api/sync_change_processor.cc
@@ -4,4 +4,6 @@
 
 #include "chrome/browser/sync/api/sync_change_processor.h"
 
+SyncChangeProcessor::SyncChangeProcessor() {}
+
 SyncChangeProcessor::~SyncChangeProcessor() {}
diff --git a/chrome/browser/sync/api/sync_change_processor.h b/chrome/browser/sync/api/sync_change_processor.h
index edc7c0f..a68dc59 100644
--- a/chrome/browser/sync/api/sync_change_processor.h
+++ b/chrome/browser/sync/api/sync_change_processor.h
@@ -1,4 +1,4 @@
-  // Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
@@ -21,6 +21,9 @@ typedef std::vector<SyncChange> SyncChangeList;
 // An interface for services that handle receiving SyncChanges.
 class SyncChangeProcessor {
  public:
+  SyncChangeProcessor();
+  virtual ~SyncChangeProcessor();
+
   // Process a list of SyncChanges.
   // Returns: A default SyncError (IsSet() == false) if no errors were
   //          encountered, and a filled SyncError (IsSet() == true)
@@ -32,8 +35,6 @@ class SyncChangeProcessor {
       const tracked_objects::Location& from_here,
       const SyncChangeList& change_list) = 0;
  protected:
-  virtual ~SyncChangeProcessor();
 };
 
-
 #endif  // CHROME_BROWSER_SYNC_API_SYNC_CHANGE_PROCESSOR_H_
diff --git a/chrome/browser/sync/api/syncable_service.h b/chrome/browser/sync/api/syncable_service.h
index 9d7ae62..a2b0213 100644
--- a/chrome/browser/sync/api/syncable_service.h
+++ b/chrome/browser/sync/api/syncable_service.h
@@ -9,6 +9,7 @@
 #include <vector>
 
 #include "base/compiler_specific.h"
+#include "base/memory/weak_ptr.h"
 #include "chrome/browser/sync/syncable/model_type.h"
 #include "chrome/browser/sync/api/sync_change_processor.h"
 #include "chrome/browser/sync/api/sync_data.h"
@@ -18,7 +19,11 @@ class SyncData;
 
 typedef std::vector<SyncData> SyncDataList;
 
-class SyncableService : public SyncChangeProcessor {
+// TODO(zea): remove SupportsWeakPtr in favor of having all SyncableService
+// implementers provide a way of getting a weak pointer to themselves.
+// See crbug.com/100114.
+class SyncableService : public SyncChangeProcessor,
+                        public base::SupportsWeakPtr<SyncableService> {
  public:
   // Informs the service to begin syncing the specified synced datatype |type|.
   // The service should then merge |initial_sync_data| into it's local data,
diff --git a/chrome/browser/sync/glue/app_data_type_controller.cc b/chrome/browser/sync/glue/app_data_type_controller.cc
index 5be7863..9406246 100644
--- a/chrome/browser/sync/glue/app_data_type_controller.cc
+++ b/chrome/browser/sync/glue/app_data_type_controller.cc
@@ -36,7 +36,13 @@ void AppDataTypeController::CreateSyncComponents() {
       profile_sync_factory_->CreateAppSyncComponents(sync_service_,
                                                      this);
   set_model_associator(sync_components.model_associator);
-  set_change_processor(sync_components.change_processor);}
+  generic_change_processor_.reset(static_cast<GenericChangeProcessor*>(
+      sync_components.change_processor));
+}
+
+GenericChangeProcessor* AppDataTypeController::change_processor() const {
+  return generic_change_processor_.get();
+}
 
 void AppDataTypeController::RecordUnrecoverableError(
     const tracked_objects::Location& from_here,
diff --git a/chrome/browser/sync/glue/app_data_type_controller.h b/chrome/browser/sync/glue/app_data_type_controller.h
index 93bd517..3d0bad4 100644
--- a/chrome/browser/sync/glue/app_data_type_controller.h
+++ b/chrome/browser/sync/glue/app_data_type_controller.h
@@ -8,6 +8,7 @@
 
 #include <string>
 
+#include "chrome/browser/sync/glue/generic_change_processor.h"
 #include "chrome/browser/sync/glue/frontend_data_type_controller.h"
 
 namespace browser_sync {
@@ -23,6 +24,9 @@ class AppDataTypeController : public FrontendDataTypeController {
   // DataTypeController implementation.
   virtual syncable::ModelType type() const;
 
+ protected:
+  virtual GenericChangeProcessor* change_processor() const OVERRIDE;
+
  private:
   // DataTypeController implementations.
   virtual bool StartModels();
@@ -33,6 +37,8 @@ class AppDataTypeController : public FrontendDataTypeController {
   virtual void RecordAssociationTime(base::TimeDelta time);
   virtual void RecordStartFailure(StartResult result);
 
+  scoped_ptr<GenericChangeProcessor> generic_change_processor_;
+
   DISALLOW_COPY_AND_ASSIGN(AppDataTypeController);
 };
 
diff --git a/chrome/browser/sync/glue/autofill_data_type_controller.cc b/chrome/browser/sync/glue/autofill_data_type_controller.cc
index cfaef42..9d258a6 100644
--- a/chrome/browser/sync/glue/autofill_data_type_controller.cc
+++ b/chrome/browser/sync/glue/autofill_data_type_controller.cc
@@ -102,8 +102,7 @@ void AutofillDataTypeController::CreateSyncComponents() {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::DB));
   DCHECK_EQ(state(), ASSOCIATING);
   ProfileSyncFactory::SyncComponents sync_components =
-      profile_sync_factory()->
-          CreateAutofillSyncComponents(
+      profile_sync_factory()->CreateAutofillSyncComponents(
           profile_sync_service(),
           web_data_service_->GetDatabase(),
           this);
diff --git a/chrome/browser/sync/glue/autofill_profile_data_type_controller.cc b/chrome/browser/sync/glue/autofill_profile_data_type_controller.cc
index e387ea4..9edf21b 100644
--- a/chrome/browser/sync/glue/autofill_profile_data_type_controller.cc
+++ b/chrome/browser/sync/glue/autofill_profile_data_type_controller.cc
@@ -5,31 +5,120 @@
 #include "chrome/browser/sync/glue/autofill_profile_data_type_controller.h"
 
 #include "base/metrics/histogram.h"
+#include "base/task.h"
+#include "chrome/browser/autofill/personal_data_manager.h"
+#include "chrome/browser/autofill/personal_data_manager_factory.h"
+#include "chrome/browser/profiles/profile.h"
+#include "chrome/browser/sync/api/sync_error.h"
+#include "chrome/browser/sync/api/syncable_service.h"
 #include "chrome/browser/sync/profile_sync_factory.h"
+#include "chrome/browser/sync/profile_sync_service.h"
 #include "chrome/browser/webdata/web_data_service.h"
+#include "chrome/common/chrome_notification_types.h"
+#include "content/browser/browser_thread.h"
+#include "content/common/notification_service.h"
+#include "content/common/notification_source.h"
 
 namespace browser_sync {
 
 AutofillProfileDataTypeController::AutofillProfileDataTypeController(
     ProfileSyncFactory* profile_sync_factory,
     Profile* profile)
-    : AutofillDataTypeController(profile_sync_factory, profile) {}
+    : NewNonFrontendDataTypeController(profile_sync_factory, profile),
+      personal_data_(NULL) {
+}
 
 AutofillProfileDataTypeController::~AutofillProfileDataTypeController() {}
 
+bool AutofillProfileDataTypeController::StartModels() {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  DCHECK_EQ(state(), MODEL_STARTING);
+  // Waiting for the personal data is subtle:  we do this as the PDM resets
+  // its cache of unique IDs once it gets loaded. If we were to proceed with
+  // association, the local ids in the mappings would wind up colliding.
+  personal_data_ = PersonalDataManagerFactory::GetForProfile(profile());
+  if (!personal_data_->IsDataLoaded()) {
+    personal_data_->SetObserver(this);
+    return false;
+  }
+
+  web_data_service_ = profile()->GetWebDataService(Profile::IMPLICIT_ACCESS);
+  if (web_data_service_.get() && web_data_service_->IsDatabaseLoaded()) {
+    return true;
+  } else {
+    notification_registrar_.Add(this, chrome::NOTIFICATION_WEB_DATABASE_LOADED,
+                                NotificationService::AllSources());
+    return false;
+  }
+}
+
+void AutofillProfileDataTypeController::OnPersonalDataChanged() {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  DCHECK_EQ(state(), MODEL_STARTING);
+  personal_data_->RemoveObserver(this);
+  web_data_service_ = profile()->GetWebDataService(Profile::IMPLICIT_ACCESS);
+  if (web_data_service_.get() && web_data_service_->IsDatabaseLoaded()) {
+    DoStartAssociationAsync();
+  } else {
+    notification_registrar_.Add(this, chrome::NOTIFICATION_WEB_DATABASE_LOADED,
+                                NotificationService::AllSources());
+  }
+}
+
+void AutofillProfileDataTypeController::Observe(
+    int notification_type,
+    const NotificationSource& source,
+    const NotificationDetails& details) {
+  notification_registrar_.RemoveAll();
+  DoStartAssociationAsync();
+}
+
+void AutofillProfileDataTypeController::DoStartAssociationAsync() {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  DCHECK_EQ(state(), MODEL_STARTING);
+  set_state(ASSOCIATING);
+  if (!StartAssociationAsync()) {
+    SyncError error(FROM_HERE,
+                    "Failed to post association task.",
+                    type());
+    StartDoneImpl(ASSOCIATION_FAILED, NOT_RUNNING, error);
+  }
+}
+
+bool AutofillProfileDataTypeController::StartAssociationAsync() {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  DCHECK_EQ(state(), ASSOCIATING);
+  return BrowserThread::PostTask(BrowserThread::DB, FROM_HERE,
+      base::Bind(&AutofillProfileDataTypeController::StartAssociation,
+                 this));
+}
+
+base::WeakPtr<SyncableService>
+    AutofillProfileDataTypeController::GetWeakPtrToSyncableService() const {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::DB));
+  return profile_sync_factory()->GetAutofillProfileSyncableService(
+      web_data_service_.get());
+}
+
+void AutofillProfileDataTypeController::StopModels() {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  DCHECK(state() == STOPPING || state() == NOT_RUNNING);
+  notification_registrar_.RemoveAll();
+  personal_data_->RemoveObserver(this);
+}
+
+void AutofillProfileDataTypeController::StopLocalServiceAsync() {
+  BrowserThread::PostTask(BrowserThread::DB, FROM_HERE,
+      base::Bind(&AutofillProfileDataTypeController::StopLocalService,
+                 this));
+}
 syncable::ModelType AutofillProfileDataTypeController::type() const {
   return syncable::AUTOFILL_PROFILE;
 }
 
-void AutofillProfileDataTypeController::CreateSyncComponents() {
-  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::DB));
-  ProfileSyncFactory::SyncComponents sync_components =
-      profile_sync_factory()->
-          CreateAutofillProfileSyncComponents(profile_sync_service(),
-                                              web_data_service(),
-                                              this);
-  set_model_associator(sync_components.model_associator);
-  set_change_processor(sync_components.change_processor);
+browser_sync::ModelSafeGroup
+    AutofillProfileDataTypeController::model_safe_group() const {
+  return browser_sync::GROUP_DB;
 }
 
 void AutofillProfileDataTypeController::RecordUnrecoverableError(
diff --git a/chrome/browser/sync/glue/autofill_profile_data_type_controller.h b/chrome/browser/sync/glue/autofill_profile_data_type_controller.h
index 64f0f49..f55dd8a 100644
--- a/chrome/browser/sync/glue/autofill_profile_data_type_controller.h
+++ b/chrome/browser/sync/glue/autofill_profile_data_type_controller.h
@@ -6,27 +6,64 @@
 #define CHROME_BROWSER_SYNC_GLUE_AUTOFILL_PROFILE_DATA_TYPE_CONTROLLER_H_
 #pragma once
 
-#include "chrome/browser/sync/glue/autofill_data_type_controller.h"
-#include "chrome/browser/sync/profile_sync_factory.h"
+#include "base/compiler_specific.h"
+#include "base/memory/ref_counted.h"
+#include "chrome/browser/autofill/personal_data_manager_observer.h"
+#include "chrome/browser/sync/glue/new_non_frontend_data_type_controller.h"
+#include "content/common/notification_observer.h"
+#include "content/common/notification_registrar.h"
+
+class NotificationDetails;
+class NotificationSource;
+class PersonalDataManager;
+class WebDataService;
 
 namespace browser_sync {
 
-class AutofillProfileDataTypeController : public AutofillDataTypeController {
+class AutofillProfileDataTypeController
+    : public NewNonFrontendDataTypeController,
+      public NotificationObserver,
+      public PersonalDataManagerObserver {
  public:
   AutofillProfileDataTypeController(
       ProfileSyncFactory* profile_sync_factory,
       Profile* profile);
   virtual ~AutofillProfileDataTypeController();
 
-  virtual syncable::ModelType type() const;
+  // NewNonFrontendDataTypeController implementation.
+  virtual syncable::ModelType type() const OVERRIDE;
+  virtual browser_sync::ModelSafeGroup model_safe_group() const OVERRIDE;
+
+  // NotificationObserver implementation.
+  virtual void Observe(int type,
+                       const NotificationSource& source,
+                       const NotificationDetails& details) OVERRIDE;
+
+  // PersonalDataManagerObserver implementation:
+  virtual void OnPersonalDataChanged() OVERRIDE;
 
  protected:
-   virtual void CreateSyncComponents();
+   // NewNonFrontendDataTypeController implementation.
+   virtual bool StartModels() OVERRIDE;
+   virtual bool StartAssociationAsync() OVERRIDE;
+   virtual base::WeakPtr<SyncableService> GetWeakPtrToSyncableService()
+      const OVERRIDE;
+   virtual void StopModels() OVERRIDE;
+   virtual void StopLocalServiceAsync() OVERRIDE;
    virtual void RecordUnrecoverableError(
        const tracked_objects::Location& from_here,
-       const std::string& message);
-   virtual void RecordAssociationTime(base::TimeDelta time);
-   virtual void RecordStartFailure(StartResult result);
+       const std::string& message) OVERRIDE;
+   virtual void RecordAssociationTime(base::TimeDelta time) OVERRIDE;
+   virtual void RecordStartFailure(StartResult result) OVERRIDE;
+
+   void DoStartAssociationAsync();
+
+ private:
+   PersonalDataManager* personal_data_;
+   scoped_refptr<WebDataService> web_data_service_;
+   NotificationRegistrar notification_registrar_;
+
+   DISALLOW_COPY_AND_ASSIGN(AutofillProfileDataTypeController);
 };
 
 }  // namespace browser_sync
diff --git a/chrome/browser/sync/glue/change_processor.cc b/chrome/browser/sync/glue/change_processor.cc
index dcf7800..7f66a81 100644
--- a/chrome/browser/sync/glue/change_processor.cc
+++ b/chrome/browser/sync/glue/change_processor.cc
@@ -39,11 +39,11 @@ bool ChangeProcessor::IsRunning() const {
 // Not implemented by default.
 void ChangeProcessor::CommitChangesFromSyncModel() {}
 
-UnrecoverableErrorHandler* ChangeProcessor::error_handler() {
+UnrecoverableErrorHandler* ChangeProcessor::error_handler() const {
   return error_handler_;
 }
 
-sync_api::UserShare* ChangeProcessor::share_handle() {
+sync_api::UserShare* ChangeProcessor::share_handle() const {
   return share_handle_;
 }
 
diff --git a/chrome/browser/sync/glue/change_processor.h b/chrome/browser/sync/glue/change_processor.h
index 2c581c9..1b6b976 100644
--- a/chrome/browser/sync/glue/change_processor.h
+++ b/chrome/browser/sync/glue/change_processor.h
@@ -75,9 +75,9 @@ class ChangeProcessor {
   virtual void StartImpl(Profile* profile) = 0;
   virtual void StopImpl() = 0;
 
-  bool running() { return running_; }
-  UnrecoverableErrorHandler* error_handler();
-  virtual sync_api::UserShare* share_handle();
+  bool running() const { return running_; }
+  UnrecoverableErrorHandler* error_handler() const;
+  virtual sync_api::UserShare* share_handle() const;
 
  private:
   bool running_;  // True if we have been told it is safe to process changes.
diff --git a/chrome/browser/sync/glue/extension_data_type_controller.cc b/chrome/browser/sync/glue/extension_data_type_controller.cc
index 11f4a72..ad977e8 100644
--- a/chrome/browser/sync/glue/extension_data_type_controller.cc
+++ b/chrome/browser/sync/glue/extension_data_type_controller.cc
@@ -36,7 +36,12 @@ void ExtensionDataTypeController::CreateSyncComponents() {
       profile_sync_factory_->CreateExtensionSyncComponents(sync_service_,
                                                      this);
   set_model_associator(sync_components.model_associator);
-  set_change_processor(sync_components.change_processor);
+  generic_change_processor_.reset(static_cast<GenericChangeProcessor*>(
+      sync_components.change_processor));
+}
+
+GenericChangeProcessor* ExtensionDataTypeController::change_processor() const {
+  return generic_change_processor_.get();
 }
 
 void ExtensionDataTypeController::RecordUnrecoverableError(
diff --git a/chrome/browser/sync/glue/extension_data_type_controller.h b/chrome/browser/sync/glue/extension_data_type_controller.h
index 761db08..61c30aa 100644
--- a/chrome/browser/sync/glue/extension_data_type_controller.h
+++ b/chrome/browser/sync/glue/extension_data_type_controller.h
@@ -8,6 +8,7 @@
 
 #include <string>
 
+#include "chrome/browser/sync/glue/generic_change_processor.h"
 #include "chrome/browser/sync/glue/frontend_data_type_controller.h"
 
 namespace browser_sync {
@@ -23,6 +24,9 @@ class ExtensionDataTypeController : public FrontendDataTypeController {
   // DataTypeController implementation.
   virtual syncable::ModelType type() const;
 
+ protected:
+  virtual GenericChangeProcessor* change_processor() const OVERRIDE;
+
  private:
   // DataTypeController implementations.
   virtual bool StartModels();
@@ -33,6 +37,8 @@ class ExtensionDataTypeController : public FrontendDataTypeController {
   virtual void RecordAssociationTime(base::TimeDelta time);
   virtual void RecordStartFailure(StartResult result);
 
+  scoped_ptr<GenericChangeProcessor> generic_change_processor_;
+
   DISALLOW_COPY_AND_ASSIGN(ExtensionDataTypeController);
 };
 
diff --git a/chrome/browser/sync/glue/frontend_data_type_controller.cc b/chrome/browser/sync/glue/frontend_data_type_controller.cc
index 71fb520..770e5c3 100644
--- a/chrome/browser/sync/glue/frontend_data_type_controller.cc
+++ b/chrome/browser/sync/glue/frontend_data_type_controller.cc
@@ -102,7 +102,7 @@ bool FrontendDataTypeController::Associate() {
   }
 
   sync_service_->ActivateDataType(type(), model_safe_group(),
-                                  change_processor_.get());
+                                  change_processor());
   state_ = RUNNING;
   // FinishStart() invokes the DataTypeManager callback, which can lead to a
   // call to Stop() if one of the other data types being started generates an
diff --git a/chrome/browser/sync/glue/generic_change_processor.cc b/chrome/browser/sync/glue/generic_change_processor.cc
index 8a064af2..56e17e6 100644
--- a/chrome/browser/sync/glue/generic_change_processor.cc
+++ b/chrome/browser/sync/glue/generic_change_processor.cc
@@ -16,27 +16,28 @@
 #include "chrome/browser/sync/internal_api/write_node.h"
 #include "chrome/browser/sync/internal_api/write_transaction.h"
 #include "chrome/browser/sync/unrecoverable_error_handler.h"
+#include "content/browser/browser_thread.h"
 
 namespace browser_sync {
 
 GenericChangeProcessor::GenericChangeProcessor(
-    SyncableService* local_service,
     UnrecoverableErrorHandler* error_handler,
+    const base::WeakPtr<SyncableService>& local_service,
     sync_api::UserShare* user_share)
     : ChangeProcessor(error_handler),
       local_service_(local_service),
-      user_share_(user_share) {
-  DCHECK(local_service_);
+      share_handle_(user_share) {
+  DCHECK(CalledOnValidThread());
 }
 
 GenericChangeProcessor::~GenericChangeProcessor() {
-  // Set to null to ensure it's not used after destruction.
-  local_service_ = NULL;
+  DCHECK(CalledOnValidThread());
 }
 
 void GenericChangeProcessor::ApplyChangesFromSyncModel(
     const sync_api::BaseTransaction* trans,
     const sync_api::ImmutableChangeRecordList& changes) {
+  DCHECK(CalledOnValidThread());
   DCHECK(running());
   DCHECK(syncer_changes_.empty());
   for (sync_api::ChangeRecordList::const_iterator it =
@@ -65,12 +66,18 @@ void GenericChangeProcessor::ApplyChangesFromSyncModel(
 }
 
 void GenericChangeProcessor::CommitChangesFromSyncModel() {
+  DCHECK(CalledOnValidThread());
   if (!running())
     return;
   if (syncer_changes_.empty())
     return;
+  if (!local_service_) {
+    syncable::ModelType type = syncer_changes_[0].sync_data().GetDataType();
+    SyncError error(FROM_HERE, "Local service destroyed.", type);
+    error_handler()->OnUnrecoverableError(error.location(), error.message());
+  }
   SyncError error = local_service_->ProcessSyncChanges(FROM_HERE,
-                                                        syncer_changes_);
+                                                       syncer_changes_);
   syncer_changes_.clear();
   if (error.IsSet()) {
     error_handler()->OnUnrecoverableError(error.location(), error.message());
@@ -80,6 +87,7 @@ void GenericChangeProcessor::CommitChangesFromSyncModel() {
 SyncError GenericChangeProcessor::GetSyncDataForType(
     syncable::ModelType type,
     SyncDataList* current_sync_data) {
+  DCHECK(CalledOnValidThread());
   std::string type_name = syncable::ModelTypeToString(type);
   sync_api::ReadTransaction trans(FROM_HERE, share_handle());
   sync_api::ReadNode root(&trans);
@@ -137,6 +145,7 @@ bool AttemptDelete(const SyncChange& change, sync_api::WriteNode* node) {
 SyncError GenericChangeProcessor::ProcessSyncChanges(
     const tracked_objects::Location& from_here,
     const SyncChangeList& list_of_changes) {
+  DCHECK(CalledOnValidThread());
   sync_api::WriteTransaction trans(from_here, share_handle());
 
   for (SyncChangeList::const_iterator iter = list_of_changes.begin();
@@ -216,6 +225,7 @@ SyncError GenericChangeProcessor::ProcessSyncChanges(
 bool GenericChangeProcessor::SyncModelHasUserCreatedNodes(
     syncable::ModelType type,
     bool* has_nodes) {
+  DCHECK(CalledOnValidThread());
   DCHECK(has_nodes);
   DCHECK_NE(type, syncable::UNSPECIFIED);
   std::string type_name = syncable::ModelTypeToString(type);
@@ -236,6 +246,7 @@ bool GenericChangeProcessor::SyncModelHasUserCreatedNodes(
 }
 
 bool GenericChangeProcessor::CryptoReadyIfNecessary(syncable::ModelType type) {
+  DCHECK(CalledOnValidThread());
   DCHECK_NE(type, syncable::UNSPECIFIED);
   // We only access the cryptographer while holding a transaction.
   sync_api::ReadTransaction trans(FROM_HERE, share_handle());
@@ -245,12 +256,17 @@ bool GenericChangeProcessor::CryptoReadyIfNecessary(syncable::ModelType type) {
          trans.GetCryptographer()->is_ready();
 }
 
-void GenericChangeProcessor::StartImpl(Profile* profile) {}
+void GenericChangeProcessor::StartImpl(Profile* profile) {
+  DCHECK(CalledOnValidThread());
+}
 
-void GenericChangeProcessor::StopImpl() {}
+void GenericChangeProcessor::StopImpl() {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+}
 
-sync_api::UserShare* GenericChangeProcessor::share_handle() {
-  return user_share_;
+sync_api::UserShare* GenericChangeProcessor::share_handle() const {
+  DCHECK(CalledOnValidThread());
+  return share_handle_;
 }
 
 }  // namespace browser_sync
diff --git a/chrome/browser/sync/glue/generic_change_processor.h b/chrome/browser/sync/glue/generic_change_processor.h
index 1afe628..9183e63 100644
--- a/chrome/browser/sync/glue/generic_change_processor.h
+++ b/chrome/browser/sync/glue/generic_change_processor.h
@@ -9,6 +9,8 @@
 #include <vector>
 
 #include "base/compiler_specific.h"
+#include "base/memory/weak_ptr.h"
+#include "base/threading/non_thread_safe.h"
 #include "chrome/browser/sync/api/sync_change_processor.h"
 #include "chrome/browser/sync/glue/change_processor.h"
 
@@ -26,11 +28,16 @@ namespace browser_sync {
 // handles all interaction with the sync api, both translating pushes from the
 // local service into transactions and receiving changes from the sync model,
 // which then get converted into SyncChange's and sent to the local service.
+//
+// As a rule, the GenericChangeProcessor is not thread safe, and should only
+// be used on the same thread in which it was created.
 class GenericChangeProcessor : public ChangeProcessor,
-                               public SyncChangeProcessor {
+                               public SyncChangeProcessor,
+                               public base::NonThreadSafe {
  public:
-  GenericChangeProcessor(SyncableService* local_service,
-                         UnrecoverableErrorHandler* error_handler,
+  // Create a change processor and connect it to the syncer.
+  GenericChangeProcessor(UnrecoverableErrorHandler* error_handler,
+                         const base::WeakPtr<SyncableService>& local_service,
                          sync_api::UserShare* user_share);
   virtual ~GenericChangeProcessor();
 
@@ -49,22 +56,26 @@ class GenericChangeProcessor : public ChangeProcessor,
       const SyncChangeList& change_list) OVERRIDE;
 
   // Fills |current_sync_data| with all the syncer data for the specified type.
-  virtual SyncError GetSyncDataForType(syncable::ModelType type,
-                                       SyncDataList* current_sync_data);
+  SyncError GetSyncDataForType(syncable::ModelType type,
+                               SyncDataList* current_sync_data);
 
   // Generic versions of AssociatorInterface methods. Called by
-  // SyncableServiceAdapter.
+  // SyncableServiceAdapter or the DataTypeController.
   bool SyncModelHasUserCreatedNodes(syncable::ModelType type,
                                     bool* has_nodes);
   bool CryptoReadyIfNecessary(syncable::ModelType type);
+
  protected:
   // ChangeProcessor interface.
-  virtual void StartImpl(Profile* profile) OVERRIDE;  // Not implemented.
-  virtual void StopImpl() OVERRIDE;  // Not implemented.
-  virtual sync_api::UserShare* share_handle() OVERRIDE;
+  virtual void StartImpl(Profile* profile) OVERRIDE;           // Does nothing.
+  // Called from UI thread (as part of deactivating datatype), but does
+  // nothing and is guaranteed to still be alive, so it's okay.
+  virtual void StopImpl() OVERRIDE;                            // Does nothing.
+  virtual sync_api::UserShare* share_handle() const OVERRIDE;
+
  private:
   // The SyncableService this change processor will forward changes on to.
-  SyncableService* local_service_;
+  const base::WeakPtr<SyncableService> local_service_;
 
   // The current list of changes received from the syncer. We buffer because
   // we must ensure no syncapi transaction is held when we pass it on to
@@ -77,7 +88,9 @@ class GenericChangeProcessor : public ChangeProcessor,
   // listening to changes (the local_service_ will be interacting with us
   // when it starts up). As such we can't wait until Start(_) has been called,
   // and have to keep a local pointer to the user_share.
-  sync_api::UserShare* user_share_;
+  sync_api::UserShare* const share_handle_;
+
+  DISALLOW_COPY_AND_ASSIGN(GenericChangeProcessor);
 };
 
 }  // namespace browser_sync
diff --git a/chrome/browser/sync/glue/new_non_frontend_data_type_controller.cc b/chrome/browser/sync/glue/new_non_frontend_data_type_controller.cc
new file mode 100644
index 0000000..0989693
--- /dev/null
+++ b/chrome/browser/sync/glue/new_non_frontend_data_type_controller.cc
@@ -0,0 +1,220 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "chrome/browser/sync/glue/new_non_frontend_data_type_controller.h"
+
+#include "base/logging.h"
+#include "chrome/browser/sync/profile_sync_factory.h"
+#include "chrome/browser/sync/profile_sync_service.h"
+#include "chrome/browser/sync/api/sync_error.h"
+#include "chrome/browser/sync/api/syncable_service.h"
+#include "chrome/browser/sync/glue/shared_change_processor_ref.h"
+#include "chrome/browser/sync/syncable/model_type.h"
+#include "content/browser/browser_thread.h"
+
+namespace browser_sync {
+
+NewNonFrontendDataTypeController::NewNonFrontendDataTypeController()
+    : shared_change_processor_(NULL) {}
+
+NewNonFrontendDataTypeController::NewNonFrontendDataTypeController(
+    ProfileSyncFactory* profile_sync_factory,
+    Profile* profile)
+    : NonFrontendDataTypeController(profile_sync_factory, profile),
+      shared_change_processor_(NULL) {
+}
+
+NewNonFrontendDataTypeController::~NewNonFrontendDataTypeController() {}
+
+void NewNonFrontendDataTypeController::Start(StartCallback* start_callback) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  DCHECK(start_callback);
+  if (state() != NOT_RUNNING) {
+    start_callback->Run(BUSY, SyncError());
+    delete start_callback;
+    return;
+  }
+
+  set_start_callback(start_callback);
+
+  set_state(MODEL_STARTING);
+  if (!StartModels()) {
+    // If we are waiting for some external service to load before associating
+    // or we failed to start the models, we exit early.
+    DCHECK(state() == MODEL_STARTING || state() == NOT_RUNNING);
+    return;
+  }
+
+  shared_change_processor_ =
+      profile_sync_factory()->CreateSharedChangeProcessor();
+
+  // Kick off association on the thread the datatype resides on.
+  set_state(ASSOCIATING);
+  if (!StartAssociationAsync()) {
+    // Releasing the shared_change_processor_ here is safe since it was never
+    // connected.
+    shared_change_processor_ = NULL;
+    SyncError error(FROM_HERE, "Failed to post StartAssociation", type());
+    StartDoneImpl(ASSOCIATION_FAILED, NOT_RUNNING, error);
+  }
+}
+
+// This method can execute after we've already stopped (and possibly even
+// destroyed) both the Syncer and the SyncableService. As a result, all actions
+// must either have no side effects outside of the DTC or must be protected
+// by |shared_change_processor_|, which is guaranteed to have been Disconnected
+// if the syncer shut down.
+void NewNonFrontendDataTypeController::StartAssociation() {
+  DCHECK(!BrowserThread::CurrentlyOn(BrowserThread::UI));
+  DCHECK_EQ(state(), ASSOCIATING);
+
+  // We're dependent on the SyncableService being destroyed on the same thread
+  // we access it. Therefore, as long as GetWeakPtrToSyncableService returns
+  // a valid WeakPtr, we can rely on it remaining initialized for the
+  // life of this method.
+  local_service_ = GetWeakPtrToSyncableService();
+  if (!local_service_.get()) {
+    // The SyncableService was destroyed before this task had a chance to
+    // execute.
+    SyncError error(FROM_HERE, "Local service destroyed before association.",
+                    type());
+    StartFailed(UNRECOVERABLE_ERROR, error);
+    return;
+  }
+
+  // Connect |shared_change_processor_| to the syncer and |local_service_|.
+  // Note that it's possible the shared_change_processor_ has already been
+  // disconnected at this point, so all our accesses to the syncer from this
+  // point on are through it.
+  if (!shared_change_processor_->Connect(profile_sync_factory(),
+                                         profile_sync_service(),
+                                         this,
+                                         local_service_)) {
+    SyncError error(FROM_HERE, "Failed to connect to syncer.", type());
+    StartFailed(UNRECOVERABLE_ERROR, error);
+  }
+
+  if (!shared_change_processor_->CryptoReadyIfNecessary(type())) {
+    StartFailed(NEEDS_CRYPTO, SyncError());
+    return;
+  }
+
+  bool sync_has_nodes = false;
+  if (!shared_change_processor_->SyncModelHasUserCreatedNodes(
+          type(), &sync_has_nodes)) {
+    SyncError error(FROM_HERE, "Failed to load sync nodes", type());
+    StartFailed(UNRECOVERABLE_ERROR, error);
+    return;
+  }
+
+  base::TimeTicks start_time = base::TimeTicks::Now();
+  SyncError error;
+  SyncDataList initial_sync_data;
+  error = shared_change_processor_->GetSyncDataForType(type(),
+                                                       &initial_sync_data);
+  if (error.IsSet()) {
+    StartFailed(ASSOCIATION_FAILED, error);
+    return;
+  }
+  // Passes a reference to the shared_change_processor_;
+  error = local_service_->MergeDataAndStartSyncing(
+      type(),
+      initial_sync_data,
+      new SharedChangeProcessorRef(shared_change_processor_));
+  RecordAssociationTime(base::TimeTicks::Now() - start_time);
+  if (error.IsSet()) {
+    local_service_->StopSyncing(type());
+    StartFailed(ASSOCIATION_FAILED, error);
+    return;
+  }
+
+  // If we've been disconnected, profile_sync_service() may return an invalid
+  // pointer, but the shared_change_processor_ protects us from attempting to
+  // access it.
+  // Note: This must be done on the datatype's thread to ensure local_service_
+  // doesn't start trying to push changes from it's thread before we activate
+  // the datatype.
+  shared_change_processor_->ActivateDataType(profile_sync_service(),
+                                             type(), model_safe_group());
+  StartDone(!sync_has_nodes ? OK_FIRST_RUN : OK, RUNNING, SyncError());
+}
+
+void NewNonFrontendDataTypeController::StartDone(
+    DataTypeController::StartResult result,
+    DataTypeController::State new_state,
+    const SyncError& error) {
+  DCHECK(!BrowserThread::CurrentlyOn(BrowserThread::UI));
+  BrowserThread::PostTask(BrowserThread::UI, FROM_HERE,
+      base::Bind(
+          &NewNonFrontendDataTypeController::StartDoneImpl,
+          this,
+          result,
+          new_state,
+          error));
+}
+
+void NewNonFrontendDataTypeController::Stop() {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+
+  // Disconnect the change processor. At this point, the SyncableService
+  // can no longer interact with the Syncer, even if it hasn't finished
+  // MergeDataAndStartSyncing. It may still have ownership of the shared
+  // change processor though.
+  if (shared_change_processor_.get())
+    shared_change_processor_->Disconnect();
+
+  // If we haven't finished starting, we need to abort the start.
+  switch (state()) {
+    case MODEL_STARTING:
+      set_state(STOPPING);
+      StartDoneImpl(ABORTED, NOT_RUNNING, SyncError());
+      return;  // The datatype was never activated, we're done.
+    case ASSOCIATING:
+      set_state(STOPPING);
+      StartDoneImpl(ABORTED, NOT_RUNNING, SyncError());
+      // We continue on to deactivate the datatype.
+      break;
+    case DISABLED:
+      // If we're disabled we never succeded assocaiting and never activated the
+      // datatype.
+      set_state(NOT_RUNNING);
+      StopModels();
+      return;
+    default:
+      // Datatype was fully started.
+      DCHECK_EQ(state(), RUNNING);
+      set_state(STOPPING);
+      StopModels();
+      break;
+  }
+
+  // Deactivate the DataType on the UI thread. We dont want to listen
+  // for any more changes or process them from the server.
+  profile_sync_service()->DeactivateDataType(type());
+
+  // Stop the local service and release our references to it and the
+  // shared change processor (posts a task to the datatype's thread).
+  StopLocalServiceAsync();
+
+  set_state(NOT_RUNNING);
+}
+
+void NewNonFrontendDataTypeController::StopLocalService() {
+  DCHECK(!BrowserThread::CurrentlyOn(BrowserThread::UI));
+  if (local_service_.get())
+    local_service_->StopSyncing(type());
+  local_service_.reset();
+  shared_change_processor_ = NULL;
+}
+
+bool NewNonFrontendDataTypeController::StopAssociationAsync() {
+  NOTIMPLEMENTED();
+  return false;
+}
+
+void NewNonFrontendDataTypeController::CreateSyncComponents() {
+  NOTIMPLEMENTED();
+}
+
+}  // namepsace browser_sync
diff --git a/chrome/browser/sync/glue/new_non_frontend_data_type_controller.h b/chrome/browser/sync/glue/new_non_frontend_data_type_controller.h
new file mode 100644
index 0000000..44cf867
--- /dev/null
+++ b/chrome/browser/sync/glue/new_non_frontend_data_type_controller.h
@@ -0,0 +1,75 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef CHROME_BROWSER_SYNC_GLUE_NEW_NON_FRONTEND_DATA_TYPE_CONTROLLER_H_
+#define CHROME_BROWSER_SYNC_GLUE_NEW_NON_FRONTEND_DATA_TYPE_CONTROLLER_H_
+#pragma once
+
+#include "base/compiler_specific.h"
+#include "base/memory/weak_ptr.h"
+#include "chrome/browser/sync/glue/non_frontend_data_type_controller.h"
+#include "chrome/browser/sync/glue/shared_change_processor.h"
+
+class SyncableService;
+
+namespace browser_sync {
+
+class NewNonFrontendDataTypeController : public NonFrontendDataTypeController {
+ public:
+  NewNonFrontendDataTypeController(
+      ProfileSyncFactory* profile_sync_factory,
+      Profile* profile);
+  virtual ~NewNonFrontendDataTypeController();
+
+  virtual void Start(StartCallback* start_callback) OVERRIDE;
+  virtual void Stop() OVERRIDE;
+
+ protected:
+  NewNonFrontendDataTypeController();
+
+  // Overrides of NonFrontendDataTypeController methods.
+  virtual void StartAssociation() OVERRIDE;
+  virtual void StartDone(DataTypeController::StartResult result,
+                         DataTypeController::State new_state,
+                         const SyncError& error) OVERRIDE;
+
+  // Calls local_service_->StopSyncing() and releases our references to it and
+  // |shared_change_processor_|.
+  virtual void StopLocalService();
+  // Posts StopLocalService() to the datatype's thread.
+  virtual void StopLocalServiceAsync() = 0;
+
+  // Extract/create the syncable service from the profile and return a
+  // WeakHandle to it.
+  virtual base::WeakPtr<SyncableService> GetWeakPtrToSyncableService()
+      const = 0;
+
+ private:
+  // Deprecated.
+  virtual bool StopAssociationAsync() OVERRIDE;
+  virtual void CreateSyncComponents() OVERRIDE;
+
+  // A weak pointer to the actual local syncable service, which performs all the
+  // real work. We do not own the object, and it is only safe to access on the
+  // DataType's thread.
+  // Lifetime: it gets set in StartAssociation() and released in
+  // StopLocalService().
+  base::WeakPtr<SyncableService> local_service_;
+
+  // The thread-safe layer between the datatype and the syncer. It allows us to
+  // disconnect both the datatype and ourselves from the syncer at shutdown
+  // time. All accesses to the syncer from any thread other than the UI thread
+  // should be through this. Once connected, it must be released on the
+  // datatype's thread (but if we never connect it we can destroy it on the UI
+  // thread as well).
+  // Lifetime: It gets created on the UI thread in Start(..), connected to
+  // the syncer in StartAssociation() on the datatype's thread, and if
+  // successfully connected released in StopLocalService() on the datatype's
+  // thread.
+  scoped_refptr<SharedChangeProcessor> shared_change_processor_;
+};
+
+}  // namespace browser_sync
+
+#endif  // CHROME_BROWSER_SYNC_GLUE_NEW_NON_FRONTEND_DATA_TYPE_CONTROLLER_H_
diff --git a/chrome/browser/sync/glue/new_non_frontend_data_type_controller_mock.cc b/chrome/browser/sync/glue/new_non_frontend_data_type_controller_mock.cc
new file mode 100644
index 0000000..a226e2d
--- /dev/null
+++ b/chrome/browser/sync/glue/new_non_frontend_data_type_controller_mock.cc
@@ -0,0 +1,13 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "chrome/browser/sync/glue/new_non_frontend_data_type_controller_mock.h"
+
+namespace browser_sync {
+
+NewNonFrontendDataTypeControllerMock::NewNonFrontendDataTypeControllerMock() {}
+
+NewNonFrontendDataTypeControllerMock::~NewNonFrontendDataTypeControllerMock() {}
+
+}  // namespace browser_sync
diff --git a/chrome/browser/sync/glue/new_non_frontend_data_type_controller_mock.h b/chrome/browser/sync/glue/new_non_frontend_data_type_controller_mock.h
new file mode 100644
index 0000000..50ca5a3
--- /dev/null
+++ b/chrome/browser/sync/glue/new_non_frontend_data_type_controller_mock.h
@@ -0,0 +1,64 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef CHROME_BROWSER_SYNC_GLUE_NEW_NON_FRONTEND_DATA_TYPE_CONTROLLER_MOCK_H_
+#define CHROME_BROWSER_SYNC_GLUE_NEW_NON_FRONTEND_DATA_TYPE_CONTROLLER_MOCK_H_
+#pragma once
+
+#include "chrome/browser/sync/api/sync_error.h"
+#include "chrome/browser/sync/glue/new_non_frontend_data_type_controller.h"
+#include "testing/gmock/include/gmock/gmock.h"
+
+namespace browser_sync {
+
+class NewNonFrontendDataTypeControllerMock
+    : public NewNonFrontendDataTypeController {
+ public:
+  NewNonFrontendDataTypeControllerMock();
+  virtual ~NewNonFrontendDataTypeControllerMock();
+
+  // DataTypeController mocks.
+  MOCK_METHOD1(Start, void(StartCallback* start_callback));
+  MOCK_METHOD0(Stop, void());
+  MOCK_METHOD0(enabled, bool());
+  MOCK_CONST_METHOD0(type, syncable::ModelType());
+  MOCK_CONST_METHOD0(name, std::string());
+  MOCK_CONST_METHOD0(model_safe_group, browser_sync::ModelSafeGroup());
+  MOCK_CONST_METHOD0(state, State());
+  MOCK_METHOD2(OnUnrecoverableError, void(const tracked_objects::Location&,
+                                          const std::string&));
+
+  // NonFrontendDataTypeController mocks.
+  MOCK_METHOD0(StartModels, bool());
+  MOCK_METHOD0(StartAssociationAsync, bool());
+  MOCK_METHOD0(StartAssociation, void());
+  MOCK_METHOD0(CreateSyncComponents, void());
+  MOCK_METHOD2(StartFailed, void(StartResult result,
+                                 const SyncError& error));
+  MOCK_METHOD3(StartDone, void(DataTypeController::StartResult result,
+                               DataTypeController::State new_state,
+                               const SyncError& error));
+  MOCK_METHOD3(StartDoneImpl, void(DataTypeController::StartResult result,
+                                   DataTypeController::State new_state,
+                                   const SyncError& error));
+  MOCK_METHOD0(StopModels, void());
+  MOCK_METHOD0(StopAssociationAsync, bool());
+  MOCK_METHOD0(StopAssociation, void());
+  MOCK_METHOD2(OnUnrecoverableErrorImpl, void(const tracked_objects::Location&,
+                                              const std::string&));
+  MOCK_METHOD2(RecordUnrecoverableError, void(const tracked_objects::Location&,
+                                              const std::string&));
+  MOCK_METHOD1(RecordAssociationTime, void(base::TimeDelta time));
+  MOCK_METHOD1(RecordStartFailure, void(StartResult result));
+
+  // NewNonFrontendDataTypeController mocks.
+  MOCK_CONST_METHOD0(GetWeakPtrToSyncableService,
+                     base::WeakPtr<SyncableService>());
+  MOCK_METHOD0(StopLocalService, void());
+  MOCK_METHOD0(StopLocalServiceAsync, void());
+};
+
+}  // namespace browser_sync
+
+#endif  // CHROME_BROWSER_SYNC_GLUE_NEW_NON_FRONTEND_DATA_TYPE_CONTROLLER_MOCK_H_
diff --git a/chrome/browser/sync/glue/new_non_frontend_data_type_controller_unittest.cc b/chrome/browser/sync/glue/new_non_frontend_data_type_controller_unittest.cc
new file mode 100644
index 0000000..5f6c566
--- /dev/null
+++ b/chrome/browser/sync/glue/new_non_frontend_data_type_controller_unittest.cc
@@ -0,0 +1,404 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "chrome/browser/sync/glue/new_non_frontend_data_type_controller.h"
+
+#include "base/callback.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/message_loop.h"
+#include "base/task.h"
+#include "base/test/test_timeouts.h"
+#include "base/tracked_objects.h"
+#include "base/synchronization/waitable_event.h"
+#include "chrome/browser/sync/api/syncable_service_mock.h"
+#include "chrome/browser/sync/engine/model_safe_worker.h"
+#include "chrome/browser/sync/glue/data_type_controller_mock.h"
+#include "chrome/browser/sync/glue/new_non_frontend_data_type_controller_mock.h"
+#include "chrome/browser/sync/glue/shared_change_processor_mock.h"
+#include "chrome/browser/sync/profile_sync_factory_mock.h"
+#include "chrome/browser/sync/profile_sync_service_mock.h"
+#include "chrome/test/base/profile_mock.h"
+#include "content/browser/browser_thread.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+using base::WaitableEvent;
+using browser_sync::GenericChangeProcessor;
+using browser_sync::SharedChangeProcessorMock;
+using browser_sync::DataTypeController;
+using browser_sync::GROUP_DB;
+using browser_sync::NewNonFrontendDataTypeController;
+using browser_sync::NewNonFrontendDataTypeControllerMock;
+using browser_sync::StartCallback;
+using syncable::AUTOFILL_PROFILE;
+using testing::_;
+using testing::DoAll;
+using testing::InvokeWithoutArgs;
+using testing::Return;
+using testing::SetArgumentPointee;
+using testing::StrictMock;
+
+namespace browser_sync {
+
+namespace {
+
+ACTION_P(WaitOnEvent, event) {
+  event->Wait();
+}
+
+ACTION_P(SignalEvent, event) {
+  event->Signal();
+}
+
+ACTION_P(SaveChangeProcessor, scoped_change_processor) {
+  scoped_change_processor->reset(arg2);
+}
+
+ACTION_P(GetWeakPtrToSyncableService, syncable_service) {
+  // Have to do this within an Action to ensure it's not evaluated on the wrong
+  // thread.
+  return syncable_service->AsWeakPtr();
+}
+
+class NewNonFrontendDataTypeControllerFake
+    : public NewNonFrontendDataTypeController {
+ public:
+  NewNonFrontendDataTypeControllerFake(
+      ProfileSyncFactory* profile_sync_factory,
+      Profile* profile,
+      NewNonFrontendDataTypeControllerMock* mock)
+      : NewNonFrontendDataTypeController(profile_sync_factory,
+                                         profile),
+        mock_(mock) {}
+
+  virtual syncable::ModelType type() const OVERRIDE {
+    return AUTOFILL_PROFILE;
+  }
+  virtual ModelSafeGroup model_safe_group() const OVERRIDE {
+    return GROUP_DB;
+  }
+
+ protected:
+  virtual base::WeakPtr<SyncableService>
+      GetWeakPtrToSyncableService() const OVERRIDE {
+    return profile_sync_factory()->GetAutofillProfileSyncableService(NULL);
+  }
+  virtual bool StartAssociationAsync() OVERRIDE {
+    mock_->StartAssociationAsync();
+    return BrowserThread::PostTask(BrowserThread::DB, FROM_HERE,
+        base::Bind(&NewNonFrontendDataTypeControllerFake::StartAssociation,
+                   this));
+  }
+
+  // We mock the following methods because their default implementations do
+  // nothing, but we still want to make sure they're called appropriately.
+  virtual bool StartModels() OVERRIDE {
+    return mock_->StartModels();
+  }
+  virtual void StopModels() OVERRIDE {
+    mock_->StopModels();
+  }
+  virtual void StopLocalServiceAsync() OVERRIDE {
+    mock_->StopLocalServiceAsync();
+    BrowserThread::PostTask(BrowserThread::DB, FROM_HERE,
+        base::Bind(&NewNonFrontendDataTypeControllerFake::StopLocalService,
+                   this));
+  }
+  virtual void RecordUnrecoverableError(
+      const tracked_objects::Location& from_here,
+      const std::string& message) OVERRIDE {
+    mock_->RecordUnrecoverableError(from_here, message);
+  }
+  virtual void RecordAssociationTime(base::TimeDelta time) OVERRIDE {
+    mock_->RecordAssociationTime(time);
+  }
+  virtual void RecordStartFailure(DataTypeController::StartResult result)
+      OVERRIDE {
+    mock_->RecordStartFailure(result);
+  }
+ private:
+  NewNonFrontendDataTypeControllerMock* mock_;
+};
+
+class NewNonFrontendDataTypeControllerTest : public testing::Test {
+ public:
+  NewNonFrontendDataTypeControllerTest()
+      : ui_thread_(BrowserThread::UI, &message_loop_),
+        db_thread_(BrowserThread::DB) {}
+
+  virtual void SetUp() OVERRIDE {
+    EXPECT_CALL(profile_, GetProfileSyncService()).WillRepeatedly(
+        Return(&service_));
+    EXPECT_CALL(service_, GetUserShare()).WillRepeatedly(
+        Return((sync_api::UserShare*)NULL));
+    db_thread_.Start();
+    profile_sync_factory_.reset(new ProfileSyncFactoryMock());
+    change_processor_ = new SharedChangeProcessorMock();
+
+    // Both of these are refcounted, so don't need to be released.
+    dtc_mock_ = new StrictMock<NewNonFrontendDataTypeControllerMock>();
+    new_non_frontend_dtc_ =
+        new NewNonFrontendDataTypeControllerFake(profile_sync_factory_.get(),
+                                                 &profile_,
+                                                 dtc_mock_.get());
+  }
+
+  virtual void TearDown() OVERRIDE {
+    db_thread_.Stop();
+  }
+
+  void WaitForDTC() {
+    WaitableEvent done(true, false);
+    BrowserThread::PostTask(BrowserThread::DB, FROM_HERE,
+       base::Bind(&NewNonFrontendDataTypeControllerTest::SignalDone,
+                  &done));
+    done.TimedWait(base::TimeDelta::FromMilliseconds(
+        TestTimeouts::action_timeout_ms()));
+    if (!done.IsSignaled()) {
+      ADD_FAILURE() << "Timed out waiting for DB thread to finish.";
+    }
+    MessageLoop::current()->RunAllPending();
+  }
+
+ protected:
+  void SetStartExpectations() {
+    EXPECT_CALL(*dtc_mock_, StartModels()).WillOnce(Return(true));
+    EXPECT_CALL(*profile_sync_factory_,
+                GetAutofillProfileSyncableService(_)).
+        WillOnce(GetWeakPtrToSyncableService(&syncable_service_));
+    EXPECT_CALL(*profile_sync_factory_,
+                CreateSharedChangeProcessor()).
+        WillOnce(Return(change_processor_.get()));
+  }
+
+  void SetAssociateExpectations() {
+    EXPECT_CALL(*dtc_mock_, StartAssociationAsync());
+    EXPECT_CALL(*change_processor_, Connect(_,_,_,_)).WillOnce(Return(true));
+    EXPECT_CALL(*change_processor_, CryptoReadyIfNecessary(_)).
+        WillOnce(Return(true));
+    EXPECT_CALL(*change_processor_, SyncModelHasUserCreatedNodes(_,_)).
+        WillOnce(DoAll(SetArgumentPointee<1>(true), Return(true)));
+    EXPECT_CALL(*change_processor_, GetSyncDataForType(_,_)).
+        WillOnce(Return(SyncError()));
+    EXPECT_CALL(syncable_service_, MergeDataAndStartSyncing(_,_,_)).
+        WillOnce(DoAll(SaveChangeProcessor(&saved_change_processor_),
+                       Return(SyncError())));
+    EXPECT_CALL(*dtc_mock_, RecordAssociationTime(_));
+  }
+
+  void SetActivateExpectations(DataTypeController::StartResult result) {
+    EXPECT_CALL(service_, ActivateDataType(_, _, _));
+    EXPECT_CALL(start_callback_, Run(result,_));
+  }
+
+  void SetStopExpectations() {
+    EXPECT_CALL(*dtc_mock_, StopModels());
+    EXPECT_CALL(*change_processor_, Disconnect()).WillOnce(Return(true));
+    EXPECT_CALL(service_, DeactivateDataType(_));
+    EXPECT_CALL(*dtc_mock_, StopLocalServiceAsync());
+    EXPECT_CALL(syncable_service_, StopSyncing(_));
+  }
+
+  void SetStartFailExpectations(DataTypeController::StartResult result) {
+    EXPECT_CALL(*dtc_mock_, StopModels());
+    EXPECT_CALL(*dtc_mock_, RecordStartFailure(result));
+    EXPECT_CALL(start_callback_, Run(result,_));
+  }
+
+  static void SignalDone(WaitableEvent* done) {
+    done->Signal();
+  }
+
+  MessageLoopForUI message_loop_;
+  BrowserThread ui_thread_;
+  BrowserThread db_thread_;
+  ProfileMock profile_;
+  scoped_ptr<ProfileSyncFactoryMock> profile_sync_factory_;
+  ProfileSyncServiceMock service_;
+  StartCallback start_callback_;
+  // Must be destroyed after new_non_frontend_dtc_.
+  SyncableServiceMock syncable_service_;
+  scoped_refptr<NewNonFrontendDataTypeControllerFake> new_non_frontend_dtc_;
+  scoped_refptr<NewNonFrontendDataTypeControllerMock> dtc_mock_;
+  scoped_refptr<SharedChangeProcessorMock> change_processor_;
+  scoped_ptr<SyncChangeProcessor> saved_change_processor_;
+};
+
+TEST_F(NewNonFrontendDataTypeControllerTest, StartOk) {
+  SetStartExpectations();
+  SetAssociateExpectations();
+  SetActivateExpectations(DataTypeController::OK);
+  EXPECT_EQ(DataTypeController::NOT_RUNNING, new_non_frontend_dtc_->state());
+  new_non_frontend_dtc_->Start(
+      NewCallback(&start_callback_, &StartCallback::Run));
+  WaitForDTC();
+  EXPECT_EQ(DataTypeController::RUNNING, new_non_frontend_dtc_->state());
+}
+
+TEST_F(NewNonFrontendDataTypeControllerTest, StartFirstRun) {
+  SetStartExpectations();
+  EXPECT_CALL(*dtc_mock_, StartAssociationAsync());
+  EXPECT_CALL(*change_processor_, Connect(_,_,_,_)).WillOnce(Return(true));
+  EXPECT_CALL(*change_processor_, CryptoReadyIfNecessary(_)).
+      WillOnce(Return(true));
+  EXPECT_CALL(*change_processor_, SyncModelHasUserCreatedNodes(_,_)).
+      WillOnce(DoAll(SetArgumentPointee<1>(false), Return(true)));
+  EXPECT_CALL(*change_processor_, GetSyncDataForType(_,_)).
+      WillOnce(Return(SyncError()));
+  EXPECT_CALL(syncable_service_, MergeDataAndStartSyncing(_,_,_)).
+    WillOnce(DoAll(SaveChangeProcessor(&saved_change_processor_),
+                   Return(SyncError())));
+  EXPECT_CALL(*dtc_mock_, RecordAssociationTime(_));
+  SetActivateExpectations(DataTypeController::OK_FIRST_RUN);
+  EXPECT_EQ(DataTypeController::NOT_RUNNING, new_non_frontend_dtc_->state());
+  new_non_frontend_dtc_->Start(
+      NewCallback(&start_callback_, &StartCallback::Run));
+  WaitForDTC();
+  EXPECT_EQ(DataTypeController::RUNNING, new_non_frontend_dtc_->state());
+}
+
+TEST_F(NewNonFrontendDataTypeControllerTest, AbortDuringStartModels) {
+  EXPECT_CALL(*dtc_mock_, StartModels()).WillOnce(Return(false));
+  SetStartFailExpectations(DataTypeController::ABORTED);
+  EXPECT_EQ(DataTypeController::NOT_RUNNING, new_non_frontend_dtc_->state());
+  new_non_frontend_dtc_->Start(
+      NewCallback(&start_callback_, &StartCallback::Run));
+  WaitForDTC();
+  EXPECT_EQ(DataTypeController::MODEL_STARTING, new_non_frontend_dtc_->state());
+  new_non_frontend_dtc_->Stop();
+  EXPECT_EQ(DataTypeController::NOT_RUNNING, new_non_frontend_dtc_->state());
+}
+
+TEST_F(NewNonFrontendDataTypeControllerTest, StartAssociationFailed) {
+  SetStartExpectations();
+  EXPECT_CALL(*dtc_mock_, StartAssociationAsync());
+  EXPECT_CALL(*change_processor_, Connect(_,_,_,_)).WillOnce(Return(true));
+  EXPECT_CALL(*change_processor_, CryptoReadyIfNecessary(_)).
+      WillOnce(Return(true));
+  EXPECT_CALL(*change_processor_, SyncModelHasUserCreatedNodes(_,_)).
+      WillOnce(DoAll(SetArgumentPointee<1>(true), Return(true)));
+  EXPECT_CALL(*change_processor_, GetSyncDataForType(_,_)).
+      WillOnce(Return(SyncError()));
+  EXPECT_CALL(syncable_service_, MergeDataAndStartSyncing(_,_,_)).
+    WillOnce(DoAll(SaveChangeProcessor(&saved_change_processor_),
+                   Return(SyncError(FROM_HERE, "failed", AUTOFILL_PROFILE))));
+  EXPECT_CALL(*dtc_mock_, RecordAssociationTime(_));
+  SetStartFailExpectations(DataTypeController::ASSOCIATION_FAILED);
+  EXPECT_CALL(syncable_service_, StopSyncing(_));
+  // Set up association to fail with an association failed error.
+  EXPECT_EQ(DataTypeController::NOT_RUNNING, new_non_frontend_dtc_->state());
+  new_non_frontend_dtc_->Start(
+      NewCallback(&start_callback_, &StartCallback::Run));
+  WaitForDTC();
+  EXPECT_EQ(DataTypeController::DISABLED, new_non_frontend_dtc_->state());
+}
+
+TEST_F(NewNonFrontendDataTypeControllerTest,
+       StartAssociationTriggersUnrecoverableError) {
+  SetStartExpectations();
+  SetStartFailExpectations(DataTypeController::UNRECOVERABLE_ERROR);
+  // Set up association to fail with an unrecoverable error.
+  EXPECT_CALL(*dtc_mock_, StartAssociationAsync());
+  EXPECT_CALL(*change_processor_, Connect(_,_,_,_)).WillOnce(Return(true));
+  EXPECT_CALL(*change_processor_, CryptoReadyIfNecessary(_)).
+      WillRepeatedly(Return(true));
+  EXPECT_CALL(*change_processor_, SyncModelHasUserCreatedNodes(_,_)).
+      WillRepeatedly(DoAll(SetArgumentPointee<1>(false), Return(false)));
+  EXPECT_EQ(DataTypeController::NOT_RUNNING, new_non_frontend_dtc_->state());
+  new_non_frontend_dtc_->Start(
+      NewCallback(&start_callback_, &StartCallback::Run));
+  WaitForDTC();
+  EXPECT_EQ(DataTypeController::NOT_RUNNING, new_non_frontend_dtc_->state());
+}
+
+TEST_F(NewNonFrontendDataTypeControllerTest, StartAssociationCryptoNotReady) {
+  SetStartExpectations();
+  SetStartFailExpectations(DataTypeController::NEEDS_CRYPTO);
+  // Set up association to fail with a NEEDS_CRYPTO error.
+  EXPECT_CALL(*dtc_mock_, StartAssociationAsync());
+  EXPECT_CALL(*change_processor_, Connect(_,_,_,_)).WillOnce(Return(true));
+  EXPECT_CALL(*change_processor_, CryptoReadyIfNecessary(_)).
+      WillRepeatedly(Return(false));
+  EXPECT_EQ(DataTypeController::NOT_RUNNING, new_non_frontend_dtc_->state());
+  new_non_frontend_dtc_->Start(
+      NewCallback(&start_callback_, &StartCallback::Run));
+  WaitForDTC();
+  EXPECT_EQ(DataTypeController::NOT_RUNNING, new_non_frontend_dtc_->state());
+}
+
+// Trigger a Stop() call when we check if the model associator has user created
+// nodes.
+TEST_F(NewNonFrontendDataTypeControllerTest, AbortDuringAssociation) {
+  WaitableEvent wait_for_db_thread_pause(false, false);
+  WaitableEvent pause_db_thread(false, false);
+
+  SetStartExpectations();
+  SetStartFailExpectations(DataTypeController::ABORTED);
+  EXPECT_CALL(*dtc_mock_, StartAssociationAsync());
+  EXPECT_CALL(*change_processor_, Connect(_,_,_,_)).WillOnce(Return(true));
+  EXPECT_CALL(*change_processor_, CryptoReadyIfNecessary(_)).
+      WillOnce(Return(true));
+  EXPECT_CALL(*change_processor_, SyncModelHasUserCreatedNodes(_,_)).
+      WillOnce(DoAll(
+          SignalEvent(&wait_for_db_thread_pause),
+          WaitOnEvent(&pause_db_thread),
+          SetArgumentPointee<1>(true),
+          Return(true)));
+  EXPECT_CALL(*change_processor_, GetSyncDataForType(_,_)).
+      WillOnce(Return(SyncError(FROM_HERE, "Disconnected.", AUTOFILL_PROFILE)));
+  EXPECT_CALL(*change_processor_, Disconnect()).
+      WillOnce(DoAll(SignalEvent(&pause_db_thread), Return(true)));
+  EXPECT_CALL(service_, DeactivateDataType(_));
+  EXPECT_CALL(*dtc_mock_, StopLocalServiceAsync());
+  EXPECT_CALL(syncable_service_, StopSyncing(_));
+  EXPECT_EQ(DataTypeController::NOT_RUNNING, new_non_frontend_dtc_->state());
+  new_non_frontend_dtc_->Start(
+      NewCallback(&start_callback_, &StartCallback::Run));
+  wait_for_db_thread_pause.Wait();
+  new_non_frontend_dtc_->Stop();
+  WaitForDTC();
+  EXPECT_EQ(DataTypeController::NOT_RUNNING, new_non_frontend_dtc_->state());
+}
+
+TEST_F(NewNonFrontendDataTypeControllerTest, Stop) {
+  SetStartExpectations();
+  SetAssociateExpectations();
+  SetActivateExpectations(DataTypeController::OK);
+  SetStopExpectations();
+  EXPECT_EQ(DataTypeController::NOT_RUNNING, new_non_frontend_dtc_->state());
+  new_non_frontend_dtc_->Start(
+      NewCallback(&start_callback_, &StartCallback::Run));
+  WaitForDTC();
+  EXPECT_EQ(DataTypeController::RUNNING, new_non_frontend_dtc_->state());
+  new_non_frontend_dtc_->Stop();
+  EXPECT_EQ(DataTypeController::NOT_RUNNING, new_non_frontend_dtc_->state());
+}
+
+TEST_F(NewNonFrontendDataTypeControllerTest, OnUnrecoverableError) {
+  SetStartExpectations();
+  SetAssociateExpectations();
+  SetActivateExpectations(DataTypeController::OK);
+  EXPECT_CALL(*dtc_mock_, RecordUnrecoverableError(_, "Test"));
+  EXPECT_CALL(service_, OnUnrecoverableError(_,_)).WillOnce(
+      InvokeWithoutArgs(new_non_frontend_dtc_.get(),
+                        &NewNonFrontendDataTypeController::Stop));
+  SetStopExpectations();
+  EXPECT_EQ(DataTypeController::NOT_RUNNING, new_non_frontend_dtc_->state());
+  new_non_frontend_dtc_->Start(
+      NewCallback(&start_callback_, &StartCallback::Run));
+  WaitForDTC();
+  EXPECT_EQ(DataTypeController::RUNNING, new_non_frontend_dtc_->state());
+  // This should cause new_non_frontend_dtc_->Stop() to be called.
+  BrowserThread::PostTask(BrowserThread::DB, FROM_HERE,
+      base::Bind(
+          &NewNonFrontendDataTypeControllerFake::OnUnrecoverableError,
+          new_non_frontend_dtc_.get(),
+          FROM_HERE,
+          std::string("Test")));
+  WaitForDTC();
+  EXPECT_EQ(DataTypeController::NOT_RUNNING, new_non_frontend_dtc_->state());
+}
+
+}  // namespace
+
+}  // namespace browser_sync
diff --git a/chrome/browser/sync/glue/non_frontend_data_type_controller.cc b/chrome/browser/sync/glue/non_frontend_data_type_controller.cc
index 75b3332..f446884 100644
--- a/chrome/browser/sync/glue/non_frontend_data_type_controller.cc
+++ b/chrome/browser/sync/glue/non_frontend_data_type_controller.cc
@@ -4,6 +4,8 @@
 
 #include "chrome/browser/sync/glue/non_frontend_data_type_controller.h"
 
+#include "base/bind.h"
+#include "base/callback.h"
 #include "base/logging.h"
 #include "chrome/browser/profiles/profile.h"
 #include "chrome/browser/sync/api/sync_error.h"
@@ -60,8 +62,7 @@ void NonFrontendDataTypeController::Start(StartCallback* start_callback) {
   state_ = MODEL_STARTING;
   if (!StartModels()) {
     // If we are waiting for some external service to load before associating
-    // or we failed to start the models, we exit early. state_ will control
-    // what we perform next.
+    // or we failed to start the models, we exit early.
     DCHECK(state_ == NOT_RUNNING || state_ == MODEL_STARTING);
     return;
   }
@@ -116,7 +117,7 @@ void NonFrontendDataTypeController::StartAssociation() {
   }
 
   profile_sync_service_->ActivateDataType(type(), model_safe_group(),
-                                          change_processor_.get());
+                                          change_processor());
   StartDone(!sync_has_nodes ? OK_FIRST_RUN : OK, RUNNING, SyncError());
 }
 
@@ -139,12 +140,11 @@ void NonFrontendDataTypeController::StartDone(
   base::AutoLock lock(abort_association_lock_);
   if (!abort_association_) {
     BrowserThread::PostTask(BrowserThread::UI, FROM_HERE,
-        NewRunnableMethod(
-            this,
-            &NonFrontendDataTypeController::StartDoneImpl,
-            result,
-            new_state,
-            error));
+        base::Bind(&NonFrontendDataTypeController::StartDoneImpl,
+                   this,
+                   result,
+                   new_state,
+                   error));
   }
 }
 
@@ -153,13 +153,14 @@ void NonFrontendDataTypeController::StartDoneImpl(
     DataTypeController::State new_state,
     const SyncError& error) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
-
+  // It's possible to have StartDoneImpl called first from the UI thread
+  // (due to Stop being called) and then posted from the non-UI thread. In
+  // this case, we drop the second call because we've already been stopped.
   if (state_ == NOT_RUNNING) {
-    // During stop it is possible startdoneimpl can be called twice. Once from
-    // the |StartDone| method and once from the |Stop| method.
     DCHECK(!start_callback_.get());
     return;
   }
+
   state_ = new_state;
   if (state_ != RUNNING) {
     // Start failed.
@@ -174,9 +175,9 @@ void NonFrontendDataTypeController::StartDoneImpl(
   callback->Run(result, error);
 }
 
-// TODO(sync): Blocking the UI thread at shutdown is bad. If we had a way of
-// distinguishing chrome shutdown from sync shutdown, we should be able to avoid
-// this (http://crbug.com/55662).
+// TODO(sync): Blocking the UI thread at shutdown is bad. The new API avoids
+// this. Once all non-frontend datatypes use the new API, we can get rid of this
+// locking (see implementation in AutofillProfileDataTypeController).
 void NonFrontendDataTypeController::Stop() {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
   // If Stop() is called while Start() is waiting for association to
@@ -253,9 +254,11 @@ void NonFrontendDataTypeController::OnUnrecoverableError(
     const std::string& message) {
   DCHECK(!BrowserThread::CurrentlyOn(BrowserThread::UI));
   RecordUnrecoverableError(from_here, message);
-  BrowserThread::PostTask(BrowserThread::UI, from_here, NewRunnableMethod(this,
-      &NonFrontendDataTypeController::OnUnrecoverableErrorImpl, from_here,
-      message));
+  BrowserThread::PostTask(BrowserThread::UI, from_here,
+      base::Bind(&NonFrontendDataTypeController::OnUnrecoverableErrorImpl,
+                 this,
+                 from_here,
+                 message));
 }
 
 void NonFrontendDataTypeController::OnUnrecoverableErrorImpl(
@@ -279,15 +282,27 @@ ProfileSyncService* NonFrontendDataTypeController::profile_sync_service()
   return profile_sync_service_;
 }
 
+void NonFrontendDataTypeController::set_start_callback(
+    StartCallback* callback) {
+  start_callback_.reset(callback);
+}
 void NonFrontendDataTypeController::set_state(State state) {
   state_ = state;
 }
 
+AssociatorInterface* NonFrontendDataTypeController::associator() const {
+  return model_associator_.get();
+}
+
 void NonFrontendDataTypeController::set_model_associator(
     AssociatorInterface* associator) {
   model_associator_.reset(associator);
 }
 
+ChangeProcessor* NonFrontendDataTypeController::change_processor() const {
+  return change_processor_.get();
+}
+
 void NonFrontendDataTypeController::set_change_processor(
     ChangeProcessor* change_processor) {
   change_processor_.reset(change_processor);
diff --git a/chrome/browser/sync/glue/non_frontend_data_type_controller.h b/chrome/browser/sync/glue/non_frontend_data_type_controller.h
index b769ac8..c48d865 100644
--- a/chrome/browser/sync/glue/non_frontend_data_type_controller.h
+++ b/chrome/browser/sync/glue/non_frontend_data_type_controller.h
@@ -133,9 +133,13 @@ class NonFrontendDataTypeController : public DataTypeController {
   ProfileSyncFactory* profile_sync_factory() const;
   Profile* profile() const;
   ProfileSyncService* profile_sync_service() const;
+  void set_start_callback(StartCallback* callback);
   void set_state(State state);
-  void set_model_associator(AssociatorInterface* associator);
-  void set_change_processor(ChangeProcessor* change_processor);
+
+  virtual AssociatorInterface* associator() const;
+  virtual void set_model_associator(AssociatorInterface* associator);
+  virtual ChangeProcessor* change_processor() const;
+  virtual void set_change_processor(ChangeProcessor* change_processor);
 
  private:
   ProfileSyncFactory* const profile_sync_factory_;
diff --git a/chrome/browser/sync/glue/preference_data_type_controller.cc b/chrome/browser/sync/glue/preference_data_type_controller.cc
index 0e52b99..c908ed3 100644
--- a/chrome/browser/sync/glue/preference_data_type_controller.cc
+++ b/chrome/browser/sync/glue/preference_data_type_controller.cc
@@ -7,7 +7,6 @@
 #include "base/metrics/histogram.h"
 #include "chrome/browser/profiles/profile.h"
 #include "chrome/browser/sync/api/syncable_service.h"
-#include "chrome/browser/sync/glue/generic_change_processor.h"
 #include "chrome/browser/sync/profile_sync_factory.h"
 
 namespace browser_sync {
@@ -33,7 +32,12 @@ void PreferenceDataTypeController::CreateSyncComponents() {
       profile_sync_factory_->CreatePreferenceSyncComponents(sync_service_,
                                                             this);
   set_model_associator(sync_components.model_associator);
-  set_change_processor(sync_components.change_processor);
+  generic_change_processor_.reset(static_cast<GenericChangeProcessor*>(
+      sync_components.change_processor));
+}
+
+GenericChangeProcessor* PreferenceDataTypeController::change_processor() const {
+  return generic_change_processor_.get();
 }
 
 void PreferenceDataTypeController::RecordUnrecoverableError(
diff --git a/chrome/browser/sync/glue/preference_data_type_controller.h b/chrome/browser/sync/glue/preference_data_type_controller.h
index 6f3d009..86f89c1 100644
--- a/chrome/browser/sync/glue/preference_data_type_controller.h
+++ b/chrome/browser/sync/glue/preference_data_type_controller.h
@@ -9,6 +9,7 @@
 #include <string>
 
 #include "base/compiler_specific.h"
+#include "chrome/browser/sync/glue/generic_change_processor.h"
 #include "chrome/browser/sync/glue/frontend_data_type_controller.h"
 
 namespace browser_sync {
@@ -24,6 +25,9 @@ class PreferenceDataTypeController : public FrontendDataTypeController {
   // FrontendDataTypeController implementation.
   virtual syncable::ModelType type() const OVERRIDE;
 
+ protected:
+  virtual GenericChangeProcessor* change_processor() const OVERRIDE;
+
  private:
   // FrontendDataTypeController implementations.
   virtual void CreateSyncComponents() OVERRIDE;
@@ -33,6 +37,8 @@ class PreferenceDataTypeController : public FrontendDataTypeController {
   virtual void RecordAssociationTime(base::TimeDelta time) OVERRIDE;
   virtual void RecordStartFailure(StartResult result) OVERRIDE;
 
+  scoped_ptr<GenericChangeProcessor> generic_change_processor_;
+
   DISALLOW_COPY_AND_ASSIGN(PreferenceDataTypeController);
 };
 
diff --git a/chrome/browser/sync/glue/search_engine_data_type_controller.cc b/chrome/browser/sync/glue/search_engine_data_type_controller.cc
index 7e434c9..9f90cd5 100644
--- a/chrome/browser/sync/glue/search_engine_data_type_controller.cc
+++ b/chrome/browser/sync/glue/search_engine_data_type_controller.cc
@@ -9,7 +9,6 @@
 #include "chrome/browser/search_engines/template_url_service.h"
 #include "chrome/browser/search_engines/template_url_service_factory.h"
 #include "chrome/browser/sync/api/syncable_service.h"
-#include "chrome/browser/sync/glue/generic_change_processor.h"
 #include "chrome/browser/sync/profile_sync_factory.h"
 #include "chrome/browser/sync/profile_sync_service.h"
 #include "chrome/common/chrome_notification_types.h"
@@ -70,7 +69,13 @@ void SearchEngineDataTypeController::CreateSyncComponents() {
       profile_sync_factory_->CreateSearchEngineSyncComponents(sync_service_,
                                                               this);
   set_model_associator(sync_components.model_associator);
-  set_change_processor(sync_components.change_processor);
+  generic_change_processor_.reset(static_cast<GenericChangeProcessor*>(
+      sync_components.change_processor));
+}
+
+GenericChangeProcessor* SearchEngineDataTypeController::change_processor()
+    const {
+  return generic_change_processor_.get();
 }
 
 void SearchEngineDataTypeController::RecordUnrecoverableError(
diff --git a/chrome/browser/sync/glue/search_engine_data_type_controller.h b/chrome/browser/sync/glue/search_engine_data_type_controller.h
index bf94854..61d9f87 100644
--- a/chrome/browser/sync/glue/search_engine_data_type_controller.h
+++ b/chrome/browser/sync/glue/search_engine_data_type_controller.h
@@ -9,6 +9,7 @@
 #include <string>
 
 #include "base/compiler_specific.h"
+#include "chrome/browser/sync/glue/generic_change_processor.h"
 #include "chrome/browser/sync/glue/frontend_data_type_controller.h"
 #include "content/common/notification_observer.h"
 #include "content/common/notification_registrar.h"
@@ -32,6 +33,9 @@ class SearchEngineDataTypeController : public FrontendDataTypeController,
                        const NotificationSource& source,
                        const NotificationDetails& details) OVERRIDE;
 
+ protected:
+  virtual GenericChangeProcessor* change_processor() const OVERRIDE;
+
  private:
   // FrontendDataTypeController implementations.
   virtual bool StartModels() OVERRIDE;
@@ -43,6 +47,8 @@ class SearchEngineDataTypeController : public FrontendDataTypeController,
   virtual void RecordAssociationTime(base::TimeDelta time) OVERRIDE;
   virtual void RecordStartFailure(StartResult result) OVERRIDE;
 
+  scoped_ptr<GenericChangeProcessor> generic_change_processor_;
+
   NotificationRegistrar registrar_;
 
   DISALLOW_COPY_AND_ASSIGN(SearchEngineDataTypeController);
diff --git a/chrome/browser/sync/glue/shared_change_processor.cc b/chrome/browser/sync/glue/shared_change_processor.cc
new file mode 100644
index 0000000..f3658b4
--- /dev/null
+++ b/chrome/browser/sync/glue/shared_change_processor.cc
@@ -0,0 +1,129 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "chrome/browser/sync/glue/shared_change_processor.h"
+
+#include "chrome/browser/sync/profile_sync_factory.h"
+#include "chrome/browser/sync/profile_sync_service.h"
+#include "chrome/browser/sync/api/sync_change.h"
+#include "chrome/browser/sync/glue/generic_change_processor.h"
+#include "content/browser/browser_thread.h"
+
+using base::AutoLock;
+
+namespace browser_sync {
+
+SharedChangeProcessor::SharedChangeProcessor()
+    : disconnected_(false) {
+  // We're always created on the UI thread.
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  DetachFromThread();
+}
+
+SharedChangeProcessor::~SharedChangeProcessor() {
+  // We can either be deleted when the DTC is destroyed (on UI thread),
+  // or when the SyncableService stop's syncing (datatype thread).
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI) ||
+         CalledOnValidThread());
+  DetachFromThread();
+}
+
+bool SharedChangeProcessor::Connect(
+    ProfileSyncFactory* sync_factory,
+    ProfileSyncService* sync_service,
+    UnrecoverableErrorHandler* error_handler,
+    const base::WeakPtr<SyncableService>& local_service) {
+  DCHECK(CalledOnValidThread());
+  AutoLock lock(monitor_lock_);
+  if (disconnected_)
+    return false;
+  if (!local_service) {
+    NOTREACHED() << "SyncableService destroyed before DTC was stopped.";
+    disconnected_ = true;
+    return false;
+  }
+  generic_change_processor_.reset(
+      sync_factory->CreateGenericChangeProcessor(sync_service,
+                                                 error_handler,
+                                                 local_service));
+  return true;
+}
+
+bool SharedChangeProcessor::Disconnect() {
+  // May be called from any thread.
+  VLOG(1) << "Disconnecting change processor.";
+  AutoLock lock(monitor_lock_);
+  bool was_connected = !disconnected_;
+  disconnected_ = true;
+  return was_connected;
+}
+
+SyncError SharedChangeProcessor::GetSyncDataForType(
+    syncable::ModelType type,
+    SyncDataList* current_sync_data) {
+  DCHECK(CalledOnValidThread());
+  AutoLock lock(monitor_lock_);
+  if (disconnected_) {
+    SyncError error(FROM_HERE, "Change processor disconnected.", type);
+    return error;
+  }
+  return generic_change_processor_->GetSyncDataForType(type, current_sync_data);
+}
+
+SyncError SharedChangeProcessor::ProcessSyncChanges(
+    const tracked_objects::Location& from_here,
+    const SyncChangeList& list_of_changes) {
+  DCHECK(CalledOnValidThread());
+  AutoLock lock(monitor_lock_);
+  if (disconnected_) {
+    // The DTC that disconnects us must ensure it posts a StopSyncing task.
+    // If we reach this, it means it just hasn't executed yet.
+    syncable::ModelType type;
+    if (list_of_changes.size() > 0) {
+      type = list_of_changes[0].sync_data().GetDataType();
+    }
+    SyncError error(FROM_HERE, "Change processor disconnected.", type);
+    return error;
+  }
+  return generic_change_processor_->ProcessSyncChanges(
+      from_here, list_of_changes);
+}
+
+bool SharedChangeProcessor::SyncModelHasUserCreatedNodes(
+    syncable::ModelType type,
+    bool* has_nodes) {
+  DCHECK(CalledOnValidThread());
+  AutoLock lock(monitor_lock_);
+  if (disconnected_) {
+    LOG(ERROR) << "Change processor disconnected.";
+    return false;
+  }
+  return generic_change_processor_->SyncModelHasUserCreatedNodes(
+      type, has_nodes);
+}
+
+bool SharedChangeProcessor::CryptoReadyIfNecessary(syncable::ModelType type) {
+  DCHECK(CalledOnValidThread());
+  AutoLock lock(monitor_lock_);
+  if (disconnected_) {
+    LOG(ERROR) << "Change processor disconnected.";
+    return true;  // Otherwise we get into infinite spin waiting.
+  }
+  return generic_change_processor_->CryptoReadyIfNecessary(type);
+}
+
+void SharedChangeProcessor::ActivateDataType(
+    ProfileSyncService* sync_service,
+    syncable::ModelType model_type,
+    browser_sync::ModelSafeGroup model_safe_group) {
+  AutoLock lock(monitor_lock_);
+  if (disconnected_) {
+    LOG(ERROR) << "Change processor disconnected.";
+    return;
+  }
+  sync_service->ActivateDataType(
+      model_type, model_safe_group, generic_change_processor_.get());
+}
+
+}  // namespace browser_sync
diff --git a/chrome/browser/sync/glue/shared_change_processor.h b/chrome/browser/sync/glue/shared_change_processor.h
new file mode 100644
index 0000000..b6d4d85
--- /dev/null
+++ b/chrome/browser/sync/glue/shared_change_processor.h
@@ -0,0 +1,113 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef CHROME_BROWSER_SYNC_GLUE_SHARED_CHANGE_PROCESSOR_H_
+#define CHROME_BROWSER_SYNC_GLUE_SHARED_CHANGE_PROCESSOR_H_
+#pragma once
+
+#include "base/location.h"
+#include "base/memory/ref_counted.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/memory/weak_ptr.h"
+#include "base/synchronization/lock.h"
+#include "base/threading/thread_checker.h"
+#include "chrome/browser/sync/api/sync_change_processor.h"
+#include "chrome/browser/sync/api/sync_error.h"
+#include "chrome/browser/sync/engine/model_safe_worker.h"
+
+class ProfileSyncFactory;
+class ProfileSyncService;
+class SyncData;
+class SyncableService;
+
+typedef std::vector<SyncData> SyncDataList;
+
+namespace browser_sync {
+
+class GenericChangeProcessor;
+class UnrecoverableErrorHandler;
+
+// A ref-counted wrapper around a GenericChangeProcessor for use with datatypes
+// that don't live on the UI thread.
+//
+// We need to make it refcounted as the ownership transfer from the
+// DataTypeController is dependent on threading, and hence racy. The
+// SharedChangeProcessor should be created on the UI thread, but should only be
+// connected and used on the same thread as the datatype it interacts with.
+//
+// The only thread-safe method is Disconnect, which will disconnect from the
+// generic change processor, letting us shut down the syncer/datatype without
+// waiting for non-UI threads.
+//
+// Note: since we control the work being done while holding the lock, we ensure
+// no I/O or other intensive work is done while blocking the UI thread (all
+// the work is in-memory sync interactions).
+//
+// We use virtual methods so that we can use mock's in testing.
+class SharedChangeProcessor
+    : public base::RefCountedThreadSafe<SharedChangeProcessor>,
+      public base::ThreadChecker {
+ public:
+  // Create an uninitialized SharedChangeProcessor (to be later connected).
+  SharedChangeProcessor();
+
+  // Connect to the Syncer. Will create and hold a new GenericChangeProcessor.
+  // Returns: true if successful, false if disconnected or |local_service| was
+  // NULL.
+  virtual bool Connect(
+    ProfileSyncFactory* sync_factory,
+    ProfileSyncService* sync_service,
+    UnrecoverableErrorHandler* error_handler,
+    const base::WeakPtr<SyncableService>& local_service);
+
+  // Disconnects from the generic change processor. May be called from any
+  // thread. After this, all attempts to interact with the change processor by
+  // |local_service_| are dropped and return errors. The syncer will be safe to
+  // shut down from the point of view of this datatype.
+  // Note: Once disconnected, you cannot reconnect without creating a new
+  // SharedChangeProcessor.
+  // Returns: true if we were previously succesfully connected, false if we were
+  // already disconnected.
+  virtual bool Disconnect();
+
+  // GenericChangeProcessor stubs (with disconnect support).
+  // Should only be called on the same thread the datatype resides.
+  virtual SyncError ProcessSyncChanges(
+      const tracked_objects::Location& from_here,
+      const SyncChangeList& change_list);
+  virtual SyncError GetSyncDataForType(syncable::ModelType type,
+                                       SyncDataList* current_sync_data);
+  virtual bool SyncModelHasUserCreatedNodes(syncable::ModelType type,
+                                            bool* has_nodes);
+  virtual bool CryptoReadyIfNecessary(syncable::ModelType type);
+
+  // Register |generic_change_processor_| as the change processor for
+  // |model_type| with the |sync_service|.
+  // Does nothing if |disconnected_| is true.
+  virtual void ActivateDataType(
+      ProfileSyncService* sync_service,
+      syncable::ModelType model_type,
+      browser_sync::ModelSafeGroup model_safe_group);
+
+ protected:
+  friend class base::RefCountedThreadSafe<SharedChangeProcessor>;
+  virtual ~SharedChangeProcessor();
+
+ private:
+  // Monitor lock for this object. All methods that interact with the change
+  // processor must aquire this lock and check whether we're disconnected or
+  // not. Once disconnected, all attempted changes to or loads from the change
+  // processor return errors. This enables us to shut down the syncer without
+  // having to wait for possibly non-UI thread datatypes to complete work.
+  mutable base::Lock monitor_lock_;
+  bool disconnected_;
+
+  scoped_ptr<GenericChangeProcessor> generic_change_processor_;
+
+  DISALLOW_COPY_AND_ASSIGN(SharedChangeProcessor);
+};
+
+}  // namespace browser_sync
+
+#endif  // CHROME_BROWSER_SYNC_GLUE_SHARED_CHANGE_PROCESSOR_H_
diff --git a/chrome/browser/sync/glue/shared_change_processor_mock.cc b/chrome/browser/sync/glue/shared_change_processor_mock.cc
new file mode 100644
index 0000000..fde6c46
--- /dev/null
+++ b/chrome/browser/sync/glue/shared_change_processor_mock.cc
@@ -0,0 +1,15 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "chrome/browser/sync/glue/shared_change_processor_mock.h"
+
+#include "base/compiler_specific.h"
+
+namespace browser_sync {
+
+SharedChangeProcessorMock::SharedChangeProcessorMock() {}
+
+SharedChangeProcessorMock::~SharedChangeProcessorMock() {}
+
+}  // namespace browser_sync.
diff --git a/chrome/browser/sync/glue/shared_change_processor_mock.h b/chrome/browser/sync/glue/shared_change_processor_mock.h
new file mode 100644
index 0000000..8e74f88
--- /dev/null
+++ b/chrome/browser/sync/glue/shared_change_processor_mock.h
@@ -0,0 +1,48 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef CHROME_BROWSER_SYNC_GLUE_SHARED_CHANGE_PROCESSOR_MOCK_H_
+#define CHROME_BROWSER_SYNC_GLUE_SHARED_CHANGE_PROCESSOR_MOCK_H_
+#pragma once
+
+#include "chrome/browser/sync/api/sync_change.h"
+#include "chrome/browser/sync/glue/shared_change_processor.h"
+#include "chrome/browser/sync/unrecoverable_error_handler.h"
+#include "testing/gmock/include/gmock/gmock.h"
+
+namespace browser_sync {
+
+class SharedChangeProcessorMock : public SharedChangeProcessor {
+ public:
+  SharedChangeProcessorMock();
+
+  MOCK_METHOD4(Connect, bool(
+      ProfileSyncFactory* sync_factory,
+      ProfileSyncService* sync_service,
+      UnrecoverableErrorHandler* error_handler,
+      const base::WeakPtr<SyncableService>& local_service));
+  MOCK_METHOD0(Disconnect, bool());
+  MOCK_METHOD2(ProcessSyncChanges,
+               SyncError(const tracked_objects::Location& from_here,
+                         const SyncChangeList& change_list));
+  MOCK_METHOD2(GetSyncDataForType,
+               SyncError(syncable::ModelType type,
+                         SyncDataList* current_sync_data));
+  MOCK_METHOD2(SyncModelHasUserCreatedNodes,
+               bool(syncable::ModelType type,
+                    bool* has_nodes));
+  MOCK_METHOD1(CryptoReadyIfNecessary, bool(syncable::ModelType type));
+
+ protected:
+  virtual ~SharedChangeProcessorMock();
+  MOCK_METHOD2(OnUnrecoverableError, void(const tracked_objects::Location&,
+                                          const std::string&));
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(SharedChangeProcessorMock);
+};
+
+}  // namespace browser_sync
+
+#endif  // CHROME_BROWSER_SYNC_GLUE_SHARED_CHANGE_PROCESSOR_MOCK_H_
diff --git a/chrome/browser/sync/glue/shared_change_processor_ref.cc b/chrome/browser/sync/glue/shared_change_processor_ref.cc
new file mode 100644
index 0000000..e0b2a52
--- /dev/null
+++ b/chrome/browser/sync/glue/shared_change_processor_ref.cc
@@ -0,0 +1,23 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "chrome/browser/sync/glue/shared_change_processor_ref.h"
+
+namespace browser_sync {
+
+SharedChangeProcessorRef::SharedChangeProcessorRef(
+    const scoped_refptr<SharedChangeProcessor>& change_processor)
+    : change_processor_(change_processor) {
+  DCHECK(change_processor_.get());
+}
+
+SharedChangeProcessorRef::~SharedChangeProcessorRef() {}
+
+SyncError SharedChangeProcessorRef::ProcessSyncChanges(
+    const tracked_objects::Location& from_here,
+    const SyncChangeList& change_list) {
+  return change_processor_->ProcessSyncChanges(from_here, change_list);
+}
+
+}  // namespace browser_sync
diff --git a/chrome/browser/sync/glue/shared_change_processor_ref.h b/chrome/browser/sync/glue/shared_change_processor_ref.h
new file mode 100644
index 0000000..4c73bac
--- /dev/null
+++ b/chrome/browser/sync/glue/shared_change_processor_ref.h
@@ -0,0 +1,38 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef CHROME_BROWSER_SYNC_GLUE_SHARED_CHANGE_PROCESSOR_REF_H_
+#define CHROME_BROWSER_SYNC_GLUE_SHARED_CHANGE_PROCESSOR_REF_H_
+#pragma once
+
+#include "base/compiler_specific.h"
+#include "base/memory/ref_counted.h"
+#include "chrome/browser/sync/api/sync_change_processor.h"
+#include "chrome/browser/sync/glue/shared_change_processor.h"
+
+namespace browser_sync {
+
+// A SyncChangeProcessor stub for interacting with a refcounted
+// SharedChangeProcessor.
+class SharedChangeProcessorRef : public SyncChangeProcessor {
+ public:
+  SharedChangeProcessorRef(
+      const scoped_refptr<browser_sync::SharedChangeProcessor>&
+          change_processor);
+  virtual ~SharedChangeProcessorRef();
+
+  // SyncChangeProcessor implementation.
+  virtual SyncError ProcessSyncChanges(
+      const tracked_objects::Location& from_here,
+      const SyncChangeList& change_list) OVERRIDE;
+
+  // Default copy and assign welcome (and safe due to refcounted-ness).
+
+ private:
+  scoped_refptr<browser_sync::SharedChangeProcessor> change_processor_;
+};
+
+}  // namespace browser_sync
+
+#endif  // CHROME_BROWSER_SYNC_GLUE_SHARED_CHANGE_PROCESSOR_REF_H_
diff --git a/chrome/browser/sync/glue/syncable_service_adapter.cc b/chrome/browser/sync/glue/syncable_service_adapter.cc
index 88b6ab1..3e8ef10 100644
--- a/chrome/browser/sync/glue/syncable_service_adapter.cc
+++ b/chrome/browser/sync/glue/syncable_service_adapter.cc
@@ -39,6 +39,10 @@ bool SyncableServiceAdapter::AssociateModels(SyncError* error) {
     *error = temp_error;
     return false;
   }
+
+  // TODO(zea): Have all datatypes take ownership of the sync_processor_.
+  // Further, refactor the DTC's to not need this class at all
+  // (crbug.com/100114).
   temp_error = service_->MergeDataAndStartSyncing(type_,
                                                   initial_sync_data,
                                                   sync_processor_);
@@ -60,8 +64,7 @@ bool SyncableServiceAdapter::SyncModelHasUserCreatedNodes(bool* has_nodes) {
 }
 
 void SyncableServiceAdapter::AbortAssociation() {
-  service_->StopSyncing(type_);
-  syncing_ = false;
+  NOTIMPLEMENTED();
 }
 
 bool SyncableServiceAdapter::CryptoReadyIfNecessary() {
diff --git a/chrome/browser/sync/profile_sync_factory.h b/chrome/browser/sync/profile_sync_factory.h
index 5b652b0..b65fe21 100644
--- a/chrome/browser/sync/profile_sync_factory.h
+++ b/chrome/browser/sync/profile_sync_factory.h
@@ -19,11 +19,14 @@ class ExtensionSettingsBackend;
 class PasswordStore;
 class PersonalDataManager;
 class ProfileSyncService;
+class SyncableService;
 class WebDatabase;
 class WebDataService;
 
 namespace browser_sync {
 class DataTypeManager;
+class GenericChangeProcessor;
+class SharedChangeProcessor;
 class SyncBackendHost;
 class UnrecoverableErrorHandler;
 }
@@ -39,6 +42,15 @@ class ProfileSyncFactory {
   // and change processors all return this struct.  This is needed
   // because the change processors typically require a type-specific
   // model associator at construction time.
+  //
+  // Note: This interface is deprecated in favor of the SyncableService API.
+  // New datatypes that do not live on the UI thread should directly return a
+  // weak pointer to a SyncableService. All others continue to return
+  // SyncComponents. It is safe to assume that the factory methods below are
+  // called on the same thread in which the datatype resides.
+  //
+  // TODO(zea): Have all datatypes using the new API switch to returning
+  // SyncableService weak pointers instead of SyncComponents (crbug.com/100114).
   struct SyncComponents {
     browser_sync::AssociatorInterface* model_associator;
     browser_sync::ChangeProcessor* change_processor;
@@ -65,6 +77,15 @@ class ProfileSyncFactory {
       browser_sync::SyncBackendHost* backend,
       const browser_sync::DataTypeController::TypeMap* controllers) = 0;
 
+  // Creating this in the factory helps us mock it out in testing.
+  virtual browser_sync::GenericChangeProcessor* CreateGenericChangeProcessor(
+      ProfileSyncService* profile_sync_service,
+      browser_sync::UnrecoverableErrorHandler* error_handler,
+      const base::WeakPtr<SyncableService>& local_service) = 0;
+
+  virtual browser_sync::SharedChangeProcessor*
+      CreateSharedChangeProcessor() = 0;
+
   // Instantiates both a model associator and change processor for the
   // app data type.  The pointers in the return struct are
   // owned by the caller.
@@ -80,13 +101,11 @@ class ProfileSyncFactory {
       WebDatabase* web_database,
       browser_sync::UnrecoverableErrorHandler* error_handler) = 0;
 
-  // Instantiates both a model associator and change processor for the
-  // autofill data type.  The pointers in the return struct are owned
-  // by the caller.
-  virtual SyncComponents CreateAutofillProfileSyncComponents(
-      ProfileSyncService* profile_sync_service,
-      WebDataService* web_data_service,
-      browser_sync::UnrecoverableErrorHandler* error_handler) = 0;
+  // Returns a weak pointer to the SyncableService associated with the datatype.
+  // The SyncableService is not owned by Sync, but by the backend service
+  // itself.
+  virtual base::WeakPtr<SyncableService> GetAutofillProfileSyncableService(
+      WebDataService* web_data_service) const = 0;
 
   // Instantiates both a model associator and change processor for the
   // bookmark data type.  The pointers in the return struct are owned
diff --git a/chrome/browser/sync/profile_sync_factory_impl.cc b/chrome/browser/sync/profile_sync_factory_impl.cc
index f38c9ba..ede961c 100644
--- a/chrome/browser/sync/profile_sync_factory_impl.cc
+++ b/chrome/browser/sync/profile_sync_factory_impl.cc
@@ -5,6 +5,7 @@
 #include "base/command_line.h"
 #include "chrome/browser/extensions/extension_service.h"
 #include "chrome/browser/extensions/extension_settings_backend.h"
+#include "chrome/browser/prefs/pref_model_associator.h"
 #include "chrome/browser/profiles/profile.h"
 #include "chrome/browser/search_engines/template_url_service.h"
 #include "chrome/browser/search_engines/template_url_service_factory.h"
@@ -29,6 +30,7 @@
 #include "chrome/browser/sync/glue/session_change_processor.h"
 #include "chrome/browser/sync/glue/session_data_type_controller.h"
 #include "chrome/browser/sync/glue/session_model_associator.h"
+#include "chrome/browser/sync/glue/shared_change_processor.h"
 #include "chrome/browser/sync/glue/syncable_service_adapter.h"
 #include "chrome/browser/sync/glue/sync_backend_host.h"
 #include "chrome/browser/sync/glue/theme_change_processor.h"
@@ -68,6 +70,7 @@ using browser_sync::SearchEngineDataTypeController;
 using browser_sync::SessionChangeProcessor;
 using browser_sync::SessionDataTypeController;
 using browser_sync::SessionModelAssociator;
+using browser_sync::SharedChangeProcessor;
 using browser_sync::SyncableServiceAdapter;
 using browser_sync::SyncBackendHost;
 using browser_sync::ThemeChangeProcessor;
@@ -182,15 +185,31 @@ DataTypeManager* ProfileSyncFactoryImpl::CreateDataTypeManager(
   return new DataTypeManagerImpl(backend, controllers);
 }
 
+browser_sync::GenericChangeProcessor*
+    ProfileSyncFactoryImpl::CreateGenericChangeProcessor(
+        ProfileSyncService* profile_sync_service,
+        browser_sync::UnrecoverableErrorHandler* error_handler,
+        const base::WeakPtr<SyncableService>& local_service) {
+  sync_api::UserShare* user_share = profile_sync_service->GetUserShare();
+  return new GenericChangeProcessor(error_handler,
+                                    local_service,
+                                    user_share);
+}
+
+browser_sync::SharedChangeProcessor* ProfileSyncFactoryImpl::
+    CreateSharedChangeProcessor() {
+  return new SharedChangeProcessor();
+}
+
 ProfileSyncFactory::SyncComponents
 ProfileSyncFactoryImpl::CreateAppSyncComponents(
     ProfileSyncService* profile_sync_service,
     UnrecoverableErrorHandler* error_handler) {
-  SyncableService* app_sync_service =
-      profile_sync_service->profile()->GetExtensionService();
+  base::WeakPtr<SyncableService> app_sync_service =
+      profile_sync_service->profile()->GetExtensionService()->AsWeakPtr();
   sync_api::UserShare* user_share = profile_sync_service->GetUserShare();
   GenericChangeProcessor* change_processor =
-      new GenericChangeProcessor(app_sync_service, error_handler, user_share);
+      new GenericChangeProcessor(error_handler, app_sync_service, user_share);
   browser_sync::SyncableServiceAdapter* sync_service_adapter =
       new browser_sync::SyncableServiceAdapter(syncable::APPS,
                                                app_sync_service,
@@ -216,21 +235,10 @@ ProfileSyncFactoryImpl::CreateAutofillSyncComponents(
   return SyncComponents(model_associator, change_processor);
 }
 
-ProfileSyncFactory::SyncComponents
-ProfileSyncFactoryImpl::CreateAutofillProfileSyncComponents(
-    ProfileSyncService* profile_sync_service,
-    WebDataService* web_data_service,
-    browser_sync::UnrecoverableErrorHandler* error_handler) {
-  AutofillProfileSyncableService* sync_service =
-      web_data_service->GetAutofillProfileSyncableService();
-  sync_api::UserShare* user_share = profile_sync_service->GetUserShare();
-  GenericChangeProcessor* change_processor =
-      new GenericChangeProcessor(sync_service, error_handler, user_share);
-  browser_sync::SyncableServiceAdapter* sync_service_adapter =
-      new browser_sync::SyncableServiceAdapter(syncable::AUTOFILL_PROFILE,
-                                               sync_service,
-                                               change_processor);
-  return SyncComponents(sync_service_adapter, change_processor);
+base::WeakPtr<SyncableService>
+ProfileSyncFactoryImpl::GetAutofillProfileSyncableService(
+    WebDataService* web_data_service) const {
+  return web_data_service->GetAutofillProfileSyncableService()->AsWeakPtr();
 }
 
 ProfileSyncFactory::SyncComponents
@@ -258,10 +266,9 @@ ProfileSyncFactoryImpl::CreateExtensionSettingSyncComponents(
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::FILE));
   sync_api::UserShare* user_share = profile_sync_service->GetUserShare();
   GenericChangeProcessor* change_processor =
-      new GenericChangeProcessor(
-          extension_settings_backend,
-          error_handler,
-          user_share);
+      new GenericChangeProcessor(error_handler,
+                                 extension_settings_backend->AsWeakPtr(),
+                                 user_share);
   browser_sync::SyncableServiceAdapter* sync_service_adapter =
       new browser_sync::SyncableServiceAdapter(syncable::EXTENSION_SETTINGS,
                                                extension_settings_backend,
@@ -273,12 +280,13 @@ ProfileSyncFactory::SyncComponents
 ProfileSyncFactoryImpl::CreateExtensionSyncComponents(
     ProfileSyncService* profile_sync_service,
     UnrecoverableErrorHandler* error_handler) {
-  SyncableService* extension_sync_service =
-      profile_sync_service->profile()->GetExtensionService();
+  base::WeakPtr<SyncableService> extension_sync_service =
+      profile_sync_service->profile()->GetExtensionService()->AsWeakPtr();
   sync_api::UserShare* user_share = profile_sync_service->GetUserShare();
   GenericChangeProcessor* change_processor =
-      new GenericChangeProcessor(extension_sync_service, error_handler,
-          user_share);
+      new GenericChangeProcessor(error_handler,
+                                 extension_sync_service,
+                                 user_share);
   browser_sync::SyncableServiceAdapter* sync_service_adapter =
       new browser_sync::SyncableServiceAdapter(syncable::EXTENSIONS,
                                                extension_sync_service,
@@ -305,11 +313,13 @@ ProfileSyncFactory::SyncComponents
 ProfileSyncFactoryImpl::CreatePreferenceSyncComponents(
     ProfileSyncService* profile_sync_service,
     UnrecoverableErrorHandler* error_handler) {
-  SyncableService* pref_sync_service =
-      profile_->GetPrefs()->GetSyncableService();
+  base::WeakPtr<SyncableService> pref_sync_service =
+      profile_->GetPrefs()->GetSyncableService()->AsWeakPtr();
   sync_api::UserShare* user_share = profile_sync_service->GetUserShare();
   GenericChangeProcessor* change_processor =
-      new GenericChangeProcessor(pref_sync_service, error_handler, user_share);
+      new GenericChangeProcessor(error_handler,
+                                 pref_sync_service,
+                                 user_share);
   SyncableServiceAdapter* sync_service_adapter =
       new SyncableServiceAdapter(syncable::PREFERENCES,
                                  pref_sync_service,
@@ -359,12 +369,14 @@ ProfileSyncFactory::SyncComponents
 ProfileSyncFactoryImpl::CreateSearchEngineSyncComponents(
     ProfileSyncService* profile_sync_service,
     UnrecoverableErrorHandler* error_handler) {
-  SyncableService* se_sync_service =
-      TemplateURLServiceFactory::GetForProfile(profile_);
+  base::WeakPtr<SyncableService> se_sync_service =
+      TemplateURLServiceFactory::GetForProfile(profile_)->AsWeakPtr();
   DCHECK(se_sync_service);
   sync_api::UserShare* user_share = profile_sync_service->GetUserShare();
   GenericChangeProcessor* change_processor =
-      new GenericChangeProcessor(se_sync_service, error_handler, user_share);
+      new GenericChangeProcessor(error_handler,
+                                 se_sync_service,
+                                 user_share);
   SyncableServiceAdapter* sync_service_adapter =
       new SyncableServiceAdapter(syncable::SEARCH_ENGINES,
                                  se_sync_service,
diff --git a/chrome/browser/sync/profile_sync_factory_impl.h b/chrome/browser/sync/profile_sync_factory_impl.h
index 5d8873b..27063a5 100644
--- a/chrome/browser/sync/profile_sync_factory_impl.h
+++ b/chrome/browser/sync/profile_sync_factory_impl.h
@@ -30,6 +30,13 @@ class ProfileSyncFactoryImpl : public ProfileSyncFactory {
       browser_sync::SyncBackendHost* backend,
       const browser_sync::DataTypeController::TypeMap* controllers);
 
+  virtual browser_sync::GenericChangeProcessor* CreateGenericChangeProcessor(
+      ProfileSyncService* profile_sync_service,
+      browser_sync::UnrecoverableErrorHandler* error_handler,
+      const base::WeakPtr<SyncableService>& local_service);
+
+  virtual browser_sync::SharedChangeProcessor* CreateSharedChangeProcessor();
+
   virtual SyncComponents CreateAppSyncComponents(
       ProfileSyncService* profile_sync_service,
       browser_sync::UnrecoverableErrorHandler* error_handler);
@@ -39,10 +46,8 @@ class ProfileSyncFactoryImpl : public ProfileSyncFactory {
       WebDatabase* web_database,
       browser_sync::UnrecoverableErrorHandler* error_handler);
 
-  virtual SyncComponents CreateAutofillProfileSyncComponents(
-      ProfileSyncService* profile_sync_service,
-      WebDataService* web_data_service,
-      browser_sync::UnrecoverableErrorHandler* error_handler);
+  virtual base::WeakPtr<SyncableService> GetAutofillProfileSyncableService(
+      WebDataService* web_data_service) const;
 
   virtual SyncComponents CreateBookmarkSyncComponents(
       ProfileSyncService* profile_sync_service,
diff --git a/chrome/browser/sync/profile_sync_factory_mock.h b/chrome/browser/sync/profile_sync_factory_mock.h
index 5bb5445..efbb7b9 100644
--- a/chrome/browser/sync/profile_sync_factory_mock.h
+++ b/chrome/browser/sync/profile_sync_factory_mock.h
@@ -33,6 +33,13 @@ class ProfileSyncFactoryMock : public ProfileSyncFactory {
                browser_sync::DataTypeManager*(
                    browser_sync::SyncBackendHost*,
                    const browser_sync::DataTypeController::TypeMap*));
+  MOCK_METHOD3(CreateGenericChangeProcessor,
+      browser_sync::GenericChangeProcessor*(
+          ProfileSyncService* profile_sync_service,
+          browser_sync::UnrecoverableErrorHandler* error_handler,
+          const base::WeakPtr<SyncableService>& local_service));
+  MOCK_METHOD0(CreateSharedChangeProcessor,
+      browser_sync::SharedChangeProcessor*());
   MOCK_METHOD2(CreateAppSyncComponents,
       SyncComponents(ProfileSyncService* profile_sync_service,
                      browser_sync::UnrecoverableErrorHandler* error_handler));
@@ -41,11 +48,9 @@ class ProfileSyncFactoryMock : public ProfileSyncFactory {
                    ProfileSyncService* profile_sync_service,
                    WebDatabase* web_database,
                    browser_sync::UnrecoverableErrorHandler* error_handler));
-  MOCK_METHOD3(CreateAutofillProfileSyncComponents,
-               SyncComponents(
-                   ProfileSyncService* profile_sync_service,
-                   WebDataService* web_data_service,
-                   browser_sync::UnrecoverableErrorHandler* error_handler));
+  MOCK_CONST_METHOD1(GetAutofillProfileSyncableService,
+                     base::WeakPtr<SyncableService>(
+                         WebDataService* web_data_service));
   MOCK_METHOD2(CreateBookmarkSyncComponents,
       SyncComponents(ProfileSyncService* profile_sync_service,
                      browser_sync::UnrecoverableErrorHandler* error_handler));
diff --git a/chrome/browser/sync/profile_sync_service.h b/chrome/browser/sync/profile_sync_service.h
index 85a69c8..97dbef8 100644
--- a/chrome/browser/sync/profile_sync_service.h
+++ b/chrome/browser/sync/profile_sync_service.h
@@ -365,7 +365,7 @@ class ProfileSyncService : public browser_sync::SyncFrontend,
   // tests.  Figure out how to pass the handle to the ModelAssociators
   // directly, figure out how to expose this to tests, and remove this
   // function.
-  sync_api::UserShare* GetUserShare() const;
+  virtual sync_api::UserShare* GetUserShare() const;
 
   // TODO(akalin): These two functions are used only by
   // ProfileSyncServiceHarness.  Figure out a different way to expose
diff --git a/chrome/browser/sync/profile_sync_service_autofill_unittest.cc b/chrome/browser/sync/profile_sync_service_autofill_unittest.cc
index 8f1a4a3..7274ede 100644
--- a/chrome/browser/sync/profile_sync_service_autofill_unittest.cc
+++ b/chrome/browser/sync/profile_sync_service_autofill_unittest.cc
@@ -30,6 +30,7 @@
 #include "chrome/browser/sync/glue/autofill_profile_data_type_controller.h"
 #include "chrome/browser/sync/glue/data_type_controller.h"
 #include "chrome/browser/sync/glue/generic_change_processor.h"
+#include "chrome/browser/sync/glue/shared_change_processor.h"
 #include "chrome/browser/sync/glue/syncable_service_adapter.h"
 #include "chrome/browser/sync/internal_api/read_node.h"
 #include "chrome/browser/sync/internal_api/read_transaction.h"
@@ -63,6 +64,7 @@ using browser_sync::AutofillModelAssociator;
 using browser_sync::AutofillProfileDataTypeController;
 using browser_sync::DataTypeController;
 using browser_sync::GenericChangeProcessor;
+using browser_sync::SharedChangeProcessor;
 using browser_sync::SyncableServiceAdapter;
 using browser_sync::GROUP_DB;
 using browser_sync::kAutofillTag;
@@ -226,26 +228,25 @@ ACTION_P3(MakeAutofillSyncComponents, service, wd, dtc) {
                                             change_processor);
 }
 
-ACTION_P3(MakeAutofillProfileSyncComponents, service, wds, dtc) {
+ACTION(MakeGenericChangeProcessor) {
+  sync_api::UserShare* user_share = arg0->GetUserShare();
+  return new GenericChangeProcessor(arg1, arg2, user_share);
+}
+
+ACTION(MakeSharedChangeProcessor) {
+  return new SharedChangeProcessor();
+}
+
+ACTION_P(MakeAutofillProfileSyncComponents, wds) {
   EXPECT_TRUE(BrowserThread::CurrentlyOn(BrowserThread::DB));
   if (!BrowserThread::CurrentlyOn(BrowserThread::DB))
-    return ProfileSyncFactory::SyncComponents(NULL, NULL);
-  AutofillProfileSyncableService* sync_service =
-      wds->GetAutofillProfileSyncableService();
-  sync_api::UserShare* user_share = service->GetUserShare();
-  GenericChangeProcessor* change_processor =
-      new GenericChangeProcessor(sync_service, dtc, user_share);
-  SyncableServiceAdapter* sync_service_adapter =
-      new SyncableServiceAdapter(syncable::AUTOFILL_PROFILE,
-                                 sync_service,
-                                 change_processor);
-  return ProfileSyncFactory::SyncComponents(sync_service_adapter,
-                                            change_processor);
+    return base::WeakPtr<SyncableService>();;
+  return wds->GetAutofillProfileSyncableService()->AsWeakPtr();
 }
 
 class AbstractAutofillFactory {
  public:
-  virtual AutofillDataTypeController* CreateDataTypeController(
+  virtual DataTypeController* CreateDataTypeController(
       ProfileSyncFactory* factory,
       ProfileMock* profile,
       ProfileSyncService* service) = 0;
@@ -258,7 +259,7 @@ class AbstractAutofillFactory {
 
 class AutofillEntryFactory : public AbstractAutofillFactory {
  public:
-  browser_sync::AutofillDataTypeController* CreateDataTypeController(
+  browser_sync::DataTypeController* CreateDataTypeController(
       ProfileSyncFactory* factory,
       ProfileMock* profile,
       ProfileSyncService* service) {
@@ -276,7 +277,7 @@ class AutofillEntryFactory : public AbstractAutofillFactory {
 
 class AutofillProfileFactory : public AbstractAutofillFactory {
  public:
-  browser_sync::AutofillDataTypeController* CreateDataTypeController(
+  browser_sync::DataTypeController* CreateDataTypeController(
       ProfileSyncFactory* factory,
       ProfileMock* profile,
       ProfileSyncService* service) {
@@ -287,8 +288,12 @@ class AutofillProfileFactory : public AbstractAutofillFactory {
                       ProfileSyncService* service,
                       WebDataService* wds,
                       DataTypeController* dtc) {
-    EXPECT_CALL(*factory, CreateAutofillProfileSyncComponents(_,_,_)).
-        WillOnce(MakeAutofillProfileSyncComponents(service, wds, dtc));
+    EXPECT_CALL(*factory, CreateGenericChangeProcessor(_,_,_)).
+        WillOnce(MakeGenericChangeProcessor());
+    EXPECT_CALL(*factory, CreateSharedChangeProcessor()).
+        WillOnce(MakeSharedChangeProcessor());
+    EXPECT_CALL(*factory, GetAutofillProfileSyncableService(_)).
+        WillOnce(MakeAutofillProfileSyncComponents(wds));
   }
 };
 
@@ -367,7 +372,7 @@ class ProfileSyncServiceAutofillTest : public AbstractProfileSyncServiceTest {
                                    task));
     EXPECT_CALL(profile_, GetProfileSyncService()).WillRepeatedly(
         Return(service_.get()));
-    AutofillDataTypeController* data_type_controller =
+    DataTypeController* data_type_controller =
         factory->CreateDataTypeController(&factory_,
             &profile_,
             service_.get());
diff --git a/chrome/browser/sync/profile_sync_service_mock.h b/chrome/browser/sync/profile_sync_service_mock.h
index e7cec33..4afd2f7 100644
--- a/chrome/browser/sync/profile_sync_service_mock.h
+++ b/chrome/browser/sync/profile_sync_service_mock.h
@@ -38,6 +38,7 @@ class ProfileSyncServiceMock : public ProfileSyncService {
   MOCK_METHOD2(OnUnrecoverableError,
                void(const tracked_objects::Location& location,
                const std::string& message));
+  MOCK_CONST_METHOD0(GetUserShare, sync_api::UserShare*());
   MOCK_METHOD3(ActivateDataType,
                void(syncable::ModelType, browser_sync::ModelSafeGroup,
                     browser_sync::ChangeProcessor*));
diff --git a/chrome/browser/sync/profile_sync_service_preference_unittest.cc b/chrome/browser/sync/profile_sync_service_preference_unittest.cc
index 1c5b08e..4299d63 100644
--- a/chrome/browser/sync/profile_sync_service_preference_unittest.cc
+++ b/chrome/browser/sync/profile_sync_service_preference_unittest.cc
@@ -50,9 +50,10 @@ typedef std::map<const std::string, const Value*> PreferenceValues;
 ACTION_P4(BuildPrefSyncComponents, profile_sync_service, pref_sync_service,
     model_associator_ptr, change_processor_ptr) {
   sync_api::UserShare* user_share = profile_sync_service->GetUserShare();
-  *change_processor_ptr = new GenericChangeProcessor(pref_sync_service,
-                                 profile_sync_service,
-                                 user_share);
+  *change_processor_ptr = new GenericChangeProcessor(
+      profile_sync_service,
+      pref_sync_service->AsWeakPtr(),
+      user_share);
   *model_associator_ptr = new browser_sync::SyncableServiceAdapter(
       syncable::PREFERENCES,
       pref_sync_service,
diff --git a/chrome/browser/webdata/autofill_profile_syncable_service.cc b/chrome/browser/webdata/autofill_profile_syncable_service.cc
index 75db0fd..7eb68fb 100644
--- a/chrome/browser/webdata/autofill_profile_syncable_service.cc
+++ b/chrome/browser/webdata/autofill_profile_syncable_service.cc
@@ -38,8 +38,7 @@ const char kAutofillProfileTag[] = "google_chrome_autofill_profiles";
 
 AutofillProfileSyncableService::AutofillProfileSyncableService(
     WebDataService* web_data_service)
-    : web_data_service_(web_data_service),
-      sync_processor_(NULL) {
+    : web_data_service_(web_data_service) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::DB));
   DCHECK(web_data_service_);
   notification_registrar_.Add(this,
@@ -52,8 +51,7 @@ AutofillProfileSyncableService::~AutofillProfileSyncableService() {
 }
 
 AutofillProfileSyncableService::AutofillProfileSyncableService()
-    : web_data_service_(NULL),
-      sync_processor_(NULL) {
+    : web_data_service_(NULL) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::DB));
 }
 
@@ -62,7 +60,7 @@ SyncError AutofillProfileSyncableService::MergeDataAndStartSyncing(
     const SyncDataList& initial_sync_data,
     SyncChangeProcessor* sync_processor) {
   DCHECK(CalledOnValidThread());
-  DCHECK(sync_processor_ == NULL);
+  DCHECK(!sync_processor_.get());
   VLOG(1) << "Associating Autofill: MergeDataAndStartSyncing";
 
   if (!LoadAutofillData(&profiles_.get())) {
@@ -85,7 +83,7 @@ SyncError AutofillProfileSyncableService::MergeDataAndStartSyncing(
     }
   }
 
-  sync_processor_ = sync_processor;
+  sync_processor_.reset(sync_processor);
 
   GUIDToProfileMap remaining_profiles;
   CreateGUIDToProfileMap(profiles_.get(), &remaining_profiles);
@@ -112,7 +110,9 @@ SyncError AutofillProfileSyncableService::MergeDataAndStartSyncing(
     profiles_map_[i->first] = i->second;
   }
 
-  SyncError error = sync_processor_->ProcessSyncChanges(FROM_HERE, new_changes);
+  SyncError error;
+  if (!new_changes.empty())
+    error = sync_processor_->ProcessSyncChanges(FROM_HERE, new_changes);
 
   WebDataService::NotifyOfMultipleAutofillChanges(web_data_service_);
 
@@ -121,10 +121,9 @@ SyncError AutofillProfileSyncableService::MergeDataAndStartSyncing(
 
 void AutofillProfileSyncableService::StopSyncing(syncable::ModelType type) {
   DCHECK(CalledOnValidThread());
-  DCHECK(sync_processor_ != NULL);
   DCHECK_EQ(type, syncable::AUTOFILL_PROFILE);
 
-  sync_processor_ = NULL;
+  sync_processor_.reset();
   profiles_.reset();
   profiles_map_.clear();
 }
@@ -132,7 +131,7 @@ void AutofillProfileSyncableService::StopSyncing(syncable::ModelType type) {
 SyncDataList AutofillProfileSyncableService::GetAllSyncData(
     syncable::ModelType type) const {
   DCHECK(CalledOnValidThread());
-  DCHECK(sync_processor_ != NULL);
+  DCHECK(sync_processor_.get());
   DCHECK_EQ(type, syncable::AUTOFILL_PROFILE);
 
   SyncDataList current_data;
@@ -148,8 +147,7 @@ SyncError AutofillProfileSyncableService::ProcessSyncChanges(
     const tracked_objects::Location& from_here,
     const SyncChangeList& change_list) {
   DCHECK(CalledOnValidThread());
-  DCHECK(sync_processor_ != NULL);
-  if (sync_processor_ == NULL) {
+  if (!sync_processor_.get()) {
     SyncError error(FROM_HERE, "Models not yet associated.",
                     syncable::AUTOFILL_PROFILE);
     return error;
@@ -196,7 +194,7 @@ void AutofillProfileSyncableService::Observe(int type,
   // up we are going to process all when MergeData..() is called. If we receive
   // notification after the sync exited, it will be sinced next time Chrome
   // starts.
-  if (!sync_processor_)
+  if (!sync_processor_.get())
     return;
 
   AutofillProfileChange* change = Details<AutofillProfileChange>(details).ptr();
@@ -351,7 +349,7 @@ void AutofillProfileSyncableService::ActOnChange(
   DCHECK((change.type() == AutofillProfileChange::REMOVE &&
           !change.profile()) ||
          (change.type() != AutofillProfileChange::REMOVE && change.profile()));
-  DCHECK(sync_processor_);
+  DCHECK(sync_processor_.get());
   SyncChangeList new_changes;
   DataBundle bundle;
   switch (change.type()) {
diff --git a/chrome/browser/webdata/autofill_profile_syncable_service.h b/chrome/browser/webdata/autofill_profile_syncable_service.h
index 362f03a..605be58 100644
--- a/chrome/browser/webdata/autofill_profile_syncable_service.h
+++ b/chrome/browser/webdata/autofill_profile_syncable_service.h
@@ -130,7 +130,7 @@ class AutofillProfileSyncableService
   // For unit-tests.
   AutofillProfileSyncableService();
   void set_sync_processor(SyncChangeProcessor* sync_processor) {
-    sync_processor_ = sync_processor;
+    sync_processor_.reset(sync_processor);
   }
 
   WebDataService* web_data_service_;  // WEAK
@@ -141,7 +141,7 @@ class AutofillProfileSyncableService
   ScopedVector<AutofillProfile> profiles_;
   GUIDToProfileMap profiles_map_;
 
-  SyncChangeProcessor* sync_processor_;
+  scoped_ptr<SyncChangeProcessor> sync_processor_;
 
   DISALLOW_COPY_AND_ASSIGN(AutofillProfileSyncableService);
 };
diff --git a/chrome/browser/webdata/autofill_profile_syncable_service_unittest.cc b/chrome/browser/webdata/autofill_profile_syncable_service_unittest.cc
index 5639c66..d86da97 100644
--- a/chrome/browser/webdata/autofill_profile_syncable_service_unittest.cc
+++ b/chrome/browser/webdata/autofill_profile_syncable_service_unittest.cc
@@ -32,21 +32,6 @@ class MockAutofillProfileSyncableService
   MOCK_METHOD1(LoadAutofillData, bool(std::vector<AutofillProfile*>*));
   MOCK_METHOD1(SaveChangesToWebData,
                bool(const AutofillProfileSyncableService::DataBundle&));
-
-  // Helper for tests that do not need to setup service completely through the
-  // MergeDataAndStartSyncing().
-  class AutoSetSyncProcessor {
-   public:
-    AutoSetSyncProcessor(MockAutofillProfileSyncableService* service,
-                         SyncChangeProcessor* sync_processor)
-         : service_(service) {
-      service->set_sync_processor(sync_processor);
-    }
-    ~AutoSetSyncProcessor() {
-      service_->set_sync_processor(NULL);
-    }
-    MockAutofillProfileSyncableService* service_;
-  };
 };
 
 ACTION_P(CopyData, data) {
@@ -112,11 +97,23 @@ class AutofillProfileSyncableServiceTest : public testing::Test {
   AutofillProfileSyncableServiceTest()
     : db_thread_(BrowserThread::DB, &message_loop_) {}
 
+  virtual void SetUp() OVERRIDE {
+    sync_processor_.reset(new MockSyncChangeProcessor);
+  }
+
+  virtual void TearDown() OVERRIDE {
+    // Each test passes ownership of the sync processor to the SyncableService.
+    // We don't release it immediately so we can verify the mock calls, so
+    // release it at teardown. Any test that doesn't call
+    // MergeDataAndStartSyncing or set_sync_processor must ensure the
+    // sync_processor_ gets properly reset.
+    ignore_result(sync_processor_.release());
+  }
  protected:
   MessageLoop message_loop_;
   BrowserThread db_thread_;
   MockAutofillProfileSyncableService autofill_syncable_service_;
-  MockSyncChangeProcessor sync_processor_;
+  scoped_ptr<MockSyncChangeProcessor> sync_processor_;
 };
 
 TEST_F(AutofillProfileSyncableServiceTest, MergeDataAndStartSyncing) {
@@ -161,15 +158,16 @@ TEST_F(AutofillProfileSyncableServiceTest, MergeDataAndStartSyncing) {
               SaveChangesToWebData(DataBundleCheck(expected_bundle)))
       .Times(1)
       .WillOnce(Return(true));
-  ON_CALL(sync_processor_, ProcessSyncChanges(_, _))
+  ON_CALL(*sync_processor_, ProcessSyncChanges(_, _))
       .WillByDefault(Return(SyncError()));
-  EXPECT_CALL(sync_processor_,
+  EXPECT_CALL(*sync_processor_,
               ProcessSyncChanges(_, CheckSyncChanges(expected_change_list)))
       .Times(1)
       .WillOnce(Return(SyncError()));
 
+  // Takes ownership of sync_processor_.
   autofill_syncable_service_.MergeDataAndStartSyncing(
-      syncable::AUTOFILL_PROFILE, data_list, &sync_processor_);
+      syncable::AUTOFILL_PROFILE, data_list, sync_processor_.get());
   autofill_syncable_service_.StopSyncing(syncable::AUTOFILL_PROFILE);
 }
 
@@ -189,16 +187,17 @@ TEST_F(AutofillProfileSyncableServiceTest, GetAllSyncData) {
   EXPECT_CALL(autofill_syncable_service_, SaveChangesToWebData(_))
       .Times(1)
       .WillOnce(Return(true));
-  ON_CALL(sync_processor_, ProcessSyncChanges(_, _))
+  ON_CALL(*sync_processor_, ProcessSyncChanges(_, _))
       .WillByDefault(Return(SyncError()));
-  EXPECT_CALL(sync_processor_,
+  EXPECT_CALL(*sync_processor_,
               ProcessSyncChanges(_, Property(&SyncChangeList::size, Eq(2U))))
       .Times(1)
       .WillOnce(Return(SyncError()));
 
   SyncDataList data_list;
+  // Takes ownership of sync_processor_.
   autofill_syncable_service_.MergeDataAndStartSyncing(
-      syncable::AUTOFILL_PROFILE, data_list, &sync_processor_);
+      syncable::AUTOFILL_PROFILE, data_list, sync_processor_.get());
 
   SyncDataList data =
       autofill_syncable_service_.GetAllSyncData(syncable::AUTOFILL_PROFILE);
@@ -235,8 +234,7 @@ TEST_F(AutofillProfileSyncableServiceTest, ProcessSyncChanges) {
       .Times(1)
       .WillOnce(Return(true));
 
-  MockAutofillProfileSyncableService::AutoSetSyncProcessor temp(
-      &autofill_syncable_service_, &sync_processor_);
+  autofill_syncable_service_.set_sync_processor(sync_processor_.get());
   SyncError error = autofill_syncable_service_.ProcessSyncChanges(
       FROM_HERE, change_list);
 
@@ -251,12 +249,11 @@ TEST_F(AutofillProfileSyncableServiceTest, ActOnChange) {
   profile.SetInfo(NAME_FIRST, UTF8ToUTF16("Jane"));
   AutofillProfileChange change1(AutofillProfileChange::ADD, guid1, &profile);
   AutofillProfileChange change2(AutofillProfileChange::REMOVE, guid2, NULL);
-  ON_CALL(sync_processor_, ProcessSyncChanges(_, _))
+  ON_CALL(*sync_processor_, ProcessSyncChanges(_, _))
       .WillByDefault(Return(SyncError(FROM_HERE, std::string("an error"),
                                       syncable::AUTOFILL_PROFILE)));
 
-  MockAutofillProfileSyncableService::AutoSetSyncProcessor temp(
-      &autofill_syncable_service_, &sync_processor_);
+  autofill_syncable_service_.set_sync_processor(sync_processor_.get());
   autofill_syncable_service_.ActOnChange(change1);
   autofill_syncable_service_.ActOnChange(change2);
 }
diff --git a/chrome/chrome_browser.gypi b/chrome/chrome_browser.gypi
index f8b90eb..a59a7a5 100644
--- a/chrome/chrome_browser.gypi
+++ b/chrome/chrome_browser.gypi
@@ -2158,6 +2158,8 @@
         'browser/sync/glue/http_bridge.cc',
         'browser/sync/glue/http_bridge.h',
         'browser/sync/glue/model_associator.h',
+        'browser/sync/glue/new_non_frontend_data_type_controller.cc',
+        'browser/sync/glue/new_non_frontend_data_type_controller.h',
         'browser/sync/glue/non_frontend_data_type_controller.cc',
         'browser/sync/glue/non_frontend_data_type_controller.h',
         'browser/sync/glue/password_change_processor.cc',
@@ -2179,6 +2181,10 @@
         'browser/sync/glue/session_model_associator.cc',
         'browser/sync/glue/session_model_associator_mac.mm',
         'browser/sync/glue/session_model_associator.h',
+        'browser/sync/glue/shared_change_processor.cc',
+        'browser/sync/glue/shared_change_processor.h',
+        'browser/sync/glue/shared_change_processor_ref.h',
+        'browser/sync/glue/shared_change_processor_ref.cc',
         'browser/sync/glue/syncable_service_adapter.cc',
         'browser/sync/glue/syncable_service_adapter.h',
         'browser/sync/glue/synced_session.h',
diff --git a/chrome/chrome_tests.gypi b/chrome/chrome_tests.gypi
index 78f01c3..d210b82 100644
--- a/chrome/chrome_tests.gypi
+++ b/chrome/chrome_tests.gypi
@@ -1080,6 +1080,7 @@
         'test_support_common',
         'test_support_sync',
         'test_support_syncapi',
+        'test_support_syncapi_service',
         'test_support_unit',
         # 3) anything tests directly depend on
         '../skia/skia.gyp:skia',
@@ -1537,12 +1538,17 @@
         'browser/sync/glue/http_bridge_unittest.cc',
         'browser/sync/glue/model_associator_mock.cc',
         'browser/sync/glue/model_associator_mock.h',
+        'browser/sync/glue/new_non_frontend_data_type_controller_mock.cc',
+        'browser/sync/glue/new_non_frontend_data_type_controller_mock.h',
+        'browser/sync/glue/new_non_frontend_data_type_controller_unittest.cc',
         'browser/sync/glue/non_frontend_data_type_controller_mock.cc',
         'browser/sync/glue/non_frontend_data_type_controller_mock.h',
         'browser/sync/glue/non_frontend_data_type_controller_unittest.cc',
         'browser/sync/glue/preference_data_type_controller_unittest.cc',
         'browser/sync/glue/search_engine_data_type_controller_unittest.cc',
         'browser/sync/glue/session_model_associator_unittest.cc',
+        'browser/sync/glue/shared_change_processor_mock.cc',
+        'browser/sync/glue/shared_change_processor_mock.h',
         'browser/sync/glue/sync_backend_host_mock.cc',
         'browser/sync/glue/sync_backend_host_mock.h',
         'browser/sync/glue/sync_backend_host_unittest.cc',