fix(fxconfig): cleaning up stale subscriber on canceled/timeout Subscribe paths (hyperledger#169)

## Problem

`NotificationClient.Subscribe()` appended a subscriber before validating
context state. If the context was canceled before queueing the upstream
request, the method returned with an error but left stale subscriber
state in memory for the same txID.

## Root cause

In `Subscribe()` the map mutation happened too early:

1. Create receiver channel
2. Append subscriber to `subscribers[txID]`
3. Later fail on canceled/timed-out context before successful queue send
4. Stale subscriber remained

## Implemented fix

Updated `Subscribe()` to keep single-upstream-subscription behavior and
clean up stale state on failed enqueue:

- Keep subscriber registration under lock
- If this is not the first subscriber for a txID, return immediately
(reuse existing upstream subscription)
- For first subscriber path, if context is canceled before enqueue,
delete `subscribers[txID]`
- If context is canceled while waiting to enqueue, delete
`subscribers[txID]`

This guarantees the map is rolled back when request enqueue does not
happen.

---------

Signed-off-by: rootp1 <arnav.iitr@gmail.com>

Author: rootp1

tools/fxconfig/internal/client/notifications.go

@@ -91,18 +91,50 @@ func (n *NotificationClient) Subscribe(ctx context.Context, txID string) (chan i
 	}
 
 	receiverCh := make(chan int, 1)
+	isFirst := func() bool {
+		n.subscribersMu.Lock()
+		defer n.subscribersMu.Unlock()
 
-	n.subscribersMu.Lock()
-	defer n.subscribersMu.Unlock()
+		subscribers := n.subscribers[txID]
+		n.subscribers[txID] = append(subscribers, receiverCh)
 
-	subscribers := n.subscribers[txID]
-	n.subscribers[txID] = append(subscribers, receiverCh)
+		return len(subscribers) == 0
+	}()
 
-	if len(subscribers) > 0 {
+	if !isFirst {
 		// we already have an active subscription for this txID
 		return receiverCh, nil
 	}
 
+	rollback := func() {
+		// rollback can race logically with dispatcher cleanup in listen(),
+		// where completed txIDs are also deleted from n.subscribers. The shared
+		// subscribersMu lock plus the missing-key guard below makes this idempotent
+		// and safe regardless of which path removes the entry first.
+		n.subscribersMu.Lock()
+		defer n.subscribersMu.Unlock()
+
+		subscribers, ok := n.subscribers[txID]
+		if !ok {
+			return
+		}
+
+		for i, ch := range subscribers {
+			if ch != receiverCh {
+				continue
+			}
+
+			subscribers = append(subscribers[:i], subscribers[i+1:]...)
+			if len(subscribers) == 0 {
+				delete(n.subscribers, txID)
+				return
+			}
+
+			n.subscribers[txID] = subscribers
+			return
+		}
+	}
+
 	// setup request
 	req := &committerpb.NotificationRequest{
 		TxStatusRequest: &committerpb.TxIDsBatch{
@@ -114,13 +146,15 @@ func (n *NotificationClient) Subscribe(ctx context.Context, txID string) (chan i
 	// check if our ctx is still open
 	select {
 	case <-ctx.Done():
+		rollback()
 		return nil, ctx.Err()
 	default:
 	}
 
 	// try to push to request queue
 	select {
 	case <-ctx.Done():
+		rollback()
 		return nil, ctx.Err()
 	case n.requestQueue <- req:
 	}

tools/fxconfig/internal/client/notifications_test.go

@@ -155,6 +155,86 @@ func TestNotificationClient_Subscribe_ContextCanceled(t *testing.T) {
 	require.ErrorIs(t, err, context.Canceled)
 }
 
+func TestNotificationClient_Subscribe_CancelKeepsExistingSubscribers(t *testing.T) {
+	t.Parallel()
+
+	nc := newTestNotificationClient(time.Second)
+	txID := "tx1"
+
+	firstCtx, cancelFirst := context.WithCancel(context.Background())
+	firstErrCh := make(chan error, 1)
+	go func() {
+		_, err := nc.Subscribe(firstCtx, txID)
+		firstErrCh <- err
+	}()
+
+	require.Eventually(t, func() bool {
+		nc.subscribersMu.RLock()
+		defer nc.subscribersMu.RUnlock()
+		return len(nc.subscribers[txID]) == 1
+	}, time.Second, time.Millisecond)
+
+	secondCh, err := nc.Subscribe(t.Context(), txID)
+	require.NoError(t, err)
+	require.NotNil(t, secondCh)
+
+	cancelFirst()
+	require.ErrorIs(t, <-firstErrCh, context.Canceled)
+
+	nc.subscribersMu.RLock()
+	subscribers := append([]chan int(nil), nc.subscribers[txID]...)
+	nc.subscribersMu.RUnlock()
+	require.Len(t, subscribers, 1)
+	require.Equal(t, secondCh, subscribers[0])
+}
+
+func TestNotificationClient_Subscribe_NoStaleSubscriberOnContextCancel(t *testing.T) {
+	t.Parallel()
+
+	nc := newTestNotificationClient(time.Second)
+	txID := "tx1"
+
+	// First call with canceled context
+	ctx1, cancel1 := context.WithCancel(context.Background())
+	cancel1()
+	_, err1 := nc.Subscribe(ctx1, txID)
+	require.ErrorIs(t, err1, context.Canceled)
+
+	// Verify no subscriber was added
+	nc.subscribersMu.RLock()
+	subscribersAfterFirstCall := len(nc.subscribers[txID])
+	nc.subscribersMu.RUnlock()
+	require.Equal(t, 0, subscribersAfterFirstCall, "No subscriber should be added when context is canceled")
+
+	// Second call with active context - should send a request
+	go func() { <-nc.requestQueue }()
+	ch2, err2 := nc.Subscribe(t.Context(), txID)
+	require.NoError(t, err2)
+	require.NotNil(t, ch2)
+
+	// Verify exactly one subscriber is added
+	nc.subscribersMu.RLock()
+	subscribersAfterSecondCall := len(nc.subscribers[txID])
+	nc.subscribersMu.RUnlock()
+	require.Equal(t, 1, subscribersAfterSecondCall, "Exactly one subscriber should be added on successful subscribe")
+}
+
+func TestNotificationClient_Subscribe_NoStaleSubscriberOnTimeout(t *testing.T) {
+	t.Parallel()
+
+	nc := newTestNotificationClient(time.Millisecond)
+	txID := "tx-timeout"
+
+	// No consumer on requestQueue, so send should timeout.
+	_, err := nc.Subscribe(t.Context(), txID)
+	require.Error(t, err)
+
+	nc.subscribersMu.RLock()
+	subscribersAfterTimeout := len(nc.subscribers[txID])
+	nc.subscribersMu.RUnlock()
+	require.Equal(t, 0, subscribersAfterTimeout, "No subscriber should remain after subscribe timeout")
+}
+
 func TestNotificationClient_Subscribe_SendsRequest(t *testing.T) {
 	t.Parallel()