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

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")
+}