From 988a0c913d1f3d45ee348fa20e7383442a9c75f7 Mon Sep 17 00:00:00 2001 From: Rahmeen14 Date: Mon, 10 Aug 2026 17:49:18 +0100 Subject: [PATCH 1/3] Run first host health check immediately and apply health state on registration --- src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.cpp | 69 ++++++++++++++----- src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.h | 18 +++-- .../rmqamqp/rmqamqp_hosthealthmonitor.t.cpp | 69 +++++++++++++++++++ 3 files changed, 133 insertions(+), 23 deletions(-) diff --git a/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.cpp b/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.cpp index 0c56872..293e440 100644 --- a/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.cpp +++ b/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.cpp @@ -42,6 +42,9 @@ HostHealthMonitor::HostHealthMonitor( rmqp::MetricPublisher* metricPublisher) : d_hostHealthConfig(hostHealthConfig) , d_currentTries(0) +// Fail-safe: unhealthy until the first check completes, so connections +// registering beforehand start paused rather than consuming blind. +, d_lastKnownHealth(UNHEALTHY) , d_timer() , d_metricPublisher(metricPublisher) { @@ -65,7 +68,9 @@ void HostHealthMonitor::start( bdlf::BindUtil::bind(&HostHealthMonitor::handleTimerFired, weak_from_this(), bdlf::PlaceHolders::_1)); - scheduleNextCheck(); + // Fire the first check immediately instead of after a full poll interval; + // checkHealth() reschedules subsequent checks at pollInterval. + d_timer->reset(bsls::TimeInterval(0)); } void HostHealthMonitor::stop() @@ -83,6 +88,23 @@ void HostHealthMonitor::registerConnection( d_metricPublisher->publishGauge(Metrics::HEALTH_AWARE_VHOSTS, static_cast(d_connections.size())); + + // Apply the current known health now (rather than waiting for the next + // poll) so consumers created on this connection open paused when unhealthy + // and active when healthy. + bsl::shared_ptr connection = conn.lock(); + if (connection) { + if (d_lastKnownHealth == UNHEALTHY) { + BALL_LOG_INFO << "healthState=UNHEALTHY action=pause " + "reason=newly-registered-connection"; + connection->pauseReceiveChannels(RESPECT_HOST_HEALTH); + } + else { + BALL_LOG_INFO << "healthState=HEALTHY action=resume " + "reason=newly-registered-connection"; + connection->resumeReceiveChannels(RESPECT_HOST_HEALTH); + } + } } void HostHealthMonitor::handleTimerFired( @@ -121,9 +143,9 @@ 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); @@ -131,23 +153,24 @@ void HostHealthMonitor::checkHealth() 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,9 +179,9 @@ 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( @@ -166,12 +189,12 @@ void HostHealthMonitor::checkHealth() static_cast(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 +213,18 @@ void HostHealthMonitor::processHealthResult(HostHealth result) { d_currentTries = 0; - BALL_LOG_DEBUG << (result == HEALTHY ? "Resuming" : "Pausing") - << " host health aware consumers."; + if (result != d_lastKnownHealth) { + BALL_LOG_INFO << "event=health-state-changed previousHealthState=" + << d_lastKnownHealth << " healthState=" << result; + } + + // Cache for registerConnection; only HEALTHY/UNHEALTHY reach here (RETRY + // returns early), so the last known state is retained across retries. + d_lastKnownHealth = 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); diff --git a/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.h b/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.h index d3d06f7..4b030a5 100644 --- a/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.h +++ b/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 { 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& 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& connection); @@ -91,6 +98,7 @@ class HostHealthMonitor rmqt::HostHealthConfig d_hostHealthConfig; bsl::list > d_connections; unsigned int d_currentTries; + HostHealth d_lastKnownHealth; bsl::shared_ptr d_timer; rmqp::MetricPublisher* d_metricPublisher; }; diff --git a/src/tests/rmqamqp/rmqamqp_hosthealthmonitor.t.cpp b/src/tests/rmqamqp/rmqamqp_hosthealthmonitor.t.cpp index 5e172c5..d7da218 100644 --- a/src/tests/rmqamqp/rmqamqp_hosthealthmonitor.t.cpp +++ b/src/tests/rmqamqp/rmqamqp_hosthealthmonitor.t.cpp @@ -236,6 +236,18 @@ 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); + + // The first health check is scheduled with a zero delay, so advancing the + // clock by zero (rather than a full poll interval) is enough to fire it. + d_timerFactory->step_time(bsls::TimeInterval(0)); +} + TEST_F(HostHealthMonitorTests, UnhealthyHostPausesConnections) { d_configurableHealthChecker.d_nextResult = false; @@ -246,6 +258,63 @@ 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(); + + // A connection that registers now (while the host is known-unhealthy) must + // be paused immediately, without waiting for the next health check. + bsl::shared_ptr lateConn = makeConnection("late-unhealthy"); + + EXPECT_CALL(*lateConn, pauseReceiveChannels(true)).Times(1); + EXPECT_CALL(*lateConn, resumeReceiveChannels(_)).Times(0); + + d_monitor->registerConnection(bsl::weak_ptr(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(); + + // A connection registering while the host is known healthy is resumed + // immediately, so consumers created on it start consuming without waiting + // for the next health check. + bsl::shared_ptr lateConn = makeConnection("late-healthy"); + + EXPECT_CALL(*lateConn, resumeReceiveChannels(true)).Times(1); + EXPECT_CALL(*lateConn, pauseReceiveChannels(_)).Times(0); + + d_monitor->registerConnection(bsl::weak_ptr(lateConn)); +} + +TEST_F(HostHealthMonitorTests, RegisterBeforeFirstCheckPausesImmediately) +{ + // Before the first health check completes, the host health is unknown. The + // monitor defaults to UNHEALTHY (fail-safe), so a connection registering in + // that window is paused immediately rather than being allowed to consume + // from a host whose health has not yet been confirmed. + bsl::shared_ptr monitor = + bsl::make_shared(d_config, d_metricPublisher.get()); + monitor->start(d_timerFactory); + + EXPECT_CALL(*d_metricPublisher, publishGauge(_, _, _)).Times(AtLeast(0)); + + bsl::shared_ptr earlyConn = makeConnection("early"); + + EXPECT_CALL(*earlyConn, pauseReceiveChannels(true)).Times(1); + EXPECT_CALL(*earlyConn, resumeReceiveChannels(_)).Times(0); + + monitor->registerConnection(bsl::weak_ptr(earlyConn)); +} + TEST_F(HostHealthMonitorTests, ExpiredConnectionIsRemovedAndNotUsed) { bsl::shared_ptr liveConn = From 27444152737b595cba4dd329b774b5ade9d7657c Mon Sep 17 00:00:00 2001 From: Rahmeen14 Date: Tue, 11 Aug 2026 14:22:12 +0100 Subject: [PATCH 2/3] fix CI --- dockerfiles/dev.Dockerfile | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/dockerfiles/dev.Dockerfile b/dockerfiles/dev.Dockerfile index b96befe..c2a1e09 100644 --- a/dockerfiles/dev.Dockerfile +++ b/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,11 @@ RUN apt-get update && apt-get install -y \ zip \ && rm -rf /var/lib/apt/lists/* +# Debian stable ships CMake 3.31, but current vcpkg port scripts call +# string(JSON ... STRING_ENCODE), which was added in CMake 4.3. Install a +# modern CMake from PyPI (arch-independent) so vcpkg dependency builds succeed. +RUN pip3 install --break-system-packages --no-cache-dir "cmake>=4.3,<5" + ENV VCPKG_FORCE_SYSTEM_BINARIES=1 # clone and install vcpkg From cafdd7f25ee54bff0de4714aaeec78b30d184e52 Mon Sep 17 00:00:00 2001 From: Rahmeen14 Date: Wed, 12 Aug 2026 15:28:39 +0100 Subject: [PATCH 3/3] refactor --- dockerfiles/dev.Dockerfile | 3 - src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.cpp | 59 ++++++++++--------- src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.h | 2 +- .../rmqamqp/rmqamqp_hosthealthmonitor.t.cpp | 14 +---- 4 files changed, 34 insertions(+), 44 deletions(-) diff --git a/dockerfiles/dev.Dockerfile b/dockerfiles/dev.Dockerfile index c2a1e09..7c7b353 100644 --- a/dockerfiles/dev.Dockerfile +++ b/dockerfiles/dev.Dockerfile @@ -24,9 +24,6 @@ RUN apt-get update && apt-get install -y \ zip \ && rm -rf /var/lib/apt/lists/* -# Debian stable ships CMake 3.31, but current vcpkg port scripts call -# string(JSON ... STRING_ENCODE), which was added in CMake 4.3. Install a -# modern CMake from PyPI (arch-independent) so vcpkg dependency builds succeed. RUN pip3 install --break-system-packages --no-cache-dir "cmake>=4.3,<5" ENV VCPKG_FORCE_SYSTEM_BINARIES=1 diff --git a/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.cpp b/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.cpp index 293e440..76af8c6 100644 --- a/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.cpp +++ b/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.cpp @@ -35,6 +35,17 @@ 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( @@ -42,9 +53,9 @@ HostHealthMonitor::HostHealthMonitor( rmqp::MetricPublisher* metricPublisher) : d_hostHealthConfig(hostHealthConfig) , d_currentTries(0) -// Fail-safe: unhealthy until the first check completes, so connections -// registering beforehand start paused rather than consuming blind. -, d_lastKnownHealth(UNHEALTHY) +// Fail-safe: assume unhealthy until the first check, so connections that +// register beforehand start paused. +, d_latestHealthCheckResult(UNHEALTHY) , d_timer() , d_metricPublisher(metricPublisher) { @@ -68,8 +79,8 @@ void HostHealthMonitor::start( bdlf::BindUtil::bind(&HostHealthMonitor::handleTimerFired, weak_from_this(), bdlf::PlaceHolders::_1)); - // Fire the first check immediately instead of after a full poll interval; - // checkHealth() reschedules subsequent checks at pollInterval. + // Fire the first check immediately; checkHealth() then reschedules at + // pollInterval. d_timer->reset(bsls::TimeInterval(0)); } @@ -89,21 +100,16 @@ void HostHealthMonitor::registerConnection( d_metricPublisher->publishGauge(Metrics::HEALTH_AWARE_VHOSTS, static_cast(d_connections.size())); - // Apply the current known health now (rather than waiting for the next - // poll) so consumers created on this connection open paused when unhealthy - // and active when healthy. + // Bring the connection into the current known state now, rather than + // waiting for the next poll. bsl::shared_ptr connection = conn.lock(); if (connection) { - if (d_lastKnownHealth == UNHEALTHY) { - BALL_LOG_INFO << "healthState=UNHEALTHY action=pause " - "reason=newly-registered-connection"; - connection->pauseReceiveChannels(RESPECT_HOST_HEALTH); - } - else { - BALL_LOG_INFO << "healthState=HEALTHY action=resume " - "reason=newly-registered-connection"; - connection->resumeReceiveChannels(RESPECT_HOST_HEALTH); - } + BALL_LOG_INFO << "healthState=" << d_latestHealthCheckResult + << " action=" + << (d_latestHealthCheckResult == HEALTHY ? "resume" + : "pause") + << " reason=newly-registered-connection"; + syncReceiveChannelsToHostHealth(*connection, d_latestHealthCheckResult); } } @@ -213,14 +219,14 @@ void HostHealthMonitor::processHealthResult(HostHealth result) { d_currentTries = 0; - if (result != d_lastKnownHealth) { + if (result != d_latestHealthCheckResult) { BALL_LOG_INFO << "event=health-state-changed previousHealthState=" - << d_lastKnownHealth << " healthState=" << result; + << d_latestHealthCheckResult << " healthState=" << result; } - // Cache for registerConnection; only HEALTHY/UNHEALTHY reach here (RETRY - // returns early), so the last known state is retained across retries. - d_lastKnownHealth = 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") @@ -258,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; } diff --git a/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.h b/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.h index 4b030a5..581c829 100644 --- a/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.h +++ b/src/rmq/rmqamqp/rmqamqp_hosthealthmonitor.h @@ -98,7 +98,7 @@ class HostHealthMonitor rmqt::HostHealthConfig d_hostHealthConfig; bsl::list > d_connections; unsigned int d_currentTries; - HostHealth d_lastKnownHealth; + HostHealth d_latestHealthCheckResult; bsl::shared_ptr d_timer; rmqp::MetricPublisher* d_metricPublisher; }; diff --git a/src/tests/rmqamqp/rmqamqp_hosthealthmonitor.t.cpp b/src/tests/rmqamqp/rmqamqp_hosthealthmonitor.t.cpp index d7da218..ac9a6ae 100644 --- a/src/tests/rmqamqp/rmqamqp_hosthealthmonitor.t.cpp +++ b/src/tests/rmqamqp/rmqamqp_hosthealthmonitor.t.cpp @@ -243,8 +243,7 @@ TEST_F(HostHealthMonitorTests, FirstCheckFiresImmediately) EXPECT_CALL(*d_connection, resumeReceiveChannels(true)).Times(1); EXPECT_CALL(*d_connection, pauseReceiveChannels(_)).Times(0); - // The first health check is scheduled with a zero delay, so advancing the - // clock by zero (rather than a full poll interval) is enough to fire it. + // First check is scheduled with zero delay, so stepping by zero fires it. d_timerFactory->step_time(bsls::TimeInterval(0)); } @@ -266,8 +265,6 @@ TEST_F(HostHealthMonitorTests, RegisterOnUnhealthyHostPausesImmediately) EXPECT_CALL(*d_connection, pauseReceiveChannels(true)).Times(1); stepAndClear(); - // A connection that registers now (while the host is known-unhealthy) must - // be paused immediately, without waiting for the next health check. bsl::shared_ptr lateConn = makeConnection("late-unhealthy"); EXPECT_CALL(*lateConn, pauseReceiveChannels(true)).Times(1); @@ -284,9 +281,6 @@ TEST_F(HostHealthMonitorTests, RegisterOnHealthyHostResumesImmediately) EXPECT_CALL(*d_connection, resumeReceiveChannels(true)).Times(1); stepAndClear(); - // A connection registering while the host is known healthy is resumed - // immediately, so consumers created on it start consuming without waiting - // for the next health check. bsl::shared_ptr lateConn = makeConnection("late-healthy"); EXPECT_CALL(*lateConn, resumeReceiveChannels(true)).Times(1); @@ -297,10 +291,8 @@ TEST_F(HostHealthMonitorTests, RegisterOnHealthyHostResumesImmediately) TEST_F(HostHealthMonitorTests, RegisterBeforeFirstCheckPausesImmediately) { - // Before the first health check completes, the host health is unknown. The - // monitor defaults to UNHEALTHY (fail-safe), so a connection registering in - // that window is paused immediately rather than being allowed to consume - // from a host whose health has not yet been confirmed. + // No check has run yet, so the monitor's fail-safe default (UNHEALTHY) + // applies. bsl::shared_ptr monitor = bsl::make_shared(d_config, d_metricPublisher.get()); monitor->start(d_timerFactory);