fix(provider): eliminate data race on concurrent init failure in Provider.Get()

tools/fxconfig/internal/client/notifications.go

@@ -37,6 +37,12 @@ type NotificationClient struct {
 	// streamErr holds the error that caused the stream to terminate.
 	// Atomically stored; checked by Subscribe() before sending requests.
 	streamErr atomic.Pointer[error]
+
+	// droppedNotifications counts status events that the dispatcher could not
+	// deliver because the subscriber's channel buffer was full or the receiver
+	// had already given up. Exposed via DroppedNotifications for diagnostics
+	// and tests.
+	droppedNotifications atomic.Uint64
 }
 
 // NewNotificationClient creates a notification client with the provided configuration.
@@ -236,11 +242,6 @@ func (n *NotificationClient) listen(ctx context.Context) error {
 
 	// spawn notification dispatcher
 	g.Go(func() error {
-		type notificationCall struct {
-			receiverQueue chan int
-			status        int
-		}
-
 		var resp *committerpb.NotificationResponse
 		for {
 			select {
@@ -249,33 +250,7 @@ func (n *NotificationClient) listen(ctx context.Context) error {
 			case resp = <-n.responseQueue:
 			}
 
-			res := parseResponse(resp)
-
-			// Collect subscribers under lock, then release before spawning goroutines.
-			// This minimizes lock hold time — only map lookups and deletes happen
-			// under the lock. Goroutine scheduling happens entirely outside.
-			var notifications []notificationCall
-
-			n.subscribersMu.Lock()
-			for txID, v := range res {
-				receivers, ok := n.subscribers[txID]
-				if !ok {
-					continue
-				}
-				delete(n.subscribers, txID)
-				for _, q := range receivers {
-					notifications = append(notifications, notificationCall{receiverQueue: q, status: v})
-				}
-			}
-			n.subscribersMu.Unlock()
-
-			for _, c := range notifications {
-				select {
-				case c.receiverQueue <- c.status:
-				default:
-					// message dropped
-				}
-			}
+			n.dispatchNotifications(n.collectNotifications(parseResponse(resp)))
 		}
 	})
 
@@ -294,6 +269,67 @@ func (n *NotificationClient) listen(ctx context.Context) error {
 	return err
 }
 
+// notificationCall is a single status event ready for delivery to a subscriber.
+type notificationCall struct {
+	txID          string
+	receiverQueue chan int
+	status        int
+}
+
+// collectNotifications snapshots and removes all subscribers whose txIDs
+// appear in res, returning one notificationCall per subscriber. Holding the
+// subscribers lock only for the duration of the lookup/delete keeps the
+// dispatcher hot path short.
+func (n *NotificationClient) collectNotifications(res map[string]int) []notificationCall {
+	var notifications []notificationCall
+
+	n.subscribersMu.Lock()
+	defer n.subscribersMu.Unlock()
+
+	for txID, v := range res {
+		receivers, ok := n.subscribers[txID]
+		if !ok {
+			continue
+		}
+		delete(n.subscribers, txID)
+		for _, q := range receivers {
+			notifications = append(notifications, notificationCall{
+				txID:          txID,
+				receiverQueue: q,
+				status:        v,
+			})
+		}
+	}
+
+	return notifications
+}
+
+// dispatchNotifications sends each pending status to its subscriber. The send
+// is non-blocking so a single slow subscriber cannot stall the dispatcher
+// goroutine, but every dropped event is logged and counted — the txID has
+// already been removed from the subscribers map, so a drop is unrecoverable
+// and must be observable.
+func (n *NotificationClient) dispatchNotifications(notifications []notificationCall) {
+	for _, c := range notifications {
+		select {
+		case c.receiverQueue <- c.status:
+		default:
+			n.droppedNotifications.Add(1)
+			logger.Warnf(
+				"notification dropped (unrecoverable): txID=%s status=%d (subscriber buffer full or receiver gone)",
+				c.txID, c.status,
+			)
+		}
+	}
+}
+
+// DroppedNotifications returns the cumulative count of status events that the
+// dispatcher could not deliver to a subscriber (buffer full or receiver
+// already gone). Useful for end-of-run diagnostics and tests.
+func (n *NotificationClient) DroppedNotifications() uint64 {
+	return n.droppedNotifications.Load()
+}
+
 // parseResponse extracts transaction statuses from a notification response,
 // mapping transaction IDs to their status codes (timeouts and status events).
 func parseResponse(resp *committerpb.NotificationResponse) map[string]int {

tools/fxconfig/internal/client/notifications_test.go

@@ -274,6 +274,93 @@ func TestNotificationClient_Subscribe_Timeout(t *testing.T) {
 	require.ErrorContains(t, err, "deadline exceeded")
 }
 
+// dispatchNotifications / collectNotifications tests
+
+func TestDispatchNotifications_DeliversToSubscriber(t *testing.T) {
+	t.Parallel()
+
+	nc := newTestNotificationClient(time.Second)
+	ch := make(chan int, 1)
+
+	nc.dispatchNotifications([]notificationCall{
+		{txID: "tx1", receiverQueue: ch, status: 7},
+	})
+
+	require.Equal(t, 7, <-ch)
+	require.Equal(t, uint64(0), nc.DroppedNotifications())
+}
+
+func TestDispatchNotifications_DropsWhenBufferFull(t *testing.T) {
+	t.Parallel()
+
+	nc := newTestNotificationClient(time.Second)
+	ch := make(chan int, 1)
+	ch <- 1 // pre-fill the single-slot buffer
+
+	nc.dispatchNotifications([]notificationCall{
+		{txID: "tx1", receiverQueue: ch, status: 2},
+	})
+
+	require.Equal(t, uint64(1), nc.DroppedNotifications())
+	// Original value is preserved; the drop did not corrupt the buffer.
+	require.Equal(t, 1, <-ch)
+}
+
+func TestDispatchNotifications_CountsEachDropIndependently(t *testing.T) {
+	t.Parallel()
+
+	nc := newTestNotificationClient(time.Second)
+	full1 := make(chan int, 1)
+	full1 <- 0
+	full2 := make(chan int, 1)
+	full2 <- 0
+	open := make(chan int, 1)
+
+	nc.dispatchNotifications([]notificationCall{
+		{txID: "tx1", receiverQueue: full1, status: 1},
+		{txID: "tx2", receiverQueue: full2, status: 2},
+		{txID: "tx3", receiverQueue: open, status: 3},
+	})
+
+	require.Equal(t, uint64(2), nc.DroppedNotifications())
+	require.Equal(t, 3, <-open)
+}
+
+func TestCollectNotifications_RemovesSubscribersAndCarriesTxID(t *testing.T) {
+	t.Parallel()
+
+	nc := newTestNotificationClient(time.Second)
+	chA := make(chan int, 1)
+	chB := make(chan int, 1)
+	nc.subscribers["tx1"] = []chan int{chA}
+	nc.subscribers["tx2"] = []chan int{chB}
+
+	got := nc.collectNotifications(map[string]int{"tx1": 11, "tx2": 22})
+
+	require.Len(t, got, 2)
+	// Subscribers must be removed so a future drop is correctly observable.
+	require.Empty(t, nc.subscribers)
+
+	byTxID := map[string]notificationCall{}
+	for _, c := range got {
+		byTxID[c.txID] = c
+	}
+	require.Equal(t, 11, byTxID["tx1"].status)
+	require.Equal(t, chA, byTxID["tx1"].receiverQueue)
+	require.Equal(t, 22, byTxID["tx2"].status)
+	require.Equal(t, chB, byTxID["tx2"].receiverQueue)
+}
+
+func TestCollectNotifications_IgnoresUnknownTxIDs(t *testing.T) {
+	t.Parallel()
+
+	nc := newTestNotificationClient(time.Second)
+	got := nc.collectNotifications(map[string]int{"unknown": 1})
+
+	require.Empty(t, got)
+	require.Equal(t, uint64(0), nc.DroppedNotifications())
+}
+
 // Close tests
 
 func TestNotificationClient_Close_CallsCloseFunc(t *testing.T) {

tools/fxconfig/internal/provider/provider.go

@@ -13,13 +13,17 @@ import (
 	"github.com/hyperledger/fabric-x/tools/fxconfig/internal/validation"
 )
 
+type initResult[T any] struct {
+	instance T
+	err      error
+}
+
 // Provider manages lazy initialization of service instances with validation support.
 // It ensures thread-safe, single initialization using sync.Once.
 type Provider[T any, K Validatable] struct {
 	once              sync.Once
 	factory           func(cfg K) (T, error)
-	instance          T
-	err               error
+	result            initResult[T]
 	cfg               K
 	validationContext validation.Context
 }
@@ -43,12 +47,13 @@ func New[T any, K Validatable](
 func (p *Provider[T, K]) Get() (T, error) {
 	p.once.Do(func() {
 		if err := p.cfg.Validate(p.validationContext); err != nil {
-			p.err = err
+			p.result = initResult[T]{err: err}
 			return
 		}
-		p.instance, p.err = p.factory(p.cfg)
+		inst, err := p.factory(p.cfg)
+		p.result = initResult[T]{instance: inst, err: err}
 	})
-	return p.instance, p.err
+	return p.result.instance, p.result.err
 }
 
 // Validate delegates to the configuration's Validate method.

tools/fxconfig/internal/provider/provider_test.go

@@ -162,6 +162,33 @@ func TestProvider_Get_ThreadSafety(t *testing.T) {
 	require.Equal(t, 1, callCount)
 }
 
+func TestProvider_Get_ThreadSafety_InitFailure(t *testing.T) {
+	t.Parallel()
+
+	validationErr := errors.New("init failure")
+	cfg := &mockConfig{validateErr: validationErr}
+	factory := func(*mockConfig) (*mockService, error) {
+		return &mockService{value: "test"}, nil
+	}
+
+	p := provider.New(factory, cfg, validation.Context{})
+
+	var wg sync.WaitGroup
+	numGoroutines := 20
+	wg.Add(numGoroutines)
+
+	for range numGoroutines {
+		go func() {
+			defer wg.Done()
+			svc, err := p.Get()
+			assert.Nil(t, svc)
+			assert.ErrorIs(t, err, validationErr)
+		}()
+	}
+
+	wg.Wait()
+}
+
 func TestProvider_Validate_Success(t *testing.T) {
 	t.Parallel()