fix: stream connections longer than 5 minutes are dropped (#244)

This fixes [streaming connection dropping after 5
minutes](#243).

The problem was applying a 5 minute timeout to an `async_read`
_operation_, when the actual intent was to apply it to the act of
receiving _any data_.

The timeout exists to prevent the client from hanging infinitely if the
server stops responding. LaunchDarkly sends heartbeats to clear the
timer.

With `async_read`, the operation won't complete until the response body
is finished. Since the body will never finish until the client shuts
down (or server interrupts it), it's not possible to institute a
timeout.

The solution is to use `async_read_some`, which completes whenever data
arrives.

This way, we can assign the 5 minute timeout as an upper bound on
receiving any data, as was originally intended.

-----

A side effect of this change was breaking the `reconnection` SSE
contract tests, which proved to be somewhat of a rabbithole.

The upshot is that we weren't handling chunked encoding properly when
chunked encoding _ends_. The original code assumes we either terminate
an async read with an error, or it goes on forever.

In fact it should perform the backoff algorithm if we detect the end of
chunked encoding.

Author: cwaldren-ld

contract-tests/sse-contract-tests/include/event_outbox.hpp

@@ -72,11 +72,11 @@ class EventOutbox : public std::enable_shared_from_this<EventOutbox> {
    private:
     RequestType build_request(
         std::size_t counter,
-        std::variant<launchdarkly::sse::Event, launchdarkly::sse::Error> ev);
+        std::variant<launchdarkly::sse::Event, launchdarkly::sse::Error> event);
     void on_resolve(beast::error_code ec, tcp::resolver::results_type results);
     void on_connect(beast::error_code ec,
                     tcp::resolver::results_type::endpoint_type);
     void on_flush_timer(boost::system::error_code ec);
     void on_write(beast::error_code ec, std::size_t);
-    void do_shutdown(beast::error_code ec, std::string what);
+    void do_shutdown(beast::error_code ec);
 };

contract-tests/sse-contract-tests/src/entity_manager.cpp

@@ -27,24 +27,30 @@ std::optional<std::string> EntityManager::create(ConfigParams const& params) {
     }
 
     if (params.body) {
-        client_builder.body(std::move(*params.body));
+        client_builder.body(*params.body);
     }
 
     if (params.readTimeoutMs) {
         client_builder.read_timeout(
             std::chrono::milliseconds(*params.readTimeoutMs));
     }
 
+    if (params.initialDelayMs) {
+        client_builder.initial_reconnect_delay(
+            std::chrono::milliseconds(*params.initialDelayMs));
+    }
+
     client_builder.logger([this](std::string msg) {
         LD_LOG(logger_, LogLevel::kDebug) << std::move(msg);
     });
 
-    client_builder.receiver([copy = poster](launchdarkly::sse::Event e) {
-        copy->post_event(std::move(e));
+    client_builder.receiver([copy = poster](launchdarkly::sse::Event event) {
+        copy->post_event(std::move(event));
     });
 
-    client_builder.errors(
-        [copy = poster](launchdarkly::sse::Error e) { copy->post_error(e); });
+    client_builder.errors([copy = poster](launchdarkly::sse::Error event) {
+        copy->post_error(event);
+    });
 
     auto client = client_builder.build();
     if (!client) {
@@ -53,7 +59,7 @@ std::optional<std::string> EntityManager::create(ConfigParams const& params) {
         return std::nullopt;
     }
 
-    client->run();
+    client->async_connect();
 
     entities_.emplace(id, std::make_pair(client, poster));
     return id;

contract-tests/sse-contract-tests/src/event_outbox.cpp

@@ -14,8 +14,6 @@ auto const kOutboxCapacity = 1023;
 EventOutbox::EventOutbox(net::any_io_executor executor,
                          std::string callback_url)
     : callback_url_{std::move(callback_url)},
-      callback_port_{},
-      callback_host_{},
       callback_counter_{0},
       executor_{executor},
       resolver_{executor},
@@ -29,7 +27,7 @@ EventOutbox::EventOutbox(net::any_io_executor executor,
     callback_port_ = uri_components->port();
 }
 
-void EventOutbox::do_shutdown(beast::error_code ec, std::string what) {
+void EventOutbox::do_shutdown(beast::error_code ec) {
     event_stream_.socket().shutdown(tcp::socket::shutdown_both, ec);
     flush_timer_.cancel();
 }
@@ -54,20 +52,18 @@ void EventOutbox::run() {
 
 void EventOutbox::stop() {
     beast::error_code ec = net::error::basic_errors::operation_aborted;
-    std::string reason = "stop";
     shutdown_ = true;
-    net::post(executor_,
-              beast::bind_front_handler(&EventOutbox::do_shutdown,
-                                        shared_from_this(), ec, reason));
+    net::post(executor_, beast::bind_front_handler(&EventOutbox::do_shutdown,
+                                                   shared_from_this(), ec));
 }
 
 EventOutbox::RequestType EventOutbox::build_request(
     std::size_t counter,
-    std::variant<launchdarkly::sse::Event, launchdarkly::sse::Error> ev) {
+    std::variant<launchdarkly::sse::Event, launchdarkly::sse::Error> event) {
     RequestType req;
 
     req.set(http::field::host, callback_host_);
-    req.method(http::verb::get);
+    req.method(http::verb::post);
     req.target(callback_url_ + "/" + std::to_string(counter));
 
     nlohmann::json json;
@@ -93,13 +89,15 @@ EventOutbox::RequestType EventOutbox::build_request(
                         break;
                     case Error::UnrecoverableClientError:
                         msg.comment = "unrecoverable client error";
+                    case Error::ReadTimeout:
+                        msg.comment = "read timeout";
                     default:
                         msg.comment = "unspecified error";
                 }
                 json = msg;
             }
         },
-        std::move(ev));
+        std::move(event));
 
     req.body() = json.dump();
     req.prepare_payload();
@@ -109,7 +107,7 @@ EventOutbox::RequestType EventOutbox::build_request(
 void EventOutbox::on_resolve(beast::error_code ec,
                              tcp::resolver::results_type results) {
     if (ec) {
-        return do_shutdown(ec, "resolve");
+        return do_shutdown(ec);
     }
 
     beast::get_lowest_layer(event_stream_)
@@ -121,7 +119,7 @@ void EventOutbox::on_resolve(beast::error_code ec,
 void EventOutbox::on_connect(beast::error_code ec,
                              tcp::resolver::results_type::endpoint_type) {
     if (ec) {
-        return do_shutdown(ec, "connect");
+        return do_shutdown(ec);
     }
 
     boost::system::error_code dummy;
@@ -131,7 +129,7 @@ void EventOutbox::on_connect(beast::error_code ec,
 
 void EventOutbox::on_flush_timer(boost::system::error_code ec) {
     if (ec && shutdown_) {
-        return do_shutdown(ec, "flush");
+        return do_shutdown(ec);
     }
 
     if (!outbox_.empty()) {
@@ -154,7 +152,7 @@ void EventOutbox::on_flush_timer(boost::system::error_code ec) {
 
 void EventOutbox::on_write(beast::error_code ec, std::size_t) {
     if (ec) {
-        return do_shutdown(ec, "write");
+        return do_shutdown(ec);
     }
     outbox_.pop();
     on_flush_timer(boost::system::error_code{});

contract-tests/sse-contract-tests/src/main.cpp

@@ -20,7 +20,7 @@ int main(int argc, char* argv[]) {
     launchdarkly::Logger logger{
         std::make_unique<ConsoleBackend>("sse-contract-tests")};
 
-    const std::string default_port = "8123";
+    std::string const default_port = "8123";
     std::string port = default_port;
     if (argc == 2) {
         port =
@@ -38,6 +38,7 @@ int main(int argc, char* argv[]) {
         srv.add_capability("report");
         srv.add_capability("post");
         srv.add_capability("reconnection");
+        srv.add_capability("read-timeout");
 
         net::signal_set signals{ioc, SIGINT, SIGTERM};
 

libs/client-sdk/src/data_sources/streaming_data_source.cpp

@@ -27,6 +27,8 @@ static char const* DataSourceErrorToString(launchdarkly::sse::Error error) {
             return "server responded with an invalid redirection";
         case sse::Error::UnrecoverableClientError:
             return "unrecoverable client-side error";
+        case sse::Error::ReadTimeout:
+            return "read timeout reached";
     }
 }
 
@@ -138,7 +140,7 @@ void StreamingDataSource::Start() {
 
     client_builder.logger([weak_self](auto msg) {
         if (auto self = weak_self.lock()) {
-            LD_LOG(self->logger_, LogLevel::kDebug) << msg;
+            LD_LOG(self->logger_, LogLevel::kDebug) << "sse-client: " << msg;
         }
     });
 
@@ -163,7 +165,7 @@ void StreamingDataSource::Start() {
             kCouldNotParseEndpoint);
         return;
     }
-    client_->run();
+    client_->async_connect();
 }
 
 void StreamingDataSource::ShutdownAsync(std::function<void()> completion) {

libs/common/src/config/logging_builder.cpp

@@ -60,7 +60,9 @@ LoggingBuilder::BasicLogging& LoggingBuilder::BasicLogging::Tag(
     return *this;
 }
 LoggingBuilder::BasicLogging::BasicLogging()
-    : level_(Defaults<AnySDK>::LogLevel()), tag_(Defaults<AnySDK>::LogTag()) {}
+    : level_(GetLogLevelEnum(std::getenv("LD_LOG_LEVEL"),
+                             Defaults<AnySDK>::LogLevel())),
+      tag_(Defaults<AnySDK>::LogTag()) {}
 
 LoggingBuilder::CustomLogging& LoggingBuilder::CustomLogging::Backend(
     std::shared_ptr<ILogBackend> backend) {

libs/server-sent-events/include/launchdarkly/sse/client.hpp

@@ -160,7 +160,7 @@ class Builder {
 class Client {
    public:
     virtual ~Client() = default;
-    virtual void run() = 0;
+    virtual void async_connect() = 0;
     virtual void async_shutdown(std::function<void()> completion) = 0;
 };
 

libs/server-sent-events/include/launchdarkly/sse/error.hpp

@@ -6,5 +6,6 @@ enum class Error {
     NoContent = 1,
     InvalidRedirectLocation = 2,
     UnrecoverableClientError = 3,
+    ReadTimeout = 4,
 };
 }