Added Backoff

Signed-off-by: Effi-S <effi.szt@gmail.com>

Author: Effi-S

token/core/common/auditor.go

@@ -8,6 +8,7 @@ package common
 
 import (
 	"context"
+	"time"
 
 	"github.com/LFDT-Panurus/panurus/token/driver"
 	"github.com/LFDT-Panurus/panurus/token/driver/protos-go/v1/request"
@@ -17,6 +18,16 @@ import (
 	"go.opentelemetry.io/otel/trace"
 )
 
+// AuditTokensNumRetries and AuditTokensRetryDelay control how RetrieveAuditTokens
+// tolerates the read-timing race in which a referenced token's producing
+// transaction is still pending, so its outputs have not yet been persisted to the
+// token store by the asynchronous finality listener. They mirror the retry/backoff
+// already applied on the sibling token.QueryEngine audit path (see token/vault.go).
+var (
+	AuditTokensNumRetries = 3
+	AuditTokensRetryDelay = 3 * time.Second
+)
+
 // AuditContext contains the context for token request auditing.
 type AuditContext[P driver.PublicParameters, IA driver.IssueAction, TA driver.TransferAction, DS driver.Deserializer] struct {
 	Logger               logging.Logger
@@ -267,7 +278,7 @@ func ExtractTokenIDsAndCheckDuplicates(
 // The returned map uses token ID pointers as keys, allowing callers to efficiently look up
 // tokens by their ID during validation.
 //
-// IMPORTANT: This function always returns a non-nil map (possibly empty) to ensure
+// This function always returns a non-nil map (possibly empty) to ensure
 // validation logic can distinguish between "no tokens requested" and "tokens not found".
 func RetrieveAuditTokens(
 	ctx context.Context,
@@ -288,7 +299,7 @@ func RetrieveAuditTokens(
 	}
 
 	logger.DebugfContext(ctx, "[%s] retrieving [%d] audit tokens...", anchor, len(tokenIDs))
-	tokens, err := queryEngine.ListAuditTokens(ctx, tokenIDs...)
+	tokens, err := listAuditTokensWithRetry(ctx, logger, queryEngine, tokenIDs, anchor)
 	if err != nil {
 		return nil, errors.WithMessagef(err, "failed to retrieve audit tokens for tx [%s]", anchor)
 	}
@@ -304,6 +315,59 @@ func RetrieveAuditTokens(
 	return auditTokens, nil
 }
 
+// listAuditTokensWithRetry calls queryEngine.ListAuditTokens, tolerating the
+// read-timing race where a referenced token is momentarily missing from the token
+// store because its producing transaction is still pending (its outputs are
+// persisted only later, by the asynchronous finality listener). On failure, it
+// checks whether any requested token belongs to a still-pending transaction and,
+// if so, waits AuditTokensRetryDelay and retries up to AuditTokensNumRetries times
+// before giving up. This mirrors the tolerance already implemented for the sibling
+// Audit() path in token/vault.go, so the earlier AuditorCheck gate no longer
+// spuriously rejects a validly-audited, quickly-chained transaction.
+func listAuditTokensWithRetry(
+	ctx context.Context,
+	logger logging.Logger,
+	queryEngine driver.QueryEngine,
+	tokenIDs []*token.ID,
+	anchor driver.TokenRequestAnchor,
+) ([]*token.Token, error) {
+	var tokens []*token.Token
+	var err error
+
+	for i := range AuditTokensNumRetries {
+		tokens, err = queryEngine.ListAuditTokens(ctx, tokenIDs...)
+		if err == nil {
+			return tokens, nil
+		}
+
+		// The lookup failed. Check whether any requested token belongs to a
+		// transaction that is still pending; if so, the row is expected to appear
+		// once the finality listener persists it, so wait a bit and retry.
+		retry := false
+		for _, id := range tokenIDs {
+			pending, pErr := queryEngine.IsPending(ctx, id)
+			if pending || pErr != nil {
+				logger.Warnf("[%s] cannot get audit token for id [%s] because the relative transaction is pending, retry [%d/%d]: with err [%v]", anchor, id, i+1, AuditTokensNumRetries, pErr)
+				if i == AuditTokensNumRetries-1 {
+					return nil, errors.Errorf("failed to get audit tokens, tx [%s] is still pending", id.TxId)
+				}
+				retry = true
+
+				break
+			}
+		}
+
+		if !retry {
+			// None of the tokens is pending: this is a genuine failure, do not retry.
+			return nil, err
+		}
+
+		time.Sleep(AuditTokensRetryDelay)
+	}
+
+	return tokens, err
+}
+
 // ValidateStructure ensures complete structural correspondence between TokenRequest and TokenRequestMetadata.
 // It validates that:
 // - Action counts match between request and metadata

token/core/common/auditor_test.go

@@ -9,12 +9,14 @@ package common
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/LFDT-Panurus/panurus/token/driver"
 	"github.com/LFDT-Panurus/panurus/token/driver/mock"
 	"github.com/LFDT-Panurus/panurus/token/driver/protos-go/v1/request"
 	"github.com/LFDT-Panurus/panurus/token/services/logging"
 	"github.com/LFDT-Panurus/panurus/token/token"
+	"github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -211,6 +213,137 @@ func TestRetrieveAuditTokens(t *testing.T) {
 		assert.Equal(t, tok1, tokens[id1.String()])
 		assert.Nil(t, tokens[id2.String()])
 	})
+
+	t.Run("RetriesWhilePendingThenSucceeds", func(t *testing.T) {
+		// The producing tx is still pending on the first lookup (row not yet
+		// persisted), then becomes available on the retry. This is the exact
+		// read-timing race from issue #2105: the token must be resolved, not
+		// spuriously rejected.
+		defer withFastAuditRetries(t)()
+
+		qe := &mock.QueryEngine{}
+		id1 := &token.ID{TxId: "tx1", Index: 0}
+		tokenIDs := []*token.ID{id1}
+
+		tok1 := &token.Token{Type: "USD", Quantity: "100"}
+		qe.ListAuditTokensReturnsOnCall(0, nil, errors.New("token not found for key [tx1:0]"))
+		qe.ListAuditTokensReturnsOnCall(1, []*token.Token{tok1}, nil)
+		qe.IsPendingReturns(true, nil)
+
+		tokens, err := RetrieveAuditTokens(ctx, logger, qe, tokenIDs, anchor)
+		require.NoError(t, err)
+		assert.Len(t, tokens, 1)
+		assert.Equal(t, tok1, tokens[id1.String()])
+		assert.Equal(t, 2, qe.ListAuditTokensCallCount())
+	})
+
+	t.Run("NoRetryWhenNotPending", func(t *testing.T) {
+		// A genuine failure (no requested token is pending) must fail fast,
+		// without spending the retry budget.
+		defer withFastAuditRetries(t)()
+
+		qe := &mock.QueryEngine{}
+		id1 := &token.ID{TxId: "tx1", Index: 0}
+		tokenIDs := []*token.ID{id1}
+
+		qe.ListAuditTokensReturns(nil, assert.AnError)
+		qe.IsPendingReturns(false, nil)
+
+		tokens, err := RetrieveAuditTokens(ctx, logger, qe, tokenIDs, anchor)
+		require.Error(t, err)
+		assert.Nil(t, tokens)
+		assert.Equal(t, 1, qe.ListAuditTokensCallCount())
+	})
+
+	t.Run("ExhaustsRetriesWhileStillPending", func(t *testing.T) {
+		// The producing tx never leaves the pending state within the grace
+		// window: we give up with a clear "still pending" error.
+		defer withFastAuditRetries(t)()
+
+		qe := &mock.QueryEngine{}
+		id1 := &token.ID{TxId: "tx1", Index: 0}
+		tokenIDs := []*token.ID{id1}
+
+		qe.ListAuditTokensReturns(nil, errors.New("token not found for key [tx1:0]"))
+		qe.IsPendingReturns(true, nil)
+
+		tokens, err := RetrieveAuditTokens(ctx, logger, qe, tokenIDs, anchor)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "still pending")
+		assert.Nil(t, tokens)
+		assert.Equal(t, AuditTokensNumRetries, qe.ListAuditTokensCallCount())
+	})
+}
+
+// withFastAuditRetries shrinks the audit-token retry delay for the duration of a
+// test so the pending-status retry path can be exercised without real sleeps, and
+// restores the original values on cleanup.
+func withFastAuditRetries(t *testing.T) func() {
+	t.Helper()
+	origDelay := AuditTokensRetryDelay
+	AuditTokensRetryDelay = time.Millisecond
+
+	return func() { AuditTokensRetryDelay = origDelay }
+}
+
+// BenchmarkRetrieveAuditTokens measures the latency added by the pending-status
+// retry/backoff introduced for issue #2105.
+//
+//   - NoRace:        the common case — the token is present on the first lookup,
+//     so no retry occurs and the added latency is only the (skipped) retry check.
+//   - RaceResolvesOnRetry: the issue #2105 race — the first lookup misses while
+//     the producing tx is pending, and the token is resolved on the retry. This
+//     is where the one AuditTokensRetryDelay backoff is paid.
+//
+// Run with a real delay to see the actual grace-window cost:
+//
+//	go test ./token/core/common/ -run '^$' -bench BenchmarkRetrieveAuditTokens -benchtime=20x
+func BenchmarkRetrieveAuditTokens(b *testing.B) {
+	ctx := context.Background()
+	logger := &logging.MockLogger{}
+	anchor := driver.TokenRequestAnchor("bench-tx")
+	id1 := &token.ID{TxId: "tx1", Index: 0}
+	tokenIDs := []*token.ID{id1}
+	tok1 := &token.Token{Type: "USD", Quantity: "100"}
+
+	b.Run("NoRace", func(b *testing.B) {
+		qe := &mock.QueryEngine{}
+		qe.ListAuditTokensReturns([]*token.Token{tok1}, nil)
+
+		b.ReportAllocs()
+		for range b.N {
+			if _, err := RetrieveAuditTokens(ctx, logger, qe, tokenIDs, anchor); err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+
+	b.Run("RaceResolvesOnRetry", func(b *testing.B) {
+		// Keep the retry mechanics but use a tiny backoff so the benchmark
+		// measures the added path cost rather than the wall-clock delay itself.
+		origDelay := AuditTokensRetryDelay
+		AuditTokensRetryDelay = time.Millisecond
+		defer func() { AuditTokensRetryDelay = origDelay }()
+
+		qe := &mock.QueryEngine{}
+		qe.IsPendingReturns(true, nil)
+		qe.ListAuditTokensStub = func(_ context.Context, _ ...*token.ID) ([]*token.Token, error) {
+			// Miss on every odd call, hit on every even call, so each iteration
+			// pays exactly one retry.
+			if qe.ListAuditTokensCallCount()%2 == 1 {
+				return nil, errors.New("token not found for key [tx1:0]")
+			}
+
+			return []*token.Token{tok1}, nil
+		}
+
+		b.ReportAllocs()
+		for range b.N {
+			if _, err := RetrieveAuditTokens(ctx, logger, qe, tokenIDs, anchor); err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
 }
 
 func TestValidateStructure(t *testing.T) {