From 3c5c79455aebecfbf74b94e85776e917be32082a Mon Sep 17 00:00:00 2001 From: mayanksharmaCSE Date: Fri, 1 May 2026 14:52:51 +0530 Subject: [PATCH 1/2] fix(fxconfig): make dropped notifications observable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The notification dispatcher in NotificationClient.listen used a bare non-blocking send with a comment, silently losing status events whenever the subscriber buffer was full or the receiver had given up. Because the txID is removed from the subscribers map before delivery is attempted, a dropped event is unrecoverable — and indistinguishable from a genuine committer timeout from the caller's perspective. Signed-off-by: mayanksharmaCSE --- .../fxconfig/internal/client/notifications.go | 100 ++++++++++++------ .../internal/client/notifications_test.go | 87 +++++++++++++++ 2 files changed, 155 insertions(+), 32 deletions(-) diff --git a/tools/fxconfig/internal/client/notifications.go b/tools/fxconfig/internal/client/notifications.go index 29af0ba..146a283 100644 --- a/tools/fxconfig/internal/client/notifications.go +++ b/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 { diff --git a/tools/fxconfig/internal/client/notifications_test.go b/tools/fxconfig/internal/client/notifications_test.go index cc6d0a2..b8b7a6b 100644 --- a/tools/fxconfig/internal/client/notifications_test.go +++ b/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) { From 042b750f1785a1088d6089fc8bb45c8e3a5559a5 Mon Sep 17 00:00:00 2001 From: mayanksharmaCSE Date: Fri, 1 May 2026 20:54:45 +0530 Subject: [PATCH 2/2] fix(provider): eliminate data race on concurrent init failure in Provider.Get() Signed-off-by: mayanksharmaCSE --- tools/fxconfig/internal/provider/provider.go | 15 +++++++---- .../internal/provider/provider_test.go | 27 +++++++++++++++++++ 2 files changed, 37 insertions(+), 5 deletions(-) diff --git a/tools/fxconfig/internal/provider/provider.go b/tools/fxconfig/internal/provider/provider.go index 9cc02c5..8db96cd 100644 --- a/tools/fxconfig/internal/provider/provider.go +++ b/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. diff --git a/tools/fxconfig/internal/provider/provider_test.go b/tools/fxconfig/internal/provider/provider_test.go index bbfd233..74cd8bc 100644 --- a/tools/fxconfig/internal/provider/provider_test.go +++ b/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()