fix(wait-tx): plug WS-subscriber leak; bind goroutine to subCtx

internal/wait-tx/subscriber.go

@@ -3,13 +3,29 @@ package waittx
 import (
 	"context"
 	"fmt"
+	"os"
 	"strings"
+	"sync/atomic"
+	"time"
 
 	rpchttp "github.com/cometbft/cometbft/rpc/client/http"
 	tmtypes "github.com/cometbft/cometbft/types"
 )
 
-const subscriberID = "sdk-go-wait"
+// subscriberSeq generates unique CometBFT subscriber IDs per process. A
+// constant subscriberID across concurrent callers risked client-side
+// collisions in cometbft's WSEvents.subscriptions[query] map. Even with one
+// rpchttp.Client per call (today), rotation keeps the design forward-safe if
+// callers ever share clients (see RCA item 6b).
+var subscriberSeq uint64
+
+func newSubscriberID() string {
+	return fmt.Sprintf("sdk-go-wait-%d-%d", os.Getpid(), atomic.AddUint64(&subscriberSeq, 1))
+}
+
+// unsubscribeTimeout bounds the lifetime of the deferred Unsubscribe call so
+// teardown cannot block on an unresponsive server forever.
+const unsubscribeTimeout = 2 * time.Second
 
 type subscriber struct {
 	endpoint string
@@ -27,14 +43,30 @@ func (s *subscriber) Wait(ctx context.Context, txHash string) (Result, error) {
 	if err := client.Start(); err != nil {
 		return Result{}, fmt.Errorf("tm client start: %w", err)
 	}
-	defer client.Stop() //nolint:errcheck
+	// Always stop the rpchttp client, even if Subscribe fails or ctx fires.
+	// This is the single point that closes the underlying gorilla WS conn
+	// (WSEvents.OnStop -> ws.Stop). Without this, on any error path the
+	// websocket to lumerad's :26657 stays ESTABLISHED until the OS or peer
+	// times out (~hours). See ops RCA on lumera-devnet-1 val3 leak (~2562
+	// sockets / 11 days).
+	defer func() { _ = client.Stop() }()
 
+	id := newSubscriberID()
 	query := fmt.Sprintf("tm.event='Tx' AND tx.hash='%s'", formatTMHash(txHash))
-	ch, err := client.Subscribe(ctx, subscriberID, query)
+	ch, err := client.Subscribe(ctx, id, query)
 	if err != nil {
 		return Result{}, fmt.Errorf("subscribe: %w", err)
 	}
-	defer client.Unsubscribe(context.Background(), subscriberID, query) //nolint:errcheck
+	// Bounded Unsubscribe context: the outer ctx may already be cancelled by
+	// the time this defer runs, and Background() with no timeout could let a
+	// slow server block teardown indefinitely. 2s is plenty for an unsubscribe
+	// JSON-RPC round-trip; if it fails we still proceed to client.Stop() which
+	// tears down the socket regardless.
+	defer func() {
+		unsubCtx, cancel := context.WithTimeout(context.Background(), unsubscribeTimeout)
+		defer cancel()
+		_ = client.Unsubscribe(unsubCtx, id, query)
+	}()
 
 	for {
 		select {

internal/wait-tx/waiter.go

@@ -49,12 +49,23 @@ func (w *Waiter) Wait(ctx context.Context, txHash string, timeout time.Duration)
 	}
 
 	if w.subscriber != nil {
+		// Bound the subscriber lifetime to setupDelay. If the WS handshake
+		// or first event delivery has not landed by then we fall through to
+		// the gRPC poller. Critically, the spawned goroutine MUST receive
+		// subCtx (not the outer ctx) so that when this select returns via
+		// the subCtx.Done() arm, the subscriber.Wait call inside the
+		// goroutine unwinds promptly and its deferred client.Stop() runs.
+		// Passing the outer ctx here was the root cause of the WS-socket
+		// leak observed on lumera-devnet-1 val3 (~2562 sockets / 11 days):
+		// subCtx fired after 5s, the caller fell through to the poller,
+		// but the goroutine kept blocking on <-ch for the lifetime of the
+		// outer ctx (often unbounded), pinning one rpchttp client open.
 		subCtx, cancel := context.WithTimeout(ctx, w.setupDelay)
 		defer cancel()
 		resCh := make(chan Result, 1)
 		errCh := make(chan error, 1)
 		go func() {
-			res, err := w.subscriber.Wait(ctx, txHash)
+			res, err := w.subscriber.Wait(subCtx, txHash)
 			if err != nil {
 				errCh <- err
 				return

internal/wait-tx/waiter_test.go

@@ -92,3 +92,82 @@ func TestNewSetsDefaults(t *testing.T) {
 		t.Fatalf("unexpected wait error: %v", err)
 	}
 }
+
+// blockingSource records the ctx it was called with and blocks on <-ctx.Done()
+// without ever returning a Result. Used to verify the subscriber goroutine
+// receives subCtx (so it unwinds promptly when setupDelay fires) rather than
+// the outer ctx (which would leave it blocking until the outer ctx expires —
+// the root-cause shape of the lumera-devnet-1 val3 WS-socket leak, RCA on file).
+type blockingSource struct {
+	received chan context.Context
+	done     chan struct{}
+}
+
+func newBlockingSource() *blockingSource {
+	return &blockingSource{received: make(chan context.Context, 1), done: make(chan struct{}, 1)}
+}
+
+func (b *blockingSource) Wait(ctx context.Context, _ string) (Result, error) {
+	b.received <- ctx
+	<-ctx.Done() // unwinds only when the supplied ctx fires
+	b.done <- struct{}{}
+	return Result{}, ctx.Err()
+}
+
+// TestWaiterSubscriberReceivesSubCtxNotOuter is the regression test for the
+// WS-socket leak fix. Before the fix, the goroutine inside Waiter.Wait was
+// invoked with the OUTER ctx; once setupDelay fired the caller fell through
+// to the poller but the goroutine kept blocking on the outer ctx (often
+// unbounded), pinning a CometBFT rpchttp client open and leaking a TCP socket
+// to :26657 for the lifetime of the outer ctx.
+//
+// This test gives the subscriber a blockingSource that records its incoming
+// ctx and asserts that, within a small multiple of setupDelay, the recorded
+// ctx is Done — which is only true if the goroutine was bound to subCtx.
+func TestWaiterSubscriberReceivesSubCtxNotOuter(t *testing.T) {
+	sub := newBlockingSource()
+	poller := &stubSource{res: Result{Code: 9}}
+	w := &Waiter{
+		subscriber: sub,
+		poller:     poller,
+		setupDelay: 25 * time.Millisecond,
+	}
+
+	// Outer ctx has a generous timeout. The test passes only if the
+	// goroutine's ctx is cancelled when setupDelay (subCtx) fires, NOT when
+	// the outer ctx fires.
+	outer, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	if _, err := w.Wait(outer, "hash", 0); err != nil {
+		t.Fatalf("Wait should fall through to poller on subscriber timeout, got err=%v", err)
+	}
+
+	var recvCtx context.Context
+	select {
+	case recvCtx = <-sub.received:
+	case <-time.After(time.Second):
+		t.Fatal("subscriber.Wait was never invoked")
+	}
+
+	// Within ~10x setupDelay the goroutine's ctx MUST be Done. If it isn't,
+	// the goroutine is leaking on the outer ctx (the bug).
+	select {
+	case <-recvCtx.Done():
+		// good
+	case <-time.After(250 * time.Millisecond):
+		t.Fatal("subscriber goroutine's ctx never fired within 10x setupDelay; goroutine is bound to outer ctx (leak)")
+	}
+
+	// And the goroutine itself must have unwound (defer client.Stop() ran).
+	select {
+	case <-sub.done:
+		// good
+	case <-time.After(time.Second):
+		t.Fatal("subscriber goroutine did not unwind after its ctx fired")
+	}
+
+	if poller.calls != 1 {
+		t.Fatalf("poller should have been invoked exactly once; got %d", poller.calls)
+	}
+}