diff --git a/docs/services/selector.md b/docs/services/selector.md index 3cb53ccf4f..fbccbcf65d 100644 --- a/docs/services/selector.md +++ b/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 diff --git a/token/services/selector/sherdlock/selector.go b/token/services/selector/sherdlock/selector.go index 71f7ae0f13..53c582cb4c 100644 --- a/token/services/selector/sherdlock/selector.go +++ b/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() diff --git a/token/services/selector/sherdlock/selector_iterator_test.go b/token/services/selector/sherdlock/selector_iterator_test.go new file mode 100644 index 0000000000..c5573123ca --- /dev/null +++ b/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") +}