feat: automatic retries in SSE client (#101)

cwaldren-ld · web-flow · commit be59e19010c6 · 2023-05-22T17:34:39.000-07:00
Adds support for automatic retries to the SSE client. I've removed the
`skip "reconnection"` suppression from the CI tests.
diff --git a/.github/workflows/sse.yml b/.github/workflows/sse.yml
@@ -36,4 +36,4 @@ jobs:
         with:
           repo: 'sse-contract-tests'
           test_service_port: ${{ env.TEST_SERVICE_PORT }}
-          extra_params: '-skip HTTP -skip reconnection'
+          extra_params: '-skip HTTP'
diff --git a/apps/sse-contract-tests/src/main.cpp b/apps/sse-contract-tests/src/main.cpp
@@ -30,14 +30,14 @@ int main(int argc, char* argv[]) {
     try {
         net::io_context ioc{1};
 
-        auto p = boost::lexical_cast<unsigned short>(port);
-        server srv(ioc, "0.0.0.0", p, logger);
+        server srv(ioc, "0.0.0.0", boost::lexical_cast<unsigned short>(port),
+                   logger);
 
         srv.add_capability("headers");
         srv.add_capability("comments");
         srv.add_capability("report");
         srv.add_capability("post");
-        srv.add_capability("read-timeout");
+        srv.add_capability("reconnection");
 
         net::signal_set signals{ioc, SIGINT, SIGTERM};
 
diff --git a/libs/server-sent-events/src/client.cpp b/libs/server-sent-events/src/client.cpp
@@ -1,9 +1,11 @@
 #include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/steady_timer.hpp>
 #include <boost/asio/strand.hpp>
 #include <boost/asio/use_future.hpp>
 #include <foxy/client_session.hpp>
 #include <launchdarkly/sse/client.hpp>
 
+#include "backoff.hpp"
 #include "parser.hpp"
 
 #include <boost/beast/http/parser.hpp>
@@ -16,6 +18,7 @@
 #include <chrono>
 #include <iostream>
 #include <memory>
+#include <sstream>
 
 namespace launchdarkly::sse {
 
@@ -40,6 +43,11 @@ static boost::optional<net::ssl::context&> ToOptRef(
 
 class FoxyClient : public Client,
                    public std::enable_shared_from_this<FoxyClient> {
+   private:
+    using cb = std::function<void(launchdarkly::sse::Event)>;
+    using body = launchdarkly::sse::detail::EventBody<cb>;
+    using response = http::response<body>;
+
    public:
     FoxyClient(boost::asio::any_io_executor executor,
                http::request<http::string_body> req,
@@ -62,16 +70,57 @@ class FoxyClient : public Client,
                    launchdarkly::foxy::session_opts{
                        ToOptRef(ssl_context_),
                        connect_timeout.value_or(kNoTimeout)}),
+          backoff_(std::chrono::seconds(1), std::chrono::seconds(30)),
+          last_event_id_(std::nullopt),
+          backoff_timer_(session_.get_executor()),
+          event_receiver_(std::move(receiver)),
           logger_(std::move(logger)) {
-        // SSE body will never end unless an error occurs, so we shouldn't set a
-        // size limit.
-        body_parser_.body_limit(boost::none);
-        body_parser_.get().body().on_event(std::move(receiver));
+        create_parser();
     }
 
-    void fail(boost::system::error_code ec, std::string const& what) {
-        logger_("sse-client: " + what + ": " + ec.message());
-        async_shutdown(nullptr);
+    /** The body parser is recreated each time a connection is made because its
+     * internal state cannot be explicitly reset.
+     *
+     * Since SSE body will never end unless
+     * an error occurs, the body size limit must be removed.
+     */
+    void create_parser() {
+        body_parser_.emplace();
+        body_parser_->body_limit(boost::none);
+        body_parser_->get().body().on_event(event_receiver_);
+    }
+
+    /**
+     * Called whenever the connection needs to be reattempted, triggering
+     * a timed wait for the current backoff duration.
+     *
+     * The body parser's last SSE event ID must be cached so it can be added
+     * as a header on the next request (since the parser is destroyed.)
+     */
+    void do_backoff(std::string const& reason) {
+        backoff_.fail();
+
+        std::stringstream msg;
+        msg << "backing off in ("
+            << std::chrono::duration_cast<std::chrono::seconds>(
+                   backoff_.delay())
+                   .count()
+            << ") seconds due to " << reason;
+
+        logger_(msg.str());
+
+        last_event_id_ = body_parser_->get().body().last_event_id();
+        create_parser();
+        backoff_timer_.expires_from_now(backoff_.delay());
+        backoff_timer_.async_wait(beast::bind_front_handler(
+            &FoxyClient::on_backoff, shared_from_this()));
+    }
+
+    void on_backoff(boost::system::error_code ec) {
+        if (ec == boost::asio::error::operation_aborted) {
+            return;
+        }
+        run();
     }
 
     void run() override {
@@ -81,24 +130,19 @@ class FoxyClient : public Client,
                                       shared_from_this()));
     }
 
-    void async_shutdown(std::function<void()> completion) override {
-        session_.async_shutdown(beast::bind_front_handler(
-            &FoxyClient::on_shutdown, std::move(completion)));
-    }
-
-    static void on_shutdown(std::function<void()> completion,
-                            boost::system::error_code ec) {
-        boost::ignore_unused(ec);
-        if (completion) {
-            completion();
-        }
-    }
-
     void on_connect(boost::system::error_code ec) {
+        if (ec == boost::asio::error::operation_aborted) {
+            return;
+        }
         if (ec) {
-            return fail(ec, "connect");
+            return do_backoff(ec.what());
         }
 
+        if (last_event_id_ && !last_event_id_->empty()) {
+            req_.set("last-event-id", *last_event_id_);
+        } else {
+            req_.erase("last-event-id");
+        }
         session_.opts.timeout = write_timeout_.value_or(kNoTimeout);
         session_.async_write(req_,
                              beast::bind_front_handler(&FoxyClient::on_write,
@@ -107,46 +151,117 @@ class FoxyClient : public Client,
 
     void on_write(boost::system::error_code ec, std::size_t amount) {
         boost::ignore_unused(amount);
+        if (ec == boost::asio::error::operation_aborted) {
+            return;
+        }
         if (ec) {
-            return fail(ec, "send request");
+            return do_backoff(ec.what());
         }
+
         session_.opts.timeout = read_timeout_.value_or(kNoTimeout);
         session_.async_read_header(
-            body_parser_, beast::bind_front_handler(&FoxyClient::on_headers,
-                                                    shared_from_this()));
+            *body_parser_, beast::bind_front_handler(&FoxyClient::on_headers,
+                                                     shared_from_this()));
     }
 
     void on_headers(boost::system::error_code ec, std::size_t amount) {
         boost::ignore_unused(amount);
+        if (ec == boost::asio::error::operation_aborted) {
+            return;
+        }
         if (ec) {
-            return fail(ec, "read header");
+            return do_backoff(ec.what());
         }
 
-        if (!body_parser_.is_header_done()) {
-            session_.async_read_header(
-                body_parser_, beast::bind_front_handler(&FoxyClient::on_headers,
-                                                        shared_from_this()));
-            return;
+        if (!body_parser_->is_header_done()) {
+            /* keep reading headers */
+            return session_.async_read_header(
+                *body_parser_,
+                beast::bind_front_handler(&FoxyClient::on_headers,
+                                          shared_from_this()));
         }
 
-        auto response = body_parser_.get();
-        if (beast::http::to_status_class(response.result()) ==
-            beast::http::status_class::successful) {
-            session_.async_read(body_parser_, beast::bind_front_handler(
-                                                  &FoxyClient::on_read_complete,
-                                                  shared_from_this()));
-        } else {
-            return fail(ec, "read response");
+        /* headers are finished, body is ready */
+        auto response = body_parser_->get();
+        auto status_class = beast::http::to_status_class(response.result());
+
+        if (status_class == beast::http::status_class::successful) {
+            if (!correct_content_type(response)) {
+                return do_backoff("invalid Content-Type");
+            }
+
+            backoff_.succeed();
+            return session_.async_read(
+                *body_parser_,
+                beast::bind_front_handler(&FoxyClient::on_read_body,
+                                          shared_from_this()));
+        }
+
+        if (status_class == beast::http::status_class::client_error) {
+            if (recoverable_client_error(response.result())) {
+                return do_backoff(backoff_reason(response.result()));
+            }
+
+            // TODO: error callback
+
+            return;
         }
+
+        do_backoff(backoff_reason(response.result()));
+    }
+
+    static std::string backoff_reason(beast::http::status status) {
+        std::stringstream ss;
+        ss << "HTTP status " << int(status) << " (" << status << ")";
+        return ss.str();
     }
 
-    void on_read_complete(boost::system::error_code ec, std::size_t amount) {
+    void on_read_body(boost::system::error_code ec, std::size_t amount) {
         boost::ignore_unused(amount);
         if (ec == boost::asio::error::operation_aborted) {
-            async_shutdown(nullptr);
-        } else {
-            return fail(ec, "read body");
+            return;
+        }
+        do_backoff(ec.what());
+    }
+
+    void async_shutdown(std::function<void()> completion) override {
+        boost::asio::post(session_.get_executor(),
+                          beast::bind_front_handler(&FoxyClient::do_shutdown,
+                                                    shared_from_this(),
+                                                    std::move(completion)));
+    }
+
+    void do_shutdown(std::function<void()> completion) {
+        session_.async_shutdown(beast::bind_front_handler(
+            &FoxyClient::on_shutdown, std::move(completion)));
+    }
+
+    static void on_shutdown(std::function<void()> completion,
+                            boost::system::error_code ec) {
+        boost::ignore_unused(ec);
+        if (completion) {
+            completion();
+        }
+    }
+
+    void fail(boost::system::error_code ec, std::string const& what) {
+        logger_("sse-client: " + what + ": " + ec.message());
+        async_shutdown(nullptr);
+    }
+
+    static bool recoverable_client_error(beast::http::status status) {
+        return (status == beast::http::status::bad_request ||
+                status == beast::http::status::request_timeout ||
+                status == beast::http::status::too_many_requests);
+    }
+
+    static bool correct_content_type(FoxyClient::response const& response) {
+        if (auto content_type = response.find("content-type");
+            content_type != response.end()) {
+            return content_type->value().find("text/event-stream") !=
+                   content_type->value().npos;
         }
+        return false;
     }
 
    private:
@@ -157,10 +272,12 @@ class FoxyClient : public Client,
     std::optional<std::chrono::milliseconds> read_timeout_;
     std::optional<std::chrono::milliseconds> write_timeout_;
     http::request<http::string_body> req_;
-    using cb = std::function<void(launchdarkly::sse::Event)>;
-    using body = launchdarkly::sse::detail::EventBody<cb>;
-    http::response_parser<body> body_parser_;
+    Builder::EventReceiver event_receiver_;
+    std::optional<http::response_parser<body> > body_parser_;
     launchdarkly::foxy::client_session session_;
+    std::optional<std::string> last_event_id_;
+    Backoff backoff_;
+    boost::asio::steady_timer backoff_timer_;
     Builder::LogCallback logger_;
 };
 
diff --git a/libs/server-sent-events/src/parser.hpp b/libs/server-sent-events/src/parser.hpp
@@ -43,8 +43,11 @@ class EventBody<EventReceiver>::value_type {
 
     EventReceiver events_;
 
+    std::optional<std::string> last_event_id_;
+
    public:
-    void on_event(EventReceiver&& receiver) { events_ = std::move(receiver); }
+    void on_event(EventReceiver receiver) { events_ = std::move(receiver); }
+    std::optional<std::string> last_event_id() const { return last_event_id_; }
 };
 
 template <class EventReceiver>
@@ -54,7 +57,7 @@ struct EventBody<EventReceiver>::reader {
     std::optional<std::string> buffered_line_;
     std::deque<std::string> complete_lines_;
     bool begin_CR_;
-    std::optional<std::string> last_event_id_;
+
     std::optional<Event> event_;
 
    public:
@@ -64,7 +67,6 @@ struct EventBody<EventReceiver>::reader {
           buffered_line_(),
           complete_lines_(),
           begin_CR_(false),
-          last_event_id_(),
           event_() {
         boost::ignore_unused(h);
     }
@@ -80,7 +82,6 @@ struct EventBody<EventReceiver>::reader {
     void init(boost::optional<std::uint64_t> const& content_length,
               error_code& ec) {
         boost::ignore_unused(content_length);
-
         // The specification requires this to indicate "no error"
         ec = {};
     }
@@ -207,7 +208,7 @@ struct EventBody<EventReceiver>::reader {
 
                 if (!event_.has_value()) {
                     event_.emplace(Event{});
-                    event_->id = last_event_id_;
+                    event_->id = body_.last_event_id_;
                 }
 
                 if (field.first == "event") {
@@ -220,8 +221,8 @@ struct EventBody<EventReceiver>::reader {
                         // ignored.
                         continue;
                     }
-                    last_event_id_ = field.second;
-                    event_->id = last_event_id_;
+                    body_.last_event_id_ = field.second;
+                    event_->id = body_.last_event_id_;
                 } else if (field.first == "retry") {
                     // todo: implement
                 }
