fix: improve handling of streaming error state changes/logging

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

@@ -21,15 +21,15 @@
 
 StreamingDataSource::StreamingDataSource(
     config::shared::built::ServiceEndpoints const& endpoints,
     config::shared::built::DataSourceConfig<config::shared::ClientSDK> const&
         data_source_config,
     config::shared::built::HttpProperties const& http_properties,
     boost::asio::any_io_executor ioc,
     Context context,
     IDataSourceUpdateSink& handler,
     DataSourceStatusManager& status_manager,
     Logger const& logger)
     : exec_(ioc),
       logger_(logger),
       context_(std::move(context)),
       status_manager_(status_manager),
@@ -99,7 +99,7 @@
     // TODO: can the read timeout be shared with *all* http requests? Or should
     // it have a default in defaults.hpp? This must be greater than the
     // heartbeat interval of the streaming service.
     client_builder.read_timeout(std::chrono::minutes(5));
 
     client_builder.write_timeout(http_config_.WriteTimeout());
 
@@ -140,12 +140,11 @@
 
     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 @@
     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