[Nearby Connections] Check for shutdown before accessing ClientProxy.

During the destruction of NearbyConnections, Core (which owns ClientProxy) is destructed before ServiceController (which owns EndpointManager), which means any pending tasks on the EndpointManager executor that use ClientProxy will be using
garbage memory. To fix this issue, EndpointManager::DiscardEndpoint will check
for an `is_shutdown` boolean set during ~EndpointManager before accessing
ClientProxy. The assumption is that any accessing of ClientProxy after the
destruction will be invalid.

PiperOrigin-RevId: 532581122

connections/implementation/BUILD

@@ -107,13 +107,11 @@ cc_library(
         "//location/nearby/cpp/sharing/implementation:__pkg__",
     ],
     deps = [
-        ":message_lite",
         "//connections:core_types",
         "//connections/implementation/analytics",
         "//connections/implementation/flags:connections_flags",
         "//connections/implementation/mediums",
         "//connections/implementation/mediums:utils",
-        "//connections/implementation/mediums/webrtc",
         "//connections/implementation/proto:offline_wire_formats_cc_proto",
         "//connections/v3:v3_types",
         "//internal/analytics:event_logger",
@@ -232,6 +230,7 @@ cc_test(
         "//internal/platform:test_util",
         "//internal/platform:types",
         "//internal/platform/implementation/g3",  # build_cleaner: keep
+        "//internal/test",
         "//proto:connections_enums_cc_proto",
         "@com_github_protobuf_matchers//protobuf-matchers",
         "@com_google_absl//absl/container:flat_hash_set",

connections/implementation/endpoint_manager.cc

@@ -287,10 +287,16 @@ bool operator<(const EndpointManager::FrameProcessor& lhs,
 }
 
 EndpointManager::EndpointManager(EndpointChannelManager* manager)
-    : channel_manager_(manager) {}
+    : EndpointManager(manager, std::make_unique<SingleThreadExecutor>()) {}
+
+EndpointManager::EndpointManager(
+    EndpointChannelManager* manager,
+    std::unique_ptr<SingleThreadExecutor> serial_executor)
+    : channel_manager_(manager), serial_executor_(std::move(serial_executor)) {}
 
 EndpointManager::~EndpointManager() {
   NEARBY_LOG(INFO, "Initiating shutdown of EndpointManager.");
+  is_shutdown_ = true;
   analytics::ThroughputRecorderContainer::GetInstance().Shutdown();
   CountDownLatch latch(1);
   RunOnEndpointManagerThread("bring-down-endpoints", [this, &latch]() {
@@ -301,7 +307,7 @@ EndpointManager::~EndpointManager() {
   latch.Await();
 
   NEARBY_LOG(INFO, "Bringing down control thread");
-  serial_executor_.Shutdown();
+  serial_executor_->Shutdown();
   NEARBY_LOG(INFO, "EndpointManager is down");
 }
 
@@ -524,10 +530,48 @@ std::vector<std::string> EndpointManager::SendPayloadChunk(
 void EndpointManager::DiscardEndpoint(ClientProxy* client,
                                       const std::string& endpoint_id) {
   NEARBY_LOGS(VERBOSE) << "DiscardEndpoint for endpoint " << endpoint_id;
-  RunOnEndpointManagerThread("discard-endpoint", [this, client, endpoint_id]() {
-    RemoveEndpoint(client, endpoint_id,
-                   /*notify=*/client->IsConnectedToEndpoint(endpoint_id));
-  });
+  RunOnEndpointManagerThread(
+      "discard-endpoint", [this, client, endpoint_id]() {
+        // `ClientProxy` is destroyed before `EndpointManager` in
+        // `~NearbyConnections`, which means "discard-endpoint" needs to check
+        // if this task is being executing during `~EndpointManager` to
+        // prevent accessing an invalid `ClientProxy` pointer. There are two
+        // cases where "discard-endpoint" can be executed during destruction,
+        // both of which can safely use `is_shutdown_` to check if this is being
+        // executed during the destruction of the object:
+        //
+        // Case 1: "discard-endpoints" is posted to the thread before
+        // destruction, but not executed yet: `~EndpointManager` blocks on
+        // "bring-down-endpoints" and because the executor is a single thread
+        // executor, tasks are guaranteed to execute sequentially
+        // (see
+        // https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newSingleThreadExecutor--)
+        // and this means that the "discard-endpoints" will be executed before
+        // "bring-down-endpoints", blocking the destruction of `is_shutdown_`
+        // and therefore `is_shutdown_` is not garbage memory.
+        //
+        // Case 2: "discard-endpoints" is posted to the thread during
+        // destruction, after "bring-down-endpoints" is called: the executor
+        // will be destructed before `is_shutdown_` because of the ordering of
+        // `EndpointManager`'s member variables, and the executor's destructor
+        // blocks on running all pending tasks
+        // (see
+        // https://source.chromium.org/chromium/chromium/src/+/refs/heads/main:chrome/services/sharing/nearby/platform/scheduled_executor.cc;l=67;drc=e0e0d24aaa54727dc0a8bc4b159ccdf80d3f5d8d),
+        // which means that "discard-endpoints" will run during the destruction
+        // of `serial_executor_` and will still have access to a valid
+        // `is_shutdown_`.
+        //
+        // TODO(b/280653613): Develop a more robost solution to prevent
+        // accessing an already destroyed `ClientProxy` during destruction.
+        if (is_shutdown_) {
+          NEARBY_LOGS(VERBOSE)
+              << "DiscardEndpoint called during destruction, returning early.";
+          return;
+        }
+
+        RemoveEndpoint(client, endpoint_id,
+                       /*notify=*/client->IsConnectedToEndpoint(endpoint_id));
+      });
 }
 
 std::vector<std::string> EndpointManager::SendControlMessage(
@@ -698,7 +742,7 @@ void EndpointManager::EndpointState::StartEndpointKeepAliveManager(
 
 void EndpointManager::RunOnEndpointManagerThread(const std::string& name,
                                                  Runnable runnable) {
-  serial_executor_.Execute(name, std::move(runnable));
+  serial_executor_->Execute(name, std::move(runnable));
 }
 
 }  // namespace connections

connections/implementation/endpoint_manager.h

@@ -153,6 +153,11 @@ class EndpointManager {
   // blocked here.
   void DiscardEndpoint(ClientProxy* client, const std::string& endpoint_id);
 
+ protected:
+  // For unit tests only to control executing tasks on the executor.
+  EndpointManager(EndpointChannelManager* manager,
+                  std::unique_ptr<SingleThreadExecutor> serial_executor);
+
  private:
   class EndpointState {
    public:
@@ -288,7 +293,18 @@ class EndpointManager {
   // We keep track of all registered channel endpoints here.
   absl::flat_hash_map<std::string, EndpointState> endpoints_;
 
-  SingleThreadExecutor serial_executor_;
+  // Indicates whether the destructor has been called yet. If `is_shutdown_`
+  // is true, assume any `ClientProxy` pointers are invalid, and should not
+  // be used.
+  //
+  // The ordering of these objects is important: `serial_executor_` must be
+  // destroyed before `is_shutdown_` because `serial_executor_` runs all
+  // pending tasks during it's destruction, and the "discard-endpoints"
+  // task checks `is_shutdown_` to prevent accessing an invalid `ClientProxy`
+  // pointer.
+  bool is_shutdown_ = false;
+
+  std::unique_ptr<SingleThreadExecutor> serial_executor_;
 };
 
 // Operator overloads when comparing FrameProcessor*.

connections/implementation/endpoint_manager_test.cc

@@ -34,6 +34,7 @@
 #include "internal/platform/count_down_latch.h"
 #include "internal/platform/exception.h"
 #include "internal/platform/logging.h"
+#include "internal/test/fake_single_thread_executor.h"
 #include "proto/connections_enums.pb.h"
 
 namespace nearby {
@@ -111,6 +112,13 @@ class MockFrameProcessor : public EndpointManager::FrameProcessor {
               (override));
 };
 
+class TestEndpointManager : public EndpointManager {
+ public:
+  TestEndpointManager(EndpointChannelManager* manager,
+                      std::unique_ptr<SingleThreadExecutor> serial_executor)
+      : EndpointManager(manager, std::move(serial_executor)) {}
+};
+
 class EndpointManagerTest : public ::testing::Test {
  protected:
   void RegisterEndpoint(std::unique_ptr<MockEndpointChannel> channel,
@@ -127,14 +135,15 @@ class EndpointManagerTest : public ::testing::Test {
     EXPECT_CALL(*channel, GetLastWriteTimestamp())
         .WillRepeatedly(Return(start_time_));
     EXPECT_CALL(mock_listener_.initiated_cb, Call).Times(1);
-    em_.RegisterEndpoint(&client_, endpoint_id_, info_, connection_options_,
-                         std::move(channel), listener_, connection_token);
+    em_.RegisterEndpoint(client_.get(), endpoint_id_, info_,
+                         connection_options_, std::move(channel), listener_,
+                         connection_token);
     if (should_close) {
       EXPECT_TRUE(done.Await(absl::Milliseconds(1000)).result());
     }
   }
 
-  ClientProxy client_;
+  std::unique_ptr<ClientProxy> client_ = std::make_unique<ClientProxy>();
   ConnectionOptions connection_options_{
       .keep_alive_interval_millis = 5000,
       .keep_alive_timeout_millis = 30000,
@@ -195,7 +204,7 @@ TEST_F(EndpointManagerTest, UnregisterEndpointCallsOnDisconnected) {
   // (IMO, it should be called as long as any connection callback was called
   // before. (in this case initiated_cb is called)).
   // Test captures current protocol behavior.
-  em_.UnregisterEndpoint(&client_, endpoint_id_);
+  em_.UnregisterEndpoint(client_.get(), endpoint_id_);
 }
 
 TEST_F(EndpointManagerTest, RegisterFrameProcessorWorks) {
@@ -250,7 +259,7 @@ TEST_F(EndpointManagerTest, UnregisterFrameProcessorWorks) {
   processors_.emplace_back(std::move(connect_request));
   // Endpoint will not send OnDisconnect notification to frame processor.
   RegisterEndpoint(std::move(endpoint_channel), false);
-  em_.UnregisterEndpoint(&client_, endpoint_id_);
+  em_.UnregisterEndpoint(client_.get(), endpoint_id_);
 }
 
 TEST_F(EndpointManagerTest, SendControlMessageWorks) {
@@ -286,7 +295,7 @@ TEST_F(EndpointManagerTest, SendControlMessageWorks) {
       em_.SendControlMessage(header, control, std::vector{endpoint_id_});
   EXPECT_EQ(failed_ids, std::vector<std::string>{});
   NEARBY_LOG(INFO, "Will unregister endpoint now");
-  em_.UnregisterEndpoint(&client_, endpoint_id_);
+  em_.UnregisterEndpoint(client_.get(), endpoint_id_);
   NEARBY_LOG(INFO, "Will call destructors now");
 }
 
@@ -301,6 +310,48 @@ TEST_F(EndpointManagerTest, SingleReadOnInvalidPayload) {
   RegisterEndpoint(std::move(endpoint_channel));
 }
 
+// Regression test for b/278729669.
+//
+// During the destruction of NearbyConnections, Core (which owns ClientProxy)
+// is destructed before ServiceController (which owns EndpointManager), which
+// means any pending tasks on the EndpointManager than use ClientProxy will
+// be using garbage memory, and cause crashes. This test enforces the fix.
+TEST_F(EndpointManagerTest, DisconnectEndpointDuringDestruction) {
+  // This test uses a `FakeSingleThreadExecutor` in order to control when
+  // tasks are executed in order to simulate the scenario where
+  // `DiscardEndpoint` is posted to the executor before the EndpointManager
+  // is destructed, and executed during it's destruction.
+  std::unique_ptr<SingleThreadExecutor> serial_executor =
+      std::make_unique<FakeSingleThreadExecutor>();
+  FakeSingleThreadExecutor* fake_serial_executor =
+      static_cast<FakeSingleThreadExecutor*>(serial_executor.get());
+  std::unique_ptr<EndpointManager> endpoint_manager =
+      std::make_unique<TestEndpointManager>(&ecm_, std::move(serial_executor));
+
+  // DiscardEndpoint posts a task to the executor to run "discard-endpoint",
+  // however the `FakeSingleThreadExecutor` will not run this task
+  // immediately.
+  fake_serial_executor->SetRunExecutablesImmediately(
+      /*run_executables_immediately=*/false);
+  endpoint_manager->DiscardEndpoint(client_.get(), endpoint_id_);
+
+  // Simulate Core destruction of ClientProxy by destroying `client_`.
+  client_.reset();
+
+  // Simulate ServiceController destruction of EndpointManager by destroying
+  // `endpoint_manager`, and set the `FakeSingleThreadExecutor` to run
+  // executables on calls `Execute`. When `endpoint_manager` is destructed, it
+  // will block on calls to `Execute` to run all pending executables, notably
+  // "discard-endpoint" from above. However, "discard-endpoint" will have a
+  // reference to a destroyed ClientProxy.
+  //
+  // Expect no crash when "discard-endpoints" is executed during the
+  // destruction.
+  fake_serial_executor->SetRunExecutablesImmediately(
+      /*run_executables_immediately=*/true);
+  endpoint_manager.reset();
+}
+
 }  // namespace
 }  // namespace connections
 }  // namespace nearby

internal/platform/single_thread_executor.h

@@ -24,7 +24,7 @@ namespace nearby {
 // queue.
 //
 // https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newSingleThreadExecutor--
-class ABSL_LOCKABLE SingleThreadExecutor final : public SubmittableExecutor {
+class ABSL_LOCKABLE SingleThreadExecutor : public SubmittableExecutor {
  public:
   using Platform = api::ImplementationPlatform;
   SingleThreadExecutor()

internal/platform/submittable_executor.h

@@ -17,6 +17,7 @@
 
 #include <cstddef>
 #include <memory>
+#include <string>
 #include <utility>
 
 #include "absl/base/thread_annotations.h"
@@ -53,7 +54,7 @@ class ABSL_LOCKABLE SubmittableExecutor : public api::SubmittableExecutor,
     }
     return *this;
   }
-  void Execute(const std::string& name, Runnable&& runnable)
+  virtual void Execute(const std::string& name, Runnable&& runnable)
       ABSL_LOCKS_EXCLUDED(mutex_) {
     MutexLock lock(&mutex_);
     if (impl_)

internal/test/BUILD

@@ -18,12 +18,14 @@ cc_library(
     name = "test",
     srcs = [
         "fake_clock.cc",
+        "fake_single_thread_executor.cc",
         "fake_task_runner.cc",
         "fake_timer.cc",
     ],
     hdrs = [
         "fake_clock.h",
         "fake_device_info.h",
+        "fake_single_thread_executor.h",
         "fake_task_runner.h",
         "fake_timer.h",
     ],

internal/test/fake_single_thread_executor.cc

@@ -0,0 +1,48 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "internal/test/fake_single_thread_executor.h"
+
+#include <string>
+#include <utility>
+#include <vector>
+
+namespace nearby {
+
+FakeSingleThreadExecutor::FakeSingleThreadExecutor() = default;
+
+FakeSingleThreadExecutor::~FakeSingleThreadExecutor() { DoShutdown(); }
+
+void FakeSingleThreadExecutor::Execute(const std::string& name,
+                                       Runnable&& runnable) {
+  runnables_.push_back(std::make_pair(name, std::move(runnable)));
+
+  if (!run_executables_immediately_) return;
+
+  RunAllExecutables();
+}
+
+void FakeSingleThreadExecutor::RunAllExecutables() {
+  // Because `SingleThreadExecutor` ensures sequencing, run all pending
+  // executables in order they were added to the vector.
+  for (auto& runnable_pair : runnables_) {
+    runnable_pair.second();
+  }
+
+  runnables_.clear();
+}
+
+void FakeSingleThreadExecutor::DoShutdown() { RunAllExecutables(); }
+
+}  // namespace nearby

internal/test/fake_single_thread_executor.h

@@ -0,0 +1,52 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef PLATFORM_PUBLIC_FAKE_SINGLE_THREAD_EXECUTOR_H_
+#define PLATFORM_PUBLIC_FAKE_SINGLE_THREAD_EXECUTOR_H_
+
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "absl/base/thread_annotations.h"
+#include "internal/platform/single_thread_executor.h"
+
+namespace nearby {
+
+class ABSL_LOCKABLE FakeSingleThreadExecutor final
+    : public SingleThreadExecutor {
+ public:
+  FakeSingleThreadExecutor();
+  ~FakeSingleThreadExecutor() override;
+  FakeSingleThreadExecutor(FakeSingleThreadExecutor&&) = default;
+  FakeSingleThreadExecutor& operator=(FakeSingleThreadExecutor&&) = default;
+
+  void Execute(const std::string& name, Runnable&& runnable) override;
+
+  void SetRunExecutablesImmediately(bool run_executables_immediately) {
+    run_executables_immediately_ = run_executables_immediately;
+  }
+
+  void RunAllExecutables();
+
+ private:
+  void DoShutdown();
+
+  bool run_executables_immediately_ = false;
+  std::vector<std::pair<std::string, Runnable>> runnables_;
+};
+
+}  // namespace nearby
+
+#endif  // PLATFORM_PUBLIC_FAKE_SINGLE_THREAD_EXECUTOR_H_