fix race issues

Signed-off-by: Roman Dmytrenko <[email protected]>

Author: erka

openfeature/multi/multiprovider.go

@@ -219,7 +219,7 @@ func NewProvider(providerMap ProviderMap, evaluationStrategy EvaluationStrategy,
 		opt(config)
 	}
 
-	providers := providerMap
+	providers := make(ProviderMap, len(providerMap))
 	collectedHooks := make([]of.Hook, 0, len(providerMap))
 	for name, provider := range providerMap {
 		// Validate Providers
@@ -233,6 +233,7 @@ func NewProvider(providerMap ProviderMap, evaluationStrategy EvaluationStrategy,
 
 		// Wrap any providers that include hooks
 		if (len(provider.Hooks()) + len(config.providerHooks[name])) == 0 {
+			providers[name] = provider
 			continue
 		}
 
@@ -251,7 +252,7 @@ func NewProvider(providerMap ProviderMap, evaluationStrategy EvaluationStrategy,
 		providers:      providers,
 		outboundEvents: make(chan of.Event, len(providers)),
 		logger:         config.logger,
-		metadata:       buildMetadata(providerMap),
+		metadata:       buildMetadata(providers),
 		overallStatus:  of.NotReadyState,
 		providerStatus: make(map[string]of.State, len(providers)),
 		globalHooks:    append(config.hooks, collectedHooks...),
@@ -373,6 +374,7 @@ func (p *Provider) Init(evalCtx of.EvaluationContext) error {
 				l.LogAttrs(context.Background(), slog.LevelDebug, "StateHandle not implemented, skipping initialization")
 			} else if err := stateHandle.Init(evalCtx); err != nil {
 				l.LogAttrs(context.Background(), slog.LevelError, "initialization failed", slog.Any("error", err))
+				p.updateProviderState(name, of.ErrorState)
 				return &ProviderError{
 					Err:          err,
 					ProviderName: name,
@@ -390,10 +392,7 @@ func (p *Provider) Init(evalCtx of.EvaluationContext) error {
 
 	if err := eg.Wait(); err != nil {
 		var pErr *ProviderError
-		if errors.As(err, &pErr) {
-			// Update provider status to error, no event needs to be emitted yet
-			p.updateProviderState(pErr.ProviderName, of.ErrorState)
-		} else {
+		if !errors.As(err, &pErr) {
 			pErr = &ProviderError{
 				Err:          err,
 				ProviderName: "unknown",
@@ -411,23 +410,30 @@ func (p *Provider) Init(evalCtx of.EvaluationContext) error {
 	p.shutdownFunc = shutdownFunc
 
 	p.workerGroup.Add(1)
-	go func() {
+	go func(ctx context.Context) {
 		workerLogger := p.logger.With(slog.String("multiprovider-worker", "event-forwarder-worker"))
 		defer p.workerGroup.Done()
-		for e := range p.inboundEvents {
-			l := workerLogger.With(
-				slog.String(MetadataProviderName, e.providerName),
-				slog.String(MetadataProviderType, e.ProviderName),
-			)
-			l.LogAttrs(context.Background(), slog.LevelDebug, "received event from provider", slog.String("event-type", string(e.EventType)))
-			if p.updateProviderStateFromEvent(e) {
-				p.outboundEvents <- e.Event
-				l.LogAttrs(context.Background(), slog.LevelDebug, "forwarded state update event")
-			} else {
-				l.LogAttrs(context.Background(), slog.LevelDebug, "total state not updated, inbound event will not be emitted")
+
+		for {
+			select {
+			case <-ctx.Done():
+				close(p.outboundEvents)
+				return
+			case e := <-p.inboundEvents:
+				l := workerLogger.With(
+					slog.String(MetadataProviderName, e.providerName),
+					slog.String(MetadataProviderType, e.ProviderName),
+				)
+				l.LogAttrs(context.Background(), slog.LevelDebug, "received event from provider", slog.String("event-type", string(e.EventType)))
+				if p.updateProviderStateFromEvent(e) {
+					p.outboundEvents <- e.Event
+					l.LogAttrs(context.Background(), slog.LevelDebug, "forwarded state update event")
+				} else {
+					l.LogAttrs(context.Background(), slog.LevelDebug, "total state not updated, inbound event will not be emitted")
+				}
 			}
 		}
-	}()
+	}(workerCtx)
 
 	p.setStatus(of.ReadyState)
 	p.initialized = true
@@ -519,13 +525,7 @@ func (p *Provider) Shutdown() {
 	}
 	// Stop all event listener workers, shutdown events should not affect overall state
 	p.shutdownFunc()
-	// Stop forwarding worker
-	close(p.inboundEvents)
-	p.logger.LogAttrs(context.Background(), slog.LevelDebug, "triggered worker shutdown")
-	// Wait for workers to stop
-	p.workerGroup.Wait()
-	p.logger.LogAttrs(context.Background(), slog.LevelDebug, "worker shutdown completed")
-	p.logger.LogAttrs(context.Background(), slog.LevelDebug, "starting provider shutdown")
+
 	var wg sync.WaitGroup
 	for _, provider := range p.providers {
 		wg.Add(1)
@@ -540,9 +540,15 @@ func (p *Provider) Shutdown() {
 
 	p.logger.LogAttrs(context.Background(), slog.LevelDebug, "waiting for provider shutdown completion")
 	wg.Wait()
+	// Stop forwarding worker
+	p.logger.LogAttrs(context.Background(), slog.LevelDebug, "triggered worker shutdown")
+	// Wait for workers to stop
+	p.workerGroup.Wait()
+	p.logger.LogAttrs(context.Background(), slog.LevelDebug, "worker shutdown completed")
+	close(p.inboundEvents)
+	p.logger.LogAttrs(context.Background(), slog.LevelDebug, "starting provider shutdown")
 	p.logger.LogAttrs(context.Background(), slog.LevelDebug, "provider shutdown completed")
 	p.setStatus(of.NotReadyState)
-	close(p.outboundEvents)
 	p.outboundEvents = nil
 	p.inboundEvents = nil
 	p.initialized = false

openfeature/multi/multiprovider_test.go

@@ -475,14 +475,8 @@ func TestMultiProvider_Track(t *testing.T) {
 		require.NoError(t, err)
 
 		// Simulate error state for one provider
-		errorProvider.eventChannel <- of.Event{
-			ProviderName: "error-provider",
-			EventType:    of.ProviderError,
-			ProviderEventDetails: of.ProviderEventDetails{
-				Message:       "error",
-				EventMetadata: make(map[string]any),
-			},
-		}
+		errorProvider.EmitEvent(of.ProviderError, "error")
+
 		// wait for event processing
 		<-mp.outboundEvents