storage: Migrate IndexedDBContextImpl::Observer to mojo

Devtools was using this interface, and when IndexedDB moves to the
storage service, it will need to access it via mojo instead.
This creates a function for observing IndexedDB changes on the
IndexedDBControl interface and replaces users and tests.

The IndexedDBDispatcherHostTest tests are disabled and were both broken
(identically without this change).  They are updated to at least pass,
although enabling them will be done in a followup to avoid getting this
reverted.

Bug: 1015214
Change-Id: I455d06545e9bbc780173244846b7e801f6bae2cb
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2052768
Commit-Queue: enne <enne@chromium.org>
Reviewed-by: Kinuko Yasuda <kinuko@chromium.org>
Reviewed-by: Daniel Murphy <dmurph@chromium.org>
Cr-Commit-Position: refs/heads/master@{#741220}

components/services/storage/public/mojom/indexed_db_control.mojom

@@ -6,6 +6,7 @@ module storage.mojom;
 
 import "components/services/storage/public/mojom/indexed_db_control_test.mojom";
 import "mojo/public/mojom/base/file_path.mojom";
+import "mojo/public/mojom/base/string16.mojom";
 import "mojo/public/mojom/base/values.mojom";
 import "mojo/public/mojom/base/time.mojom";
 import "url/mojom/origin.mojom";
@@ -33,6 +34,19 @@ struct IndexedDBStorageUsageInfo {
   mojo_base.mojom.Time last_modified_time;
 };
 
+// Communicates with IndexedDB clients about changes in IndexedDB.
+interface IndexedDBObserver {
+  // This function is called when the size of the usage for a particular origin
+  // changes (both in memory and on disk).
+  OnIndexedDBListChanged(url.mojom.Origin origin);
+
+  // This function is called when the content of a particular object store
+  // for a particular origin has been modified.
+  OnIndexedDBContentChanged(url.mojom.Origin origin,
+                            mojo_base.mojom.String16 database_name,
+                            mojo_base.mojom.String16 object_store_name);
+};
+
 // Controls the state of IndexedDB within a partition. This is a privileged
 // interface and must not be brokered to untrusted clients.
 //
@@ -70,6 +84,9 @@ interface IndexedDBControl {
   // restarts can preserve that data.
   SetForceKeepSessionState();
 
+  // Adds an observer to be notified about modifications to IndexedDB.
+  AddObserver(pending_remote<IndexedDBObserver> observer);
+
   // Binds the testing interface for extra functionality only available in
   // tests.
   BindTestInterface(pending_receiver<IndexedDBControlTest> receiver);

content/browser/devtools/protocol/storage_handler.cc

@@ -172,70 +172,64 @@ class StorageHandler::CacheStorageObserver : CacheStorageContextImpl::Observer {
 // thread.
 // Registered on creation as an observer in IndexedDBContextImpl, unregistered
 // on destruction.
-class StorageHandler::IndexedDBObserver : IndexedDBContextImpl::Observer {
+class StorageHandler::IndexedDBObserver
+    : public storage::mojom::IndexedDBObserver {
  public:
-  IndexedDBObserver(base::WeakPtr<StorageHandler> owner_storage_handler,
-                    IndexedDBContextImpl* indexed_db_context)
-      : owner_(owner_storage_handler), context_(indexed_db_context) {
-    TaskRunner()->PostTask(
-        FROM_HERE, base::BindOnce(&IndexedDBObserver::AddObserverOnIDBThread,
-                                  base::Unretained(this)));
+  explicit IndexedDBObserver(
+      base::WeakPtr<StorageHandler> owner_storage_handler)
+      : owner_(owner_storage_handler), receiver_(this) {
+    DCHECK_CURRENTLY_ON(BrowserThread::UI);
+    if (!owner_storage_handler)
+      return;
+    auto& control =
+        owner_storage_handler->storage_partition_->GetIndexedDBControl();
+    mojo::PendingRemote<storage::mojom::IndexedDBObserver> remote;
+    receiver_.Bind(remote.InitWithNewPipeAndPassReceiver());
+    control.AddObserver(std::move(remote));
   }
 
-  ~IndexedDBObserver() override {
-    DCHECK(TaskRunner()->RunsTasksInCurrentSequence());
-    context_->RemoveObserver(this);
-  }
+  ~IndexedDBObserver() override { DCHECK_CURRENTLY_ON(BrowserThread::UI); }
 
-  void TrackOriginOnIDBThread(const url::Origin& origin) {
-    DCHECK(TaskRunner()->RunsTasksInCurrentSequence());
+  void TrackOrigin(const url::Origin& origin) {
+    DCHECK_CURRENTLY_ON(BrowserThread::UI);
     if (origins_.find(origin) != origins_.end())
       return;
     origins_.insert(origin);
   }
 
-  void UntrackOriginOnIDBThread(const url::Origin& origin) {
-    DCHECK(TaskRunner()->RunsTasksInCurrentSequence());
+  void UntrackOrigin(const url::Origin& origin) {
+    DCHECK_CURRENTLY_ON(BrowserThread::UI);
     origins_.erase(origin);
   }
 
   void OnIndexedDBListChanged(const url::Origin& origin) override {
+    DCHECK_CURRENTLY_ON(BrowserThread::UI);
+    if (!owner_)
+      return;
     auto found = origins_.find(origin);
     if (found == origins_.end())
       return;
-    base::PostTask(FROM_HERE, {BrowserThread::UI},
-                   base::BindOnce(&StorageHandler::NotifyIndexedDBListChanged,
-                                  owner_, origin.Serialize()));
+    owner_->NotifyIndexedDBListChanged(origin.Serialize());
   }
 
   void OnIndexedDBContentChanged(
       const url::Origin& origin,
       const base::string16& database_name,
       const base::string16& object_store_name) override {
+    DCHECK_CURRENTLY_ON(BrowserThread::UI);
+    if (!owner_)
+      return;
     auto found = origins_.find(origin);
     if (found == origins_.end())
       return;
-    base::PostTask(
-        FROM_HERE, {BrowserThread::UI},
-        base::BindOnce(&StorageHandler::NotifyIndexedDBContentChanged, owner_,
-                       origin.Serialize(), database_name, object_store_name));
-  }
-
-  base::SequencedTaskRunner* TaskRunner() const {
-    return context_->IDBTaskRunner();
+    owner_->NotifyIndexedDBContentChanged(origin.Serialize(), database_name,
+                                          object_store_name);
   }
 
  private:
-  void AddObserverOnIDBThread() {
-    DCHECK(TaskRunner()->RunsTasksInCurrentSequence());
-    context_->AddObserver(this);
-  }
-
-  // Maintained on the IDB thread to avoid thread contention.
   base::flat_set<url::Origin> origins_;
-
   base::WeakPtr<StorageHandler> owner_;
-  scoped_refptr<IndexedDBContextImpl> context_;
+  mojo::Receiver<storage::mojom::IndexedDBObserver> receiver_;
 
   DISALLOW_COPY_AND_ASSIGN(IndexedDBObserver);
 };
@@ -262,13 +256,7 @@ void StorageHandler::SetRenderer(int process_host_id,
 
 Response StorageHandler::Disable() {
   cache_storage_observer_.reset();
-  if (indexed_db_observer_) {
-    scoped_refptr<base::SequencedTaskRunner> observer_task_runner =
-        indexed_db_observer_->TaskRunner();
-    observer_task_runner->DeleteSoon(FROM_HERE,
-                                     std::move(indexed_db_observer_));
-  }
-
+  indexed_db_observer_.reset();
   return Response::OK();
 }
 
@@ -431,10 +419,7 @@ Response StorageHandler::TrackIndexedDBForOrigin(const std::string& origin) {
   if (!origin_url.is_valid())
     return Response::InvalidParams(origin + " is not a valid URL");
 
-  GetIndexedDBObserver()->TaskRunner()->PostTask(
-      FROM_HERE, base::BindOnce(&IndexedDBObserver::TrackOriginOnIDBThread,
-                                base::Unretained(GetIndexedDBObserver()),
-                                url::Origin::Create(origin_url)));
+  GetIndexedDBObserver()->TrackOrigin(url::Origin::Create(origin_url));
   return Response::OK();
 }
 
@@ -446,10 +431,7 @@ Response StorageHandler::UntrackIndexedDBForOrigin(const std::string& origin) {
   if (!origin_url.is_valid())
     return Response::InvalidParams(origin + " is not a valid URL");
 
-  GetIndexedDBObserver()->TaskRunner()->PostTask(
-      FROM_HERE, base::BindOnce(&IndexedDBObserver::UntrackOriginOnIDBThread,
-                                base::Unretained(GetIndexedDBObserver()),
-                                url::Origin::Create(origin_url)));
+  GetIndexedDBObserver()->UntrackOrigin(url::Origin::Create(origin_url));
   return Response::OK();
 }
 
@@ -468,10 +450,8 @@ StorageHandler::GetCacheStorageObserver() {
 StorageHandler::IndexedDBObserver* StorageHandler::GetIndexedDBObserver() {
   DCHECK_CURRENTLY_ON(BrowserThread::UI);
   if (!indexed_db_observer_) {
-    indexed_db_observer_ = std::make_unique<IndexedDBObserver>(
-        weak_ptr_factory_.GetWeakPtr(),
-        static_cast<IndexedDBContextImpl*>(
-            storage_partition_->GetIndexedDBContext()));
+    indexed_db_observer_ =
+        std::make_unique<IndexedDBObserver>(weak_ptr_factory_.GetWeakPtr());
   }
   return indexed_db_observer_.get();
 }

content/browser/indexed_db/indexed_db_context_impl.cc

@@ -424,6 +424,19 @@ void IndexedDBContextImpl::ResetCachesForTesting(base::OnceClosure callback) {
   std::move(callback).Run();
 }
 
+void IndexedDBContextImpl::AddObserver(
+    mojo::PendingRemote<storage::mojom::IndexedDBObserver> observer) {
+  IDBTaskRunner()->PostTask(
+      FROM_HERE,
+      base::BindOnce(
+          [](IndexedDBContextImpl* context,
+             mojo::PendingRemote<storage::mojom::IndexedDBObserver> observer) {
+            context->observers_.Add(std::move(observer));
+          },
+          // As |this| is destroyed on the IDBTaskRunner it is safe to post raw.
+          base::Unretained(this), std::move(observer)));
+}
+
 void IndexedDBContextImpl::ForceCloseSync(
     const Origin& origin,
     storage::mojom::ForceCloseReason reason) {
@@ -623,20 +636,9 @@ void IndexedDBContextImpl::BlobFilesCleaned(const url::Origin& origin) {
   QueryDiskAndUpdateQuotaUsage(origin);
 }
 
-void IndexedDBContextImpl::AddObserver(
-    IndexedDBContextImpl::Observer* observer) {
-  DCHECK(!observers_.HasObserver(observer));
-  observers_.AddObserver(observer);
-}
-
-void IndexedDBContextImpl::RemoveObserver(
-    IndexedDBContextImpl::Observer* observer) {
-  observers_.RemoveObserver(observer);
-}
-
 void IndexedDBContextImpl::NotifyIndexedDBListChanged(const Origin& origin) {
   for (auto& observer : observers_)
-    observer.OnIndexedDBListChanged(origin);
+    observer->OnIndexedDBListChanged(origin);
 }
 
 void IndexedDBContextImpl::NotifyIndexedDBContentChanged(
@@ -644,8 +646,8 @@ void IndexedDBContextImpl::NotifyIndexedDBContentChanged(
     const base::string16& database_name,
     const base::string16& object_store_name) {
   for (auto& observer : observers_) {
-    observer.OnIndexedDBContentChanged(origin, database_name,
-                                       object_store_name);
+    observer->OnIndexedDBContentChanged(origin, database_name,
+                                        object_store_name);
   }
 }
 

content/browser/indexed_db/indexed_db_context_impl.h

@@ -27,6 +27,7 @@
 #include "mojo/public/cpp/bindings/pending_remote.h"
 #include "mojo/public/cpp/bindings/receiver_set.h"
 #include "mojo/public/cpp/bindings/remote.h"
+#include "mojo/public/cpp/bindings/remote_set.h"
 #include "storage/browser/blob/mojom/blob_storage_context.mojom.h"
 #include "storage/browser/quota/quota_manager_proxy.h"
 #include "storage/browser/quota/special_storage_policy.h"
@@ -52,18 +53,6 @@ class CONTENT_EXPORT IndexedDBContextImpl
       public storage::mojom::IndexedDBControl,
       public storage::mojom::IndexedDBControlTest {
  public:
-  class Observer {
-   public:
-    virtual void OnIndexedDBListChanged(const url::Origin& origin) = 0;
-    virtual void OnIndexedDBContentChanged(
-        const url::Origin& origin,
-        const base::string16& database_name,
-        const base::string16& object_store_name) = 0;
-
-   protected:
-    virtual ~Observer() {}
-  };
-
   // The indexed db directory.
   static const base::FilePath::CharType kIndexedDBDirectory[];
 
@@ -105,6 +94,8 @@ class CONTENT_EXPORT IndexedDBContextImpl
   void GetFilePathForTesting(const url::Origin& origin,
                              GetFilePathForTestingCallback callback) override;
   void ResetCachesForTesting(base::OnceClosure callback) override;
+  void AddObserver(
+      mojo::PendingRemote<storage::mojom::IndexedDBObserver> observer) override;
 
   // TODO(enne): fix internal indexeddb callers to use ForceClose async instead.
   void ForceCloseSync(const url::Origin& origin,
@@ -173,10 +164,6 @@ class CONTENT_EXPORT IndexedDBContextImpl
                                        : nullptr;
   }
 
-  // Only callable on the IDB task runner.
-  void AddObserver(Observer* observer);
-  void RemoveObserver(Observer* observer);
-
   void NotifyIndexedDBListChanged(const url::Origin& origin);
   void NotifyIndexedDBContentChanged(const url::Origin& origin,
                                      const base::string16& database_name,
@@ -228,11 +215,11 @@ class CONTENT_EXPORT IndexedDBContextImpl
   scoped_refptr<base::SequencedTaskRunner> io_task_runner_;
   std::unique_ptr<std::set<url::Origin>> origin_set_;
   std::map<url::Origin, int64_t> origin_size_map_;
-  base::ObserverList<Observer>::Unchecked observers_;
   base::Clock* clock_;
 
   mojo::ReceiverSet<storage::mojom::IndexedDBControl> receivers_;
   mojo::ReceiverSet<storage::mojom::IndexedDBControlTest> test_receivers_;
+  mojo::RemoteSet<storage::mojom::IndexedDBObserver> observers_;
 
   DISALLOW_COPY_AND_ASSIGN(IndexedDBContextImpl);
 };

content/browser/indexed_db/indexed_db_dispatcher_host_unittest.cc

@@ -36,6 +36,7 @@
 #include "mojo/public/cpp/bindings/associated_remote.h"
 #include "mojo/public/cpp/bindings/pending_associated_remote.h"
 #include "mojo/public/cpp/bindings/pending_remote.h"
+#include "mojo/public/cpp/bindings/receiver.h"
 #include "mojo/public/cpp/bindings/remote.h"
 #include "storage/browser/test/mock_quota_manager.h"
 #include "storage/browser/test/mock_quota_manager_proxy.h"
@@ -145,8 +146,12 @@ void StatusCallback(base::OnceClosure callback,
   std::move(callback).Run();
 }
 
-class TestIndexedDBObserver : public IndexedDBContextImpl::Observer {
+class TestIndexedDBObserver : public storage::mojom::IndexedDBObserver {
  public:
+  explicit TestIndexedDBObserver(
+      mojo::PendingReceiver<storage::mojom::IndexedDBObserver> receiver)
+      : receiver_(this, std::move(receiver)) {}
+
   void OnIndexedDBListChanged(const url::Origin& origin) override {
     ++notify_list_changed_count;
   }
@@ -160,6 +165,9 @@ class TestIndexedDBObserver : public IndexedDBContextImpl::Observer {
 
   int notify_list_changed_count = 0;
   int notify_content_changed_count = 0;
+
+ private:
+  mojo::Receiver<storage::mojom::IndexedDBObserver> receiver_;
 };
 
 }  // namespace
@@ -1015,172 +1023,221 @@ TEST_F(IndexedDBDispatcherHostTest, DISABLED_NotifyIndexedDBListChanged) {
   const char kObjectStoreName[] = "os";
   const char kIndexName[] = "index";
 
-  TestIndexedDBObserver observer;
-  context_impl_->AddObserver(&observer);
+  mojo::PendingReceiver<storage::mojom::IndexedDBObserver> receiver;
+  mojo::PendingRemote<storage::mojom::IndexedDBObserver> remote;
+  TestIndexedDBObserver observer(remote.InitWithNewPipeAndPassReceiver());
+  context_impl_->AddObserver(std::move(remote));
 
   // Open connection 1.
-  TestDatabaseConnection connection1(
-      context_impl_->IDBTaskRunner(), ToOrigin(kOrigin),
-      base::UTF8ToUTF16(kDatabaseName), kDBVersion1, kTransactionId1);
+  std::unique_ptr<TestDatabaseConnection> connection1;
+
   IndexedDBDatabaseMetadata metadata1;
   mojo::PendingAssociatedRemote<blink::mojom::IDBDatabase> pending_database1;
   EXPECT_EQ(0, observer.notify_list_changed_count);
   {
     base::RunLoop loop;
-    EXPECT_CALL(
-        *connection1.open_callbacks,
-        MockedUpgradeNeeded(IsAssociatedInterfacePtrInfoValid(true),
-                            IndexedDBDatabaseMetadata::NO_VERSION,
-                            blink::mojom::IDBDataLoss::None, std::string(), _))
-        .WillOnce(testing::DoAll(MoveArg<0>(&pending_database1),
-                                 testing::SaveArg<4>(&metadata1),
-                                 QuitLoop(&loop)));
-
-    // Queue open request message.
-    connection1.Open(idb_mojo_factory_.get());
+    context_impl_->IDBTaskRunner()->PostTask(
+        FROM_HERE, base::BindLambdaForTesting([&]() {
+          connection1 = std::make_unique<TestDatabaseConnection>(
+              context_impl_->IDBTaskRunner(), ToOrigin(kOrigin),
+              base::UTF8ToUTF16(kDatabaseName), kDBVersion1, kTransactionId1);
+
+          EXPECT_CALL(*connection1->open_callbacks,
+                      MockedUpgradeNeeded(
+                          IsAssociatedInterfacePtrInfoValid(true),
+                          IndexedDBDatabaseMetadata::NO_VERSION,
+                          blink::mojom::IDBDataLoss::None, std::string(), _))
+              .WillOnce(testing::DoAll(MoveArg<0>(&pending_database1),
+                                       testing::SaveArg<4>(&metadata1),
+                                       QuitLoop(&loop)));
+
+          // Queue open request message.
+          connection1->Open(idb_mojo_factory_.get());
+        }));
     loop.Run();
   }
   EXPECT_TRUE(pending_database1.is_valid());
-  EXPECT_EQ(connection1.version, metadata1.version);
-  EXPECT_EQ(connection1.db_name, metadata1.name);
+  EXPECT_EQ(connection1->version, metadata1.version);
+  EXPECT_EQ(connection1->db_name, metadata1.name);
 
-  // Create object store and index.
-  connection1.database.Bind(std::move(pending_database1));
-  ASSERT_TRUE(connection1.database.is_bound());
-  ASSERT_TRUE(connection1.version_change_transaction.is_bound());
   {
     ::testing::InSequence dummy;
     base::RunLoop loop;
     base::RepeatingClosure quit_closure =
         base::BarrierClosure(2, loop.QuitClosure());
-
-    EXPECT_CALL(*connection1.connection_callbacks, Complete(kTransactionId1))
-        .Times(1)
-        .WillOnce(RunClosure(quit_closure));
-    EXPECT_CALL(
-        *connection1.open_callbacks,
-        MockedSuccessDatabase(IsAssociatedInterfacePtrInfoValid(false), _))
-        .Times(1)
-        .WillOnce(RunClosure(std::move(quit_closure)));
-
-    ASSERT_TRUE(connection1.database.is_bound());
-    connection1.version_change_transaction->CreateObjectStore(
-        kObjectStoreId, base::UTF8ToUTF16(kObjectStoreName),
-        blink::IndexedDBKeyPath(), false);
-    connection1.database->CreateIndex(kTransactionId1, kObjectStoreId, kIndexId,
-                                      base::UTF8ToUTF16(kIndexName),
-                                      blink::IndexedDBKeyPath(), false, false);
-    connection1.version_change_transaction->Commit(0);
+    context_impl_->IDBTaskRunner()->PostTask(
+        FROM_HERE, base::BindLambdaForTesting([&]() {
+          // Create object store and index.
+          connection1->database.Bind(std::move(pending_database1));
+          ASSERT_TRUE(connection1->database.is_bound());
+          ASSERT_TRUE(connection1->version_change_transaction.is_bound());
+
+          EXPECT_CALL(*connection1->connection_callbacks,
+                      Complete(kTransactionId1))
+              .Times(1)
+              .WillOnce(RunClosure(quit_closure));
+          EXPECT_CALL(*connection1->open_callbacks,
+                      MockedSuccessDatabase(
+                          IsAssociatedInterfacePtrInfoValid(false), _))
+              .Times(1)
+              .WillOnce(RunClosure(std::move(quit_closure)));
+
+          ASSERT_TRUE(connection1->database.is_bound());
+          connection1->version_change_transaction->CreateObjectStore(
+              kObjectStoreId, base::UTF8ToUTF16(kObjectStoreName),
+              blink::IndexedDBKeyPath(), false);
+          connection1->database->CreateIndex(
+              kTransactionId1, kObjectStoreId, kIndexId,
+              base::UTF8ToUTF16(kIndexName), blink::IndexedDBKeyPath(), false,
+              false);
+          connection1->version_change_transaction->Commit(0);
+        }));
     loop.Run();
   }
+
   EXPECT_EQ(2, observer.notify_list_changed_count);
-  connection1.database->Close();
 
   // Open connection 2.
-  TestDatabaseConnection connection2(
-      context_impl_->IDBTaskRunner(), url::Origin::Create(GURL(kOrigin)),
-      base::UTF8ToUTF16(kDatabaseName), kDBVersion2, kTransactionId2);
+  std::unique_ptr<TestDatabaseConnection> connection2;
+
   IndexedDBDatabaseMetadata metadata2;
   mojo::PendingAssociatedRemote<blink::mojom::IDBDatabase> pending_database2;
   {
     ::testing::InSequence dummy;
     base::RunLoop loop;
-    EXPECT_CALL(*connection2.open_callbacks,
-                MockedUpgradeNeeded(
-                    IsAssociatedInterfacePtrInfoValid(true), kDBVersion1,
-                    blink::mojom::IDBDataLoss::None, std::string(), _))
-        .WillOnce(testing::DoAll(MoveArg<0>(&pending_database2),
-                                 testing::SaveArg<4>(&metadata2),
-                                 QuitLoop(&loop)));
-
-    // Queue open request message.
-    connection2.Open(idb_mojo_factory_.get());
+    base::RepeatingClosure quit_closure =
+        base::BarrierClosure(2, loop.QuitClosure());
+    context_impl_->IDBTaskRunner()->PostTask(
+        FROM_HERE, base::BindLambdaForTesting([&]() {
+          connection1->database->Close();
+
+          connection2 = std::make_unique<TestDatabaseConnection>(
+              context_impl_->IDBTaskRunner(),
+              url::Origin::Create(GURL(kOrigin)),
+              base::UTF8ToUTF16(kDatabaseName), kDBVersion2, kTransactionId2);
+
+          EXPECT_CALL(*connection2->open_callbacks,
+                      MockedUpgradeNeeded(
+                          IsAssociatedInterfacePtrInfoValid(true), kDBVersion1,
+                          blink::mojom::IDBDataLoss::None, std::string(), _))
+              .WillOnce(testing::DoAll(MoveArg<0>(&pending_database2),
+                                       testing::SaveArg<4>(&metadata2),
+                                       QuitLoop(&loop)));
+
+          // Queue open request message.
+          connection2->Open(idb_mojo_factory_.get());
+        }));
     loop.Run();
   }
   EXPECT_TRUE(pending_database2.is_valid());
-  EXPECT_EQ(connection2.version, metadata2.version);
-  EXPECT_EQ(connection2.db_name, metadata2.name);
+  EXPECT_EQ(connection2->version, metadata2.version);
+  EXPECT_EQ(connection2->db_name, metadata2.name);
 
-  // Delete index.
-  connection2.database.Bind(std::move(pending_database2));
-  ASSERT_TRUE(connection2.database.is_bound());
-  ASSERT_TRUE(connection2.version_change_transaction.is_bound());
   {
     ::testing::InSequence dummy;
     base::RunLoop loop;
     base::RepeatingClosure quit_closure =
         base::BarrierClosure(2, loop.QuitClosure());
-
-    EXPECT_CALL(*connection2.connection_callbacks, Complete(kTransactionId2))
-        .Times(1)
-        .WillOnce(RunClosure(quit_closure));
-    EXPECT_CALL(
-        *connection2.open_callbacks,
-        MockedSuccessDatabase(IsAssociatedInterfacePtrInfoValid(false), _))
-        .Times(1)
-        .WillOnce(RunClosure(std::move(quit_closure)));
-
-    ASSERT_TRUE(connection2.database.is_bound());
-    connection2.database->DeleteIndex(kTransactionId2, kObjectStoreId,
-                                      kIndexId);
-    connection2.version_change_transaction->Commit(0);
+    context_impl_->IDBTaskRunner()->PostTask(
+        FROM_HERE, base::BindLambdaForTesting([&]() {
+          // Delete index.
+          connection2->database.Bind(std::move(pending_database2));
+          ASSERT_TRUE(connection2->database.is_bound());
+          ASSERT_TRUE(connection2->version_change_transaction.is_bound());
+
+          EXPECT_CALL(*connection2->connection_callbacks,
+                      Complete(kTransactionId2))
+              .Times(1)
+              .WillOnce(RunClosure(quit_closure));
+          EXPECT_CALL(*connection2->open_callbacks,
+                      MockedSuccessDatabase(
+                          IsAssociatedInterfacePtrInfoValid(false), _))
+              .Times(1)
+              .WillOnce(RunClosure(std::move(quit_closure)));
+
+          ASSERT_TRUE(connection2->database.is_bound());
+          connection2->database->DeleteIndex(kTransactionId2, kObjectStoreId,
+                                             kIndexId);
+          connection2->version_change_transaction->Commit(0);
+        }));
     loop.Run();
   }
   EXPECT_EQ(3, observer.notify_list_changed_count);
-  connection2.database->Close();
 
   // Open connection 3.
-  TestDatabaseConnection connection3(
-      context_impl_->IDBTaskRunner(), ToOrigin(kOrigin),
-      base::UTF8ToUTF16(kDatabaseName), kDBVersion3, kTransactionId3);
+  std::unique_ptr<TestDatabaseConnection> connection3;
+
   IndexedDBDatabaseMetadata metadata3;
   mojo::PendingAssociatedRemote<blink::mojom::IDBDatabase> pending_database3;
   {
     ::testing::InSequence dummy;
     base::RunLoop loop;
-    EXPECT_CALL(*connection3.open_callbacks,
-                MockedUpgradeNeeded(
-                    IsAssociatedInterfacePtrInfoValid(true), kDBVersion2,
-                    blink::mojom::IDBDataLoss::None, std::string(), _))
-        .WillOnce(testing::DoAll(MoveArg<0>(&pending_database3),
-                                 testing::SaveArg<4>(&metadata3),
-                                 QuitLoop(&loop)));
-
-    // Queue open request message.
-    connection3.Open(idb_mojo_factory_.get());
+    context_impl_->IDBTaskRunner()->PostTask(
+        FROM_HERE, base::BindLambdaForTesting([&]() {
+          connection2->database->Close();
+          connection3 = std::make_unique<TestDatabaseConnection>(
+              context_impl_->IDBTaskRunner(), ToOrigin(kOrigin),
+              base::UTF8ToUTF16(kDatabaseName), kDBVersion3, kTransactionId3);
+
+          EXPECT_CALL(*connection3->open_callbacks,
+                      MockedUpgradeNeeded(
+                          IsAssociatedInterfacePtrInfoValid(true), kDBVersion2,
+                          blink::mojom::IDBDataLoss::None, std::string(), _))
+              .WillOnce(testing::DoAll(MoveArg<0>(&pending_database3),
+                                       testing::SaveArg<4>(&metadata3),
+                                       QuitLoop(&loop)));
+
+          // Queue open request message.
+          connection3->Open(idb_mojo_factory_.get());
+        }));
     loop.Run();
   }
   EXPECT_TRUE(pending_database3.is_valid());
-  EXPECT_EQ(connection3.version, metadata3.version);
-  EXPECT_EQ(connection3.db_name, metadata3.name);
+  EXPECT_EQ(connection3->version, metadata3.version);
+  EXPECT_EQ(connection3->db_name, metadata3.name);
 
-  // Delete object store.
-  connection3.database.Bind(std::move(pending_database3));
-  ASSERT_TRUE(connection3.database.is_bound());
-  ASSERT_TRUE(connection3.version_change_transaction.is_bound());
   {
     ::testing::InSequence dummy;
     base::RunLoop loop;
     base::RepeatingClosure quit_closure =
         base::BarrierClosure(2, loop.QuitClosure());
-
-    EXPECT_CALL(*connection3.connection_callbacks, Complete(kTransactionId3))
-        .Times(1)
-        .WillOnce(RunClosure(quit_closure));
-    EXPECT_CALL(
-        *connection3.open_callbacks,
-        MockedSuccessDatabase(IsAssociatedInterfacePtrInfoValid(false), _))
-        .Times(1)
-        .WillOnce(RunClosure(std::move(quit_closure)));
-
-    ASSERT_TRUE(connection3.database.is_bound());
-    connection3.version_change_transaction->DeleteObjectStore(kObjectStoreId);
-    connection3.version_change_transaction->Commit(0);
+    context_impl_->IDBTaskRunner()->PostTask(
+        FROM_HERE, base::BindLambdaForTesting([&]() {
+          // Delete object store.
+          connection3->database.Bind(std::move(pending_database3));
+          ASSERT_TRUE(connection3->database.is_bound());
+          ASSERT_TRUE(connection3->version_change_transaction.is_bound());
+
+          EXPECT_CALL(*connection3->connection_callbacks,
+                      Complete(kTransactionId3))
+              .Times(1)
+              .WillOnce(RunClosure(quit_closure));
+          EXPECT_CALL(*connection3->open_callbacks,
+                      MockedSuccessDatabase(
+                          IsAssociatedInterfacePtrInfoValid(false), _))
+              .Times(1)
+              .WillOnce(RunClosure(std::move(quit_closure)));
+
+          ASSERT_TRUE(connection3->database.is_bound());
+          connection3->version_change_transaction->DeleteObjectStore(
+              kObjectStoreId);
+          connection3->version_change_transaction->Commit(0);
+        }));
     loop.Run();
   }
   EXPECT_EQ(4, observer.notify_list_changed_count);
 
-  context_impl_->RemoveObserver(&observer);
+  {
+    base::RunLoop loop;
+    context_impl_->IDBTaskRunner()->PostTask(FROM_HERE,
+                                             base::BindLambdaForTesting([&]() {
+                                               connection1.reset();
+                                               connection2.reset();
+                                               connection3.reset();
+                                               loop.Quit();
+                                             }));
+    loop.Run();
+  }
 }
 
 MATCHER(IsSuccessKey, "") {
@@ -1196,8 +1253,10 @@ TEST_F(IndexedDBDispatcherHostTest, DISABLED_NotifyIndexedDBContentChanged) {
   const int64_t kObjectStoreId = 10;
   const char kObjectStoreName[] = "os";
 
-  TestIndexedDBObserver observer;
-  context_impl_->AddObserver(&observer);
+  mojo::PendingReceiver<storage::mojom::IndexedDBObserver> receiver;
+  mojo::PendingRemote<storage::mojom::IndexedDBObserver> remote;
+  TestIndexedDBObserver observer(remote.InitWithNewPipeAndPassReceiver());
+  context_impl_->AddObserver(std::move(remote));
   EXPECT_EQ(0, observer.notify_list_changed_count);
   EXPECT_EQ(0, observer.notify_content_changed_count);
 
@@ -1357,7 +1416,8 @@ TEST_F(IndexedDBDispatcherHostTest, DISABLED_NotifyIndexedDBContentChanged) {
       }));
   loop5.Run();
 
-  EXPECT_EQ(3, observer.notify_list_changed_count);
+  // +2 list changed, one for the transaction, the other for the ~DatabaseImpl
+  EXPECT_EQ(4, observer.notify_list_changed_count);
   EXPECT_EQ(2, observer.notify_content_changed_count);
 
   base::RunLoop loop6;
@@ -1368,8 +1428,6 @@ TEST_F(IndexedDBDispatcherHostTest, DISABLED_NotifyIndexedDBContentChanged) {
                                              loop6.Quit();
                                            }));
   loop6.Run();
-
-  context_impl_->RemoveObserver(&observer);
 }
 
 }  // namespace content