Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions libs/server-sdk/src/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ target_sources(${LIBNAME}
data_components/big_segments/big_segment_store_status_provider.hpp
data_components/big_segments/big_segment_store_status_provider.cpp
data_interfaces/destination/itransactional_destination.hpp
data_interfaces/source/fdv2_source_result.cpp
data_components/memory_store/memory_store.hpp
data_components/memory_store/memory_store.cpp
data_components/serialization_adapters/json_deserializer.hpp
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
#include "fdv2_source_result.hpp"

#include <charconv>
#include <cstdint>
#include <random>
#include <system_error>

namespace launchdarkly::server_side::data_interfaces {

namespace {

// Subtracts a value drawn uniformly from [0, ttl/2], so that SDKs which fell
// back within the same window do not all re-attempt FDv2 at once.
std::chrono::seconds Jitter(std::chrono::seconds ttl) {
static thread_local std::mt19937_64 generator{std::random_device{}()};
std::uniform_int_distribution<std::chrono::seconds::rep> distribution(
0, ttl.count() / 2);
return ttl - std::chrono::seconds(distribution(generator));
}

} // namespace

FDv1FallbackDirective FDv1FallbackDirective::DefaultTtl() {
return FDv1FallbackDirective{Jitter(kDefaultTtl)};
}

FDv1FallbackDirective FDv1FallbackDirective::FromServiceTtl(
std::chrono::seconds ttl) {
if (ttl > std::chrono::seconds::zero() && ttl <= kDefaultTtl) {
return FDv1FallbackDirective{ttl};
}
return DefaultTtl();
}

FDv1FallbackDirective FDv1FallbackDirective::FromServiceTtl(
std::string_view ttl) {
std::uint64_t seconds = 0;
auto const* begin = ttl.data();
auto const* end = begin + ttl.size();

Check warning on line 39 in libs/server-sdk/src/data_interfaces/source/fdv2_source_result.cpp

View workflow job for this annotation

GitHub Actions / cpp-linter

libs/server-sdk/src/data_interfaces/source/fdv2_source_result.cpp:39:29 [cppcoreguidelines-pro-bounds-pointer-arithmetic]

do not use pointer arithmetic
auto const [ptr, ec] = std::from_chars(begin, end, seconds);
if (ec != std::errc{} || ptr != end) {
return DefaultTtl();
}
return FromServiceTtl(std::chrono::seconds(seconds));
}

} // namespace launchdarkly::server_side::data_interfaces
41 changes: 19 additions & 22 deletions libs/server-sdk/src/data_interfaces/source/fdv2_source_result.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -5,13 +5,10 @@
#include <launchdarkly/data_model/change_set.hpp>
#include <launchdarkly/data_sources/data_source_status_error_info.hpp>

#include <charconv>
#include <chrono>
#include <cstdint>
#include <optional>
#include <string>
#include <string_view>
#include <system_error>
#include <variant>

namespace launchdarkly::server_side::data_interfaces {
Expand All @@ -22,32 +19,32 @@ namespace launchdarkly::server_side::data_interfaces {
* X-LD-FD-Fallback: true response header).
*/
struct FDv1FallbackDirective {
/** Default TTL used when the directive carries no TTL of its own. */
/** Used whenever the service supplies no usable TTL. */
static constexpr std::chrono::seconds kDefaultTtl = std::chrono::hours(1);

/**
* Parse the value of an X-LD-FD-Fallback-TTL header (or equivalent
* protocolFallbackTTL field from a goodbye message). Returns
* std::nullopt if the value is malformed.
* Builds a directive using the jittered default TTL, so that SDKs which
* fell back together do not all retry at once. Used when the service
* supplies no usable TTL. Safe to call from any thread.
*/
static std::optional<std::chrono::seconds> ParseTtl(
std::string_view value) {
std::uint64_t seconds = 0;
auto const* begin = value.data();
auto const* end = begin + value.size();
auto const [ptr, ec] = std::from_chars(begin, end, seconds);
if (ec != std::errc{} || ptr != end) {
return std::nullopt;
}
return std::chrono::seconds(seconds);
}
static FDv1FallbackDirective DefaultTtl();

/**
* How long to stay on FDv1 before attempting to recover to FDv2.
* A value of 0 seconds means the SDK should remain on FDv1
* indefinitely.
* Builds a directive from a service-supplied TTL, used as given. A value
* outside (0, 1 hour] uses the jittered default instead, so a fallback is
* never indefinite.
*/
std::chrono::seconds ttl = kDefaultTtl;
static FDv1FallbackDirective FromServiceTtl(std::chrono::seconds ttl);

/**
* Builds a directive from the raw value of an X-LD-FD-Fallback-TTL
* header. A value that is not a whole number of seconds uses the
* jittered default.
*/
static FDv1FallbackDirective FromServiceTtl(std::string_view ttl);

/** How long to stay off FDv2 before attempting to recover to it. */
std::chrono::seconds ttl;
};

/**
Expand Down
3 changes: 0 additions & 3 deletions libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -24,10 +24,10 @@
// Reduces a source result to the signal the conditions act on.
SourceSignal ClassifyResult(data_interfaces::FDv2SourceResult const& result) {
using Result = data_interfaces::FDv2SourceResult;
if (std::get_if<Result::ChangeSet>(&result.value)) {

Check warning on line 27 in libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp

View workflow job for this annotation

GitHub Actions / cpp-linter

libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp:27:9 [readability-implicit-bool-conversion]

implicit conversion 'add_pointer_t<const launchdarkly::server_side::data_interfaces::FDv2SourceResult::ChangeSet>' (aka 'const launchdarkly::server_side::data_interfaces::FDv2SourceResult::ChangeSet *') -> bool
return SourceSignal::kChangeSet;
}
if (std::get_if<Result::Interrupted>(&result.value)) {

Check warning on line 30 in libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp

View workflow job for this annotation

GitHub Actions / cpp-linter

libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp:30:9 [readability-implicit-bool-conversion]

implicit conversion 'add_pointer_t<const launchdarkly::server_side::data_interfaces::FDv2SourceResult::Interrupted>' (aka 'const launchdarkly::server_side::data_interfaces::FDv2SourceResult::Interrupted *') -> bool
return SourceSignal::kInterrupted;
}
return SourceSignal::kOther;
Expand All @@ -44,14 +44,14 @@
std::unique_ptr<IFDv2ConditionFactory> recovery_condition_factory,
boost::asio::any_io_executor ioc,
data_components::DataSourceStatusManager* status_manager,
Logger const& logger)

Check warning on line 47 in libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp

View workflow job for this annotation

GitHub Actions / cpp-linter

libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp:47:5 [modernize-pass-by-value]

pass by value and use std::move
: logger_(logger),
ioc_(std::move(ioc)),
initializer_factories_(std::move(initializer_factories)),
fallback_condition_factory_(std::move(fallback_condition_factory)),
recovery_condition_factory_(std::move(recovery_condition_factory)),
status_manager_(status_manager),
store_(),

Check warning on line 54 in libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp

View workflow job for this annotation

GitHub Actions / cpp-linter

libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp:54:7 [readability-redundant-member-init]

initializer for member 'store_' is redundant
change_notifier_(store_, store_),
initialize_called_(false),
last_logged_synchronizer_interrupted_(false),
Expand Down Expand Up @@ -131,7 +131,7 @@
if (initializer_index_ >= initializer_factories_.size()) {
exhausted = true;
} else {
auto& factory = initializer_factories_[initializer_index_++];

Check warning on line 134 in libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp

View workflow job for this annotation

GitHub Actions / cpp-linter

libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp:134:13 [readability-qualified-auto]

'auto &factory' can be declared as 'const auto &factory'
active_initializer_ = factory->Build();
LD_LOG(logger_, LogLevel::kInfo)
<< Identity() << ": starting initializer "
Expand Down Expand Up @@ -182,7 +182,7 @@
DataSourceStatus::DataSourceState::kInterrupted,
iv.error.Kind(), iv.error.Message());
},
[&](Result::TerminalError const& te) {

Check warning on line 185 in libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp

View workflow job for this annotation

GitHub Actions / cpp-linter

libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp:185:46 [readability-identifier-length]

parameter name 'te' is too short, expected at least 3 characters
LD_LOG(logger_, LogLevel::kWarn)
<< Identity()
<< ": initializer terminal error: " << te.error.Message();
Expand Down Expand Up @@ -380,7 +380,7 @@
DataSourceStatus::DataSourceState::kInterrupted,
iv.error.Kind(), iv.error.Message());
},
[&](Result::TerminalError const& te) {

Check warning on line 383 in libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp

View workflow job for this annotation

GitHub Actions / cpp-linter

libs/server-sdk/src/data_systems/fdv2/fdv2_data_system.cpp:383:46 [readability-identifier-length]

parameter name 'te' is too short, expected at least 3 characters
LD_LOG(logger_, LogLevel::kWarn)
<< Identity()
<< ": synchronizer terminal error: " << te.error.Message();
Expand Down Expand Up @@ -441,9 +441,6 @@
}

void FDv2DataSystem::ScheduleFDv2RetryLocked(std::chrono::seconds ttl) {
if (ttl == std::chrono::seconds::zero()) {
return;
}
// Cancel any in-flight retry and start fresh; CancellationSource is
// one-shot, so reusing the same source for successive schedules would
// leak prior timers.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -258,7 +258,7 @@ class FDv2DataSystem final : public data_interfaces::IDataSystem {
void OnConditionFired(IFDv2Condition::Type type);

// Schedules an FDv2 recovery attempt after the given TTL. Called with
// mutex_ held. TTL of 0 disables the recovery and is a no-op.
// mutex_ held.
void ScheduleFDv2RetryLocked(std::chrono::seconds ttl);

// Invoked when the FDv1 fallback TTL expires. Switches the source list
Expand Down
18 changes: 7 additions & 11 deletions libs/server-sdk/src/data_systems/fdv2/fdv2_polling_impl.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -41,16 +41,11 @@
if (it == headers.end() || !boost::iequals(it->second, "true")) {
return std::nullopt;
}
data_interfaces::FDv1FallbackDirective directive;
auto const ttl_it = headers.find(kFDv1FallbackTtlHeader);
if (ttl_it != headers.end()) {
auto const ttl =
data_interfaces::FDv1FallbackDirective::ParseTtl(ttl_it->second);
if (ttl) {
directive.ttl = *ttl;
}
}
return directive;
return ttl_it == headers.end()
? data_interfaces::FDv1FallbackDirective::DefaultTtl()
: data_interfaces::FDv1FallbackDirective::FromServiceTtl(
ttl_it->second);
}

network::HttpRequest MakeFDv2PollRequest(
Expand All @@ -67,7 +62,7 @@
network::HttpRequest::BodyType{}};
}

boost::urls::url u = parsed.value();

Check warning on line 65 in libs/server-sdk/src/data_systems/fdv2/fdv2_polling_impl.cpp

View workflow job for this annotation

GitHub Actions / cpp-linter

libs/server-sdk/src/data_systems/fdv2/fdv2_polling_impl.cpp:65:22 [readability-identifier-length]

variable name 'u' is too short, expected at least 3 characters
// A trailing '/' on the base URL appears as an empty final segment;
// remove it so subsequent push_backs don't produce a double slash.
auto segs = u.segments();
Expand Down Expand Up @@ -99,7 +94,7 @@
Logger const& logger) {
for (auto const& event_val : events) {
auto const* event_obj = event_val.if_object();
if (!event_obj) {

Check warning on line 97 in libs/server-sdk/src/data_systems/fdv2/fdv2_polling_impl.cpp

View workflow job for this annotation

GitHub Actions / cpp-linter

libs/server-sdk/src/data_systems/fdv2/fdv2_polling_impl.cpp:97:14 [readability-implicit-bool-conversion]

implicit conversion 'const boost::json::object *' -> bool
continue;
}

Expand Down Expand Up @@ -131,8 +126,9 @@
if (auto* goodbye = std::get_if<Goodbye>(&result)) {
FDv2SourceResult result{FDv2SourceResult::Goodbye{goodbye->reason}};
if (goodbye->protocol_fallback_ttl) {
result.fdv1_fallback = data_interfaces::FDv1FallbackDirective{
std::chrono::seconds(*goodbye->protocol_fallback_ttl)};
result.fdv1_fallback =
data_interfaces::FDv1FallbackDirective::FromServiceTtl(
std::chrono::seconds(*goodbye->protocol_fallback_ttl));
}
return result;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -192,16 +192,13 @@ void FDv2StreamingSynchronizer::State::OnResponse(
latest_fdv1_fallback_.reset();
return;
}
data_interfaces::FDv1FallbackDirective directive;
auto const ttl_it = headers.find("X-LD-FD-Fallback-TTL");
if (ttl_it != headers.end()) {
auto const value = ttl_it->value();
auto const ttl = data_interfaces::FDv1FallbackDirective::ParseTtl(
std::string_view{value.data(), value.size()});
if (ttl) {
directive.ttl = *ttl;
}
}
auto const directive =
ttl_it == headers.end()
? data_interfaces::FDv1FallbackDirective::DefaultTtl()
: data_interfaces::FDv1FallbackDirective::FromServiceTtl(
std::string_view{ttl_it->value().data(),
ttl_it->value().size()});
std::lock_guard lock(mutex_);
latest_fdv1_fallback_ = directive;
}
Expand Down Expand Up @@ -257,8 +254,8 @@ void FDv2StreamingSynchronizer::State::OnEvent(sse::Event const& event) {
FDv2SourceResult::Goodbye{r.reason}};
if (r.protocol_fallback_ttl) {
goodbye_result.fdv1_fallback =
data_interfaces::FDv1FallbackDirective{
std::chrono::seconds(*r.protocol_fallback_ttl)};
data_interfaces::FDv1FallbackDirective::FromServiceTtl(
std::chrono::seconds(*r.protocol_fallback_ttl));
}
Notify(std::move(goodbye_result));
// Drop the current connection and reconnect; the protocol
Expand Down
79 changes: 61 additions & 18 deletions libs/server-sdk/tests/fdv2_data_system_test.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,13 @@ Logger MakeNullLogger() {
return Logger{std::make_shared<NullBackend>()};
}

// Runs handlers until none are ready, without waiting on pending timers
// (e.g. the FDv2 recovery timer scheduled by an FDv1 fallback directive).
void RunReadyHandlers(boost::asio::io_context& ioc) {
while (ioc.poll() > 0) {
}
}

// Initializer that resolves Run() with a single pre-set result.
class MockInitializer : public IFDv2Initializer {
public:
Expand Down Expand Up @@ -1097,8 +1104,7 @@ TEST(FDv2DataSystemTest, SynchronizerFdv1FlagSwitchesToFdv1Adapter) {
data_model::ChangeSetType::kNone,
{},
data_model::Selector{}}}};
r.fdv1_fallback =
data_interfaces::FDv1FallbackDirective{std::chrono::seconds{0}};
r.fdv1_fallback = data_interfaces::FDv1FallbackDirective{1h};
return r;
}()});
auto fdv2_factory =
Expand All @@ -1120,7 +1126,7 @@ TEST(FDv2DataSystemTest, SynchronizerFdv1FlagSwitchesToFdv1Adapter) {
/*recovery_condition_factory=*/nullptr,
ioc.get_executor(), &status_manager, logger);
ds.Initialize();
ioc.run();
RunReadyHandlers(ioc);

EXPECT_EQ(1, fdv1_factory_ptr->build_count_);
}
Expand All @@ -1131,16 +1137,14 @@ TEST(FDv2DataSystemTest,
boost::asio::io_context ioc;
data_components::DataSourceStatusManager status_manager;

// Directive with TTL=0: indefinite, no automatic FDv2 retry.
auto fdv2_sync =
std::make_unique<MockSynchronizer>(std::vector<FDv2SourceResult>{[]() {
FDv2SourceResult r{
FDv2SourceResult::Interrupted{FDv2SourceResult::ErrorInfo{
FDv2SourceResult::ErrorInfo::ErrorKind::kErrorResponse,
/*status_code=*/418, "directive",
std::chrono::system_clock::now()}}};
r.fdv1_fallback =
data_interfaces::FDv1FallbackDirective{std::chrono::seconds{0}};
r.fdv1_fallback = data_interfaces::FDv1FallbackDirective{1h};
return r;
}()});
auto fdv2_factory =
Expand All @@ -1154,12 +1158,55 @@ TEST(FDv2DataSystemTest,
/*recovery_condition_factory=*/nullptr,
ioc.get_executor(), &status_manager, logger);
ds.Initialize();
ioc.run();
RunReadyHandlers(ioc);

EXPECT_NE(DataSourceStatus::DataSourceState::kOff,
status_manager.Status().State());
}

TEST(FDv2DataSystemTest,
SynchronizerFdv1FlagWithoutAdapterStillRecoversToFDv2AfterTtl) {
auto logger = MakeNullLogger();
boost::asio::io_context ioc;
data_components::DataSourceStatusManager status_manager;

// First build: emits the directive with a 1s TTL; no FDv1 fallback
// synchronizer is configured. Second build (after the TTL elapses):
// returns no results, so MockSynchronizer emits Shutdown.
auto first_sync =
std::make_unique<MockSynchronizer>(std::vector<FDv2SourceResult>{[]() {
FDv2SourceResult r{
FDv2SourceResult::Interrupted{FDv2SourceResult::ErrorInfo{
FDv2SourceResult::ErrorInfo::ErrorKind::kErrorResponse,
/*status_code=*/418, "directive",
std::chrono::system_clock::now()}}};
r.fdv1_fallback = data_interfaces::FDv1FallbackDirective{1s};
return r;
}()});
auto second_sync =
std::make_unique<MockSynchronizer>(std::vector<FDv2SourceResult>{});

std::vector<std::unique_ptr<IFDv2Synchronizer>> sources;
sources.push_back(std::move(first_sync));
sources.push_back(std::move(second_sync));
auto factory =
std::make_unique<MultiShotSynchronizerFactory>(std::move(sources));
auto* factory_ptr = factory.get();

std::vector<std::unique_ptr<IFDv2SynchronizerFactory>> synchronizers;
synchronizers.push_back(std::move(factory));

FDv2DataSystem ds({}, std::move(synchronizers),
/*fallback_condition_factory=*/nullptr,
/*recovery_condition_factory=*/nullptr,
ioc.get_executor(), &status_manager, logger);
ds.Initialize();
ioc.run();

// The fallback was not indefinite: FDv2 was rebuilt after the TTL.
EXPECT_EQ(2, factory_ptr->build_count_);
}

TEST(FDv2DataSystemTest, InitializerFdv1FlagSwitchesToFdv1Adapter) {
auto logger = MakeNullLogger();
boost::asio::io_context ioc;
Expand All @@ -1171,8 +1218,7 @@ TEST(FDv2DataSystemTest, InitializerFdv1FlagSwitchesToFdv1Adapter) {
FDv2SourceResult::ErrorInfo::ErrorKind::kErrorResponse,
/*status_code=*/418, "directive",
std::chrono::system_clock::now()}}};
init_result.fdv1_fallback =
data_interfaces::FDv1FallbackDirective{std::chrono::seconds{0}};
init_result.fdv1_fallback = data_interfaces::FDv1FallbackDirective{1h};
auto initializer =
std::make_unique<MockInitializer>(std::move(init_result));

Expand Down Expand Up @@ -1201,7 +1247,7 @@ TEST(FDv2DataSystemTest, InitializerFdv1FlagSwitchesToFdv1Adapter) {
/*recovery_condition_factory=*/nullptr,
ioc.get_executor(), &status_manager, logger);
ds.Initialize();
ioc.run();
RunReadyHandlers(ioc);

// FDv2 synchronizer was skipped; FDv1 adapter was built and ran.
EXPECT_EQ(0, fdv2_factory_ptr->build_count_);
Expand All @@ -1226,8 +1272,7 @@ TEST(FDv2DataSystemTest,
ItemChange{"flagA", data_model::FlagDescriptor(flag_a)},
},
MakeSelector(1, "state-1"));
init_result.fdv1_fallback =
data_interfaces::FDv1FallbackDirective{std::chrono::seconds{0}};
init_result.fdv1_fallback = data_interfaces::FDv1FallbackDirective{1h};

auto initializer =
std::make_unique<MockInitializer>(std::move(init_result));
Expand Down Expand Up @@ -1257,7 +1302,7 @@ TEST(FDv2DataSystemTest,
/*recovery_condition_factory=*/nullptr,
ioc.get_executor(), &status_manager, logger);
ds.Initialize();
ioc.run();
RunReadyHandlers(ioc);

// Basis applied before the switch.
EXPECT_TRUE(ds.Initialized());
Expand Down Expand Up @@ -1383,8 +1428,7 @@ TEST(FDv2DataSystemTest, FDv1SourceSelfDirectiveDoesNotRebuildFDv1) {
data_model::ChangeSetType::kNone,
{},
data_model::Selector{}}}};
r.fdv1_fallback =
data_interfaces::FDv1FallbackDirective{std::chrono::seconds{0}};
r.fdv1_fallback = data_interfaces::FDv1FallbackDirective{1h};
return r;
}()});
auto fdv2_factory =
Expand All @@ -1400,8 +1444,7 @@ TEST(FDv2DataSystemTest, FDv1SourceSelfDirectiveDoesNotRebuildFDv1) {
FDv2SourceResult::ErrorInfo::ErrorKind::kErrorResponse,
/*status_code=*/418, "self-trigger",
std::chrono::system_clock::now()}}};
r.fdv1_fallback =
data_interfaces::FDv1FallbackDirective{std::chrono::seconds{0}};
r.fdv1_fallback = data_interfaces::FDv1FallbackDirective{1h};
return r;
}()});
auto fdv1_factory =
Expand All @@ -1417,7 +1460,7 @@ TEST(FDv2DataSystemTest, FDv1SourceSelfDirectiveDoesNotRebuildFDv1) {
/*recovery_condition_factory=*/nullptr,
ioc.get_executor(), &status_manager, logger);
ds.Initialize();
ioc.run();
RunReadyHandlers(ioc);

EXPECT_EQ(1, fdv1_factory_ptr->build_count_);
}
Expand Down
Loading
Loading