From 8a1e4fa18b50afe7047a407b6ad6ae1b7c553c69 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Tue, 28 Jul 2026 12:54:09 -0400 Subject: [PATCH 1/7] chore: Use go-jsonstream RawValue/Offset prerelease from branch Temporary, swap before merge: requires the pseudo-version of go-jsonstream/v3 branch mk/SDK-2760/jreader-raw-value (PR #45, which is stacked on the RFC 8259 compliance work in PR #46) so CI can build against the new jreader.RawValue and jreader.Offset APIs. Replace with the released jsonstream version once those PRs ship. --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 9a345af1..b53c2838 100644 --- a/go.mod +++ b/go.mod @@ -8,7 +8,7 @@ require ( github.com/gregjones/httpcache v0.0.0-20171119193500-2bcd89a1743f github.com/launchdarkly/ccache v1.1.0 github.com/launchdarkly/eventsource v1.10.0 - github.com/launchdarkly/go-jsonstream/v3 v3.1.1 + github.com/launchdarkly/go-jsonstream/v3 v3.1.2-0.20260728165303-975e0be34b70 github.com/launchdarkly/go-ntlm-proxy-auth v1.0.3 github.com/launchdarkly/go-sdk-common/v3 v3.5.0 github.com/launchdarkly/go-sdk-events/v3 v3.6.2 diff --git a/go.sum b/go.sum index 3f81bb71..fb636df7 100644 --- a/go.sum +++ b/go.sum @@ -19,8 +19,8 @@ github.com/launchdarkly/ccache v1.1.0 h1:voD1M+ZJXR3MREOKtBwgTF9hYHl1jg+vFKS/+VA github.com/launchdarkly/ccache v1.1.0/go.mod h1:TlxzrlnzvYeXiLHmesMuvoZetu4Z97cV1SsdqqBJi1Q= github.com/launchdarkly/eventsource v1.10.0 h1:H9Tp6AfGu/G2qzBJC26iperrvwhzdbiA/gx7qE2nDFI= github.com/launchdarkly/eventsource v1.10.0/go.mod h1:J3oa50bPvJesZqNAJtb5btSIo5N6roDWhiAS3IpsKck= -github.com/launchdarkly/go-jsonstream/v3 v3.1.1 h1:ugupp2eNtwVbr69KCdeUrm1vUf1/3ju4Wdliaob95uY= -github.com/launchdarkly/go-jsonstream/v3 v3.1.1/go.mod h1:ZBjhKq8mhArCtqotGRGnteY6eXpNm1GaOdUSZHh+ZjM= +github.com/launchdarkly/go-jsonstream/v3 v3.1.2-0.20260728165303-975e0be34b70 h1:Z4GlojB8T/d8jeAKFbEOpVYOOvCbIcNJYGeB0YIauiE= +github.com/launchdarkly/go-jsonstream/v3 v3.1.2-0.20260728165303-975e0be34b70/go.mod h1:ZBjhKq8mhArCtqotGRGnteY6eXpNm1GaOdUSZHh+ZjM= github.com/launchdarkly/go-ntlm-proxy-auth v1.0.3 h1:i3V0N+R0Fd2nXfGEVKCBIZ8kyttZ+SRKvBG8cdcphO4= github.com/launchdarkly/go-ntlm-proxy-auth v1.0.3/go.mod h1:kU5uMfNSTpYE6fIzmAXjFxUdmnaDPUEQ5zKm3RVKUsY= github.com/launchdarkly/go-ntlmssp v1.0.3 h1:rFxOnnEJ2DzJ+NU0plhXqnldJUwn3wWJFTWKCmaiQdE= From f4673b593c70edc1320ef9af8f327702028a74ef Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Tue, 28 Jul 2026 11:05:34 -0400 Subject: [PATCH 2/7] perf: Parse FDv2 payloads in a single jsonstream pass The FDv2 data sources previously scanned every item's JSON multiple times: the polling path ran encoding/json reflection over the whole {"events":[...]} envelope (copying each event into a RawEvent), unmarshaled each event again into PutObject (copying the item JSON a second time), and finally parsed the model with jreader in ChangeSet.Collections(). The streaming path did the same minus the envelope pass. This cost lands on every client init, since the default data system mode runs a polling initializer first, and on every changeset an application (or Relay) ingests. The default build now decodes each payload in one pass with jsonstream. Scalars are read in place, and a recognized item is model-decoded directly from the stream -- which fully validates it, the tokenizer's grammar being RFC 8259 compliant -- while the reader offsets recorded around the decode capture the item's raw bytes as a zero-copy slice of the input (jreader.Offset). Values that are never model-parsed here -- objects of unrecognized kinds, and event data arriving before its event name -- are captured with jreader.RawValue, whose own validation is then the only line of defense before the bytes can be relayed downstream. The decoders are order-independent; LaunchDarkly services always write "kind" before "object" and "event" before "data", but nothing requires it. The launchdarkly_easyjson build keeps the previous reflection-based decode instead: easyjson support is planned for removal from go-jsonstream, so no SDK code depends on its token reader (which provides neither RawValue nor Offset). Both variants produce identical results, including the eager item deserialization, and share the test suite; the two intentional single-pass divergences (fractional-version truncation, trailing bytes ignored after payload-transferred) are documented in default-only tests. ChangeSetBuilder gains AddParsedPut so parsed items travel with the raw changes, and Finish pre-populates the ChangeSet's collections cache when every recognized put is parsed, making Collections() allocation-free. Change.Object raw bytes are preserved verbatim (surrounding JSON whitespace trimmed, cap == len) for consumers such as Relay, which re-serializes them downstream unmodified. Benchmark (2000 flags, parse + Collections): the previous decode measures 21.6ms / 8.12MB / 64.1k allocs; the single-pass decode measures 8.5ms / 4.61MB / 42.1k allocs (2.5x faster), within ~2% of the per-byte throughput of an FDv1-equivalent parse of the same flags. A reference implementation of the old decode is kept in the benchmarks. One error kind changed: per-event JSON errors in a polling payload now surface as invalid-data (malformedJSONError) instead of a network error, matching the envelope error handling. --- internal/datasourcev2/event_parsing.go | 220 +++++++++ .../event_parsing_benchmark_test.go | 43 ++ .../datasourcev2/event_parsing_default.go | 233 ++++++++++ .../event_parsing_default_test.go | 28 ++ .../datasourcev2/event_parsing_easyjson.go | 95 ++++ internal/datasourcev2/event_parsing_test.go | 428 ++++++++++++++++++ internal/datasourcev2/polling_http_request.go | 53 +-- .../datasourcev2/streaming_data_source.go | 21 +- subsystems/changeset.go | 108 ++++- subsystems/changeset_parsed_put_test.go | 133 ++++++ 10 files changed, 1295 insertions(+), 67 deletions(-) create mode 100644 internal/datasourcev2/event_parsing.go create mode 100644 internal/datasourcev2/event_parsing_benchmark_test.go create mode 100644 internal/datasourcev2/event_parsing_default.go create mode 100644 internal/datasourcev2/event_parsing_default_test.go create mode 100644 internal/datasourcev2/event_parsing_easyjson.go create mode 100644 internal/datasourcev2/event_parsing_test.go create mode 100644 subsystems/changeset_parsed_put_test.go diff --git a/internal/datasourcev2/event_parsing.go b/internal/datasourcev2/event_parsing.go new file mode 100644 index 00000000..1e671717 --- /dev/null +++ b/internal/datasourcev2/event_parsing.go @@ -0,0 +1,220 @@ +package datasourcev2 + +import ( + "encoding/json" + "errors" + + "github.com/launchdarkly/go-jsonstream/v3/jreader" + "github.com/launchdarkly/go-server-sdk/v7/subsystems" + "github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoretypes" +) + +// This file contains the decoders for the FDv2 protocol events that are shared by both build +// variants. The payload-level decoders -- the ones responsible for capturing each put-object +// item's raw JSON alongside its parsed form -- are split by build tag: +// +// - event_parsing_default.go: a single-pass jsonstream decoder using jreader.RawValue for +// zero-copy raw capture. Each event, and in particular each item's JSON, is scanned once. +// - event_parsing_easyjson.go: the previous reflection-based decode (kept for the +// launchdarkly_easyjson build so that no new code depends on the easyjson token reader, +// which is planned for removal). +// +// Both variants produce the same results: a parsedPutObject carrying the item's raw bytes and, +// for recognized kinds, its parsed form, so that the ChangeSet's collections are assembled +// without a later re-parse. The decoders are tolerant of property ordering: nothing here assumes +// that, for example, "kind" appears before "object" in a put-object event, even though +// LaunchDarkly services always write them in that order. + +// JSON property names shared by the FDv2 protocol event decoders. +const ( + propEvents = "events" + propEvent = "event" + propData = "data" + propKind = "kind" + propKey = "key" + propVersion = "version" + propObject = "object" + propReason = "reason" + propState = "state" + propPayloads = "payloads" + propID = "id" + propTarget = "target" + propIntentCode = "intentCode" + propPayloadID = "payloadId" +) + +const errNoKnownPollingEvents = "didn't receive any known protocol events in polling payload" + +// parsedPutObject is the result of decoding a put-object event's data. +type parsedPutObject struct { + kind subsystems.ObjectKind + key string + version int + // object is the item's raw JSON. In the default build it references the input buffer passed + // to parsePutObjectEventData and is valid only as long as that buffer is. + object json.RawMessage + // item is the parsed representation of object, set only when hasItem is true. hasItem is + // false when the kind is unrecognized, in which case the raw bytes are still retained for + // forwards compatibility. + item ldstoretypes.ItemDescriptor + hasItem bool +} + +// addTo adds the put to a change-set builder, carrying the parsed item along when there is one. +func (p parsedPutObject) addTo(builder *subsystems.ChangeSetBuilder) { + if p.hasItem { + builder.AddParsedPut(p.kind, p.key, p.version, p.object, p.item) + } else { + builder.AddPut(p.kind, p.key, p.version, p.object) + } +} + +// parseServerIntentEventData decodes the data of a server-intent event. The intent is required to +// have at least one payload (at index 0) at this time. +func parseServerIntentEventData(data []byte) (subsystems.ServerIntent, error) { + r := jreader.NewReader(data) + intent, err := readServerIntent(&r) + if err == nil { + err = r.Error() + } + return intent, err +} + +func readServerIntent(r *jreader.Reader) (subsystems.ServerIntent, error) { + var intent subsystems.ServerIntent + gotPayload := false + for obj := r.Object(); obj.Next(); { + if string(obj.Name()) != propPayloads { + _ = r.SkipValue() + continue + } + for arr := r.Array(); arr.Next(); { + // The protocol allows more than one payload, but SDKs currently only support one; + // any additional payloads are skipped. + if gotPayload { + _ = r.SkipValue() + continue + } + for payloadObj := r.Object(); payloadObj.Next(); { + switch string(payloadObj.Name()) { + case propID: + intent.Payload.ID = r.String() + case propTarget: + intent.Payload.Target = r.Int() + case propIntentCode: + intent.Payload.Code = subsystems.IntentCode(r.String()) + case propReason: + intent.Payload.Reason = r.String() + default: + _ = r.SkipValue() + } + } + gotPayload = true + } + } + if err := r.Error(); err != nil { + return intent, err + } + if !gotPayload { + // It is a protocol error for the payload list to be missing or empty. + return intent, errors.New("changeset: server-intent event has no payloads") + } + return intent, nil +} + +// parseDeleteObjectEventData decodes the data of a delete-object event. +func parseDeleteObjectEventData(data []byte) (subsystems.DeleteObject, error) { + r := jreader.NewReader(data) + deleteObject, err := readDeleteObject(&r) + if err == nil { + err = r.Error() + } + return deleteObject, err +} + +func readDeleteObject(r *jreader.Reader) (subsystems.DeleteObject, error) { + var d subsystems.DeleteObject + for obj := r.Object(); obj.Next(); { + switch string(obj.Name()) { + case propKind: + d.Kind = subsystems.ObjectKind(r.String()) + case propKey: + d.Key = r.String() + case propVersion: + d.Version = r.Int() + default: + _ = r.SkipValue() + } + } + return d, r.Error() +} + +// parseSelectorEventData decodes the data of a payload-transferred event. Both the state and the +// version are required. +func parseSelectorEventData(data []byte) (subsystems.Selector, error) { + r := jreader.NewReader(data) + selector, err := readSelector(&r) + if err == nil { + err = r.Error() + } + return selector, err +} + +func readSelector(r *jreader.Reader) (subsystems.Selector, error) { + var state string + var version int + gotState, gotVersion := false, false + for obj := r.Object(); obj.Next(); { + switch string(obj.Name()) { + case propState: + state = r.String() + gotState = true + case propVersion: + version = r.Int() + gotVersion = true + default: + _ = r.SkipValue() + } + } + if err := r.Error(); err != nil { + return subsystems.NoSelector(), err + } + if !gotState { + return subsystems.NoSelector(), errors.New("unmarshal selector: missing state field") + } + if !gotVersion { + return subsystems.NoSelector(), errors.New("unmarshal selector: missing version field") + } + return subsystems.NewSelector(state, version), nil +} + +// parseGoodbyeEventData decodes the data of a goodbye event. +func parseGoodbyeEventData(data []byte) (subsystems.Goodbye, error) { + r := jreader.NewReader(data) + var goodbye subsystems.Goodbye + for obj := r.Object(); obj.Next(); { + if string(obj.Name()) == propReason { + goodbye.Reason = r.String() + } else { + _ = r.SkipValue() + } + } + return goodbye, r.Error() +} + +// parseErrorEventData decodes the data of an error event. +func parseErrorEventData(data []byte) (subsystems.Error, error) { + r := jreader.NewReader(data) + var errorData subsystems.Error + for obj := r.Object(); obj.Next(); { + switch string(obj.Name()) { + case propPayloadID: + errorData.PayloadID = r.String() + case propReason: + errorData.Reason = r.String() + default: + _ = r.SkipValue() + } + } + return errorData, r.Error() +} diff --git a/internal/datasourcev2/event_parsing_benchmark_test.go b/internal/datasourcev2/event_parsing_benchmark_test.go new file mode 100644 index 00000000..cf4ae1a4 --- /dev/null +++ b/internal/datasourcev2/event_parsing_benchmark_test.go @@ -0,0 +1,43 @@ +package datasourcev2 + +import ( + "context" + "testing" +) + +// These benchmarks guard the single-pass polling-payload parser against regressions, and keep the +// previous reflection-based decode path measurable for comparison. The dominant cost of client +// initialization over FDv2 is parsing the initial polling payload, so this path is +// performance-sensitive: every SDK start pays it once per payload. + +func BenchmarkParsePollingPayload(b *testing.B) { + body := makePollingBody(b, 2000) + b.SetBytes(int64(len(body))) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + changeSet, err := parsePollingPayload(context.Background(), body) + if err != nil { + b.Fatal(err) + } + if _, err := changeSet.Collections(); err != nil { + b.Fatal(err) + } + } +} + +func BenchmarkParsePollingPayloadReflectionReference(b *testing.B) { + body := makePollingBody(b, 2000) + b.SetBytes(int64(len(body))) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + changeSet, err := parsePollingPayloadReflectionReference(body) + if err != nil { + b.Fatal(err) + } + if _, err := changeSet.Collections(); err != nil { + b.Fatal(err) + } + } +} diff --git a/internal/datasourcev2/event_parsing_default.go b/internal/datasourcev2/event_parsing_default.go new file mode 100644 index 00000000..214f1b3f --- /dev/null +++ b/internal/datasourcev2/event_parsing_default.go @@ -0,0 +1,233 @@ +//go:build !launchdarkly_easyjson +// +build !launchdarkly_easyjson + +package datasourcev2 + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + + "github.com/launchdarkly/go-jsonstream/v3/jreader" + "github.com/launchdarkly/go-server-sdk/v7/subsystems" +) + +// This file contains the single-pass payload decoders used by the default build. Each event -- +// and in particular each put-object's item JSON -- is scanned only once: scalars are read in +// place, and a recognized item's model is decoded directly from the stream while its raw bytes +// are captured as a zero-copy slice of the input by recording the reader's offset around the +// decode (the RFC 8259 compliant tokenizer fully validates everything it parses, so the captured +// span is known-valid JSON). Only values that are never model-parsed here -- objects of +// unrecognized kinds, and event data that arrives before its event name -- go through +// jreader.RawValue, which performs its own boundary scan and validation. + +// parsePollingPayload walks a polling response body of the form {"events":[...]} in a single +// jsonstream pass and returns the completed change set. It mirrors the event semantics of the +// streaming data source: a server-intent of "none" short-circuits to a no-changes result, and a +// payload-transferred event completes the change set. Unknown event names are ignored for +// forwards compatibility. +// +// The returned ChangeSet's raw change objects reference the body slice, so the body must not be +// reused or modified while the ChangeSet is alive. +func parsePollingPayload(ctx context.Context, body []byte) (*subsystems.ChangeSet, error) { + builder := subsystems.NewChangeSetBuilder() + r := jreader.NewReader(body) + for topObj := r.Object(); topObj.Next(); { + if string(topObj.Name()) != propEvents { + _ = r.SkipValue() + continue + } + for arr := r.Array(); arr.Next(); { + select { + case <-ctx.Done(): + return nil, ctx.Err() + default: + } + changeSet, done, err := readPollingEvent(&r, body, builder) + if err != nil { + return nil, err + } + if done { + return changeSet, nil + } + } + } + if err := r.Error(); err != nil { + return nil, err + } + return nil, errors.New(errNoKnownPollingEvents) +} + +// readPollingEvent consumes one {"event":...,"data":...} object from the events array. When the +// event completes the payload (server-intent "none" or payload-transferred), it returns the +// resulting change set with done set to true. +// +// The common case dispatches on the event name and decodes the data in place as soon as it is +// reached. If the data property appears before the event name, its raw bytes are captured +// (zero-copy) and decoded once the name is known. +func readPollingEvent( + r *jreader.Reader, + body []byte, + builder *subsystems.ChangeSetBuilder, +) (changeSet *subsystems.ChangeSet, done bool, err error) { + var name subsystems.EventName + var deferredData []byte + gotData := false + // dataInput must be the byte slice that dataReader was created over, so that offset-based + // span capture inside readPutObject can slice it. + dispatch := func(dataReader *jreader.Reader, dataInput []byte) error { + switch name { + case subsystems.EventServerIntent: + intent, err := readServerIntent(dataReader) + if err != nil { + return err + } + if intent.Payload.Code == subsystems.IntentNone { + changeSet, done = builder.NoChanges(), true + return nil + } + builder.Start(intent) + case subsystems.EventPutObject: + put, err := readPutObject(dataReader, dataInput) + if err != nil { + return err + } + put.addTo(builder) + case subsystems.EventDeleteObject: + deleteObject, err := readDeleteObject(dataReader) + if err != nil { + return err + } + builder.AddDelete(deleteObject.Kind, deleteObject.Key, deleteObject.Version) + case subsystems.EventPayloadTransferred: + selector, err := readSelector(dataReader) + if err != nil { + return err + } + finished, err := builder.Finish(selector) + if err != nil { + return err + } + changeSet, done = finished, true + default: + // An unknown event name is ignored for forwards compatibility. + return dataReader.SkipValue() + } + return nil + } + for obj := r.Object(); obj.Next(); { + switch string(obj.Name()) { + case propEvent: + name = subsystems.EventName(r.String()) + case propData: + gotData = true + if name == "" { + deferredData = r.RawValue() + } else if err := dispatch(r, body); err != nil { + return nil, false, err + } + default: + _ = r.SkipValue() + } + if done { + // The payload is complete; the rest of the input is intentionally left unread. + return changeSet, true, nil + } + } + if err := r.Error(); err != nil { + return nil, false, err + } + if deferredData != nil { + dataReader := jreader.NewReader(deferredData) + if err := dispatch(&dataReader, deferredData); err != nil { + return nil, false, err + } + return changeSet, done, nil + } + if !gotData { + switch name { + case subsystems.EventServerIntent, subsystems.EventPutObject, + subsystems.EventDeleteObject, subsystems.EventPayloadTransferred: + // A payload-affecting event without data cannot be applied; treat the payload as + // malformed rather than silently dropping the event. + return nil, false, fmt.Errorf("polling payload event %q has no data", name) + } + } + return changeSet, done, nil +} + +// parsePutObjectEventData decodes the data of a put-object event in a single pass, capturing the +// item's raw JSON and its parsed form together. +func parsePutObjectEventData(data []byte) (parsedPutObject, error) { + r := jreader.NewReader(data) + p, err := readPutObject(&r, data) + if err == nil { + err = r.Error() + } + return p, err +} + +// jsonWhitespace is the set of insignificant whitespace bytes RFC 8259 allows between tokens; a +// span captured via reader offsets may include them around the value and they are trimmed off. +const jsonWhitespace = " \t\r\n" + +// readPutObject decodes a put-object event's data from r, which must have been created over +// input (offset-based span capture slices it). +func readPutObject(r *jreader.Reader, input []byte) (parsedPutObject, error) { + var p parsedPutObject + var object json.RawMessage + for obj := r.Object(); obj.Next(); { + switch string(obj.Name()) { + case propKind: + p.kind = subsystems.ObjectKind(r.String()) + case propKey: + p.key = r.String() + case propVersion: + p.version = r.Int() + case propObject: + if kind, recognized := p.kind.ToFDV1(); recognized { + // The kind is already known (LaunchDarkly services always write "kind" before + // "object"), so the item is model-decoded directly from the stream -- which + // fully validates it -- while the reader offsets around the decode capture the + // value's raw bytes with no additional scan. + start := r.Offset() + item, err := kind.DeserializeFromJSONReader(r) + if err != nil { + return p, err + } + span := bytes.Trim(input[start:r.Offset()], jsonWhitespace) + object = json.RawMessage(span[:len(span):len(span)]) + p.item, p.hasItem = item, true + } else { + // The kind is unrecognized or has not been read yet, so the value cannot be + // model-decoded here; RawValue captures it with full validation, and if the + // kind turns out to be recognized it is deserialized after the loop. + object = json.RawMessage(r.RawValue()) + } + default: + _ = r.SkipValue() + } + } + if err := r.Error(); err != nil { + return p, err + } + p.object = object + if p.hasItem || object == nil { + return p, nil + } + if kind, recognized := p.kind.ToFDV1(); recognized { + // The object appeared before the kind; parse it now. + itemReader := jreader.NewReader(object) + item, err := kind.DeserializeFromJSONReader(&itemReader) + if err != nil { + return p, err + } + p.item, p.hasItem = item, true + } + // If the kind is unrecognized, the object is kept raw (and unparsed) for forwards + // compatibility. No further validation is needed: RawValue already fully validated the + // bytes, so a malformed object has failed the payload above regardless of kind. + return p, nil +} diff --git a/internal/datasourcev2/event_parsing_default_test.go b/internal/datasourcev2/event_parsing_default_test.go new file mode 100644 index 00000000..458bd7a5 --- /dev/null +++ b/internal/datasourcev2/event_parsing_default_test.go @@ -0,0 +1,28 @@ +//go:build !launchdarkly_easyjson +// +build !launchdarkly_easyjson + +package datasourcev2 + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// These tests document behavior specific to the default build's single-pass parser. The easyjson +// build's reflection-based decode parses the whole envelope up front, so it does not share them. + +func TestParsePollingPayloadStopsReadingAfterTransfer(t *testing.T) { + // Once payload-transferred completes the change set, the single-pass parser never reads the + // rest of the body -- even content that is not valid JSON. + body := []byte(`{"events":[` + + `{"event":"server-intent","data":{"payloads":[{"id":"p1","target":1,"intentCode":"xfer-full","reason":"r"}]}},` + + `{"event":"payload-transferred","data":{"state":"s","version":1}},` + + `{"event":"put-object","data":{"this is": "not even valid put data`, + ) + changeSet, err := parsePollingPayload(context.Background(), body) + require.NoError(t, err) + assert.Empty(t, changeSet.Changes()) +} diff --git a/internal/datasourcev2/event_parsing_easyjson.go b/internal/datasourcev2/event_parsing_easyjson.go new file mode 100644 index 00000000..333303a0 --- /dev/null +++ b/internal/datasourcev2/event_parsing_easyjson.go @@ -0,0 +1,95 @@ +//go:build launchdarkly_easyjson +// +build launchdarkly_easyjson + +package datasourcev2 + +import ( + "context" + "encoding/json" + "errors" + + "github.com/launchdarkly/go-jsonstream/v3/jreader" + "github.com/launchdarkly/go-server-sdk/v7/subsystems" +) + +// This file contains the payload decoders used by the launchdarkly_easyjson build. It retains +// the reflection-based decode (encoding/json over the polling envelope and each put-object +// event) so that nothing in this build depends on jreader.RawValue: easyjson support is planned +// for removal from go-jsonstream, and no new code should rely on its token reader. +// +// The results are identical to the default build's single-pass decoder -- including the eager +// item deserialization that lets the ChangeSet pre-populate its collections -- at the cost of +// the extra reflection scans and byte copies that the default build avoids. + +// parsePollingPayload decodes a polling response body of the form {"events":[...]} and returns +// the completed change set. Refer to the default build's implementation for the event semantics; +// the two are behaviorally equivalent. +func parsePollingPayload(ctx context.Context, body []byte) (*subsystems.ChangeSet, error) { + var payload subsystems.PollingPayload + if err := json.Unmarshal(body, &payload); err != nil { + return nil, err + } + builder := subsystems.NewChangeSetBuilder() + for _, event := range payload.Events { + select { + case <-ctx.Done(): + return nil, ctx.Err() + default: + } + switch event.Name { + case subsystems.EventServerIntent: + intent, err := parseServerIntentEventData(event.Data) + if err != nil { + return nil, err + } + if intent.Payload.Code == subsystems.IntentNone { + return builder.NoChanges(), nil + } + builder.Start(intent) + case subsystems.EventPutObject: + put, err := parsePutObjectEventData(event.Data) + if err != nil { + return nil, err + } + put.addTo(builder) + case subsystems.EventDeleteObject: + deleteObject, err := parseDeleteObjectEventData(event.Data) + if err != nil { + return nil, err + } + builder.AddDelete(deleteObject.Kind, deleteObject.Key, deleteObject.Version) + case subsystems.EventPayloadTransferred: + selector, err := parseSelectorEventData(event.Data) + if err != nil { + return nil, err + } + return builder.Finish(selector) + default: + // An unknown event name is ignored for forwards compatibility. + } + } + return nil, errors.New(errNoKnownPollingEvents) +} + +// parsePutObjectEventData decodes the data of a put-object event via encoding/json, then eagerly +// deserializes the item so the change set can carry the parsed form alongside the raw bytes. +func parsePutObjectEventData(data []byte) (parsedPutObject, error) { + var put subsystems.PutObject + if err := json.Unmarshal(data, &put); err != nil { + return parsedPutObject{}, err + } + p := parsedPutObject{kind: put.Kind, key: put.Key, version: put.Version, object: put.Object} + kind, recognized := put.Kind.ToFDV1() + if !recognized || put.Object == nil { + // An unrecognized kind is kept raw for forwards compatibility; its JSON syntax was + // already validated by the json.RawMessage decode above. + return p, nil + } + itemReader := jreader.NewReader(put.Object) + item, err := kind.DeserializeFromJSONReader(&itemReader) + if err != nil { + return p, err + } + p.item, p.hasItem = item, true + return p, nil +} diff --git a/internal/datasourcev2/event_parsing_test.go b/internal/datasourcev2/event_parsing_test.go new file mode 100644 index 00000000..cc73d339 --- /dev/null +++ b/internal/datasourcev2/event_parsing_test.go @@ -0,0 +1,428 @@ +package datasourcev2 + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "testing" + + "github.com/launchdarkly/go-sdk-common/v3/ldvalue" + "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldbuilders" + "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldmodel" + "github.com/launchdarkly/go-server-sdk/v7/internal/datakinds" + "github.com/launchdarkly/go-server-sdk/v7/subsystems" + "github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoretypes" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func makeTestFlagJSON(t testing.TB, key string, version int) []byte { + t.Helper() + flag := ldbuilders.NewFlagBuilder(key).Version(version). + On(true). + Variations(ldvalue.Bool(true), ldvalue.Bool(false)). + FallthroughVariation(0). + AddTarget(0, "user-a", "user-b"). + AddRule(ldbuilders.NewRuleBuilder().ID("rule-1").Variation(1).Clauses( + ldbuilders.Clause("email", "in", ldvalue.String("test@example.com")), + )). + Build() + return datakinds.Features.Serialize(ldstoretypes.ItemDescriptor{Version: version, Item: &flag}) +} + +func makeTestSegmentJSON(t testing.TB, key string, version int) []byte { + t.Helper() + segment := ldbuilders.NewSegmentBuilder(key).Version(version).Included("user-a").Build() + return datakinds.Segments.Serialize(ldstoretypes.ItemDescriptor{Version: version, Item: &segment}) +} + +func TestParsePutObjectEventDataPropertyOrderings(t *testing.T) { + flagJSON := makeTestFlagJSON(t, "flagkey", 3) + + kindFirst := fmt.Sprintf(`{"kind":"flag","key":"flagkey","version":3,"object":%s}`, flagJSON) + objectFirst := fmt.Sprintf(`{"object":%s,"version":3,"key":"flagkey","kind":"flag"}`, flagJSON) + + for name, data := range map[string]string{"kind before object": kindFirst, "object before kind": objectFirst} { + t.Run(name, func(t *testing.T) { + p, err := parsePutObjectEventData([]byte(data)) + require.NoError(t, err) + assert.Equal(t, subsystems.FlagKind, p.kind) + assert.Equal(t, "flagkey", p.key) + assert.Equal(t, 3, p.version) + assert.JSONEq(t, string(flagJSON), string(p.object)) + require.True(t, p.hasItem) + assert.Equal(t, 3, p.item.Version) + require.IsType(t, &ldmodel.FeatureFlag{}, p.item.Item) + assert.Equal(t, "flagkey", p.item.Item.(*ldmodel.FeatureFlag).Key) + }) + } +} + +func TestParsePutObjectEventDataSegmentKind(t *testing.T) { + segmentJSON := makeTestSegmentJSON(t, "segmentkey", 2) + data := fmt.Sprintf(`{"kind":"segment","key":"segmentkey","version":2,"object":%s}`, segmentJSON) + p, err := parsePutObjectEventData([]byte(data)) + require.NoError(t, err) + assert.Equal(t, subsystems.SegmentKind, p.kind) + require.True(t, p.hasItem) + require.IsType(t, &ldmodel.Segment{}, p.item.Item) + assert.Equal(t, "segmentkey", p.item.Item.(*ldmodel.Segment).Key) +} + +func TestParsePutObjectEventDataUnknownKind(t *testing.T) { + t.Run("valid object is retained raw", func(t *testing.T) { + data := `{"kind":"future-kind","key":"k","version":1,"object":{"key":"k","futureProp":[1,2]}}` + p, err := parsePutObjectEventData([]byte(data)) + require.NoError(t, err) + assert.Equal(t, subsystems.ObjectKind("future-kind"), p.kind) + assert.False(t, p.hasItem) + assert.JSONEq(t, `{"key":"k","futureProp":[1,2]}`, string(p.object)) + }) + + t.Run("malformed object is still an error", func(t *testing.T) { + // The braces balance, so this proves the object bytes are fully validated even though + // an unrecognized kind is never model-parsed (in the default build that validation + // comes from jreader.RawValue itself; in the easyjson build, from the json.RawMessage + // decode). + data := `{"kind":"future-kind","key":"k","version":1,"object":{"bad":}}` + _, err := parsePutObjectEventData([]byte(data)) + assert.Error(t, err) + }) +} + +func TestParsePutObjectEventDataMalformedObjectOfKnownKind(t *testing.T) { + data := `{"kind":"flag","key":"k","version":1,"object":{"key":12345,"on":"not-a-bool"}}` + _, err := parsePutObjectEventData([]byte(data)) + assert.Error(t, err) +} + +func TestParsePutObjectEventDataPreservesObjectBytesExactly(t *testing.T) { + // Whitespace surrounding the object value must not leak into the captured raw bytes, and + // whitespace inside the value must be preserved verbatim -- relay embeds these bytes + // unmodified into downstream events. + data := `{ "kind": "flag", "key": "flagkey", "version": 1,` + "\n\t" + + `"object": { "key": "flagkey", "version": 1, "on": true } }` + p, err := parsePutObjectEventData([]byte(data)) + require.NoError(t, err) + assert.Equal(t, `{ "key": "flagkey", "version": 1, "on": true }`, string(p.object)) + require.True(t, p.hasItem) + assert.Equal(t, "flagkey", p.item.Item.(*ldmodel.FeatureFlag).Key) +} + +func TestParsePutObjectEventDataIgnoresExtraProperties(t *testing.T) { + flagJSON := makeTestFlagJSON(t, "flagkey", 1) + data := fmt.Sprintf(`{"kind":"flag","futureField":{"a":[1]},"key":"flagkey","version":1,"object":%s}`, flagJSON) + p, err := parsePutObjectEventData([]byte(data)) + require.NoError(t, err) + assert.True(t, p.hasItem) +} + +func TestParseServerIntentEventData(t *testing.T) { + t.Run("single payload", func(t *testing.T) { + data := `{"payloads":[{"id":"p1","target":3,"intentCode":"xfer-full","reason":"payload-missing"}]}` + intent, err := parseServerIntentEventData([]byte(data)) + require.NoError(t, err) + assert.Equal(t, "p1", intent.Payload.ID) + assert.Equal(t, 3, intent.Payload.Target) + assert.Equal(t, subsystems.IntentTransferFull, intent.Payload.Code) + assert.Equal(t, "payload-missing", intent.Payload.Reason) + }) + + t.Run("only the first of multiple payloads is used", func(t *testing.T) { + data := `{"payloads":[{"id":"p1","intentCode":"none"},{"id":"p2","intentCode":"xfer-full"}]}` + intent, err := parseServerIntentEventData([]byte(data)) + require.NoError(t, err) + assert.Equal(t, "p1", intent.Payload.ID) + assert.Equal(t, subsystems.IntentNone, intent.Payload.Code) + }) + + t.Run("empty payloads is an error", func(t *testing.T) { + _, err := parseServerIntentEventData([]byte(`{"payloads":[]}`)) + assert.Error(t, err) + }) + + t.Run("missing payloads is an error", func(t *testing.T) { + _, err := parseServerIntentEventData([]byte(`{}`)) + assert.Error(t, err) + }) +} + +func TestParseDeleteObjectEventData(t *testing.T) { + data := `{"version":7,"kind":"segment","key":"gone"}` + d, err := parseDeleteObjectEventData([]byte(data)) + require.NoError(t, err) + assert.Equal(t, subsystems.DeleteObject{Version: 7, Kind: subsystems.SegmentKind, Key: "gone"}, d) +} + +func TestParseSelectorEventData(t *testing.T) { + t.Run("valid", func(t *testing.T) { + selector, err := parseSelectorEventData([]byte(`{"state":"s1","version":9}`)) + require.NoError(t, err) + assert.Equal(t, "s1", selector.State()) + assert.Equal(t, 9, selector.Version()) + }) + + t.Run("missing state is an error", func(t *testing.T) { + _, err := parseSelectorEventData([]byte(`{"version":9}`)) + assert.Error(t, err) + }) + + t.Run("missing version is an error", func(t *testing.T) { + _, err := parseSelectorEventData([]byte(`{"state":"s1"}`)) + assert.Error(t, err) + }) +} + +func TestParseGoodbyeEventData(t *testing.T) { + goodbye, err := parseGoodbyeEventData([]byte(`{"reason":"see ya","silent":true}`)) + require.NoError(t, err) + assert.Equal(t, "see ya", goodbye.Reason) +} + +func TestParseErrorEventData(t *testing.T) { + errorData, err := parseErrorEventData([]byte(`{"payloadId":"p1","reason":"broke"}`)) + require.NoError(t, err) + assert.Equal(t, subsystems.Error{PayloadID: "p1", Reason: "broke"}, errorData) +} + +func makePollingBody(t testing.TB, flagCount int) []byte { + t.Helper() + var sb strings.Builder + sb.WriteString(`{"events":[`) + sb.WriteString(`{"event":"server-intent","data":{"payloads":[{"id":"p1","target":1,"intentCode":"xfer-full","reason":"payload-missing"}]}}`) + for i := 0; i < flagCount; i++ { + fmt.Fprintf(&sb, `,{"event":"put-object","data":{"kind":"flag","key":"flag-%d","version":%d,"object":%s}}`, + i, i+1, makeTestFlagJSON(t, fmt.Sprintf("flag-%d", i), i+1)) + } + fmt.Fprintf(&sb, `,{"event":"put-object","data":{"kind":"segment","key":"segment-0","version":1,"object":%s}}`, + makeTestSegmentJSON(t, "segment-0", 1)) + sb.WriteString(`,{"event":"delete-object","data":{"kind":"flag","key":"deleted-flag","version":99}}`) + sb.WriteString(`,{"event":"payload-transferred","data":{"state":"p1:10","version":10}}`) + sb.WriteString(`]}`) + return []byte(sb.String()) +} + +func TestParsePollingPayloadFullTransfer(t *testing.T) { + body := makePollingBody(t, 2) + changeSet, err := parsePollingPayload(context.Background(), body) + require.NoError(t, err) + + assert.Equal(t, subsystems.IntentTransferFull, changeSet.IntentCode()) + assert.Equal(t, "p1:10", changeSet.Selector().State()) + assert.Equal(t, 10, changeSet.Selector().Version()) + + changes := changeSet.Changes() + require.Len(t, changes, 4) // 2 flags + 1 segment + 1 delete + assert.Equal(t, subsystems.ChangeTypePut, changes[0].Action) + assert.Equal(t, "flag-0", changes[0].Key) + assert.JSONEq(t, string(makeTestFlagJSON(t, "flag-0", 1)), string(changes[0].Object)) + assert.Equal(t, subsystems.ChangeTypeDelete, changes[3].Action) + assert.Equal(t, "deleted-flag", changes[3].Key) + + collections, err := changeSet.Collections() + require.NoError(t, err) + itemsByKind := map[string]int{} + for _, coll := range collections { + itemsByKind[coll.Kind.GetName()] = len(coll.Items) + } + assert.Equal(t, map[string]int{"features": 3, "segments": 1}, itemsByKind) // 2 puts + 1 tombstone + for _, coll := range collections { + if coll.Kind.GetName() != "features" { + continue + } + for _, item := range coll.Items { + if item.Key == "deleted-flag" { + assert.Nil(t, item.Item.Item) + assert.Equal(t, 99, item.Item.Version) + } else { + assert.NotNil(t, item.Item.Item) + } + } + } +} + +func TestParsePollingPayloadIntentNone(t *testing.T) { + body := []byte(`{"events":[{"event":"server-intent","data":{"payloads":[{"id":"p1","target":1,"intentCode":"none","reason":"up-to-date"}]}}]}`) + changeSet, err := parsePollingPayload(context.Background(), body) + require.NoError(t, err) + assert.Equal(t, subsystems.IntentNone, changeSet.IntentCode()) + assert.Empty(t, changeSet.Changes()) +} + +func TestParsePollingPayloadNoKnownEventsIsError(t *testing.T) { + for name, body := range map[string]string{ + "empty events": `{"events":[]}`, + "no events property": `{"other":true}`, + "only unknown events": `{"events":[{"event":"future-event","data":{"a":1}}]}`, + } { + t.Run(name, func(t *testing.T) { + _, err := parsePollingPayload(context.Background(), []byte(body)) + assert.Error(t, err) + }) + } +} + +func TestParsePollingPayloadMalformedBody(t *testing.T) { + for _, body := range []string{ + `{`, + `[]`, + `{"events":[{"event":"server-intent","data":{"payloads":[{}]}}`, + `{"events":[{"event":"put-object","data":{"kind":"flag","key":"k","version":1,"object":{"key":}}}]}`, + } { + t.Run(body, func(t *testing.T) { + _, err := parsePollingPayload(context.Background(), []byte(body)) + assert.Error(t, err) + }) + } +} + +func TestParsePollingPayloadUnknownEventsAreIgnored(t *testing.T) { + body := []byte(`{"events":[` + + `{"event":"future-event","data":{"whatever":[1,2,3]}},` + + `{"event":"server-intent","data":{"payloads":[{"id":"p1","target":1,"intentCode":"xfer-full","reason":"r"}]}},` + + `{"event":"another-future-event","data":"scalar data"},` + + `{"event":"payload-transferred","data":{"state":"s","version":1}}` + + `]}`) + changeSet, err := parsePollingPayload(context.Background(), body) + require.NoError(t, err) + assert.Equal(t, subsystems.IntentTransferFull, changeSet.IntentCode()) + assert.Equal(t, "s", changeSet.Selector().State()) +} + +func TestParsePollingPayloadDataBeforeEventName(t *testing.T) { + flagJSON := makeTestFlagJSON(t, "flagkey", 1) + body := []byte(fmt.Sprintf(`{"events":[`+ + `{"data":{"payloads":[{"id":"p1","target":1,"intentCode":"xfer-full","reason":"r"}]},"event":"server-intent"},`+ + `{"data":{"kind":"flag","key":"flagkey","version":1,"object":%s},"event":"put-object"},`+ + `{"data":{"state":"s","version":1},"event":"payload-transferred"}`+ + `]}`, flagJSON)) + changeSet, err := parsePollingPayload(context.Background(), body) + require.NoError(t, err) + changes := changeSet.Changes() + require.Len(t, changes, 1) + assert.Equal(t, "flagkey", changes[0].Key) + assert.JSONEq(t, string(flagJSON), string(changes[0].Object)) + collections, err := changeSet.Collections() + require.NoError(t, err) + require.Len(t, collections, 1) + require.Len(t, collections[0].Items, 1) + assert.NotNil(t, collections[0].Items[0].Item.Item) +} + +func TestParsePollingPayloadEventWithoutDataIsError(t *testing.T) { + body := []byte(`{"events":[` + + `{"event":"server-intent","data":{"payloads":[{"id":"p1","target":1,"intentCode":"xfer-full","reason":"r"}]}},` + + `{"event":"put-object"},` + + `{"event":"payload-transferred","data":{"state":"s","version":1}}` + + `]}`) + _, err := parsePollingPayload(context.Background(), body) + assert.Error(t, err) +} + +func TestParsePollingPayloadEventsAfterTransferAreIgnored(t *testing.T) { + // The trailing put-object is valid JSON but is never dispatched, because the payload is + // already complete. (The default build's single-pass parser stops reading the body entirely + // at that point; the easyjson build's envelope decode still requires the remainder to be + // well-formed JSON, which is why this event must be syntactically valid.) + body := []byte(`{"events":[` + + `{"event":"server-intent","data":{"payloads":[{"id":"p1","target":1,"intentCode":"xfer-full","reason":"r"}]}},` + + `{"event":"payload-transferred","data":{"state":"s","version":1}},` + + `{"event":"put-object","data":{"kind":"flag","key":"never-dispatched","version":1,"object":{"key":12345}}}` + + `]}`, + ) + changeSet, err := parsePollingPayload(context.Background(), body) + require.NoError(t, err) + assert.Empty(t, changeSet.Changes()) +} + +func TestParsePollingPayloadHonorsContextCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, err := parsePollingPayload(ctx, makePollingBody(t, 2)) + assert.ErrorIs(t, err, context.Canceled) +} + +// parsePollingPayloadReflectionReference replicates the previous reflection-based parse +// (PollingPayload envelope unmarshal, per-event unmarshal, raw puts) so that tests and +// benchmarks can verify the single-pass parser against it. +func parsePollingPayloadReflectionReference(body []byte) (*subsystems.ChangeSet, error) { + var payload subsystems.PollingPayload + if err := json.Unmarshal(body, &payload); err != nil { + return nil, err + } + builder := subsystems.NewChangeSetBuilder() + for _, event := range payload.Events { + switch event.Name { + case subsystems.EventServerIntent: + var serverIntent subsystems.ServerIntent + if err := json.Unmarshal(event.Data, &serverIntent); err != nil { + return nil, err + } + if serverIntent.Payload.Code == subsystems.IntentNone { + return builder.NoChanges(), nil + } + builder.Start(serverIntent) + case subsystems.EventPutObject: + var put subsystems.PutObject + if err := json.Unmarshal(event.Data, &put); err != nil { + return nil, err + } + builder.AddPut(put.Kind, put.Key, put.Version, put.Object) + case subsystems.EventDeleteObject: + var deleteObject subsystems.DeleteObject + if err := json.Unmarshal(event.Data, &deleteObject); err != nil { + return nil, err + } + builder.AddDelete(deleteObject.Kind, deleteObject.Key, deleteObject.Version) + case subsystems.EventPayloadTransferred: + var selector subsystems.Selector + if err := json.Unmarshal(event.Data, &selector); err != nil { + return nil, err + } + return builder.Finish(selector) + } + } + return nil, fmt.Errorf("didn't receive any known protocol events in polling payload") +} + +func TestParsePollingPayloadMatchesReflectionReference(t *testing.T) { + body := makePollingBody(t, 20) + + got, err := parsePollingPayload(context.Background(), body) + require.NoError(t, err) + want, err := parsePollingPayloadReflectionReference(body) + require.NoError(t, err) + + assert.Equal(t, want.IntentCode(), got.IntentCode()) + assert.Equal(t, want.Selector(), got.Selector()) + + wantChanges, gotChanges := want.Changes(), got.Changes() + require.Equal(t, len(wantChanges), len(gotChanges)) + for i := range wantChanges { + assert.Equal(t, wantChanges[i].Action, gotChanges[i].Action, "change %d", i) + assert.Equal(t, wantChanges[i].Kind, gotChanges[i].Kind, "change %d", i) + assert.Equal(t, wantChanges[i].Key, gotChanges[i].Key, "change %d", i) + assert.Equal(t, wantChanges[i].Version, gotChanges[i].Version, "change %d", i) + assert.Equal(t, string(wantChanges[i].Object), string(gotChanges[i].Object), "change %d", i) + } + + wantCollections, err := want.Collections() + require.NoError(t, err) + gotCollections, err := got.Collections() + require.NoError(t, err) + wantItems := flattenCollectionsForComparison(wantCollections) + gotItems := flattenCollectionsForComparison(gotCollections) + assert.Equal(t, wantItems, gotItems) +} + +func flattenCollectionsForComparison(collections []ldstoretypes.Collection) map[string]ldstoretypes.ItemDescriptor { + result := map[string]ldstoretypes.ItemDescriptor{} + for _, coll := range collections { + for _, item := range coll.Items { + result[coll.Kind.GetName()+"/"+item.Key] = item.Item + } + } + return result +} diff --git a/internal/datasourcev2/polling_http_request.go b/internal/datasourcev2/polling_http_request.go index b44554bb..83f49a99 100644 --- a/internal/datasourcev2/polling_http_request.go +++ b/internal/datasourcev2/polling_http_request.go @@ -2,7 +2,7 @@ package datasourcev2 import ( "context" - "encoding/json" + "errors" "fmt" "io" "maps" @@ -83,53 +83,14 @@ func (r *pollingRequester) Request( return subsystems.NewChangeSetBuilder().NoChanges(), headers, nil } - var payload subsystems.PollingPayload - if err = json.Unmarshal(body, &payload); err != nil { - return nil, headers, malformedJSONError{err} - } - - changeSet := subsystems.NewChangeSetBuilder() - - for _, event := range payload.Events { - select { - case <-ctx.Done(): - return nil, headers, ctx.Err() - default: - switch event.Name { - case subsystems.EventServerIntent: - var serverIntent subsystems.ServerIntent - err := json.Unmarshal(event.Data, &serverIntent) - if err != nil { - return nil, headers, err - } - if serverIntent.Payload.Code == subsystems.IntentNone { - return changeSet.NoChanges(), headers, nil - } - changeSet.Start(serverIntent) - case subsystems.EventPutObject: - var put subsystems.PutObject - if err := json.Unmarshal(event.Data, &put); err != nil { - return nil, headers, err - } - changeSet.AddPut(put.Kind, put.Key, put.Version, put.Object) - case subsystems.EventDeleteObject: - var deleteObject subsystems.DeleteObject - if err := json.Unmarshal(event.Data, &deleteObject); err != nil { - return nil, headers, err - } - changeSet.AddDelete(deleteObject.Kind, deleteObject.Key, deleteObject.Version) - case subsystems.EventPayloadTransferred: - var selector subsystems.Selector - if err := json.Unmarshal(event.Data, &selector); err != nil { - return nil, headers, err - } - changeset, err := changeSet.Finish(selector) - return changeset, headers, err - } + changeSet, err := parsePollingPayload(ctx, body) + if err != nil { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return nil, headers, err } + return nil, headers, malformedJSONError{err} } - - return nil, headers, fmt.Errorf("didn't receive any known protocol events in polling payload") + return changeSet, headers, nil } func (r *pollingRequester) makeRequest( diff --git a/internal/datasourcev2/streaming_data_source.go b/internal/datasourcev2/streaming_data_source.go index 15f6fe19..d477e125 100644 --- a/internal/datasourcev2/streaming_data_source.go +++ b/internal/datasourcev2/streaming_data_source.go @@ -2,7 +2,6 @@ package datasourcev2 import ( "context" - "encoding/json" "errors" "maps" "net/http" @@ -214,8 +213,7 @@ func (sp *StreamProcessor) consumeStream(stream *es.Stream, resultChan chan<- su // Swallow the event and move on. case subsystems.EventServerIntent: - var serverIntent subsystems.ServerIntent - err := json.Unmarshal([]byte(event.Data()), &serverIntent) + serverIntent, err := parseServerIntentEventData([]byte(event.Data())) if err != nil { gotMalformedEvent(event, err) break @@ -241,24 +239,21 @@ func (sp *StreamProcessor) consumeStream(stream *es.Stream, resultChan chan<- su } case subsystems.EventPutObject: - var p subsystems.PutObject - err := json.Unmarshal([]byte(event.Data()), &p) + p, err := parsePutObjectEventData([]byte(event.Data())) if err != nil { gotMalformedEvent(event, err) break } - changeSetBuilder.AddPut(p.Kind, p.Key, p.Version, p.Object) + p.addTo(changeSetBuilder) case subsystems.EventDeleteObject: - var d subsystems.DeleteObject - err := json.Unmarshal([]byte(event.Data()), &d) + d, err := parseDeleteObjectEventData([]byte(event.Data())) if err != nil { gotMalformedEvent(event, err) break } changeSetBuilder.AddDelete(d.Kind, d.Key, d.Version) case subsystems.EventGoodbye: - var goodbye subsystems.Goodbye - err := json.Unmarshal([]byte(event.Data()), &goodbye) + goodbye, err := parseGoodbyeEventData([]byte(event.Data())) if err != nil { gotMalformedEvent(event, err) break @@ -266,8 +261,7 @@ func (sp *StreamProcessor) consumeStream(stream *es.Stream, resultChan chan<- su sp.loggers.Infof("SSE server sent goodbye: %s", goodbye.Reason) case subsystems.EventError: - var errorData subsystems.Error - err := json.Unmarshal([]byte(event.Data()), &errorData) + errorData, err := parseErrorEventData([]byte(event.Data())) if err != nil { gotMalformedEvent(event, err) break @@ -284,8 +278,7 @@ func (sp *StreamProcessor) consumeStream(stream *es.Stream, resultChan chan<- su changeSetBuilder.Reset() case subsystems.EventPayloadTransferred: - var selector subsystems.Selector - err := json.Unmarshal([]byte(event.Data()), &selector) + selector, err := parseSelectorEventData([]byte(event.Data())) if err != nil { gotMalformedEvent(event, err) break diff --git a/subsystems/changeset.go b/subsystems/changeset.go index ed294b10..94bfe7c3 100644 --- a/subsystems/changeset.go +++ b/subsystems/changeset.go @@ -122,7 +122,17 @@ func toStorableItems(deltas []Change) ([]ldstoretypes.Collection, error) { // ChangeSetBuilder is a helper for constructing a ChangeSet. type ChangeSetBuilder struct { intent *ServerIntent - changes []Change + changes []changeRecord +} + +// changeRecord is a Change plus, optionally, the already-parsed representation of the change's +// object. When every put of a recognized kind carries a parsed item, the builder can assemble +// the ChangeSet's collections directly at Finish time, so that Collections() does not need to +// re-parse the raw JSON. +type changeRecord struct { + change Change + item ldstoretypes.ItemDescriptor + hasItem bool } // NewChangeSetBuilder creates a new ChangeSetBuilder, which is empty by default. @@ -265,12 +275,22 @@ func (c *ChangeSetBuilder) Finish(selector Selector) (*ChangeSet, error) { if c.intent == nil { return nil, errors.New("changeset: cannot complete without a server-intent") } - changes := &ChangeSet{ + var changes []Change + if c.changes != nil { + changes = make([]Change, 0, len(c.changes)) + for _, record := range c.changes { + changes = append(changes, record.change) + } + } + changeSet := &ChangeSet{ intentCode: c.intent.Payload.Code, selector: selector, - changes: c.changes, + changes: changes, mu: &sync.Mutex{}, } + if collections, ok := collectionsFromRecords(c.changes); ok { + changeSet.collection = collections + } c.changes = nil // Once a full transfer has been processed, all future changes should be @@ -279,17 +299,91 @@ func (c *ChangeSetBuilder) Finish(selector Selector) (*ChangeSet, error) { if c.intent.Payload.Code == IntentTransferFull { c.intent.Payload.Code = IntentTransferChanges } - return changes, nil + return changeSet, nil +} + +// collectionsFromRecords assembles storable collections from the builder's records without +// parsing any JSON, if possible. This requires every put of a recognized kind to carry a +// pre-parsed item; otherwise it returns (nil, false) and the ChangeSet falls back to parsing +// lazily in Collections(). +func collectionsFromRecords(records []changeRecord) ([]ldstoretypes.Collection, bool) { + if records == nil { + return nil, false + } + for _, record := range records { + if record.change.Action == ChangeTypePut && !record.hasItem { + if _, recognized := record.change.Kind.ToFDV1(); recognized { + return nil, false + } + } + } + collections := make(kindMap) + for _, record := range records { + kind, ok := record.change.Kind.ToFDV1() + if !ok { + // Unrecognized kinds are not errors and are ignored for forwards compatibility. + continue + } + switch record.change.Action { + case ChangeTypePut: + collections[kind] = append(collections[kind], ldstoretypes.KeyedItemDescriptor{ + Key: record.change.Key, + Item: record.item, + }) + case ChangeTypeDelete: + // A deletion is represented by a tombstone, which is an ItemDescriptor with a version and nil item. + collections[kind] = append(collections[kind], ldstoretypes.KeyedItemDescriptor{ + Key: record.change.Key, + Item: ldstoretypes.ItemDescriptor{Version: record.change.Version, Item: nil}, + }) + default: + // An unknown action isn't an error, and should be ignored for forwards compatibility. + continue + } + } + result := collections.flatten() + if result == nil { + result = []ldstoretypes.Collection{} + } + return result, true } // AddPut adds a new object to the changeset. func (c *ChangeSetBuilder) AddPut(kind ObjectKind, key string, version int, object json.RawMessage) *ChangeSetBuilder { - c.changes = append(c.changes, Change{ + c.changes = append(c.changes, changeRecord{change: Change{ Action: ChangeTypePut, Kind: kind, Key: key, Version: version, Object: object, + }}) + + return c +} + +// AddParsedPut adds a new object to the changeset, along with the already-parsed representation +// of that object. The object parameter is the raw JSON of the item, exactly as it would be passed +// to AddPut, and item is the result of deserializing those same bytes. +// +// Providing the parsed item up front allows the ChangeSet returned by Finish to assemble its +// Collections() result directly, instead of re-parsing the raw JSON of every put. +func (c *ChangeSetBuilder) AddParsedPut( + kind ObjectKind, + key string, + version int, + object json.RawMessage, + item ldstoretypes.ItemDescriptor, +) *ChangeSetBuilder { + c.changes = append(c.changes, changeRecord{ + change: Change{ + Action: ChangeTypePut, + Kind: kind, + Key: key, + Version: version, + Object: object, + }, + item: item, + hasItem: true, }) return c @@ -297,12 +391,12 @@ func (c *ChangeSetBuilder) AddPut(kind ObjectKind, key string, version int, obje // AddDelete adds a deletion to the changeset. func (c *ChangeSetBuilder) AddDelete(kind ObjectKind, key string, version int) *ChangeSetBuilder { - c.changes = append(c.changes, Change{ + c.changes = append(c.changes, changeRecord{change: Change{ Action: ChangeTypeDelete, Kind: kind, Key: key, Version: version, - }) + }}) return c } diff --git a/subsystems/changeset_parsed_put_test.go b/subsystems/changeset_parsed_put_test.go new file mode 100644 index 00000000..73afde6a --- /dev/null +++ b/subsystems/changeset_parsed_put_test.go @@ -0,0 +1,133 @@ +package subsystems + +import ( + "testing" + + "github.com/launchdarkly/go-sdk-common/v3/ldvalue" + "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldbuilders" + "github.com/launchdarkly/go-server-sdk-evaluation/v3/ldmodel" + "github.com/launchdarkly/go-server-sdk/v7/internal/datakinds" + "github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoretypes" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func makeParsedFlagItem(key string, version int) ldstoretypes.ItemDescriptor { + flag := ldbuilders.NewFlagBuilder(key).Version(version).SingleVariation(ldvalue.Bool(true)).Build() + return ldstoretypes.ItemDescriptor{Version: version, Item: &flag} +} + +func serializeFlagItem(item ldstoretypes.ItemDescriptor) []byte { + return datakinds.Features.Serialize(item) +} + +func TestChangeSetBuilderAddParsedPutPrePopulatesCollections(t *testing.T) { + builder := NewChangeSetBuilder() + builder.Start(ServerIntent{Payload: Payload{Code: IntentTransferFull}}) + + // The raw object bytes are deliberately NOT valid JSON. If Collections() were to re-parse + // them, it would fail; succeeding proves the pre-parsed items were used instead. + builder.AddParsedPut(FlagKind, "flag-1", 1, []byte("deliberately not JSON"), makeParsedFlagItem("flag-1", 1)) + builder.AddDelete(FlagKind, "flag-2", 2) + + changeSet, err := builder.Finish(NewSelector("state", 1)) + require.NoError(t, err) + + // The raw bytes are preserved verbatim for consumers of Changes(). + changes := changeSet.Changes() + require.Len(t, changes, 2) + assert.Equal(t, "deliberately not JSON", string(changes[0].Object)) + + collections, err := changeSet.Collections() + require.NoError(t, err) + require.Len(t, collections, 1) + assert.Equal(t, "features", collections[0].Kind.GetName()) + require.Len(t, collections[0].Items, 2) + + byKey := map[string]ldstoretypes.ItemDescriptor{} + for _, item := range collections[0].Items { + byKey[item.Key] = item.Item + } + require.IsType(t, &ldmodel.FeatureFlag{}, byKey["flag-1"].Item) + assert.Equal(t, 1, byKey["flag-1"].Version) + assert.Nil(t, byKey["flag-2"].Item) // the delete is a tombstone + assert.Equal(t, 2, byKey["flag-2"].Version) +} + +func TestChangeSetBuilderMixedPutsFallBackToLazyParsing(t *testing.T) { + builder := NewChangeSetBuilder() + builder.Start(ServerIntent{Payload: Payload{Code: IntentTransferFull}}) + + item1 := makeParsedFlagItem("flag-1", 1) + builder.AddParsedPut(FlagKind, "flag-1", 1, serializeFlagItem(item1), item1) + // This put has no parsed item, so the whole change set must be parsed lazily. + builder.AddPut(FlagKind, "flag-2", 2, serializeFlagItem(makeParsedFlagItem("flag-2", 2))) + + changeSet, err := builder.Finish(NewSelector("state", 1)) + require.NoError(t, err) + + collections, err := changeSet.Collections() + require.NoError(t, err) + require.Len(t, collections, 1) + assert.Len(t, collections[0].Items, 2) + for _, item := range collections[0].Items { + assert.NotNil(t, item.Item.Item, "item %s should have been parsed", item.Key) + } +} + +func TestChangeSetBuilderUnparsedPutOfUnknownKindDoesNotBlockPrePopulation(t *testing.T) { + builder := NewChangeSetBuilder() + builder.Start(ServerIntent{Payload: Payload{Code: IntentTransferFull}}) + + // An unrecognized kind never has a parsed item, and is excluded from collections; its + // presence must not force the lazy path (which the invalid raw bytes of the parsed put + // would then fail on). + builder.AddPut("future-kind", "future-key", 1, []byte(`{"anything":true}`)) + builder.AddParsedPut(FlagKind, "flag-1", 1, []byte("deliberately not JSON"), makeParsedFlagItem("flag-1", 1)) + + changeSet, err := builder.Finish(NewSelector("state", 1)) + require.NoError(t, err) + + // Both changes are visible to Changes() consumers... + require.Len(t, changeSet.Changes(), 2) + assert.Equal(t, ObjectKind("future-kind"), changeSet.Changes()[0].Kind) + + // ...but only the recognized kind appears in collections. + collections, err := changeSet.Collections() + require.NoError(t, err) + require.Len(t, collections, 1) + assert.Equal(t, "features", collections[0].Kind.GetName()) + assert.Len(t, collections[0].Items, 1) +} + +func TestChangeSetBuilderParsedPutsAcrossFinishReuse(t *testing.T) { + builder := NewChangeSetBuilder() + builder.Start(ServerIntent{Payload: Payload{Code: IntentTransferFull}}) + builder.AddParsedPut(FlagKind, "flag-1", 1, []byte("raw-1"), makeParsedFlagItem("flag-1", 1)) + first, err := builder.Finish(NewSelector("state", 1)) + require.NoError(t, err) + + // The builder is reusable after Finish; the next change set starts empty and the intent + // becomes xfer-changes. + builder.AddParsedPut(FlagKind, "flag-2", 2, []byte("raw-2"), makeParsedFlagItem("flag-2", 2)) + second, err := builder.Finish(NewSelector("state", 2)) + require.NoError(t, err) + + require.Len(t, first.Changes(), 1) + assert.Equal(t, "flag-1", first.Changes()[0].Key) + assert.Equal(t, IntentTransferFull, first.IntentCode()) + + require.Len(t, second.Changes(), 1) + assert.Equal(t, "flag-2", second.Changes()[0].Key) + assert.Equal(t, IntentTransferChanges, second.IntentCode()) + + firstCollections, err := first.Collections() + require.NoError(t, err) + secondCollections, err := second.Collections() + require.NoError(t, err) + require.Len(t, firstCollections, 1) + require.Len(t, secondCollections, 1) + assert.Equal(t, "flag-1", firstCollections[0].Items[0].Key) + assert.Equal(t, "flag-2", secondCollections[0].Items[0].Key) +} From 47c52c487c546ec7c3877baa8248ff2e37922943 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Wed, 29 Jul 2026 15:52:11 -0400 Subject: [PATCH 3/7] chore: Point go-jsonstream at the merged v3 RawValue commit The single-pass decoder depended on a local, unmerged RawValue commit that does not exist on origin. RawValue is now merged to go-jsonstream v3, so pin the resolvable v3 tip and drop the reliance on a local-only pseudo-version. --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index b53c2838..0e724529 100644 --- a/go.mod +++ b/go.mod @@ -8,7 +8,7 @@ require ( github.com/gregjones/httpcache v0.0.0-20171119193500-2bcd89a1743f github.com/launchdarkly/ccache v1.1.0 github.com/launchdarkly/eventsource v1.10.0 - github.com/launchdarkly/go-jsonstream/v3 v3.1.2-0.20260728165303-975e0be34b70 + github.com/launchdarkly/go-jsonstream/v3 v3.1.2-0.20260729181618-28aa1e35cda3 github.com/launchdarkly/go-ntlm-proxy-auth v1.0.3 github.com/launchdarkly/go-sdk-common/v3 v3.5.0 github.com/launchdarkly/go-sdk-events/v3 v3.6.2 diff --git a/go.sum b/go.sum index fb636df7..51e6ba63 100644 --- a/go.sum +++ b/go.sum @@ -19,8 +19,8 @@ github.com/launchdarkly/ccache v1.1.0 h1:voD1M+ZJXR3MREOKtBwgTF9hYHl1jg+vFKS/+VA github.com/launchdarkly/ccache v1.1.0/go.mod h1:TlxzrlnzvYeXiLHmesMuvoZetu4Z97cV1SsdqqBJi1Q= github.com/launchdarkly/eventsource v1.10.0 h1:H9Tp6AfGu/G2qzBJC26iperrvwhzdbiA/gx7qE2nDFI= github.com/launchdarkly/eventsource v1.10.0/go.mod h1:J3oa50bPvJesZqNAJtb5btSIo5N6roDWhiAS3IpsKck= -github.com/launchdarkly/go-jsonstream/v3 v3.1.2-0.20260728165303-975e0be34b70 h1:Z4GlojB8T/d8jeAKFbEOpVYOOvCbIcNJYGeB0YIauiE= -github.com/launchdarkly/go-jsonstream/v3 v3.1.2-0.20260728165303-975e0be34b70/go.mod h1:ZBjhKq8mhArCtqotGRGnteY6eXpNm1GaOdUSZHh+ZjM= +github.com/launchdarkly/go-jsonstream/v3 v3.1.2-0.20260729181618-28aa1e35cda3 h1:wT4WhK2sqM4j0nJwa+eO7g+XCblsT5myDhiadFaXeA8= +github.com/launchdarkly/go-jsonstream/v3 v3.1.2-0.20260729181618-28aa1e35cda3/go.mod h1:ZBjhKq8mhArCtqotGRGnteY6eXpNm1GaOdUSZHh+ZjM= github.com/launchdarkly/go-ntlm-proxy-auth v1.0.3 h1:i3V0N+R0Fd2nXfGEVKCBIZ8kyttZ+SRKvBG8cdcphO4= github.com/launchdarkly/go-ntlm-proxy-auth v1.0.3/go.mod h1:kU5uMfNSTpYE6fIzmAXjFxUdmnaDPUEQ5zKm3RVKUsY= github.com/launchdarkly/go-ntlmssp v1.0.3 h1:rFxOnnEJ2DzJ+NU0plhXqnldJUwn3wWJFTWKCmaiQdE= From dbeab421f9c717bc9676411d6f25a207e0747855 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Wed, 29 Jul 2026 15:52:11 -0400 Subject: [PATCH 4/7] fix: Reject fractional protocol versions and collapse duplicate polling data Two behavioral divergences between the default single-pass build and the launchdarkly_easyjson build: - A fractional version/target (e.g. 1.9) was silently truncated by jreader.Int in the default build but rejected by the reflection decode. Add a strict readInt used at every integer field so both builds reject it. - A duplicated "data" property on a polling event was applied more than once by the default build. Capture the event data and dispatch once, so it resolves last-wins exactly as encoding/json does. Both variants now behave identically. Adds regression tests that run under both build tags. --- internal/datasourcev2/event_parsing.go | 20 +++- .../datasourcev2/event_parsing_default.go | 113 ++++++++---------- internal/datasourcev2/event_parsing_test.go | 63 ++++++++++ 3 files changed, 127 insertions(+), 69 deletions(-) diff --git a/internal/datasourcev2/event_parsing.go b/internal/datasourcev2/event_parsing.go index 1e671717..c80832cc 100644 --- a/internal/datasourcev2/event_parsing.go +++ b/internal/datasourcev2/event_parsing.go @@ -3,6 +3,7 @@ package datasourcev2 import ( "encoding/json" "errors" + "math" "github.com/launchdarkly/go-jsonstream/v3/jreader" "github.com/launchdarkly/go-server-sdk/v7/subsystems" @@ -45,6 +46,19 @@ const ( const errNoKnownPollingEvents = "didn't receive any known protocol events in polling payload" +// readInt reads a JSON number that must be an integer. jreader.Int coerces via int(Float64()), +// which silently truncates a fractional number (for example 1.9 becomes 1). The FDv2 protocol's +// version and target fields are integers, so a fractional value is a malformed payload and is +// rejected here, matching the reflection-based decode used by the launchdarkly_easyjson build. +func readInt(r *jreader.Reader) int { + f := r.Float64() + if f != math.Trunc(f) { + r.AddError(jreader.SyntaxError{Message: "expected integer, got fractional number"}) + return 0 + } + return int(f) +} + // parsedPutObject is the result of decoding a put-object event's data. type parsedPutObject struct { kind subsystems.ObjectKind @@ -100,7 +114,7 @@ func readServerIntent(r *jreader.Reader) (subsystems.ServerIntent, error) { case propID: intent.Payload.ID = r.String() case propTarget: - intent.Payload.Target = r.Int() + intent.Payload.Target = readInt(r) case propIntentCode: intent.Payload.Code = subsystems.IntentCode(r.String()) case propReason: @@ -141,7 +155,7 @@ func readDeleteObject(r *jreader.Reader) (subsystems.DeleteObject, error) { case propKey: d.Key = r.String() case propVersion: - d.Version = r.Int() + d.Version = readInt(r) default: _ = r.SkipValue() } @@ -170,7 +184,7 @@ func readSelector(r *jreader.Reader) (subsystems.Selector, error) { state = r.String() gotState = true case propVersion: - version = r.Int() + version = readInt(r) gotVersion = true default: _ = r.SkipValue() diff --git a/internal/datasourcev2/event_parsing_default.go b/internal/datasourcev2/event_parsing_default.go index 214f1b3f..ace336e1 100644 --- a/internal/datasourcev2/event_parsing_default.go +++ b/internal/datasourcev2/event_parsing_default.go @@ -45,7 +45,7 @@ func parsePollingPayload(ctx context.Context, body []byte) (*subsystems.ChangeSe return nil, ctx.Err() default: } - changeSet, done, err := readPollingEvent(&r, body, builder) + changeSet, done, err := readPollingEvent(&r, builder) if err != nil { return nil, err } @@ -64,88 +64,30 @@ func parsePollingPayload(ctx context.Context, body []byte) (*subsystems.ChangeSe // event completes the payload (server-intent "none" or payload-transferred), it returns the // resulting change set with done set to true. // -// The common case dispatches on the event name and decodes the data in place as soon as it is -// reached. If the data property appears before the event name, its raw bytes are captured -// (zero-copy) and decoded once the name is known. +// The event's data is captured verbatim and decoded once the whole event object has been read. +// This keeps the decode independent of property order (the event name may follow the data) and +// makes a duplicated data property resolve last-wins, exactly as the reflection-based decode does. func readPollingEvent( r *jreader.Reader, - body []byte, builder *subsystems.ChangeSetBuilder, ) (changeSet *subsystems.ChangeSet, done bool, err error) { var name subsystems.EventName - var deferredData []byte + var data []byte gotData := false - // dataInput must be the byte slice that dataReader was created over, so that offset-based - // span capture inside readPutObject can slice it. - dispatch := func(dataReader *jreader.Reader, dataInput []byte) error { - switch name { - case subsystems.EventServerIntent: - intent, err := readServerIntent(dataReader) - if err != nil { - return err - } - if intent.Payload.Code == subsystems.IntentNone { - changeSet, done = builder.NoChanges(), true - return nil - } - builder.Start(intent) - case subsystems.EventPutObject: - put, err := readPutObject(dataReader, dataInput) - if err != nil { - return err - } - put.addTo(builder) - case subsystems.EventDeleteObject: - deleteObject, err := readDeleteObject(dataReader) - if err != nil { - return err - } - builder.AddDelete(deleteObject.Kind, deleteObject.Key, deleteObject.Version) - case subsystems.EventPayloadTransferred: - selector, err := readSelector(dataReader) - if err != nil { - return err - } - finished, err := builder.Finish(selector) - if err != nil { - return err - } - changeSet, done = finished, true - default: - // An unknown event name is ignored for forwards compatibility. - return dataReader.SkipValue() - } - return nil - } for obj := r.Object(); obj.Next(); { switch string(obj.Name()) { case propEvent: name = subsystems.EventName(r.String()) case propData: gotData = true - if name == "" { - deferredData = r.RawValue() - } else if err := dispatch(r, body); err != nil { - return nil, false, err - } + data = r.RawValue() default: _ = r.SkipValue() } - if done { - // The payload is complete; the rest of the input is intentionally left unread. - return changeSet, true, nil - } } if err := r.Error(); err != nil { return nil, false, err } - if deferredData != nil { - dataReader := jreader.NewReader(deferredData) - if err := dispatch(&dataReader, deferredData); err != nil { - return nil, false, err - } - return changeSet, done, nil - } if !gotData { switch name { case subsystems.EventServerIntent, subsystems.EventPutObject, @@ -154,8 +96,47 @@ func readPollingEvent( // malformed rather than silently dropping the event. return nil, false, fmt.Errorf("polling payload event %q has no data", name) } + return nil, false, nil + } + + dataReader := jreader.NewReader(data) + switch name { + case subsystems.EventServerIntent: + intent, err := readServerIntent(&dataReader) + if err != nil { + return nil, false, err + } + if intent.Payload.Code == subsystems.IntentNone { + return builder.NoChanges(), true, nil + } + builder.Start(intent) + case subsystems.EventPutObject: + put, err := readPutObject(&dataReader, data) + if err != nil { + return nil, false, err + } + put.addTo(builder) + case subsystems.EventDeleteObject: + deleteObject, err := readDeleteObject(&dataReader) + if err != nil { + return nil, false, err + } + builder.AddDelete(deleteObject.Kind, deleteObject.Key, deleteObject.Version) + case subsystems.EventPayloadTransferred: + selector, err := readSelector(&dataReader) + if err != nil { + return nil, false, err + } + finished, err := builder.Finish(selector) + if err != nil { + return nil, false, err + } + return finished, true, nil + default: + // An unknown event name is ignored for forwards compatibility. RawValue already validated + // the data above. } - return changeSet, done, nil + return nil, false, nil } // parsePutObjectEventData decodes the data of a put-object event in a single pass, capturing the @@ -185,7 +166,7 @@ func readPutObject(r *jreader.Reader, input []byte) (parsedPutObject, error) { case propKey: p.key = r.String() case propVersion: - p.version = r.Int() + p.version = readInt(r) case propObject: if kind, recognized := p.kind.ToFDV1(); recognized { // The kind is already known (LaunchDarkly services always write "kind" before diff --git a/internal/datasourcev2/event_parsing_test.go b/internal/datasourcev2/event_parsing_test.go index cc73d339..12134c98 100644 --- a/internal/datasourcev2/event_parsing_test.go +++ b/internal/datasourcev2/event_parsing_test.go @@ -417,6 +417,69 @@ func TestParsePollingPayloadMatchesReflectionReference(t *testing.T) { assert.Equal(t, wantItems, gotItems) } +// A protocol integer field (version, target) must reject a fractional JSON number rather than +// silently truncating it. jreader.Int coerces via int(Float64()) -- so 1.9 would become 1 -- and +// the single-pass decoder relies on readInt to reject it, matching the reflection-based decode +// that the launchdarkly_easyjson build (and the pre-single-pass code) used. Both build variants +// run this test. +func TestFractionalIntegerProtocolFieldsAreRejected(t *testing.T) { + flagJSON := makeTestFlagJSON(t, "flagkey", 1) + cases := map[string]func() error{ + "put-object version": func() error { + _, err := parsePutObjectEventData([]byte(fmt.Sprintf( + `{"kind":"flag","key":"flagkey","version":1.9,"object":%s}`, flagJSON))) + return err + }, + "delete-object version": func() error { + _, err := parseDeleteObjectEventData([]byte(`{"kind":"flag","key":"k","version":1.9}`)) + return err + }, + "selector version": func() error { + _, err := parseSelectorEventData([]byte(`{"state":"s","version":1.9}`)) + return err + }, + "server-intent target": func() error { + _, err := parseServerIntentEventData([]byte( + `{"payloads":[{"id":"p1","target":1.9,"intentCode":"xfer-full","reason":"r"}]}`)) + return err + }, + } + for name, run := range cases { + t.Run(name, func(t *testing.T) { + assert.Error(t, run()) + }) + } +} + +// A polling event that carries the "data" property more than once (non-conformant, but handled for +// parity with encoding/json) resolves last-wins: the event is applied exactly once, using the final +// value. Both build variants run this test. +func TestParsePollingPayloadDuplicateDataPropertyUsesLastValue(t *testing.T) { + flagJSON := makeTestFlagJSON(t, "flagkey", 1) + body := []byte(fmt.Sprintf(`{"events":[`+ + `{"event":"server-intent","data":{"payloads":[{"id":"p1","target":1,"intentCode":"xfer-full","reason":"r"}]}},`+ + `{"event":"put-object",`+ + `"data":{"kind":"flag","key":"first","version":1,"object":%s},`+ + `"data":{"kind":"flag","key":"last","version":1,"object":%s}},`+ + `{"event":"payload-transferred","data":{"state":"s","version":1}}`+ + `]}`, flagJSON, flagJSON)) + + changeSet, err := parsePollingPayload(context.Background(), body) + require.NoError(t, err) + + changes := changeSet.Changes() + require.Len(t, changes, 1) + assert.Equal(t, "last", changes[0].Key) + + collections, err := changeSet.Collections() + require.NoError(t, err) + total := 0 + for _, coll := range collections { + total += len(coll.Items) + } + assert.Equal(t, 1, total) +} + func flattenCollectionsForComparison(collections []ldstoretypes.Collection) map[string]ldstoretypes.ItemDescriptor { result := map[string]ldstoretypes.ItemDescriptor{} for _, coll := range collections { From 55e2664fc75313051f2ae42f5d6e1d68b7c07667 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Thu, 30 Jul 2026 12:53:55 -0400 Subject: [PATCH 5/7] fix: Match encoding/json integer parsing in both FDv2 build variants readInt went through jreader.Float64, so it truncated fractional versions and produced an implementation-defined (CPU-architecture-dependent) result for a value outside the int range -- diverging from the reflection decode the launchdarkly_easyjson build uses. Split the scalar event decoders by build tag so both variants agree with encoding/json: the default build parses the raw number literal with strconv (rejecting fractional, exponent, and out-of-range values while preserving large integers exactly and treating null as zero), and the easyjson build decodes each scalar event with json.Unmarshal into the subsystems types. The payload- transferred selector is kept lenient in both builds to match Selector's float64-based decode. Adds a differential test that checks every integer field against an encoding/json oracle under both build tags. --- internal/datasourcev2/event_parsing.go | 211 ++---------------- .../datasourcev2/event_parsing_default.go | 198 ++++++++++++++++ .../datasourcev2/event_parsing_easyjson.go | 39 ++++ internal/datasourcev2/event_parsing_test.go | 98 +++++--- 4 files changed, 319 insertions(+), 227 deletions(-) diff --git a/internal/datasourcev2/event_parsing.go b/internal/datasourcev2/event_parsing.go index c80832cc..7428ee6e 100644 --- a/internal/datasourcev2/event_parsing.go +++ b/internal/datasourcev2/event_parsing.go @@ -2,63 +2,30 @@ package datasourcev2 import ( "encoding/json" - "errors" - "math" - "github.com/launchdarkly/go-jsonstream/v3/jreader" "github.com/launchdarkly/go-server-sdk/v7/subsystems" "github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoretypes" ) -// This file contains the decoders for the FDv2 protocol events that are shared by both build -// variants. The payload-level decoders -- the ones responsible for capturing each put-object -// item's raw JSON alongside its parsed form -- are split by build tag: +// This file contains the parts of FDv2 protocol event decoding that are shared by both build +// variants. The event decoders themselves are split by build tag: // -// - event_parsing_default.go: a single-pass jsonstream decoder using jreader.RawValue for -// zero-copy raw capture. Each event, and in particular each item's JSON, is scanned once. -// - event_parsing_easyjson.go: the previous reflection-based decode (kept for the -// launchdarkly_easyjson build so that no new code depends on the easyjson token reader, -// which is planned for removal). +// - event_parsing_default.go: a single-pass jsonstream decoder using jreader.RawValue/Offset for +// zero-copy raw capture. Each event, and in particular each put-object item's JSON, is scanned +// once. Its scalar decoders are written to match encoding/json's acceptance rules (see below). +// - event_parsing_easyjson.go: a reflection-based decode (encoding/json) for the +// launchdarkly_easyjson build, so that no new code depends on the easyjson token reader, which +// is planned for removal from go-jsonstream. // -// Both variants produce the same results: a parsedPutObject carrying the item's raw bytes and, -// for recognized kinds, its parsed form, so that the ChangeSet's collections are assembled -// without a later re-parse. The decoders are tolerant of property ordering: nothing here assumes -// that, for example, "kind" appears before "object" in a put-object event, even though -// LaunchDarkly services always write them in that order. - -// JSON property names shared by the FDv2 protocol event decoders. -const ( - propEvents = "events" - propEvent = "event" - propData = "data" - propKind = "kind" - propKey = "key" - propVersion = "version" - propObject = "object" - propReason = "reason" - propState = "state" - propPayloads = "payloads" - propID = "id" - propTarget = "target" - propIntentCode = "intentCode" - propPayloadID = "payloadId" -) +// The two variants accept and reject the same payloads. The easyjson build decodes each scalar +// event with encoding/json into the subsystems types; the default build's jsonstream decoders are +// written to mirror that behavior exactly -- including rejecting non-integer version/target values +// on the plain-int types and matching Selector's float64-based (truncating) version handling. Both +// produce a parsedPutObject carrying the item's raw bytes and, for recognized kinds, its parsed +// form, so the ChangeSet's collections are assembled without a later re-parse. const errNoKnownPollingEvents = "didn't receive any known protocol events in polling payload" -// readInt reads a JSON number that must be an integer. jreader.Int coerces via int(Float64()), -// which silently truncates a fractional number (for example 1.9 becomes 1). The FDv2 protocol's -// version and target fields are integers, so a fractional value is a malformed payload and is -// rejected here, matching the reflection-based decode used by the launchdarkly_easyjson build. -func readInt(r *jreader.Reader) int { - f := r.Float64() - if f != math.Trunc(f) { - r.AddError(jreader.SyntaxError{Message: "expected integer, got fractional number"}) - return 0 - } - return int(f) -} - // parsedPutObject is the result of decoding a put-object event's data. type parsedPutObject struct { kind subsystems.ObjectKind @@ -82,153 +49,3 @@ func (p parsedPutObject) addTo(builder *subsystems.ChangeSetBuilder) { builder.AddPut(p.kind, p.key, p.version, p.object) } } - -// parseServerIntentEventData decodes the data of a server-intent event. The intent is required to -// have at least one payload (at index 0) at this time. -func parseServerIntentEventData(data []byte) (subsystems.ServerIntent, error) { - r := jreader.NewReader(data) - intent, err := readServerIntent(&r) - if err == nil { - err = r.Error() - } - return intent, err -} - -func readServerIntent(r *jreader.Reader) (subsystems.ServerIntent, error) { - var intent subsystems.ServerIntent - gotPayload := false - for obj := r.Object(); obj.Next(); { - if string(obj.Name()) != propPayloads { - _ = r.SkipValue() - continue - } - for arr := r.Array(); arr.Next(); { - // The protocol allows more than one payload, but SDKs currently only support one; - // any additional payloads are skipped. - if gotPayload { - _ = r.SkipValue() - continue - } - for payloadObj := r.Object(); payloadObj.Next(); { - switch string(payloadObj.Name()) { - case propID: - intent.Payload.ID = r.String() - case propTarget: - intent.Payload.Target = readInt(r) - case propIntentCode: - intent.Payload.Code = subsystems.IntentCode(r.String()) - case propReason: - intent.Payload.Reason = r.String() - default: - _ = r.SkipValue() - } - } - gotPayload = true - } - } - if err := r.Error(); err != nil { - return intent, err - } - if !gotPayload { - // It is a protocol error for the payload list to be missing or empty. - return intent, errors.New("changeset: server-intent event has no payloads") - } - return intent, nil -} - -// parseDeleteObjectEventData decodes the data of a delete-object event. -func parseDeleteObjectEventData(data []byte) (subsystems.DeleteObject, error) { - r := jreader.NewReader(data) - deleteObject, err := readDeleteObject(&r) - if err == nil { - err = r.Error() - } - return deleteObject, err -} - -func readDeleteObject(r *jreader.Reader) (subsystems.DeleteObject, error) { - var d subsystems.DeleteObject - for obj := r.Object(); obj.Next(); { - switch string(obj.Name()) { - case propKind: - d.Kind = subsystems.ObjectKind(r.String()) - case propKey: - d.Key = r.String() - case propVersion: - d.Version = readInt(r) - default: - _ = r.SkipValue() - } - } - return d, r.Error() -} - -// parseSelectorEventData decodes the data of a payload-transferred event. Both the state and the -// version are required. -func parseSelectorEventData(data []byte) (subsystems.Selector, error) { - r := jreader.NewReader(data) - selector, err := readSelector(&r) - if err == nil { - err = r.Error() - } - return selector, err -} - -func readSelector(r *jreader.Reader) (subsystems.Selector, error) { - var state string - var version int - gotState, gotVersion := false, false - for obj := r.Object(); obj.Next(); { - switch string(obj.Name()) { - case propState: - state = r.String() - gotState = true - case propVersion: - version = readInt(r) - gotVersion = true - default: - _ = r.SkipValue() - } - } - if err := r.Error(); err != nil { - return subsystems.NoSelector(), err - } - if !gotState { - return subsystems.NoSelector(), errors.New("unmarshal selector: missing state field") - } - if !gotVersion { - return subsystems.NoSelector(), errors.New("unmarshal selector: missing version field") - } - return subsystems.NewSelector(state, version), nil -} - -// parseGoodbyeEventData decodes the data of a goodbye event. -func parseGoodbyeEventData(data []byte) (subsystems.Goodbye, error) { - r := jreader.NewReader(data) - var goodbye subsystems.Goodbye - for obj := r.Object(); obj.Next(); { - if string(obj.Name()) == propReason { - goodbye.Reason = r.String() - } else { - _ = r.SkipValue() - } - } - return goodbye, r.Error() -} - -// parseErrorEventData decodes the data of an error event. -func parseErrorEventData(data []byte) (subsystems.Error, error) { - r := jreader.NewReader(data) - var errorData subsystems.Error - for obj := r.Object(); obj.Next(); { - switch string(obj.Name()) { - case propPayloadID: - errorData.PayloadID = r.String() - case propReason: - errorData.Reason = r.String() - default: - _ = r.SkipValue() - } - } - return errorData, r.Error() -} diff --git a/internal/datasourcev2/event_parsing_default.go b/internal/datasourcev2/event_parsing_default.go index ace336e1..8bbdbbe3 100644 --- a/internal/datasourcev2/event_parsing_default.go +++ b/internal/datasourcev2/event_parsing_default.go @@ -9,11 +9,55 @@ import ( "encoding/json" "errors" "fmt" + "strconv" "github.com/launchdarkly/go-jsonstream/v3/jreader" "github.com/launchdarkly/go-server-sdk/v7/subsystems" ) +// JSON property names used by the single-pass event decoders. +const ( + propEvents = "events" + propEvent = "event" + propData = "data" + propKind = "kind" + propKey = "key" + propVersion = "version" + propObject = "object" + propReason = "reason" + propState = "state" + propPayloads = "payloads" + propID = "id" + propTarget = "target" + propIntentCode = "intentCode" + propPayloadID = "payloadId" +) + +// readInt reads a JSON number that must fit in a Go int, with the same acceptance rules as decoding +// into an int via encoding/json (the reflection decode used by the launchdarkly_easyjson build for +// the plain-int fields), so the two build variants stay in agreement. jreader.Int cannot be used: +// it coerces via int(Float64()), which silently truncates a fractional number (1.9 -> 1), loses +// precision beyond 2^53, and is implementation-defined out of range. Parsing the raw number literal +// instead means a fractional value (1.9), exponent notation (1e2), or an out-of-range value is +// rejected, large integers up to the int range are preserved exactly, and a JSON null yields the +// zero value -- exactly as encoding/json does. +func readInt(r *jreader.Reader) int { + raw := r.RawValue() + if r.Error() != nil { + return 0 + } + if string(raw) == "null" { + // encoding/json leaves the zero value for a null; match that rather than erroring. + return 0 + } + n, err := strconv.ParseInt(string(raw), 10, 0) + if err != nil { + r.AddError(jreader.SyntaxError{Message: "expected integer"}) + return 0 + } + return int(n) +} + // This file contains the single-pass payload decoders used by the default build. Each event -- // and in particular each put-object's item JSON -- is scanned only once: scalars are read in // place, and a recognized item's model is decoded directly from the stream while its raw bytes @@ -212,3 +256,157 @@ func readPutObject(r *jreader.Reader, input []byte) (parsedPutObject, error) { // bytes, so a malformed object has failed the payload above regardless of kind. return p, nil } + +// parseServerIntentEventData decodes the data of a server-intent event. The intent is required to +// have at least one payload (at index 0) at this time, matching subsystems.ServerIntent's decode. +func parseServerIntentEventData(data []byte) (subsystems.ServerIntent, error) { + r := jreader.NewReader(data) + intent, err := readServerIntent(&r) + if err == nil { + err = r.Error() + } + return intent, err +} + +func readServerIntent(r *jreader.Reader) (subsystems.ServerIntent, error) { + var intent subsystems.ServerIntent + gotPayload := false + for obj := r.Object(); obj.Next(); { + if string(obj.Name()) != propPayloads { + _ = r.SkipValue() + continue + } + for arr := r.Array(); arr.Next(); { + // The protocol allows more than one payload, but SDKs currently only support one; + // any additional payloads are skipped. + if gotPayload { + _ = r.SkipValue() + continue + } + for payloadObj := r.Object(); payloadObj.Next(); { + switch string(payloadObj.Name()) { + case propID: + intent.Payload.ID = r.String() + case propTarget: + intent.Payload.Target = readInt(r) + case propIntentCode: + intent.Payload.Code = subsystems.IntentCode(r.String()) + case propReason: + intent.Payload.Reason = r.String() + default: + _ = r.SkipValue() + } + } + gotPayload = true + } + } + if err := r.Error(); err != nil { + return intent, err + } + if !gotPayload { + // It is a protocol error for the payload list to be missing or empty. + return intent, errors.New("changeset: server-intent event has no payloads") + } + return intent, nil +} + +// parseDeleteObjectEventData decodes the data of a delete-object event. +func parseDeleteObjectEventData(data []byte) (subsystems.DeleteObject, error) { + r := jreader.NewReader(data) + deleteObject, err := readDeleteObject(&r) + if err == nil { + err = r.Error() + } + return deleteObject, err +} + +func readDeleteObject(r *jreader.Reader) (subsystems.DeleteObject, error) { + var d subsystems.DeleteObject + for obj := r.Object(); obj.Next(); { + switch string(obj.Name()) { + case propKind: + d.Kind = subsystems.ObjectKind(r.String()) + case propKey: + d.Key = r.String() + case propVersion: + d.Version = readInt(r) + default: + _ = r.SkipValue() + } + } + return d, r.Error() +} + +// parseSelectorEventData decodes the data of a payload-transferred event. Both the state and the +// version are required. +func parseSelectorEventData(data []byte) (subsystems.Selector, error) { + r := jreader.NewReader(data) + selector, err := readSelector(&r) + if err == nil { + err = r.Error() + } + return selector, err +} + +func readSelector(r *jreader.Reader) (subsystems.Selector, error) { + var state string + var version int + gotState, gotVersion := false, false + for obj := r.Object(); obj.Next(); { + switch string(obj.Name()) { + case propState: + state = r.String() + gotState = true + case propVersion: + // subsystems.Selector's decode (the reflection reference for this event) reads version + // through float64 and truncates, so use the lenient jreader.Int here rather than the + // strict readInt used for the other integer fields, keeping the two builds in + // agreement. The value is deprecated and not used by consumers. + version = r.Int() + gotVersion = true + default: + _ = r.SkipValue() + } + } + if err := r.Error(); err != nil { + return subsystems.NoSelector(), err + } + if !gotState { + return subsystems.NoSelector(), errors.New("unmarshal selector: missing state field") + } + if !gotVersion { + return subsystems.NoSelector(), errors.New("unmarshal selector: missing version field") + } + return subsystems.NewSelector(state, version), nil +} + +// parseGoodbyeEventData decodes the data of a goodbye event. +func parseGoodbyeEventData(data []byte) (subsystems.Goodbye, error) { + r := jreader.NewReader(data) + var goodbye subsystems.Goodbye + for obj := r.Object(); obj.Next(); { + if string(obj.Name()) == propReason { + goodbye.Reason = r.String() + } else { + _ = r.SkipValue() + } + } + return goodbye, r.Error() +} + +// parseErrorEventData decodes the data of an error event. +func parseErrorEventData(data []byte) (subsystems.Error, error) { + r := jreader.NewReader(data) + var errorData subsystems.Error + for obj := r.Object(); obj.Next(); { + switch string(obj.Name()) { + case propPayloadID: + errorData.PayloadID = r.String() + case propReason: + errorData.Reason = r.String() + default: + _ = r.SkipValue() + } + } + return errorData, r.Error() +} diff --git a/internal/datasourcev2/event_parsing_easyjson.go b/internal/datasourcev2/event_parsing_easyjson.go index 333303a0..e36643d3 100644 --- a/internal/datasourcev2/event_parsing_easyjson.go +++ b/internal/datasourcev2/event_parsing_easyjson.go @@ -93,3 +93,42 @@ func parsePutObjectEventData(data []byte) (parsedPutObject, error) { p.item, p.hasItem = item, true return p, nil } + +// The scalar event decoders below use encoding/json into the subsystems types. This is the +// reflection decode the default build's jsonstream decoders are written to match; keeping it here +// means the launchdarkly_easyjson build depends only on encoding/json for these events. + +// parseServerIntentEventData decodes the data of a server-intent event. +func parseServerIntentEventData(data []byte) (subsystems.ServerIntent, error) { + var intent subsystems.ServerIntent + err := json.Unmarshal(data, &intent) + return intent, err +} + +// parseDeleteObjectEventData decodes the data of a delete-object event. +func parseDeleteObjectEventData(data []byte) (subsystems.DeleteObject, error) { + var deleteObject subsystems.DeleteObject + err := json.Unmarshal(data, &deleteObject) + return deleteObject, err +} + +// parseSelectorEventData decodes the data of a payload-transferred event. +func parseSelectorEventData(data []byte) (subsystems.Selector, error) { + var selector subsystems.Selector + err := json.Unmarshal(data, &selector) + return selector, err +} + +// parseGoodbyeEventData decodes the data of a goodbye event. +func parseGoodbyeEventData(data []byte) (subsystems.Goodbye, error) { + var goodbye subsystems.Goodbye + err := json.Unmarshal(data, &goodbye) + return goodbye, err +} + +// parseErrorEventData decodes the data of an error event. +func parseErrorEventData(data []byte) (subsystems.Error, error) { + var errorData subsystems.Error + err := json.Unmarshal(data, &errorData) + return errorData, err +} diff --git a/internal/datasourcev2/event_parsing_test.go b/internal/datasourcev2/event_parsing_test.go index 12134c98..0cc601e6 100644 --- a/internal/datasourcev2/event_parsing_test.go +++ b/internal/datasourcev2/event_parsing_test.go @@ -417,38 +417,76 @@ func TestParsePollingPayloadMatchesReflectionReference(t *testing.T) { assert.Equal(t, wantItems, gotItems) } -// A protocol integer field (version, target) must reject a fractional JSON number rather than -// silently truncating it. jreader.Int coerces via int(Float64()) -- so 1.9 would become 1 -- and -// the single-pass decoder relies on readInt to reject it, matching the reflection-based decode -// that the launchdarkly_easyjson build (and the pre-single-pass code) used. Both build variants -// run this test. -func TestFractionalIntegerProtocolFieldsAreRejected(t *testing.T) { +// The scalar event decoders must accept and reject an integer field exactly as decoding the +// corresponding subsystems type via encoding/json does -- encoding/json is what the +// launchdarkly_easyjson build uses, so this is the parity contract between the two build variants. +// The oracle is computed live from encoding/json into the same struct that build decodes into, so +// per-type behavior is captured automatically: PutObject, DeleteObject, and Payload have plain int +// fields and reject non-integers, while Selector reads its version through float64 and truncates. +// Both build variants run this test. +func TestScalarEventIntegerParsingMatchesEncodingJSON(t *testing.T) { flagJSON := makeTestFlagJSON(t, "flagkey", 1) - cases := map[string]func() error{ - "put-object version": func() error { - _, err := parsePutObjectEventData([]byte(fmt.Sprintf( - `{"kind":"flag","key":"flagkey","version":1.9,"object":%s}`, flagJSON))) - return err - }, - "delete-object version": func() error { - _, err := parseDeleteObjectEventData([]byte(`{"kind":"flag","key":"k","version":1.9}`)) - return err - }, - "selector version": func() error { - _, err := parseSelectorEventData([]byte(`{"state":"s","version":1.9}`)) - return err - }, - "server-intent target": func() error { - _, err := parseServerIntentEventData([]byte( - `{"payloads":[{"id":"p1","target":1.9,"intentCode":"xfer-full","reason":"r"}]}`)) - return err - }, - } - for name, run := range cases { - t.Run(name, func(t *testing.T) { - assert.Error(t, run()) - }) + values := []string{ + "0", "1", "-1", "100", "2147483647", "-2147483648", + "9007199254740993", // > 2^53, must round-trip exactly for the strict fields + "9223372036854775807", // MaxInt64 + "9223372036854775808", // MaxInt64 + 1 (out of range) + "1e20", "-1e20", + "1.9", "1.0", "1e2", // non-integer literals + "null", } + + t.Run("put-object version", func(t *testing.T) { + for _, v := range values { + data := []byte(fmt.Sprintf(`{"kind":"flag","key":"k","version":%s,"object":%s}`, v, flagJSON)) + var oracle subsystems.PutObject + wantErr := json.Unmarshal(data, &oracle) != nil + p, err := parsePutObjectEventData(data) + require.Equalf(t, wantErr, err != nil, "version=%s: encoding/json err=%v, decoder err=%v", v, wantErr, err) + if !wantErr { + assert.Equalf(t, oracle.Version, p.version, "version=%s", v) + } + } + }) + + t.Run("delete-object version", func(t *testing.T) { + for _, v := range values { + data := []byte(fmt.Sprintf(`{"kind":"flag","key":"k","version":%s}`, v)) + var oracle subsystems.DeleteObject + wantErr := json.Unmarshal(data, &oracle) != nil + d, err := parseDeleteObjectEventData(data) + require.Equalf(t, wantErr, err != nil, "version=%s: encoding/json err=%v, decoder err=%v", v, wantErr, err) + if !wantErr { + assert.Equalf(t, oracle.Version, d.Version, "version=%s", v) + } + } + }) + + t.Run("server-intent target", func(t *testing.T) { + for _, v := range values { + data := []byte(fmt.Sprintf(`{"payloads":[{"id":"p1","target":%s,"intentCode":"xfer-full","reason":"r"}]}`, v)) + var oracle subsystems.ServerIntent + wantErr := json.Unmarshal(data, &oracle) != nil + intent, err := parseServerIntentEventData(data) + require.Equalf(t, wantErr, err != nil, "target=%s: encoding/json err=%v, decoder err=%v", v, wantErr, err) + if !wantErr { + assert.Equalf(t, oracle.Payload.Target, intent.Payload.Target, "target=%s", v) + } + } + }) + + t.Run("selector version", func(t *testing.T) { + for _, v := range values { + data := []byte(fmt.Sprintf(`{"state":"s","version":%s}`, v)) + var oracle subsystems.Selector + wantErr := json.Unmarshal(data, &oracle) != nil + sel, err := parseSelectorEventData(data) + require.Equalf(t, wantErr, err != nil, "version=%s: encoding/json err=%v, decoder err=%v", v, wantErr, err) + if !wantErr { + assert.Equalf(t, oracle.Version(), sel.Version(), "version=%s", v) + } + } + }) } // A polling event that carries the "data" property more than once (non-conformant, but handled for From 8e3d4ed246786c3121041390c5943a3628a05853 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Thu, 30 Jul 2026 12:58:29 -0400 Subject: [PATCH 6/7] fix: Prevent duplicate-kind type confusion in put-object decoding The default build model-decodes a put-object's item as soon as the object is reached, under whatever kind has been seen so far. A later "kind" property (non-conformant, but json resolves it last-wins) overrode p.kind without re-parsing, leaving the item typed as the earlier kind -- e.g. a *FeatureFlag filed under the Segments collection, which serializes as an empty value into a persistent store. Track the kind the item was decoded under and, after the object is fully read, re-parse under the final kind if it changed, or drop the item and keep the object raw if the final kind is unrecognized. This matches the easyjson build, which decodes under the final kind. --- .../datasourcev2/event_parsing_default.go | 41 +++++++++++++------ internal/datasourcev2/event_parsing_test.go | 26 ++++++++++++ 2 files changed, 54 insertions(+), 13 deletions(-) diff --git a/internal/datasourcev2/event_parsing_default.go b/internal/datasourcev2/event_parsing_default.go index 8bbdbbe3..fb1353e9 100644 --- a/internal/datasourcev2/event_parsing_default.go +++ b/internal/datasourcev2/event_parsing_default.go @@ -203,6 +203,11 @@ const jsonWhitespace = " \t\r\n" func readPutObject(r *jreader.Reader, input []byte) (parsedPutObject, error) { var p parsedPutObject var object json.RawMessage + // decodedKind records the kind under which the item was eagerly model-decoded while reading + // the object, so that a "kind" property appearing after "object" (non-conformant, but json's + // last-wins behavior handles it) can be detected and the item re-parsed under the final kind + // rather than left mis-typed. + var decodedKind subsystems.ObjectKind for obj := r.Object(); obj.Next(); { switch string(obj.Name()) { case propKind: @@ -224,7 +229,7 @@ func readPutObject(r *jreader.Reader, input []byte) (parsedPutObject, error) { } span := bytes.Trim(input[start:r.Offset()], jsonWhitespace) object = json.RawMessage(span[:len(span):len(span)]) - p.item, p.hasItem = item, true + p.item, p.hasItem, decodedKind = item, true, p.kind } else { // The kind is unrecognized or has not been read yet, so the value cannot be // model-decoded here; RawValue captures it with full validation, and if the @@ -239,21 +244,31 @@ func readPutObject(r *jreader.Reader, input []byte) (parsedPutObject, error) { return p, err } p.object = object - if p.hasItem || object == nil { + if object == nil { return p, nil } - if kind, recognized := p.kind.ToFDV1(); recognized { - // The object appeared before the kind; parse it now. - itemReader := jreader.NewReader(object) - item, err := kind.DeserializeFromJSONReader(&itemReader) - if err != nil { - return p, err - } - p.item, p.hasItem = item, true + kind, recognized := p.kind.ToFDV1() + if !recognized { + // The final kind is unrecognized -- either it was never a known kind, or a later "kind" + // property overrode an earlier recognized one. Keep the object raw and unparsed for + // forwards compatibility, discarding any item eagerly decoded under the earlier kind. The + // bytes were already validated (by the eager decode or by RawValue). + p.hasItem = false + return p, nil + } + if p.hasItem && decodedKind == p.kind { + // The item was already decoded under the final kind while reading the object. + return p, nil + } + // The object was captured before its kind was known, or a later "kind" changed the type after + // the object was decoded; parse the captured bytes under the final kind so that the stored item + // and kind always agree. + itemReader := jreader.NewReader(object) + item, err := kind.DeserializeFromJSONReader(&itemReader) + if err != nil { + return p, err } - // If the kind is unrecognized, the object is kept raw (and unparsed) for forwards - // compatibility. No further validation is needed: RawValue already fully validated the - // bytes, so a malformed object has failed the payload above regardless of kind. + p.item, p.hasItem = item, true return p, nil } diff --git a/internal/datasourcev2/event_parsing_test.go b/internal/datasourcev2/event_parsing_test.go index 0cc601e6..a62b829a 100644 --- a/internal/datasourcev2/event_parsing_test.go +++ b/internal/datasourcev2/event_parsing_test.go @@ -527,3 +527,29 @@ func flattenCollectionsForComparison(collections []ldstoretypes.Collection) map[ } return result } + +// A duplicated "kind" property (non-conformant, but json resolves it last-wins) must not leave the +// parsed item typed as the earlier kind. The item and the stored kind must always agree: if the +// final kind is recognized the item is (re-)parsed under it, and if the final kind is unrecognized +// the object is kept raw with no item. Both build variants run this test. +func TestParsePutObjectEventDataDuplicateKind(t *testing.T) { + flagJSON := makeTestFlagJSON(t, "dup", 5) + + t.Run("later kind wins and item is re-typed", func(t *testing.T) { + data := []byte(fmt.Sprintf(`{"kind":"flag","object":%s,"key":"dup","version":5,"kind":"segment"}`, flagJSON)) + p, err := parsePutObjectEventData(data) + require.NoError(t, err) + assert.Equal(t, subsystems.SegmentKind, p.kind) + require.True(t, p.hasItem) + assert.IsType(t, &ldmodel.Segment{}, p.item.Item) + }) + + t.Run("later unrecognized kind keeps object raw", func(t *testing.T) { + data := []byte(fmt.Sprintf(`{"kind":"flag","object":%s,"key":"dup","version":5,"kind":"future-kind"}`, flagJSON)) + p, err := parsePutObjectEventData(data) + require.NoError(t, err) + assert.Equal(t, subsystems.ObjectKind("future-kind"), p.kind) + assert.False(t, p.hasItem) + assert.JSONEq(t, string(flagJSON), string(p.object)) + }) +} From b6fe049435b88b073f07f2830d49bef6ecadb0c9 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Thu, 30 Jul 2026 13:06:57 -0400 Subject: [PATCH 7/7] fix: Bound nesting depth when skipping unknown FDv2 fields jreader.SkipValue recurses one stack frame per level of nesting with no depth limit, so a deeply nested value in an unknown field of an untrusted payload could overflow the stack and crash the process -- a regression from the previous encoding/json decode, which capped nesting at 10000. Skip unknown values with RawValue instead: it scans the value iteratively and validates it with encoding/json, whose nesting cap turns an over-deep value into an error rather than a crash. The easyjson build already went through encoding/json and is unaffected. --- .../datasourcev2/event_parsing_default.go | 30 ++++++++++++------- internal/datasourcev2/event_parsing_test.go | 27 +++++++++++++++++ 2 files changed, 47 insertions(+), 10 deletions(-) diff --git a/internal/datasourcev2/event_parsing_default.go b/internal/datasourcev2/event_parsing_default.go index fb1353e9..b6c505ff 100644 --- a/internal/datasourcev2/event_parsing_default.go +++ b/internal/datasourcev2/event_parsing_default.go @@ -58,6 +58,16 @@ func readInt(r *jreader.Reader) int { return int(n) } +// skipValue discards the next JSON value. It deliberately uses RawValue rather than +// jreader.SkipValue: SkipValue recurses one stack frame per level of nesting with no depth limit, +// so a deeply nested value in an unknown field of untrusted input could overflow the stack and +// crash the process. RawValue scans the value iteratively and validates it with encoding/json, +// whose nesting is capped -- turning an over-deep value into an error rather than a crash, matching +// the encoding/json decode the SDK used before this parser. +func skipValue(r *jreader.Reader) { + _ = r.RawValue() +} + // This file contains the single-pass payload decoders used by the default build. Each event -- // and in particular each put-object's item JSON -- is scanned only once: scalars are read in // place, and a recognized item's model is decoded directly from the stream while its raw bytes @@ -80,7 +90,7 @@ func parsePollingPayload(ctx context.Context, body []byte) (*subsystems.ChangeSe r := jreader.NewReader(body) for topObj := r.Object(); topObj.Next(); { if string(topObj.Name()) != propEvents { - _ = r.SkipValue() + skipValue(&r) continue } for arr := r.Array(); arr.Next(); { @@ -126,7 +136,7 @@ func readPollingEvent( gotData = true data = r.RawValue() default: - _ = r.SkipValue() + skipValue(r) } } if err := r.Error(); err != nil { @@ -237,7 +247,7 @@ func readPutObject(r *jreader.Reader, input []byte) (parsedPutObject, error) { object = json.RawMessage(r.RawValue()) } default: - _ = r.SkipValue() + skipValue(r) } } if err := r.Error(); err != nil { @@ -288,14 +298,14 @@ func readServerIntent(r *jreader.Reader) (subsystems.ServerIntent, error) { gotPayload := false for obj := r.Object(); obj.Next(); { if string(obj.Name()) != propPayloads { - _ = r.SkipValue() + skipValue(r) continue } for arr := r.Array(); arr.Next(); { // The protocol allows more than one payload, but SDKs currently only support one; // any additional payloads are skipped. if gotPayload { - _ = r.SkipValue() + skipValue(r) continue } for payloadObj := r.Object(); payloadObj.Next(); { @@ -309,7 +319,7 @@ func readServerIntent(r *jreader.Reader) (subsystems.ServerIntent, error) { case propReason: intent.Payload.Reason = r.String() default: - _ = r.SkipValue() + skipValue(r) } } gotPayload = true @@ -346,7 +356,7 @@ func readDeleteObject(r *jreader.Reader) (subsystems.DeleteObject, error) { case propVersion: d.Version = readInt(r) default: - _ = r.SkipValue() + skipValue(r) } } return d, r.Error() @@ -380,7 +390,7 @@ func readSelector(r *jreader.Reader) (subsystems.Selector, error) { version = r.Int() gotVersion = true default: - _ = r.SkipValue() + skipValue(r) } } if err := r.Error(); err != nil { @@ -403,7 +413,7 @@ func parseGoodbyeEventData(data []byte) (subsystems.Goodbye, error) { if string(obj.Name()) == propReason { goodbye.Reason = r.String() } else { - _ = r.SkipValue() + skipValue(&r) } } return goodbye, r.Error() @@ -420,7 +430,7 @@ func parseErrorEventData(data []byte) (subsystems.Error, error) { case propReason: errorData.Reason = r.String() default: - _ = r.SkipValue() + skipValue(&r) } } return errorData, r.Error() diff --git a/internal/datasourcev2/event_parsing_test.go b/internal/datasourcev2/event_parsing_test.go index a62b829a..01e3f212 100644 --- a/internal/datasourcev2/event_parsing_test.go +++ b/internal/datasourcev2/event_parsing_test.go @@ -553,3 +553,30 @@ func TestParsePutObjectEventDataDuplicateKind(t *testing.T) { assert.JSONEq(t, string(flagJSON), string(p.object)) }) } + +// A deeply nested value in an unknown/skipped field must be rejected as an error rather than +// recursing until the stack overflows (a remote-crash DoS from untrusted input). The value exceeds +// encoding/json's nesting cap; both build variants must return an error, not crash. Both build +// variants run this test. +func TestDeeplyNestedSkippedValueIsRejected(t *testing.T) { + const depth = 100000 + nested := strings.Repeat("[", depth) + strings.Repeat("]", depth) + + t.Run("polling top-level unknown property", func(t *testing.T) { + body := []byte(`{"junk":` + nested + `,"events":[]}`) + _, err := parsePollingPayload(context.Background(), body) + assert.Error(t, err) + }) + + t.Run("put-object unknown property", func(t *testing.T) { + data := []byte(`{"kind":"flag","junk":` + nested + `,"key":"k","version":1}`) + _, err := parsePutObjectEventData(data) + assert.Error(t, err) + }) + + t.Run("server-intent unknown property", func(t *testing.T) { + data := []byte(`{"junk":` + nested + `,"payloads":[{"id":"p1","intentCode":"xfer-full"}]}`) + _, err := parseServerIntentEventData(data) + assert.Error(t, err) + }) +}