diff --git a/cpp/core/internal/BUILD b/cpp/core/internal/BUILD
index ff017980..88b74ceb 100644
--- a/cpp/core/internal/BUILD
+++ b/cpp/core/internal/BUILD
@@ -185,6 +185,7 @@ cc_test(
         ":internal_test",
         "//core:core_types",
         "//core/internal/mediums",
+        "//core/internal/mediums:utils",
         "//proto/connections:offline_wire_formats_portable_proto",
         "//platform/base",
         "//platform/base:test_util",
diff --git a/cpp/core/internal/base_endpoint_channel_test.cc b/cpp/core/internal/base_endpoint_channel_test.cc
index e71aebcf..80668a44 100644
--- a/cpp/core/internal/base_endpoint_channel_test.cc
+++ b/cpp/core/internal/base_endpoint_channel_test.cc
@@ -107,9 +107,10 @@ DoDhKeyExchange(BaseEndpointChannel* channel_a,
       {
           .on_success_cb =
               [&latch, &context_a](
-                  const string& endpoint_id,
+                  const std::string& endpoint_id,
                   std::unique_ptr<securegcm::UKey2Handshake> ukey2,
-                  const string& auth_token, const ByteArray& raw_auth_token) {
+                  const std::string& auth_token,
+                  const ByteArray& raw_auth_token) {
                 NEARBY_LOG(INFO, "client-A side key negotiation done");
                 EXPECT_TRUE(ukey2->VerifyHandshake());
                 auto context = ukey2->ToConnectionContext();
@@ -118,7 +119,8 @@ DoDhKeyExchange(BaseEndpointChannel* channel_a,
                 latch.CountDown();
               },
           .on_failure_cb =
-              [&latch](const string& endpoint_id, EndpointChannel* channel) {
+              [&latch](const std::string& endpoint_id,
+                       EndpointChannel* channel) {
                 NEARBY_LOG(INFO, "client-A side key negotiation failed");
                 latch.CountDown();
               },
@@ -128,9 +130,10 @@ DoDhKeyExchange(BaseEndpointChannel* channel_a,
       {
           .on_success_cb =
               [&latch, &context_b](
-                  const string& endpoint_id,
+                  const std::string& endpoint_id,
                   std::unique_ptr<securegcm::UKey2Handshake> ukey2,
-                  const string& auth_token, const ByteArray& raw_auth_token) {
+                  const std::string& auth_token,
+                  const ByteArray& raw_auth_token) {
                 NEARBY_LOG(INFO, "client-B side key negotiation done");
                 EXPECT_TRUE(ukey2->VerifyHandshake());
                 auto context = ukey2->ToConnectionContext();
@@ -139,7 +142,8 @@ DoDhKeyExchange(BaseEndpointChannel* channel_a,
                 latch.CountDown();
               },
           .on_failure_cb =
-              [&latch](const string& endpoint_id, EndpointChannel* channel) {
+              [&latch](const std::string& endpoint_id,
+                       EndpointChannel* channel) {
                 NEARBY_LOG(INFO, "client-B side key negotiation failed");
                 latch.CountDown();
               },
diff --git a/cpp/core/internal/base_pcp_handler_test.cc b/cpp/core/internal/base_pcp_handler_test.cc
index 24034e9b..2433916b 100644
--- a/cpp/core/internal/base_pcp_handler_test.cc
+++ b/cpp/core/internal/base_pcp_handler_test.cc
@@ -29,6 +29,7 @@
 #include "proto/connections/offline_wire_formats.pb.h"
 #include "platform/base/byte_array.h"
 #include "platform/base/exception.h"
+#include "platform/base/medium_environment.h"
 #include "platform/public/count_down_latch.h"
 #include "platform/public/pipe.h"
 #include "proto/connections_enums.pb.h"
@@ -130,7 +131,7 @@ class MockPcpHandler : public BasePcpHandler {
               (override));
   MOCK_METHOD(Status, StopAdvertisingImpl, (ClientProxy * client), (override));
   MOCK_METHOD(StartOperationResult, StartDiscoveryImpl,
-              (ClientProxy * client, const string& service_id,
+              (ClientProxy * client, const std::string& service_id,
                const ConnectionOptions& options),
               (override));
   MOCK_METHOD(Status, StopDiscoveryImpl, (ClientProxy * client), (override));
@@ -399,9 +400,11 @@ class BasePcpHandlerTest
       .endpoint_distance_changed_cb =
           mock_discovery_listener_.endpoint_distance_changed_cb.AsStdFunction(),
   };
+  MediumEnvironment& env_ = MediumEnvironment::Instance();
 };
 
 TEST_P(BasePcpHandlerTest, ConstructorDestructorWorks) {
+  env_.Start();
   Mediums m;
   EndpointChannelManager ecm;
   EndpointManager em(&ecm);
@@ -409,9 +412,11 @@ TEST_P(BasePcpHandlerTest, ConstructorDestructorWorks) {
   MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
   SUCCEED();
   bwu.Shutdown();
+  env_.Stop();
 }
 
 TEST_P(BasePcpHandlerTest, StartAdvertisingChangesState) {
+  env_.Start();
   ClientProxy client;
   Mediums m;
   EndpointChannelManager ecm;
@@ -420,9 +425,11 @@ TEST_P(BasePcpHandlerTest, StartAdvertisingChangesState) {
   MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
   StartAdvertising(&client, &pcp_handler);
   bwu.Shutdown();
+  env_.Stop();
 }
 
 TEST_P(BasePcpHandlerTest, StopAdvertisingChangesState) {
+  env_.Start();
   ClientProxy client;
   Mediums m;
   EndpointChannelManager ecm;
@@ -435,9 +442,11 @@ TEST_P(BasePcpHandlerTest, StopAdvertisingChangesState) {
   pcp_handler.StopAdvertising(&client);
   EXPECT_FALSE(client.IsAdvertising());
   bwu.Shutdown();
+  env_.Stop();
 }
 
 TEST_P(BasePcpHandlerTest, StartDiscoveryChangesState) {
+  env_.Start();
   ClientProxy client;
   Mediums m;
   EndpointChannelManager ecm;
@@ -446,9 +455,11 @@ TEST_P(BasePcpHandlerTest, StartDiscoveryChangesState) {
   MockPcpHandler pcp_handler(&m, &em, &ecm, &bwu);
   StartDiscovery(&client, &pcp_handler);
   bwu.Shutdown();
+  env_.Stop();
 }
 
 TEST_P(BasePcpHandlerTest, StopDiscoveryChangesState) {
+  env_.Start();
   ClientProxy client;
   Mediums m;
   EndpointChannelManager ecm;
@@ -461,9 +472,11 @@ TEST_P(BasePcpHandlerTest, StopDiscoveryChangesState) {
   pcp_handler.StopDiscovery(&client);
   EXPECT_FALSE(client.IsDiscovering());
   bwu.Shutdown();
+  env_.Stop();
 }
 
 TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) {
+  env_.Start();
   std::string endpoint_id{"1234"};
   ClientProxy client;
   Mediums m;
@@ -486,9 +499,11 @@ TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) {
   channel_b->Close();
   bwu.Shutdown();
   pcp_handler.DisconnectFromEndpointManager();
+  env_.Stop();
 }
 
 TEST_P(BasePcpHandlerTest, IoError_RequestConnectionFails) {
+  env_.Start();
   std::string endpoint_id{"1234"};
   ClientProxy client;
   Mediums m;
@@ -513,9 +528,11 @@ TEST_P(BasePcpHandlerTest, IoError_RequestConnectionFails) {
   channel_b->Close();
   bwu.Shutdown();
   pcp_handler.DisconnectFromEndpointManager();
+  env_.Stop();
 }
 
 TEST_P(BasePcpHandlerTest, AcceptConnectionChangesState) {
+  env_.Start();
   std::string endpoint_id{"1234"};
   ClientProxy client;
   Mediums m;
@@ -542,9 +559,11 @@ TEST_P(BasePcpHandlerTest, AcceptConnectionChangesState) {
   channel_b->Close();
   bwu.Shutdown();
   pcp_handler.DisconnectFromEndpointManager();
+  env_.Stop();
 }
 
 TEST_P(BasePcpHandlerTest, RejectConnectionChangesState) {
+  env_.Start();
   std::string endpoint_id{"1234"};
   ClientProxy client;
   Mediums m;
@@ -567,9 +586,11 @@ TEST_P(BasePcpHandlerTest, RejectConnectionChangesState) {
   channel_b->Close();
   bwu.Shutdown();
   pcp_handler.DisconnectFromEndpointManager();
+  env_.Stop();
 }
 
 TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) {
+  env_.Start();
   std::string endpoint_id{"1234"};
   ClientProxy client;
   Mediums m;
@@ -602,9 +623,11 @@ TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) {
   channel_b->Close();
   bwu.Shutdown();
   pcp_handler.DisconnectFromEndpointManager();
+  env_.Stop();
 }
 
 TEST_P(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) {
+  env_.Start();
   std::atomic_int destroyed_flag = 0;
   int mediums_count = 0;
   {
@@ -638,9 +661,11 @@ TEST_P(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) {
     bwu.Shutdown();
   }
   EXPECT_EQ(destroyed_flag.load(), mediums_count);
+  env_.Stop();
 }
 
 TEST_P(BasePcpHandlerTest, MultipleMediumsProduceSingleEndpointLostEvent) {
+  env_.Start();
   BooleanMediumSelector allowed = GetParam();
   if (allowed.Count(true) < 2) {
     // Ignore single-medium test cases, and implicit "all mediums" case.
@@ -685,12 +710,14 @@ TEST_P(BasePcpHandlerTest, MultipleMediumsProduceSingleEndpointLostEvent) {
     pcp_handler.DisconnectFromEndpointManager();
   }
   EXPECT_EQ(destroyed_flag.load(), mediums_count);
+  env_.Stop();
 }
 
 INSTANTIATE_TEST_SUITE_P(ParameterizedBasePcpHandlerTest, BasePcpHandlerTest,
                          ::testing::ValuesIn(kTestCases));
 
 TEST_F(BasePcpHandlerTest, InjectEndpoint) {
+  env_.Start();
   std::string service_id{"service"};
   std::string endpoint_id{"ABCD"};
   ClientProxy client;
@@ -742,6 +769,7 @@ TEST_F(BasePcpHandlerTest, InjectEndpoint) {
           .remote_bluetooth_mac_address = ByteArray(kFakeMacAddress),
       });
   bwu.Shutdown();
+  env_.Stop();
 }
 
 }  // namespace
diff --git a/cpp/core/internal/bluetooth_device_name.cc b/cpp/core/internal/bluetooth_device_name.cc
index f968aff2..49e7838f 100644
--- a/cpp/core/internal/bluetooth_device_name.cc
+++ b/cpp/core/internal/bluetooth_device_name.cc
@@ -62,12 +62,7 @@ BluetoothDeviceName::BluetoothDeviceName(
     absl::string_view bluetooth_device_name_string) {
   ByteArray bluetooth_device_name_bytes =
       Base64Utils::Decode(bluetooth_device_name_string);
-
   if (bluetooth_device_name_bytes.Empty()) {
-    NEARBY_LOG(
-        INFO,
-        "Cannot deserialize BluetoothDeviceName: failed Base64 decoding of %s",
-        std::string(bluetooth_device_name_string).c_str());
     return;
   }
 
diff --git a/cpp/core/internal/bwu_manager.cc b/cpp/core/internal/bwu_manager.cc
index e2bbe186..cf2ad287 100644
--- a/cpp/core/internal/bwu_manager.cc
+++ b/cpp/core/internal/bwu_manager.cc
@@ -205,12 +205,18 @@ void BwuManager::OnIncomingFrame(OfflineFrame& frame,
   if (parser::GetFrameType(frame) != V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION)
     return;
   auto bwu_frame = frame.v1().bandwidth_upgrade_negotiation();
-  CountDownLatch latch(1);
-  RunOnBwuManagerThread([this, client, endpoint_id, &bwu_frame, &latch]() {
-    OnBwuNegotiationFrame(client, bwu_frame, endpoint_id);
-    latch.CountDown();
-  });
-  latch.Await();
+  if (FeatureFlags::GetInstance().GetFlags().enable_async_bandwidth_upgrade) {
+    RunOnBwuManagerThread([this, client, endpoint_id, bwu_frame]() {
+      OnBwuNegotiationFrame(client, bwu_frame, endpoint_id);
+    });
+  } else {
+    CountDownLatch latch(1);
+    RunOnBwuManagerThread([this, client, endpoint_id, bwu_frame, &latch]() {
+      OnBwuNegotiationFrame(client, bwu_frame, endpoint_id);
+      latch.CountDown();
+    });
+    latch.Await();
+  }
 }
 
 void BwuManager::OnEndpointDisconnect(ClientProxy* client,
@@ -274,7 +280,7 @@ void BwuManager::Revert() {
 }
 
 void BwuManager::OnBwuNegotiationFrame(ClientProxy* client,
-                                       const BwuNegotiationFrame& frame,
+                                       const BwuNegotiationFrame frame,
                                        const string& endpoint_id) {
   NEARBY_LOG(INFO, "OnBwuNegotiationFrame for endpoint %s",
              endpoint_id.c_str());
@@ -426,7 +432,7 @@ void BwuManager::ProcessBwuPathAvailableEvent(
         std::shared_ptr<EndpointChannel> previous_endpoint_channel =
             item.mapped();
         if (previous_endpoint_channel) {
-          previous_endpoint_channel->Close(DisconnectionReason::IO_ERROR);
+          previous_endpoint_channel->Close(DisconnectionReason::UNFINISHED);
         }
       }
       std::shared_ptr<EndpointChannel> new_channel =
@@ -435,7 +441,7 @@ void BwuManager::ProcessBwuPathAvailableEvent(
         // The upgraded channel never finished upgrading, and therefore is still
         // paused.
         new_channel->Resume();
-        new_channel->Close(DisconnectionReason::IO_ERROR);
+        new_channel->Close(DisconnectionReason::UNFINISHED);
       }
 
       return;
diff --git a/cpp/core/internal/bwu_manager.h b/cpp/core/internal/bwu_manager.h
index 5fcf54c7..f0916eec 100644
--- a/cpp/core/internal/bwu_manager.h
+++ b/cpp/core/internal/bwu_manager.h
@@ -114,7 +114,7 @@ class BwuManager : public EndpointManager::FrameProcessor {
   // Processes the BwuNegotiationFrames that come over the
   // EndpointChannel on both initiator and responder side of the upgrade.
   void OnBwuNegotiationFrame(ClientProxy* client,
-                             const BwuNegotiationFrame& frame,
+                             const BwuNegotiationFrame frame,
                              const string& endpoint_id);
 
   // Called to revert any state changed by the Initiator or Responder in the
diff --git a/cpp/core/internal/bwu_manager_test.cc b/cpp/core/internal/bwu_manager_test.cc
index 269c0812..52ab8278 100644
--- a/cpp/core/internal/bwu_manager_test.cc
+++ b/cpp/core/internal/bwu_manager_test.cc
@@ -20,6 +20,7 @@
 #include "core/internal/endpoint_channel_manager.h"
 #include "core/internal/endpoint_manager.h"
 #include "core/internal/mediums/mediums.h"
+#include "core/internal/mediums/utils.h"
 #include "platform/public/system_clock.h"
 #include "gmock/gmock.h"
 #include "gtest/gtest.h"
@@ -56,6 +57,23 @@ TEST(BwuManagerTest, CanInitiateBwu) {
   bwu_manager.Shutdown();
 }
 
+TEST(BwuManagerTest, CanProcessPathAvailableFrame) {
+  ClientProxy client;
+  std::string endpoint_id("EP_A");
+  Mediums mediums;
+  EndpointChannelManager ecm;
+  EndpointManager em{&ecm};
+  BwuManager bwu_manager{mediums, em, ecm, {}, {}};
+
+  LocationHint location_hint = Utils::BuildLocationHint("US");
+  ExceptionOr<OfflineFrame> wrapped_frame = parser::FromBytes(
+      parser::ForBwuWebrtcPathAvailable("my_id", location_hint));
+
+  bwu_manager.OnIncomingFrame(wrapped_frame.result(), endpoint_id, &client,
+                              Medium::WEB_RTC);
+  bwu_manager.Shutdown();
+}
+
 }  // namespace
 }  // namespace connections
 }  // namespace nearby
diff --git a/cpp/core/internal/encryption_runner_test.cc b/cpp/core/internal/encryption_runner_test.cc
index 3fdc0180..2dc3cfc4 100644
--- a/cpp/core/internal/encryption_runner_test.cc
+++ b/cpp/core/internal/encryption_runner_test.cc
@@ -102,15 +102,16 @@ TEST(EncryptionRunnerTest, ReadWrite) {
       &user_a.client, "endpoint_id", &user_a.channel,
       {
           .on_success_cb =
-              [&response](const string& endpoint_id,
+              [&response](const std::string& endpoint_id,
                           std::unique_ptr<securegcm::UKey2Handshake> ukey2,
-                          const string& auth_token,
+                          const std::string& auth_token,
                           const ByteArray& raw_auth_token) {
                 response.server_status = Response::Status::kDone;
                 response.latch.CountDown();
               },
           .on_failure_cb =
-              [&response](const string& endpoint_id, EndpointChannel* channel) {
+              [&response](const std::string& endpoint_id,
+                          EndpointChannel* channel) {
                 response.server_status = Response::Status::kFailed;
                 response.latch.CountDown();
               },
@@ -119,15 +120,16 @@ TEST(EncryptionRunnerTest, ReadWrite) {
       &user_b.client, "endpoint_id", &user_b.channel,
       {
           .on_success_cb =
-              [&response](const string& endpoint_id,
+              [&response](const std::string& endpoint_id,
                           std::unique_ptr<securegcm::UKey2Handshake> ukey2,
-                          const string& auth_token,
+                          const std::string& auth_token,
                           const ByteArray& raw_auth_token) {
                 response.client_status = Response::Status::kDone;
                 response.latch.CountDown();
               },
           .on_failure_cb =
-              [&response](const string& endpoint_id, EndpointChannel* channel) {
+              [&response](const std::string& endpoint_id,
+                          EndpointChannel* channel) {
                 response.client_status = Response::Status::kFailed;
                 response.latch.CountDown();
               },
diff --git a/cpp/core/internal/endpoint_manager.cc b/cpp/core/internal/endpoint_manager.cc
index 95616f3d..4f20786e 100644
--- a/cpp/core/internal/endpoint_manager.cc
+++ b/cpp/core/internal/endpoint_manager.cc
@@ -20,10 +20,10 @@
 #include "core/internal/endpoint_channel.h"
 #include "core/internal/offline_frames.h"
 #include "platform/base/exception.h"
+#include "platform/base/feature_flags.h"
 #include "platform/public/count_down_latch.h"
 #include "platform/public/logging.h"
 #include "proto/connections_enums.pb.h"
-
 namespace location {
 namespace nearby {
 namespace connections {
@@ -301,21 +301,19 @@ void EndpointManager::UnregisterFrameProcessor(V1Frame::FrameType frame_type,
 EndpointManager::FrameProcessor* EndpointManager::GetFrameProcessor(
     V1Frame::FrameType frame_type) {
   EndpointManager::FrameProcessor* processor = nullptr;
-  CountDownLatch latch(1);
-  RunOnEndpointManagerThread([this, frame_type, &processor, &latch]() {
-    auto it = frame_processors_.find(frame_type);
-    if (it != frame_processors_.end()) {
-      processor = it->second;
-    }
-    latch.CountDown();
-  });
-  latch.Await();
+  auto it = frame_processors_.find(frame_type);
+  if (it != frame_processors_.end()) {
+    processor = it->second;
+  }
   return processor;
 }
 
 void EndpointManager::EnsureWorkersTerminated(const std::string& endpoint_id) {
+  NEARBY_LOG(ERROR, "EnsureWorkersTerminated for endpoint %s",
+             endpoint_id.c_str());
   auto item = endpoints_.find(endpoint_id);
   if (item != endpoints_.end()) {
+    NEARBY_LOGS(INFO) << "EndpointState found for id: " << endpoint_id;
     // If another instance of data and keep-alive handlers is running, it will
     // terminate soon; we should block until it happens.
     EndpointState& endpoint_state = item->second;
@@ -346,13 +344,21 @@ void EndpointManager::RegisterEndpoint(ClientProxy* client,
   RunOnEndpointManagerThread([this, client, channel = channel.release(),
                               &endpoint_id, &info, &options, &listener,
                               &latch]() {
+    if (endpoints_.contains(endpoint_id)) {
+      NEARBY_LOG(WARNING, "Registing duplicate endpoint %s",
+                 endpoint_id.c_str());
+      if (!FeatureFlags::GetInstance()
+               .GetFlags()
+               .endpoint_manager_ensure_workers_terminated_inside_remove) {
+        EnsureWorkersTerminated(endpoint_id);
+      }
+    }
     // Pass ownership of channel to EndpointChannelManager
     NEARBY_LOG(INFO, "Registering endpoint with channel manager: id=%s",
                endpoint_id.c_str());
     channel_manager_->RegisterChannelForEndpoint(
         client, endpoint_id, std::unique_ptr<EndpointChannel>(channel));
 
-    EnsureWorkersTerminated(endpoint_id);
     EndpointState& endpoint_state =
         endpoints_.emplace(endpoint_id, EndpointState()).first->second;
     endpoint_state.client = client;
@@ -406,6 +412,7 @@ void EndpointManager::RegisterEndpoint(ClientProxy* client,
 
 void EndpointManager::UnregisterEndpoint(ClientProxy* client,
                                          const std::string& endpoint_id) {
+  NEARBY_LOG(ERROR, "UnregisterEndpoint for endpoint %s", endpoint_id.c_str());
   CountDownLatch latch(1);
   RunOnEndpointManagerThread([this, client, endpoint_id, &latch]() {
     RemoveEndpoint(client, endpoint_id,
@@ -442,6 +449,7 @@ std::vector<std::string> EndpointManager::SendPayloadChunk(
 // allow synchronous behavior here it will cause a live lock.
 void EndpointManager::DiscardEndpoint(ClientProxy* client,
                                       const std::string& endpoint_id) {
+  NEARBY_LOG(ERROR, "DiscardEndpoint for endpoint %s", endpoint_id.c_str());
   RunOnEndpointManagerThread([this, client, endpoint_id]() {
     RemoveEndpoint(client, endpoint_id,
                    /*notify=*/
@@ -464,6 +472,7 @@ std::vector<std::string> EndpointManager::SendControlMessage(
 void EndpointManager::RemoveEndpoint(ClientProxy* client,
                                      const std::string& endpoint_id,
                                      bool notify) {
+  NEARBY_LOG(ERROR, "RemoveEndpoint for endpoint %s", endpoint_id.c_str());
   // Unregistering from channel_manager_ will also serve to terminate
   // the dedicated handler and KeepAlive threads we started when we registered
   // this endpoint.
@@ -478,6 +487,11 @@ void EndpointManager::RemoveEndpoint(ClientProxy* client,
     client->OnDisconnected(endpoint_id, notify);
     NEARBY_LOG(INFO, "Removed endpoint; id=%s", endpoint_id.c_str());
   }
+  if (FeatureFlags::GetInstance()
+          .GetFlags()
+          .endpoint_manager_ensure_workers_terminated_inside_remove) {
+    EnsureWorkersTerminated(endpoint_id);
+  }
 }
 
 // @EndpointManagerThread
diff --git a/cpp/core/internal/mediums/webrtc/connection_flow.cc b/cpp/core/internal/mediums/webrtc/connection_flow.cc
index 40dcf7ee..15b25609 100644
--- a/cpp/core/internal/mediums/webrtc/connection_flow.cc
+++ b/cpp/core/internal/mediums/webrtc/connection_flow.cc
@@ -343,11 +343,15 @@ bool ConnectionFlow::TransitionState(State current_state, State new_state) {
 }
 
 bool ConnectionFlow::CloseLocked() {
+  NEARBY_LOG(INFO, "Closing WebRTC connection.");
   if (state_ == State::kEnded) {
     return false;
   }
   state_ = State::kEnded;
 
+  single_threaded_signaling_offloader_.Shutdown();
+  peer_connection_observer_.Shutdown();
+
   if (peer_connection_) peer_connection_->Close();
 
   data_channel_observer_.reset();
diff --git a/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.cc b/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.cc
index 8fb6c8e1..75be6c7c 100644
--- a/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.cc
+++ b/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.cc
@@ -28,6 +28,10 @@ PeerConnectionObserverImpl::PeerConnectionObserverImpl(
     : connection_flow_(connection_flow),
       local_ice_candidate_listener_(std::move(local_ice_candidate_listener)) {}
 
+PeerConnectionObserverImpl::~PeerConnectionObserverImpl() {
+    Shutdown();
+}
+
 void PeerConnectionObserverImpl::OnIceCandidate(
     const webrtc::IceCandidateInterface* candidate) {
   local_ice_candidate_listener_.local_ice_candidate_found_cb(candidate);
@@ -38,8 +42,10 @@ void PeerConnectionObserverImpl::OnSignalingChange(
   NEARBY_LOG(INFO, "OnSignalingChange: %d", new_state);
 
   OffloadFromSignalingThread([this, new_state]() {
-    if (new_state == webrtc::PeerConnectionInterface::SignalingState::kStable)
+    if (new_state == webrtc::PeerConnectionInterface::SignalingState::kStable &&
+        connection_flow_) {
       connection_flow_->OnSignalingStable();
+    }
   });
 }
 
@@ -47,8 +53,10 @@ void PeerConnectionObserverImpl::OnDataChannel(
     rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel) {
   NEARBY_LOG(INFO, "OnDataChannel");
 
-  data_channel->RegisterObserver(
-      connection_flow_->CreateDataChannelObserver(data_channel));
+  if (connection_flow_) {
+    data_channel->RegisterObserver(
+        connection_flow_->CreateDataChannelObserver(data_channel));
+  }
 }
 
 void PeerConnectionObserverImpl::OnIceGatheringChange(
@@ -61,7 +69,9 @@ void PeerConnectionObserverImpl::OnConnectionChange(
   NEARBY_LOG(INFO, "OnConnectionChange: %d", new_state);
 
   OffloadFromSignalingThread([this, new_state]() {
-    connection_flow_->ProcessOnPeerConnectionChange(new_state);
+    if (connection_flow_) {
+      connection_flow_->ProcessOnPeerConnectionChange(new_state);
+    }
   });
 }
 
@@ -69,6 +79,11 @@ void PeerConnectionObserverImpl ::OnRenegotiationNeeded() {
   NEARBY_LOG(INFO, "OnRenegotiationNeeded");
 }
 
+void PeerConnectionObserverImpl::Shutdown() {
+  single_threaded_signaling_offloader_.Shutdown();
+  connection_flow_ = nullptr;
+}
+
 void PeerConnectionObserverImpl::OffloadFromSignalingThread(Runnable runnable) {
   single_threaded_signaling_offloader_.Execute(std::move(runnable));
 }
diff --git a/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.h b/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.h
index b7d3be47..a2906c20 100644
--- a/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.h
+++ b/cpp/core/internal/mediums/webrtc/peer_connection_observer_impl.h
@@ -28,10 +28,10 @@ class ConnectionFlow;
 
 class PeerConnectionObserverImpl : public webrtc::PeerConnectionObserver {
  public:
-  ~PeerConnectionObserverImpl() override = default;
   PeerConnectionObserverImpl(
       ConnectionFlow* connection_flow,
       LocalIceCandidateListener local_ice_candidate_listener);
+  ~PeerConnectionObserverImpl() override;
 
   // webrtc::PeerConnectionObserver:
   void OnIceCandidate(const webrtc::IceCandidateInterface* candidate) override;
@@ -45,10 +45,12 @@ class PeerConnectionObserverImpl : public webrtc::PeerConnectionObserver {
       webrtc::PeerConnectionInterface::PeerConnectionState new_state) override;
   void OnRenegotiationNeeded() override;
 
+  void Shutdown();
+
  private:
   void OffloadFromSignalingThread(Runnable runnable);
 
-  ConnectionFlow* connection_flow_;
+  ConnectionFlow* volatile connection_flow_;
   LocalIceCandidateListener local_ice_candidate_listener_;
   SingleThreadExecutor single_threaded_signaling_offloader_;
 };
diff --git a/cpp/platform/base/feature_flags.h b/cpp/platform/base/feature_flags.h
index 3d8456a5..ee4a89cd 100644
--- a/cpp/platform/base/feature_flags.h
+++ b/cpp/platform/base/feature_flags.h
@@ -35,6 +35,13 @@ class FeatureFlags {
     // Ignore subsequent BWU Available events when we're still processing the
     // first one.
     bool disallow_out_of_order_bwu_avail_event = true;
+    bool enable_async_bandwidth_upgrade = true;
+    // Let endpoint_manager erase deleted endpoint from endpoints_ inside
+    // function RemoveEndpoint.
+    bool endpoint_manager_ensure_workers_terminated_inside_remove = true;
+    // If a scheduled runnable is already running, Cancel() will synchronously
+    // wait for the task to complete.
+    bool cancel_waits_for_running_tasks = true;
   };
 
   static const FeatureFlags& GetInstance() {
diff --git a/cpp/platform/impl/g3/bluetooth_classic.cc b/cpp/platform/impl/g3/bluetooth_classic.cc
index 4c7c7901..7af8b6c4 100644
--- a/cpp/platform/impl/g3/bluetooth_classic.cc
+++ b/cpp/platform/impl/g3/bluetooth_classic.cc
@@ -106,9 +106,8 @@ BluetoothDevice* BluetoothSocket::GetRemoteDevice() {
 
 std::unique_ptr<api::BluetoothSocket> BluetoothServerSocket::Accept() {
   absl::MutexLock lock(&mutex_);
-  while (pending_sockets_.empty()) {
+  while (!closed_ && pending_sockets_.empty()) {
     cond_.Wait(&mutex_);
-    if (closed_) break;
   }
   // whether or not we were running in the wait loop, return early if closed.
   if (closed_) return {};
diff --git a/cpp/platform/public/BUILD b/cpp/platform/public/BUILD
index 6b394eec..24b4878e 100644
--- a/cpp/platform/public/BUILD
+++ b/cpp/platform/public/BUILD
@@ -22,6 +22,7 @@ cc_library(
         "atomic_reference.h",
         "cancelable.h",
         "cancelable_alarm.h",
+        "cancellable_task.h",
         "condition_variable.h",
         "count_down_latch.h",
         "crypto.h",
@@ -107,6 +108,7 @@ cc_library(
 cc_test(
     name = "public_test",
     size = "small",
+    timeout = "moderate",
     srcs = [
         "atomic_boolean_test.cc",
         "atomic_reference_test.cc",
diff --git a/cpp/platform/public/ble.cc b/cpp/platform/public/ble.cc
index 41ce8d3f..cb277a29 100644
--- a/cpp/platform/public/ble.cc
+++ b/cpp/platform/public/ble.cc
@@ -53,9 +53,6 @@ bool BleMedium::StartScanning(
                 auto& context = *pair.first->second;
                 if (pair.second) {
                   context.peripheral = BlePeripheral(&peripheral);
-                  NEARBY_LOG(INFO,
-                             "Discovered peripheral '%s'",
-                             peripheral.GetName().c_str());
                   discovered_peripheral_callback_.peripheral_discovered_cb(
                       context.peripheral, service_id,
                       context.peripheral.GetAdvertisementBytes(service_id),
diff --git a/cpp/platform/public/cancelable.h b/cpp/platform/public/cancelable.h
index ef9fd6ae..89b957b1 100644
--- a/cpp/platform/public/cancelable.h
+++ b/cpp/platform/public/cancelable.h
@@ -19,6 +19,7 @@
 #include <utility>
 
 #include "platform/api/cancelable.h"
+#include "platform/public/cancellable_task.h"
 
 namespace location {
 namespace nearby {
@@ -35,14 +36,21 @@ class Cancelable final {
 
   // This constructor is used internally only,
   // by other classes in "//platform/public/".
-  explicit Cancelable(std::shared_ptr<api::Cancelable> impl)
-      : impl_(std::move(impl)) {}
+  explicit Cancelable(std::shared_ptr<CancellableTask> task,
+                      std::shared_ptr<api::Cancelable> impl)
+      : task_{task}, impl_(std::move(impl)) {}
 
-  bool Cancel() { return impl_ ? impl_->Cancel() : false; }
+  bool Cancel() {
+    if (!impl_) return false;
+    bool result = impl_->Cancel();
+    task_->CancelAndWaitIfStarted();
+    return result;
+  }
 
   bool IsValid() { return impl_ != nullptr; }
 
  private:
+  std::shared_ptr<CancellableTask> task_;
   std::shared_ptr<api::Cancelable> impl_;
 };
 
diff --git a/cpp/platform/public/cancellable_task.h b/cpp/platform/public/cancellable_task.h
new file mode 100644
index 00000000..70c6e013
--- /dev/null
+++ b/cpp/platform/public/cancellable_task.h
@@ -0,0 +1,56 @@
+#ifndef PLATFORM_PUBLIC_CANCELLABLE_TASK_H_
+#define PLATFORM_PUBLIC_CANCELLABLE_TASK_H_
+
+#include <utility>
+
+#include "platform/base/feature_flags.h"
+#include "platform/base/runnable.h"
+#include "platform/public/atomic_boolean.h"
+#include "platform/public/future.h"
+
+namespace location {
+namespace nearby {
+
+/**
+ * Runnable wrapper that allows one to wait for the task
+ * to complete if it is already running.
+ */
+class CancellableTask {
+ public:
+  explicit CancellableTask(Runnable&& runnable)
+      : runnable_{std::move(runnable)} {}
+
+  /**
+   * Try to cancel the task and wait until completion if the task is already
+   * running.
+   */
+  void CancelAndWaitIfStarted() {
+    if (started_or_cancelled_.Set(true)) {
+      if (FeatureFlags::GetInstance()
+              .GetFlags()
+              .cancel_waits_for_running_tasks) {
+        // task could still be running, wait until finish
+        finished_.Get();
+      }
+    } else {
+      // mark as finished to support multiple calls to this method
+      finished_.Set(true);
+    }
+  }
+
+  void operator()() {
+    if (started_or_cancelled_.Set(true)) return;
+    runnable_();
+    finished_.Set(true);
+  }
+
+ private:
+  AtomicBoolean started_or_cancelled_;
+  Future<bool> finished_;
+  Runnable runnable_;
+};
+
+}  // namespace nearby
+}  // namespace location
+
+#endif  // PLATFORM_PUBLIC_CANCELLABLE_TASK_H_
diff --git a/cpp/platform/public/scheduled_executor.h b/cpp/platform/public/scheduled_executor.h
index 4dba913f..4ce64867 100644
--- a/cpp/platform/public/scheduled_executor.h
+++ b/cpp/platform/public/scheduled_executor.h
@@ -23,6 +23,7 @@
 #include "platform/api/scheduled_executor.h"
 #include "platform/base/runnable.h"
 #include "platform/public/cancelable.h"
+#include "platform/public/cancellable_task.h"
 #include "platform/public/mutex.h"
 #include "platform/public/mutex_lock.h"
 #include "absl/time/time.h"
@@ -73,8 +74,13 @@ class ScheduledExecutor final {
   Cancelable Schedule(Runnable&& runnable, absl::Duration duration)
       ABSL_LOCKS_EXCLUDED(mutex_) {
     MutexLock lock(&mutex_);
-    return impl_ ? Cancelable(impl_->Schedule(std::move(runnable), duration))
-                 : Cancelable();
+    if (impl_) {
+      auto task = std::make_shared<CancellableTask>(std::move(runnable));
+      return Cancelable(task,
+                        impl_->Schedule([task]() { (*task)(); }, duration));
+    } else {
+      return Cancelable();
+    }
   }
 
  private:
diff --git a/cpp/platform/public/scheduled_executor_test.cc b/cpp/platform/public/scheduled_executor_test.cc
index d7e371ec..f2469373 100644
--- a/cpp/platform/public/scheduled_executor_test.cc
+++ b/cpp/platform/public/scheduled_executor_test.cc
@@ -18,6 +18,7 @@
 #include <functional>
 
 #include "platform/base/exception.h"
+#include "platform/public/count_down_latch.h"
 #include "gtest/gtest.h"
 #include "absl/synchronization/mutex.h"
 #include "absl/time/clock.h"
@@ -95,6 +96,20 @@ TEST(ScheduledExecutorTest, CanCancel) {
   EXPECT_EQ(value, 0);
 }
 
+TEST(ScheduledExecutorTest, CanCancelTwice) {
+  ScheduledExecutor executor;
+  std::atomic_int value = 0;
+  Cancelable cancelable =
+      executor.Schedule([&value]() { value += 1; }, kShortDelay);
+  EXPECT_EQ(value, 0);
+
+  cancelable.Cancel();
+  cancelable.Cancel();
+
+  absl::SleepFor(kLongDelay);
+  EXPECT_EQ(value, 0);
+}
+
 TEST(ScheduledExecutorTest, FailToCancel) {
   absl::Mutex mutex;
   absl::CondVar cond;
@@ -118,5 +133,47 @@ TEST(ScheduledExecutorTest, FailToCancel) {
   EXPECT_EQ(value, 1);
 }
 
+TEST(ScheduledExecutorTest,
+     CancelWhileRunning_TaskCompletesBeforeCancelReturns) {
+  CountDownLatch start_latch(1);
+  ScheduledExecutor executor;
+  std::atomic_int value = 0;
+  // A task that takes a little bit of time to complete
+  Cancelable cancelable = executor.Schedule(
+      [&start_latch, &value]() {
+        start_latch.CountDown();
+        absl::SleepFor(kLongDelay);
+        value += 1;
+      },
+      absl::ZeroDuration());
+
+  start_latch.Await();
+  cancelable.Cancel();
+
+  EXPECT_EQ(value, 1);
+}
+
+TEST(ScheduledExecutorTest,
+     CancelTwiceWhileRunning_TaskCompletesBeforeCancelReturns) {
+  CountDownLatch start_latch(1);
+  ScheduledExecutor executor;
+  std::atomic_int value = 0;
+  // A task that takes a little bit of time to complete
+  Cancelable cancelable = executor.Schedule(
+      [&start_latch, &value]() {
+        start_latch.CountDown();
+        absl::SleepFor(kLongDelay);
+        value += 1;
+      },
+      absl::ZeroDuration());
+
+  start_latch.Await();
+
+  cancelable.Cancel();
+  cancelable.Cancel();
+
+  EXPECT_EQ(value, 1);
+}
+
 }  // namespace nearby
 }  // namespace location