From 770018c640f7c5eba31e5c48dc59c46cae211d22 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Wed, 2 Sep 2026 03:11:29 +0000 Subject: [PATCH 1/7] feat: Add the FDv2 streaming synchronizer to the client --- libs/client-sdk/src/CMakeLists.txt | 2 + .../fdv2/fdv2_response_headers.cpp | 25 + .../fdv2/fdv2_response_headers.hpp | 5 + .../fdv2/streaming_synchronizer.cpp | 473 ++++++++++++++++ .../fdv2/streaming_synchronizer.hpp | 178 ++++++ libs/client-sdk/tests/CMakeLists.txt | 2 +- .../fdv2_streaming_synchronizer_test.cpp | 523 ++++++++++++++++++ 7 files changed, 1207 insertions(+), 1 deletion(-) create mode 100644 libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.cpp create mode 100644 libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.hpp create mode 100644 libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp diff --git a/libs/client-sdk/src/CMakeLists.txt b/libs/client-sdk/src/CMakeLists.txt index 589fb66f0..dcf8025a4 100644 --- a/libs/client-sdk/src/CMakeLists.txt +++ b/libs/client-sdk/src/CMakeLists.txt @@ -20,6 +20,7 @@ target_sources(${LIBNAME} PRIVATE data_sources/fdv2/fdv2_polling_impl.cpp data_sources/fdv2/polling_initializer.cpp data_sources/fdv2/polling_synchronizer.cpp + data_sources/fdv2/streaming_synchronizer.cpp data_sources/data_source_event_handler.cpp data_sources/polling_data_source.cpp flag_manager/flag_store.cpp @@ -45,6 +46,7 @@ target_sources(${LIBNAME} PRIVATE data_sources/fdv2/fdv2_polling_impl.hpp data_sources/fdv2/polling_initializer.hpp data_sources/fdv2/polling_synchronizer.hpp + data_sources/fdv2/streaming_synchronizer.hpp flag_manager/flag_store.hpp flag_manager/flag_updater.hpp bindings/c/sdk.cpp diff --git a/libs/client-sdk/src/data_sources/fdv2/fdv2_response_headers.cpp b/libs/client-sdk/src/data_sources/fdv2/fdv2_response_headers.cpp index 912e46d66..37e84ec77 100644 --- a/libs/client-sdk/src/data_sources/fdv2/fdv2_response_headers.cpp +++ b/libs/client-sdk/src/data_sources/fdv2/fdv2_response_headers.cpp @@ -32,4 +32,29 @@ FDv2ResponseHeaders ReadFDv2ResponseHeaders( return result; } +FDv2ResponseHeaders ReadFDv2ResponseHeaders( + boost::beast::http::response_header<> const& headers) { + FDv2ResponseHeaders result; + + if (auto const it = headers.find(kEnvironmentIdHeader); + it != headers.end()) { + result.environment_id = std::string{it->value()}; + } + + auto const fallback = headers.find(kFDv1FallbackHeader); + if (fallback == headers.end() || + !boost::iequals(fallback->value(), "true")) { + return result; + } + + auto const ttl = headers.find(kFDv1FallbackTtlHeader); + result.fdv1_fallback = + ttl == headers.end() + ? FDv1FallbackDirective::FromServiceTtl(std::nullopt) + : FDv1FallbackDirective::FromServiceTtl( + std::string_view{ttl->value().data(), ttl->value().size()}); + + return result; +} + } // namespace launchdarkly::client_side::data_sources diff --git a/libs/client-sdk/src/data_sources/fdv2/fdv2_response_headers.hpp b/libs/client-sdk/src/data_sources/fdv2/fdv2_response_headers.hpp index 959030d8b..4504a7a9d 100644 --- a/libs/client-sdk/src/data_sources/fdv2/fdv2_response_headers.hpp +++ b/libs/client-sdk/src/data_sources/fdv2/fdv2_response_headers.hpp @@ -4,6 +4,8 @@ #include +#include + #include #include @@ -20,4 +22,7 @@ struct FDv2ResponseHeaders { FDv2ResponseHeaders ReadFDv2ResponseHeaders( network::HttpResult::HeadersType const& headers); +FDv2ResponseHeaders ReadFDv2ResponseHeaders( + boost::beast::http::response_header<> const& headers); + } // namespace launchdarkly::client_side::data_sources diff --git a/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.cpp b/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.cpp new file mode 100644 index 000000000..5cc214e67 --- /dev/null +++ b/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.cpp @@ -0,0 +1,473 @@ +#include "streaming_synchronizer.hpp" +#include "fdv2_changeset_translation.hpp" +#include "fdv2_polling_impl.hpp" +#include "fdv2_response_headers.hpp" + +#include + +#include +#include + +#include + +namespace launchdarkly::client_side::data_sources { + +static char const* const kIdentity = "FDv2 streaming synchronizer"; + +static char const* const kPingEvent = "ping"; + +// Maximum time between bytes read from the stream before the SSE client +// declares the connection dead and reconnects. Must be greater than the +// streaming service's heartbeat interval. Hardcoded rather than read from +// HttpProperties, whose default ReadTimeout is sized for one-shot HTTP +// requests and would cause spurious disconnects on a long-lived stream. +static constexpr std::chrono::minutes kDeadConnectionInterval{5}; + +using ErrorInfo = FDv2SourceResult::ErrorInfo; +using ErrorKind = ErrorInfo::ErrorKind; + +static ErrorInfo MakeError(ErrorKind kind, + ErrorInfo::StatusCodeType status, + std::string message) { + return ErrorInfo{kind, status, std::move(message), + std::chrono::system_clock::now()}; +} + +template +inline constexpr bool always_false_v = false; + +FDv2StreamingSynchronizer::State::State( + Logger const& logger, + boost::asio::any_io_executor const& executor, + FDv2RequestConfig const& stream_config, + FDv2RequestConfig const& poll_config, + std::chrono::milliseconds initial_reconnect_delay) + : logger_(logger), + stream_config_(stream_config), + poll_config_(poll_config), + initial_reconnect_delay_(initial_reconnect_delay), + executor_(executor), + requester_(executor, poll_config.http_properties.Tls()) {} + +void FDv2StreamingSynchronizer::State::EnsureStarted( + data_model::Selector const& selector, + std::shared_ptr self) { + { + std::lock_guard lock(mutex_); + latest_selector_ = selector; + if (closed_ || started_) { + return; + } + started_ = true; + } + + bool const post = + stream_config_.transport == FDv2ContextTransport::kPostBody; + + auto parsed = boost::urls::parse_uri(stream_config_.base_url); + if (!parsed) { + // started_ intentionally left true: a bad endpoint URL is a + // configuration error that won't fix itself. The TerminalError + // result tells the orchestrator to stop retrying this synchronizer. + LD_LOG(logger_, LogLevel::kError) + << kIdentity << ": could not parse streaming endpoint URL"; + Notify(FDv2SourceResult{FDv2SourceResult::TerminalError{ + MakeError(ErrorKind::kNetworkError, 0, + "could not parse streaming endpoint URL")}}); + return; + } + + boost::urls::url url = parsed.value(); + + // A trailing '/' on the base URL appears as an empty final segment. + // Remove it so the pushed segments do not produce a double slash. + auto segments = url.segments(); + if (!segments.empty() && segments.back().empty()) { + segments.pop_back(); + } + segments.push_back("sdk"); + segments.push_back("stream"); + segments.push_back("eval"); + if (!post) { + segments.push_back( + encoding::Base64UrlEncode(stream_config_.serialized_context)); + } + if (stream_config_.with_reasons) { + url.params().set("withReasons", "true"); + } + + // The basis parameter is added by the on-connect hook instead, so that + // each reconnection uses the freshest selector. + { + std::lock_guard lock(mutex_); + base_url_ = url; + } + + auto builder = sse::Builder(executor_, std::string(url.buffer())); + + builder.method(post ? boost::beast::http::verb::post + : boost::beast::http::verb::get); + if (post) { + builder.header("content-type", "application/json"); + builder.body(stream_config_.serialized_context); + } + builder.read_timeout(kDeadConnectionInterval); + builder.write_timeout(stream_config_.http_properties.WriteTimeout()); + builder.connect_timeout(stream_config_.http_properties.ConnectTimeout()); + builder.initial_reconnect_delay(initial_reconnect_delay_); + + for (auto const& [key, value] : + stream_config_.http_properties.BaseHeaders()) { + builder.header(key, value); + } + if (stream_config_.http_properties.Tls().PeerVerifyMode() == + config::shared::built::TlsOptions::VerifyMode::kVerifyNone) { + builder.skip_verify_peer(true); + } + if (auto ca_file = stream_config_.http_properties.Tls().CustomCAFile()) { + builder.custom_ca_file(*ca_file); + } + if (auto proxy_url = stream_config_.http_properties.Proxy().Url()) { + builder.proxy(*proxy_url); + } + + std::weak_ptr weak = self; + builder.on_connect([weak](HttpRequest* req) { + if (auto s = weak.lock()) { + s->OnConnect(req); + } + }); + builder.on_response([weak](HttpResponseHeader const& headers) { + if (auto s = weak.lock()) { + s->OnResponse(headers); + } + }); + builder.receiver([weak](sse::Event const& event) { + if (auto s = weak.lock()) { + s->OnEvent(event); + } + }); + builder.logger([weak](std::string msg) { + if (auto s = weak.lock()) { + LD_LOG(s->logger_, LogLevel::kDebug) << "sse-client: " << msg; + } + }); + builder.errors([weak](sse::Error error) { + if (auto s = weak.lock()) { + s->OnError(error); + } + }); + + auto client = builder.build(); + if (!client) { + // started_ intentionally left true: same reasoning as above. + LD_LOG(logger_, LogLevel::kError) + << kIdentity << ": could not build SSE client"; + Notify(FDv2SourceResult{FDv2SourceResult::TerminalError{MakeError( + ErrorKind::kNetworkError, 0, "could not build SSE client")}}); + return; + } + + // Publishing the client and connecting are atomic with respect to + // Shutdown(). A client built after Shutdown ran is discarded without + // connecting, so there is nothing to clean up. + std::lock_guard lock(mutex_); + if (closed_) { + return; + } + sse_client_ = client; + client->async_connect(); +} + +void FDv2StreamingSynchronizer::State::OnConnect(HttpRequest* req) { + std::lock_guard lock(mutex_); + // base_url_ is guaranteed populated. EnsureStarted publishes it before + // calling async_connect, which is what eventually triggers this hook. + boost::urls::url url = *base_url_; + if (latest_selector_.value) { + url.params().set("basis", latest_selector_.value->state); + } + req->target(url.encoded_target()); +} + +void FDv2StreamingSynchronizer::State::OnResponse( + HttpResponseHeader const& headers) { + auto read = ReadFDv2ResponseHeaders(headers); + + std::lock_guard lock(mutex_); + latest_fdv1_fallback_ = std::move(read.fdv1_fallback); + if (read.environment_id) { + latest_environment_id_ = std::move(read.environment_id); + } +} + +void FDv2StreamingSynchronizer::State::PollForPing( + std::shared_ptr self) { + { + std::lock_guard lock(mutex_); + if (closed_ || ping_poll_in_flight_) { + return; + } + ping_poll_in_flight_ = true; + } + + LD_LOG(logger_, LogLevel::kDebug) + << kIdentity << ": ping received, polling for the current payload"; + + data_model::Selector selector; + { + std::lock_guard lock(mutex_); + selector = latest_selector_; + } + + auto request = MakeFDv2PollRequest(poll_config_, selector); + requester_.Request( + request, [self = std::move(self)](network::HttpResult const& res) { + FDv2ProtocolHandler handler; + auto result = + HandleFDv2PollResponse(res, &handler, self->logger_, kIdentity); + { + std::lock_guard lock(self->mutex_); + self->ping_poll_in_flight_ = false; + } + self->Notify(std::move(result)); + }); +} + +void FDv2StreamingSynchronizer::State::OnEvent(sse::Event const& event) { + if (event.type() == kPingEvent) { + PollForPing(shared_from_this()); + return; + } + + if (!FDv2ProtocolHandler::IsKnownEvent(event.type())) { + return; + } + + boost::system::error_code ec; + auto data = boost::json::parse(event.data(), ec); + if (ec) { + protocol_handler_.Reset(); + std::string msg = "could not parse FDv2 streaming event payload"; + LD_LOG(logger_, LogLevel::kError) << kIdentity << ": " << msg; + Notify(FDv2SourceResult{FDv2SourceResult::Interrupted{ + MakeError(ErrorKind::kInvalidData, 0, std::move(msg))}}); + std::lock_guard lock(mutex_); + if (sse_client_) { + sse_client_->async_restart("FDv2 parse error"); + } + return; + } + + auto result = protocol_handler_.HandleEvent(event.type(), data); + + std::visit( + [this](auto const& r) { + using T = std::decay_t; + if constexpr (std::is_same_v) { + // Accumulating, heartbeat, or unknown event — nothing to do. + } else if constexpr (std::is_same_v) { + auto typed = TranslateChangeSet(r, logger_); + if (!typed) { + std::string msg = + "FDv2 streaming changeset could not be translated"; + LD_LOG(logger_, LogLevel::kError) + << kIdentity << ": " << msg; + Notify(FDv2SourceResult{ + FDv2SourceResult::Interrupted{MakeError( + ErrorKind::kInvalidData, 0, std::move(msg))}}); + return; + } + Notify(FDv2SourceResult{ + FDv2SourceResult::ChangeSet{std::move(*typed)}}); + } else if constexpr (std::is_same_v) { + LD_LOG(logger_, LogLevel::kInfo) + << kIdentity + << ": Goodbye was received from the LaunchDarkly " + "connection with reason: '" + << r.reason.value_or("") << "'."; + FDv2SourceResult goodbye_result{ + FDv2SourceResult::Goodbye{r.reason}}; + if (r.protocol_fallback_ttl) { + goodbye_result.fdv1_fallback = + FDv1FallbackDirective::FromServiceTtl( + std::chrono::seconds(*r.protocol_fallback_ttl)); + } + Notify(std::move(goodbye_result)); + // Drop the current connection and reconnect. The protocol + // handler is reset so the new connection starts in a clean + // state. + protocol_handler_.Reset(); + std::lock_guard lock(mutex_); + if (sse_client_) { + sse_client_->async_restart("FDv2 goodbye received"); + } + } else if constexpr (std::is_same_v) { + if (r.kind == FDv2ProtocolHandler::Error::Kind::kServerError) { + auto const& id = r.server_error.value().id; + std::string msg = + "An issue was encountered receiving updates for " + "payload '" + + id.value_or("") + "' with reason: '" + r.message + + "'. Automatic retry will occur."; + LD_LOG(logger_, LogLevel::kInfo) + << kIdentity << ": " << msg; + Notify(FDv2SourceResult{ + FDv2SourceResult::Interrupted{MakeError( + ErrorKind::kErrorResponse, 0, std::move(msg))}}); + return; + } + LD_LOG(logger_, LogLevel::kError) + << kIdentity << ": " << r.message; + Notify(FDv2SourceResult{FDv2SourceResult::Interrupted{ + MakeError(ErrorKind::kInvalidData, 0, r.message)}}); + std::lock_guard lock(mutex_); + if (sse_client_) { + sse_client_->async_restart("FDv2 protocol error"); + } + } else { + static_assert(always_false_v, "non-exhaustive visitor"); + } + }, + result); +} + +void FDv2StreamingSynchronizer::State::OnError(sse::Error const& error) { + protocol_handler_.Reset(); + + std::string msg = sse::ErrorToString(error); + + if (sse::IsRecoverable(error)) { + LD_LOG(logger_, LogLevel::kWarn) << kIdentity << ": " << msg; + Notify(FDv2SourceResult{FDv2SourceResult::Interrupted{ + MakeError(ErrorKind::kNetworkError, 0, std::move(msg))}}); + return; + } + + LD_LOG(logger_, LogLevel::kError) << kIdentity << ": " << msg; + + if (auto const* client_error = + std::get_if(&error)) { + Notify(FDv2SourceResult{FDv2SourceResult::TerminalError{MakeError( + ErrorKind::kErrorResponse, + static_cast(client_error->status), + std::move(msg))}}); + return; + } + + Notify(FDv2SourceResult{FDv2SourceResult::TerminalError{ + MakeError(ErrorKind::kNetworkError, 0, std::move(msg))}}); +} + +void FDv2StreamingSynchronizer::State::Notify(FDv2SourceResult result) { + std::optional> promise; + { + std::lock_guard lock(mutex_); + // A directive parsed from the stream, such as one on a goodbye + // message, takes precedence over the most recent response header. + if (!result.fdv1_fallback) { + result.fdv1_fallback = latest_fdv1_fallback_; + } + if (!result.environment_id) { + result.environment_id = latest_environment_id_; + } + if (pending_promise_) { + promise = std::move(pending_promise_); + pending_promise_.reset(); + } else { + result_queue_.push_back(std::move(result)); + return; + } + } + // Resolve outside the lock. Promise::Resolve may invoke inline + // continuations that could call back into Notify or Next. + promise->Resolve(std::move(result)); +} + +async::Future FDv2StreamingSynchronizer::State::Next( + data_model::Selector const& selector, + std::shared_ptr self) { + EnsureStarted(selector, std::move(self)); + + std::lock_guard lock(mutex_); + if (!result_queue_.empty()) { + auto result = std::move(result_queue_.front()); + result_queue_.pop_front(); + return async::MakeFuture(std::move(result)); + } + return pending_promise_.emplace().GetFuture(); +} + +void FDv2StreamingSynchronizer::State::ClearPendingPromise() { + std::lock_guard lock(mutex_); + pending_promise_.reset(); +} + +void FDv2StreamingSynchronizer::State::Shutdown() { + std::shared_ptr client; + { + std::lock_guard lock(mutex_); + closed_ = true; + client = std::exchange(sse_client_, nullptr); + } + if (client) { + client->async_shutdown([] {}); + } +} + +FDv2StreamingSynchronizer::FDv2StreamingSynchronizer( + boost::asio::any_io_executor const& executor, + Logger const& logger, + FDv2RequestConfig const& stream_config, + FDv2RequestConfig const& poll_config, + std::chrono::milliseconds initial_reconnect_delay) + : state_(std::make_shared(logger, + executor, + stream_config, + poll_config, + initial_reconnect_delay)) {} + +FDv2StreamingSynchronizer::~FDv2StreamingSynchronizer() { + Close(); +} + +async::Future FDv2StreamingSynchronizer::Next( + data_model::Selector selector) { + auto closed = close_promise_.GetFuture(); + if (closed.IsFinished()) { + return async::MakeFuture( + FDv2SourceResult{FDv2SourceResult::Shutdown{}}); + } + + auto result_future = state_->Next(selector, state_); + if (result_future.IsFinished()) { + return result_future; + } + + return async::WhenAny(closed, result_future) + .Then( + [state = state_, result_future]( + std::size_t const& idx) mutable -> FDv2SourceResult { + if (idx == 0) { + state->ClearPendingPromise(); + return FDv2SourceResult{FDv2SourceResult::Shutdown{}}; + } + return *result_future.GetResult(); + }, + async::kInlineExecutor); +} + +void FDv2StreamingSynchronizer::Close() { + if (!close_promise_.Resolve(std::monostate{})) { + return; + } + state_->Shutdown(); +} + +std::string const& FDv2StreamingSynchronizer::Identity() const { + static std::string const identity = kIdentity; + return identity; +} + +} // namespace launchdarkly::client_side::data_sources diff --git a/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.hpp b/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.hpp new file mode 100644 index 000000000..94024ea9f --- /dev/null +++ b/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.hpp @@ -0,0 +1,178 @@ +#pragma once + +#include "fdv2_request_config.hpp" +#include "ifdv2_synchronizer.hpp" + +#include +#include +#include +#include +#include + +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace launchdarkly::client_side::data_sources { + +class FDv2StreamingSynchronizerTestPeer; + +/** + * Keeps flag data current over a long-lived connection to the FDv2 client + * streaming endpoint, turning the push-based event stream into the pull-based + * IFDv2Synchronizer::Next() interface. + * + * A `ping` event on the stream carries no data, so it is answered with a poll + * for the current payload. The service sends either data events or pings on a + * given stream, not both. + * + * Threading model: + * Next() should only be called once at a time. + * Close() may be called concurrently with Next(). + * This object may be safely destroyed once no call to Next() or Close() is + * in progress. + */ +class FDv2StreamingSynchronizer final : public IFDv2Synchronizer { + friend class FDv2StreamingSynchronizerTestPeer; + + public: + /** + * @param executor Runs the stream, the ping-triggered polls, and the + * reconnection backoff. + * @param logger Receives a description of any failure. + * @param stream_config How to reach the streaming endpoint, and which + * context to evaluate. + * @param poll_config Where to poll in answer to a `ping` event. Must + * describe the same context as stream_config. + * @param initial_reconnect_delay Where the reconnection backoff starts. + */ + FDv2StreamingSynchronizer( + boost::asio::any_io_executor const& executor, + Logger const& logger, + FDv2RequestConfig const& stream_config, + FDv2RequestConfig const& poll_config, + std::chrono::milliseconds initial_reconnect_delay); + + ~FDv2StreamingSynchronizer() override; + + async::Future Next( + data_model::Selector selector) override; + + void Close() override; + + [[nodiscard]] std::string const& Identity() const override; + + private: + // Any state that async SSE callbacks may touch lives here, held by + // shared_ptr so those callbacks can outlive the synchronizer. + class State : public std::enable_shared_from_this { + friend class FDv2StreamingSynchronizerTestPeer; + + public: + State(Logger const& logger, + boost::asio::any_io_executor const& executor, + FDv2RequestConfig const& stream_config, + FDv2RequestConfig const& poll_config, + std::chrono::milliseconds initial_reconnect_delay); + + /** + * Records the selector to send on the next connection attempt, starts + * the stream if it is not already running, and returns a Future + * resolving with the next result. + * + * If a result is already buffered the Future is resolved + * immediately. Otherwise it resolves when the next event arrives. + * + * @param self The shared_ptr owning this State, used to form the weak + * references the SSE callbacks capture. + */ + async::Future Next( + data_model::Selector const& selector, + std::shared_ptr self); + + /** + * Abandons an outstanding Next() call without delivering a result. + * Any result that arrives afterwards is buffered for the next call. + */ + void ClearPendingPromise(); + + /** + * Marks the State closed and shuts down the stream if one was + * started. After Shutdown returns, no new stream can start. + * Idempotent. + */ + void Shutdown(); + + private: + using HttpRequest = + boost::beast::http::request; + using HttpResponseHeader = boost::beast::http::response_header<>; + + /** Starts the stream if it is not already running. */ + void EnsureStarted(data_model::Selector const& selector, + std::shared_ptr self); + + /** + * Delivers a result to the caller of Next(), or buffers it if no + * caller is waiting. + */ + void Notify(FDv2SourceResult result); + + /** + * Issues the poll a `ping` event calls for, delivering its result + * when it arrives. A ping received while an answering poll is still + * in flight is dropped, so that a burst of pings cannot pile up + * requests. + */ + void PollForPing(std::shared_ptr self); + + // SSE client callbacks. + void OnConnect(HttpRequest* req); + void OnResponse(HttpResponseHeader const& headers); + void OnEvent(sse::Event const& event); + void OnError(sse::Error const& error); + + // Logger is itself thread-safe. + Logger const logger_; + + // Immutable state. + FDv2RequestConfig const stream_config_; + FDv2RequestConfig const poll_config_; + std::chrono::milliseconds const initial_reconnect_delay_; + boost::asio::any_io_executor const executor_; + network::Requester const requester_; + + // Touched only from SSE callbacks, which all run on the same strand. + // No lock required. + FDv2ProtocolHandler protocol_handler_; + + std::mutex mutex_; + // All protected by mutex_. + bool started_ = false; + bool closed_ = false; + bool ping_poll_in_flight_ = false; + // From the most recent stream response. + std::optional latest_fdv1_fallback_; + std::optional latest_environment_id_; + data_model::Selector latest_selector_; + std::optional base_url_; + std::shared_ptr sse_client_; + std::optional> pending_promise_; + std::deque result_queue_; + }; + + // Resolved by Close() or on destruction, cancelling any outstanding + // Next() call. + async::Promise close_promise_; + + // Shared with async SSE callbacks. + std::shared_ptr state_; +}; + +} // namespace launchdarkly::client_side::data_sources diff --git a/libs/client-sdk/tests/CMakeLists.txt b/libs/client-sdk/tests/CMakeLists.txt index 5f53df407..0f3f2c8a2 100644 --- a/libs/client-sdk/tests/CMakeLists.txt +++ b/libs/client-sdk/tests/CMakeLists.txt @@ -16,6 +16,6 @@ endif () add_executable(gtest_${LIBNAME} ${tests}) -target_link_libraries(gtest_${LIBNAME} launchdarkly::client launchdarkly::internal GTest::gtest_main) +target_link_libraries(gtest_${LIBNAME} launchdarkly::client launchdarkly::internal launchdarkly::sse GTest::gtest_main) gtest_discover_tests(gtest_${LIBNAME}) diff --git a/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp b/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp new file mode 100644 index 000000000..84485c849 --- /dev/null +++ b/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp @@ -0,0 +1,523 @@ +#include + +#include + +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +#include +#include +#include +#include + +namespace launchdarkly::client_side::data_sources { + +// Drives the State's per-event, per-error, and per-connect entry points +// directly, so that no real SSE connection is required. +class FDv2StreamingSynchronizerTestPeer { + public: + static void OnEvent(FDv2StreamingSynchronizer& sync, + sse::Event const& event) { + sync.state_->OnEvent(event); + } + static void OnError(FDv2StreamingSynchronizer& sync, + sse::Error const& error) { + sync.state_->OnError(error); + } + static void OnConnect( + FDv2StreamingSynchronizer& sync, + boost::beast::http::request* req) { + sync.state_->OnConnect(req); + } + static void OnResponse( + FDv2StreamingSynchronizer& sync, + boost::beast::http::response_header<> const& headers) { + sync.state_->OnResponse(headers); + } + static void MarkStarted(FDv2StreamingSynchronizer& sync) { + std::lock_guard lock(sync.state_->mutex_); + sync.state_->started_ = true; + } + static void SetBaseUrl(FDv2StreamingSynchronizer& sync, + boost::urls::url url) { + std::lock_guard lock(sync.state_->mutex_); + sync.state_->base_url_ = std::move(url); + } + static void SetLatestSelector(FDv2StreamingSynchronizer& sync, + data_model::Selector selector) { + std::lock_guard lock(sync.state_->mutex_); + sync.state_->latest_selector_ = std::move(selector); + } + static void SetSseClient(FDv2StreamingSynchronizer& sync, + std::shared_ptr client) { + std::lock_guard lock(sync.state_->mutex_); + sync.state_->sse_client_ = std::move(client); + } +}; + +} // namespace launchdarkly::client_side::data_sources + +using namespace launchdarkly; +using namespace launchdarkly::client_side::data_sources; +using namespace std::chrono_literals; + +namespace { + +Logger MakeNullLogger() { + struct NullBackend : ILogBackend { + bool Enabled(LogLevel) noexcept override { return false; } + void Write(LogLevel, std::string) noexcept override {} + }; + return Logger{std::make_shared()}; +} + +class IoContextRunner { + public: + IoContextRunner() : work_guard_(boost::asio::make_work_guard(ioc_)) { + thread_ = std::thread([this] { ioc_.run(); }); + } + ~IoContextRunner() { + work_guard_.reset(); + ioc_.stop(); + if (thread_.joinable()) { + thread_.join(); + } + } + boost::asio::io_context& context() { return ioc_; } + + private: + boost::asio::io_context ioc_; + boost::asio::executor_work_guard + work_guard_; + std::thread thread_; +}; + +FDv2RequestConfig MakeConfig( + std::string base_url, + FDv2ContextTransport transport = FDv2ContextTransport::kGetPath, + bool with_reasons = false) { + return FDv2RequestConfig{ + std::move(base_url), + config::shared::Defaults::HttpProperties(), + R"({"kind":"user","key":"user-key"})", transport, with_reasons}; +} + +// Records calls to the sse::Client interface, so tests can verify how the +// synchronizer drives the connection without a real network client. +class MockSseClient : public sse::Client { + public: + void async_connect() override { ++connect_count_; } + void async_shutdown(std::function completion) override { + ++shutdown_count_; + if (completion) { + completion(); + } + } + void async_restart(std::string const& reason) override { + ++restart_count_; + last_restart_reason_ = reason; + } + + int connect_count_ = 0; + int shutdown_count_ = 0; + int restart_count_ = 0; + std::string last_restart_reason_; +}; + +boost::beast::http::response_header<> MakeResponseHeaders( + std::vector> const& headers) { + boost::beast::http::response_header<> result; + for (auto const& [name, value] : headers) { + result.set(name, value); + } + return result; +} + +} // namespace + +// ============================================================================ +// Lifecycle +// ============================================================================ + +TEST(ClientFDv2StreamingSynchronizerTest, UnparseableEndpointIsTerminal) { + auto logger = MakeNullLogger(); + IoContextRunner runner; + + FDv2StreamingSynchronizer synchronizer(runner.context().get_executor(), + logger, MakeConfig("not a url"), + MakeConfig("http://localhost"), 1s); + + auto result = synchronizer.Next(data_model::Selector{}).WaitForResult(2s); + + ASSERT_TRUE(result.has_value()); + auto* terminal = + std::get_if(&result->value); + ASSERT_NE(nullptr, terminal); + EXPECT_EQ(FDv2SourceResult::ErrorInfo::ErrorKind::kNetworkError, + terminal->error.Kind()); +} + +TEST(ClientFDv2StreamingSynchronizerTest, NextAfterCloseIsShutdown) { + auto logger = MakeNullLogger(); + IoContextRunner runner; + + FDv2StreamingSynchronizer synchronizer( + runner.context().get_executor(), logger, MakeConfig("http://localhost"), + MakeConfig("http://localhost"), 1s); + synchronizer.Close(); + + auto result = synchronizer.Next(data_model::Selector{}).WaitForResult(2s); + + ASSERT_TRUE(result.has_value()); + EXPECT_TRUE( + std::holds_alternative(result->value)); +} + +TEST(ClientFDv2StreamingSynchronizerTest, CloseUnblocksAPendingNext) { + auto logger = MakeNullLogger(); + IoContextRunner runner; + + FDv2StreamingSynchronizer synchronizer( + runner.context().get_executor(), logger, MakeConfig("http://localhost"), + MakeConfig("http://localhost"), 1s); + + // Skip the SSE setup, so that Next is pending purely on the close race + // rather than on real network activity. + FDv2StreamingSynchronizerTestPeer::MarkStarted(synchronizer); + + auto future = synchronizer.Next(data_model::Selector{}); + synchronizer.Close(); + auto result = future.WaitForResult(2s); + + ASSERT_TRUE(result.has_value()); + EXPECT_TRUE( + std::holds_alternative(result->value)); +} + +// ============================================================================ +// Request construction +// ============================================================================ + +TEST(ClientFDv2StreamingSynchronizerTest, GetTargetCarriesTheEncodedContext) { + auto logger = MakeNullLogger(); + IoContextRunner runner; + auto client = std::make_shared(); + + FDv2StreamingSynchronizer synchronizer( + runner.context().get_executor(), logger, + MakeConfig("https://stream.example.com"), + MakeConfig("http://localhost"), 1s); + + // The connection is not made, but the target is built during setup. + synchronizer.Next(data_model::Selector{}); + boost::beast::http::request req; + FDv2StreamingSynchronizerTestPeer::OnConnect(synchronizer, &req); + + EXPECT_EQ("/sdk/stream/eval/eyJraW5kIjoidXNlciIsImtleSI6InVzZXIta2V5In0=", + req.target()); +} + +TEST(ClientFDv2StreamingSynchronizerTest, PostTargetOmitsTheContext) { + auto logger = MakeNullLogger(); + IoContextRunner runner; + + FDv2StreamingSynchronizer synchronizer( + runner.context().get_executor(), logger, + MakeConfig("https://stream.example.com", + FDv2ContextTransport::kPostBody), + MakeConfig("http://localhost"), 1s); + + synchronizer.Next(data_model::Selector{}); + boost::beast::http::request req; + FDv2StreamingSynchronizerTestPeer::OnConnect(synchronizer, &req); + + EXPECT_EQ("/sdk/stream/eval", req.target()); +} + +TEST(ClientFDv2StreamingSynchronizerTest, TargetCarriesWithReasons) { + auto logger = MakeNullLogger(); + IoContextRunner runner; + + FDv2StreamingSynchronizer synchronizer( + runner.context().get_executor(), logger, + MakeConfig("https://stream.example.com", + FDv2ContextTransport::kPostBody, + /* with_reasons= */ true), + MakeConfig("http://localhost"), 1s); + + synchronizer.Next(data_model::Selector{}); + boost::beast::http::request req; + FDv2StreamingSynchronizerTestPeer::OnConnect(synchronizer, &req); + + EXPECT_EQ("/sdk/stream/eval?withReasons=true", req.target()); +} + +TEST(ClientFDv2StreamingSynchronizerTest, EmptySelectorSendsNoBasis) { + auto logger = MakeNullLogger(); + IoContextRunner runner; + + FDv2StreamingSynchronizer synchronizer( + runner.context().get_executor(), logger, + MakeConfig("https://stream.example.com", + FDv2ContextTransport::kPostBody), + MakeConfig("http://localhost"), 1s); + + boost::urls::url base = + boost::urls::parse_uri("https://stream.example.com/sdk/stream/eval") + .value(); + FDv2StreamingSynchronizerTestPeer::SetBaseUrl(synchronizer, base); + + boost::beast::http::request req; + FDv2StreamingSynchronizerTestPeer::OnConnect(synchronizer, &req); + + EXPECT_EQ("/sdk/stream/eval", req.target()); +} + +// Each connection attempt uses the freshest selector, which is why the basis +// is appended per connect rather than baked into the base URL. +TEST(ClientFDv2StreamingSynchronizerTest, SelectorIsSentAsTheBasisPerConnect) { + auto logger = MakeNullLogger(); + IoContextRunner runner; + + FDv2StreamingSynchronizer synchronizer( + runner.context().get_executor(), logger, + MakeConfig("https://stream.example.com", + FDv2ContextTransport::kPostBody), + MakeConfig("http://localhost"), 1s); + + boost::urls::url base = + boost::urls::parse_uri("https://stream.example.com/sdk/stream/eval") + .value(); + FDv2StreamingSynchronizerTestPeer::SetBaseUrl(synchronizer, base); + FDv2StreamingSynchronizerTestPeer::SetLatestSelector( + synchronizer, + data_model::Selector{data_model::Selector::State{3, "state-3"}}); + + boost::beast::http::request req; + FDv2StreamingSynchronizerTestPeer::OnConnect(synchronizer, &req); + + EXPECT_EQ("/sdk/stream/eval?basis=state-3", req.target()); +} + +// ============================================================================ +// Events +// ============================================================================ + +namespace { + +// Builds a synchronizer that believes it is already streaming, so that tests +// can push events at it without a connection. +struct StreamingFixture { + Logger logger = MakeNullLogger(); + IoContextRunner runner; + std::shared_ptr client = std::make_shared(); + std::unique_ptr synchronizer; + + explicit StreamingFixture(std::string poll_base_url = "http://localhost") { + synchronizer = std::make_unique( + runner.context().get_executor(), logger, + MakeConfig("https://stream.example.com"), + MakeConfig(std::move(poll_base_url)), 1s); + FDv2StreamingSynchronizerTestPeer::MarkStarted(*synchronizer); + FDv2StreamingSynchronizerTestPeer::SetSseClient(*synchronizer, client); + } + + void Push(std::string type, std::string data) { + FDv2StreamingSynchronizerTestPeer::OnEvent( + *synchronizer, sse::Event(std::move(type), std::move(data))); + } + + std::optional NextResult() { + return synchronizer->Next(data_model::Selector{}).WaitForResult(2s); + } +}; + +} // namespace + +TEST(ClientFDv2StreamingSynchronizerTest, FullTransferBecomesAChangeSet) { + StreamingFixture f; + + f.Push("server-intent", R"({"payloads":[{"id":"p1","target":1,)" + R"("intentCode":"xfer-full"}]})"); + f.Push("put-object", R"({"version":7,"kind":"flag-eval","key":"my-flag",)" + R"("object":{"value":"a","variation":1}})"); + f.Push("payload-transferred", R"({"state":"abc","version":7})"); + + auto result = f.NextResult(); + + ASSERT_TRUE(result.has_value()); + auto* change_set = std::get_if(&result->value); + ASSERT_NE(nullptr, change_set); + ASSERT_EQ(1u, change_set->change_set.data.size()); + EXPECT_EQ("my-flag", change_set->change_set.data[0].key); + ASSERT_TRUE(change_set->change_set.selector.value.has_value()); + EXPECT_EQ("abc", change_set->change_set.selector.value->state); +} + +TEST(ClientFDv2StreamingSynchronizerTest, GoodbyeReportsAndReconnects) { + StreamingFixture f; + + f.Push("goodbye", R"({"reason":"bye"})"); + auto result = f.NextResult(); + + ASSERT_TRUE(result.has_value()); + auto* goodbye = std::get_if(&result->value); + ASSERT_NE(nullptr, goodbye); + EXPECT_EQ("bye", goodbye->reason.value_or("")); + EXPECT_EQ(1, f.client->restart_count_); +} + +TEST(ClientFDv2StreamingSynchronizerTest, GoodbyeCarriesItsFallbackTtl) { + StreamingFixture f; + + f.Push("goodbye", R"({"reason":"bye","protocolFallbackTTL":90})"); + auto result = f.NextResult(); + + ASSERT_TRUE(result.has_value()); + ASSERT_TRUE(result->fdv1_fallback.has_value()); + EXPECT_EQ(90s, result->fdv1_fallback->ttl); +} + +TEST(ClientFDv2StreamingSynchronizerTest, UnparseableEventDataReconnects) { + StreamingFixture f; + + f.Push("put-object", "{not json"); + auto result = f.NextResult(); + + ASSERT_TRUE(result.has_value()); + EXPECT_TRUE( + std::holds_alternative(result->value)); + EXPECT_EQ(1, f.client->restart_count_); +} + +TEST(ClientFDv2StreamingSynchronizerTest, + UntranslatableChangeSetIsInterrupted) { + StreamingFixture f; + + f.Push("server-intent", R"({"payloads":[{"id":"p1","target":1,)" + R"("intentCode":"xfer-full"}]})"); + f.Push("put-object", R"({"version":7,"kind":"flag-eval","key":"my-flag",)" + R"("object":["not-an-object"]})"); + f.Push("payload-transferred", R"({"state":"abc","version":7})"); + + auto result = f.NextResult(); + + ASSERT_TRUE(result.has_value()); + EXPECT_TRUE( + std::holds_alternative(result->value)); +} + +TEST(ClientFDv2StreamingSynchronizerTest, UnrecognizedEventIsIgnored) { + StreamingFixture f; + + f.Push("something-new", R"({"anything":true})"); + + // Nothing but a real result or Close can resolve the future. + auto future = f.synchronizer->Next(data_model::Selector{}); + EXPECT_FALSE(future.IsFinished()); +} + +// A ping carries no data, so the SDK asks for the current payload. Pointing +// the poll at an unusable URL makes the answering request observable without +// a network. +TEST(ClientFDv2StreamingSynchronizerTest, PingTriggersAPoll) { + StreamingFixture f("not a url"); + + f.Push("ping", ""); + auto result = f.NextResult(); + + ASSERT_TRUE(result.has_value()); + EXPECT_TRUE( + std::holds_alternative(result->value)); + EXPECT_EQ(0, f.client->restart_count_); +} + +// ============================================================================ +// Response headers +// ============================================================================ + +TEST(ClientFDv2StreamingSynchronizerTest, ResultsCarryTheEnvironmentId) { + StreamingFixture f; + + FDv2StreamingSynchronizerTestPeer::OnResponse( + *f.synchronizer, MakeResponseHeaders({{"X-LD-EnvId", "env-1234"}})); + f.Push("goodbye", R"({"reason":"bye"})"); + + auto result = f.NextResult(); + + ASSERT_TRUE(result.has_value()); + ASSERT_TRUE(result->environment_id.has_value()); + EXPECT_EQ("env-1234", *result->environment_id); +} + +TEST(ClientFDv2StreamingSynchronizerTest, ResultsCarryTheFallbackDirective) { + StreamingFixture f; + + FDv2StreamingSynchronizerTestPeer::OnResponse( + *f.synchronizer, + MakeResponseHeaders( + {{"X-LD-FD-Fallback", "true"}, {"X-LD-FD-Fallback-TTL", "120"}})); + f.Push("server-intent", R"({"payloads":[{"id":"p1","target":1,)" + R"("intentCode":"xfer-full"}]})"); + f.Push("payload-transferred", R"({"state":"abc","version":7})"); + + auto result = f.NextResult(); + + ASSERT_TRUE(result.has_value()); + ASSERT_TRUE(result->fdv1_fallback.has_value()); + EXPECT_EQ(120s, result->fdv1_fallback->ttl); +} + +TEST(ClientFDv2StreamingSynchronizerTest, ReconnectWithoutTheHeaderClearsIt) { + StreamingFixture f; + + FDv2StreamingSynchronizerTestPeer::OnResponse( + *f.synchronizer, MakeResponseHeaders({{"X-LD-FD-Fallback", "true"}})); + FDv2StreamingSynchronizerTestPeer::OnResponse(*f.synchronizer, + MakeResponseHeaders({})); + f.Push("goodbye", R"({"reason":"bye"})"); + + auto result = f.NextResult(); + + ASSERT_TRUE(result.has_value()); + EXPECT_FALSE(result->fdv1_fallback.has_value()); +} + +// ============================================================================ +// Errors +// ============================================================================ + +TEST(ClientFDv2StreamingSynchronizerTest, RecoverableSseErrorIsInterrupted) { + StreamingFixture f; + + FDv2StreamingSynchronizerTestPeer::OnError(*f.synchronizer, + sse::errors::ReadTimeout{}); + auto result = f.NextResult(); + + ASSERT_TRUE(result.has_value()); + EXPECT_TRUE( + std::holds_alternative(result->value)); +} + +TEST(ClientFDv2StreamingSynchronizerTest, UnrecoverableSseErrorIsTerminal) { + StreamingFixture f; + + FDv2StreamingSynchronizerTestPeer::OnError( + *f.synchronizer, sse::errors::UnrecoverableClientError{ + boost::beast::http::status::unauthorized}); + auto result = f.NextResult(); + + ASSERT_TRUE(result.has_value()); + auto* terminal = + std::get_if(&result->value); + ASSERT_NE(nullptr, terminal); + EXPECT_EQ(401u, terminal->error.StatusCode()); +} From b67efe282404a5b5541bbcdb4491e1970f46028c Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Thu, 10 Sep 2026 18:31:16 -0700 Subject: [PATCH 2/7] refactor: Use DefaultTtl for the absent-TTL streaming header case --- libs/client-sdk/src/data_sources/fdv2/fdv2_response_headers.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libs/client-sdk/src/data_sources/fdv2/fdv2_response_headers.cpp b/libs/client-sdk/src/data_sources/fdv2/fdv2_response_headers.cpp index 37e84ec77..30f25f019 100644 --- a/libs/client-sdk/src/data_sources/fdv2/fdv2_response_headers.cpp +++ b/libs/client-sdk/src/data_sources/fdv2/fdv2_response_headers.cpp @@ -50,7 +50,7 @@ FDv2ResponseHeaders ReadFDv2ResponseHeaders( auto const ttl = headers.find(kFDv1FallbackTtlHeader); result.fdv1_fallback = ttl == headers.end() - ? FDv1FallbackDirective::FromServiceTtl(std::nullopt) + ? FDv1FallbackDirective::DefaultTtl() : FDv1FallbackDirective::FromServiceTtl( std::string_view{ttl->value().data(), ttl->value().size()}); From 98f3592740f348b317503406a3bd7d571cfe6520 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Thu, 10 Sep 2026 20:25:10 -0700 Subject: [PATCH 3/7] test: Fold the streaming goodbye tests and add a TTL-precedence test --- .../fdv2_streaming_synchronizer_test.cpp | 31 ++++++++++++------- 1 file changed, 20 insertions(+), 11 deletions(-) diff --git a/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp b/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp index 84485c849..9eac8e87a 100644 --- a/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp +++ b/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp @@ -362,10 +362,11 @@ TEST(ClientFDv2StreamingSynchronizerTest, FullTransferBecomesAChangeSet) { EXPECT_EQ("abc", change_set->change_set.selector.value->state); } -TEST(ClientFDv2StreamingSynchronizerTest, GoodbyeReportsAndReconnects) { +TEST(ClientFDv2StreamingSynchronizerTest, + GoodbyeReportsReconnectsAndCarriesItsTtl) { StreamingFixture f; - f.Push("goodbye", R"({"reason":"bye"})"); + f.Push("goodbye", R"({"reason":"bye","protocolFallbackTTL":90})"); auto result = f.NextResult(); ASSERT_TRUE(result.has_value()); @@ -373,15 +374,6 @@ TEST(ClientFDv2StreamingSynchronizerTest, GoodbyeReportsAndReconnects) { ASSERT_NE(nullptr, goodbye); EXPECT_EQ("bye", goodbye->reason.value_or("")); EXPECT_EQ(1, f.client->restart_count_); -} - -TEST(ClientFDv2StreamingSynchronizerTest, GoodbyeCarriesItsFallbackTtl) { - StreamingFixture f; - - f.Push("goodbye", R"({"reason":"bye","protocolFallbackTTL":90})"); - auto result = f.NextResult(); - - ASSERT_TRUE(result.has_value()); ASSERT_TRUE(result->fdv1_fallback.has_value()); EXPECT_EQ(90s, result->fdv1_fallback->ttl); } @@ -476,6 +468,23 @@ TEST(ClientFDv2StreamingSynchronizerTest, ResultsCarryTheFallbackDirective) { EXPECT_EQ(120s, result->fdv1_fallback->ttl); } +TEST(ClientFDv2StreamingSynchronizerTest, GoodbyeTtlWinsOverTheHeader) { + StreamingFixture f; + + FDv2StreamingSynchronizerTestPeer::OnResponse( + *f.synchronizer, + MakeResponseHeaders( + {{"X-LD-FD-Fallback", "true"}, {"X-LD-FD-Fallback-TTL", "120"}})); + f.Push("goodbye", R"({"reason":"bye","protocolFallbackTTL":90})"); + + auto result = f.NextResult(); + + ASSERT_TRUE(result.has_value()); + // The goodbye's own TTL (90) wins over the header's (120). + ASSERT_TRUE(result->fdv1_fallback.has_value()); + EXPECT_EQ(90s, result->fdv1_fallback->ttl); +} + TEST(ClientFDv2StreamingSynchronizerTest, ReconnectWithoutTheHeaderClearsIt) { StreamingFixture f; From 99ba02189b4391de8e49721c77ede69246abaf94 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Thu, 10 Sep 2026 22:01:54 -0700 Subject: [PATCH 4/7] test: Tidy the FDv2 streaming synchronizer tests and comments --- .../fdv2/streaming_synchronizer.cpp | 19 ++-- .../fdv2_streaming_synchronizer_test.cpp | 102 ++++++++---------- .../fdv2_streaming_synchronizer_test.cpp | 15 +-- 3 files changed, 58 insertions(+), 78 deletions(-) diff --git a/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.cpp b/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.cpp index 5cc214e67..c56ef552a 100644 --- a/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.cpp +++ b/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.cpp @@ -16,11 +16,8 @@ static char const* const kIdentity = "FDv2 streaming synchronizer"; static char const* const kPingEvent = "ping"; -// Maximum time between bytes read from the stream before the SSE client -// declares the connection dead and reconnects. Must be greater than the -// streaming service's heartbeat interval. Hardcoded rather than read from -// HttpProperties, whose default ReadTimeout is sized for one-shot HTTP -// requests and would cause spurious disconnects on a long-lived stream. +// Read-idle timeout for the long-lived stream, larger than the service +// heartbeat so a live connection is not declared dead. static constexpr std::chrono::minutes kDeadConnectionInterval{5}; using ErrorInfo = FDv2SourceResult::ErrorInfo; @@ -66,9 +63,8 @@ void FDv2StreamingSynchronizer::State::EnsureStarted( auto parsed = boost::urls::parse_uri(stream_config_.base_url); if (!parsed) { - // started_ intentionally left true: a bad endpoint URL is a - // configuration error that won't fix itself. The TerminalError - // result tells the orchestrator to stop retrying this synchronizer. + // A bad endpoint URL is a configuration error that won't fix itself, + // so started_ stays true and this synchronizer does not reconnect. LD_LOG(logger_, LogLevel::kError) << kIdentity << ": could not parse streaming endpoint URL"; Notify(FDv2SourceResult{FDv2SourceResult::TerminalError{ @@ -160,7 +156,6 @@ void FDv2StreamingSynchronizer::State::EnsureStarted( auto client = builder.build(); if (!client) { - // started_ intentionally left true: same reasoning as above. LD_LOG(logger_, LogLevel::kError) << kIdentity << ": could not build SSE client"; Notify(FDv2SourceResult{FDv2SourceResult::TerminalError{MakeError( @@ -168,9 +163,7 @@ void FDv2StreamingSynchronizer::State::EnsureStarted( return; } - // Publishing the client and connecting are atomic with respect to - // Shutdown(). A client built after Shutdown ran is discarded without - // connecting, so there is nothing to clean up. + // If Close() ran while we were building, drop the client and stop. std::lock_guard lock(mutex_); if (closed_) { return; @@ -265,7 +258,7 @@ void FDv2StreamingSynchronizer::State::OnEvent(sse::Event const& event) { [this](auto const& r) { using T = std::decay_t; if constexpr (std::is_same_v) { - // Accumulating, heartbeat, or unknown event — nothing to do. + // Accumulating, heartbeat, or unknown event -- nothing to do. } else if constexpr (std::is_same_v) { auto typed = TranslateChangeSet(r, logger_); if (!typed) { diff --git a/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp b/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp index 9eac8e87a..553a75e47 100644 --- a/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp +++ b/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp @@ -79,6 +79,25 @@ Logger MakeNullLogger() { return Logger{std::make_shared()}; } +FDv2RequestConfig MakeConfig( + std::string base_url, + FDv2ContextTransport transport = FDv2ContextTransport::kGetPath, + bool with_reasons = false) { + return FDv2RequestConfig{ + std::move(base_url), + config::shared::Defaults::HttpProperties(), + R"({"kind":"user","key":"user-key"})", transport, with_reasons}; +} + +boost::beast::http::response_header<> MakeResponseHeaders( + std::vector> const& headers) { + boost::beast::http::response_header<> result; + for (auto const& [name, value] : headers) { + result.set(name, value); + } + return result; +} + class IoContextRunner { public: IoContextRunner() : work_guard_(boost::asio::make_work_guard(ioc_)) { @@ -100,23 +119,12 @@ class IoContextRunner { std::thread thread_; }; -FDv2RequestConfig MakeConfig( - std::string base_url, - FDv2ContextTransport transport = FDv2ContextTransport::kGetPath, - bool with_reasons = false) { - return FDv2RequestConfig{ - std::move(base_url), - config::shared::Defaults::HttpProperties(), - R"({"kind":"user","key":"user-key"})", transport, with_reasons}; -} - // Records calls to the sse::Client interface, so tests can verify how the // synchronizer drives the connection without a real network client. class MockSseClient : public sse::Client { public: - void async_connect() override { ++connect_count_; } + void async_connect() override {} void async_shutdown(std::function completion) override { - ++shutdown_count_; if (completion) { completion(); } @@ -126,20 +134,36 @@ class MockSseClient : public sse::Client { last_restart_reason_ = reason; } - int connect_count_ = 0; - int shutdown_count_ = 0; int restart_count_ = 0; std::string last_restart_reason_; }; -boost::beast::http::response_header<> MakeResponseHeaders( - std::vector> const& headers) { - boost::beast::http::response_header<> result; - for (auto const& [name, value] : headers) { - result.set(name, value); +// Builds a synchronizer that believes it is already streaming, so that tests +// can push events at it without a connection. +struct StreamingFixture { + Logger logger = MakeNullLogger(); + IoContextRunner runner; + std::shared_ptr client = std::make_shared(); + std::unique_ptr synchronizer; + + explicit StreamingFixture(std::string poll_base_url = "http://localhost") { + synchronizer = std::make_unique( + runner.context().get_executor(), logger, + MakeConfig("https://stream.example.com"), + MakeConfig(std::move(poll_base_url)), 1s); + FDv2StreamingSynchronizerTestPeer::MarkStarted(*synchronizer); + FDv2StreamingSynchronizerTestPeer::SetSseClient(*synchronizer, client); + } + + void Push(std::string type, std::string data) { + FDv2StreamingSynchronizerTestPeer::OnEvent( + *synchronizer, sse::Event(std::move(type), std::move(data))); } - return result; -} + + std::optional NextResult() { + return synchronizer->Next(data_model::Selector{}).WaitForResult(2s); + } +}; } // namespace @@ -311,37 +335,6 @@ TEST(ClientFDv2StreamingSynchronizerTest, SelectorIsSentAsTheBasisPerConnect) { // Events // ============================================================================ -namespace { - -// Builds a synchronizer that believes it is already streaming, so that tests -// can push events at it without a connection. -struct StreamingFixture { - Logger logger = MakeNullLogger(); - IoContextRunner runner; - std::shared_ptr client = std::make_shared(); - std::unique_ptr synchronizer; - - explicit StreamingFixture(std::string poll_base_url = "http://localhost") { - synchronizer = std::make_unique( - runner.context().get_executor(), logger, - MakeConfig("https://stream.example.com"), - MakeConfig(std::move(poll_base_url)), 1s); - FDv2StreamingSynchronizerTestPeer::MarkStarted(*synchronizer); - FDv2StreamingSynchronizerTestPeer::SetSseClient(*synchronizer, client); - } - - void Push(std::string type, std::string data) { - FDv2StreamingSynchronizerTestPeer::OnEvent( - *synchronizer, sse::Event(std::move(type), std::move(data))); - } - - std::optional NextResult() { - return synchronizer->Next(data_model::Selector{}).WaitForResult(2s); - } -}; - -} // namespace - TEST(ClientFDv2StreamingSynchronizerTest, FullTransferBecomesAChangeSet) { StreamingFixture f; @@ -374,6 +367,7 @@ TEST(ClientFDv2StreamingSynchronizerTest, ASSERT_NE(nullptr, goodbye); EXPECT_EQ("bye", goodbye->reason.value_or("")); EXPECT_EQ(1, f.client->restart_count_); + EXPECT_EQ("FDv2 goodbye received", f.client->last_restart_reason_); ASSERT_TRUE(result->fdv1_fallback.has_value()); EXPECT_EQ(90s, result->fdv1_fallback->ttl); } @@ -417,10 +411,8 @@ TEST(ClientFDv2StreamingSynchronizerTest, UnrecognizedEventIsIgnored) { EXPECT_FALSE(future.IsFinished()); } -// A ping carries no data, so the SDK asks for the current payload. Pointing -// the poll at an unusable URL makes the answering request observable without -// a network. TEST(ClientFDv2StreamingSynchronizerTest, PingTriggersAPoll) { + // An invalid URL makes the answering request observable without network. StreamingFixture f("not a url"); f.Push("ping", ""); diff --git a/libs/server-sdk/tests/fdv2_streaming_synchronizer_test.cpp b/libs/server-sdk/tests/fdv2_streaming_synchronizer_test.cpp index 25bebefc9..7fd28185d 100644 --- a/libs/server-sdk/tests/fdv2_streaming_synchronizer_test.cpp +++ b/libs/server-sdk/tests/fdv2_streaming_synchronizer_test.cpp @@ -115,9 +115,8 @@ config::shared::built::HttpProperties MakeHttpProperties() { // requiring a real network client. class MockSseClient : public sse::Client { public: - void async_connect() override { ++connect_count_; } + void async_connect() override {} void async_shutdown(std::function completion) override { - ++shutdown_count_; if (completion) { completion(); } @@ -127,8 +126,6 @@ class MockSseClient : public sse::Client { last_restart_reason_ = reason; } - int connect_count_ = 0; - int shutdown_count_ = 0; int restart_count_ = 0; std::string last_restart_reason_; }; @@ -833,9 +830,8 @@ TEST(FDv2StreamingSynchronizerTest, DirectiveWithTtlHeaderParsesValue) { IoContextRunner runner; FDv2StreamingSynchronizer synchronizer( - runner.context().get_executor(), logger, - "http://localhost", MakeHttpProperties(), std::nullopt, - 1s); + runner.context().get_executor(), logger, "http://localhost", + MakeHttpProperties(), std::nullopt, 1s); FDv2StreamingSynchronizerTestPeer::MarkStarted(synchronizer); // Server sends the directive with an explicit TTL. @@ -862,9 +858,8 @@ TEST(FDv2StreamingSynchronizerTest, DirectiveWithoutTtlHeaderUsesDefault) { IoContextRunner runner; FDv2StreamingSynchronizer synchronizer( - runner.context().get_executor(), logger, - "http://localhost", MakeHttpProperties(), std::nullopt, - 1s); + runner.context().get_executor(), logger, "http://localhost", + MakeHttpProperties(), std::nullopt, 1s); FDv2StreamingSynchronizerTestPeer::MarkStarted(synchronizer); // Server sends the directive with no TTL header. From 9c5c0719f52c638dec9450d38e7debca0c452d8c Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Fri, 11 Sep 2026 14:46:27 -0700 Subject: [PATCH 5/7] docs: Drop the ping detail from the FDv2 streaming synchronizer class comment --- .../src/data_sources/fdv2/streaming_synchronizer.hpp | 4 ---- 1 file changed, 4 deletions(-) diff --git a/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.hpp b/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.hpp index 94024ea9f..7d16979fe 100644 --- a/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.hpp +++ b/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.hpp @@ -28,10 +28,6 @@ class FDv2StreamingSynchronizerTestPeer; * streaming endpoint, turning the push-based event stream into the pull-based * IFDv2Synchronizer::Next() interface. * - * A `ping` event on the stream carries no data, so it is answered with a poll - * for the current payload. The service sends either data events or pings on a - * given stream, not both. - * * Threading model: * Next() should only be called once at a time. * Close() may be called concurrently with Next(). From bb0d4356a0c55c0bb63cab928a4d1c7ff684a5a9 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Fri, 11 Sep 2026 17:20:46 -0700 Subject: [PATCH 6/7] fix: Guard the SSE ReadTimeout formatter against a missing timeout --- libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp | 2 +- libs/server-sent-events/src/error.cpp | 7 +++++-- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp b/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp index 553a75e47..a99dc7bec 100644 --- a/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp +++ b/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp @@ -500,7 +500,7 @@ TEST(ClientFDv2StreamingSynchronizerTest, RecoverableSseErrorIsInterrupted) { StreamingFixture f; FDv2StreamingSynchronizerTestPeer::OnError(*f.synchronizer, - sse::errors::ReadTimeout{}); + sse::errors::ReadTimeout{100ms}); auto result = f.NextResult(); ASSERT_TRUE(result.has_value()); diff --git a/libs/server-sent-events/src/error.cpp b/libs/server-sent-events/src/error.cpp index b4cf1c44c..9d277ad82 100644 --- a/libs/server-sent-events/src/error.cpp +++ b/libs/server-sent-events/src/error.cpp @@ -19,8 +19,11 @@ std::ostream& operator<<(std::ostream& out, NotRedirectable const&) { } std::ostream& operator<<(std::ostream& out, ReadTimeout const& err) { - out << "timed out reading response body (exceeded " << err.timeout->count() - << "ms) - will retry"; + out << "timed out reading response body"; + if (err.timeout) { + out << " (exceeded " << err.timeout->count() << "ms)"; + } + out << " - will retry"; return out; } From bd73ba950e087030bda08f8a978564853981a11e Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Mon, 14 Sep 2026 17:28:58 -0700 Subject: [PATCH 7/7] fix: Reset the FDv2 streaming handler and reconnect on a translation failure --- .../data_sources/fdv2/streaming_synchronizer.cpp | 6 ++++++ .../tests/fdv2_streaming_synchronizer_test.cpp | 15 ++++++++++++++- 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.cpp b/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.cpp index c56ef552a..2720b39cc 100644 --- a/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.cpp +++ b/libs/client-sdk/src/data_sources/fdv2/streaming_synchronizer.cpp @@ -262,6 +262,8 @@ void FDv2StreamingSynchronizer::State::OnEvent(sse::Event const& event) { } else if constexpr (std::is_same_v) { auto typed = TranslateChangeSet(r, logger_); if (!typed) { + // Discard the accepted-but-unstored payload. + protocol_handler_.Reset(); std::string msg = "FDv2 streaming changeset could not be translated"; LD_LOG(logger_, LogLevel::kError) @@ -269,6 +271,10 @@ void FDv2StreamingSynchronizer::State::OnEvent(sse::Event const& event) { Notify(FDv2SourceResult{ FDv2SourceResult::Interrupted{MakeError( ErrorKind::kInvalidData, 0, std::move(msg))}}); + std::lock_guard lock(mutex_); + if (sse_client_) { + sse_client_->async_restart("FDv2 translation error"); + } return; } Notify(FDv2SourceResult{ diff --git a/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp b/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp index a99dc7bec..9bd7f6e01 100644 --- a/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp +++ b/libs/client-sdk/tests/fdv2_streaming_synchronizer_test.cpp @@ -385,7 +385,7 @@ TEST(ClientFDv2StreamingSynchronizerTest, UnparseableEventDataReconnects) { } TEST(ClientFDv2StreamingSynchronizerTest, - UntranslatableChangeSetIsInterrupted) { + UntranslatableChangeSetResetsTheHandlerAndReconnects) { StreamingFixture f; f.Push("server-intent", R"({"payloads":[{"id":"p1","target":1,)" @@ -399,6 +399,19 @@ TEST(ClientFDv2StreamingSynchronizerTest, ASSERT_TRUE(result.has_value()); EXPECT_TRUE( std::holds_alternative(result->value)); + // The connection is dropped and reconnected for a fresh basis. + EXPECT_EQ(1, f.client->restart_count_); + + // A payload-transferred is only valid mid-transfer, i.e. after a + // server-intent. The reset ended the transfer, so this one is now a + // protocol error. Without the reset the handler would still be mid-transfer + // and would emit a partial changeset over the data we just discarded. + f.Push("payload-transferred", R"({"state":"def","version":8})"); + auto after = f.NextResult(); + + ASSERT_TRUE(after.has_value()); + EXPECT_TRUE( + std::holds_alternative(after->value)); } TEST(ClientFDv2StreamingSynchronizerTest, UnrecognizedEventIsIgnored) {