feat(requester): auto-heal stuck nonce markers via reconciliation loop

README.md

@@ -271,6 +271,8 @@ The application can be configured using the following flags at runtime:
 | `tx-pool-ttl`                  | `30s`            | How long the transaction mempool holds an out-of-order transaction waiting for its nonce gap to fill, before submitting it anyway. Only applies when `tx-mempool-mode=true`.                                       |
 | `tx-max-batch-size`            | `5`              | Maximum number of EVM transactions per `EVM.batchRun` Cadence transaction in the transaction mempool. Only applies when `tx-mempool-mode=true`.                                                                        |
 | `tx-max-nonce-gap`             | `500`            | How far ahead of an EOA's on-chain nonce the transaction mempool accepts a nonce; nonces beyond `indexedNonce + gap` are rejected as nonce-too-high. `0` disables the upper bound. A nonce below the indexed nonce is always rejected as nonce-too-low. Only applies when `tx-mempool-mode=true`. |
+| `tx-reconcile-interval`        | `1s`             | How often the mempool reconciliation loop polls each active EOA's most recent Cadence tx wrapper. Only applies when `tx-mempool-mode=true`.                                                                                                              |
+| `tx-reconcile-stale-after`     | `30s`            | Grace period past a wrapper's submission time after which, if it still is not sealed, the reconciler treats it as dropped and clears the EOA's in-flight marker. Must be >> Flow sealing latency (~6-8s). Only applies when `tx-mempool-mode=true`.       |
 
 
 # EVM Gateway Endpoints

cmd/run/cmd.go

@@ -252,6 +252,12 @@ func parseConfigFromFlags() error {
 		if cfg.TxMaxBatchSize < 1 {
 			return fmt.Errorf("tx-max-batch-size must be >= 1 when tx-mempool-mode is enabled")
 		}
+		if cfg.TxReconcileInterval <= 0 {
+			return fmt.Errorf("tx-reconcile-interval must be > 0 when tx-mempool-mode is enabled")
+		}
+		if cfg.TxReconcileStaleAfter <= 0 {
+			return fmt.Errorf("tx-reconcile-stale-after must be > 0 when tx-mempool-mode is enabled")
+		}
 	}
 
 	if !cfg.ExperimentalSoftFinalityEnabled && cfg.ExperimentalSealingVerificationEnabled {
@@ -330,6 +336,8 @@ func init() {
 	Cmd.Flags().DurationVar(&cfg.TxPoolTTL, "tx-pool-ttl", 30*time.Second, "How long the transaction mempool holds an out-of-order transaction waiting for its nonce gap to fill, before submitting it anyway.")
 	Cmd.Flags().IntVar(&cfg.TxMaxBatchSize, "tx-max-batch-size", 5, "Maximum number of EVM transactions per EVM.batchRun Cadence transaction in the transaction mempool.")
 	Cmd.Flags().Uint64Var(&cfg.TxMaxNonceGap, "tx-max-nonce-gap", 500, "How far ahead of an EOA's on-chain nonce the transaction mempool accepts a nonce; nonces beyond indexedNonce+gap are rejected as nonce-too-high. 0 means no upper bound. A nonce below the indexed nonce is always rejected as nonce-too-low regardless of this setting.")
+	Cmd.Flags().DurationVar(&cfg.TxReconcileInterval, "tx-reconcile-interval", time.Second, "How often the mempool reconciliation loop polls each active EOA's most recent Cadence tx wrapper. Only used when --tx-mempool-mode=true.")
+	Cmd.Flags().DurationVar(&cfg.TxReconcileStaleAfter, "tx-reconcile-stale-after", 30*time.Second, "How long to wait for a submitted wrapping Cadence tx to seal before the reconciliation loop treats it as dropped and clears the EOA's in-flight marker. Must be >> Flow sealing latency.")
 	Cmd.Flags().DurationVar(&cfg.RpcRequestTimeout, "rpc-request-timeout", time.Second*120, "Sets the maximum duration at which JSON-RPC requests should generate a response, before they timeout. The default is 120 seconds.")
 
 	err := Cmd.Flags().MarkDeprecated("init-cadence-height", "This flag is no longer necessary and will be removed in future version. The initial Cadence height is known for testnet/mainnet and this was only required for fresh deployments of EVM Gateway. Once the DB has been initialized, the latest index Cadence height will be used upon start-up.")

config/config.go

@@ -164,6 +164,16 @@ type Config struct {
 	// bounds only the upper end: a nonce below the indexed nonce is always
 	// rejected with ErrNonceTooLow regardless of this setting.
 	TxMaxNonceGap uint64
+	// TxReconcileInterval is how often the mempool reconciliation loop wakes to
+	// poll each active EOA's last-submitted wrapping Cadence tx status against the
+	// chain. A stale marker (reverted wrapper or silent drop) is cleared here so
+	// the wedge does not persist until idle-eviction.
+	TxReconcileInterval time.Duration
+	// TxReconcileStaleAfter is the grace period past a wrapper's submission time
+	// after which, if it still is not sealed, the reconciler treats it as dropped
+	// and clears the marker. Must comfortably exceed Flow's sealing latency
+	// (~6-8s) to avoid false positives.
+	TxReconcileStaleAfter time.Duration
 	// RpcRequestTimeout is the maximum duration at which JSON-RPC requests should generate
 	// a response, before they timeout.
 	RpcRequestTimeout time.Duration

metrics/collector.go

@@ -120,6 +120,11 @@ var txPoolNonceViewCache = prometheus.NewCounterVec(prometheus.CounterOpts{
 	Help: "Block-view cache accesses when reading EOA nonces (hit = reused the view built for the indexed height; miss = rebuilt it)",
 }, []string{"result"})
 
+var txPoolReconcileResets = prometheus.NewCounterVec(prometheus.CounterOpts{
+	Name: prefixedName("txpool_reconcile_resets_total"),
+	Help: "Total mempool in-flight nonce markers cleared by the reconciliation loop, by reason (wrapper-reverted, unsealed-past-threshold)",
+}, []string{"reason"})
+
 var metrics = []prometheus.Collector{
 	apiErrors,
 	serverPanicsCounters,
@@ -142,6 +147,7 @@ var metrics = []prometheus.Collector{
 	txPoolQueuedTransactions,
 	txPoolSubmissions,
 	txPoolNonceViewCache,
+	txPoolReconcileResets,
 }
 
 type Collector interface {
@@ -164,6 +170,7 @@ type Collector interface {
 	TxPoolSize(queues int, queuedTransactions int)
 	TxPoolSubmission(reason string)
 	NonceViewCache(hit bool)
+	TxPoolReconcileReset(reason string)
 }
 
 var _ Collector = &DefaultCollector{}
@@ -193,6 +200,7 @@ type DefaultCollector struct {
 	txPoolQueuedTransactions       prometheus.Gauge
 	txPoolSubmissions              *prometheus.CounterVec
 	txPoolNonceViewCache           *prometheus.CounterVec
+	txPoolReconcileResets          *prometheus.CounterVec
 }
 
 func NewCollector(logger zerolog.Logger) Collector {
@@ -224,6 +232,7 @@ func NewCollector(logger zerolog.Logger) Collector {
 		txPoolQueuedTransactions:       txPoolQueuedTransactions,
 		txPoolSubmissions:              txPoolSubmissions,
 		txPoolNonceViewCache:           txPoolNonceViewCache,
+		txPoolReconcileResets:          txPoolReconcileResets,
 	}
 }
 
@@ -340,6 +349,10 @@ func (c *DefaultCollector) NonceViewCache(hit bool) {
 	c.txPoolNonceViewCache.With(prometheus.Labels{"result": result}).Inc()
 }
 
+func (c *DefaultCollector) TxPoolReconcileReset(reason string) {
+	c.txPoolReconcileResets.With(prometheus.Labels{"reason": reason}).Inc()
+}
+
 func prefixedName(name string) string {
 	return fmt.Sprintf("evm_gateway_%s", name)
 }

metrics/nop.go

@@ -30,3 +30,4 @@ func (c *nopCollector) FlowTotalSupply(totalSupply *big.Int)       {}
 func (c *nopCollector) TxPoolSize(queues int, queued int)          {}
 func (c *nopCollector) TxPoolSubmission(reason string)             {}
 func (c *nopCollector) NonceViewCache(hit bool)                    {}
+func (c *nopCollector) TxPoolReconcileReset(reason string)         {}

services/requester/tx_mempool.go

@@ -100,6 +100,36 @@ import (
 //       Example: frontier 5, TxMaxNonceGap 500, tx with nonce 600 arrives →
 //       rejected (it cannot execute until ~595 intervening nonces are filled).
 //
+// Recovery
+//   13. Reconciliation: a background loop (reconcileLoop) ticks every
+//       TxReconcileInterval (default 1s). For each EOA with an outstanding
+//       submission marker (highestSent() set and lastFlowTxID != zero) it
+//       calls GetTransactionResult(lastFlowTxID) and resets the marker in
+//       two cases:
+//         (a) the wrapper is SEALED with a non-nil Error — the wrapper
+//             reverted. Canonical case: two consecutive-nonce wrappers land
+//             in the same Flow block, the collector executes them out of
+//             order, and the higher-nonce wrapper's run.cdc assertion trips
+//             with "nonce too high". No EVM.TransactionExecuted event fires
+//             for the reverted wrapper, so without reconciliation the pool
+//             would never learn.
+//         (b) the wrapper is not sealed and now - lastSubmittedAt exceeds
+//             TxReconcileStaleAfter (default 30s) — probable silent drop or
+//             any never-lands case.
+//       The reset clears lastConsecutivelySubmitted, submitting, and
+//       lastFlowTxID so the next Add() re-classifies against on-chain state.
+//       Concurrency: the (eoa, flowTxID, lastSubmittedAt) snapshot is taken
+//       under queueMux, the GetTransactionResult call runs OUTSIDE the lock,
+//       and the reset re-acquires the lock and (i) re-checks lastFlowTxID
+//       still matches and (ii) requires q.nonces.inFlight() to be false — so
+//       neither a superseding ack'd submission nor a freshly-in-flight one
+//       is ever clobbered.
+//       Observability: each reset emits a WARN log line with eoa,
+//       flow_tx_id and reason ("wrapper-reverted" | "unsealed-past-threshold")
+//       and increments the TxPoolReconcileReset counter. Wedge duration is
+//       bounded to ~one sealing window (~6-8s) instead of the full
+//       idleQueueRetention (60s).
+//
 // Cross-cutting invariants
 //   - No silent drops: for any accepted tx id you can either find it on-chain
 //     (submitted) or find a WARN log saying it was dropped (submit failure or
@@ -110,7 +140,9 @@ import (
 //     and never wedges the EOA (the in-flight marker is rolled back). The pool
 //     does NOT retry internally.
 //   - Concurrency: one background goroutine (processQueues) flushes due queues
-//     and Add runs under the same pool-wide queueMux; see the note on TxMemPool.
+//     and Add runs under the same pool-wide queueMux; a second goroutine
+//     (reconcileLoop) polls Cadence tx status outside the lock and only
+//     acquires it briefly to reset stuck markers. See the note on TxMemPool.
 
 // heldTx is a transaction held in the mempool, waiting for its
 // collection window to elapse or its nonce gap to be filled.
@@ -194,6 +226,18 @@ const fastPathSubmitTimeout = 10 * time.Second
 // recent activity is kept before being removed, to bound memory usage.
 const idleQueueRetention = time.Minute
 
+// defaultTxReconcileInterval and defaultTxReconcileStaleAfter are the fallback
+// values used by NewTxMemPool when the config leaves them at zero. This
+// protects programmatic callers (e.g. e2e tests constructing a Config directly)
+// from the time.NewTicker(0) panic and from a zero staleness threshold that
+// would treat every unsealed tx as instantly stale. The CLI-flag defaults in
+// cmd/run/cmd.go match these — the double-source-of-truth is intentional so
+// both flag-driven and Go-driven constructors behave sanely.
+const (
+	defaultTxReconcileInterval   = time.Second
+	defaultTxReconcileStaleAfter = 30 * time.Second
+)
+
 // nonceWrapper is a nonce that may be unset (set == false). It disambiguates the
 // otherwise ambiguous value 0, which is both a valid nonce and the zero value.
 // An unset nonceWrapper behaves as -∞ in the comparisons below (atLeast/is/max):
@@ -410,6 +454,16 @@ func (n *nonceTracker) refreshNextNonce(nextNonce uint64) {
 	n.localNextNonce = nextNonce
 }
 
+// resetSubmissionState clears both submission markers so subsequent Add() calls
+// re-classify against on-chain state. Called by the reconciliation loop when
+// the wrapping Cadence tx demonstrably did not advance the on-chain nonce
+// (reverted, or unsealed past the stale-after threshold). The corresponding
+// eoaQueue's lastFlowTxID must be cleared alongside this call.
+func (n *nonceTracker) resetSubmissionState() {
+	n.lastConsecutivelySubmitted = nonceWrapper{}
+	n.submitting = nonceWrapper{}
+}
+
 // eoaQueue tracks the held transactions and submission state for one EOA.
 type eoaQueue struct {
 	// txs holds pending transactions keyed by nonce. Keying by nonce gives
@@ -434,6 +488,11 @@ type eoaQueue struct {
 	lastActivity time.Time
 	// nonces is the submission-state machine for this EOA.
 	nonces nonceTracker
+	// lastFlowTxID is the Flow transaction ID of the most recent Cadence submission
+	// for this EOA. Zero until the first successful submission. Read by the
+	// reconciliation loop to poll the wrapper's on-chain status; rolled back to
+	// zero when reconciliation detects the wrapper reverted or never sealed.
+	lastFlowTxID flow.Identifier
 }
 
 // isEmpty reports whether the queue holds no transactions. Callers must hold
@@ -474,6 +533,10 @@ type TxMemPool struct {
 	// before a Flow tx is signed) so logSubmission can record it, letting an
 	// operator correlate a wedged EVM nonce to the specific Cadence tx.
 	submitBatch func(ctx context.Context, txs []heldTx) (flow.Identifier, error)
+	// getTxResult retrieves the sealed status of a Cadence transaction. It defaults
+	// to t.client.GetTransactionResult and exists as a field so tests can inject a
+	// fake without a live Access Node.
+	getTxResult func(ctx context.Context, id flow.Identifier) (*flow.TransactionResult, error)
 	// now returns the current time. It defaults to time.Now and exists as a
 	// field so tests can drive the collection window, flush deadline, submission
 	// spacing, TTL expiry and idle-queue retention with a controllable clock
@@ -507,8 +570,19 @@ func NewTxMemPool(
 		now:           time.Now,
 	}
 	pool.submitBatch = pool.submitTxBatch
+	pool.getTxResult = pool.client.GetTransactionResult
+
+	// Backfill reconcile-loop knobs when a programmatic caller leaves them at
+	// zero. Also protects against time.NewTicker(0) which panics.
+	if pool.config.TxReconcileInterval <= 0 {
+		pool.config.TxReconcileInterval = defaultTxReconcileInterval
+	}
+	if pool.config.TxReconcileStaleAfter <= 0 {
+		pool.config.TxReconcileStaleAfter = defaultTxReconcileStaleAfter
+	}
 
 	go pool.processQueues(ctx)
+	go pool.reconcileLoop(ctx)
 
 	return pool, nil
 }
@@ -626,6 +700,7 @@ func (t *TxMemPool) Add(
 			}
 			q.nonces.markSubmitted(tx.Nonce())
 			q.lastSubmittedAt = t.now()
+			q.lastFlowTxID = flowTxID
 			return nil
 		}
 		t.enqueue(q, held, now)
@@ -735,7 +810,7 @@ func (t *TxMemPool) processQueues(ctx context.Context) {
 func (t *TxMemPool) submitWork(ctx context.Context, w flushWork) error {
 	flowTxID, err := t.submitBatch(ctx, w.txs)
 	t.logSubmission(w.from, w.txs, w.reason, w.localNextNonce, flowTxID, err)
-	t.reconcileSubmission(w, err)
+	t.reconcileSubmission(w, flowTxID, err)
 	return err
 }
 
@@ -852,7 +927,12 @@ func (t *TxMemPool) logSubmission(
 //
 // TTL-expiry batches (w.needsReconcile == false) never mark the tracker, so
 // there is nothing to reconcile for them.
-func (t *TxMemPool) reconcileSubmission(w flushWork, submitErr error) {
+//
+// flowTxID is recorded on the queue when submission succeeds so the
+// reconciliation loop can later poll the wrapping Cadence tx status against the
+// chain (see reconcileLoop). It is ignored on failure and for batches that
+// don't require reconciliation.
+func (t *TxMemPool) reconcileSubmission(w flushWork, flowTxID flow.Identifier, submitErr error) {
 	if !w.needsReconcile {
 		return
 	}
@@ -871,6 +951,7 @@ func (t *TxMemPool) reconcileSubmission(w flushWork, submitErr error) {
 		return
 	}
 	q.nonces.markSubmitted(highNonce)
+	q.lastFlowTxID = flowTxID
 }
 
 // collectDueBatches selects, under the queue lock, every batch that is due for
@@ -1114,3 +1195,118 @@ func (t *TxMemPool) submitTxBatch(ctx context.Context, txs []heldTx) (flow.Ident
 
 	return flowTx.ID(), nil
 }
+
+// reconcileLoop periodically inspects each active EOA queue's most recent
+// wrapping Cadence transaction and clears the in-flight nonce marker when the
+// wrapper is provably not going to advance the on-chain nonce. Two cases
+// trigger a reset:
+//  1. The wrapper is SEALED with a non-nil Error (it reverted — e.g. the
+//     intra-block reordering "nonce too high" assertion in run.cdc). This is
+//     the primary DFNS-observed failure mode.
+//  2. The wrapper is not sealed and more than TxReconcileStaleAfter has
+//     elapsed since submission. Catches silent AN drops and any other
+//     never-lands case.
+//
+// The "chain advanced past highestSent" case is not handled here on purpose:
+// if the wrapper actually landed, it seals successfully and the next
+// legitimate submission moves lastConsecutivelySubmitted forward on its own;
+// there is no wedge to clear.
+func (t *TxMemPool) reconcileLoop(ctx context.Context) {
+	ticker := time.NewTicker(t.config.TxReconcileInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			t.reconcileOnce(ctx)
+		}
+	}
+}
+
+// reconcileOnce performs one pass of reconciliation across all active EOA
+// queues. Callers must NOT hold queueMux — this method acquires it in short
+// sections around each EOA's read/reset, and the network call (getTxResult)
+// happens outside the lock so a slow AN cannot pin the pool.
+func (t *TxMemPool) reconcileOnce(ctx context.Context) {
+	// Snapshot the (eoa, flowTxID, lastSubmittedAt) tuples for EOAs with an
+	// outstanding submission marker. Copy under the lock so we can release it
+	// before doing network I/O.
+	type snapshot struct {
+		from            gethCommon.Address
+		flowTxID        flow.Identifier
+		lastSubmittedAt time.Time
+	}
+	var snaps []snapshot
+	t.queueMux.Lock()
+	for from, q := range t.queues {
+		if !q.nonces.highestSent().set {
+			continue
+		}
+		if q.lastFlowTxID == (flow.Identifier{}) {
+			continue
+		}
+		snaps = append(snaps, snapshot{from, q.lastFlowTxID, q.lastSubmittedAt})
+	}
+	t.queueMux.Unlock()
+
+	now := t.now()
+	for _, s := range snaps {
+		getTxResultCtx, cancel := context.WithTimeout(ctx, t.config.TxReconcileInterval)
+		result, err := t.getTxResult(getTxResultCtx, s.flowTxID)
+		cancel()
+		// Fall through: even on error we may still want to check the staleness
+		// path below. But avoid touching state on transient AN errors — only
+		// reset if we have concrete evidence (SEALED-with-error) OR the
+		// staleness threshold is exceeded.
+		sealed := err == nil && result != nil && result.Status == flow.TransactionStatusSealed
+		reverted := sealed && result.Error != nil
+		stale := (!sealed || result.Error != nil) && now.Sub(s.lastSubmittedAt) > t.config.TxReconcileStaleAfter
+
+		if !reverted && !stale {
+			continue
+		}
+
+		t.queueMux.Lock()
+		q, ok := t.queues[s.from]
+		// If the queue was evicted or already advanced (different flow_tx_id
+		// now), do nothing — a fresher submission has superseded this state.
+		if !ok || q.lastFlowTxID != s.flowTxID {
+			t.queueMux.Unlock()
+			continue
+		}
+		// A newer batch entered flight between our snapshot and this reset.
+		// lastFlowTxID is only advanced together with markSubmitted (in
+		// reconcileSubmission), so a matching lastFlowTxID means the batch that
+		// set it has already returned; any q.nonces.submitting we see now must
+		// belong to a strictly newer batch. Clobbering its submitting marker
+		// would let a client retry duplicate a nonce that is legitimately in
+		// flight — reintroducing the very failure mode this loop exists to
+		// prevent. Skip and let the next tick handle whichever wrapper needs it.
+		if q.nonces.inFlight() {
+			t.queueMux.Unlock()
+			continue
+		}
+
+		reason := "unsealed-past-threshold"
+		if reverted {
+			reason = "wrapper-reverted"
+		}
+		elapsed := now.Sub(s.lastSubmittedAt)
+		event := t.logger.Warn().
+			Str("eoa", s.from.Hex()).
+			Str("flow_tx_id", s.flowTxID.Hex()).
+			Str("reason", reason).
+			Dur("elapsed-since-submit", elapsed)
+
+		if reverted {
+			event = event.Str("wrapper-error", result.Error.Error())
+		}
+		event.Msg("reconciliation clearing stuck in-flight marker; subsequent Add() calls will re-classify against chain")
+
+		q.nonces.resetSubmissionState()
+		q.lastFlowTxID = flow.Identifier{}
+		t.collector.TxPoolReconcileReset(reason)
+		t.queueMux.Unlock()
+	}
+}