Fix 3 flaky thrift tests: InteractionTest, ServiceHealthPoller, RocketNetwork

Summary:
Fix race conditions and timing-sensitivity in three flaky thrift tests.

1. InteractionTest::PrioritizedInteractionRequest (cpp2/test/InteractionTest.cpp):
   The test had a race between TilePromise::fulfill() executing asynchronously
   on the EventBase thread and the concurrency controller dequeuing the next
   request. If the interaction request was dequeued before the tile was fully
   registered, it would hit an uninitialized TilePromise. Fixed by completing
   the factory method synchronously (get() on sf1) before proceeding, and
   using a separate "blocker" addPrimitive(0,0) call to occupy the worker
   thread instead of relying on the interaction setup.

2. ServiceHealthPollerTest::DynamicLiveness (cpp2/test/ServiceHealthPollerTest.cpp):
   The test used co_await sleep(50ms) then checked the result, which was
   timing-dependent and could fail under CI load. Replaced with event-driven
   co_await result->co_next() which waits for the actual state change. Also
   increased liveness interval from 200ms to 500ms to ensure the "unchanged
   because liveness is large" assertion has sufficient margin.

3. RocketNetworkTest::ObserverIsNotInstalledWhenFlagIsFalse
   (transport/rocket/test/network/RocketNetworkTest.cpp):
   After reconnect(), the old server-side connection may linger briefly while
   the new one is already created. The retry loop in getServerConnection()
   only tried 10 times which was insufficient under CI load. Increased to
   50 retries.

Reviewed By: evanjzou

Differential Revision: D94987644

fbshipit-source-id: 16a0e176290f9ca42859e15ee21fa993ced89e0d

Author: sazonovkirill

third-party/thrift/src/thrift/lib/cpp2/test/InteractionTest.cpp

@@ -108,18 +108,18 @@ TEST(InteractionTest, PrioritizedInteractionRequest) {
   // test.
   FLAGS_thrift_experimental_use_resource_pools = true;
   struct BlockingCalculatorHandler : public SemiCalculatorHandler {
-    folly::Baton<> blockInteraction, blockNormal, startedInteraction;
-
-    TileAndResponse<AdditionIf, int> initializedAddition(int x) override {
-      startedInteraction.post();
-      EXPECT_TRUE(blockInteraction.try_wait_for(std::chrono::seconds(5)));
-      return SemiCalculatorHandler::initializedAddition(x);
-    }
+    folly::Baton<> blockBlocker, blockNormal, startedBlocker;
 
     folly::SemiFuture<int32_t> semifuture_addPrimitive(
         int32_t a, int32_t b) override {
-      EXPECT_TRUE(blockNormal.try_wait_for(std::chrono::seconds(5)));
-      blockNormal.reset();
+      if (a == 0 && b == 0) {
+        // This is the "blocker" call used to occupy the worker thread.
+        startedBlocker.post();
+        EXPECT_TRUE(blockBlocker.try_wait_for(std::chrono::seconds(5)));
+      } else {
+        EXPECT_TRUE(blockNormal.try_wait_for(std::chrono::seconds(5)));
+        blockNormal.reset();
+      }
       return a + b;
     }
   };
@@ -134,18 +134,26 @@ TEST(InteractionTest, PrioritizedInteractionRequest) {
   auto client = runner.newClient<CalculatorAsyncClient>(
       nullptr, RocketClientChannel::newChannel);
 
+  // Create the interaction and let the factory method complete. This ensures
+  // the tile is fully registered, avoiding a race where TilePromise::fulfill()
+  // runs asynchronously on the EventBase thread after the concurrency
+  // controller has already released the worker to dequeue the next request.
   RpcOptions opts;
   auto [adder, sf1] = client->eager_semifuture_initializedAddition(opts, 42);
-  auto sf2 = adder.semifuture_getPrimitive();
+  EXPECT_EQ(std::move(sf1).get(), 42);
 
-  // Wait for the interaction requests to reach the server
-  handler->startedInteraction.wait();
+  // Block the sole worker thread with a normal request.
+  auto blocker = client->semifuture_addPrimitive(0, 0);
+  handler->startedBlocker.wait();
 
+  // While the worker is blocked, send both an interaction request (MID_PRI)
+  // and a normal request (LO_PRI). Both enter the resource pool's RequestPile.
+  auto sf2 = adder.semifuture_getPrimitive();
   auto normal = client->semifuture_addPrimitive(1, 2);
 
-  // Wait for the normal requests to reach the server
+  // Wait for both requests to be queued in the resource pool.
   auto start = std::chrono::steady_clock::now();
-  while (runner.getThriftServer().resourcePoolSet().numQueued() != 1) {
+  while (runner.getThriftServer().resourcePoolSet().numQueued() != 2) {
     /* sleep override */
     std::this_thread::sleep_for(std::chrono::milliseconds(10));
     if (std::chrono::steady_clock::now() - start > std::chrono::seconds(5)) {
@@ -154,14 +162,15 @@ TEST(InteractionTest, PrioritizedInteractionRequest) {
     }
   }
 
-  // Unblock the interaction request and wait for it to finish
-  handler->blockInteraction.post();
-  EXPECT_EQ(std::move(sf1).get(), 42);
+  // Unblock the blocker so the worker can process queued requests.
+  handler->blockBlocker.post();
+  EXPECT_EQ(std::move(blocker).get(), 0);
 
-  // Make sure interaction request is executed before normal request.
+  // The interaction request (MID_PRI) should be dequeued before the normal
+  // request (LO_PRI).
   EXPECT_EQ(std::move(sf2).get(), 42);
 
-  // Unblock the normal request and wait for it to finish
+  // Unblock the normal request and wait for it to finish.
   handler->blockNormal.post();
   EXPECT_EQ(std::move(normal).get(), 3);
 }

third-party/thrift/src/thrift/lib/cpp2/test/ServiceHealthPollerTest.cpp

@@ -243,17 +243,17 @@ CO_TEST(ServiceHealthPoller, DynamicLiveness) {
   EXPECT_EQ(co_await result->co_next(), ServiceHealth::ERROR);
 
   handler.set(ServiceHealth::OK);
-  co_await folly::coro::sleep(50ms);
-  EXPECT_EQ(result->get(), ServiceHealth::OK);
+  // Use event-driven wait instead of sleep to avoid flakiness under load
+  EXPECT_EQ(co_await result->co_next(), ServiceHealth::OK);
 
-  liveness.setValue(200ms);
+  liveness.setValue(500ms);
   folly::observer_detail::ObserverManager::waitForAllUpdates();
   // wait for any existing polling to be done
   co_await result->co_next();
 
   handler.set(ServiceHealth::ERROR);
   co_await folly::coro::sleep(50ms);
-  // unchanged because liveness is large
+  // unchanged because liveness is large (500ms >> 50ms)
   EXPECT_EQ(result->get(), ServiceHealth::OK);
 
   EXPECT_EQ(co_await result->co_next(), ServiceHealth::ERROR);

third-party/thrift/src/thrift/lib/cpp2/transport/rocket/test/network/RocketNetworkTest.cpp

@@ -106,8 +106,10 @@ class RocketNetworkTest : public testing::Test {
   void unsetExpectedSetupMetadata() { server_->setExpectedSetupMetadata({}); }
 
   // Returns the server connection once it is established.
+  // After reconnect(), the old server-side connection may linger briefly while
+  // the new one is already created, so we retry until exactly 1 remains.
   RocketServerConnection* getServerConnection() {
-    for (int i = 0, numTries = 10; i < numTries; ++i) {
+    for (int i = 0, numTries = 50; i < numTries; ++i) {
       auto connCount = 0;
       RocketServerConnection* conn = nullptr;
       this->server_->getEventBase().runInEventBaseThreadAndWait([&] {