diff --git a/docs/configuration.md b/docs/configuration.md index 436a4aeca7..b458fded70 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -415,7 +415,24 @@ token: # Adds random jitter to break symmetry when multiple auditors retry simultaneously # Default: 0.3 (30%) jitterFactor: 0.3 - Security: 256 + + # auditTokensRetry controls how the auditor's early validation gate + # (AuditorCheck) 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. When a lookup misses and the producing tx is still pending, + # the gate waits and retries instead of spuriously rejecting the request. + auditTokensRetry: + # numRetries is the number of token-lookup attempts before giving up. + # There are numRetries-1 delays between attempts. Values <= 0 (or an + # omitted key) keep the default; a single attempt is always made. + # Default: 3 + numRetries: 3 + # retryDelay is the backoff slept between attempts, as a duration string + # (e.g. 500ms, 3s). Must be > 0; an invalid or non-positive value falls + # back to the default. The backoff honors context cancellation. + # Default: 3s + retryDelay: 3s ``` ## Minimal Configuration @@ -821,6 +838,59 @@ Default values: --- +### Optional: token.tms..auditor.auditTokensRetry + +Controls how the auditor's early validation gate (`AuditorCheck`) 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. On a missing lookup whose producing transaction is +still `Pending`, the gate waits `retryDelay` and retries up to `numRetries` +attempts before failing, instead of spuriously rejecting a validly-audited, +quickly-chained transaction. This mirrors the tolerance already applied on the +sibling `Audit()` path. Applies to both the `fabtoken` and `dlog` drivers. + +If not specified, the default configuration is: + +```yaml +token: + tms: + : + auditor: + auditTokensRetry: + numRetries: 3 + retryDelay: 3s +``` + +Default values: + +- numRetries: 3 +- retryDelay: 3s + +**Parameter Descriptions:** + +- **numRetries**: Number of token-lookup attempts made before giving up, with + `numRetries - 1` backoff delays between them (so the defaults give a ~6s grace + window: 3 attempts, 2 delays of 3s). A value `<= 0`, or an omitted key, keeps the + default; at least one attempt is always made. Retries are only spent while a + referenced transaction is genuinely `Pending` — a hard lookup failure (or a + failure to determine pending status) fails fast without consuming the budget. +- **retryDelay**: Backoff slept between attempts, expressed as a Go duration string + (e.g. `500ms`, `3s`). Must be `> 0`; an invalid or non-positive value logs a + warning and falls back to the default. The backoff honors context cancellation, + so a cancelled or timed-out request returns immediately rather than pinning a + goroutine for the full grace window. + +**Tuning Recommendations:** + +- **For quickly-chained, high-throughput workloads** (an output spent shortly after + it was audited), raise `numRetries` and/or `retryDelay` to widen the grace window + and further reduce spurious rejections when finality persistence lags. +- **For latency-sensitive gates**, lower `retryDelay` (e.g. `500ms`) so the audit + gate fails faster when the producing transaction never becomes final, at the cost + of tolerating a shorter persistence lag. + +--- + ### Optional: token.tms..auditor.locker Controls the distributed locking strategy used by the auditor to serialise diff --git a/token/core/common/audit_retry_config.go b/token/core/common/audit_retry_config.go new file mode 100644 index 0000000000..dbe918d996 --- /dev/null +++ b/token/core/common/audit_retry_config.go @@ -0,0 +1,93 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package common + +import ( + "time" + + "github.com/LFDT-Panurus/panurus/token/services/logging" +) + +// AuditRetryConfigKey is the per-TMS configuration key (relative to the TMS +// configuration block) holding the audit-token retry/backoff settings consumed by +// AuditorCheck's RetrieveAuditTokens call. +const AuditRetryConfigKey = "auditor.auditTokensRetry" + +// AuditRetryConfig holds the retry budget and backoff used by RetrieveAuditTokens +// to tolerate the pending-transaction read-timing race (issue #2105). +type AuditRetryConfig struct { + // NumRetries is the number of ListAuditTokens attempts made before giving up. + // A value <= 0 is clamped to a single attempt by RetrieveAuditTokens. + NumRetries int + // RetryDelay is the backoff slept between attempts (NumRetries-1 delays). + RetryDelay time.Duration +} + +// DefaultAuditRetryConfig returns the audit-token retry configuration using the +// package default constants. +func DefaultAuditRetryConfig() AuditRetryConfig { + return AuditRetryConfig{ + NumRetries: DefaultAuditTokensNumRetries, + RetryDelay: DefaultAuditTokensRetryDelay, + } +} + +// AuditConfigProvider is the minimal configuration surface needed to load the +// audit-token retry configuration. It is satisfied by driver.Configuration (the +// per-TMS config) and is kept minimal so it is trivial to mock in tests. +type AuditConfigProvider interface { + // IsSet checks whether a configuration key is defined. + IsSet(key string) bool + // UnmarshalKey decodes the configuration value associated with a key into rawVal. + UnmarshalKey(key string, rawVal any) error +} + +// auditRetryConfigRaw is the yaml-facing shape of AuditRetryConfigKey. RetryDelay +// is decoded as a duration string (e.g. "3s") so operators can express it in +// human-readable units. +type auditRetryConfigRaw struct { + NumRetries int `yaml:"numRetries"` + RetryDelay string `yaml:"retryDelay"` +} + +// LoadAuditRetryConfig loads the audit-token retry configuration from the passed +// provider, overlaying any valid configured value onto DefaultAuditRetryConfig(). +// A missing key, an unmarshal failure, or an individually invalid field leaves the +// corresponding default in place (a warning is logged for invalid values), so this +// function never fails: the audit gate always has a usable configuration. +func LoadAuditRetryConfig(cp AuditConfigProvider) AuditRetryConfig { + cfg := DefaultAuditRetryConfig() + + if cp == nil || !cp.IsSet(AuditRetryConfigKey) { + return cfg + } + + var raw auditRetryConfigRaw + if err := cp.UnmarshalKey(AuditRetryConfigKey, &raw); err != nil { + logging.MustGetLogger().Warnf("failed to unmarshal audit-token retry configuration [%s], using defaults: %v", AuditRetryConfigKey, err) + + return cfg + } + + // Apply the retry count if valid. A value <= 0 keeps the default; disabling + // retries entirely is intentionally not expressible here, matching the + // "at least one attempt" clamp in RetrieveAuditTokens. + if raw.NumRetries > 0 { + cfg.NumRetries = raw.NumRetries + } + + // Apply the retry delay if valid. + if raw.RetryDelay != "" { + if duration, err := time.ParseDuration(raw.RetryDelay); err == nil && duration > 0 { + cfg.RetryDelay = duration + } else { + logging.MustGetLogger().Warnf("invalid retryDelay value [%s] for key [%s], using default", raw.RetryDelay, AuditRetryConfigKey) + } + } + + return cfg +} diff --git a/token/core/common/audit_retry_config_test.go b/token/core/common/audit_retry_config_test.go new file mode 100644 index 0000000000..0a8ea785e6 --- /dev/null +++ b/token/core/common/audit_retry_config_test.go @@ -0,0 +1,132 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package common + +import ( + "testing" + "time" + + "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" + "github.com/stretchr/testify/assert" +) + +// stubAuditConfigProvider is a minimal AuditConfigProvider whose IsSet result and +// UnmarshalKey behavior are supplied per test. +type stubAuditConfigProvider struct { + set bool + unmarshal func(key string, rawVal any) error + lastKey string + unmarshaled bool +} + +func (s *stubAuditConfigProvider) IsSet(string) bool { return s.set } + +func (s *stubAuditConfigProvider) UnmarshalKey(key string, rawVal any) error { + s.lastKey = key + s.unmarshaled = true + if s.unmarshal == nil { + return nil + } + + return s.unmarshal(key, rawVal) +} + +func TestDefaultAuditRetryConfig(t *testing.T) { + cfg := DefaultAuditRetryConfig() + assert.Equal(t, DefaultAuditTokensNumRetries, cfg.NumRetries) + assert.Equal(t, DefaultAuditTokensRetryDelay, cfg.RetryDelay) +} + +func TestLoadAuditRetryConfig(t *testing.T) { + t.Run("NilProviderReturnsDefaults", func(t *testing.T) { + cfg := LoadAuditRetryConfig(nil) + assert.Equal(t, DefaultAuditRetryConfig(), cfg) + }) + + t.Run("KeyNotSetReturnsDefaults", func(t *testing.T) { + cp := &stubAuditConfigProvider{set: false} + cfg := LoadAuditRetryConfig(cp) + assert.Equal(t, DefaultAuditRetryConfig(), cfg) + assert.False(t, cp.unmarshaled, "UnmarshalKey must not be called when the key is unset") + }) + + t.Run("UsesConfiguredKey", func(t *testing.T) { + cp := &stubAuditConfigProvider{set: true} + LoadAuditRetryConfig(cp) + assert.Equal(t, AuditRetryConfigKey, cp.lastKey) + }) + + t.Run("OverridesBothFields", func(t *testing.T) { + cp := &stubAuditConfigProvider{set: true, unmarshal: func(_ string, rawVal any) error { + raw := rawVal.(*auditRetryConfigRaw) + raw.NumRetries = 7 + raw.RetryDelay = "500ms" + + return nil + }} + cfg := LoadAuditRetryConfig(cp) + assert.Equal(t, 7, cfg.NumRetries) + assert.Equal(t, 500*time.Millisecond, cfg.RetryDelay) + }) + + t.Run("UnmarshalErrorReturnsDefaults", func(t *testing.T) { + cp := &stubAuditConfigProvider{set: true, unmarshal: func(_ string, _ any) error { + return errors.New("boom") + }} + cfg := LoadAuditRetryConfig(cp) + assert.Equal(t, DefaultAuditRetryConfig(), cfg) + }) + + t.Run("NonPositiveNumRetriesKeepsDefault", func(t *testing.T) { + cp := &stubAuditConfigProvider{set: true, unmarshal: func(_ string, rawVal any) error { + raw := rawVal.(*auditRetryConfigRaw) + raw.NumRetries = 0 + raw.RetryDelay = "2s" + + return nil + }} + cfg := LoadAuditRetryConfig(cp) + assert.Equal(t, DefaultAuditTokensNumRetries, cfg.NumRetries) + assert.Equal(t, 2*time.Second, cfg.RetryDelay) + }) + + t.Run("InvalidRetryDelayKeepsDefault", func(t *testing.T) { + cp := &stubAuditConfigProvider{set: true, unmarshal: func(_ string, rawVal any) error { + raw := rawVal.(*auditRetryConfigRaw) + raw.NumRetries = 5 + raw.RetryDelay = "not-a-duration" + + return nil + }} + cfg := LoadAuditRetryConfig(cp) + assert.Equal(t, 5, cfg.NumRetries) + assert.Equal(t, DefaultAuditTokensRetryDelay, cfg.RetryDelay) + }) + + t.Run("NonPositiveRetryDelayKeepsDefault", func(t *testing.T) { + cp := &stubAuditConfigProvider{set: true, unmarshal: func(_ string, rawVal any) error { + raw := rawVal.(*auditRetryConfigRaw) + raw.RetryDelay = "0s" + + return nil + }} + cfg := LoadAuditRetryConfig(cp) + assert.Equal(t, DefaultAuditTokensRetryDelay, cfg.RetryDelay) + }) + + t.Run("EmptyRetryDelayKeepsDefault", func(t *testing.T) { + cp := &stubAuditConfigProvider{set: true, unmarshal: func(_ string, rawVal any) error { + raw := rawVal.(*auditRetryConfigRaw) + raw.NumRetries = 4 + + return nil + }} + cfg := LoadAuditRetryConfig(cp) + assert.Equal(t, 4, cfg.NumRetries) + assert.Equal(t, DefaultAuditTokensRetryDelay, cfg.RetryDelay) + }) +} diff --git a/token/core/common/auditor.go b/token/core/common/auditor.go index 96f7ad6e20..da3f455f3f 100644 --- a/token/core/common/auditor.go +++ b/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,18 @@ import ( "go.opentelemetry.io/otel/trace" ) +// DefaultAuditTokensNumRetries and DefaultAuditTokensRetryDelay are the default +// retry budget and backoff used by RetrieveAuditTokens to tolerate 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). Per-instance tuning is done via +// the AuditorService fields that default to these values, not by mutating a global. +const ( + DefaultAuditTokensNumRetries = 3 + DefaultAuditTokensRetryDelay = 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,14 +280,20 @@ 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". +// +// numRetries and retryDelay control the tolerance for the pending-transaction +// read-timing race (see DefaultAuditTokensNumRetries / DefaultAuditTokensRetryDelay). +// A numRetries <= 0 is clamped to a single attempt. func RetrieveAuditTokens( ctx context.Context, logger logging.Logger, queryEngine driver.QueryEngine, tokenIDs []*token.ID, anchor driver.TokenRequestAnchor, + numRetries int, + retryDelay time.Duration, ) (map[string]*token.Token, error) { if logger == nil { return nil, errors.Errorf("logger cannot be nil for tx [%s]", anchor) @@ -288,7 +307,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, numRetries, retryDelay) if err != nil { return nil, errors.WithMessagef(err, "failed to retrieve audit tokens for tx [%s]", anchor) } @@ -304,6 +323,89 @@ 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 retryDelay and retries. It makes up to numRetries attempts total, +// sleeping retryDelay between them (numRetries-1 delays); numRetries <= 0 is +// clamped to a single attempt. 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. +// +// The backoff honors ctx cancellation, so a cancelled or timed-out request returns +// immediately instead of pinning a goroutine for the full grace window. A genuine +// (non-pending) lookup failure, or a failure to determine pending status, is +// returned rather than retried and never masked as "still pending". +func listAuditTokensWithRetry( + ctx context.Context, + logger logging.Logger, + queryEngine driver.QueryEngine, + tokenIDs []*token.ID, + anchor driver.TokenRequestAnchor, + numRetries int, + retryDelay time.Duration, +) ([]*token.Token, error) { + attempts := max(1, numRetries) + + var tokens []*token.Token + var err error + + for i := range attempts { + 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 pErr != nil { + // We could not even determine the pending status: this is a hard + // failure, not a pending transaction. Surface both errors instead + // of masking them as "still pending". + return nil, errors.Wrapf(errors.Join(err, pErr), "failed to retrieve audit tokens, tx [%s]: cannot determine pending status of token [%s]", anchor, id) + } + if pending { + 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, attempts, err) + retry = true + + break + } + } + + if !retry { + // None of the tokens is pending: this is a genuine failure, do not retry. + return nil, err + } + + if i == attempts-1 { + // Retry budget exhausted while a token is still pending. Report that, + // but keep the underlying lookup error so operators can diagnose it. + return nil, errors.Wrapf(err, "failed to get audit tokens for tx [%s], transaction is still pending", anchor) + } + + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(retryDelay): + } + } + + // Unreachable: the loop above returns on success, on genuine failure, and on + // the final pending attempt. Return an explicit error so a nil/nil pair can + // never escape to callers that index the returned slice. + if err == nil { + err = errors.Errorf("failed to retrieve audit tokens for tx [%s]", anchor) + } + + return nil, err +} + // ValidateStructure ensures complete structural correspondence between TokenRequest and TokenRequestMetadata. // It validates that: // - Action counts match between request and metadata diff --git a/token/core/common/auditor_test.go b/token/core/common/auditor_test.go index b25eebb2bf..30dcfda061 100644 --- a/token/core/common/auditor_test.go +++ b/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" ) @@ -154,10 +156,15 @@ func TestRetrieveAuditTokens(t *testing.T) { logger := &logging.MockLogger{} anchor := driver.TokenRequestAnchor("test-tx") + // testNumRetries mirrors the production default; testRetryDelay is tiny so the + // pending-status retry path can be exercised without real sleeps. + const testNumRetries = DefaultAuditTokensNumRetries + const testRetryDelay = time.Millisecond + t.Run("EmptyTokenIDs", func(t *testing.T) { qe := &mock.QueryEngine{} - tokens, err := RetrieveAuditTokens(ctx, logger, qe, nil, anchor) + tokens, err := RetrieveAuditTokens(ctx, logger, qe, nil, anchor, testNumRetries, testRetryDelay) require.NoError(t, err) assert.NotNil(t, tokens) assert.Empty(t, tokens) @@ -174,7 +181,7 @@ func TestRetrieveAuditTokens(t *testing.T) { tok2 := &token.Token{Type: "USD", Quantity: "200"} qe.ListAuditTokensReturns([]*token.Token{tok1, tok2}, nil) - tokens, err := RetrieveAuditTokens(ctx, logger, qe, tokenIDs, anchor) + tokens, err := RetrieveAuditTokens(ctx, logger, qe, tokenIDs, anchor, testNumRetries, testRetryDelay) require.NoError(t, err) assert.Len(t, tokens, 2) assert.Equal(t, tok1, tokens[id1.String()]) @@ -190,7 +197,7 @@ func TestRetrieveAuditTokens(t *testing.T) { qe.ListAuditTokensReturns(nil, assert.AnError) - tokens, err := RetrieveAuditTokens(ctx, logger, qe, tokenIDs, anchor) + tokens, err := RetrieveAuditTokens(ctx, logger, qe, tokenIDs, anchor, testNumRetries, testRetryDelay) require.Error(t, err) assert.Contains(t, err.Error(), "failed to retrieve audit tokens") assert.Nil(t, tokens) @@ -205,12 +212,182 @@ func TestRetrieveAuditTokens(t *testing.T) { tok1 := &token.Token{Type: "USD", Quantity: "100"} qe.ListAuditTokensReturns([]*token.Token{tok1, nil}, nil) - tokens, err := RetrieveAuditTokens(ctx, logger, qe, tokenIDs, anchor) + tokens, err := RetrieveAuditTokens(ctx, logger, qe, tokenIDs, anchor, testNumRetries, testRetryDelay) require.NoError(t, err) assert.Len(t, tokens, 2) 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. + 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, testNumRetries, testRetryDelay) + 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. + 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, testNumRetries, testRetryDelay) + 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 that still + // carries the underlying lookup error. + 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, testNumRetries, testRetryDelay) + require.Error(t, err) + assert.Contains(t, err.Error(), "still pending") + assert.Contains(t, err.Error(), "token not found for key [tx1:0]") + assert.Nil(t, tokens) + assert.Equal(t, testNumRetries, qe.ListAuditTokensCallCount()) + }) + + t.Run("IsPendingErrorIsSurfacedNotMaskedAsPending", func(t *testing.T) { + // When IsPending itself fails (e.g. a store outage), we must not report + // the tx as "still pending" nor burn the retry budget: the underlying + // lookup error and the IsPending error are surfaced immediately. + qe := &mock.QueryEngine{} + id1 := &token.ID{TxId: "tx1", Index: 0} + tokenIDs := []*token.ID{id1} + + qe.ListAuditTokensReturns(nil, errors.New("connection refused")) + qe.IsPendingReturns(false, errors.New("db is down")) + + tokens, err := RetrieveAuditTokens(ctx, logger, qe, tokenIDs, anchor, testNumRetries, testRetryDelay) + require.Error(t, err) + assert.NotContains(t, err.Error(), "still pending") + assert.Contains(t, err.Error(), "connection refused") + assert.Contains(t, err.Error(), "db is down") + assert.Nil(t, tokens) + assert.Equal(t, 1, qe.ListAuditTokensCallCount()) + }) + + t.Run("ContextCancellationInterruptsBackoff", func(t *testing.T) { + // A cancelled context must abort the backoff immediately instead of + // pinning the goroutine for the full grace window. + cancelCtx, cancel := context.WithCancel(ctx) + cancel() + + 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) + + // Use a long delay: if ctx cancellation were ignored the test would hang. + tokens, err := RetrieveAuditTokens(cancelCtx, logger, qe, tokenIDs, anchor, testNumRetries, time.Hour) + require.Error(t, err) + require.ErrorIs(t, err, context.Canceled) + assert.Nil(t, tokens) + assert.Equal(t, 1, qe.ListAuditTokensCallCount()) + }) + + t.Run("ZeroRetriesMakesSingleAttemptAndNeverReturnsNilNil", func(t *testing.T) { + // numRetries <= 0 must be clamped to a single attempt; on failure it must + // return an error (never a nil slice with a nil error, which would panic + // the caller when it indexes the slice). + 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(false, nil) + + tokens, err := RetrieveAuditTokens(ctx, logger, qe, tokenIDs, anchor, 0, testRetryDelay) + require.Error(t, err) + assert.Nil(t, tokens) + assert.Equal(t, 1, qe.ListAuditTokensCallCount()) + }) +} + +// 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 retryDelay 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, DefaultAuditTokensNumRetries, time.Millisecond); 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. + 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, DefaultAuditTokensNumRetries, time.Millisecond); err != nil { + b.Fatal(err) + } + } + }) } func TestValidateStructure(t *testing.T) { diff --git a/token/core/fabtoken/v1/auditor.go b/token/core/fabtoken/v1/auditor.go index ee8af486d0..86ebd93f94 100644 --- a/token/core/fabtoken/v1/auditor.go +++ b/token/core/fabtoken/v1/auditor.go @@ -8,6 +8,7 @@ package v1 import ( "context" + "time" "github.com/LFDT-Panurus/panurus/token/core/common" "github.com/LFDT-Panurus/panurus/token/core/fabtoken/v1/audit" @@ -26,15 +27,27 @@ type AuditorService struct { Deserializer driver.Deserializer QueryEngine driver.QueryEngine tracer trace.Tracer + + // AuditTokensNumRetries and AuditTokensRetryDelay control how AuditorCheck's + // token lookup tolerates the pending-transaction read-timing race (issue #2105). + // They default to common.DefaultAuditTokensNumRetries / DefaultAuditTokensRetryDelay + // and can be tuned per TMS. + AuditTokensNumRetries int + AuditTokensRetryDelay time.Duration } // NewAuditorService returns a new instance of AuditorService. +// +// retryConfig sets the audit-token retry/backoff behavior of AuditorCheck; pass +// common.DefaultAuditRetryConfig() for the built-in defaults or +// common.LoadAuditRetryConfig(tmsConfig) to honor the per-TMS configuration file. func NewAuditorService( logger logging.Logger, publicParametersManager common.PublicParametersManager[*setup.PublicParams], deserializer driver.Deserializer, queryEngine driver.QueryEngine, tracerProvider trace.TracerProvider, + retryConfig common.AuditRetryConfig, ) *AuditorService { return &AuditorService{ Logger: logger, @@ -42,6 +55,8 @@ func NewAuditorService( Deserializer: deserializer, QueryEngine: queryEngine, tracer: tracerProvider.Tracer("auditor_service", tracing.WithMetricsOpts(tracing.MetricsOpts{})), + AuditTokensNumRetries: retryConfig.NumRetries, + AuditTokensRetryDelay: retryConfig.RetryDelay, } } @@ -58,7 +73,7 @@ func (s *AuditorService) AuditorCheck(ctx context.Context, request *driver.Token } // Retrieve audit tokens from the query engine - auditTokens, err := common.RetrieveAuditTokens(ctx, s.Logger, s.QueryEngine, tokenIDs, anchor) + auditTokens, err := common.RetrieveAuditTokens(ctx, s.Logger, s.QueryEngine, tokenIDs, anchor, s.AuditTokensNumRetries, s.AuditTokensRetryDelay) if err != nil { return err } diff --git a/token/core/fabtoken/v1/auditor_test.go b/token/core/fabtoken/v1/auditor_test.go index 55e8f074a3..1626bc0be7 100644 --- a/token/core/fabtoken/v1/auditor_test.go +++ b/token/core/fabtoken/v1/auditor_test.go @@ -11,6 +11,7 @@ import ( "math/big" "testing" + "github.com/LFDT-Panurus/panurus/token/core/common" v1 "github.com/LFDT-Panurus/panurus/token/core/fabtoken/v1" "github.com/LFDT-Panurus/panurus/token/core/fabtoken/v1/actions" "github.com/LFDT-Panurus/panurus/token/core/fabtoken/v1/setup" @@ -111,7 +112,7 @@ func newAuditEnv(benchmarkCase *benchmark2.Case) (*auditEnv, error) { queryEngine := &mockQueryEngine{} tracerProvider := noop.NewTracerProvider() - as := v1.NewAuditorService(logger, publicParamsManager, deserializer, queryEngine, tracerProvider) + as := v1.NewAuditorService(logger, publicParamsManager, deserializer, queryEngine, tracerProvider, common.DefaultAuditRetryConfig()) // Create test data structures issueAction := &actions.IssueAction{ diff --git a/token/core/fabtoken/v1/driver/driver.go b/token/core/fabtoken/v1/driver/driver.go index 5f5c4cc00c..9417d2b03d 100644 --- a/token/core/fabtoken/v1/driver/driver.go +++ b/token/core/fabtoken/v1/driver/driver.go @@ -182,7 +182,7 @@ func (d *Driver) NewTokenService(tmsID driver.TMSID, publicParams []byte) (drive tmsConfig, metrics.NewIssueService(v1.NewIssueService(publicParamsManager, ws, deserializer), metricsProvider), metrics.NewTransferService(v1.NewTransferService(logger, publicParamsManager, ws, common.NewVaultTokenLoader(qe), deserializer), metricsProvider), - metrics.NewAuditorService(v1.NewAuditorService(logger, publicParamsManager, deserializer, qe, d.tracerProvider), metricsProvider), + metrics.NewAuditorService(v1.NewAuditorService(logger, publicParamsManager, deserializer, qe, d.tracerProvider, common.LoadAuditRetryConfig(tmsConfig)), metricsProvider), metrics.NewTokensService(tokensService, metricsProvider), metrics.NewTokensUpgradeService(&v1.TokensUpgradeService{}, metricsProvider), authorization, diff --git a/token/core/zkatdlog/nogh/v1/auditor.go b/token/core/zkatdlog/nogh/v1/auditor.go index 698cfa9b8e..e90ed2e410 100644 --- a/token/core/zkatdlog/nogh/v1/auditor.go +++ b/token/core/zkatdlog/nogh/v1/auditor.go @@ -8,6 +8,7 @@ package v1 import ( "context" + "time" math "github.com/IBM/mathlib" "github.com/LFDT-Panurus/panurus/token/core/common" @@ -26,14 +27,27 @@ type AuditorService struct { Deserializer driver.Deserializer QueryEngine driver.QueryEngine tracer trace.Tracer + + // AuditTokensNumRetries and AuditTokensRetryDelay control how AuditorCheck's + // token lookup tolerates the pending-transaction read-timing race (issue #2105). + // They default to common.DefaultAuditTokensNumRetries / DefaultAuditTokensRetryDelay + // and can be tuned per TMS. + AuditTokensNumRetries int + AuditTokensRetryDelay time.Duration } +// NewAuditorService returns a new instance of AuditorService. +// +// retryConfig sets the audit-token retry/backoff behavior of AuditorCheck; pass +// common.DefaultAuditRetryConfig() for the built-in defaults or +// common.LoadAuditRetryConfig(tmsConfig) to honor the per-TMS configuration file. func NewAuditorService( logger logging.Logger, publicParametersManager common.PublicParametersManager[*setup.PublicParams], deserializer driver.Deserializer, queryEngine driver.QueryEngine, tracerProvider trace.TracerProvider, + retryConfig common.AuditRetryConfig, ) *AuditorService { return &AuditorService{ Logger: logger, @@ -41,6 +55,8 @@ func NewAuditorService( Deserializer: deserializer, QueryEngine: queryEngine, tracer: tracerProvider.Tracer("auditor_service", tracing.WithMetricsOpts(tracing.MetricsOpts{})), + AuditTokensNumRetries: retryConfig.NumRetries, + AuditTokensRetryDelay: retryConfig.RetryDelay, } } @@ -55,7 +71,7 @@ func (s *AuditorService) AuditorCheck(ctx context.Context, request *driver.Token } // Retrieve audit tokens from the query engine - auditTokens, err := common.RetrieveAuditTokens(ctx, s.Logger, s.QueryEngine, tokenIDs, anchor) + auditTokens, err := common.RetrieveAuditTokens(ctx, s.Logger, s.QueryEngine, tokenIDs, anchor, s.AuditTokensNumRetries, s.AuditTokensRetryDelay) if err != nil { return err } diff --git a/token/core/zkatdlog/nogh/v1/benchmark/auditor_setup.go b/token/core/zkatdlog/nogh/v1/benchmark/auditor_setup.go index dc9556ef2d..4c7a00f11a 100644 --- a/token/core/zkatdlog/nogh/v1/benchmark/auditor_setup.go +++ b/token/core/zkatdlog/nogh/v1/benchmark/auditor_setup.go @@ -114,6 +114,7 @@ func NewAuditCheckSetup(conf *SetupConfiguration) (*AuditCheckSetup, error) { deserializer, &mock.QueryEngine{}, noop.NewTracerProvider(), + tokcommon.DefaultAuditRetryConfig(), ) return &AuditCheckSetup{ diff --git a/token/core/zkatdlog/nogh/v1/driver/driver.go b/token/core/zkatdlog/nogh/v1/driver/driver.go index 30a8b9feb0..025383a0a9 100644 --- a/token/core/zkatdlog/nogh/v1/driver/driver.go +++ b/token/core/zkatdlog/nogh/v1/driver/driver.go @@ -203,6 +203,7 @@ func (d *Driver) NewTokenService(tmsID driver.TMSID, publicParams []byte) (drive deserializer, qe, d.tracerProvider, + common.LoadAuditRetryConfig(tmsConfig), ), metricsProvider), metrics.NewTokensService(tokensService, metricsProvider), metrics.NewTokensUpgradeService(tokensUpgradeService, metricsProvider),