From b8c669f1e00c99f35aa0f70c4a7e549affb8f809 Mon Sep 17 00:00:00 2001 From: Michael Buntarman Date: Fri, 3 Jul 2026 18:12:49 +0700 Subject: [PATCH 1/2] chore: stop re-broadcasting permanently-failed market settlements --- .../tn_settlement/internal/engine_ops.go | 72 +++++ .../tn_settlement/internal/engine_ops_test.go | 66 ++++ .../tn_settlement/scheduler/constants.go | 10 + .../tn_settlement/scheduler/scheduler.go | 286 ++++++++++-------- .../tn_settlement/scheduler/scheduler_test.go | 120 +++++++- 5 files changed, 432 insertions(+), 122 deletions(-) diff --git a/extensions/tn_settlement/internal/engine_ops.go b/extensions/tn_settlement/internal/engine_ops.go index e818830a8..5f3cc8aec 100644 --- a/extensions/tn_settlement/internal/engine_ops.go +++ b/extensions/tn_settlement/internal/engine_ops.go @@ -2,6 +2,7 @@ package internal import ( "context" + "errors" "fmt" "strings" "time" @@ -261,6 +262,17 @@ func (e *EngineOperations) BroadcastSettleMarketWithRetry( "tx_hash", hash.String(), "is_nonce_error", isNonceError(err), "error", err) + + // A permanent (deterministic) failure recurs identically on every retry, + // so stop now instead of burning more nonces and committing more failed + // txs to blocks. Returning the wrapped sentinel lets the scheduler + // quarantine the market and flag it for manual intervention. + if isPermanentSettleError(err) { + e.logger.Error("settle_market permanently failed; not retrying (needs manual intervention)", + "query_id", queryID, + "error", err) + return fmt.Errorf("%w (query_id=%d): %w", ErrPermanentSettleFailure, queryID, err) + } } return fmt.Errorf("settle_market failed after %d retries: %w", maxRetries, lastErr) @@ -476,6 +488,66 @@ func isNonceError(err error) bool { return strings.Contains(msg, "nonce") || strings.Contains(msg, "invalid nonce") } +// ErrPermanentSettleFailure marks a settle_market failure that will recur +// identically on every retry. A market's attestation is signed and immutable, so +// a deterministic parse/decode failure of it can never succeed by re-broadcasting +// — the retry only burns nonces and commits another failed tx to a block. The +// scheduler detects this with errors.Is and quarantines the market for manual +// intervention (re-attestation or admin_force_settle_market) instead of retrying. +var ErrPermanentSettleFailure = errors.New("permanent settle_market failure") + +// permanentSettleErrorSignatures are lowercased substrings that identify a +// settle_market revert caused by an unparseable or malformed attestation. These +// come from tn_utils.parse_attestation_boolean / parseBinaryActionResult +// (extensions/tn_utils/precompiles.go) and are deterministic on-chain action +// errors: the signed attestation cannot change, so the same parse fails forever. +// The list is deliberately narrow — only errors that are provably permanent — so +// a transient or unfamiliar failure keeps the existing retry behavior rather than +// being mistakenly quarantined. +var permanentSettleErrorSignatures = []string{ + // Binary-action payload wrong width, e.g. an empty 128-byte result on a binary + // market whose data landed after the attestation was captured (the market-368 + // case): "binary action result must be 32 bytes (abi-encoded bool), got 128". + "binary action result must be", + "abi-encoded bool", + "failed to decode boolean abi result", + "expected 1 value from boolean decode", + "decoded value is not boolean", + // Numeric-action payload (action_id 1-5): parse_attestation_boolean routes a + // numeric-settled market to parseNumericActionResult, which fails + // deterministically on an empty or malformed immutable payload — the same + // late-arriving-data failure mode as market 368, for numeric markets. + "result payload contains no values", + "failed to decode abi result payload", + "expected 2 arrays (timestamps, values)", + "values must be []*big.int", + // Malformed / empty canonical result — an immutable attestation that can never + // be parsed. + "invalid result_canonical", + "result_canonical cannot be empty", + "unsupported action_id", +} + +// isPermanentSettleError reports whether a settle_market broadcast error is a +// deterministic, non-recoverable failure that must not be retried. Nonce and +// network/broadcast errors mean the tx never executed deterministically, so they +// are explicitly excluded and left to the retry path. +func isPermanentSettleError(err error) bool { + if err == nil { + return false + } + if isNonceError(err) { + return false + } + msg := strings.ToLower(err.Error()) + for _, sig := range permanentSettleErrorSignatures { + if strings.Contains(msg, sig) { + return true + } + } + return false +} + // RequestAttestationForMarket broadcasts a request_attestation transaction for a market // with retry logic (exponential backoff for transient errors like nonce conflicts) func (e *EngineOperations) RequestAttestationForMarket( diff --git a/extensions/tn_settlement/internal/engine_ops_test.go b/extensions/tn_settlement/internal/engine_ops_test.go index aefcd0f4a..efa344605 100644 --- a/extensions/tn_settlement/internal/engine_ops_test.go +++ b/extensions/tn_settlement/internal/engine_ops_test.go @@ -693,3 +693,69 @@ func (m *mockReadDBForQueryComponents) Execute(ctx context.Context, stmt string, func (m *mockReadDBForQueryComponents) BeginTx(ctx context.Context) (sql.Tx, error) { return nil, fmt.Errorf("mock read handle does not support transactions") } + +// ============================================================================= +// Test: Permanent-failure classification (infinite-retry fix) +// ============================================================================= + +func TestIsPermanentSettleError(t *testing.T) { + cases := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"market-368 empty binary payload", errors.New("transaction failed with code 65535: binary action result must be 32 bytes (abi-encoded bool), got 128"), true}, + {"wrapped malformed canonical", fmt.Errorf("settle: %w", errors.New("invalid result_canonical: too short for version")), true}, + {"empty canonical", errors.New("result_canonical cannot be empty"), true}, + {"boolean decode failure", errors.New("failed to decode boolean ABI result: bad payload"), true}, + {"numeric market empty payload (368 analog)", errors.New("transaction failed with code 65535: result payload contains no values"), true}, + {"numeric market malformed payload", errors.New("transaction failed with code 65535: failed to decode ABI result payload: bad"), true}, + {"numeric market wrong array count", errors.New("expected 2 arrays (timestamps, values), got 1"), true}, + {"nonce conflict is transient", errors.New("invalid nonce: expected 5 got 4"), false}, + {"network/broadcast is transient", errors.New("broadcast tx: connection refused"), false}, + {"unrelated action revert stays transient (conservative)", errors.New("transaction failed with code 1: insufficient collateral"), false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, isPermanentSettleError(tc.err)) + }) + } +} + +// TestBroadcastSettleMarketWithRetry_PermanentFailureStopsImmediately asserts a +// permanent (deterministic) settle_market failure is attempted exactly once — +// never retried — and returns the ErrPermanentSettleFailure sentinel so the +// scheduler can quarantine the market. This is the nonce-burn / block-spam fix. +func TestBroadcastSettleMarketWithRetry_PermanentFailureStopsImmediately(t *testing.T) { + accounts := &mockAccounts{} + broadcaster := &mockBroadcaster{ + failUntil: 10, // always return the failed result below + successResult: &ktypes.TxResult{ + Code: 65535, + Log: "ERROR: binary action result must be 32 bytes (abi-encoded bool), got 128", + }, + } + + priv, _, err := crypto.GenerateSecp256k1Key(nil) + require.NoError(t, err) + signer := auth.GetUserSigner(priv) + + ops := &EngineOperations{ + logger: log.New(), + accounts: accounts, + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err = ops.BroadcastSettleMarketWithRetry( + ctx, "test-chain", signer, broadcaster.broadcast, + 368, // queryID + 3, // maxRetries — must be ignored for a permanent failure + ) + + require.Error(t, err) + require.ErrorIs(t, err, ErrPermanentSettleFailure) + require.Equal(t, 1, broadcaster.attempts, "a permanent failure must be attempted exactly once (no retries)") +} diff --git a/extensions/tn_settlement/scheduler/constants.go b/extensions/tn_settlement/scheduler/constants.go index 3be1a633e..77f928b81 100644 --- a/extensions/tn_settlement/scheduler/constants.go +++ b/extensions/tn_settlement/scheduler/constants.go @@ -16,3 +16,13 @@ const RetryBackoffInitial = 2 * time.Second // RetryBackoffMax is the maximum backoff duration for retries const RetryBackoffMax = 30 * time.Second + +// PermanentFailureReprobeCooldown is how long a market whose settlement failed +// with a permanent (deterministic) error is quarantined before the scheduler +// re-probes it once. A permanent failure — e.g. a malformed, immutable +// attestation — recurs identically on every attempt, so re-broadcasting it each +// poll only burns nonces and commits failed txs to blocks. Quarantining bounds +// that to at most one failed tx per cooldown per stuck market, while still +// auto-recovering a market an operator repairs (re-attestation): after the +// cooldown, one probe is allowed through, and a success clears the quarantine. +const PermanentFailureReprobeCooldown = 1 * time.Hour diff --git a/extensions/tn_settlement/scheduler/scheduler.go b/extensions/tn_settlement/scheduler/scheduler.go index 35c9de4ab..e83c154a0 100644 --- a/extensions/tn_settlement/scheduler/scheduler.go +++ b/extensions/tn_settlement/scheduler/scheduler.go @@ -2,6 +2,7 @@ package scheduler import ( "context" + "errors" "fmt" "runtime/debug" "sync" @@ -57,6 +58,15 @@ type SettlementScheduler struct { // processingGuard prevents overlapping settlement runs (nonce conflicts) processingGuard ProcessingGuard + + // quarantine tracks query_ids whose settlement permanently failed, each + // mapped to the time after which the scheduler may re-probe it once. Such a + // market is skipped until then so a deterministically-failing settle_market + // tx is not re-broadcast every poll (which burns nonces and commits failed + // txs to blocks). In-memory and leader-local: on restart/leadership change it + // re-probes once and re-quarantines if still permanent. Protected by quarantineMu. + quarantine map[int]time.Time + quarantineMu sync.Mutex } type NewSettlementSchedulerParams struct { @@ -90,7 +100,38 @@ func NewSettlementScheduler(params NewSettlementSchedulerParams) *SettlementSche maxMarketsPerRun: maxMarkets, retryAttempts: retries, processingGuard: params.ProcessingGuard, + quarantine: make(map[int]time.Time), + } +} + +// isQuarantined reports whether a market is currently quarantined after a +// permanent settlement failure. Once past its re-probe deadline it returns false +// so exactly one attempt is allowed through — the entry is kept until that +// attempt either succeeds (clearQuarantine) or fails permanently again +// (quarantineMarket re-arms the cooldown). +func (s *SettlementScheduler) isQuarantined(queryID int) bool { + s.quarantineMu.Lock() + defer s.quarantineMu.Unlock() + reprobeAt, ok := s.quarantine[queryID] + if !ok { + return false } + return time.Now().Before(reprobeAt) +} + +// quarantineMarket flags a market as permanently failing and (re)arms its +// re-probe cooldown. +func (s *SettlementScheduler) quarantineMarket(queryID int) { + s.quarantineMu.Lock() + defer s.quarantineMu.Unlock() + s.quarantine[queryID] = time.Now().Add(PermanentFailureReprobeCooldown) +} + +// clearQuarantine removes a market's quarantine entry (called when it settles). +func (s *SettlementScheduler) clearQuarantine(queryID int) { + s.quarantineMu.Lock() + defer s.quarantineMu.Unlock() + delete(s.quarantine, queryID) } func (s *SettlementScheduler) SetSigner(sig auth.Signer) { @@ -161,92 +202,23 @@ func (s *SettlementScheduler) Start(ctx context.Context, cronExpr string) error "max_markets", maxMarkets, "retry_attempts", retries) - // Query for unsettled markets - markets, err := engineOps.FindUnsettledMarkets(jobCtx, maxMarkets) + settled, failed, skipped, err := s.runSettlementCycle( + jobCtx, engineOps, broadcaster, signer, chainID, maxMarkets, retries) if err != nil { - s.logger.Error("failed to query unsettled markets", "error", err) - return - } - - if len(markets) == 0 { - s.logger.Debug("no unsettled markets found") + // Query error and cancellation are already logged inside the cycle. return } - s.logger.Info("found unsettled markets", "count", len(markets)) - - // Attempt to settle each market - settled := 0 - failed := 0 - skipped := 0 - - for _, market := range markets { - // Check for cancellation - select { - case <-jobCtx.Done(): - s.logger.Info("settlement job cancelled", - "settled", settled, - "failed", failed, - "skipped", skipped) - return - default: - } - - // Check if attestation exists and is signed - hasAttestation, err := engineOps.AttestationExists(jobCtx, market.Hash) - if err != nil { - s.logger.Warn("failed to check attestation", - "query_id", market.ID, - "error", err) - failed++ - continue - } - if !hasAttestation { - // Request attestation for this market - s.logger.Info("attestation not available, requesting attestation", - "query_id", market.ID, - "settle_time", market.SettleTime) - - if err := engineOps.RequestAttestationForMarket(jobCtx, chainID, signer, broadcaster.BroadcastTx, market); err != nil { - s.logger.Warn("failed to request attestation", - "query_id", market.ID, - "error", err) - failed++ - } else { - s.logger.Info("attestation requested successfully", - "query_id", market.ID) - skipped++ // Will settle on next run after signing - } - continue - } - - // Broadcast settle_market transaction with retry - err = engineOps.BroadcastSettleMarketWithRetry( - jobCtx, - chainID, - signer, - broadcaster.BroadcastTx, - market.ID, - retries, - ) - - if err != nil { - s.logger.Warn("failed to settle market after retries", - "query_id", market.ID, - "settle_time", market.SettleTime, - "error", err) - failed++ - } else { - s.logger.Info("market settled successfully", "query_id", market.ID) - settled++ - } + // Only emit the completion summary when markets were actually processed; + // an idle poll already logged a Debug line and would otherwise spam an + // Info line every poll interval. + if settled+failed+skipped > 0 { + s.logger.Info("settlement job completed", + "total_markets", settled+failed+skipped, + "settled", settled, + "failed", failed, + "skipped", skipped) } - - s.logger.Info("settlement job completed", - "total_markets", len(markets), - "settled", settled, - "failed", failed, - "skipped", skipped) } if j, err := s.cron.Cron(cronExpr).Do(jobFunc); err != nil { @@ -277,6 +249,119 @@ func (s *SettlementScheduler) Stop() error { return nil } +// runSettlementCycle performs one settlement pass: find unsettled markets, and +// for each ensure a signed attestation exists then broadcast settle_market. +// Markets whose settlement PERMANENTLY fails (ErrPermanentSettleFailure — e.g. an +// immutable, malformed attestation) are quarantined and skipped on later cycles +// so the scheduler stops re-broadcasting a tx that reverts identically forever +// (which burns nonces and commits failed txs to blocks). Returns per-cycle +// counts; err is non-nil only for a query failure or context cancellation. +func (s *SettlementScheduler) runSettlementCycle( + ctx context.Context, + engineOps EngineOps, + broadcaster txBroadcaster, + signer auth.Signer, + chainID string, + maxMarkets, retries int, +) (settled, failed, skipped int, err error) { + markets, err := engineOps.FindUnsettledMarkets(ctx, maxMarkets) + if err != nil { + s.logger.Error("failed to query unsettled markets", "error", err) + return 0, 0, 0, fmt.Errorf("query unsettled markets: %w", err) + } + + if len(markets) == 0 { + s.logger.Debug("no unsettled markets found") + return 0, 0, 0, nil + } + + s.logger.Info("found unsettled markets", "count", len(markets)) + + for _, market := range markets { + // Check for cancellation (e.g. leadership loss / shutdown) + select { + case <-ctx.Done(): + s.logger.Info("settlement cycle cancelled", + "settled", settled, + "failed", failed, + "skipped", skipped) + return settled, failed, skipped, ctx.Err() + default: + } + + // Skip markets quarantined by a prior permanent failure until their + // re-probe deadline. + if s.isQuarantined(market.ID) { + s.logger.Warn("skipping market quarantined after a permanent settlement failure; awaiting manual intervention", + "query_id", market.ID) + skipped++ + continue + } + + // Check if a signed attestation exists + hasAttestation, attErr := engineOps.AttestationExists(ctx, market.Hash) + if attErr != nil { + s.logger.Warn("failed to check attestation", + "query_id", market.ID, + "error", attErr) + failed++ + continue + } + if !hasAttestation { + // Request attestation for this market + s.logger.Info("attestation not available, requesting attestation", + "query_id", market.ID, + "settle_time", market.SettleTime) + + if reqErr := engineOps.RequestAttestationForMarket(ctx, chainID, signer, broadcaster.BroadcastTx, market); reqErr != nil { + s.logger.Warn("failed to request attestation", + "query_id", market.ID, + "error", reqErr) + failed++ + } else { + s.logger.Info("attestation requested successfully", + "query_id", market.ID) + skipped++ // Will settle on next run after signing + } + continue + } + + // Broadcast settle_market transaction with retry + settleErr := engineOps.BroadcastSettleMarketWithRetry( + ctx, + chainID, + signer, + broadcaster.BroadcastTx, + market.ID, + retries, + ) + switch { + case settleErr == nil: + s.logger.Info("market settled successfully", "query_id", market.ID) + s.clearQuarantine(market.ID) + settled++ + case errors.Is(settleErr, internal.ErrPermanentSettleFailure): + // Deterministic failure — will recur identically. Quarantine so it is + // not re-broadcast every poll, and flag it for manual intervention. + s.quarantineMarket(market.ID) + s.logger.Error("market permanently unsettleable; quarantined and flagged for manual intervention", + "query_id", market.ID, + "settle_time", market.SettleTime, + "reprobe_after", PermanentFailureReprobeCooldown, + "error", settleErr) + failed++ + default: + s.logger.Warn("failed to settle market after retries", + "query_id", market.ID, + "settle_time", market.SettleTime, + "error", settleErr) + failed++ + } + } + + return settled, failed, skipped, nil +} + // RunOnce executes the settlement job payload once (for tests and manual triggering) func (s *SettlementScheduler) RunOnce(ctx context.Context) error { // Check processing guard to prevent overlapping runs @@ -305,41 +390,6 @@ func (s *SettlementScheduler) RunOnce(ctx context.Context) error { } chainID := kwilService.GenesisConfig.ChainID - markets, err := engineOps.FindUnsettledMarkets(ctx, maxMarkets) - if err != nil { - return fmt.Errorf("query unsettled markets: %w", err) - } - - for _, market := range markets { - hasAttestation, err := engineOps.AttestationExists(ctx, market.Hash) - if err != nil { - s.logger.Error("failed to check attestation existence", - "query_id", market.ID, - "error", err) - return fmt.Errorf("check attestation for market %d: %w", market.ID, err) - } - if !hasAttestation { - // Request attestation for this market - if err := engineOps.RequestAttestationForMarket(ctx, chainID, signer, broadcaster.BroadcastTx, market); err != nil { - s.logger.Warn("failed to request attestation in RunOnce", - "query_id", market.ID, - "error", err) - } - continue - } - - err = engineOps.BroadcastSettleMarketWithRetry( - ctx, - chainID, - signer, - broadcaster.BroadcastTx, - market.ID, - retries, - ) - if err != nil { - return fmt.Errorf("settle market %d: %w", market.ID, err) - } - } - - return nil + _, _, _, err := s.runSettlementCycle(ctx, engineOps, broadcaster, signer, chainID, maxMarkets, retries) + return err } diff --git a/extensions/tn_settlement/scheduler/scheduler_test.go b/extensions/tn_settlement/scheduler/scheduler_test.go index 6ae31dcc5..05d81542e 100644 --- a/extensions/tn_settlement/scheduler/scheduler_test.go +++ b/extensions/tn_settlement/scheduler/scheduler_test.go @@ -2,6 +2,7 @@ package scheduler import ( "context" + "fmt" "testing" "time" @@ -93,9 +94,20 @@ func (m *mockSigner) PubKey() crypto.PublicKey { // mockEngineOps implements EngineOps interface for testing. // It signals when methods are called to verify job execution. // It also checks context cancellation to ensure the scheduler uses its own context. +// +// The zero value preserves the lifecycle tests' behavior (no markets, no +// attestation, settle succeeds as a no-op). The optional fields drive the +// settlement-flow tests: markets to return, whether a signed attestation exists, +// the error BroadcastSettleMarketWithRetry returns, and a record of the query_ids +// it was called with (to assert a quarantined market is not re-broadcast). type mockEngineOps struct { t *testing.T onFindUnsettledMarkets func() + + markets []*internal.UnsettledMarket + attestationExists bool + settleErr error + settleCalls []int } func (m *mockEngineOps) FindUnsettledMarkets(ctx context.Context, limit int) ([]*internal.UnsettledMarket, error) { @@ -110,12 +122,11 @@ func (m *mockEngineOps) FindUnsettledMarkets(ctx context.Context, limit int) ([] if m.onFindUnsettledMarkets != nil { m.onFindUnsettledMarkets() } - // Return empty list - no markets to settle - return []*internal.UnsettledMarket{}, nil + return m.markets, nil } func (m *mockEngineOps) AttestationExists(ctx context.Context, marketHash []byte) (bool, error) { - return false, nil + return m.attestationExists, nil } func (m *mockEngineOps) RequestAttestationForMarket(ctx context.Context, chainID string, signer auth.Signer, broadcaster func(context.Context, *ktypes.Transaction, uint8) (ktypes.Hash, *ktypes.TxResult, error), market *internal.UnsettledMarket) error { @@ -123,7 +134,108 @@ func (m *mockEngineOps) RequestAttestationForMarket(ctx context.Context, chainID } func (m *mockEngineOps) BroadcastSettleMarketWithRetry(ctx context.Context, chainID string, signer auth.Signer, broadcaster func(context.Context, *ktypes.Transaction, uint8) (ktypes.Hash, *ktypes.TxResult, error), queryID int, maxRetries int) error { - return nil + m.settleCalls = append(m.settleCalls, queryID) + return m.settleErr +} + +// ============================================================================= +// Test: Permanent settlement failure quarantines the market +// ============================================================================= + +// TestRunSettlementCycle_PermanentFailureQuarantinesMarket asserts the fix for +// the infinite-retry bug: a market whose settlement fails permanently +// (ErrPermanentSettleFailure, e.g. a malformed immutable attestation) is +// attempted once, quarantined, then SKIPPED on the next cycle instead of being +// re-broadcast every poll (which burned nonces and spammed failed txs to blocks). +func TestRunSettlementCycle_PermanentFailureQuarantinesMarket(t *testing.T) { + broadcaster := &mockTxBroadcaster{} + signer := &mockSigner{} + + mockOps := &mockEngineOps{ + markets: []*internal.UnsettledMarket{{ID: 368, Hash: []byte{0xab}, SettleTime: 1}}, + attestationExists: true, + settleErr: fmt.Errorf("%w: transaction failed with code 65535: binary action result must be 32 bytes (abi-encoded bool), got 128", + internal.ErrPermanentSettleFailure), + } + + s := NewSettlementScheduler(NewSettlementSchedulerParams{ + Service: &common.Service{Logger: log.New()}, + Logger: log.New(), + EngineOps: mockOps, + Tx: broadcaster, + Signer: signer, + MaxMarketsPerRun: 10, + RetryAttempts: 3, + }) + + ctx := context.Background() + + // Cycle 1: 368 fails permanently — attempted exactly once, then quarantined. + settled, failed, skipped, err := s.runSettlementCycle(ctx, mockOps, broadcaster, signer, "test-chain", 10, 3) + if err != nil { + t.Fatalf("cycle 1 unexpected error: %v", err) + } + if settled != 0 || failed != 1 || skipped != 0 { + t.Fatalf("cycle 1 counts = settled=%d failed=%d skipped=%d; want 0/1/0", settled, failed, skipped) + } + if len(mockOps.settleCalls) != 1 || mockOps.settleCalls[0] != 368 { + t.Fatalf("cycle 1 expected exactly one settle attempt for 368, got %v", mockOps.settleCalls) + } + if !s.isQuarantined(368) { + t.Fatal("market 368 should be quarantined after a permanent failure") + } + + // Cycle 2: 368 is quarantined — skipped, and NOT re-broadcast. + settled, failed, skipped, err = s.runSettlementCycle(ctx, mockOps, broadcaster, signer, "test-chain", 10, 3) + if err != nil { + t.Fatalf("cycle 2 unexpected error: %v", err) + } + if settled != 0 || failed != 0 || skipped != 1 { + t.Fatalf("cycle 2 counts = settled=%d failed=%d skipped=%d; want 0/0/1", settled, failed, skipped) + } + if len(mockOps.settleCalls) != 1 { + t.Fatalf("cycle 2 must not re-broadcast a quarantined market; settleCalls=%v", mockOps.settleCalls) + } +} + +// TestRunSettlementCycle_SuccessClearsQuarantine asserts a market that later +// settles (e.g. after an operator re-attests) has its quarantine cleared. +func TestRunSettlementCycle_SuccessClearsQuarantine(t *testing.T) { + broadcaster := &mockTxBroadcaster{} + signer := &mockSigner{} + + mockOps := &mockEngineOps{ + markets: []*internal.UnsettledMarket{{ID: 42, Hash: []byte{0xcd}, SettleTime: 1}}, + attestationExists: true, + } + + s := NewSettlementScheduler(NewSettlementSchedulerParams{ + Service: &common.Service{Logger: log.New()}, + Logger: log.New(), + EngineOps: mockOps, + Tx: broadcaster, + Signer: signer, + MaxMarketsPerRun: 10, + RetryAttempts: 3, + }) + + // Pre-quarantine the market, then let it settle successfully after the + // re-probe deadline by clearing the cooldown (simulate cooldown elapsed). + s.quarantineMarket(42) + s.quarantineMu.Lock() + s.quarantine[42] = time.Now().Add(-time.Minute) // deadline in the past → re-probe allowed + s.quarantineMu.Unlock() + + settled, failed, skipped, err := s.runSettlementCycle(context.Background(), mockOps, broadcaster, signer, "test-chain", 10, 3) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if settled != 1 || failed != 0 || skipped != 0 { + t.Fatalf("counts = settled=%d failed=%d skipped=%d; want 1/0/0", settled, failed, skipped) + } + if s.isQuarantined(42) { + t.Fatal("a successful settlement must clear the quarantine entry") + } } // ============================================================================= From 32226ca9b39839c5db1781f190faaaa50d803a5f Mon Sep 17 00:00:00 2001 From: Michael Buntarman Date: Fri, 3 Jul 2026 18:55:39 +0700 Subject: [PATCH 2/2] chore: settle newer markets even when older ones are quarantined --- .../tn_settlement/scheduler/scheduler.go | 31 +++++++++++- .../tn_settlement/scheduler/scheduler_test.go | 50 ++++++++++++++++++- 2 files changed, 78 insertions(+), 3 deletions(-) diff --git a/extensions/tn_settlement/scheduler/scheduler.go b/extensions/tn_settlement/scheduler/scheduler.go index e83c154a0..cfcef39dd 100644 --- a/extensions/tn_settlement/scheduler/scheduler.go +++ b/extensions/tn_settlement/scheduler/scheduler.go @@ -134,6 +134,16 @@ func (s *SettlementScheduler) clearQuarantine(queryID int) { delete(s.quarantine, queryID) } +// quarantineCount returns the number of quarantine entries. It bounds how many +// quarantined rows a settlement query might return, so runSettlementCycle can +// over-fetch by this amount and still process up to maxMarkets non-quarantined +// markets. +func (s *SettlementScheduler) quarantineCount() int { + s.quarantineMu.Lock() + defer s.quarantineMu.Unlock() + return len(s.quarantine) +} + func (s *SettlementScheduler) SetSigner(sig auth.Signer) { s.mu.Lock() s.signer = sig @@ -264,7 +274,12 @@ func (s *SettlementScheduler) runSettlementCycle( chainID string, maxMarkets, retries int, ) (settled, failed, skipped int, err error) { - markets, err := engineOps.FindUnsettledMarkets(ctx, maxMarkets) + // Over-fetch by the number of quarantined markets. FindUnsettledMarkets returns + // oldest-first, so a quarantined older market would otherwise fill a limited page + // ahead of a newer, settleable one and starve it. Requesting maxMarkets plus the + // quarantine count guarantees the page still holds up to maxMarkets non-quarantined + // markets (len(quarantine) is a safe upper bound on how many rows we may skip). + markets, err := engineOps.FindUnsettledMarkets(ctx, maxMarkets+s.quarantineCount()) if err != nil { s.logger.Error("failed to query unsettled markets", "error", err) return 0, 0, 0, fmt.Errorf("query unsettled markets: %w", err) @@ -277,7 +292,17 @@ func (s *SettlementScheduler) runSettlementCycle( s.logger.Info("found unsettled markets", "count", len(markets)) + // processed counts the non-quarantined markets handled this cycle — it is the + // maxMarkets budget. Quarantine skips do NOT consume it, so a quarantined market + // cannot crowd out settleable ones. + processed := 0 for _, market := range markets { + // Stop once the budget of non-quarantined markets is met; any remaining + // fetched rows are only the over-fetch headroom. + if processed >= maxMarkets { + break + } + // Check for cancellation (e.g. leadership loss / shutdown) select { case <-ctx.Done(): @@ -290,7 +315,7 @@ func (s *SettlementScheduler) runSettlementCycle( } // Skip markets quarantined by a prior permanent failure until their - // re-probe deadline. + // re-probe deadline. A quarantine skip does not consume the budget. if s.isQuarantined(market.ID) { s.logger.Warn("skipping market quarantined after a permanent settlement failure; awaiting manual intervention", "query_id", market.ID) @@ -298,6 +323,8 @@ func (s *SettlementScheduler) runSettlementCycle( continue } + processed++ + // Check if a signed attestation exists hasAttestation, attErr := engineOps.AttestationExists(ctx, market.Hash) if attErr != nil { diff --git a/extensions/tn_settlement/scheduler/scheduler_test.go b/extensions/tn_settlement/scheduler/scheduler_test.go index 05d81542e..775111e49 100644 --- a/extensions/tn_settlement/scheduler/scheduler_test.go +++ b/extensions/tn_settlement/scheduler/scheduler_test.go @@ -122,7 +122,12 @@ func (m *mockEngineOps) FindUnsettledMarkets(ctx context.Context, limit int) ([] if m.onFindUnsettledMarkets != nil { m.onFindUnsettledMarkets() } - return m.markets, nil + // Mirror the real query's LIMIT: oldest-first, truncated to `limit`. + markets := m.markets + if limit >= 0 && len(markets) > limit { + markets = markets[:limit] + } + return markets, nil } func (m *mockEngineOps) AttestationExists(ctx context.Context, marketHash []byte) (bool, error) { @@ -238,6 +243,49 @@ func TestRunSettlementCycle_SuccessClearsQuarantine(t *testing.T) { } } +// TestRunSettlementCycle_QuarantinedMarketDoesNotStarveNewer asserts a quarantined +// older market does not consume the maxMarkets budget and starve a newer eligible +// market. With maxMarkets=1 and the (oldest-first) page truncating to the limit, the +// naive fetch would return only the quarantined market and never reach the newer one; +// over-fetching by the quarantine count keeps the newer market settleable. +func TestRunSettlementCycle_QuarantinedMarketDoesNotStarveNewer(t *testing.T) { + broadcaster := &mockTxBroadcaster{} + signer := &mockSigner{} + + older := &internal.UnsettledMarket{ID: 1, Hash: []byte{0x01}, SettleTime: 100} + newer := &internal.UnsettledMarket{ID: 2, Hash: []byte{0x02}, SettleTime: 200} + + mockOps := &mockEngineOps{ + markets: []*internal.UnsettledMarket{older, newer}, // oldest-first + attestationExists: true, + } + + s := NewSettlementScheduler(NewSettlementSchedulerParams{ + Service: &common.Service{Logger: log.New()}, + Logger: log.New(), + EngineOps: mockOps, + Tx: broadcaster, + Signer: signer, + MaxMarketsPerRun: 1, + RetryAttempts: 3, + }) + + // The older market was quarantined by a prior permanent failure. + s.quarantineMarket(older.ID) + + settled, failed, skipped, err := s.runSettlementCycle(context.Background(), mockOps, broadcaster, signer, "test-chain", 1, 3) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if settled != 1 || failed != 0 || skipped != 1 { + t.Fatalf("counts = settled=%d failed=%d skipped=%d; want 1/0/1 (older skipped, newer settled)", settled, failed, skipped) + } + // Only the newer market may be settled; the quarantined older one must be skipped. + if len(mockOps.settleCalls) != 1 || mockOps.settleCalls[0] != newer.ID { + t.Fatalf("expected exactly the newer market (%d) settled, got settleCalls=%v", newer.ID, mockOps.settleCalls) + } +} + // ============================================================================= // Test: Scheduler Start/Stop // =============================================================================