[Sync] Replace uses of ObserverListThreadSafe with WeakHandles

ObserverListThreadSafe was overkill since there was only two threads
involved, and only one observer.

Add MessageLoop member variables to various test classes that need it
(ObserverListThreadSafe had a hack that made it unnecessary before.)

BUG=103732
TEST=Start with a clean profile and set up sync.  about:profiler shouldn't
show ObserverListThreadSafe::Notify calls with sync threads anymore.


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

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

23 files changed, 303 insertions, 181 deletions

diff --git a/chrome/browser/sync/engine/syncer_proto_util_unittest.cc b/chrome/browser/sync/engine/syncer_proto_util_unittest.cc
index a22e1a3..809ce1e 100644
--- a/chrome/browser/sync/engine/syncer_proto_util_unittest.cc
+++ b/chrome/browser/sync/engine/syncer_proto_util_unittest.cc
@@ -8,6 +8,7 @@
 
 #include "base/basictypes.h"
 #include "base/compiler_specific.h"
+#include "base/message_loop.h"
 #include "chrome/browser/sync/engine/syncproto.h"
 #include "chrome/browser/sync/syncable/blob.h"
 #include "chrome/browser/sync/syncable/directory_manager.h"
@@ -137,6 +138,7 @@ class SyncerProtoUtilTest : public testing::Test {
   }
 
  protected:
+  MessageLoop message_loop_;
   browser_sync::TestDirectorySetterUpper setter_upper_;
 };
 
diff --git a/chrome/browser/sync/engine/syncer_unittest.cc b/chrome/browser/sync/engine/syncer_unittest.cc
index 9985120..b4b4003 100644
--- a/chrome/browser/sync/engine/syncer_unittest.cc
+++ b/chrome/browser/sync/engine/syncer_unittest.cc
@@ -16,6 +16,7 @@
 #include "base/compiler_specific.h"
 #include "base/location.h"
 #include "base/memory/scoped_ptr.h"
+#include "base/message_loop.h"
 #include "base/string_number_conversions.h"
 #include "base/stringprintf.h"
 #include "base/time.h"
@@ -468,6 +469,8 @@ class SyncerTest : public testing::Test,
     return GetField(metahandle, field, false);
   }
 
+  MessageLoop message_loop_;
+
   // Some ids to aid tests. Only the root one's value is specific. The rest
   // are named for test clarity.
   // TODO(chron): Get rid of these inbuilt IDs. They only make it
diff --git a/chrome/browser/sync/internal_api/debug_info_event_listener.h b/chrome/browser/sync/internal_api/debug_info_event_listener.h
index bc7b592..52f8fcd 100644
--- a/chrome/browser/sync/internal_api/debug_info_event_listener.h
+++ b/chrome/browser/sync/internal_api/debug_info_event_listener.h
@@ -5,6 +5,7 @@
 #ifndef CHROME_BROWSER_SYNC_INTERNAL_API_DEBUG_INFO_EVENT_LISTENER_H_
 #define CHROME_BROWSER_SYNC_INTERNAL_API_DEBUG_INFO_EVENT_LISTENER_H_
 
+#include <queue>
 #include <string>
 
 #include "chrome/browser/sync/internal_api/sync_manager.h"
diff --git a/chrome/browser/sync/internal_api/sync_manager.cc b/chrome/browser/sync/internal_api/sync_manager.cc
index 1719a57..07801ef 100644
--- a/chrome/browser/sync/internal_api/sync_manager.cc
+++ b/chrome/browser/sync/internal_api/sync_manager.cc
@@ -12,7 +12,6 @@
 #include "base/json/json_writer.h"
 #include "base/memory/ref_counted.h"
 #include "base/observer_list.h"
-#include "base/observer_list_threadsafe.h"
 #include "base/string_number_conversions.h"
 #include "base/values.h"
 #include "chrome/browser/sync/engine/all_status.h"
@@ -131,8 +130,6 @@ class SyncManager::SyncInternal
   explicit SyncInternal(const std::string& name)
       : name_(name),
         weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)),
-        change_observers_(
-            new ObserverListThreadSafe<SyncManager::ChangeObserver>()),
         registrar_(NULL),
         change_delegate_(NULL),
         initialized_(false),
@@ -277,9 +274,6 @@ class SyncManager::SyncInternal
 
   virtual void StoreState(const std::string& cookie) OVERRIDE;
 
-  void AddChangeObserver(SyncManager::ChangeObserver* observer);
-  void RemoveChangeObserver(SyncManager::ChangeObserver* observer);
-
   void AddObserver(SyncManager::Observer* observer);
   void RemoveObserver(SyncManager::Observer* observer);
 
@@ -495,11 +489,9 @@ class SyncManager::SyncInternal
   // constructing any transaction type.
   UserShare share_;
 
-  // Even though observers are always added/removed from the sync
-  // thread, we still need to use a thread-safe observer list as we
-  // can notify from any thread.
-  scoped_refptr<ObserverListThreadSafe<SyncManager::ChangeObserver> >
-      change_observers_;
+  // This can be called from any thread, but only between calls to
+  // OpenDirectory() and ShutdownOnSyncThread().
+  browser_sync::WeakHandle<SyncManager::ChangeObserver> change_observer_;
 
   ObserverList<SyncManager::Observer> observers_;
 
@@ -893,7 +885,16 @@ void SyncManager::SyncInternal::StartSyncingNormally() {
 bool SyncManager::SyncInternal::OpenDirectory() {
   DCHECK(!initialized_) << "Should only happen once";
 
-  bool share_opened = dir_manager()->Open(username_for_share(), this);
+  // Set before Open().
+  change_observer_ =
+      browser_sync::MakeWeakHandle(js_mutation_event_observer_.AsWeakPtr());
+
+  bool share_opened =
+      dir_manager()->Open(
+          username_for_share(),
+          this,
+          browser_sync::MakeWeakHandle(
+              js_mutation_event_observer_.AsWeakPtr()));
   if (!share_opened) {
     LOG(ERROR) << "Could not open share for:" << username_for_share();
     return false;
@@ -907,8 +908,6 @@ bool SyncManager::SyncInternal::OpenDirectory() {
   }
 
   connection_manager()->set_client_id(lookup->cache_guid());
-  lookup->AddTransactionObserver(&js_mutation_event_observer_);
-  AddChangeObserver(&js_mutation_event_observer_);
   return true;
 }
 
@@ -1223,16 +1222,6 @@ SyncManager::~SyncManager() {
   delete data_;
 }
 
-void SyncManager::AddChangeObserver(ChangeObserver* observer) {
-  DCHECK(thread_checker_.CalledOnValidThread());
-  data_->AddChangeObserver(observer);
-}
-
-void SyncManager::RemoveChangeObserver(ChangeObserver* observer) {
-  DCHECK(thread_checker_.CalledOnValidThread());
-  data_->RemoveChangeObserver(observer);
-}
-
 void SyncManager::AddObserver(Observer* observer) {
   DCHECK(thread_checker_.CalledOnValidThread());
   data_->AddObserver(observer);
@@ -1268,8 +1257,9 @@ void SyncManager::SyncInternal::ShutdownOnSyncThread() {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   // Prevent any in-flight method calls from running.  Also
-  // invalidates |weak_handle_this_|.
+  // invalidates |weak_handle_this_| and |change_observer_|.
   weak_ptr_factory_.InvalidateWeakPtrs();
+  js_mutation_event_observer_.InvalidateWeakPtrs();
 
   scheduler_.reset();
 
@@ -1297,9 +1287,6 @@ void SyncManager::SyncInternal::ShutdownOnSyncThread() {
       // transaction.
       ReadTransaction trans(FROM_HERE, GetUserShare());
       trans.GetCryptographer()->RemoveObserver(this);
-      trans.GetLookup()->
-          RemoveTransactionObserver(&js_mutation_event_observer_);
-      RemoveChangeObserver(&js_mutation_event_observer_);
     }
     dir_manager()->FinalSaveChangesForAll();
     dir_manager()->Close(username_for_share());
@@ -1315,8 +1302,9 @@ void SyncManager::SyncInternal::ShutdownOnSyncThread() {
 
   initialized_ = false;
 
-  // We reset this here, since only now we know it will not be
+  // We reset these here, since only now we know they will not be
   // accessed from other threads (since we shut down everything).
+  change_observer_.Reset();
   weak_handle_this_.Reset();
 }
 
@@ -1382,7 +1370,7 @@ void SyncManager::SyncInternal::HandleTransactionCompleteChangeEvent(
     const syncable::ModelType type = syncable::ModelTypeFromInt(i);
     if (models_with_changes.test(type)) {
       change_delegate_->OnChangesComplete(type);
-      change_observers_->Notify(
+      change_observer_.Call(FROM_HERE,
           &SyncManager::ChangeObserver::OnChangesComplete, type);
     }
   }
@@ -1418,7 +1406,7 @@ ModelTypeBitSet SyncManager::SyncInternal::HandleTransactionEndingChangeEvent(
     if (!ordered_changes.Get().empty()) {
       change_delegate_->
           OnChangesApplied(type, &read_trans, ordered_changes);
-      change_observers_->Notify(
+      change_observer_.Call(FROM_HERE,
           &SyncManager::ChangeObserver::OnChangesApplied,
           type, write_transaction_info.Get().id, ordered_changes);
       models_with_changes.set(i, true);
@@ -1962,16 +1950,6 @@ void SyncManager::SyncInternal::StoreState(
   lookup->SaveChanges();
 }
 
-void SyncManager::SyncInternal::AddChangeObserver(
-    SyncManager::ChangeObserver* observer) {
-  change_observers_->AddObserver(observer);
-}
-
-void SyncManager::SyncInternal::RemoveChangeObserver(
-    SyncManager::ChangeObserver* observer) {
-  change_observers_->RemoveObserver(observer);
-}
-
 void SyncManager::SyncInternal::AddObserver(
     SyncManager::Observer* observer) {
   observers_.AddObserver(observer);
diff --git a/chrome/browser/sync/internal_api/sync_manager.h b/chrome/browser/sync/internal_api/sync_manager.h
index 3133f78..eaf5ccb 100644
--- a/chrome/browser/sync/internal_api/sync_manager.h
+++ b/chrome/browser/sync/internal_api/sync_manager.h
@@ -496,10 +496,6 @@ class SyncManager {
   // Request a clearing of all data on the server
   void RequestClearServerData();
 
-  // Add/remove change observers.
-  void AddChangeObserver(ChangeObserver* observer);
-  void RemoveChangeObserver(ChangeObserver* observer);
-
   // Adds a listener to be notified of sync events.
   // NOTE: It is OK (in fact, it's probably a good idea) to call this before
   // having received OnInitializationCompleted.
diff --git a/chrome/browser/sync/internal_api/syncapi_unittest.cc b/chrome/browser/sync/internal_api/syncapi_unittest.cc
index d8b9277..6e3c296 100644
--- a/chrome/browser/sync/internal_api/syncapi_unittest.cc
+++ b/chrome/browser/sync/internal_api/syncapi_unittest.cc
@@ -192,6 +192,7 @@ class SyncApiTest : public testing::Test {
   }
 
  protected:
+  MessageLoop message_loop_;
   browser_sync::TestUserShare test_user_share_;
 };
 
diff --git a/chrome/browser/sync/js/js_mutation_event_observer.cc b/chrome/browser/sync/js/js_mutation_event_observer.cc
index 5087a23..2c323d1 100644
--- a/chrome/browser/sync/js/js_mutation_event_observer.cc
+++ b/chrome/browser/sync/js/js_mutation_event_observer.cc
@@ -15,12 +15,21 @@
 
 namespace browser_sync {
 
-JsMutationEventObserver::JsMutationEventObserver() {}
+JsMutationEventObserver::JsMutationEventObserver()
+    : weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)) {}
 
 JsMutationEventObserver::~JsMutationEventObserver() {
   DCHECK(non_thread_safe_.CalledOnValidThread());
 }
 
+base::WeakPtr<JsMutationEventObserver> JsMutationEventObserver::AsWeakPtr() {
+  return weak_ptr_factory_.GetWeakPtr();
+}
+
+void JsMutationEventObserver::InvalidateWeakPtrs() {
+  weak_ptr_factory_.InvalidateWeakPtrs();
+}
+
 void JsMutationEventObserver::SetJsEventHandler(
     const WeakHandle<JsEventHandler>& event_handler) {
   event_handler_ = event_handler;
diff --git a/chrome/browser/sync/js/js_mutation_event_observer.h b/chrome/browser/sync/js/js_mutation_event_observer.h
index ced1d91..c4d7d12 100644
--- a/chrome/browser/sync/js/js_mutation_event_observer.h
+++ b/chrome/browser/sync/js/js_mutation_event_observer.h
@@ -10,6 +10,7 @@
 
 #include "base/basictypes.h"
 #include "base/compiler_specific.h"
+#include "base/memory/weak_ptr.h"
 #include "base/threading/non_thread_safe.h"
 #include "chrome/browser/sync/internal_api/sync_manager.h"
 #include "chrome/browser/sync/syncable/transaction_observer.h"
@@ -34,6 +35,10 @@ class JsMutationEventObserver
 
   virtual ~JsMutationEventObserver();
 
+  base::WeakPtr<JsMutationEventObserver> AsWeakPtr();
+
+  void InvalidateWeakPtrs();
+
   void SetJsEventHandler(const WeakHandle<JsEventHandler>& event_handler);
 
   // sync_api::SyncManager::ChangeObserver implementation.
@@ -56,6 +61,7 @@ class JsMutationEventObserver
 
  private:
   base::NonThreadSafe non_thread_safe_;
+  base::WeakPtrFactory<JsMutationEventObserver> weak_ptr_factory_;
   WeakHandle<JsEventHandler> event_handler_;
 
   void HandleJsEvent(
diff --git a/chrome/browser/sync/notifier/non_blocking_invalidation_notifier.cc b/chrome/browser/sync/notifier/non_blocking_invalidation_notifier.cc
index 40a1f91..674c223 100644
--- a/chrome/browser/sync/notifier/non_blocking_invalidation_notifier.cc
+++ b/chrome/browser/sync/notifier/non_blocking_invalidation_notifier.cc
@@ -7,10 +7,8 @@
 #include "base/logging.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/message_loop.h"
-#include "base/observer_list_threadsafe.h"
 #include "base/threading/thread.h"
 #include "chrome/browser/sync/notifier/invalidation_notifier.h"
-#include "chrome/browser/sync/notifier/sync_notifier_observer.h"
 
 namespace sync_notifier {
 
@@ -18,12 +16,11 @@ class NonBlockingInvalidationNotifier::Core
     : public base::RefCountedThreadSafe<NonBlockingInvalidationNotifier::Core>,
       public SyncNotifierObserver {
  public:
-  // Called on parent thread.
-  Core();
-
-  // Called on parent thread.
-  void AddObserver(SyncNotifierObserver* observer);
-  void RemoveObserver(SyncNotifierObserver* observer);
+  // Called on parent thread.  |delegate_observer| should be
+  // initialized.
+  explicit Core(
+      const browser_sync::WeakHandle<SyncNotifierObserver>&
+          delegate_observer);
 
   // Helpers called on I/O thread.
   void Initialize(
@@ -50,14 +47,19 @@ class NonBlockingInvalidationNotifier::Core
   // Called on parent or I/O thread.
   ~Core();
 
+  // The variables below should be used only on the I/O thread.
+  const browser_sync::WeakHandle<SyncNotifierObserver> delegate_observer_;
   scoped_ptr<InvalidationNotifier> invalidation_notifier_;
   scoped_refptr<base::MessageLoopProxy> io_message_loop_proxy_;
-  scoped_refptr<ObserverListThreadSafe<SyncNotifierObserver> > observers_;
+
   DISALLOW_COPY_AND_ASSIGN(Core);
 };
 
-NonBlockingInvalidationNotifier::Core::Core()
-    : observers_(new ObserverListThreadSafe<SyncNotifierObserver>()) {
+NonBlockingInvalidationNotifier::Core::Core(
+    const browser_sync::WeakHandle<SyncNotifierObserver>&
+        delegate_observer)
+    : delegate_observer_(delegate_observer) {
+  DCHECK(delegate_observer_.IsInitialized());
 }
 
 NonBlockingInvalidationNotifier::Core::~Core() {
@@ -92,16 +94,6 @@ void NonBlockingInvalidationNotifier::Core::Teardown() {
   io_message_loop_proxy_ = NULL;
 }
 
-void NonBlockingInvalidationNotifier::Core::AddObserver(
-    SyncNotifierObserver* observer) {
-  observers_->AddObserver(observer);
-}
-
-void NonBlockingInvalidationNotifier::Core::RemoveObserver(
-    SyncNotifierObserver* observer) {
-  observers_->RemoveObserver(observer);
-}
-
 void NonBlockingInvalidationNotifier::Core::SetUniqueId(
     const std::string& unique_id) {
   DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
@@ -129,21 +121,24 @@ void NonBlockingInvalidationNotifier::Core::UpdateEnabledTypes(
 void NonBlockingInvalidationNotifier::Core::OnIncomingNotification(
         const syncable::ModelTypePayloadMap& type_payloads) {
   DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
-  observers_->Notify(&SyncNotifierObserver::OnIncomingNotification,
-                     type_payloads);
+  delegate_observer_.Call(FROM_HERE,
+                          &SyncNotifierObserver::OnIncomingNotification,
+                          type_payloads);
 }
 
 void NonBlockingInvalidationNotifier::Core::OnNotificationStateChange(
         bool notifications_enabled) {
   DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
-  observers_->Notify(&SyncNotifierObserver::OnNotificationStateChange,
-                     notifications_enabled);
+  delegate_observer_.Call(FROM_HERE,
+                          &SyncNotifierObserver::OnNotificationStateChange,
+                          notifications_enabled);
 }
 
 void NonBlockingInvalidationNotifier::Core::StoreState(
     const std::string& state) {
   DCHECK(io_message_loop_proxy_->BelongsToCurrentThread());
-  observers_->Notify(&SyncNotifierObserver::StoreState, state);
+  delegate_observer_.Call(FROM_HERE,
+                          &SyncNotifierObserver::StoreState, state);
 }
 
 NonBlockingInvalidationNotifier::NonBlockingInvalidationNotifier(
@@ -152,7 +147,10 @@ NonBlockingInvalidationNotifier::NonBlockingInvalidationNotifier(
     const browser_sync::WeakHandle<InvalidationVersionTracker>&
         invalidation_version_tracker,
     const std::string& client_info)
-        : core_(new Core),
+        : weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)),
+          core_(
+              new Core(browser_sync::MakeWeakHandle(
+                  weak_ptr_factory_.GetWeakPtr()))),
           parent_message_loop_proxy_(
               base::MessageLoopProxy::current()),
           io_message_loop_proxy_(notifier_options.request_context_getter->
@@ -183,13 +181,13 @@ NonBlockingInvalidationNotifier::~NonBlockingInvalidationNotifier() {
 void NonBlockingInvalidationNotifier::AddObserver(
     SyncNotifierObserver* observer) {
   DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  core_->AddObserver(observer);
+  observers_.AddObserver(observer);
 }
 
 void NonBlockingInvalidationNotifier::RemoveObserver(
     SyncNotifierObserver* observer) {
   DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  core_->RemoveObserver(observer);
+  observers_.RemoveObserver(observer);
 }
 
 void NonBlockingInvalidationNotifier::SetUniqueId(
@@ -242,4 +240,25 @@ void NonBlockingInvalidationNotifier::SendNotification(
   // need to forward on the call.
 }
 
+void NonBlockingInvalidationNotifier::OnIncomingNotification(
+        const syncable::ModelTypePayloadMap& type_payloads) {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  FOR_EACH_OBSERVER(SyncNotifierObserver, observers_,
+                    OnIncomingNotification(type_payloads));
+}
+
+void NonBlockingInvalidationNotifier::OnNotificationStateChange(
+        bool notifications_enabled) {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  FOR_EACH_OBSERVER(SyncNotifierObserver, observers_,
+                    OnNotificationStateChange(notifications_enabled));
+}
+
+void NonBlockingInvalidationNotifier::StoreState(
+    const std::string& state) {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  FOR_EACH_OBSERVER(SyncNotifierObserver, observers_,
+                    StoreState(state));
+}
+
 }  // namespace sync_notifier
diff --git a/chrome/browser/sync/notifier/non_blocking_invalidation_notifier.h b/chrome/browser/sync/notifier/non_blocking_invalidation_notifier.h
index 351d67f..45da2f6 100644
--- a/chrome/browser/sync/notifier/non_blocking_invalidation_notifier.h
+++ b/chrome/browser/sync/notifier/non_blocking_invalidation_notifier.h
@@ -14,8 +14,11 @@
 #include "base/basictypes.h"
 #include "base/compiler_specific.h"
 #include "base/memory/ref_counted.h"
+#include "base/memory/weak_ptr.h"
+#include "base/observer_list.h"
 #include "chrome/browser/sync/notifier/invalidation_version_tracker.h"
 #include "chrome/browser/sync/notifier/sync_notifier.h"
+#include "chrome/browser/sync/notifier/sync_notifier_observer.h"
 #include "chrome/browser/sync/util/weak_handle.h"
 #include "jingle/notifier/base/notifier_options.h"
 
@@ -25,7 +28,9 @@ class MessageLoopProxy;
 
 namespace sync_notifier {
 
-class NonBlockingInvalidationNotifier : public SyncNotifier {
+class NonBlockingInvalidationNotifier
+    : public SyncNotifier,
+      public SyncNotifierObserver {
  public:
   // |invalidation_version_tracker| must be initialized.
   NonBlockingInvalidationNotifier(
@@ -49,13 +54,26 @@ class NonBlockingInvalidationNotifier : public SyncNotifier {
   virtual void SendNotification(
       const syncable::ModelTypeSet& changed_types) OVERRIDE;
 
+  // SyncNotifierObserver implementation.
+  virtual void OnIncomingNotification(
+      const syncable::ModelTypePayloadMap& type_payloads) OVERRIDE;
+  virtual void OnNotificationStateChange(bool notifications_enabled) OVERRIDE;
+  virtual void StoreState(const std::string& state) OVERRIDE;
+
  private:
-  // The real guts of NonBlockingInvalidationNotifier, which allows this class
-  // to not be refcounted.
   class Core;
+
+  base::WeakPtrFactory<NonBlockingInvalidationNotifier> weak_ptr_factory_;
+
+  // Our observers (which must live on the parent thread).
+  ObserverList<SyncNotifierObserver> observers_;
+
+  // The real guts of NonBlockingInvalidationNotifier, which allows
+  // this class to live completely on the parent thread.
   scoped_refptr<Core> core_;
   scoped_refptr<base::MessageLoopProxy> parent_message_loop_proxy_;
   scoped_refptr<base::MessageLoopProxy> io_message_loop_proxy_;
+
   DISALLOW_COPY_AND_ASSIGN(NonBlockingInvalidationNotifier);
 };
 
diff --git a/chrome/browser/sync/notifier/non_blocking_invalidation_notifier_unittest.cc b/chrome/browser/sync/notifier/non_blocking_invalidation_notifier_unittest.cc
index 2dc621a..d401742 100644
--- a/chrome/browser/sync/notifier/non_blocking_invalidation_notifier_unittest.cc
+++ b/chrome/browser/sync/notifier/non_blocking_invalidation_notifier_unittest.cc
@@ -66,18 +66,29 @@ class NonBlockingInvalidationNotifierTest : public testing::Test {
 };
 
 TEST_F(NonBlockingInvalidationNotifierTest, Basic) {
-  syncable::ModelTypeSet types;
-  types.insert(syncable::BOOKMARKS);
-  types.insert(syncable::AUTOFILL);
+  InSequence dummy;
+
+  syncable::ModelTypePayloadMap type_payloads;
+  type_payloads[syncable::PREFERENCES] = "payload";
+  type_payloads[syncable::BOOKMARKS] = "";
+  type_payloads[syncable::AUTOFILL] = "";
+
+  EXPECT_CALL(mock_observer_, OnNotificationStateChange(true));
+  EXPECT_CALL(mock_observer_, StoreState("new_fake_state"));
+  EXPECT_CALL(mock_observer_, OnIncomingNotification(type_payloads));
+  EXPECT_CALL(mock_observer_, OnNotificationStateChange(false));
 
-  invalidation_notifier_->SetUniqueId("fake_id");
   invalidation_notifier_->SetState("fake_state");
+  invalidation_notifier_->SetUniqueId("fake_id");
   invalidation_notifier_->UpdateCredentials("foo@bar.com", "fake_token");
-  invalidation_notifier_->UpdateEnabledTypes(types);
-}
 
-// TODO(akalin): Add synchronous operations for testing to
-// NonBlockingInvalidationNotifierTest and use that to test it.
+  invalidation_notifier_->OnNotificationStateChange(true);
+  invalidation_notifier_->StoreState("new_fake_state");
+  invalidation_notifier_->OnIncomingNotification(type_payloads);
+  invalidation_notifier_->OnNotificationStateChange(false);
+
+  ui_loop_.RunAllPending();
+}
 
 }  // namespace
 
diff --git a/chrome/browser/sync/sessions/sync_session_unittest.cc b/chrome/browser/sync/sessions/sync_session_unittest.cc
index 8c46802..5ee4ac6 100644
--- a/chrome/browser/sync/sessions/sync_session_unittest.cc
+++ b/chrome/browser/sync/sessions/sync_session_unittest.cc
@@ -7,6 +7,7 @@
 #include "base/compiler_specific.h"
 #include "base/location.h"
 #include "base/memory/ref_counted.h"
+#include "base/message_loop.h"
 #include "chrome/browser/sync/engine/conflict_resolver.h"
 #include "chrome/browser/sync/engine/mock_model_safe_workers.h"
 #include "chrome/browser/sync/engine/syncer_types.h"
@@ -101,6 +102,7 @@ class SyncSessionTest : public testing::Test,
     return request_params;
   }
 
+  MessageLoop message_loop_;
   bool controller_invocations_allowed_;
   scoped_ptr<SyncSession> session_;
   scoped_ptr<SyncSessionContext> context_;
diff --git a/chrome/browser/sync/syncable/directory_manager.cc b/chrome/browser/sync/syncable/directory_manager.cc
index b98aacb..c207b3c 100644
--- a/chrome/browser/sync/syncable/directory_manager.cc
+++ b/chrome/browser/sync/syncable/directory_manager.cc
@@ -42,19 +42,26 @@ DirectoryManager::~DirectoryManager() {
       << "Dir " << managed_directory_->name() << " not closed!";
 }
 
-bool DirectoryManager::Open(const std::string& name,
-                            DirectoryChangeDelegate* delegate) {
+bool DirectoryManager::Open(
+    const std::string& name,
+    DirectoryChangeDelegate* delegate,
+    const browser_sync::WeakHandle<TransactionObserver>&
+        transaction_observer) {
   bool was_open = false;
   const DirOpenResult result =
-      OpenImpl(name, GetSyncDataDatabasePath(), delegate, &was_open);
+      OpenImpl(name, GetSyncDataDatabasePath(), delegate,
+               transaction_observer, &was_open);
   return syncable::OPENED == result;
 }
 
 // Opens a directory.  Returns false on error.
-DirOpenResult DirectoryManager::OpenImpl(const std::string& name,
-                                         const FilePath& path,
-                                         DirectoryChangeDelegate* delegate,
-                                         bool* was_open) {
+DirOpenResult DirectoryManager::OpenImpl(
+    const std::string& name,
+    const FilePath& path,
+    DirectoryChangeDelegate* delegate,
+    const browser_sync::WeakHandle<TransactionObserver>&
+        transaction_observer,
+    bool* was_open) {
   bool opened = false;
   {
     base::AutoLock lock(lock_);
@@ -72,7 +79,8 @@ DirOpenResult DirectoryManager::OpenImpl(const std::string& name,
   // Otherwise, open it.
 
   scoped_ptr<Directory> dir(new Directory);
-  const DirOpenResult result = dir->Open(path, name, delegate);
+  const DirOpenResult result =
+      dir->Open(path, name, delegate, transaction_observer);
   if (syncable::OPENED == result) {
     base::AutoLock lock(lock_);
     managed_directory_ = dir.release();
diff --git a/chrome/browser/sync/syncable/directory_manager.h b/chrome/browser/sync/syncable/directory_manager.h
index 4a7316f..243000b 100644
--- a/chrome/browser/sync/syncable/directory_manager.h
+++ b/chrome/browser/sync/syncable/directory_manager.h
@@ -22,6 +22,7 @@
 #include "chrome/browser/sync/syncable/dir_open_result.h"
 #include "chrome/browser/sync/syncable/syncable.h"
 #include "chrome/browser/sync/util/cryptographer.h"
+#include "chrome/browser/sync/util/weak_handle.h"
 
 namespace sync_api { class BaseTransaction; }
 namespace syncable { class BaseTransaction; }
@@ -45,8 +46,10 @@ class DirectoryManager {
   // common case.  Does not take ownership of |delegate|, which must
   // be non-NULL.  Starts sending events to |delegate| if the returned
   // result is true.  Note that events to |delegate| may be sent from
-  // *any* thread.
-  bool Open(const std::string& name, DirectoryChangeDelegate* delegate);
+  // *any* thread.  |transaction_observer| must be initialized.
+  bool Open(const std::string& name, DirectoryChangeDelegate* delegate,
+            const browser_sync::WeakHandle<TransactionObserver>&
+                transaction_observer);
 
   // Marks a directory as closed and stops sending events to the
   // delegate.  It might take a while until all the file handles and
@@ -74,8 +77,12 @@ class DirectoryManager {
     return cryptographer_.get();
   }
 
-  DirOpenResult OpenImpl(const std::string& name, const FilePath& path,
-                         DirectoryChangeDelegate* delegate, bool* was_open);
+  DirOpenResult OpenImpl(
+      const std::string& name, const FilePath& path,
+      DirectoryChangeDelegate* delegate,
+      const browser_sync::WeakHandle<TransactionObserver>&
+          transaction_observer,
+      bool* was_open);
 
   // Helpers for friend class ScopedDirLookup:
   friend class ScopedDirLookup;
diff --git a/chrome/browser/sync/syncable/syncable.cc b/chrome/browser/sync/syncable/syncable.cc
index 747e6bc..b1548f0 100644
--- a/chrome/browser/sync/syncable/syncable.cc
+++ b/chrome/browser/sync/syncable/syncable.cc
@@ -349,10 +349,14 @@ DictionaryValue* EntryKernel::ToValue() const {
 ///////////////////////////////////////////////////////////////////////////
 // Directory
 
-void Directory::InitKernel(const std::string& name,
-                           DirectoryChangeDelegate* delegate) {
-  DCHECK(kernel_ == NULL);
-  kernel_ = new Kernel(FilePath(), name, KernelLoadInfo(), delegate);
+void Directory::InitKernelForTest(
+    const std::string& name,
+    DirectoryChangeDelegate* delegate,
+    const browser_sync::WeakHandle<TransactionObserver>&
+        transaction_observer) {
+  DCHECK(!kernel_);
+  kernel_ =  new Kernel(FilePath(), name, KernelLoadInfo(),
+                        delegate, transaction_observer);
 }
 
 Directory::PersistedKernelInfo::PersistedKernelInfo()
@@ -378,10 +382,11 @@ Directory::SaveChangesSnapshot::SaveChangesSnapshot()
 
 Directory::SaveChangesSnapshot::~SaveChangesSnapshot() {}
 
-Directory::Kernel::Kernel(const FilePath& db_path,
-                          const string& name,
-                          const KernelLoadInfo& info,
-                          DirectoryChangeDelegate* delegate)
+Directory::Kernel::Kernel(
+    const FilePath& db_path, const string& name,
+    const KernelLoadInfo& info, DirectoryChangeDelegate* delegate,
+    const browser_sync::WeakHandle<TransactionObserver>&
+        transaction_observer)
     : db_path(db_path),
       refcount(1),
       next_write_transaction_id(0),
@@ -399,8 +404,9 @@ Directory::Kernel::Kernel(const FilePath& db_path,
       cache_guid(info.cache_guid),
       next_metahandle(info.max_metahandle + 1),
       delegate(delegate),
-      observers(new ObserverListThreadSafe<TransactionObserver>()) {
+      transaction_observer(transaction_observer) {
   DCHECK(delegate);
+  DCHECK(transaction_observer.IsInitialized());
 }
 
 void Directory::Kernel::AddRef() {
@@ -432,9 +438,13 @@ Directory::~Directory() {
   Close();
 }
 
-DirOpenResult Directory::Open(const FilePath& file_path, const string& name,
-                              DirectoryChangeDelegate* delegate) {
-  const DirOpenResult result = OpenImpl(file_path, name, delegate);
+DirOpenResult Directory::Open(
+    const FilePath& file_path, const string& name,
+    DirectoryChangeDelegate* delegate,
+    const browser_sync::WeakHandle<TransactionObserver>&
+        transaction_observer) {
+  const DirOpenResult result =
+      OpenImpl(file_path, name, delegate, transaction_observer);
   if (OPENED != result)
     Close();
   return result;
@@ -461,9 +471,12 @@ DirectoryBackingStore* Directory::CreateBackingStore(
   return new DirectoryBackingStore(dir_name, backing_filepath);
 }
 
-DirOpenResult Directory::OpenImpl(const FilePath& file_path,
-                                  const string& name,
-                                  DirectoryChangeDelegate* delegate) {
+DirOpenResult Directory::OpenImpl(
+    const FilePath& file_path,
+    const string& name,
+    DirectoryChangeDelegate* delegate,
+    const browser_sync::WeakHandle<TransactionObserver>&
+        transaction_observer) {
   DCHECK_EQ(static_cast<DirectoryBackingStore*>(NULL), store_);
   FilePath db_path(file_path);
   file_util::AbsolutePath(&db_path);
@@ -477,7 +490,7 @@ DirOpenResult Directory::OpenImpl(const FilePath& file_path,
   if (OPENED != result)
     return result;
 
-  kernel_ = new Kernel(db_path, name, info, delegate);
+  kernel_ = new Kernel(db_path, name, info, delegate, transaction_observer);
   kernel_->metahandles_index->swap(metas_bucket);
   InitializeIndices();
   return OPENED;
@@ -1109,14 +1122,6 @@ void Directory::CheckTreeInvariants(syncable::BaseTransaction* trans,
   }
 }
 
-void Directory::AddTransactionObserver(TransactionObserver* observer) {
-  kernel_->observers->AddObserver(observer);
-}
-
-void Directory::RemoveTransactionObserver(TransactionObserver* observer) {
-  kernel_->observers->RemoveObserver(observer);
-}
-
 ///////////////////////////////////////////////////////////////////////////////
 // ScopedKernelLock
 
@@ -1153,13 +1158,13 @@ BaseTransaction::BaseTransaction(const tracked_objects::Location& from_here,
                                  Directory* directory)
     : from_here_(from_here), name_(name), writer_(writer),
       directory_(directory), dirkernel_(directory->kernel_) {
-  dirkernel_->observers->Notify(
+  dirkernel_->transaction_observer.Call(FROM_HERE,
       &TransactionObserver::OnTransactionStart, from_here_, writer_);
 }
 
 BaseTransaction::~BaseTransaction() {
   if (writer_ != INVALID) {
-    dirkernel_->observers->Notify(
+    dirkernel_->transaction_observer.Call(FROM_HERE,
         &TransactionObserver::OnTransactionEnd, from_here_, writer_);
   }
 }
@@ -1268,7 +1273,7 @@ ModelTypeBitSet WriteTransaction::NotifyTransactionChangingAndEnding(
       delegate->HandleTransactionEndingChangeEvent(
           immutable_write_transaction_info, this);
 
-  dirkernel_->observers->Notify(
+  dirkernel_->transaction_observer.Call(FROM_HERE,
       &TransactionObserver::OnTransactionWrite,
       immutable_write_transaction_info, models_with_changes);
 
diff --git a/chrome/browser/sync/syncable/syncable.h b/chrome/browser/sync/syncable/syncable.h
index 0f6ecee..54fe4c0 100644
--- a/chrome/browser/sync/syncable/syncable.h
+++ b/chrome/browser/sync/syncable/syncable.h
@@ -24,7 +24,6 @@
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/memory/ref_counted.h"
-#include "base/observer_list_threadsafe.h"
 #include "base/synchronization/lock.h"
 #include "base/time.h"
 #include "chrome/browser/sync/protocol/sync.pb.h"
@@ -35,6 +34,7 @@
 #include "chrome/browser/sync/syncable/model_type.h"
 #include "chrome/browser/sync/util/immutable.h"
 #include "chrome/browser/sync/util/time.h"
+#include "chrome/browser/sync/util/weak_handle.h"
 
 namespace base {
 class DictionaryValue;
@@ -821,11 +821,14 @@ class Directory {
   // Does not take ownership of |delegate|, which must not be NULL.
   // Starts sending events to |delegate| if the returned result is
   // OPENED.  Note that events to |delegate| may be sent from *any*
-  // thread.
+  // thread.  |transaction_observer| must be initialized.
   DirOpenResult Open(const FilePath& file_path, const std::string& name,
-                     DirectoryChangeDelegate* delegate);
+                     DirectoryChangeDelegate* delegate,
+                     const browser_sync::WeakHandle<TransactionObserver>&
+                         transaction_observer);
 
-  // Stops sending events to the delegate.
+  // Stops sending events to the delegate and the transaction
+  // observer.
   void Close();
 
   int64 NextMetahandle();
@@ -866,12 +869,6 @@ class Directory {
   // Unique to each account / client pair.
   std::string cache_guid() const;
 
-  // These are backed by a thread-safe observer list, and so can be
-  // called on any thread, and events will be sent to the observer on
-  // the same thread that it was added on.
-  void AddTransactionObserver(TransactionObserver* observer);
-  void RemoveTransactionObserver(TransactionObserver* observer);
-
  protected:  // for friends, mainly used by Entry constructors
   virtual EntryKernel* GetEntryByHandle(int64 handle);
   virtual EntryKernel* GetEntryByHandle(int64 metahandle,
@@ -900,8 +897,11 @@ class Directory {
   // before calling.
   EntryKernel* GetEntryById(const Id& id, ScopedKernelLock* const lock);
 
-  DirOpenResult OpenImpl(const FilePath& file_path, const std::string& name,
-                         DirectoryChangeDelegate* delegate);
+  DirOpenResult OpenImpl(
+      const FilePath& file_path, const std::string& name,
+      DirectoryChangeDelegate* delegate,
+      const browser_sync::WeakHandle<TransactionObserver>&
+          transaction_observer);
 
   template <class T> void TestAndSet(T* kernel_data, const T* data_to_set);
 
@@ -1049,14 +1049,22 @@ class Directory {
   typedef Index<ClientTagIndexer>::Set ClientTagIndex;
 
  protected:
-  // Used by tests.
-  void InitKernel(const std::string& name, DirectoryChangeDelegate* delegate);
+  // Used by tests. |delegate| must not be NULL.
+  // |transaction_observer| must be initialized.
+  void InitKernelForTest(
+      const std::string& name,
+      DirectoryChangeDelegate* delegate,
+      const browser_sync::WeakHandle<TransactionObserver>&
+          transaction_observer);
 
  private:
   struct Kernel {
-    // |delegate| can be NULL.
+    // |delegate| must not be NULL.  |transaction_observer| must be
+    // initialized.
     Kernel(const FilePath& db_path, const std::string& name,
-           const KernelLoadInfo& info, DirectoryChangeDelegate* delegate);
+           const KernelLoadInfo& info, DirectoryChangeDelegate* delegate,
+           const browser_sync::WeakHandle<TransactionObserver>&
+               transaction_observer);
 
     ~Kernel();
 
@@ -1125,11 +1133,11 @@ class Directory {
     // The next metahandle is protected by kernel mutex.
     int64 next_metahandle;
 
-    // The delegate for directory change events.  Can be NULL.
+    // The delegate for directory change events.  Must not be NULL.
     DirectoryChangeDelegate* const delegate;
 
-    // The transaction observers.
-    scoped_refptr<ObserverListThreadSafe<TransactionObserver> > observers;
+    // The transaction observer.
+    const browser_sync::WeakHandle<TransactionObserver> transaction_observer;
   };
 
   // Helper method used to do searches on |parent_id_child_index|.
diff --git a/chrome/browser/sync/syncable/syncable_mock.cc b/chrome/browser/sync/syncable/syncable_mock.cc
index 701c7bf..b21decf 100644
--- a/chrome/browser/sync/syncable/syncable_mock.cc
+++ b/chrome/browser/sync/syncable/syncable_mock.cc
@@ -5,9 +5,10 @@
 #include "chrome/browser/sync/syncable/syncable_mock.h"
 
 #include "base/location.h"
+#include "chrome/browser/sync/test/null_transaction_observer.h"
 
 MockDirectory::MockDirectory() {
-  InitKernel("myk", &delegate_);
+  InitKernelForTest("myk", &delegate_, syncable::NullTransactionObserver());
 }
 
 MockDirectory::~MockDirectory() {}
diff --git a/chrome/browser/sync/syncable/syncable_unittest.cc b/chrome/browser/sync/syncable/syncable_unittest.cc
index 6ce650d..1b6fc38 100644
--- a/chrome/browser/sync/syncable/syncable_unittest.cc
+++ b/chrome/browser/sync/syncable/syncable_unittest.cc
@@ -4,28 +4,15 @@
 
 #include "chrome/browser/sync/syncable/syncable.h"
 
-#include "build/build_config.h"
-
-#include <sys/types.h>
-
-#include <limits>
 #include <string>
 
-#if !defined(OS_WIN)
-#define MAX_PATH PATH_MAX
-#include <ostream>
-#include <stdio.h>
-#include <sys/ipc.h>
-#include <sys/sem.h>
-#include <sys/times.h>
-#endif  // !defined(OS_WIN)
-
 #include "base/compiler_specific.h"
 #include "base/file_path.h"
 #include "base/file_util.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/memory/scoped_ptr.h"
+#include "base/message_loop.h"
 #include "base/scoped_temp_dir.h"
 #include "base/stringprintf.h"
 #include "base/synchronization/condition_variable.h"
@@ -39,6 +26,7 @@
 #include "chrome/browser/sync/test/engine/test_id_factory.h"
 #include "chrome/browser/sync/test/engine/test_syncable_utils.h"
 #include "chrome/browser/sync/test/null_directory_change_delegate.h"
+#include "chrome/browser/sync/test/null_transaction_observer.h"
 #include "chrome/test/base/values_test_util.h"
 #include "testing/gtest/include/gtest/gtest.h"
 #include "third_party/sqlite/sqlite3.h"
@@ -103,6 +91,7 @@ class SyncableGeneralTest : public testing::Test {
   virtual void TearDown() {
   }
  protected:
+  MessageLoop message_loop_;
   ScopedTempDir temp_dir_;
   NullDirectoryChangeDelegate delegate_;
   FilePath db_path_;
@@ -110,7 +99,7 @@ class SyncableGeneralTest : public testing::Test {
 
 TEST_F(SyncableGeneralTest, General) {
   Directory dir;
-  dir.Open(db_path_, "SimpleTest", &delegate_);
+  dir.Open(db_path_, "SimpleTest", &delegate_, NullTransactionObserver());
 
   int64 root_metahandle;
   {
@@ -209,7 +198,7 @@ TEST_F(SyncableGeneralTest, General) {
 
 TEST_F(SyncableGeneralTest, ChildrenOps) {
   Directory dir;
-  dir.Open(db_path_, "SimpleTest", &delegate_);
+  dir.Open(db_path_, "SimpleTest", &delegate_, NullTransactionObserver());
 
   int64 root_metahandle;
   {
@@ -289,7 +278,7 @@ TEST_F(SyncableGeneralTest, ClientIndexRebuildsProperly) {
   // Test creating a new meta entry.
   {
     Directory dir;
-    dir.Open(db_path_, "IndexTest", &delegate_);
+    dir.Open(db_path_, "IndexTest", &delegate_, NullTransactionObserver());
     {
       WriteTransaction wtrans(FROM_HERE, UNITTEST, &dir);
       MutableEntry me(&wtrans, CREATE, wtrans.root_id(), name);
@@ -305,7 +294,7 @@ TEST_F(SyncableGeneralTest, ClientIndexRebuildsProperly) {
   // The DB was closed. Now reopen it. This will cause index regeneration.
   {
     Directory dir;
-    dir.Open(db_path_, "IndexTest", &delegate_);
+    dir.Open(db_path_, "IndexTest", &delegate_, NullTransactionObserver());
 
     ReadTransaction trans(FROM_HERE, &dir);
     Entry me(&trans, GET_BY_CLIENT_TAG, tag);
@@ -325,7 +314,7 @@ TEST_F(SyncableGeneralTest, ClientIndexRebuildsDeletedProperly) {
   // Test creating a deleted, unsynced, server meta entry.
   {
     Directory dir;
-    dir.Open(db_path_, "IndexTest", &delegate_);
+    dir.Open(db_path_, "IndexTest", &delegate_, NullTransactionObserver());
     {
       WriteTransaction wtrans(FROM_HERE, UNITTEST, &dir);
       MutableEntry me(&wtrans, CREATE, wtrans.root_id(), "deleted");
@@ -343,7 +332,7 @@ TEST_F(SyncableGeneralTest, ClientIndexRebuildsDeletedProperly) {
   // Should still be present and valid in the client tag index.
   {
     Directory dir;
-    dir.Open(db_path_, "IndexTest", &delegate_);
+    dir.Open(db_path_, "IndexTest", &delegate_, NullTransactionObserver());
 
     ReadTransaction trans(FROM_HERE, &dir);
     Entry me(&trans, GET_BY_CLIENT_TAG, tag);
@@ -357,7 +346,7 @@ TEST_F(SyncableGeneralTest, ClientIndexRebuildsDeletedProperly) {
 
 TEST_F(SyncableGeneralTest, ToValue) {
   Directory dir;
-  dir.Open(db_path_, "SimpleTest", &delegate_);
+  dir.Open(db_path_, "SimpleTest", &delegate_, NullTransactionObserver());
 
   const Id id = TestIdFactory::FromNumber(99);
   {
@@ -412,6 +401,7 @@ class TestUnsaveableDirectory : public Directory {
 // Test suite for syncable::Directory.
 class SyncableDirectoryTest : public testing::Test {
  protected:
+  MessageLoop message_loop_;
   ScopedTempDir temp_dir_;
   static const char kName[];
   static const Id kId;
@@ -425,7 +415,8 @@ class SyncableDirectoryTest : public testing::Test {
     file_util::Delete(file_path_, true);
     dir_.reset(new Directory());
     ASSERT_TRUE(dir_.get());
-    ASSERT_TRUE(OPENED == dir_->Open(file_path_, kName, &delegate_));
+    ASSERT_EQ(OPENED, dir_->Open(file_path_, kName,
+                                 &delegate_, NullTransactionObserver()));
     ASSERT_TRUE(dir_->good());
   }
 
@@ -439,7 +430,8 @@ class SyncableDirectoryTest : public testing::Test {
   void ReloadDir() {
     dir_.reset(new Directory());
     ASSERT_TRUE(dir_.get());
-    ASSERT_TRUE(OPENED == dir_->Open(file_path_, kName, &delegate_));
+    ASSERT_EQ(OPENED, dir_->Open(file_path_, kName,
+                                 &delegate_, NullTransactionObserver()));
   }
 
   void SaveAndReloadDir() {
@@ -1202,7 +1194,8 @@ TEST_F(SyncableDirectoryTest, TestSimpleFieldsPreservedDuringSaveChanges) {
   dir_->SaveChanges();
   dir_.reset(new Directory());
   ASSERT_TRUE(dir_.get());
-  ASSERT_TRUE(OPENED == dir_->Open(file_path_, kName, &delegate_));
+  ASSERT_EQ(OPENED, dir_->Open(file_path_, kName,
+                               &delegate_, NullTransactionObserver()));
   ASSERT_TRUE(dir_->good());
 
   {
@@ -1301,7 +1294,8 @@ TEST_F(SyncableDirectoryTest, TestSaveChangesFailure) {
   // always fail.
   dir_.reset(new TestUnsaveableDirectory());
   ASSERT_TRUE(dir_.get());
-  ASSERT_TRUE(OPENED == dir_->Open(file_path_, kName, &delegate_));
+  ASSERT_EQ(OPENED, dir_->Open(file_path_, kName,
+                               &delegate_, NullTransactionObserver()));
   ASSERT_TRUE(dir_->good());
   int64 handle2 = 0;
   {
@@ -1374,7 +1368,8 @@ TEST_F(SyncableDirectoryTest, TestSaveChangesFailureWithPurge) {
   // always fail.
   dir_.reset(new TestUnsaveableDirectory());
   ASSERT_TRUE(dir_.get());
-  ASSERT_TRUE(OPENED == dir_->Open(file_path_, kName, &delegate_));
+  ASSERT_EQ(OPENED, dir_->Open(file_path_, kName,
+                               &delegate_, NullTransactionObserver()));
   ASSERT_TRUE(dir_->good());
 
   ModelTypeSet set;
@@ -1496,13 +1491,14 @@ class SyncableDirectoryManager : public testing::Test {
   virtual void TearDown() {
   }
  protected:
+  MessageLoop message_loop_;
   ScopedTempDir temp_dir_;
   NullDirectoryChangeDelegate delegate_;
 };
 
 TEST_F(SyncableDirectoryManager, TestFileRelease) {
   DirectoryManager dm(FilePath(temp_dir_.path()));
-  ASSERT_TRUE(dm.Open("ScopeTest", &delegate_));
+  ASSERT_TRUE(dm.Open("ScopeTest", &delegate_, NullTransactionObserver()));
   {
     ScopedDirLookup(&dm, "ScopeTest");
   }
@@ -1520,7 +1516,8 @@ class ThreadOpenTestDelegate : public base::PlatformThread::Delegate {
  private:
   // PlatformThread::Delegate methods:
   virtual void ThreadMain() {
-    CHECK(directory_manager_->Open("Open", &delegate_));
+    CHECK(directory_manager_->Open(
+        "Open", &delegate_, NullTransactionObserver()));
   }
 
   DISALLOW_COPY_AND_ASSIGN(ThreadOpenTestDelegate);
@@ -1564,6 +1561,7 @@ class ThreadBugDelegate : public base::PlatformThread::Delegate {
 
   // PlatformThread::Delegate methods:
   virtual void ThreadMain() {
+    MessageLoop message_loop;
     const std::string dirname = "ThreadBug1";
     base::AutoLock scoped_lock(step_->mutex);
 
@@ -1573,12 +1571,14 @@ class ThreadBugDelegate : public base::PlatformThread::Delegate {
       }
       switch (step_->number) {
       case 0:
-        directory_manager_->Open(dirname, &delegate_);
+        directory_manager_->Open(
+            dirname, &delegate_, NullTransactionObserver());
         break;
       case 1:
         {
           directory_manager_->Close(dirname);
-          directory_manager_->Open(dirname, &delegate_);
+          directory_manager_->Open(
+              dirname, &delegate_, NullTransactionObserver());
           ScopedDirLookup dir(directory_manager_, dirname);
           CHECK(dir.good());
           WriteTransaction trans(FROM_HERE, UNITTEST, dir);
@@ -1636,6 +1636,7 @@ class DirectoryKernelStalenessBugDelegate : public ThreadBugDelegate {
     : ThreadBugDelegate(role, step, dirman) {}
 
   virtual void ThreadMain() {
+    MessageLoop message_loop;
     const char test_bytes[] = "test data";
     const std::string dirname = "DirectoryKernelStalenessBug";
     base::AutoLock scoped_lock(step_->mutex);
@@ -1652,7 +1653,8 @@ class DirectoryKernelStalenessBugDelegate : public ThreadBugDelegate {
           file_util::Delete(directory_manager_->GetSyncDataDatabasePath(),
                             true);
           // Test.
-          directory_manager_->Open(dirname, &delegate_);
+          directory_manager_->Open(
+              dirname, &delegate_, NullTransactionObserver());
           ScopedDirLookup dir(directory_manager_, dirname);
           CHECK(dir.good());
           WriteTransaction trans(FROM_HERE, UNITTEST, dir);
@@ -1671,7 +1673,8 @@ class DirectoryKernelStalenessBugDelegate : public ThreadBugDelegate {
         break;
       case 1:
         {
-          directory_manager_->Open(dirname, &delegate_);
+          directory_manager_->Open(
+              dirname, &delegate_, NullTransactionObserver());
           ScopedDirLookup dir(directory_manager_, dirname);
           CHECK(dir.good());
         }
@@ -1770,13 +1773,14 @@ class StressTransactionsDelegate : public base::PlatformThread::Delegate {
 };
 
 TEST(SyncableDirectory, StressTransactions) {
+  MessageLoop message_loop;
   ScopedTempDir temp_dir;
   ASSERT_TRUE(temp_dir.CreateUniqueTempDir());
   DirectoryManager dirman(FilePath(temp_dir.path()));
   std::string dirname = "stress";
   file_util::Delete(dirman.GetSyncDataDatabasePath(), true);
   NullDirectoryChangeDelegate delegate;
-  dirman.Open(dirname, &delegate);
+  dirman.Open(dirname, &delegate, NullTransactionObserver());
 
   const int kThreadCount = 7;
   base::PlatformThreadHandle threads[kThreadCount];
diff --git a/chrome/browser/sync/test/engine/syncer_command_test.h b/chrome/browser/sync/test/engine/syncer_command_test.h
index 55460b7..c6fccbb 100644
--- a/chrome/browser/sync/test/engine/syncer_command_test.h
+++ b/chrome/browser/sync/test/engine/syncer_command_test.h
@@ -11,6 +11,7 @@
 #include <vector>
 
 #include "base/compiler_specific.h"
+#include "base/message_loop.h"
 #include "chrome/browser/sync/engine/model_safe_worker.h"
 #include "chrome/browser/sync/sessions/debug_info_getter.h"
 #include "chrome/browser/sync/sessions/sync_session.h"
@@ -157,6 +158,7 @@ class SyncerCommandTestWithParam : public testing::TestWithParam<T>,
   }
 
  private:
+  MessageLoop message_loop_;
   scoped_ptr<TestDirectorySetterUpper> syncdb_;
   scoped_ptr<sessions::SyncSessionContext> context_;
   scoped_ptr<MockConnectionManager> mock_server_;
diff --git a/chrome/browser/sync/test/engine/test_directory_setter_upper.cc b/chrome/browser/sync/test/engine/test_directory_setter_upper.cc
index 4123540..7cdb2a4 100644
--- a/chrome/browser/sync/test/engine/test_directory_setter_upper.cc
+++ b/chrome/browser/sync/test/engine/test_directory_setter_upper.cc
@@ -10,9 +10,11 @@
 #include "base/string_util.h"
 #include "chrome/browser/sync/syncable/directory_manager.h"
 #include "chrome/browser/sync/syncable/syncable.h"
+#include "chrome/browser/sync/test/null_transaction_observer.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 using syncable::DirectoryManager;
+using syncable::NullTransactionObserver;
 using syncable::ReadTransaction;
 using syncable::ScopedDirLookup;
 
@@ -38,7 +40,7 @@ void TestDirectorySetterUpper::reset_directory_manager(DirectoryManager* d) {
 
 void TestDirectorySetterUpper::SetUp() {
   Init();
-  ASSERT_TRUE(manager()->Open(name(), &delegate_));
+  ASSERT_TRUE(manager()->Open(name(), &delegate_, NullTransactionObserver()));
 }
 
 void TestDirectorySetterUpper::TearDown() {
@@ -80,7 +82,8 @@ void ManuallyOpenedTestDirectorySetterUpper::SetUp() {
 }
 
 void ManuallyOpenedTestDirectorySetterUpper::Open() {
-  ASSERT_TRUE(manager()->Open(name(), &delegate_));
+  ASSERT_TRUE(
+      manager()->Open(name(), &delegate_, NullTransactionObserver()));
   was_opened_ = true;
 }
 
@@ -111,7 +114,7 @@ void TriggeredOpenTestDirectorySetterUpper::TearDown() {
 
 MockDirectorySetterUpper::MockDirectory::MockDirectory(
     const std::string& name) {
-  InitKernel(name, &delegate_);
+  InitKernelForTest(name, &delegate_, NullTransactionObserver());
 }
 
 MockDirectorySetterUpper::MockDirectory::~MockDirectory() {}
diff --git a/chrome/browser/sync/test/null_transaction_observer.cc b/chrome/browser/sync/test/null_transaction_observer.cc
new file mode 100644
index 0000000..4fe914f
--- /dev/null
+++ b/chrome/browser/sync/test/null_transaction_observer.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/test/null_transaction_observer.h"
+
+#include "base/memory/weak_ptr.h"
+
+namespace syncable {
+
+browser_sync::WeakHandle<TransactionObserver> NullTransactionObserver() {
+  return browser_sync::MakeWeakHandle(base::WeakPtr<TransactionObserver>());
+}
+
+}  // namespace syncable
diff --git a/chrome/browser/sync/test/null_transaction_observer.h b/chrome/browser/sync/test/null_transaction_observer.h
new file mode 100644
index 0000000..74aa9d7
--- /dev/null
+++ b/chrome/browser/sync/test/null_transaction_observer.h
@@ -0,0 +1,21 @@
+// 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_TEST_NULL_TRANSACTION_OBSERVER_H_
+#define CHROME_BROWSER_SYNC_TEST_NULL_TRANSACTION_OBSERVER_H_
+#pragma once
+
+#include "chrome/browser/sync/util/weak_handle.h"
+
+namespace syncable {
+
+class TransactionObserver;
+
+// Returns an initialized weak handle to a transaction observer that
+// does nothing.
+browser_sync::WeakHandle<TransactionObserver> NullTransactionObserver();
+
+}  // namespace syncable
+
+#endif  // CHROME_BROWSER_SYNC_TEST_NULL_TRANSACTION_OBSERVER_H_
diff --git a/chrome/chrome_tests.gypi b/chrome/chrome_tests.gypi
index a9cc2c7..5673b04 100644
--- a/chrome/chrome_tests.gypi
+++ b/chrome/chrome_tests.gypi
@@ -356,6 +356,8 @@
         'browser/sync/js/js_test_util.h',
         'browser/sync/test/null_directory_change_delegate.cc',
         'browser/sync/test/null_directory_change_delegate.h',
+        'browser/sync/test/null_transaction_observer.cc',
+        'browser/sync/test/null_transaction_observer.h',
         'browser/sync/test/engine/test_directory_setter_upper.cc',
         'browser/sync/test/engine/test_directory_setter_upper.h',
       ],