[M119] Reland: Fix IPC Channel pipe teardown

This is a reland with the new test temporarily disabled on Android
until it can run without disrupting other tests.

(cherry picked from commit cd4c1f1)

Fixed: 1494461
Change-Id: If1d83c2dce62020f78dd50abc460973759002a1a
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/5015115
Commit-Queue: Ken Rockot <rockot@google.com>
Reviewed-by: Robert Sesek <rsesek@chromium.org>
Cr-Original-Commit-Position: refs/heads/main@{#1221953}
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/5038080
Auto-Submit: Ken Rockot <rockot@google.com>
Commit-Queue: Daniel Cheng <dcheng@chromium.org>
Reviewed-by: Daniel Cheng <dcheng@chromium.org>
Cr-Commit-Position: refs/branch-heads/6045@{#1383}
Cr-Branched-From: 905e8bd-refs/heads/main@{#1204232}

ipc/ipc_mojo_bootstrap.cc

@@ -793,13 +793,12 @@ class ChannelAssociatedGroupController
         // handle.
         DCHECK(!endpoint->client());
         DCHECK(endpoint->peer_closed());
-        MarkClosedAndMaybeRemove(endpoint);
+        MarkClosed(endpoint);
       } else {
-        MarkPeerClosedAndMaybeRemove(endpoint);
+        MarkPeerClosed(endpoint);
       }
     }
-
-    DCHECK(endpoints_.empty());
+    endpoints_.clear();
 
     GetMemoryDumpProvider().RemoveController(this);
   }
@@ -844,15 +843,19 @@ class ChannelAssociatedGroupController
     base::AutoLock locker(lock_);
     encountered_error_ = true;
 
+    std::vector<uint32_t> endpoints_to_remove;
     std::vector<scoped_refptr<Endpoint>> endpoints_to_notify;
     for (auto iter = endpoints_.begin(); iter != endpoints_.end();) {
       Endpoint* endpoint = iter->second.get();
       ++iter;
 
-      if (endpoint->client())
+      if (endpoint->client()) {
         endpoints_to_notify.push_back(endpoint);
+      }
 
-      MarkPeerClosedAndMaybeRemove(endpoint);
+      if (MarkPeerClosed(endpoint)) {
+        endpoints_to_remove.push_back(endpoint->id());
+      }
     }
 
     for (auto& endpoint : endpoints_to_notify) {
@@ -861,6 +864,10 @@ class ChannelAssociatedGroupController
       if (endpoint->client())
         NotifyEndpointOfError(endpoint.get(), false /* force_async */);
     }
+
+    for (uint32_t id : endpoints_to_remove) {
+      endpoints_.erase(id);
+    }
   }
 
   void NotifyEndpointOfError(Endpoint* endpoint, bool force_async) {
@@ -899,19 +906,33 @@ class ChannelAssociatedGroupController
     NotifyEndpointOfError(endpoint, false /* force_async */);
   }
 
-  void MarkClosedAndMaybeRemove(Endpoint* endpoint) {
+  // Marks `endpoint` as closed and returns true if and only if its peer was
+  // also already closed.
+  bool MarkClosed(Endpoint* endpoint) {
     lock_.AssertAcquired();
     endpoint->set_closed();
-    if (endpoint->closed() && endpoint->peer_closed())
-      endpoints_.erase(endpoint->id());
+    return endpoint->peer_closed();
   }
 
-  void MarkPeerClosedAndMaybeRemove(Endpoint* endpoint) {
+  // Marks `endpoint` as having a closed peer and returns true if and only if
+  // `endpoint` itself was also already closed.
+  bool MarkPeerClosed(Endpoint* endpoint) {
     lock_.AssertAcquired();
     endpoint->set_peer_closed();
     endpoint->SignalSyncMessageEvent();
-    if (endpoint->closed() && endpoint->peer_closed())
+    return endpoint->closed();
+  }
+
+  void MarkClosedAndMaybeRemove(Endpoint* endpoint) {
+    if (MarkClosed(endpoint)) {
       endpoints_.erase(endpoint->id());
+    }
+  }
+
+  void MarkPeerClosedAndMaybeRemove(Endpoint* endpoint) {
+    if (MarkPeerClosed(endpoint)) {
+      endpoints_.erase(endpoint->id());
+    }
   }
 
   Endpoint* FindOrInsertEndpoint(mojo::InterfaceId id, bool* inserted) {