fix(selector): close the token iterator displaced on retry

Selector.selectInternal replaced s.cache on every immediate retry without
closing the iterator it displaced, so a single Select could abandon up to
maxImmediateRetries iterators. On the lazy fetcher path an iterator wraps
sql.Rows, so each one held a database cursor and its pooled connection until
the finalizer ran — draining, under contention, the same pool the vault and
commit path use.

s.mu is documented as protecting the cache field, but selectInternal read
s.cache.Next() and wrote s.cache without holding it. A concurrent Close()
nils the field, so it could be dereferenced after the isClosed() check.

Route both accesses through mutex-guarded helpers: next() reads the current
iterator under s.mu, and swapCache() closes the displaced iterator before
installing the new one. If the selector was closed while the fetch was in
flight, swapCache closes the new iterator too and reports the error, so no
iterator is left unclosed on any path.

Fixes #2019

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

Author: AkramBitar

docs/services/selector.md

@@ -163,6 +163,14 @@ The selector uses a **Token Fetcher** to retrieve available tokens from the data
 5. Selector iterates through tokens, attempting to lock each one
 6. If insufficient tokens, selector requests fresh data and retries
 
+**Iterator lifecycle:** the iterator the fetcher hands out owns a resource — on the lazy
+path it wraps the query's `sql.Rows`, and therefore a database cursor and its pooled
+connection — so exactly one `Close()` per iterator is required. The selector holds at most
+one iterator at a time: step 6 above closes the iterator it displaces before installing the
+refreshed one, and `Selector.Close()` closes whichever is current. Both happen under the
+selector's mutex, so closing a selector while a retry is in flight neither leaks an iterator
+nor races with the retry.
+
 **Adaptive refresh strategy** with two triggers:
 - **Time-based**: Refreshes when data is older than `fetcherCacheRefresh`
 - **Query-based**: Refreshes after `fetcherCacheMaxQueries` queries to prevent serving stale data in high-throughput scenarios

token/services/selector/sherdlock/selector.go

@@ -164,7 +164,7 @@ func (s *Selector) selectInternal(ctx context.Context, owner token.OwnerFilter,
 	}
 	sum, selected, tokensLockedByOthersExist, immediateRetries := token2.NewZeroQuantity(s.precision), collections.NewSet[*token2.ID](), true, 0
 	for {
-		if t, err := s.cache.Next(); err != nil {
+		if t, err := s.next(); err != nil {
 			return nil, nil, immediateRetries, errors.Wrapf(err, "failed to get tokens for [%s:%s]", owner.ID(), tokenType)
 		} else if t == nil {
 			if !tokensLockedByOthersExist {
@@ -189,9 +189,13 @@ func (s *Selector) selectInternal(ctx context.Context, owner token.OwnerFilter,
 			}
 
 			s.logger.DebugfContext(ctx, "Fetch all non-deleted tokens from the DB and refresh the token cache.")
-			if s.cache, err = s.fetcher.UnspentTokensIteratorBy(ctx, owner.ID(), tokenType); err != nil {
+			it, err := s.fetcher.UnspentTokensIteratorBy(ctx, owner.ID(), tokenType)
+			if err != nil {
 				return nil, nil, immediateRetries, errors.Wrapf(err, "failed to reload tokens for retry %d [%s:%s]", immediateRetries, owner.ID(), tokenType)
 			}
+			if err := s.swapCache(it); err != nil {
+				return nil, nil, immediateRetries, err
+			}
 
 			immediateRetries++
 			tokensLockedByOthersExist = false
@@ -222,6 +226,39 @@ func (s *Selector) selectInternal(ctx context.Context, owner token.OwnerFilter,
 	}
 }
 
+// next returns the next token of the current cache. It holds s.mu for the whole
+// call so that a concurrent Close cannot swap the iterator out, or nil it, while
+// it is being read. It reports an error if the selector has already been closed.
+func (s *Selector) next() (*token2.UnspentTokenInWallet, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.cache == nil {
+		return nil, errors.New("selector is already closed")
+	}
+
+	return s.cache.Next()
+}
+
+// swapCache installs it as the new token cache and closes the iterator it
+// replaces, so a refresh on retry does not abandon a database cursor and its
+// pooled connection. If the selector was closed in the meantime, it closes it
+// too and reports an error: no iterator is ever left unclosed.
+func (s *Selector) swapCache(it Iterator[*token2.UnspentTokenInWallet]) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.cache == nil {
+		it.Close()
+
+		return errors.New("selector is already closed")
+	}
+	s.cache.Close()
+	s.cache = it
+
+	return nil
+}
+
 func (s *Selector) Close() error {
 	s.mu.Lock()
 	defer s.mu.Unlock()

token/services/selector/sherdlock/selector_iterator_test.go

@@ -0,0 +1,119 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package sherdlock_test
+
+import (
+	"context"
+	"runtime"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/LFDT-Panurus/panurus/token/services/selector/sherdlock"
+	"github.com/LFDT-Panurus/panurus/token/services/selector/sherdlock/mocks"
+	token2 "github.com/LFDT-Panurus/panurus/token/token"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestSelectorClosesDisplacedIterators verifies that every iterator the retry
+// loop replaces is closed. Previously `selectInternal` overwrote `s.cache` on
+// each immediate retry without closing the iterator it displaced, leaking a
+// database cursor and its pooled connection per retry. See #2019.
+func TestSelectorClosesDisplacedIterators(t *testing.T) {
+	_, metrics := setupMetricsMocks()
+
+	mockFetcher := &mocks.FakeTokenFetcher{}
+	mockLocker := &mocks.FakeTokenLocker{}
+
+	var mu sync.Mutex
+	var iterators []*mocks.FakeIterator[*token2.UnspentTokenInWallet]
+	mockFetcher.UnspentTokensIteratorByStub = func(_ context.Context, _ string, _ token2.Type) (sherdlock.Iterator[*token2.UnspentTokenInWallet], error) {
+		it := &mocks.FakeIterator[*token2.UnspentTokenInWallet]{}
+		// One token, always locked by someone else, then exhausted: this drives a
+		// refresh of the cache on every pass through the loop.
+		it.NextReturnsOnCall(0, &token2.UnspentTokenInWallet{
+			Id:       token2.ID{TxId: "tx1", Index: 0},
+			Type:     "ABC",
+			Quantity: "100",
+		}, nil)
+		it.NextReturnsOnCall(1, nil, nil)
+
+		mu.Lock()
+		defer mu.Unlock()
+		iterators = append(iterators, it)
+
+		return it, nil
+	}
+	// Every token appears locked by another process, so the selector exhausts its
+	// immediate retries and gives up with SufficientButLockedFunds.
+	mockLocker.TryLockReturns(false, nil)
+
+	s := sherdlock.NewSelector(sherdlock.Logger(), mockFetcher, mockLocker, 64, metrics)
+	_, _, err := s.Select(t.Context(), &unitTestMockOwnerFilter{id: "alice"}, "50", "ABC")
+	require.Error(t, err)
+
+	mu.Lock()
+	defer mu.Unlock()
+	require.NotEmpty(t, iterators, "the retry loop must have refreshed the cache at least once")
+	// All but the last iterator were displaced by a refresh and must be closed.
+	for i, it := range iterators[:len(iterators)-1] {
+		assert.Equal(t, 1, it.CloseCallCount(), "displaced iterator %d was leaked", i)
+	}
+	// The last one is still the current cache; Close must release it.
+	require.NoError(t, s.Close())
+	assert.Equal(t, 1, iterators[len(iterators)-1].CloseCallCount(), "the current iterator must be closed by Close")
+}
+
+// TestSelectorCloseDuringRetryIsRaceFree runs Close concurrently with a retrying
+// Select. Under `-race` this covers the unsynchronized reads and writes of
+// `s.cache` inside `selectInternal`, which could also nil-dereference when Close
+// won the race between the closed check and the iterator read. See #2019.
+func TestSelectorCloseDuringRetryIsRaceFree(t *testing.T) {
+	_, metrics := setupMetricsMocks()
+
+	mockFetcher := &mocks.FakeTokenFetcher{}
+	mockLocker := &mocks.FakeTokenLocker{}
+
+	var created atomic.Int64
+	mockFetcher.UnspentTokensIteratorByStub = func(_ context.Context, _ string, _ token2.Type) (sherdlock.Iterator[*token2.UnspentTokenInWallet], error) {
+		created.Add(1)
+		it := &mocks.FakeIterator[*token2.UnspentTokenInWallet]{}
+		it.NextReturnsOnCall(0, &token2.UnspentTokenInWallet{
+			Id:       token2.ID{TxId: "tx1", Index: 0},
+			Type:     "ABC",
+			Quantity: "100",
+		}, nil)
+		it.NextReturnsOnCall(1, nil, nil)
+
+		return it, nil
+	}
+	mockLocker.TryLockReturns(false, nil)
+
+	s := sherdlock.NewSelector(sherdlock.Logger(), mockFetcher, mockLocker, 64, metrics)
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		// Either outcome is valid: locked funds, or "already closed" if Close won.
+		_, _, _ = s.Select(t.Context(), &unitTestMockOwnerFilter{id: "alice"}, "50", "ABC")
+	}()
+	go func() {
+		defer wg.Done()
+		// Give the selector a chance to enter the retry loop before closing it.
+		for created.Load() == 0 {
+			runtime.Gosched()
+		}
+		_ = s.Close()
+	}()
+	wg.Wait()
+
+	// Whoever ran last, the selector ends up closed and stays closed.
+	_, _, err := s.Select(t.Context(), &unitTestMockOwnerFilter{id: "alice"}, "50", "ABC")
+	require.ErrorContains(t, err, "selector is already closed")
+}