Add endpoint connection-lifecycle hooks and a configurable establishment timeout

src/rmq/rmqa/rmqa_rabbitcontextimpl.cpp

@@ -214,6 +214,7 @@ RabbitContextImpl::RabbitContextImpl(
             d_connectionMonitor,
             options.clientProperties(),
             options.connectionErrorThreshold(),
+            options.connectionEstablishmentTimeout(),
             isHostHealthMonitoringEnabled);
 
     if (!d_threadPool) {

src/rmq/rmqa/rmqa_rabbitcontextoptions.cpp

@@ -21,9 +21,14 @@
 #include <bdlf_bind.h>
 #include <bdls_osutil.h>
 #include <bdls_processutil.h>
+#include <bsls_assert.h>
+#include <bsls_timeinterval.h>
 
 namespace BloombergLP {
 namespace rmqa {
+
+BALL_LOG_SET_NAMESPACE_CATEGORY("RMQA.RABBITCONTEXTOPTIONS")
+
 namespace {
 
 void populateUsefulInformation(rmqt::FieldTable* propertiesPtr)
@@ -66,6 +71,7 @@ RabbitContextOptions::RabbitContextOptions()
 , d_messageProcessingTimeout(DEFAULT_MESSAGE_PROCESSING_TIMEOUT)
 , d_tunables()
 , d_connectionErrorThreshold()
+, d_connectionEstablishmentTimeout()
 , d_shuffleConnectionEndpoints()
 , d_hostHealthConfig()
 {
@@ -115,6 +121,30 @@ RabbitContextOptions& RabbitContextOptions::setConnectionErrorThreshold(
     return *this;
 }
 
+RabbitContextOptions& RabbitContextOptions::setConnectionEstablishmentTimeout(
+    const bsl::optional<bsls::TimeInterval>& timeout)
+{
+    // The hung timer applies this at millisecond granularity (it uses
+    // bsls::TimeInterval::totalMilliseconds()), so any value below 1ms --
+    // including a positive sub-millisecond interval -- rounds down to zero and
+    // would expire the bound immediately on every attempt, producing a
+    // connect -> hung -> retry loop that never establishes. Require at least
+    // 1ms rather than merely positive; log as well as assert so the
+    // misconfiguration is diagnosable even in builds where assertions are
+    // compiled out.
+    if (timeout.has_value() && *timeout < bsls::TimeInterval(0, 1000 * 1000)) {
+        BALL_LOG_ERROR << "setConnectionEstablishmentTimeout given "
+                       << (timeout->totalSecondsAsDouble() * 1000.0)
+                       << "ms; values below 1ms round down to zero at the "
+                          "timer's millisecond granularity and prevent any "
+                          "connection from establishing";
+    }
+    BSLS_ASSERT(!timeout.has_value() ||
+                *timeout >= bsls::TimeInterval(0, 1000 * 1000));
+    d_connectionEstablishmentTimeout = timeout;
+    return *this;
+}
+
 RabbitContextOptions& RabbitContextOptions::setConsumerTracing(
     const bsl::shared_ptr<rmqp::ConsumerTracing>& consumerTracing)
 {

src/rmq/rmqa/rmqa_rabbitcontextoptions.h

@@ -113,6 +113,27 @@ class RabbitContextOptions {
     RabbitContextOptions& setConnectionErrorThreshold(
         const bsl::optional<bsls::TimeInterval>& timeout);
 
+    /// \brief Set the maximum time a single connection-establishment attempt
+    /// (TCP connect + TLS handshake + AMQP handshake) may take before it is
+    /// aborted and retried. The library already bounds establishment with a
+    /// default; this makes that bound configurable.
+    /// \param timeout the establishment timeout, as a `bsls::TimeInterval` so
+    /// that any resolution (down to nanoseconds) can be expressed -- e.g.
+    /// `bsls::TimeInterval(5.0)` for five seconds, or
+    /// `bsls::TimeInterval(0, 500 * 1000 * 1000)` for 500 milliseconds. If not
+    /// set (`bsl::nullopt`), the library default is used. This only ever
+    /// applies to the establishment phase -- it is cancelled once a connection
+    /// is established -- so it does not affect long-lived connections. Lowering
+    /// it makes the client give up on a stalled attempt and retry sooner;
+    /// raising it tolerates slower networks/handshakes.
+    /// \note The timeout is applied at millisecond granularity: the behavior
+    /// is undefined unless `timeout`, when set, is at least one millisecond. A
+    /// smaller (sub-millisecond) interval rounds down to zero, which would
+    /// expire the bound immediately and prevent any connection from
+    /// establishing.
+    RabbitContextOptions& setConnectionEstablishmentTimeout(
+        const bsl::optional<bsls::TimeInterval>& timeout);
+
     /// \brief will be called back to create a context which spans for the
     /// lifetime of the messageguard _before_ it is passed to its consumer
     /// message processor if there has
@@ -183,6 +204,12 @@ class RabbitContextOptions {
         return d_connectionErrorThreshold;
     }
 
+    const bsl::optional<bsls::TimeInterval>&
+    connectionEstablishmentTimeout() const
+    {
+        return d_connectionEstablishmentTimeout;
+    }
+
     const rmqt::Tunables& tunables() const { return d_tunables; }
 
     const bsl::shared_ptr<rmqp::ConsumerTracing>& consumerTracing() const
@@ -221,6 +248,7 @@ class RabbitContextOptions {
     bsls::TimeInterval d_messageProcessingTimeout;
     rmqt::Tunables d_tunables;
     bsl::optional<bsls::TimeInterval> d_connectionErrorThreshold;
+    bsl::optional<bsls::TimeInterval> d_connectionEstablishmentTimeout;
     bsl::shared_ptr<rmqp::ConsumerTracing> d_consumerTracing;
     bsl::shared_ptr<rmqp::ProducerTracing> d_producerTracing;
     bsl::optional<bool> d_shuffleConnectionEndpoints;

src/rmq/rmqamqp/rmqamqp_connection.cpp

@@ -48,6 +48,7 @@
 #include <bdlt_currenttime.h>
 #include <bsl_algorithm.h>
 #include <bsl_cstring.h>
+#include <bsl_exception.h>
 #include <bsl_functional.h>
 #include <bsl_limits.h>
 #include <bsl_memory.h>
@@ -80,7 +81,7 @@ const bsl::uint16_t k_MAX_CHANNEL_NUM = bsl::numeric_limits<uint16_t>::max();
 // k_MAX_HEARTBEAT_TIMEOUT_SEC cannot be zero as the HeartbeatManager does
 // not support this
 const bsl::uint16_t k_MAX_HEARTBEAT_TIMEOUT_SEC = 60;
-const bsl::uint16_t k_HUNG_TIMER_SEC            = 60;
+const bsl::uint16_t k_DEFAULT_HUNG_TIMER_SEC    = 60;
 
 BALL_LOG_SET_NAMESPACE_CATEGORY("RMQAMQP.CONNECTION")
 
@@ -220,6 +221,7 @@ Connection::Connection(
     const bsl::shared_ptr<rmqt::Endpoint>& endpoint,
     const bsl::shared_ptr<rmqt::Credentials>& credentials,
     const rmqt::FieldTable& clientProperties,
+    const bsl::optional<bsls::TimeInterval>& connectionEstablishmentTimeout,
     bsl::string_view name)
 : d_resolver(resolver)
 , d_retryHandler(retryHandler)
@@ -233,8 +235,9 @@ Connection::Connection(
 , d_state(Connection::DISCONNECTED)
 , d_clientProperties(clientProperties)
 , d_channels()
-, d_hungTimer(
-      timerFactory->createWithTimeout(bsls::TimeInterval(k_HUNG_TIMER_SEC)))
+, d_connectionEstablishmentTimeout(connectionEstablishmentTimeout.value_or(
+      bsls::TimeInterval(k_DEFAULT_HUNG_TIMER_SEC)))
+, d_hungTimer(timerFactory->createWithTimeout(d_connectionEstablishmentTimeout))
 , d_timerFactory(timerFactory)
 , d_firstConnectCb()
 , d_closeCb()
@@ -601,8 +604,32 @@ void Connection::socketShutdown(DisconnectType dcType,
     d_state = Connection::DISCONNECTED;
     BALL_LOG_TRACE << "State now set to: " << d_state;
     if (dcType == WILL_RETRY) {
+        // The connection is down and will be retried: either an establishment
+        // attempt failed or an established connection was lost. Schedule the
+        // retry before invoking the observational hook: the hook is a
+        // caller-supplied override, and if it were to throw, the retry must
+        // already have been scheduled so the connection is not left dead.
+        // Ordering is otherwise immaterial -- retry only arms a timer, so the
+        // next initiateConnect() (which reads the endpoint) runs well after
+        // this returns.
         d_retryHandler->retry(
             bdlf::BindUtil::bind(&Connection::retry, weak_from_this()));
+        // The hook is a caller-supplied override; catch any exception so it
+        // cannot propagate into the event loop and terminate the process. The
+        // retry above is already scheduled, so the connection is unaffected.
+        try {
+            d_endpoint->onConnectFailed();
+        }
+        catch (const bsl::exception& e) {
+            BALL_LOG_ERROR << "rmqt::Endpoint::onConnectFailed threw ("
+                           << e.what() << "); ignoring for "
+                           << connectionDebugName();
+        }
+        catch (...) {
+            BALL_LOG_ERROR << "rmqt::Endpoint::onConnectFailed threw a "
+                              "non-standard exception; ignoring for "
+                           << connectionDebugName();
+        }
     }
     else {
         if (d_firstConnectCb) {
@@ -670,6 +697,26 @@ class Connection::ConnectionMethodProcessor {
                 conn.d_firstConnectCb(result);
                 conn.d_firstConnectCb = ConnectedCallback();
             }
+
+            // Invoke the observational hook last: it is a caller-supplied
+            // override, so the critical connect path (timer cancel, retry
+            // success, channel open, connect callback) must complete first. By
+            // this point the connection is fully established, and any exception
+            // is caught so it cannot propagate into the event loop and
+            // terminate the process.
+            try {
+                conn.d_endpoint->onConnectSuccess();
+            }
+            catch (const bsl::exception& e) {
+                BALL_LOG_ERROR << "rmqt::Endpoint::onConnectSuccess threw ("
+                               << e.what() << "); ignoring for "
+                               << conn.connectionDebugName();
+            }
+            catch (...) {
+                BALL_LOG_ERROR << "rmqt::Endpoint::onConnectSuccess threw a "
+                                  "non-standard exception; ignoring for "
+                               << conn.connectionDebugName();
+            }
         }
     }
 
@@ -1010,9 +1057,9 @@ void Connection::connectionHung(rmqio::Timer::InterruptReason reason)
         d_metricPublisher->publishCounter(
             "hung_connection_reset", 1, d_vhostTags);
 
-        BALL_LOG_FATAL << "Timed out after " << k_HUNG_TIMER_SEC
-                       << " seconds while connecting: "
-                       << connectionDebugName();
+        BALL_LOG_ERROR << "Timed out after "
+                       << d_connectionEstablishmentTimeout.totalMilliseconds()
+                       << " ms while connecting: " << connectionDebugName();
         closeSocket(WILL_RETRY);
     }
 }
@@ -1056,6 +1103,7 @@ Connection::Factory::Factory(
     const bsl::shared_ptr<ConnectionMonitor>& connectionMonitor,
     const rmqt::FieldTable& clientProperties,
     const bsl::optional<bsls::TimeInterval>& connectionErrorThreshold,
+    const bsl::optional<bsls::TimeInterval>& connectionEstablishmentTimeout,
     const bool isHostHealthMonitoringEnabled)
 : d_errorCb(errorCb)
 , d_clientProperties(clientProperties)
@@ -1064,6 +1112,7 @@ Connection::Factory::Factory(
 , d_timerFactory(timerFactory)
 , d_connectionMonitor(connectionMonitor)
 , d_connectionErrorThreshold(connectionErrorThreshold)
+, d_connectionEstablishmentTimeout(connectionEstablishmentTimeout)
 , d_isHostHealthMonitoringEnabled(isHostHealthMonitoringEnabled)
 {
 }
@@ -1083,6 +1132,7 @@ bsl::shared_ptr<Connection> Connection::Factory::create(
         endpoint,
         credentials,
         generateClientProperties(d_clientProperties, name),
+        d_connectionEstablishmentTimeout,
         name));
 
     d_connectionMonitor->addConnection(bsl::weak_ptr<Connection>(result));

src/rmq/rmqamqp/rmqamqp_connection.h

@@ -172,16 +172,18 @@ class Connection : public bsl::enable_shared_from_this<Connection>,
     /// \param endpoint        references the vhost this connection is
     ///                        joining
     /// \param credentials     Credentials used to connect to the broker
-    Connection(const bsl::shared_ptr<rmqio::Resolver>& resolver,
-               const bsl::shared_ptr<rmqio::RetryHandler>& retryHandler,
-               const bsl::shared_ptr<rmqamqp::HeartbeatManager>& hbManager,
-               const bsl::shared_ptr<rmqio::TimerFactory>& timerFactory,
-               const bsl::shared_ptr<rmqamqp::ChannelFactory>& channelFactory,
-               const bsl::shared_ptr<rmqp::MetricPublisher>& metricPublisher,
-               const bsl::shared_ptr<rmqt::Endpoint>& endpoint,
-               const bsl::shared_ptr<rmqt::Credentials>& credentials,
-               const rmqt::FieldTable& clientProperties,
-               bsl::string_view name);
+    Connection(
+        const bsl::shared_ptr<rmqio::Resolver>& resolver,
+        const bsl::shared_ptr<rmqio::RetryHandler>& retryHandler,
+        const bsl::shared_ptr<rmqamqp::HeartbeatManager>& hbManager,
+        const bsl::shared_ptr<rmqio::TimerFactory>& timerFactory,
+        const bsl::shared_ptr<rmqamqp::ChannelFactory>& channelFactory,
+        const bsl::shared_ptr<rmqp::MetricPublisher>& metricPublisher,
+        const bsl::shared_ptr<rmqt::Endpoint>& endpoint,
+        const bsl::shared_ptr<rmqt::Credentials>& credentials,
+        const rmqt::FieldTable& clientProperties,
+        const bsl::optional<bsls::TimeInterval>& connectionEstablishmentTimeout,
+        bsl::string_view name);
 
   private:
     bsl::shared_ptr<rmqio::Resolver> d_resolver;
@@ -199,6 +201,7 @@ class Connection : public bsl::enable_shared_from_this<Connection>,
     State d_state;
     rmqt::FieldTable d_clientProperties;
     ChannelMap d_channels;
+    bsls::TimeInterval d_connectionEstablishmentTimeout;
     bsl::shared_ptr<rmqio::Timer> d_hungTimer;
 
     bsl::shared_ptr<rmqio::TimerFactory> d_timerFactory;
@@ -318,14 +321,16 @@ class Connection::Factory {
   public:
     // CREATORS
 
-    Factory(const bsl::shared_ptr<rmqio::Resolver>& resolver,
-            const bsl::shared_ptr<rmqio::TimerFactory>& timerFactory,
-            const rmqt::ErrorCallback& errorCb,
-            const bsl::shared_ptr<rmqp::MetricPublisher>& metricPublisher,
-            const bsl::shared_ptr<ConnectionMonitor>& connectionMonitor,
-            const rmqt::FieldTable& clientProperties,
-            const bsl::optional<bsls::TimeInterval>& connectionErrorThreshold,
-            const bool isHostHealthMonitoringEnabled);
+    Factory(
+        const bsl::shared_ptr<rmqio::Resolver>& resolver,
+        const bsl::shared_ptr<rmqio::TimerFactory>& timerFactory,
+        const rmqt::ErrorCallback& errorCb,
+        const bsl::shared_ptr<rmqp::MetricPublisher>& metricPublisher,
+        const bsl::shared_ptr<ConnectionMonitor>& connectionMonitor,
+        const rmqt::FieldTable& clientProperties,
+        const bsl::optional<bsls::TimeInterval>& connectionErrorThreshold,
+        const bsl::optional<bsls::TimeInterval>& connectionEstablishmentTimeout,
+        const bool isHostHealthMonitoringEnabled);
 
     virtual ~Factory() {}
 
@@ -350,6 +355,7 @@ class Connection::Factory {
     const bsl::shared_ptr<rmqio::TimerFactory> d_timerFactory;
     const bsl::shared_ptr<ConnectionMonitor> d_connectionMonitor;
     const bsl::optional<bsls::TimeInterval> d_connectionErrorThreshold;
+    const bsl::optional<bsls::TimeInterval> d_connectionEstablishmentTimeout;
     const bool d_isHostHealthMonitoringEnabled;
 }; // class Connection::Factory
 } // namespace rmqamqp

src/rmq/rmqt/rmqt_endpoint.h

@@ -36,6 +36,37 @@ class Endpoint {
     virtual bsl::uint16_t port() const        = 0;
     virtual bsl::shared_ptr<rmqt::SecurityParameters>
     securityParameters() const;
+
+    /// \brief Invoked each time an AMQP connection to this endpoint is
+    /// established (Connection.Open-Ok received). Defaults to a no-op;
+    /// implementations may override it to observe the connection lifecycle,
+    /// e.g. for monitoring or metrics.
+    ///
+    /// \warning Invoked synchronously on the connection's event-loop thread
+    /// (the same context as the connection's other callbacks). Overrides must
+    /// be cheap and non-blocking and must not call back into the connection.
+    /// A single `Endpoint` may be shared (via `shared_ptr`) by more than one
+    /// connection, and those connections may run on different event-loop
+    /// threads; an override must therefore synchronise any state it mutates.
+    virtual void onConnectSuccess() {}
+
+    /// \brief Invoked each time the connection to this endpoint is not
+    /// established and will be retried -- either an establishment attempt
+    /// failed, or a previously-established connection was lost and will
+    /// reconnect. Defaults to a no-op; implementations may override it to
+    /// observe the connection lifecycle, e.g. for monitoring or metrics.
+    ///
+    /// \warning Invoked synchronously on the connection's event-loop thread
+    /// (the same context as the connection's other callbacks). Overrides must
+    /// be cheap and non-blocking and must not call back into the connection.
+    /// A single `Endpoint` may be shared (via `shared_ptr`) by more than one
+    /// connection, and those connections may run on different event-loop
+    /// threads; an override must therefore synchronise any state it mutates.
+    ///
+    /// \note Only invoked when the connection will be retried. It is not
+    /// invoked on a terminal failure (e.g. when the connection error threshold
+    /// is breached and the failure is surfaced to the application).
+    virtual void onConnectFailed() {}
 };
 
 } // namespace rmqt

src/tests/rmqa/rmqa_connectionimpl.t.cpp

@@ -73,6 +73,7 @@ class MockConnection : public rmqamqp::Connection {
                           endpoint,
                           credentials,
                           clientProperties,
+                          bsl::optional<bsls::TimeInterval>(),
                           "Connection Name")
     , d_ackQueue(bsl::make_shared<rmqt::ConsumerAckQueue>())
     , d_receiveChannel(