sync: Implement disabling of non blocking types

Adds logic to have the model-thread NonBlockingDataTypeProcessor send
a message to the sync-thread SyncCore when it receives notifcation from
the ui-thread NonBlockingDataTypeController that it should stop syncing.
This message will allow the sync thread to stop requeting updates and
commits on behalf of the now-disabled type.

Fixes the handling of a race in the NonBlockingDataTypeProcessor.  The
race is as follows:
1. NBDTP receives a request to enable sync from the UI thread, and sends
   a connection request to the sync thread via SyncCoreProxy.
2. NBDTP receives a request to disable sync from the UI thread.  It
   updates its internal state accordingly.
3. NBDTP receives the connection OK response from the sync thread,
   which was genrated in response to its request in step 1.

Previously, the processor would wrongly set itself to the enable state
in step 3.  The fix is to use a new WeakPtrFactory and invalidate the
pointers it has issued in step 2 in order to prevent the response seen
in step 3 from being run.

Adds some more tests for this scenario and more.  The test framework had
to be made a bit more complicated in order to handle these tests, but
I think the extra complexity is worth it.  I don't know of any other
way to reliably defend against these race cases.

BUG=351005

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

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

10 files changed, 75 insertions, 14 deletions

diff --git a/sync/internal_api/non_blocking_type_processor.cc b/sync/internal_api/non_blocking_type_processor.cc
index d426443..870c296 100644
--- a/sync/internal_api/non_blocking_type_processor.cc
+++ b/sync/internal_api/non_blocking_type_processor.cc
@@ -11,10 +11,12 @@
 namespace syncer {
 
 NonBlockingTypeProcessor::NonBlockingTypeProcessor(ModelType type)
-  : type_(type),
-    is_preferred_(false),
-    is_connected_(false),
-    weak_ptr_factory_(this) {}
+    : type_(type),
+      is_preferred_(false),
+      is_connected_(false),
+      weak_ptr_factory_for_ui_(this),
+      weak_ptr_factory_for_sync_(this) {
+}
 
 NonBlockingTypeProcessor::~NonBlockingTypeProcessor() {
 }
@@ -34,10 +36,14 @@ ModelType NonBlockingTypeProcessor::GetModelType() const {
   return type_;
 }
 
-void NonBlockingTypeProcessor::Enable(SyncCoreProxy* core_proxy) {
+void NonBlockingTypeProcessor::Enable(
+    scoped_ptr<SyncCoreProxy> sync_core_proxy) {
   DCHECK(CalledOnValidThread());
+  DVLOG(1) << "Asked to enable " << ModelTypeToString(type_);
   is_preferred_ = true;
-  core_proxy->ConnectTypeToCore(GetModelType(), AsWeakPtr());
+  sync_core_proxy_ = sync_core_proxy.Pass();
+  sync_core_proxy_->ConnectTypeToCore(GetModelType(),
+                                      weak_ptr_factory_for_sync_.GetWeakPtr());
 }
 
 void NonBlockingTypeProcessor::Disable() {
@@ -48,20 +54,30 @@ void NonBlockingTypeProcessor::Disable() {
 
 void NonBlockingTypeProcessor::Disconnect() {
   DCHECK(CalledOnValidThread());
+  DVLOG(1) << "Asked to disconnect " << ModelTypeToString(type_);
   is_connected_ = false;
+
+  if (sync_core_proxy_) {
+    sync_core_proxy_->Disconnect(GetModelType());
+    sync_core_proxy_.reset();
+  }
+
+  weak_ptr_factory_for_sync_.InvalidateWeakPtrs();
   core_ = base::WeakPtr<NonBlockingTypeProcessorCore>();
   sync_thread_ = scoped_refptr<base::SequencedTaskRunner>();
 }
 
-base::WeakPtr<NonBlockingTypeProcessor> NonBlockingTypeProcessor::AsWeakPtr() {
+base::WeakPtr<NonBlockingTypeProcessor>
+NonBlockingTypeProcessor::AsWeakPtrForUI() {
   DCHECK(CalledOnValidThread());
-  return weak_ptr_factory_.GetWeakPtr();
+  return weak_ptr_factory_for_ui_.GetWeakPtr();
 }
 
 void NonBlockingTypeProcessor::OnConnect(
     base::WeakPtr<NonBlockingTypeProcessorCore> core,
     scoped_refptr<base::SequencedTaskRunner> sync_thread) {
   DCHECK(CalledOnValidThread());
+  DVLOG(1) << "Successfully connected " << ModelTypeToString(type_);
   is_connected_ = true;
   core_ = core;
   sync_thread_ = sync_thread;
diff --git a/sync/internal_api/public/non_blocking_type_processor.h b/sync/internal_api/public/non_blocking_type_processor.h
index 3745088..2db2c32 100644
--- a/sync/internal_api/public/non_blocking_type_processor.h
+++ b/sync/internal_api/public/non_blocking_type_processor.h
@@ -43,7 +43,7 @@ class SYNC_EXPORT_PRIVATE NonBlockingTypeProcessor : base::NonThreadSafe {
   ModelType GetModelType() const;
 
   // Starts the handshake with the sync thread.
-  void Enable(SyncCoreProxy* core_proxy);
+  void Enable(scoped_ptr<SyncCoreProxy> core_proxy);
 
   // Severs all ties to the sync thread and may delete local sync state.
   // Another call to Enable() can be used to re-establish this connection.
@@ -57,7 +57,9 @@ class SYNC_EXPORT_PRIVATE NonBlockingTypeProcessor : base::NonThreadSafe {
   void OnConnect(base::WeakPtr<NonBlockingTypeProcessorCore> core,
                  scoped_refptr<base::SequencedTaskRunner> sync_thread);
 
-  base::WeakPtr<NonBlockingTypeProcessor> AsWeakPtr();
+  // Returns the long-lived WeakPtr that is intended to be registered with the
+  // ProfileSyncService.
+  base::WeakPtr<NonBlockingTypeProcessor> AsWeakPtrForUI();
 
  private:
   ModelType type_;
@@ -71,10 +73,20 @@ class SYNC_EXPORT_PRIVATE NonBlockingTypeProcessor : base::NonThreadSafe {
   // SyncCoreProxy.
   bool is_connected_;
 
+  // Our link to data type management on the sync thread.
+  // Used for enabling and disabling sync for this type.
+  scoped_ptr<SyncCoreProxy> sync_core_proxy_;
+
   base::WeakPtr<NonBlockingTypeProcessorCore> core_;
   scoped_refptr<base::SequencedTaskRunner> sync_thread_;
 
-  base::WeakPtrFactory<NonBlockingTypeProcessor> weak_ptr_factory_;
+  // We use two different WeakPtrFactories because we want the pointers they
+  // issue to have different lifetimes.  When asked to disconnect from the sync
+  // thread, we want to make sure that no tasks generated as part of the
+  // now-obsolete connection to affect us.  But we also want the WeakPtr we
+  // sent to the UI thread to remain valid.
+  base::WeakPtrFactory<NonBlockingTypeProcessor> weak_ptr_factory_for_ui_;
+  base::WeakPtrFactory<NonBlockingTypeProcessor> weak_ptr_factory_for_sync_;
 };
 
 }  // namespace syncer
diff --git a/sync/internal_api/public/sync_core_proxy.h b/sync/internal_api/public/sync_core_proxy.h
index 5ec8efe..98f481d 100644
--- a/sync/internal_api/public/sync_core_proxy.h
+++ b/sync/internal_api/public/sync_core_proxy.h
@@ -27,6 +27,13 @@ class SYNC_EXPORT_PRIVATE SyncCoreProxy {
       syncer::ModelType type,
       base::WeakPtr<NonBlockingTypeProcessor> type_processor) = 0;
 
+  // Tells the syncer that we're no longer interested in syncing this type.
+  //
+  // Once this takes effect, the syncer can assume that it will no longer
+  // receive commit requests for this type.  It should also stop requesting
+  // and applying updates for this type, too.
+  virtual void Disconnect(syncer::ModelType type) = 0;
+
   // Creates a clone of this SyncCoreProxy.
   virtual scoped_ptr<SyncCoreProxy> Clone() const = 0;
 };
diff --git a/sync/internal_api/public/test/null_sync_core_proxy.h b/sync/internal_api/public/test/null_sync_core_proxy.h
index a491de9..d957aec5 100644
--- a/sync/internal_api/public/test/null_sync_core_proxy.h
+++ b/sync/internal_api/public/test/null_sync_core_proxy.h
@@ -23,6 +23,7 @@ class NullSyncCoreProxy : public SyncCoreProxy {
   virtual void ConnectTypeToCore(
       syncer::ModelType type,
       base::WeakPtr<NonBlockingTypeProcessor> processor) OVERRIDE;
+  virtual void Disconnect(syncer::ModelType type) OVERRIDE;
   virtual scoped_ptr<SyncCoreProxy> Clone() const OVERRIDE;
 };
 
diff --git a/sync/internal_api/sync_core.cc b/sync/internal_api/sync_core.cc
index 9c26095..dbeb76d 100644
--- a/sync/internal_api/sync_core.cc
+++ b/sync/internal_api/sync_core.cc
@@ -25,6 +25,10 @@ void SyncCore::ConnectSyncTypeToCore(
   model_type_registry_->InitializeNonBlockingType(type, task_runner, processor);
 }
 
+void SyncCore::Disconnect(ModelType type) {
+  model_type_registry_->RemoveNonBlockingType(type);
+}
+
 base::WeakPtr<SyncCore> SyncCore::AsWeakPtr() {
   return weak_ptr_factory_.GetWeakPtr();
 }
diff --git a/sync/internal_api/sync_core.h b/sync/internal_api/sync_core.h
index 504304c..6f63070 100644
--- a/sync/internal_api/sync_core.h
+++ b/sync/internal_api/sync_core.h
@@ -36,6 +36,17 @@ class SYNC_EXPORT_PRIVATE SyncCore {
     scoped_refptr<base::SequencedTaskRunner> datatype_task_runner,
     base::WeakPtr<NonBlockingTypeProcessor> sync_client);
 
+  // Disconnects the syncer from the model and stops syncing the type.
+  //
+  // By the time this is called, the model thread should have already
+  // invalidated the WeakPtr it sent to us in the connect request.  Any
+  // messages sent to that NonBlockingTypeProcessor will not be recived.
+  //
+  // This is the sync thread's chance to clear state associated with the type.
+  // It also causes the syncer to stop requesting updates for this type, and to
+  // abort any in-progress commit requests.
+  void Disconnect(ModelType type);
+
   base::WeakPtr<SyncCore> AsWeakPtr();
 
  private:
diff --git a/sync/internal_api/sync_core_proxy_impl.cc b/sync/internal_api/sync_core_proxy_impl.cc
index a75041ac..7e4ecce 100644
--- a/sync/internal_api/sync_core_proxy_impl.cc
+++ b/sync/internal_api/sync_core_proxy_impl.cc
@@ -32,6 +32,11 @@ void SyncCoreProxyImpl::ConnectTypeToCore(
                  type_processor));
 }
 
+void SyncCoreProxyImpl::Disconnect(ModelType type) {
+  sync_task_runner_->PostTask(
+      FROM_HERE, base::Bind(&SyncCore::Disconnect, sync_core_, type));
+}
+
 scoped_ptr<SyncCoreProxy> SyncCoreProxyImpl::Clone() const {
   return scoped_ptr<SyncCoreProxy>(
       new SyncCoreProxyImpl(sync_task_runner_, sync_core_));
diff --git a/sync/internal_api/sync_core_proxy_impl.h b/sync/internal_api/sync_core_proxy_impl.h
index 99e1b8f..3622d2b 100644
--- a/sync/internal_api/sync_core_proxy_impl.h
+++ b/sync/internal_api/sync_core_proxy_impl.h
@@ -40,6 +40,9 @@ class SYNC_EXPORT_PRIVATE SyncCoreProxyImpl : public SyncCoreProxy {
       syncer::ModelType type,
       base::WeakPtr<NonBlockingTypeProcessor> type_processor) OVERRIDE;
 
+  // Disables syncing for the given type on the sync thread.
+  virtual void Disconnect(syncer::ModelType type) OVERRIDE;
+
   virtual scoped_ptr<SyncCoreProxy> Clone() const OVERRIDE;
 
  private:
diff --git a/sync/internal_api/sync_core_proxy_impl_unittest.cc b/sync/internal_api/sync_core_proxy_impl_unittest.cc
index 2e306ee..d1913b0 100644
--- a/sync/internal_api/sync_core_proxy_impl_unittest.cc
+++ b/sync/internal_api/sync_core_proxy_impl_unittest.cc
@@ -31,9 +31,7 @@ class SyncCoreProxyImplTest : public ::testing::Test {
     core_.reset();
   }
 
-  SyncCoreProxy* GetProxy() {
-    return &core_proxy_;
-  }
+  scoped_ptr<SyncCoreProxy> GetProxy() { return core_proxy_.Clone(); }
 
  private:
   base::MessageLoop loop_;
diff --git a/sync/internal_api/test/null_sync_core_proxy.cc b/sync/internal_api/test/null_sync_core_proxy.cc
index 10fc757..e0c93f6 100644
--- a/sync/internal_api/test/null_sync_core_proxy.cc
+++ b/sync/internal_api/test/null_sync_core_proxy.cc
@@ -16,6 +16,10 @@ void NullSyncCoreProxy::ConnectTypeToCore(
   NOTREACHED() << "NullSyncCoreProxy is not meant to be used";
 }
 
+void NullSyncCoreProxy::Disconnect(syncer::ModelType type) {
+  NOTREACHED() << "NullSyncCoreProxy is not meant to be used";
+}
+
 scoped_ptr<SyncCoreProxy> NullSyncCoreProxy::Clone() const {
   return scoped_ptr<SyncCoreProxy>(new NullSyncCoreProxy());
 }