Run first host health check immediately and apply health state on registration + fix CI by upgrading CMake version

dockerfiles/dev.Dockerfile

@@ -3,7 +3,6 @@ FROM debian:stable
 RUN apt-get update && apt-get install -y \
     build-essential \
     clang-format \
-    cmake \
     curl \
     gcc \
     gdb \
@@ -25,6 +24,8 @@ RUN apt-get update && apt-get install -y \
     zip \
     && rm -rf /var/lib/apt/lists/*
 
+RUN pip3 install --break-system-packages --no-cache-dir "cmake>=4.3,<5"
+
 ENV VCPKG_FORCE_SYSTEM_BINARIES=1
 
 # clone and install vcpkg

src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.cpp

@@ -35,13 +35,27 @@ BALL_LOG_SET_NAMESPACE_CATEGORY("RMQAMQP.HOSTHEALTHMONITOR")
 
 const bool RESPECT_HOST_HEALTH = true;
 
+void syncReceiveChannelsToHostHealth(rmqamqp::Connection& connection,
+                                     HostHealthMonitor::HostHealth health)
+{
+    if (health == HostHealthMonitor::HEALTHY) {
+        connection.resumeReceiveChannels(RESPECT_HOST_HEALTH);
+    }
+    else if (health == HostHealthMonitor::UNHEALTHY) {
+        connection.pauseReceiveChannels(RESPECT_HOST_HEALTH);
+    }
+}
+
 } // namespace
 
 HostHealthMonitor::HostHealthMonitor(
     const rmqt::HostHealthConfig& hostHealthConfig,
     rmqp::MetricPublisher* metricPublisher)
 : d_hostHealthConfig(hostHealthConfig)
 , d_currentTries(0)
+// Fail-safe: assume unhealthy until the first check, so connections that
+// register beforehand start paused.
+, d_latestHealthCheckResult(UNHEALTHY)
 , d_timer()
 , d_metricPublisher(metricPublisher)
 {
@@ -65,7 +79,9 @@ void HostHealthMonitor::start(
         bdlf::BindUtil::bind(&HostHealthMonitor::handleTimerFired,
                              weak_from_this(),
                              bdlf::PlaceHolders::_1));
-    scheduleNextCheck();
+    // Fire the first check immediately; checkHealth() then reschedules at
+    // pollInterval.
+    d_timer->reset(bsls::TimeInterval(0));
 }
 
 void HostHealthMonitor::stop()
@@ -83,6 +99,18 @@ void HostHealthMonitor::registerConnection(
 
     d_metricPublisher->publishGauge(Metrics::HEALTH_AWARE_VHOSTS,
                                     static_cast<double>(d_connections.size()));
+
+    // Bring the connection into the current known state now, rather than
+    // waiting for the next poll.
+    bsl::shared_ptr<rmqamqp::Connection> connection = conn.lock();
+    if (connection) {
+        BALL_LOG_INFO << "healthState=" << d_latestHealthCheckResult
+                      << " action="
+                      << (d_latestHealthCheckResult == HEALTHY ? "resume"
+                                                               : "pause")
+                      << " reason=newly-registered-connection";
+        syncReceiveChannelsToHostHealth(*connection, d_latestHealthCheckResult);
+    }
 }
 
 void HostHealthMonitor::handleTimerFired(
@@ -121,33 +149,34 @@ void HostHealthMonitor::checkHealth()
             bsls::SystemTime::nowMonotonicClock();
         const double durationMs = (endTime - startTime).totalMilliseconds();
 
-        BALL_LOG_INFO << "Health check completed in " << durationMs
-                      << " ms with result: " << healthCheckerResult
-                      << ". Set health state to: " << result;
+        BALL_LOG_DEBUG << "event=health-check-completed durationMs="
+                       << durationMs << " result=" << healthCheckerResult
+                       << " healthState=" << result;
 
         d_metricPublisher->publishSummary(Metrics::HEALTH_CHECK_DURATION_MS,
                                           durationMs);
 
         const double HEALTHCHECK_DURATION_REPORTING_THRESHOLD =
             bsl::min(d_hostHealthConfig.pollInterval() * 1000.0 * 0.8, 1000.0);
         if (durationMs > HEALTHCHECK_DURATION_REPORTING_THRESHOLD) {
-            BALL_LOG_WARN << "Host health check took " << durationMs
-                          << " ms which exceeds the threshold of "
-                          << HEALTHCHECK_DURATION_REPORTING_THRESHOLD << " ms.";
+            BALL_LOG_WARN << "event=health-check-duration-exceeds-threshold "
+                             "durationMs="
+                          << durationMs << " thresholdMs="
+                          << HEALTHCHECK_DURATION_REPORTING_THRESHOLD;
             d_metricPublisher->publishCounter(
                 Metrics::HEALTH_CHECK_BLOCKED_EVENT_LOOP, 1.0);
         }
     }
     catch (const bsl::exception& e) {
-        BALL_LOG_ERROR << "Host health check failed with exception: "
-                       << e.what();
+        BALL_LOG_ERROR << "event=health-check-failed exception=\"" << e.what()
+                       << "\"";
         result = RETRY;
 
         d_metricPublisher->publishCounter(Metrics::HEALTH_CHECK_FAILURES_TOTAL,
                                           1.0);
     }
     catch (...) {
-        BALL_LOG_ERROR << "Host health check failed with unknown exception.";
+        BALL_LOG_ERROR << "event=health-check-failed exception=unknown";
         result = RETRY;
 
         d_metricPublisher->publishCounter(Metrics::HEALTH_CHECK_FAILURES_TOTAL,
@@ -156,22 +185,22 @@ void HostHealthMonitor::checkHealth()
 
     if (result == RETRY) {
         if (d_currentTries++ > d_hostHealthConfig.maxRetriesOnFailure()) {
-            BALL_LOG_ERROR << "Exceeded max retries on failure of "
+            BALL_LOG_ERROR << "event=max-retries-exceeded maxRetries="
                            << d_hostHealthConfig.maxRetriesOnFailure()
-                           << ". Marking host as UNHEALTHY.";
+                           << " action=mark-unhealthy";
             result = UNHEALTHY;
 
             d_metricPublisher->publishGauge(
                 Metrics::HEALTH_CHECK_CONSECUTIVE_FAILURES,
                 static_cast<double>(d_currentTries));
         }
         else {
-            BALL_LOG_WARN << "Current tries " << d_currentTries
-                          << " do not exceed max retries on failure of "
+            BALL_LOG_WARN << "event=health-check-retry currentTries="
+                          << d_currentTries << " maxRetries="
                           << d_hostHealthConfig.maxRetriesOnFailure()
-                          << ". Will retry after "
+                          << " retryAfterSeconds="
                           << d_hostHealthConfig.pollInterval()
-                          << " seconds. Will NOT pause the consumers yet.";
+                          << " action=none";
 
             if (d_metricPublisher) {
                 d_metricPublisher->publishGauge(
@@ -190,8 +219,18 @@ void HostHealthMonitor::processHealthResult(HostHealth result)
 {
     d_currentTries = 0;
 
-    BALL_LOG_DEBUG << (result == HEALTHY ? "Resuming" : "Pausing")
-                   << " host health aware consumers.";
+    if (result != d_latestHealthCheckResult) {
+        BALL_LOG_INFO << "event=health-state-changed previousHealthState="
+                      << d_latestHealthCheckResult << " healthState=" << result;
+    }
+
+    // Cache for registerConnection. RETRY returns early, so this only ever
+    // holds HEALTHY or UNHEALTHY.
+    d_latestHealthCheckResult = result;
+
+    BALL_LOG_DEBUG << "healthState=" << result
+                   << " action=" << (result == HEALTHY ? "resume" : "pause")
+                   << " target=health-aware-consumers";
 
     const double statusValue = (result == HEALTHY) ? 1.0 : 0.0;
     d_metricPublisher->publishGauge(Metrics::HEALTH_CHECK_STATUS, statusValue);
@@ -225,12 +264,7 @@ void HostHealthMonitor::processHealthResult(HostHealth result)
             continue;
         }
 
-        if (result == HEALTHY) {
-            connection->resumeReceiveChannels(RESPECT_HOST_HEALTH);
-        }
-        else {
-            connection->pauseReceiveChannels(RESPECT_HOST_HEALTH);
-        }
+        syncReceiveChannelsToHostHealth(*connection, result);
 
         ++conn;
     }

src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.h

@@ -39,6 +39,12 @@ namespace rmqamqp {
 /// complete promptly, as it will delay AMQP heartbeats and message delivery
 /// for all connections by its execution time every \c pollInterval seconds.
 /// Each check schedules the next one via a timer (self-rescheduling pattern).
+///
+/// One monitor exists per context: the checker runs once per \c pollInterval
+/// regardless of connection or consumer count. The latest result is cached and
+/// applied to every registered connection (and to connections registering
+/// between checks, see \c registerConnection). Defaults to unhealthy until the
+/// first check.
 class HostHealthMonitor
 : public bsl::enable_shared_from_this<HostHealthMonitor> {
   public:
@@ -64,16 +70,17 @@ class HostHealthMonitor
 
     ~HostHealthMonitor();
 
-    /// Start the health monitoring timer. The first health check will fire
-    /// after one poll interval.
+    /// Start the health monitoring timer. The first health check fires
+    /// immediately (on the event loop); subsequent checks run every
+    /// \c pollInterval seconds.
     void start(const bsl::shared_ptr<rmqio::TimerFactory>& timerFactory);
 
     /// Stop the health monitoring timer.
     void stop();
 
-    /// Register a connection to be notified about host health changes.
-    /// When the host becomes unhealthy, the connection's receive channels
-    /// will be paused. When the host recovers, they will be resumed.
+    /// Register a connection to be paused/resumed as host health changes.
+    /// At registration the connection is brought into the current known state
+    /// immediately: resumed if the host is known healthy, paused otherwise.
     void
     registerConnection(const bsl::weak_ptr<rmqamqp::Connection>& connection);
 
@@ -91,6 +98,7 @@ class HostHealthMonitor
     rmqt::HostHealthConfig d_hostHealthConfig;
     bsl::list<bsl::weak_ptr<rmqamqp::Connection> > d_connections;
     unsigned int d_currentTries;
+    HostHealth d_latestHealthCheckResult;
     bsl::shared_ptr<rmqio::Timer> d_timer;
     rmqp::MetricPublisher* d_metricPublisher;
 };

src/tests/rmqamqp/rmqamqp_hosthealthmonitor.t.cpp

@@ -236,6 +236,17 @@ TEST_F(HostHealthMonitorTests, HealthyHostResumesConnections)
     stepOnePollInterval();
 }
 
+TEST_F(HostHealthMonitorTests, FirstCheckFiresImmediately)
+{
+    d_configurableHealthChecker.d_nextResult = true;
+
+    EXPECT_CALL(*d_connection, resumeReceiveChannels(true)).Times(1);
+    EXPECT_CALL(*d_connection, pauseReceiveChannels(_)).Times(0);
+
+    // First check is scheduled with zero delay, so stepping by zero fires it.
+    d_timerFactory->step_time(bsls::TimeInterval(0));
+}
+
 TEST_F(HostHealthMonitorTests, UnhealthyHostPausesConnections)
 {
     d_configurableHealthChecker.d_nextResult = false;
@@ -246,6 +257,56 @@ TEST_F(HostHealthMonitorTests, UnhealthyHostPausesConnections)
     stepOnePollInterval();
 }
 
+TEST_F(HostHealthMonitorTests, RegisterOnUnhealthyHostPausesImmediately)
+{
+    // Drive one check that marks the host UNHEALTHY.
+    d_configurableHealthChecker.d_nextResult = false;
+
+    EXPECT_CALL(*d_connection, pauseReceiveChannels(true)).Times(1);
+    stepAndClear();
+
+    bsl::shared_ptr<MockConnection> lateConn = makeConnection("late-unhealthy");
+
+    EXPECT_CALL(*lateConn, pauseReceiveChannels(true)).Times(1);
+    EXPECT_CALL(*lateConn, resumeReceiveChannels(_)).Times(0);
+
+    d_monitor->registerConnection(bsl::weak_ptr<rmqamqp::Connection>(lateConn));
+}
+
+TEST_F(HostHealthMonitorTests, RegisterOnHealthyHostResumesImmediately)
+{
+    // Drive one check that marks the host HEALTHY.
+    d_configurableHealthChecker.d_nextResult = true;
+
+    EXPECT_CALL(*d_connection, resumeReceiveChannels(true)).Times(1);
+    stepAndClear();
+
+    bsl::shared_ptr<MockConnection> lateConn = makeConnection("late-healthy");
+
+    EXPECT_CALL(*lateConn, resumeReceiveChannels(true)).Times(1);
+    EXPECT_CALL(*lateConn, pauseReceiveChannels(_)).Times(0);
+
+    d_monitor->registerConnection(bsl::weak_ptr<rmqamqp::Connection>(lateConn));
+}
+
+TEST_F(HostHealthMonitorTests, RegisterBeforeFirstCheckPausesImmediately)
+{
+    // No check has run yet, so the monitor's fail-safe default (UNHEALTHY)
+    // applies.
+    bsl::shared_ptr<HostHealthMonitor> monitor =
+        bsl::make_shared<HostHealthMonitor>(d_config, d_metricPublisher.get());
+    monitor->start(d_timerFactory);
+
+    EXPECT_CALL(*d_metricPublisher, publishGauge(_, _, _)).Times(AtLeast(0));
+
+    bsl::shared_ptr<MockConnection> earlyConn = makeConnection("early");
+
+    EXPECT_CALL(*earlyConn, pauseReceiveChannels(true)).Times(1);
+    EXPECT_CALL(*earlyConn, resumeReceiveChannels(_)).Times(0);
+
+    monitor->registerConnection(bsl::weak_ptr<rmqamqp::Connection>(earlyConn));
+}
+
 TEST_F(HostHealthMonitorTests, ExpiredConnectionIsRemovedAndNotUsed)
 {
     bsl::shared_ptr<MockConnection> liveConn =