fix(network): remove goroutine leak in transfer-metadata lookup wait

LookupTransferMetadataKey waited for its lookupListener by spawning a
goroutine parked on sync.WaitGroup.Wait(). On the timeout path the
function returns and its deferred RemoveLookupListener tears the listener
down, so OnStatus/OnError -- the only callers of wg.Done() -- never fire
and the goroutine is parked for the life of the process. ScanForPreImage
uses a 5 minute timeout, so every HTLC pre-image that is never revealed
stranded one goroutine.

Buffering the channel does not help: the goroutine blocked on wg.Wait(),
not on a channel send, and the send it replaced was a close(), which
never blocks.

Remove the goroutine instead. lookupListener now owns a done channel that
the first matching notification closes, and the waiter selects on it
against a timer. With nothing parked on the notification there is nothing
to leak. Closing exactly once also removes the "negative WaitGroup
counter" panic a duplicate notification could trigger, and the timer is
explicitly stopped so a lookup that completes early no longer retains one
for the rest of its timeout.

The timeout is reported as a plain error naming the key rather than a bare
context sentinel: a lookup that does not complete in time is an expected
outcome here, and surfacing context.DeadlineExceeded would make it
indistinguishable from caller cancellation to errors.Is-based retry logic.
The context parameter is dropped rather than left dead -- neither
driver.Network.LookupTransferMetadataKey nor htlc.ScanForPreImage carries
a context, so the only caller could pass nothing but context.Background().

The tests are rewritten to actually reproduce the bug: they never release
the listener, since releasing it also let the leaked goroutine exit, which
is why the previous ones passed against the unfixed code. Verified that
goleak flags the pre-fix implementation under these assertions and passes
the new one.

Fixes #2124

Signed-off-by: AkramBitar <akram@il.ibm.com>

Author: AkramBitar

go.mod

@@ -26,6 +26,7 @@ require (
 	github.com/tidwall/gjson v1.19.0
 	go.opentelemetry.io/otel/trace v1.44.0
 	go.uber.org/dig v1.19.0
+	go.uber.org/goleak v1.3.0
 	go.uber.org/zap v1.28.0
 	go.yaml.in/yaml/v3 v3.0.4
 	golang.org/x/crypto v0.54.0

token/services/network/fabric/lookup_listener_test.go

@@ -0,0 +1,126 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package fabric
+
+import (
+	"context"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+)
+
+const testKey = "transfer-metadata-key"
+
+// TestLookupListenerWaitOnStatus verifies the happy path: the value reported by the
+// lookup manager is returned to the waiter.
+func TestLookupListenerWaitOnStatus(t *testing.T) {
+	l := newLookupListener(testKey)
+
+	go l.OnStatus(context.Background(), testKey, []byte("pre-image"))
+
+	value, err := l.wait(5 * time.Second)
+	require.NoError(t, err)
+	assert.Equal(t, []byte("pre-image"), value)
+}
+
+// TestLookupListenerWaitOnError verifies that a failure reported by the lookup manager
+// is surfaced to the waiter.
+func TestLookupListenerWaitOnError(t *testing.T) {
+	l := newLookupListener(testKey)
+	expected := errors.New("scan failed")
+
+	go l.OnError(context.Background(), testKey, expected)
+
+	value, err := l.wait(5 * time.Second)
+	require.ErrorIs(t, err, expected)
+	assert.Nil(t, value)
+}
+
+// TestLookupListenerIgnoresOtherKeys verifies that notifications for a different key do
+// not release the waiter, which must still time out.
+func TestLookupListenerIgnoresOtherKeys(t *testing.T) {
+	l := newLookupListener(testKey)
+
+	l.OnStatus(context.Background(), "some-other-key", []byte("not mine"))
+	l.OnError(context.Background(), "some-other-key", errors.New("not mine either"))
+
+	_, err := l.wait(20 * time.Millisecond)
+	require.ErrorContains(t, err, "timed out")
+}
+
+// TestLookupListenerFirstNotificationWins verifies that done is closed exactly once, so a
+// duplicate or racing notification neither panics nor overwrites the reported result.
+func TestLookupListenerFirstNotificationWins(t *testing.T) {
+	l := newLookupListener(testKey)
+
+	l.OnStatus(context.Background(), testKey, []byte("first"))
+	l.OnStatus(context.Background(), testKey, []byte("second"))
+	l.OnError(context.Background(), testKey, errors.New("late failure"))
+
+	value, err := l.wait(5 * time.Second)
+	require.NoError(t, err)
+	assert.Equal(t, []byte("first"), value)
+}
+
+// TestLookupListenerConcurrentNotifications verifies that concurrent notifications for the
+// same key are safe. Run with -race, this covers the close-once path under contention.
+func TestLookupListenerConcurrentNotifications(t *testing.T) {
+	l := newLookupListener(testKey)
+
+	var wg sync.WaitGroup
+	for i := range 16 {
+		wg.Go(func() {
+			if i%2 == 0 {
+				l.OnStatus(context.Background(), testKey, []byte(strconv.Itoa(i)))
+			} else {
+				l.OnError(context.Background(), testKey, errors.New(strconv.Itoa(i)))
+			}
+		})
+	}
+	wg.Wait()
+
+	_, err := l.wait(5 * time.Second)
+	// Which notification won is undefined; not panicking and not blocking is the point.
+	_ = err
+}
+
+// TestLookupListenerWaitTimeoutDoesNotLeak is the regression test for issue #2124.
+//
+// The waiter is abandoned exactly as LookupTransferMetadataKey abandons it in production:
+// the timeout fires and the listener is never notified, because the deferred
+// RemoveLookupListener has torn it down, so OnStatus/OnError will never be called.
+//
+// The test deliberately does NOT release the listener afterwards. That is what made the
+// original tests vacuous — releasing the signal also let the leaked goroutine exit, so they
+// passed against the buggy implementation. goleak fails here if anything is still parked.
+func TestLookupListenerWaitTimeoutDoesNotLeak(t *testing.T) {
+	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
+
+	l := newLookupListener(testKey)
+
+	_, err := l.wait(20 * time.Millisecond)
+	require.ErrorContains(t, err, "timed out")
+	require.ErrorContains(t, err, testKey, "the timeout error should name the key being looked up")
+}
+
+// TestLookupListenerRepeatedTimeoutsDoNotLeak covers the accumulating case from #2124: every
+// unrevealed HTLC pre-image used to strand one goroutine for the life of the process. As
+// above, no listener is ever notified.
+func TestLookupListenerRepeatedTimeoutsDoNotLeak(t *testing.T) {
+	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
+
+	for range 32 {
+		_, err := newLookupListener(testKey).wait(time.Millisecond)
+		require.ErrorContains(t, err, "timed out")
+	}
+}

token/services/network/fabric/network.go

@@ -407,9 +407,7 @@ func (n *Network) LookupTransferMetadataKey(namespace string, key string, timeou
 		return nil, errors.Wrapf(err, "failed to generate transfer action metadata key from [%s]", key)
 	}
 	logger.Debugf("lookup transfer metadata key [%s] from [%s] in namespace [%s]", key, transferMetadataKey, namespace)
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
-	l := &lookupListener{wg: wg, key: transferMetadataKey}
+	l := newLookupListener(transferMetadataKey)
 	if err := n.llm.AddLookupListener(namespace, transferMetadataKey, l); err != nil {
 		return nil, errors.Wrapf(err, "failed to add lookup listener")
 	}
@@ -418,12 +416,10 @@ func (n *Network) LookupTransferMetadataKey(namespace string, key string, timeou
 			logger.Debugf("failed to remove lookup listener [%s]: %v", transferMetadataKey, err)
 		}
 	}()
-	if err := waitTimeout(wg, timeout); err != nil {
-		return nil, err
-	}
-	logger.Debugf("lookup transfer metadata key [%s] from [%s] in namespace [%s], done, result [%s][%s]", key, transferMetadataKey, namespace, l.value, l.err)
+	value, err := l.wait(timeout)
+	logger.Debugf("lookup transfer metadata key [%s] from [%s] in namespace [%s], done, result [%s][%v]", key, transferMetadataKey, namespace, value, err)
 
-	return l.value, l.err
+	return value, err
 }
 
 // Ledger returns direct access to the ledger querying layer.
@@ -552,44 +548,59 @@ func (n *Network) createCleanupManager(tmsID token2.TMSID) (*cleanup.Manager, er
 	return manager, nil
 }
 
+// lookupListener waits for the lookup manager to report the value of a single key.
+//
+// Completion is signalled by closing done, so a waiter that gives up on timeout leaves
+// nothing behind: there is no goroutine parked on the notification (#2124). done is closed
+// exactly once, which also makes a duplicate or racing notification harmless.
 type lookupListener struct {
 	key   string
-	wg    *sync.WaitGroup
+	done  chan struct{}
+	once  sync.Once
 	value []byte
 	err   error
 }
 
+// newLookupListener returns a listener waiting for the passed key.
+func newLookupListener(key string) *lookupListener {
+	return &lookupListener{key: key, done: make(chan struct{})}
+}
+
 func (l *lookupListener) OnStatus(ctx context.Context, key string, value []byte) {
 	logger.DebugfContext(ctx, "lookup transfer metadata key [%s], got value [%s][%v]", l.key, key, value)
-	if l.key == key {
-		l.value = value
-		l.wg.Done()
-
+	if l.key != key {
 		return
 	}
+	l.once.Do(func() {
+		l.value = value
+		close(l.done)
+	})
 }
 
 func (l *lookupListener) OnError(ctx context.Context, key string, err error) {
 	logger.DebugfContext(ctx, "lookup transfer metadata key [%s], got error [%s][%s]", l.key, key, err)
-	if l.key == key {
-		l.err = err
-		l.wg.Done()
-
+	if l.key != key {
 		return
 	}
+	l.once.Do(func() {
+		l.err = err
+		close(l.done)
+	})
 }
 
-func waitTimeout(wg *sync.WaitGroup, timeout time.Duration) error {
-	c := make(chan struct{})
-	go func() {
-		defer close(c)
-		wg.Wait()
-	}()
+// wait blocks until the listener has been notified or the timeout expires, and returns
+// whatever the lookup manager reported for the key. The timeout is reported as a plain
+// error rather than a context sentinel: a lookup that does not complete in time is an
+// expected outcome here, not a cancellation of the caller.
+func (l *lookupListener) wait(timeout time.Duration) ([]byte, error) {
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+
 	select {
-	case <-c:
-		return nil
-	case <-time.After(timeout):
-		return errors.Errorf("context done")
+	case <-l.done:
+		return l.value, l.err
+	case <-timer.C:
+		return nil, errors.Errorf("timed out after [%s] waiting for lookup of key [%s]", timeout, l.key)
 	}
 }