fix: make SetProviderAndWait behavior consistent with SetProvider

OpenFeature specification defines SetProviderAndWait as a waiting
version of SetProvider, or as a shortcut for waiting on provider ready
event. However, currently SetProvider and SetProviderAndWait exhibit
non-trivial behavior differences besides waiting.

SetProvider runs initialization asynchronously and potentially
concurrently with shutdown of the old provider. The API is not blocked
and the application author may initialize other providers
concurrently, run evaluations, etc.

🐛: in this mode, the error from initializer is ignored when updating
the provider state, so fatal/error states may be not set properly.

SetProviderAndWait runs initialization synchronously while holding
exclusive api.mu lock. This almost completely locks OpenFeature SDK:
the application author cannot initialize other providers (for
different domains), configure context or hooks, evaluate feature
flags, or shutdown SDK. If a provider initialization blocks
forever'ish, the SDK remains unusable and is unrecoverable.

Another difference is that old provider is shutdown only after new
provider has successfully initialized.

🐛: if the new provider fails to initialize, the old provider is
already unset in API but will never be shutdown, and the new provider
is not registered with api.eventExecutor (so if it comes back online
after some time, nobody listens to its events, and the state will go
out of sync if old provider continues emitting events).

🐛: in both modes, given that shutdown is run concurrently with
updating subscriptions in eventExecutor, it is possible for the old
provider to override the state of the new provider:
1. init finishes, emits provider ready event (directly from
   goroutine), updates state
2. old provider emits some event during shutdown (e.g., PROVIDER_ERROR
   or PROVIDER_STALE), eventExecutor receives the event and updates
   the state to error/stale
3. new provider is registered with eventExecutor but the state is
   already wrong.

This PR introduces a couple of changes:

Make initialization flow consistent across both modes: always
initialize async but make "AndWait" methods wait for initialization
outside of critical section. Make init respect returned error.

Always call shutdown on old provider (if it is no longer used).

Always register new provider with event executor. Do this before we
start init/shutdown, so the old provider cannot influence state of the
new provider. Make event executor registration non-erroring by making
shutdown channel buffered (there were no good way to recover from
registration error).

Signed-off-by: Oleksii Shmalko <oleksii.shmalko@datadoghq.com>

Author: dd-oleksii

openfeature/event_executor.go

@@ -1,12 +1,10 @@
 package openfeature
 
 import (
-	"fmt"
 	"log/slog"
 	"maps"
 	"slices"
 	"sync"
-	"time"
 )
 
 const defaultDomain = ""
@@ -192,32 +190,32 @@ func (e *eventExecutor) State(domain string) State {
 }
 
 // registerDefaultProvider registers the default FeatureProvider and remove the old default provider if available
-func (e *eventExecutor) registerDefaultProvider(provider FeatureProvider) error {
+func (e *eventExecutor) registerDefaultProvider(provider FeatureProvider) {
 	e.mu.Lock()
 	defer e.mu.Unlock()
 
 	newProvider := newProviderRef(provider)
 	oldProvider := e.defaultProviderReference
 	e.defaultProviderReference = newProvider
 
-	return e.startListeningAndShutdownOld(newProvider, oldProvider)
+	e.startListeningAndShutdownOld(newProvider, oldProvider)
 }
 
 // registerNamedEventingProvider registers a named FeatureProvider and remove event listener for old named provider
-func (e *eventExecutor) registerNamedEventingProvider(associatedClient string, provider FeatureProvider) error {
+func (e *eventExecutor) registerNamedEventingProvider(associatedClient string, provider FeatureProvider) {
 	e.mu.Lock()
 	defer e.mu.Unlock()
 	newProvider := newProviderRef(provider)
 
 	oldProvider := e.namedProviderReference[associatedClient]
 	e.namedProviderReference[associatedClient] = newProvider
 
-	return e.startListeningAndShutdownOld(newProvider, oldProvider)
+	e.startListeningAndShutdownOld(newProvider, oldProvider)
 }
 
 // startListeningAndShutdownOld is a helper to start concurrent listening to new provider events and  invoke shutdown
 // hook of the old provider if it's not bound by another subscription
-func (e *eventExecutor) startListeningAndShutdownOld(newProvider providerReference, oldReference providerReference) error {
+func (e *eventExecutor) startListeningAndShutdownOld(newProvider providerReference, oldReference providerReference) {
 	// check if this provider already actively handled - 1:N binding capability
 	if !isRunning(newProvider, e.activeSubscriptions) {
 		e.activeSubscriptions = append(e.activeSubscriptions, newProvider)
@@ -247,7 +245,7 @@ func (e *eventExecutor) startListeningAndShutdownOld(newProvider providerReferen
 
 	// check if this provider is still bound - 1:N binding capability
 	if isBound(oldReference, e.defaultProviderReference, slices.Collect(maps.Values(e.namedProviderReference))) {
-		return nil
+		return
 	}
 
 	// drop from active references
@@ -258,16 +256,28 @@ func (e *eventExecutor) startListeningAndShutdownOld(newProvider providerReferen
 	_, ok := oldReference.featureProvider.(EventHandler)
 	if !ok {
 		// no shutdown for non event handling provider
-		return nil
+		return
 	}
 
 	// avoid shutdown lockouts
 	select {
 	case oldReference.shutdownSemaphore <- "":
-		return nil
-	case <-time.After(200 * time.Millisecond):
-		return fmt.Errorf("old event handler %s timeout waiting for handler shutdown",
-			oldReference.featureProvider.Metadata().Name)
+	default:
+		// This should never happen:
+		//
+		// providerReference.shutdownSemaphore is created with
+		// a buffer size of 1, so it should allow sending at
+		// least one shutdown signal without blocking. Locking
+		// should prevent us from sending more than one
+		// signal.
+		//
+		// In the unlikely case that it does not, this
+		// shouldn't be a big deal: we have already swapped
+		// references in eventExecutor and openfeatureAPI, and
+		// the handler should be able to receive at least one
+		// shutdown signal later.
+		slog.Info("OF BUG: failed to send shutdown to old event handler",
+			"provider", oldReference.featureProvider.Metadata().Name)
 	}
 }
 

openfeature/event_executor_test.go

@@ -30,19 +30,13 @@ func TestEventHandler_RegisterUnregisterEventProvider(t *testing.T) {
 		}
 
 		executor := newEventExecutor()
-		err := executor.registerDefaultProvider(eventingProvider)
-		if err != nil {
-			t.Fatal(err)
-		}
+		executor.registerDefaultProvider(eventingProvider)
 
 		if executor.defaultProviderReference.featureProvider != eventingProvider {
 			t.Error("implementation should register default eventing provider")
 		}
 
-		err = executor.registerNamedEventingProvider("domain", eventingProvider)
-		if err != nil {
-			t.Fatal(err)
-		}
+		executor.registerNamedEventingProvider("domain", eventingProvider)
 
 		if _, ok := executor.namedProviderReference["domain"]; !ok {
 			t.Errorf("implementation should register named eventing provider")
@@ -966,7 +960,6 @@ func TestEventHandler_HandlersRunImmediately(t *testing.T) {
 		eventingImpl := &ProviderEventing{
 			c: make(chan Event, 1),
 		}
-		eventingImpl.Invoke(Event{EventType: ProviderError})
 
 		provider := struct {
 			FeatureProvider
@@ -990,6 +983,9 @@ func TestEventHandler_HandlersRunImmediately(t *testing.T) {
 		AddHandler(ProviderStale, &callback)
 		AddHandler(ProviderConfigChange, &callback)
 
+		// Emit error event from the provider
+		eventingImpl.Invoke(Event{EventType: ProviderError})
+
 		// assert client transitioned to ERROR
 		eventually(t, func() bool {
 			return NewDefaultClient().State() == ErrorState
@@ -1009,7 +1005,6 @@ func TestEventHandler_HandlersRunImmediately(t *testing.T) {
 		eventingImpl := &ProviderEventing{
 			c: make(chan Event, 1),
 		}
-		eventingImpl.Invoke(Event{EventType: ProviderStale})
 
 		provider := struct {
 			FeatureProvider
@@ -1033,6 +1028,9 @@ func TestEventHandler_HandlersRunImmediately(t *testing.T) {
 		AddHandler(ProviderError, &callback)
 		AddHandler(ProviderConfigChange, &callback)
 
+		// Emit stale event from the provider.
+		eventingImpl.Invoke(Event{EventType: ProviderStale})
+
 		// assert client transitioned to STALE
 		eventually(t, func() bool {
 			return NewDefaultClient().State() == StaleState
@@ -1198,25 +1196,16 @@ func TestEventHandler_1ToNMapping(t *testing.T) {
 
 		executor := newEventExecutor()
 
-		err := executor.registerDefaultProvider(eventingProvider)
-		if err != nil {
-			t.Fatal(err)
-		}
+		executor.registerDefaultProvider(eventingProvider)
 
 		if len(executor.activeSubscriptions) != 1 &&
 			executor.activeSubscriptions[0].featureProvider != eventingProvider.FeatureProvider {
 			t.Fatal("provider not added to active provider subscriptions")
 		}
 
-		err = executor.registerNamedEventingProvider("clientA", eventingProvider)
-		if err != nil {
-			t.Fatal(err)
-		}
+		executor.registerNamedEventingProvider("clientA", eventingProvider)
 
-		err = executor.registerNamedEventingProvider("clientB", eventingProvider)
-		if err != nil {
-			t.Fatal(err)
-		}
+		executor.registerNamedEventingProvider("clientB", eventingProvider)
 
 		if len(executor.activeSubscriptions) != 1 {
 			t.Fatal("multiple provided in active subscriptions")
@@ -1236,15 +1225,9 @@ func TestEventHandler_1ToNMapping(t *testing.T) {
 
 		executor := newEventExecutor()
 
-		err := executor.registerDefaultProvider(eventingProvider)
-		if err != nil {
-			t.Fatal(err)
-		}
+		executor.registerDefaultProvider(eventingProvider)
 
-		err = executor.registerNamedEventingProvider("clientA", eventingProvider)
-		if err != nil {
-			t.Fatal(err)
-		}
+		executor.registerNamedEventingProvider("clientA", eventingProvider)
 
 		overridingProvider := struct {
 			FeatureProvider
@@ -1256,10 +1239,7 @@ func TestEventHandler_1ToNMapping(t *testing.T) {
 			},
 		}
 
-		err = executor.registerNamedEventingProvider("clientA", overridingProvider)
-		if err != nil {
-			t.Fatal(err)
-		}
+		executor.registerNamedEventingProvider("clientA", overridingProvider)
 
 		if len(executor.activeSubscriptions) != 2 {
 			t.Fatal("error with active provider subscriptions")
@@ -1279,15 +1259,9 @@ func TestEventHandler_1ToNMapping(t *testing.T) {
 
 		executor := newEventExecutor()
 
-		err := executor.registerNamedEventingProvider("clientA", eventingProvider)
-		if err != nil {
-			t.Fatal(err)
-		}
+		executor.registerNamedEventingProvider("clientA", eventingProvider)
 
-		err = executor.registerNamedEventingProvider("clientB", eventingProvider)
-		if err != nil {
-			t.Fatal(err)
-		}
+		executor.registerNamedEventingProvider("clientB", eventingProvider)
 
 		overridingProvider := struct {
 			FeatureProvider
@@ -1299,10 +1273,7 @@ func TestEventHandler_1ToNMapping(t *testing.T) {
 			},
 		}
 
-		err = executor.registerNamedEventingProvider("clientA", overridingProvider)
-		if err != nil {
-			t.Fatal(err)
-		}
+		executor.registerNamedEventingProvider("clientA", overridingProvider)
 
 		if len(executor.activeSubscriptions) != 2 {
 			t.Fatal("error with active provider subscriptions")
@@ -1322,10 +1293,7 @@ func TestEventHandler_1ToNMapping(t *testing.T) {
 
 		executor := newEventExecutor()
 
-		err := executor.registerNamedEventingProvider("clientA", eventingProvider)
-		if err != nil {
-			t.Fatal(err)
-		}
+		executor.registerNamedEventingProvider("clientA", eventingProvider)
 
 		overridingProvider := struct {
 			FeatureProvider
@@ -1337,10 +1305,7 @@ func TestEventHandler_1ToNMapping(t *testing.T) {
 			},
 		}
 
-		err = executor.registerNamedEventingProvider("clientA", overridingProvider)
-		if err != nil {
-			t.Fatal(err)
-		}
+		executor.registerNamedEventingProvider("clientA", overridingProvider)
 
 		if len(executor.activeSubscriptions) != 1 &&
 			executor.activeSubscriptions[0].featureProvider != overridingProvider.FeatureProvider {
@@ -1560,8 +1525,7 @@ func TestEventHandler_ChannelClosure(t *testing.T) {
 
 	executor := newEventExecutor()
 
-	err := executor.registerDefaultProvider(eventingProvider)
-	require.NoError(t, err)
+	executor.registerDefaultProvider(eventingProvider)
 	require.Len(t, executor.activeSubscriptions, 1)
 
 	var eventCount atomic.Int64