-
Notifications
You must be signed in to change notification settings - Fork 4
refactor: implement fdv2 polling initializer / synchronizer #519
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
bf945b3
cc59635
82d09d2
259d8e9
f5b82bf
7826357
651a780
fe37f55
c470763
ed22857
f0013da
29a94c8
9e75b47
c8e1c7f
0996cca
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| #pragma once | ||
|
|
||
| #include <launchdarkly/data_model/fdv2_change.hpp> | ||
| #include <launchdarkly/serialization/json_fdv2_events.hpp> | ||
|
|
||
| #include <boost/json/value.hpp> | ||
|
|
||
| #include <string_view> | ||
| #include <variant> | ||
| #include <vector> | ||
|
|
||
| namespace launchdarkly { | ||
|
|
||
| /** | ||
| * Protocol state machine for the FDv2 wire format. | ||
| * | ||
| * Accumulates put-object and delete-object events between a server-intent | ||
| * and payload-transferred event, then emits a complete FDv2ChangeSet. | ||
| * | ||
| * Shared between the polling and streaming synchronizers. | ||
| */ | ||
| class FDv2ProtocolHandler { | ||
| public: | ||
| /** | ||
| * Result of handling a single FDv2 event: | ||
| * - monostate: no output yet (accumulating, heartbeat, or unknown event) | ||
| * - FDv2ChangeSet: complete changeset ready to apply | ||
| * - FDv2Error: server reported an error; discard partial data | ||
| * - Goodbye: server is closing; caller should rotate sources | ||
| */ | ||
| using Result = std::variant<std::monostate, | ||
| data_model::FDv2ChangeSet, | ||
| FDv2Error, | ||
| Goodbye>; | ||
|
|
||
| /** | ||
| * Process one FDv2 event. | ||
| * | ||
| * @param event_type The event type string (e.g. "server-intent", | ||
| * "put-object", "payload-transferred"). | ||
| * @param data The parsed JSON value for the event's data field. | ||
| * @return A Result indicating what (if anything) the caller | ||
| * should act on. | ||
| */ | ||
| Result HandleEvent(std::string_view event_type, | ||
| boost::json::value const& data); | ||
|
|
||
| /** | ||
| * Reset accumulated state. Call on reconnect before processing new events. | ||
| */ | ||
| void Reset(); | ||
|
|
||
| FDv2ProtocolHandler() = default; | ||
|
|
||
| private: | ||
| enum class State { kInactive, kFull, kPartial }; | ||
|
|
||
| State state_ = State::kInactive; | ||
| std::vector<data_model::FDv2Change> changes_; | ||
| }; | ||
|
|
||
| } // namespace launchdarkly |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,220 @@ | ||
| #include <launchdarkly/fdv2_protocol_handler.hpp> | ||
|
|
||
| #include <launchdarkly/data_model/flag.hpp> | ||
| #include <launchdarkly/data_model/item_descriptor.hpp> | ||
| #include <launchdarkly/data_model/segment.hpp> | ||
| #include <launchdarkly/serialization/json_flag.hpp> | ||
| #include <launchdarkly/serialization/json_segment.hpp> | ||
|
|
||
| #include <boost/json.hpp> | ||
| #include <tl/expected.hpp> | ||
|
|
||
| namespace launchdarkly { | ||
|
|
||
| static char const* const kServerIntent = "server-intent"; | ||
| static char const* const kPutObject = "put-object"; | ||
| static char const* const kDeleteObject = "delete-object"; | ||
| static char const* const kPayloadTransferred = "payload-transferred"; | ||
| static char const* const kError = "error"; | ||
| static char const* const kGoodbye = "goodbye"; | ||
|
|
||
| // Returns the parsed FDv2Change on success, nullopt for unknown kinds (which | ||
| // should be silently skipped for forward-compatibility), or an error string if | ||
| // a known kind fails to deserialize. | ||
| static tl::expected<std::optional<data_model::FDv2Change>, std::string> | ||
| ParsePut(PutObject const& put) { | ||
| if (put.kind == "flag") { | ||
| auto result = boost::json::value_to< | ||
| tl::expected<std::optional<data_model::Flag>, JsonError>>( | ||
| put.object); | ||
| // One bad flag aborts the entire transfer so the store is never | ||
| // left in a partially-updated state. | ||
| if (!result) { | ||
| return tl::make_unexpected("could not deserialize flag '" + | ||
| put.key + "'"); | ||
| } | ||
| if (!result->has_value()) { | ||
| return tl::make_unexpected("flag '" + put.key + "' object was null"); | ||
| } | ||
| return data_model::FDv2Change{ | ||
| put.key, | ||
| data_model::ItemDescriptor<data_model::Flag>{std::move(**result)}}; | ||
| } | ||
| if (put.kind == "segment") { | ||
| auto result = boost::json::value_to< | ||
| tl::expected<std::optional<data_model::Segment>, JsonError>>( | ||
| put.object); | ||
| // One bad segment aborts the entire transfer so the store is never | ||
| // left in a partially-updated state. | ||
| if (!result) { | ||
| return tl::make_unexpected("could not deserialize segment '" + | ||
| put.key + "'"); | ||
| } | ||
| if (!result->has_value()) { | ||
| return tl::make_unexpected("segment '" + put.key + | ||
| "' object was null"); | ||
| } | ||
| return data_model::FDv2Change{ | ||
| put.key, | ||
| data_model::ItemDescriptor<data_model::Segment>{ | ||
| std::move(**result)}}; | ||
| } | ||
| // Silently skip unknown kinds for forward-compatibility. | ||
| return std::nullopt; | ||
| } | ||
|
|
||
| static data_model::FDv2Change MakeDeleteChange(DeleteObject const& del) { | ||
| if (del.kind == "flag") { | ||
| return data_model::FDv2Change{ | ||
| del.key, | ||
| data_model::ItemDescriptor<data_model::Flag>{ | ||
| data_model::Tombstone{static_cast<uint64_t>(del.version)}}}; | ||
| } | ||
| return data_model::FDv2Change{ | ||
| del.key, | ||
| data_model::ItemDescriptor<data_model::Segment>{ | ||
| data_model::Tombstone{static_cast<uint64_t>(del.version)}}}; | ||
| } | ||
|
|
||
| FDv2ProtocolHandler::Result FDv2ProtocolHandler::HandleEvent( | ||
| std::string_view event_type, | ||
| boost::json::value const& data) { | ||
| if (event_type == kServerIntent) { | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. It might be nice to have this handle which event type it is in a more compact way, like a switch, and then handle the message is functions. Just makes it easier to see at a glance what is handled independent of how it is handled. |
||
| auto result = boost::json::value_to< | ||
| tl::expected<std::optional<ServerIntent>, JsonError>>(data); | ||
| if (!result) { | ||
| Reset(); | ||
| return FDv2Error{std::nullopt, "could not deserialize server-intent"}; | ||
| } | ||
| if (!result->has_value()) { | ||
| Reset(); | ||
| return FDv2Error{std::nullopt, "server-intent data was null"}; | ||
| } | ||
| auto const& intent = **result; | ||
| if (intent.payloads.empty()) { | ||
| return std::monostate{}; | ||
| } | ||
| auto const& code = intent.payloads[0].intent_code; | ||
| changes_.clear(); | ||
| if (code == IntentCode::kTransferFull) { | ||
| state_ = State::kFull; | ||
| } else if (code == IntentCode::kTransferChanges) { | ||
| state_ = State::kPartial; | ||
| } else { | ||
| // kNone or kUnknown: emit an empty changeset immediately. | ||
| state_ = State::kInactive; | ||
| return data_model::FDv2ChangeSet{data_model::FDv2ChangeSet::Type::kNone, | ||
| {}, | ||
| data_model::Selector{}}; | ||
| } | ||
| return std::monostate{}; | ||
| } | ||
|
|
||
| if (event_type == kPutObject) { | ||
| if (state_ == State::kInactive) { | ||
| return std::monostate{}; | ||
| } | ||
| auto result = boost::json::value_to< | ||
| tl::expected<std::optional<PutObject>, JsonError>>(data); | ||
| if (!result) { | ||
| Reset(); | ||
| return FDv2Error{std::nullopt, "could not deserialize put-object"}; | ||
| } | ||
| if (!result->has_value()) { | ||
| Reset(); | ||
| return FDv2Error{std::nullopt, "put-object data was null"}; | ||
| } | ||
| auto change = ParsePut(**result); | ||
| if (!change) { | ||
| Reset(); | ||
| return FDv2Error{std::nullopt, std::move(change.error())}; | ||
| } | ||
| if (*change) { | ||
| changes_.push_back(std::move(**change)); | ||
| } | ||
| return std::monostate{}; | ||
| } | ||
|
|
||
| if (event_type == kDeleteObject) { | ||
| if (state_ == State::kInactive) { | ||
| return std::monostate{}; | ||
| } | ||
| auto result = boost::json::value_to< | ||
| tl::expected<std::optional<DeleteObject>, JsonError>>(data); | ||
| if (!result) { | ||
| Reset(); | ||
| return FDv2Error{std::nullopt, "could not deserialize delete-object"}; | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We do need differentiation for malformed data errors, as we typically will restart on those errors versus just potentially logging another error type. |
||
| } | ||
| if (!result->has_value()) { | ||
| Reset(); | ||
| return FDv2Error{std::nullopt, "delete-object data was null"}; | ||
| } | ||
| auto const& del = **result; | ||
| // Silently skip unknown kinds for forward-compatibility. | ||
| if (del.kind != "flag" && del.kind != "segment") { | ||
| return std::monostate{}; | ||
| } | ||
| changes_.push_back(MakeDeleteChange(del)); | ||
| return std::monostate{}; | ||
| } | ||
|
|
||
| if (event_type == kPayloadTransferred) { | ||
| auto result = boost::json::value_to< | ||
| tl::expected<std::optional<PayloadTransferred>, JsonError>>(data); | ||
| if (!result) { | ||
| Reset(); | ||
| return FDv2Error{std::nullopt, | ||
| "could not deserialize payload-transferred"}; | ||
| } | ||
| if (!result->has_value()) { | ||
| Reset(); | ||
| return FDv2Error{std::nullopt, "payload-transferred data was null"}; | ||
| } | ||
| auto const& transferred = **result; | ||
| auto type = (state_ == State::kPartial) | ||
| ? data_model::FDv2ChangeSet::Type::kPartial | ||
| : data_model::FDv2ChangeSet::Type::kFull; | ||
| data_model::FDv2ChangeSet changeset{ | ||
| type, | ||
| std::move(changes_), | ||
| data_model::Selector{data_model::Selector::State{ | ||
| transferred.version, transferred.state}}}; | ||
| Reset(); | ||
| return changeset; | ||
| } | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Missing kInactive guard in payload-transferred handlerHigh Severity The Reviewed by Cursor Bugbot for commit 0996cca. Configure here. |
||
|
|
||
| if (event_type == kError) { | ||
| auto result = boost::json::value_to< | ||
| tl::expected<std::optional<FDv2Error>, JsonError>>(data); | ||
| Reset(); | ||
| if (!result) { | ||
| return FDv2Error{std::nullopt, "could not deserialize error event"}; | ||
| } | ||
| if (!result->has_value()) { | ||
| return FDv2Error{std::nullopt, "error event data was null"}; | ||
| } | ||
| return **result; | ||
| } | ||
|
|
||
| if (event_type == kGoodbye) { | ||
| auto result = boost::json::value_to< | ||
| tl::expected<std::optional<Goodbye>, JsonError>>(data); | ||
| if (!result) { | ||
| return Goodbye{std::nullopt}; | ||
| } | ||
| if (!result->has_value()) { | ||
| return Goodbye{std::nullopt}; | ||
| } | ||
| return **result; | ||
| } | ||
|
|
||
| // heartbeat and unrecognized events: no-op. | ||
| return std::monostate{}; | ||
| } | ||
|
|
||
| void FDv2ProtocolHandler::Reset() { | ||
| state_ = State::kInactive; | ||
| changes_.clear(); | ||
| } | ||
|
|
||
| } // namespace launchdarkly | ||


There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This ends up with maybe a slightly different layer responsibility, but I think it is likely fine. I think Java returns it, and then it is discarded a layer up, but we aren't acting on it, so it doesn't really make a difference.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Actually I am a bit concerned that we may be losing granularity of data source reporting. Though maybe we don't have that concern at the moment.