Fix flaky test.

PiperOrigin-RevId: 864541156

connections/implementation/mediums/ble/BUILD

@@ -176,7 +176,6 @@ cc_test(
         ":ble_advertisement_header",
         ":bloom_filter",
         "//connections/implementation:types",
-        "//connections/implementation/flags:connections_flags",
         "//connections/implementation/mediums:utils",
         "//connections/implementation/mediums/advertisements:dct_advertisement",
         "//internal/flags:nearby_flags",
@@ -193,6 +192,7 @@ cc_test(
         "@com_github_protobuf_matchers//protobuf-matchers",
         "@com_google_absl//absl/base:core_headers",
         "@com_google_absl//absl/hash:hash_testing",
+        "@com_google_absl//absl/memory",
         "@com_google_absl//absl/status",
         "@com_google_absl//absl/status:statusor",
         "@com_google_absl//absl/strings",

connections/implementation/mediums/ble/discovered_peripheral_tracker.cc

@@ -63,18 +63,30 @@ constexpr absl::Duration kExtendedAdvertisementHeaderDelay = absl::Seconds(3);
 constexpr absl::Duration kAdvertisementHeaderExpiry = absl::Seconds(15);
 }  // namespace
 
+// Private c'tor for testing.
+DiscoveredPeripheralTracker::DiscoveredPeripheralTracker(
+    bool is_extended_advertisement_available, bool start_fetch_executor)
+    : is_extended_advertisement_available_(
+          is_extended_advertisement_available),
+      start_fetch_executor_(start_fetch_executor) {}
+
 DiscoveredPeripheralTracker::DiscoveredPeripheralTracker(
     bool is_extended_advertisement_available)
-    : is_extended_advertisement_available_(
-          is_extended_advertisement_available) {
-  executor_ = std::make_unique<MultiThreadExecutor>(kGattThreadCount);
-  executor_->Execute([this]() { GattFetchingLoop(); });
-}
+    : DiscoveredPeripheralTracker(is_extended_advertisement_available,
+                                  /*start_fetch_executor=*/true) {}
 
 DiscoveredPeripheralTracker::~DiscoveredPeripheralTracker() {
   Shutdown();
 }
 
+void DiscoveredPeripheralTracker::StartFetchExecutorIfNeeded() {
+  if (executor_ != nullptr) {
+    return;
+  }
+  executor_ = std::make_unique<MultiThreadExecutor>(kGattThreadCount);
+  executor_->Execute([this]() { GattFetchingLoop(); });
+}
+
 void DiscoveredPeripheralTracker::Shutdown() {
   if (shutting_down_.Set(true)) {
     return;
@@ -101,6 +113,9 @@ void DiscoveredPeripheralTracker::StartTracking(
     const Uuid& fast_advertisement_service_uuid) {
   MutexLock lock(&mutex_);
 
+  if (start_fetch_executor_) {
+    StartFetchExecutorIfNeeded();
+  }
   ServiceIdInfo service_id_info = {
       .discovered_peripheral_callback =
           std::move(discovered_peripheral_callback),
@@ -318,9 +333,13 @@ bool DiscoveredPeripheralTracker::IsSkippableGattAdvertisement(
       ExtractAdvertisementHeaderBytes(advertisement_data));
 
   if (!advertisement_header.IsValid()) {
+    // Don't skip any advertisement if the header is not valid.  It may be one
+    // of the legacy advertisement formats that is dealt with later.
     return false;
   }
 
+  // Delay processing GATT advertisements that have corresponding extended
+  // advertisements for 3s.  After that all GATT advertisements are processed.
   if (advertisement_header.IsSupportExtendedAdvertisement() &&
       (SystemClock::ElapsedRealtime() - medium_start_scanning_time_) <
           kExtendedAdvertisementHeaderDelay) {
@@ -740,16 +759,11 @@ void DiscoveredPeripheralTracker::HandleAdvertisementHeader(
   }
 
   // Determine whether or not we need to read a fresh GATT advertisement.
-  VLOG(1) << ": Handle GATT advertisement header with hash "
+  VLOG(1) << "Handle GATT advertisement header with hash "
           << absl::BytesToHexString(
                   advertisement_header.GetAdvertisementHash().AsStringView())
           << " in thread";
 
-  if (executor_ == nullptr) {
-    // The situation happens when flag value changed
-    executor_ = std::make_unique<MultiThreadExecutor>(kGattThreadCount);
-  }
-
   if (!ShouldReadRawAdvertisementFromServer(advertisement_header)) {
     UpdateCommonStateForFoundBleAdvertisement(advertisement_header);
     return;
@@ -831,7 +845,6 @@ bool DiscoveredPeripheralTracker::ShouldReadRawAdvertisementFromServer(
     const BleAdvertisementHeader& advertisement_header) {
   // Check if we have never seen this header. New headers should always be
   // read.
-  ByteArray advertisement_header_bytes(advertisement_header);
   const auto it = advertisement_read_results_.find(advertisement_header);
   if (it == advertisement_read_results_.end()) {
     LOG(INFO) << "Received advertisement header with hash "
@@ -965,7 +978,8 @@ void DiscoveredPeripheralTracker::FetchRawAdvertisementsInThread(
 
 void DiscoveredPeripheralTracker::GattFetchingLoop() {
   while (true) {
-    GattFetchTask task;
+    GattFetchTask* task = nullptr;
+    bool is_extended_advertisement = false;
     {
       MutexLock lock(&task_mutex_);
       if (shutting_down_) {
@@ -986,27 +1000,40 @@ void DiscoveredPeripheralTracker::GattFetchingLoop() {
       // Non-extended advertisements are prioritized over extended
       // advertisements.
       if (!gatt_fetch_tasks_.empty()) {
-        task = std::move(gatt_fetch_tasks_.front());
-        gatt_fetch_tasks_.pop_front();
+        task = &gatt_fetch_tasks_.front();
       } else if (!gatt_extended_fetch_tasks_.empty()) {
-        task = std::move(gatt_extended_fetch_tasks_.front());
-        gatt_extended_fetch_tasks_.pop_front();
+        is_extended_advertisement = true;
+        task = &gatt_extended_fetch_tasks_.front();
       }
     }
+    if (task == nullptr) {
+      LOG(WARNING) << "No task found, skip to fetch raw advertisement.";
+      continue;;
+    }
 
     // Check if the task is expired.
-    if (SystemClock::ElapsedRealtime() - task.scheduled_time >
+    if (SystemClock::ElapsedRealtime() - task->scheduled_time >
         kAdvertisementHeaderExpiry) {
       VLOG(1) << "GATT advertisement with hash: "
               << absl::BytesToHexString(
-                     task.advertisement_header.GetAdvertisementHash()
+                     task->advertisement_header.GetAdvertisementHash()
                          .AsStringView())
               << " is expired, skip to fetch raw advertisement.";
-      continue;
+    } else {
+      FetchRawAdvertisementsInThread(task->peripheral,
+                                     task->advertisement_header,
+                                     std::move(task->advertisement_fetcher));
+    }
+    // Remove task from queue after fetching is done.  This allows newly scanned
+    // advertisements to be deduped against a running fetch task.
+    {
+      MutexLock lock(&task_mutex_);
+      if (is_extended_advertisement) {
+        gatt_extended_fetch_tasks_.pop_front();
+      } else {
+        gatt_fetch_tasks_.pop_front();
+      }
     }
-
-    FetchRawAdvertisementsInThread(task.peripheral, task.advertisement_header,
-                                   std::move(task.advertisement_fetcher));
   }
 }
 

connections/implementation/mediums/ble/discovered_peripheral_tracker.h

@@ -70,7 +70,7 @@ class DiscoveredPeripheralTracker {
       mediums::AdvertisementReadResult& advertisement_read_result)>;
 
   explicit DiscoveredPeripheralTracker(
-      bool is_extended_advertisement_available = false);
+      bool is_extended_advertisement_available);
 
   ~DiscoveredPeripheralTracker();
 
@@ -112,6 +112,11 @@ class DiscoveredPeripheralTracker {
   // Shuts down the GATT fetching thread.
   void Shutdown() ABSL_LOCKS_EXCLUDED(mutex_);
 
+  void StartFetchExecutorForTesting() {
+    MutexLock lock(&mutex_);
+    StartFetchExecutorIfNeeded();
+  }
+
  private:
   using BleAdvertisementSet = absl::flat_hash_set<BleAdvertisement>;
 
@@ -161,6 +166,16 @@ class DiscoveredPeripheralTracker {
     absl::Time scheduled_time;
   };
 
+  friend class DiscoveredPeripheralTrackerTest;
+
+  // C'tor used for tests only.
+  DiscoveredPeripheralTracker(
+      bool is_extended_advertisement_available, bool start_fetch_executor);
+
+  // Initiatializes the executor used for handling GATT fetch tasks if not
+  // already started.
+  void StartFetchExecutorIfNeeded() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
   // Clears stale data from any previous sessions.
   void ClearDataForServiceId(const std::string& service_id)
       ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
@@ -306,7 +321,8 @@ class DiscoveredPeripheralTracker {
       ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
 
   Mutex mutex_;
-  bool is_extended_advertisement_available_;
+  const bool is_extended_advertisement_available_;
+  const bool start_fetch_executor_;
 
   absl::Time medium_start_scanning_time_ ABSL_GUARDED_BY(mutex_) =
       absl::InfiniteFuture();

connections/implementation/mediums/ble/discovered_peripheral_tracker_test.cc

@@ -27,10 +27,10 @@
 #include "protobuf-matchers/protocol-buffer-matchers.h"
 #include "gtest/gtest.h"
 #include "absl/base/thread_annotations.h"
+#include "absl/memory/memory.h"
 #include "absl/strings/string_view.h"
 #include "absl/time/clock.h"
 #include "absl/time/time.h"
-#include "connections/implementation/flags/nearby_connections_feature_flags.h"
 #include "connections/implementation/mediums/advertisements/dct_advertisement.h"
 #include "connections/implementation/mediums/ble/advertisement_read_result.h"
 #include "connections/implementation/mediums/ble/ble_advertisement.h"
@@ -187,6 +187,8 @@ class MockDiscoveredPeripheralCallback : public DiscoveredPeripheralCallback {
   MOCK_METHOD(void, OnLegacyDeviceDiscovered, (), ());
 };
 
+}  // namespace
+
 class DiscoveredPeripheralTrackerTest
     : public testing::TestWithParam<
           std::tuple</*is_extended_advertisement_available=*/bool>> {
@@ -198,9 +200,11 @@ class DiscoveredPeripheralTrackerTest
         .use_simulated_clock = true,
     };
     MediumEnvironment::Instance().Start(config);
-    discovered_peripheral_tracker_ =
-        std::make_unique<DiscoveredPeripheralTracker>(
-            is_extended_advertisement_available);
+    discovered_peripheral_tracker_ = absl::WrapUnique(
+        new DiscoveredPeripheralTracker(is_extended_advertisement_available,
+                                        /*start_fetch_executor=*/false));
+    // Do not start the fetch executor since we will manually run the GATT
+    // fetch loop in the test.
     adapter_peripheral_ = std::make_unique<BluetoothAdapter>();
     adapter_central_ = std::make_unique<BluetoothAdapter>();
     ble_peripheral_ = std::make_unique<BleMedium>(*adapter_peripheral_);
@@ -350,6 +354,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
               },
       },
       Uuid(kFastAdvertisementServiceUuid));
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!fast_advertisement_bytes.Empty()) {
@@ -385,6 +390,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, DctAdvertisementPeripheralDiscovered) {
               },
       },
       Uuid(kFastAdvertisementServiceUuid));
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   advertisement_data.service_data.insert(
@@ -424,6 +430,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
        .legacy_device_discovered_cb =
            [&legacy_found_latch]() { legacy_found_latch.CountDown(); }},
       {});
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!advertisement_header_bytes.Empty()) {
@@ -464,6 +471,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
               },
       },
       Uuid(kFastAdvertisementServiceUuid));
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!fast_advertisement_bytes.Empty()) {
@@ -535,6 +543,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
               },
       },
       Uuid(kFastAdvertisementServiceUuid));
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!fast_advertisement_bytes.Empty()) {
@@ -579,6 +588,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
               },
       },
       Uuid("FE3C"));
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!fast_advertisement_bytes.Empty()) {
@@ -622,6 +632,8 @@ TEST_P(DiscoveredPeripheralTrackerTest,
               },
       },
       Uuid(kFastAdvertisementServiceUuid));
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
+
   discovered_peripheral_tracker_->StartTracking(
       std::string(kServiceIdB), false, Pcp::kP2pPointToPoint,
       {
@@ -681,6 +693,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
               },
       },
       {});
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!advertisement_header_bytes.Empty()) {
@@ -717,6 +730,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
                   bool fast_advertisement) { found_latch.CountDown(); },
       },
       {});
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!advertisement_header_bytes.Empty()) {
@@ -763,6 +777,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
               },
       },
       {});
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!advertisement_header_bytes.Empty()) {
@@ -809,6 +824,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
               },
       },
       {});
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!advertisement_header_bytes.Empty()) {
@@ -852,6 +868,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
                   bool fast_advertisement) { found_latch.CountDown(); },
       },
       {});
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!advertisement_header_bytes.Empty()) {
@@ -901,6 +918,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
               },
       },
       Uuid(kFastAdvertisementServiceUuid));
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!advertisement_header_bytes.Empty()) {
@@ -961,6 +979,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
                   bool fast_advertisement) { lost_latch.CountDown(); },
       },
       Uuid(kFastAdvertisementServiceUuid));
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!advertisement_header_bytes.Empty()) {
@@ -1018,6 +1037,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, LostPeripheralForAdvertisementLost) {
                   bool fast_advertisement) { lost_latch.CountDown(); },
       },
       {});
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!advertisement_header_bytes.Empty()) {
@@ -1096,6 +1116,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
                   bool fast_advertisement) { lost_latch_b.CountDown(); },
       },
       {});
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!advertisement_header_bytes.Empty()) {
@@ -1158,6 +1179,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
                   bool fast_advertisement) { lost_latch.CountDown(); },
       },
       {});
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!advertisement_header_bytes.Empty()) {
@@ -1212,6 +1234,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, InstantLostPeripheralForInstantOnLost) {
                   bool fast_advertisement) { lost_latch.CountDown(); },
       },
       {});
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!advertisement_header_bytes.Empty()) {
@@ -1284,6 +1307,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
               },
       },
       {});
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!advertisement_header_bytes.Empty()) {
@@ -1360,6 +1384,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
                   bool fast_advertisement) { lost_latch.CountDown(); },
       },
       Uuid(kFastAdvertisementServiceUuid));
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!fast_advertisement_bytes.Empty()) {
@@ -1416,6 +1441,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, HandleDummyAdvertisement) {
               },
       },
       {});
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   FindAdvertisement(advertising_data, {}, fetch_latch);
 
@@ -1452,6 +1478,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, SkipDummyAdvertisement) {
               },
       },
       {});
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   FindAdvertisement(advertising_data, {}, fetch_latch);
 
@@ -1483,6 +1510,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, FetchGattAdvertisementInThread) {
               },
       },
       {});
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!advertisement_header_bytes.Empty()) {
@@ -1524,6 +1552,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
               },
       },
       {});
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!advertisement_header_bytes.Empty()) {
@@ -1571,6 +1600,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
               },
       },
       {});
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   api::ble::BleAdvertisementData advertisement_data{};
   if (!advertisement_header_bytes.Empty()) {
@@ -1625,6 +1655,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
               },
       },
       bleutils::kCopresenceServiceUuid);
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   // 1. First receive a GATT advertisement data, it should be skipped.
   api::ble::BleAdvertisementData advertisement_data{};
@@ -1675,7 +1706,7 @@ TEST_P(DiscoveredPeripheralTrackerTest, SkipExpiredGattAdvertisement) {
       },
       bleutils::kCopresenceServiceUuid);
 
-  // 1. First receive a GATT advertisement data, it should be skipped.
+  // 1. First receive a GATT advertisement data, it will expire after 15s.
   api::ble::BleAdvertisementData advertisement_data{};
   if (!advertisement_header_bytes.Empty()) {
     advertisement_data.service_data.insert(
@@ -1685,11 +1716,12 @@ TEST_P(DiscoveredPeripheralTrackerTest, SkipExpiredGattAdvertisement) {
   FindAdvertisementWithDelay(advertisement_data, {advertisement_bytes},
                              fetch_latch, kDefaultGattFetchDelay);
 
-  // 2. The GATT advertisement will be expired after 20 seconds.
+  // 2. The GATT advertisement is already queued and will be skipped.
   FindAdvertisement(advertisement_data, {advertisement_bytes}, fetch_latch);
   (*fake_clock)->FastForward(absl::Seconds(20));
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
-  // We should receive a client callback of a peripheral discovery.
+  // We should not receive a client callback of a peripheral discovery.
   fetch_latch.Await(kWaitDuration);
   EXPECT_FALSE(found_latch.Await(kWaitDuration).result());
   EXPECT_EQ(GetFetchAdvertisementCallbackCount(), 0);
@@ -1725,6 +1757,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
       },
       bleutils::kCopresenceServiceUuid);
   (*fake_clock)->FastForward(absl::Seconds(4));
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   // 1. Received extended advertisement.
   api::ble::BleAdvertisementData extended_advertisement_data{};
@@ -1791,6 +1824,7 @@ TEST_P(DiscoveredPeripheralTrackerTest,
       },
       bleutils::kCopresenceServiceUuid);
   (*fake_clock)->FastForward(absl::Seconds(4));
+  discovered_peripheral_tracker_->StartFetchExecutorForTesting();
 
   // 1. Find peripheral A with GATT advertisement.
   api::ble::BleAdvertisementData advertisement_data_a{};
@@ -1824,8 +1858,6 @@ INSTANTIATE_TEST_SUITE_P(
     ::testing::Combine(
         /*is_extended_advertisement_available=*/testing::Bool()));
 
-}  // namespace
-
 }  // namespace mediums
 }  // namespace connections
 }  // namespace nearby