Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: Expand error handling to not require an explicit server intent #272

Merged
merged 2 commits into from
Dec 20, 2024
Merged

fix: Expand error handling to not require an explicit server intent #272

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e72d534
fix: Expand error handling to not require an explicit server intent
keelerm84 Dec 20, 2024
ba09b67
Add additional multiple-update test
keelerm84 Dec 20, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 30 additions & 0 deletions mockld/streaming_service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,6 +164,36 @@ func (s *StreamingService) makeXferFull() []eventsource.Event {
return events
}

func (s *StreamingService) PushServerIntent(intentCode, intentReason string) {
event := eventImpl{
name: "server-intent",
data: framework.ServerIntent{
Payloads: []framework.Payload{
{
ID: "payloadID",
Target: 1,
Code: intentCode,
Reason: intentReason,
},
},
},
}

s.lock.Lock()
alreadyStarted := s.started
if !alreadyStarted {
s.queuedEvents = append(s.queuedEvents, event)
}
s.lock.Unlock()

if alreadyStarted {
s.logEvent(event)
s.streams.Publish([]string{allDataChannel}, event)
} else {
s.debugLogger.Println("Will send server-intent event after connection has started")
}
}

// PushEvent sends an SSE event to all clients that are currently connected to the stream-- or, if no client
// has connected yet, queues the event so that it will be sent (after the initial data) to the
// first client that connects. (The latter is necessary to avoid race conditions, since even after
Expand Down
77 changes: 65 additions & 12 deletions sdktests/common_tests_stream_fdv2.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
var (
initialValue = ldvalue.String("initial value") //nolint:gochecknoglobals
updatedValue = ldvalue.String("updated value") //nolint:gochecknoglobals
finalValue = ldvalue.String("final value") //nolint:gochecknoglobals

newInitialValue = ldvalue.String("new initial value") //nolint:gochecknoglobals

Expand All @@ -31,9 +32,11 @@ func (c CommonStreamingTests) FDv2(t *ldtest.T) {
t.Run(
"updates are not complete until payload transferred is sent",
c.UpdatesAreNotCompleteUntilPayloadTransferredIsSent)
t.Run("handles multiple updates", c.HandlesMultipleUpdates)
t.Run("ignores model version", c.IgnoresModelVersion)
t.Run("ignores heart beat", c.IgnoresHeartBeat)
t.Run("discards events on errors", c.DiscardsEventsOnError)
t.Run("can discard partial events on errors", c.CanDiscardPartialEventsOnError)
t.Run("can discard full events on errors", c.CanDiscardFullEventsOnError)
t.Run("disconnects on goodbye", c.DisconnectsOnGoodbye)
}

Expand Down Expand Up @@ -213,6 +216,32 @@ func (c CommonStreamingTests) UpdatesAreNotCompleteUntilPayloadTransferredIsSent
pollUntilFlagValueUpdated(t, client, "new-flag-key", context, defaultValue, newInitialValue, defaultValue)
}

func (c CommonStreamingTests) HandlesMultipleUpdates(t *ldtest.T) {
dataSystem, configurers := c.setupDataSystems(t, c.makeSDKDataWithFlag(1, initialValue))
client := NewSDKClient(t, c.baseSDKConfigurationPlus(configurers...)...)
context := ldcontext.New("context-key")

expectedEvaluations := map[string]ldvalue.Value{"flag-key": initialValue}
validatePayloadReceived(t, dataSystem.PrimarySync().Endpoint(), client, "", expectedEvaluations)

dataSystem.PrimarySync().streaming.PushUpdate(
"flag", "flag-key", 2, c.makeFlagData("flag-key", 2, updatedValue))
dataSystem.PrimarySync().streaming.PushPayloadTransferred("updated", 2)
pollUntilFlagValueUpdated(t, client, "flag-key", context, initialValue, updatedValue, defaultValue)

dataSystem.PrimarySync().streaming.PushUpdate(
"flag", "new-flag-key", 2, c.makeFlagData("new-flag-key", 2, newInitialValue))
dataSystem.PrimarySync().streaming.PushPayloadTransferred("updated", 3)
pollUntilFlagValueUpdated(t, client, "new-flag-key", context, defaultValue, newInitialValue, defaultValue)

dataSystem.PrimarySync().streaming.PushServerIntent("xfer-full", "stale")
dataSystem.PrimarySync().streaming.PushUpdate(
"flag", "flag-key", 3, c.makeFlagData("flag-key", 3, finalValue))
dataSystem.PrimarySync().streaming.PushPayloadTransferred("updated", 4)
pollUntilFlagValueUpdated(t, client, "flag-key", context, updatedValue, finalValue, defaultValue)
pollUntilFlagValueUpdated(t, client, "new-flag-key", context, newInitialValue, defaultValue, defaultValue)
}

func (c CommonStreamingTests) IgnoresModelVersion(t *ldtest.T) {
dataSystem, configurers := c.setupDataSystems(t, c.makeSDKDataWithFlag(100, initialValue))
client := NewSDKClient(t, c.baseSDKConfigurationPlus(configurers...)...)
Expand Down Expand Up @@ -252,16 +281,12 @@ func (c CommonStreamingTests) IgnoresHeartBeat(t *ldtest.T) {
pollUntilFlagValueUpdated(t, client, "flag-key", context, initialValue, updatedValue, defaultValue)
}

func (c CommonStreamingTests) DiscardsEventsOnError(t *ldtest.T) {
func (c CommonStreamingTests) CanDiscardPartialEventsOnError(t *ldtest.T) {
dataSystem, configurers := c.setupDataSystems(t, c.makeSDKDataWithFlag(1, initialValue))
client := NewSDKClient(t, c.baseSDKConfigurationPlus(configurers...)...)

_, err := dataSystem.PrimarySync().endpoint.AwaitConnection(time.Second)
require.NoError(t, err)

context := ldcontext.New("context-key")
flagKeyValue := basicEvaluateFlag(t, client, "flag-key", context, defaultValue)
m.In(t).Assert(flagKeyValue, m.JSONEqual(initialValue))
expectedEvaluations := map[string]ldvalue.Value{"flag-key": initialValue}
validatePayloadReceived(t, dataSystem.PrimarySync().Endpoint(), client, "", expectedEvaluations)

// The error should cause this update to be discard.
dataSystem.PrimarySync().streaming.PushUpdate(
Expand All @@ -272,6 +297,7 @@ func (c CommonStreamingTests) DiscardsEventsOnError(t *ldtest.T) {
"flag", "new-flag-key", 2, c.makeFlagData("new-flag-key", 2, newInitialValue))
dataSystem.PrimarySync().streaming.PushPayloadTransferred("updated", 2)

context := ldcontext.New("context-key")
require.Never(
t,
checkForUpdatedValue(t, client, "flag-key", context, initialValue, updatedValue, defaultValue),
Expand All @@ -281,6 +307,34 @@ func (c CommonStreamingTests) DiscardsEventsOnError(t *ldtest.T) {
)

pollUntilFlagValueUpdated(t, client, "new-flag-key", context, defaultValue, newInitialValue, defaultValue)

// Original flag value should still be the same.
value := basicEvaluateFlag(t, client, "flag-key", context, defaultValue)
require.Equal(t, initialValue, value)
}

func (c CommonStreamingTests) CanDiscardFullEventsOnError(t *ldtest.T) {
dataSystem, configurers := c.setupDataSystems(t, c.makeSDKDataWithFlag(1, initialValue))
client := NewSDKClient(t, c.baseSDKConfigurationPlus(configurers...)...)

expectedEvaluations := map[string]ldvalue.Value{"flag-key": initialValue}
validatePayloadReceived(t, dataSystem.PrimarySync().Endpoint(), client, "", expectedEvaluations)

// The error should cause this update to be discard.
dataSystem.PrimarySync().streaming.PushUpdate(
"flag", "flag-key", 2, c.makeFlagData("flag-key", 2, updatedValue))
dataSystem.PrimarySync().streaming.PushError("some-id", "some reason")
// But this change should be applied.
dataSystem.PrimarySync().streaming.PushServerIntent("xfer-full", "stale")
dataSystem.PrimarySync().streaming.PushUpdate(
"flag", "new-flag-key", 2, c.makeFlagData("new-flag-key", 2, newInitialValue))
dataSystem.PrimarySync().streaming.PushPayloadTransferred("updated", 2)

context := ldcontext.New("context-key")

// Previous flag should be removed, reverting to a default value being served.
pollUntilFlagValueUpdated(t, client, "flag-key", context, initialValue, defaultValue, defaultValue)
pollUntilFlagValueUpdated(t, client, "new-flag-key", context, defaultValue, newInitialValue, defaultValue)
}

func (c CommonStreamingTests) DisconnectsOnGoodbye(t *ldtest.T) {
Expand All @@ -290,16 +344,15 @@ func (c CommonStreamingTests) DisconnectsOnGoodbye(t *ldtest.T) {
t.Defer(streamEndpoint.Close)
client := NewSDKClient(t, WithPrimaryStreamingSynchronizer(baseStreamConfig(streamEndpoint)))

_, err := streamEndpoint.AwaitConnection(time.Second)
require.NoError(t, err)
conn := streamEndpoint.RequireConnection(t, time.Second)

dataSystems[0].PrimarySync().streaming.PushUpdate(
"flag", "flag-key", 2, c.makeFlagData("flag-key", 2, updatedValue))
// This should prompt the SDK to discard previous events, disconnect, and then re-connect.
dataSystems[0].PrimarySync().streaming.PushGoodbye("some-reason", false, false)
conn.Cancel()

_, err = streamEndpoint.AwaitConnection(time.Second)
require.NoError(t, err)
_ = streamEndpoint.RequireConnection(t, time.Second)

context := ldcontext.New("context-key")
require.Never(
Expand Down
Loading