fix: improve handling of streaming error state changes/logging (#439)

The existing handling for data source state transitions when receiving
`sse` errors was incorrect. It was treating any error as a state
transition to `kOff/kShutdown`, and logging at the `error` level.

The correct behavior would be making that transition only if the error
is unrecoverable. Additionally, recoverable errors should be logged at
the debug level, in order to not overstate the impact of the issue.

This bug did not actually affect the backoff behavior of the
eventsource, but it did impact status listeners and any code that queried the existing status within the SDK.

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

@@ -140,12 +140,11 @@ void StreamingDataSource::Start() {
 
     client_builder.errors([weak_self](auto error) {
         if (auto self = weak_self.lock()) {
-            auto error_string = launchdarkly::sse::ErrorToString(error);
-            LD_LOG(self->logger_, LogLevel::kError) << error_string;
-            self->status_manager_.SetState(
-                DataSourceStatus::DataSourceState::kShutdown,
-                DataSourceStatus::ErrorInfo::ErrorKind::kErrorResponse,
-                std::move(error_string));
+            std::string error_string = sse::ErrorToString(error);
+            LD_LOG(self->logger_, sse::IsRecoverable(error) ? LogLevel::kDebug
+                                                            : LogLevel::kError);
+            self->HandleErrorStateChange(std::move(error),
+                                         std::move(error_string));
         }
     });
 
@@ -162,6 +161,52 @@ void StreamingDataSource::Start() {
     client_->async_connect();
 }
 
+template <class>
+inline constexpr bool always_false_v = false;
+
+void StreamingDataSource::HandleErrorStateChange(sse::Error error,
+                                                 std::string error_string) {
+    auto const state = sse::IsRecoverable(error) ? DataSourceState::kInterrupted
+                                                 : DataSourceState::kShutdown;
+    std::visit(
+        [this, state, error_string = std::move(error_string)](auto error) {
+            using T = std::decay_t<decltype(error)>;
+            if constexpr (std::is_same_v<T, sse::errors::ReadTimeout>) {
+                this->status_manager_.SetState(
+                    state,
+                    DataSourceStatus::ErrorInfo::ErrorKind::kNetworkError,
+                    std::move(error_string));
+
+            } else if constexpr (std::is_same_v<
+                                     T,
+                                     sse::errors::UnrecoverableClientError>) {
+                this->status_manager_.SetState(
+                    state,
+                    static_cast<DataSourceStatusManager::StatusCodeType>(
+                        error.status),
+                    std::move(error_string));
+
+            } else if constexpr (std::is_same_v<
+                                     T, sse::errors::InvalidRedirectLocation>) {
+                this->status_manager_.SetState(
+                    state,
+                    DataSourceStatus::ErrorInfo::ErrorKind::kNetworkError,
+                    std::move(error_string));
+
+            } else if constexpr (std::is_same_v<T,
+                                                sse::errors::NotRedirectable>) {
+                this->status_manager_.SetState(
+                    state,
+                    DataSourceStatus::ErrorInfo::ErrorKind::kNetworkError,
+                    std::move(error_string));
+            } else {
+                static_assert(always_false_v<decltype(error)>,
+                              "non-exhaustive visitor");
+            }
+        },
+        std::move(error));
+}
+
 void StreamingDataSource::ShutdownAsync(std::function<void()> completion) {
     if (client_) {
         status_manager_.SetState(

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

@@ -39,6 +39,8 @@ class StreamingDataSource final
     void ShutdownAsync(std::function<void()>) override;
 
    private:
+    void HandleErrorStateChange(sse::Error error, std::string error_string);
+
     Context context_;
     boost::asio::any_io_executor exec_;
     DataSourceStatusManager& status_manager_;

libs/internal/include/launchdarkly/data_sources/data_source_status_manager_base.hpp

@@ -18,6 +18,9 @@ namespace launchdarkly::internal::data_sources {
 template <typename TDataSourceStatus, typename TInterface>
 class DataSourceStatusManagerBase : public TInterface {
    public:
+    using StatusCodeType =
+        typename TDataSourceStatus::ErrorInfo::StatusCodeType;
+
     /**
      * Set the state.
      *
@@ -39,7 +42,7 @@ class DataSourceStatusManagerBase : public TInterface {
      * @param message The message to associate with the error.
      */
     void SetState(typename TDataSourceStatus::DataSourceState state,
-                  typename TDataSourceStatus::ErrorInfo::StatusCodeType code,
+                  StatusCodeType code,
                   std::string message) {
         {
             std::lock_guard lock(status_mutex_);

libs/server-sdk/src/data_systems/background_sync/sources/streaming/streaming_data_source.cpp

@@ -137,12 +137,11 @@ void StreamingDataSource::StartAsync(
 
     client_builder.errors([weak_self](auto error) {
         if (auto self = weak_self.lock()) {
-            std::string error_string = launchdarkly::sse::ErrorToString(error);
-            LD_LOG(self->logger_, LogLevel::kError) << error_string;
-            self->status_manager_.SetState(
-                DataSourceStatus::DataSourceState::kOff,
-                DataSourceStatus::ErrorInfo::ErrorKind::kErrorResponse,
-                std::move(error_string));
+            std::string error_string = sse::ErrorToString(error);
+            LD_LOG(self->logger_, sse::IsRecoverable(error) ? LogLevel::kDebug
+                                                            : LogLevel::kError);
+            self->HandleErrorStateChange(std::move(error),
+                                         std::move(error_string));
         }
     });
 
@@ -159,6 +158,52 @@ void StreamingDataSource::StartAsync(
     client_->async_connect();
 }
 
+template <class>
+inline constexpr bool always_false_v = false;
+
+void StreamingDataSource::HandleErrorStateChange(sse::Error error,
+                                                 std::string error_string) {
+    auto const state = sse::IsRecoverable(error) ? DataSourceState::kInterrupted
+                                                 : DataSourceState::kOff;
+    std::visit(
+        [this, state, error_string = std::move(error_string)](auto error) {
+            using T = std::decay_t<decltype(error)>;
+            if constexpr (std::is_same_v<T, sse::errors::ReadTimeout>) {
+                this->status_manager_.SetState(
+                    state,
+                    DataSourceStatus::ErrorInfo::ErrorKind::kNetworkError,
+                    std::move(error_string));
+
+            } else if constexpr (std::is_same_v<
+                                     T,
+                                     sse::errors::UnrecoverableClientError>) {
+                this->status_manager_.SetState(
+                    state,
+                    static_cast<data_components::DataSourceStatusManager::
+                                    StatusCodeType>(error.status),
+                    std::move(error_string));
+
+            } else if constexpr (std::is_same_v<
+                                     T, sse::errors::InvalidRedirectLocation>) {
+                this->status_manager_.SetState(
+                    state,
+                    DataSourceStatus::ErrorInfo::ErrorKind::kNetworkError,
+                    std::move(error_string));
+
+            } else if constexpr (std::is_same_v<T,
+                                                sse::errors::NotRedirectable>) {
+                this->status_manager_.SetState(
+                    state,
+                    DataSourceStatus::ErrorInfo::ErrorKind::kNetworkError,
+                    std::move(error_string));
+            } else {
+                static_assert(always_false_v<decltype(error)>,
+                              "non-exhaustive visitor");
+            }
+        },
+        std::move(error));
+}
+
 void StreamingDataSource::ShutdownAsync(std::function<void()> completion) {
     if (client_) {
         status_manager_.SetState(

libs/server-sdk/src/data_systems/background_sync/sources/streaming/streaming_data_source.hpp

@@ -33,6 +33,8 @@ class StreamingDataSource final
     [[nodiscard]] std::string const& Identity() const override;
 
    private:
+    void HandleErrorStateChange(sse::Error error, std::string error_string);
+
     boost::asio::any_io_executor io_;
     Logger const& logger_;
 

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

@@ -9,9 +9,6 @@
 namespace launchdarkly::sse {
 namespace errors {
 
-struct NoContent {};
-std::ostream& operator<<(std::ostream& out, NoContent const&);
-
 struct InvalidRedirectLocation {
     std::string location;
 };
@@ -32,12 +29,13 @@ std::ostream& operator<<(std::ostream& out, UnrecoverableClientError const&);
 
 }  // namespace errors
 
-using Error = std::variant<errors::NoContent,
-                           errors::InvalidRedirectLocation,
+using Error = std::variant<errors::InvalidRedirectLocation,
                            errors::NotRedirectable,
                            errors::ReadTimeout,
                            errors::UnrecoverableClientError>;
 
+bool IsRecoverable(Error const& error);
+
 std::ostream& operator<<(std::ostream& out, Error const& error);
 
 std::string ErrorToString(Error const& error);

libs/server-sent-events/src/client.cpp

@@ -222,7 +222,8 @@ class FoxyClient : public Client,
 
         if (status_class == beast::http::status_class::successful) {
             if (response.result() == beast::http::status::no_content) {
-                errors_(errors::NoContent{});
+                errors_(
+                    errors::UnrecoverableClientError{http::status::no_content});
                 return;
             }
             if (!correct_content_type(response)) {
@@ -355,7 +356,6 @@ class FoxyClient : public Client,
             logger_("exception closing stream: " + std::string(err.what()));
         }
 
-
         // Ideally we would call session_->async_shutdown() here to gracefully
         // terminate the SSL session. For unknown reasons, this call appears to
         // hang indefinitely and never complete until the SDK client is

libs/server-sent-events/src/error.cpp

@@ -5,12 +5,6 @@
 namespace launchdarkly::sse {
 namespace errors {
 
-std::ostream& operator<<(std::ostream& out, NoContent const&) {
-    out << "received HTTP error 204 (no content) - giving up "
-           "permanently";
-    return out;
-}
-
 std::ostream& operator<<(std::ostream& out,
                          InvalidRedirectLocation const& invalid) {
     out << "received invalid redirect from server, cannot follow ("
@@ -60,4 +54,8 @@ std::string ErrorToString(Error const& error) {
     return ss.str();
 }
 
+bool IsRecoverable(Error const& error) {
+    return std::holds_alternative<errors::ReadTimeout>(error);
+}
+
 }  // namespace launchdarkly::sse