From ee2d4fbbceddc017c31d9c900106b6ec8c9bb384 Mon Sep 17 00:00:00 2001
From: vishal <1117327+vishalchangrani@users.noreply.github.com>
Date: Wed, 29 Jul 2026 11:20:11 -0400
Subject: [PATCH 01/10] feat(requester): add mempool reconciliation loop to
 auto-heal stuck nonce markers
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adds a background reconciliation loop to TxMemPool that polls the most recent
wrapping Cadence tx for each active EOA and clears the in-flight nonce marker
when the wrapper cannot advance the on-chain nonce. Bounds wedge duration to
one sealing window (~6-8s) instead of the full idleQueueRetention (~1 minute).

Motivation: the DFNS silent-drop incident (2026-07-29T03:09:11Z, EOA
0xdEA9...58B3, nonce 25882) showed the wrapping Cadence tx sealing with
evm_error=nonce too high after intra-block Collection Node reordering. In
that case the mempool has already advanced `lastConsecutivelySubmitted` on
successful send, so subsequent retries got ErrInFlightNonce until the
idle-eviction retention expired — that is the wedge this loop closes.

Implementation:
  * eoaQueue gains lastFlowTxID, set on every successful submission (fast
    path in Add, and reconcileSubmission after a background submitWork).
  * reconcileSubmission now takes the flowTxID so it can persist it on the
    queue alongside markSubmitted (existing behavior otherwise unchanged).
  * nonceTracker.resetSubmissionState clears both submission markers so the
    next Add re-classifies against on-chain state.
  * reconcileLoop ticks at TxReconcileInterval, snapshots per-EOA
    (flowTxID, lastSubmittedAt) under the lock, calls GetTransactionResult
    outside the lock, and resets state only on concrete evidence:
    SEALED-with-error, or unsealed past TxReconcileStaleAfter. Transient AN
    errors do not touch state. A short critical section around the reset
    re-checks that the same flowTxID still owns the queue.
  * Injectable getTxResult field mirrors the existing submitBatch pattern
    so tests can drive reconciliation without a live Access Node.
  * New metric TxPoolReconcileReset(reason) with labels
    {wrapper-reverted, unsealed-past-threshold}.
  * New config knobs TxReconcileInterval (default 1s) and
    TxReconcileStaleAfter (default 30s), both validated > 0 when
    --tx-mempool-mode=true.

Test file change is minimal: only satisfies the new reconcileSubmission
signature (added flow.Identifier{} argument at four existing call sites)
and defaults getTxResult in newTestPool to a no-op fake. No new tests are
added here — a follow-up subagent covers behavior.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 cmd/run/cmd.go                        |   8 ++
 config/config.go                      |  10 ++
 metrics/collector.go                  |  13 +++
 metrics/nop.go                        |   1 +
 services/requester/tx_mempool.go      | 139 +++++++++++++++++++++++++-
 services/requester/tx_mempool_test.go |  10 ++
 6 files changed, 178 insertions(+), 3 deletions(-)

diff --git a/cmd/run/cmd.go b/cmd/run/cmd.go
index 33fd33eb..40ffd35a 100644
--- a/cmd/run/cmd.go
+++ b/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.")
diff --git a/config/config.go b/config/config.go
index 05ff7c91..686e5b23 100644
--- a/config/config.go
+++ b/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
diff --git a/metrics/collector.go b/metrics/collector.go
index 4f9307ca..819840c6 100644
--- a/metrics/collector.go
+++ b/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)
 }
diff --git a/metrics/nop.go b/metrics/nop.go
index d9a71fed..e5b7b1d5 100644
--- a/metrics/nop.go
+++ b/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)         {}
diff --git a/services/requester/tx_mempool.go b/services/requester/tx_mempool.go
index e3e4a1be..0bb2d6ae 100644
--- a/services/requester/tx_mempool.go
+++ b/services/requester/tx_mempool.go
@@ -100,6 +100,16 @@ 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, tick
+//       TxReconcileInterval) polls the wrapping Cadence tx status for each EOA
+//       with an outstanding in-flight marker. If the wrapper is sealed with an
+//       error (e.g. the run.cdc "nonce too high" assertion after intra-block
+//       reordering) OR it has not sealed within TxReconcileStaleAfter, the
+//       marker is cleared so the next Add() re-classifies against on-chain
+//       state — bounding wedge duration to ~one sealing window instead of the
+//       full idle-eviction retention.
+//
 // 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 +120,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.
@@ -410,6 +422,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 +456,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 +501,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 +538,10 @@ func NewTxMemPool(
 		now:           time.Now,
 	}
 	pool.submitBatch = pool.submitTxBatch
+	pool.getTxResult = pool.client.GetTransactionResult
 
 	go pool.processQueues(ctx)
+	go pool.reconcileLoop(ctx)
 
 	return pool, nil
 }
@@ -626,6 +659,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 +769,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 +886,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 +910,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 +1154,96 @@ 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. Three cases
+// trigger a reset:
+//  1. The wrapper is SEALED with a non-zero ErrorCode (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.
+//  3. The on-chain nextNonce has advanced past highestSent — the chain moved
+//     on without our help (unlikely with the other two cases, but keeps the
+//     marker precisely in sync).
+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 {
+		result, err := t.getTxResult(ctx, s.flowTxID)
+		// 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 := 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
+		}
+
+		reason := "unsealed-past-threshold"
+		if reverted {
+			reason = "wrapper-reverted"
+		}
+		t.logger.Warn().
+			Str("eoa", s.from.Hex()).
+			Str("flow-tx-id", s.flowTxID.Hex()).
+			Str("reason", reason).
+			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()
+	}
+}
diff --git a/services/requester/tx_mempool_test.go b/services/requester/tx_mempool_test.go
index feacb209..f9e72b3e 100644
--- a/services/requester/tx_mempool_test.go
+++ b/services/requester/tx_mempool_test.go
@@ -150,6 +150,12 @@ func newTestPool(
 	pool.submitBatch = func(ctx context.Context, txs []heldTx) (flow.Identifier, error) {
 		return flow.Identifier{}, submit(ctx, txs)
 	}
+	// The reconciliation loop is not started by newTestPool, but the field is
+	// defaulted to a no-op fake so any direct call to reconcileOnce from a test
+	// works without a live Access Node.
+	pool.getTxResult = func(context.Context, flow.Identifier) (*flow.TransactionResult, error) {
+		return nil, nil
+	}
 	return pool
 }
 
@@ -438,6 +444,7 @@ func Test_ReconcileSubmission_OnlyReconcilesMatchingInFlightBatch(t *testing.T)
 	// A different (newer) in-flight nonce owns the marker: not cleared.
 	pool.reconcileSubmission(
 		flushWork{from: from, txs: []heldTx{makeHeldTx(5, time.Time{})}, needsReconcile: true},
+		flow.Identifier{},
 		submitErr,
 	)
 	assert.True(t, pool.queues[from].nonces.inFlight())
@@ -445,6 +452,7 @@ func Test_ReconcileSubmission_OnlyReconcilesMatchingInFlightBatch(t *testing.T)
 	// A TTL-expiry batch (needsReconcile false) never touches the tracker.
 	pool.reconcileSubmission(
 		flushWork{from: from, txs: []heldTx{makeHeldTx(7, time.Time{})}, needsReconcile: false},
+		flow.Identifier{},
 		submitErr,
 	)
 	assert.True(t, pool.queues[from].nonces.inFlight())
@@ -452,6 +460,7 @@ func Test_ReconcileSubmission_OnlyReconcilesMatchingInFlightBatch(t *testing.T)
 	// The failed in-flight batch still owns the marker: cleared.
 	pool.reconcileSubmission(
 		flushWork{from: from, txs: []heldTx{makeHeldTx(7, time.Time{})}, needsReconcile: true},
+		flow.Identifier{},
 		submitErr,
 	)
 	assert.False(t, pool.queues[from].nonces.inFlight())
@@ -459,6 +468,7 @@ func Test_ReconcileSubmission_OnlyReconcilesMatchingInFlightBatch(t *testing.T)
 	// Unknown EOA: no panic.
 	pool.reconcileSubmission(
 		flushWork{from: gethCommon.HexToAddress("0xdef"), txs: []heldTx{makeHeldTx(7, time.Time{})}, needsReconcile: true},
+		flow.Identifier{},
 		submitErr,
 	)
 }

From 3c4b78a62dcb38e51d8175465475cf386d547b7d Mon Sep 17 00:00:00 2001
From: vishal <1117327+vishalchangrani@users.noreply.github.com>
Date: Wed, 29 Jul 2026 11:24:55 -0400
Subject: [PATCH 02/10] test(requester): unit tests for mempool reconciliation
 loop

Adds seven table-tests exercising TxMemPool.reconcileOnce, covering every
branch of the recovery path (behavior-spec case 13):

  - wrapper reverted (Sealed + non-nil Error) clears the marker
  - wrapper unsealed past TxReconcileStaleAfter clears the marker
  - wrapper Sealed with no error leaves the marker in place
  - wrapper unsealed and fresh leaves the marker in place
  - queue without an outstanding marker is skipped (no getTxResult call)
  - fresher submission superseding lastFlowTxID between snapshot and
    reset is respected (no wrongful reset)
  - end-to-end: post-reconcile, a same-nonce retry is accepted rather
    than rejected as in flight

All tests drive reconcileOnce synchronously through a fake clock plus
injected submitBatch/getTxResult, so no goroutines and no wall-clock
sleeps are involved. testPoolConfig() now seeds TxReconcileInterval and
TxReconcileStaleAfter to match the CLI defaults (1s / 30s).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 services/requester/tx_mempool_test.go | 348 +++++++++++++++++++++++++-
 1 file changed, 344 insertions(+), 4 deletions(-)

diff --git a/services/requester/tx_mempool_test.go b/services/requester/tx_mempool_test.go
index f9e72b3e..3cdd23a7 100644
--- a/services/requester/tx_mempool_test.go
+++ b/services/requester/tx_mempool_test.go
@@ -183,10 +183,12 @@ func (c *fakeClock) advance(d time.Duration) {
 
 func testPoolConfig() config.Config {
 	return config.Config{
-		TxCollectionWindow:  100 * time.Millisecond,
-		TxSubmissionSpacing: time.Second,
-		TxPoolTTL:           time.Minute,
-		TxMaxBatchSize:      10,
+		TxCollectionWindow:    100 * time.Millisecond,
+		TxSubmissionSpacing:   time.Second,
+		TxPoolTTL:             time.Minute,
+		TxMaxBatchSize:        10,
+		TxReconcileInterval:   time.Second,
+		TxReconcileStaleAfter: 30 * time.Second,
 	}
 }
 
@@ -1188,3 +1190,341 @@ func noncesOf(txs []heldTx) []uint64 {
 	}
 	return ns
 }
+
+// --- Reconciliation loop tests -------------------------------------------
+// These drive reconcileOnce directly (rather than through the background
+// goroutine) so behavior can be asserted synchronously and without wall-clock
+// sleeps. The 7 cases below map to the recovery spec (behavior spec case 13
+// in tx_mempool.go).
+
+// primeReconcilePool sets up a pool with one EOA (`from`) whose fast-path
+// submission has succeeded: the nonce tracker records nonce N as consecutively
+// submitted, no submission is in flight, and lastFlowTxID / lastSubmittedAt
+// are set from the returned values. Returns the flow-tx-id that identifies the
+// most-recent wrapper (what reconcileOnce polls) so tests can assert the
+// getTxResult callback receives the expected identifier.
+func primeReconcilePool(
+	t *testing.T,
+	pool *TxMemPool,
+	clk *fakeClock,
+	key *ecdsa.PrivateKey,
+	nonce uint64,
+	flowTxID flow.Identifier,
+) gethCommon.Address {
+	t.Helper()
+	from := crypto.PubkeyToAddress(key.PublicKey)
+
+	// Fast-path submit sets lastConsecutivelySubmitted, lastFlowTxID, and
+	// lastSubmittedAt from a real Add() flow — exercising the actual submission
+	// path (rather than seeding fields by hand).
+	pool.submitBatch = func(_ context.Context, _ []heldTx) (flow.Identifier, error) {
+		return flowTxID, nil
+	}
+	require.NoError(t, pool.Add(context.Background(), signedTestTx(t, key, nonce, 1)))
+
+	q := pool.queues[from]
+	require.NotNil(t, q)
+	require.True(t, q.nonces.lastConsecutivelySubmitted.set,
+		"precondition: fast-path submission must have advanced lastConsecutivelySubmitted")
+	require.Equal(t, nonce, q.nonces.lastConsecutivelySubmitted.v)
+	require.Equal(t, flowTxID, q.lastFlowTxID)
+	require.Equal(t, clk.now(), q.lastSubmittedAt)
+	return from
+}
+
+// After a fast-path submission the wrapping Cadence tx can seal with an error
+// (e.g. the run.cdc "nonce too high" assertion after intra-block reordering).
+// reconcileOnce must observe this and clear the in-flight state so a subsequent
+// Add() re-classifies against the on-chain frontier rather than staying wedged
+// behind the stale marker.
+func Test_TxMemPool_ReconcileClearsMarkerWhenWrapperReverted(t *testing.T) {
+	key, err := crypto.GenerateKey()
+	require.NoError(t, err)
+
+	clk := newFakeClock(timingClockBase)
+	pool := newTestPool(
+		&fakeNonceProvider{nonce: 3},
+		func(_ context.Context, _ []heldTx) error { return nil },
+		testPoolConfig(),
+	)
+	pool.now = clk.now
+
+	flowTxID := flow.HexToID(
+		"1111111111111111111111111111111111111111111111111111111111111111",
+	)
+	from := primeReconcilePool(t, pool, clk, key, 3, flowTxID)
+
+	var polledID flow.Identifier
+	pool.getTxResult = func(_ context.Context, id flow.Identifier) (*flow.TransactionResult, error) {
+		polledID = id
+		return &flow.TransactionResult{
+			Status: flow.TransactionStatusSealed,
+			Error:  errors.New("evm_error=nonce too high"),
+		}, nil
+	}
+
+	pool.reconcileOnce(context.Background())
+
+	assert.Equal(t, flowTxID, polledID, "reconciler must poll the recorded wrapping tx id")
+
+	q := pool.queues[from]
+	require.NotNil(t, q)
+	assert.False(t, q.nonces.lastConsecutivelySubmitted.set,
+		"reverted wrapper must clear lastConsecutivelySubmitted")
+	assert.False(t, q.nonces.submitting.set,
+		"reverted wrapper must clear submitting")
+	assert.Equal(t, flow.Identifier{}, q.lastFlowTxID,
+		"reverted wrapper must zero lastFlowTxID so a fresher submission owns the slot")
+}
+
+// A wrapper that never seals within TxReconcileStaleAfter is treated as
+// dropped: reconcileOnce clears the in-flight marker so the EOA can recover
+// without waiting for the idle-queue eviction window.
+func Test_TxMemPool_ReconcileClearsMarkerWhenWrapperStale(t *testing.T) {
+	key, err := crypto.GenerateKey()
+	require.NoError(t, err)
+
+	clk := newFakeClock(timingClockBase)
+	pool := newTestPool(
+		&fakeNonceProvider{nonce: 3},
+		func(_ context.Context, _ []heldTx) error { return nil },
+		testPoolConfig(),
+	)
+	pool.now = clk.now
+
+	flowTxID := flow.HexToID(
+		"2222222222222222222222222222222222222222222222222222222222222222",
+	)
+	from := primeReconcilePool(t, pool, clk, key, 3, flowTxID)
+
+	// Wrapper has not sealed — could be an AN drop or a slow seal. Either way,
+	// past the stale threshold reconcileOnce must clear the marker.
+	pool.getTxResult = func(_ context.Context, _ flow.Identifier) (*flow.TransactionResult, error) {
+		return &flow.TransactionResult{Status: flow.TransactionStatusExecuted}, nil
+	}
+
+	// Advance past the stale threshold. lastSubmittedAt was stamped at
+	// timingClockBase inside primeReconcilePool.
+	clk.advance(pool.config.TxReconcileStaleAfter + time.Second)
+
+	pool.reconcileOnce(context.Background())
+
+	q := pool.queues[from]
+	require.NotNil(t, q)
+	assert.False(t, q.nonces.lastConsecutivelySubmitted.set,
+		"stale unsealed wrapper must clear lastConsecutivelySubmitted")
+	assert.False(t, q.nonces.submitting.set)
+	assert.Equal(t, flow.Identifier{}, q.lastFlowTxID)
+}
+
+// A wrapper that sealed cleanly (Status Sealed, no Error) is the healthy path:
+// the on-chain nonce has advanced and reconcileOnce must leave the marker
+// alone. Resetting here would let a client's retry with the same nonce
+// double-spend against the freshly-advanced frontier.
+func Test_TxMemPool_ReconcileLeavesMarkerWhenWrapperSealedSuccessfully(t *testing.T) {
+	key, err := crypto.GenerateKey()
+	require.NoError(t, err)
+
+	clk := newFakeClock(timingClockBase)
+	pool := newTestPool(
+		&fakeNonceProvider{nonce: 3},
+		func(_ context.Context, _ []heldTx) error { return nil },
+		testPoolConfig(),
+	)
+	pool.now = clk.now
+
+	flowTxID := flow.HexToID(
+		"3333333333333333333333333333333333333333333333333333333333333333",
+	)
+	from := primeReconcilePool(t, pool, clk, key, 3, flowTxID)
+
+	pool.getTxResult = func(_ context.Context, _ flow.Identifier) (*flow.TransactionResult, error) {
+		return &flow.TransactionResult{Status: flow.TransactionStatusSealed, Error: nil}, nil
+	}
+
+	pool.reconcileOnce(context.Background())
+
+	q := pool.queues[from]
+	require.NotNil(t, q)
+	assert.True(t, q.nonces.lastConsecutivelySubmitted.set,
+		"sealed-successful wrapper must NOT clear the marker")
+	assert.Equal(t, uint64(3), q.nonces.lastConsecutivelySubmitted.v)
+	assert.Equal(t, flowTxID, q.lastFlowTxID,
+		"sealed-successful wrapper must preserve lastFlowTxID")
+}
+
+// A wrapper that is still in flight (not yet sealed) within the stale window
+// is normal steady-state operation. reconcileOnce must not reset in this case;
+// resetting would race the imminent seal and could allow a duplicate submission.
+func Test_TxMemPool_ReconcileLeavesMarkerWhenWrapperUnsealedAndFresh(t *testing.T) {
+	key, err := crypto.GenerateKey()
+	require.NoError(t, err)
+
+	clk := newFakeClock(timingClockBase)
+	pool := newTestPool(
+		&fakeNonceProvider{nonce: 3},
+		func(_ context.Context, _ []heldTx) error { return nil },
+		testPoolConfig(),
+	)
+	pool.now = clk.now
+
+	flowTxID := flow.HexToID(
+		"4444444444444444444444444444444444444444444444444444444444444444",
+	)
+	from := primeReconcilePool(t, pool, clk, key, 3, flowTxID)
+
+	pool.getTxResult = func(_ context.Context, _ flow.Identifier) (*flow.TransactionResult, error) {
+		return &flow.TransactionResult{Status: flow.TransactionStatusExecuted}, nil
+	}
+
+	// Advance a small amount — well within the stale threshold.
+	clk.advance(pool.config.TxReconcileStaleAfter / 3)
+
+	pool.reconcileOnce(context.Background())
+
+	q := pool.queues[from]
+	require.NotNil(t, q)
+	assert.True(t, q.nonces.lastConsecutivelySubmitted.set,
+		"unsealed fresh wrapper must NOT clear the marker")
+	assert.Equal(t, uint64(3), q.nonces.lastConsecutivelySubmitted.v)
+	assert.Equal(t, flowTxID, q.lastFlowTxID)
+}
+
+// An EOA with no outstanding submission marker is not a candidate for
+// reconciliation — reconcileOnce must skip it entirely and never issue a
+// getTxResult call for it (avoids unnecessary AN traffic on idle EOAs and
+// starts up scenarios where every queue is fresh).
+func Test_TxMemPool_ReconcileSkipsQueueWithoutMarker(t *testing.T) {
+	pool := newTestPool(
+		&fakeNonceProvider{nonce: 0},
+		func(_ context.Context, _ []heldTx) error { return nil },
+		testPoolConfig(),
+	)
+	from := gethCommon.HexToAddress("0xabc")
+
+	// Empty tracker, no lastFlowTxID: no work for the reconciler.
+	pool.queues[from] = &eoaQueue{
+		txs:          map[uint64]heldTx{},
+		lastActivity: time.Now(),
+	}
+
+	pool.getTxResult = func(_ context.Context, _ flow.Identifier) (*flow.TransactionResult, error) {
+		t.Fatalf("getTxResult must not be called for a queue without an outstanding marker")
+		return nil, nil
+	}
+
+	pool.reconcileOnce(context.Background())
+
+	// The queue must be untouched.
+	q := pool.queues[from]
+	require.NotNil(t, q)
+	assert.False(t, q.nonces.lastConsecutivelySubmitted.set)
+	assert.False(t, q.nonces.submitting.set)
+	assert.Equal(t, flow.Identifier{}, q.lastFlowTxID)
+}
+
+// While reconcileOnce is polling the AN outside the lock, a concurrent
+// submission may advance the EOA's lastFlowTxID to a fresher wrapper. When the
+// reconciler re-acquires the lock to reset, it must notice that the flow-tx-id
+// has moved on and leave the (now-current) marker alone — otherwise a
+// successful just-submitted batch would be wrongly cleared.
+func Test_TxMemPool_ReconcileSkipsSupersededFlowTxID(t *testing.T) {
+	key, err := crypto.GenerateKey()
+	require.NoError(t, err)
+
+	clk := newFakeClock(timingClockBase)
+	pool := newTestPool(
+		&fakeNonceProvider{nonce: 3},
+		func(_ context.Context, _ []heldTx) error { return nil },
+		testPoolConfig(),
+	)
+	pool.now = clk.now
+
+	originalFlowTxID := flow.HexToID(
+		"5555555555555555555555555555555555555555555555555555555555555555",
+	)
+	newerFlowTxID := flow.HexToID(
+		"6666666666666666666666666666666666666666666666666666666666666666",
+	)
+	from := primeReconcilePool(t, pool, clk, key, 3, originalFlowTxID)
+
+	// The getTxResult call simulates the race: while the reconciler is polling
+	// outside the lock, a concurrent successful submission bumps the queue's
+	// lastFlowTxID. When the reconciler re-acquires the lock to reset, the
+	// stored id will no longer match its snapshot and the reset must be skipped.
+	pool.getTxResult = func(_ context.Context, _ flow.Identifier) (*flow.TransactionResult, error) {
+		pool.queueMux.Lock()
+		pool.queues[from].lastFlowTxID = newerFlowTxID
+		pool.queueMux.Unlock()
+		return &flow.TransactionResult{
+			Status: flow.TransactionStatusSealed,
+			Error:  errors.New("wrapper reverted"),
+		}, nil
+	}
+
+	pool.reconcileOnce(context.Background())
+
+	q := pool.queues[from]
+	require.NotNil(t, q)
+	assert.True(t, q.nonces.lastConsecutivelySubmitted.set,
+		"a fresher submission has superseded the snapshot; reconciler must not reset")
+	assert.Equal(t, uint64(3), q.nonces.lastConsecutivelySubmitted.v)
+	assert.Equal(t, newerFlowTxID, q.lastFlowTxID,
+		"the newer lastFlowTxID must survive the reconciler pass")
+}
+
+// End-to-end: after reconciliation clears a wedged marker, a client's retry
+// with the same nonce is accepted (fast-paths) rather than being rejected as
+// in flight. This is the operational point of the reconciliation loop.
+func Test_TxMemPool_ReconcileClearsAllowsSubsequentAddToBeAccepted(t *testing.T) {
+	key, err := crypto.GenerateKey()
+	require.NoError(t, err)
+
+	clk := newFakeClock(timingClockBase)
+	pool := newTestPool(
+		// Frontier stays at 3: the reverted wrapper did not advance the chain.
+		&fakeNonceProvider{nonce: 3},
+		func(_ context.Context, _ []heldTx) error { return nil },
+		testPoolConfig(),
+	)
+	pool.now = clk.now
+
+	flowTxID := flow.HexToID(
+		"7777777777777777777777777777777777777777777777777777777777777777",
+	)
+	from := primeReconcilePool(t, pool, clk, key, 3, flowTxID)
+
+	// While the marker is set, a retry of the same nonce is rejected as in flight.
+	err = pool.Add(context.Background(), signedTestTx(t, key, 3, 2))
+	require.ErrorIs(t, err, errs.ErrInFlightNonce,
+		"before reconciliation, retry with the same nonce must be rejected as in flight")
+
+	// Reconcile with a reverted wrapper: the marker clears.
+	pool.getTxResult = func(_ context.Context, _ flow.Identifier) (*flow.TransactionResult, error) {
+		return &flow.TransactionResult{
+			Status: flow.TransactionStatusSealed,
+			Error:  errors.New("evm_error=nonce too high"),
+		}, nil
+	}
+	// Advance past the submission-spacing gap so the subsequent Add can fast-path.
+	clk.advance(pool.config.TxSubmissionSpacing + time.Second)
+	pool.reconcileOnce(context.Background())
+
+	// After reconciliation, a retry of the same nonce is accepted (fast-paths).
+	var retrySubmitted bool
+	pool.submitBatch = func(_ context.Context, txs []heldTx) (flow.Identifier, error) {
+		retrySubmitted = true
+		require.Len(t, txs, 1)
+		assert.Equal(t, uint64(3), txs[0].nonce)
+		return flow.HexToID("8888888888888888888888888888888888888888888888888888888888888888"), nil
+	}
+	require.NoError(t, pool.Add(context.Background(), signedTestTx(t, key, 3, 2)),
+		"after reconciliation, retry with the same nonce must be accepted")
+	assert.True(t, retrySubmitted, "retry must reach the submit path (fast-path)")
+
+	// And the EOA is no longer wedged: the new submission owns lastFlowTxID.
+	q := pool.queues[from]
+	require.NotNil(t, q)
+	assert.NotEqual(t, flowTxID, q.lastFlowTxID,
+		"lastFlowTxID must now reference the retry's wrapping tx, not the reverted one")
+}

From ebc4ecb22a197fb19c4aebdf1dd52300066b6ec4 Mon Sep 17 00:00:00 2001
From: vishal <1117327+vishalchangrani@users.noreply.github.com>
Date: Wed, 29 Jul 2026 11:29:47 -0400
Subject: [PATCH 03/10] docs(requester): reconciliation loop behavior spec and
 README

Extend the tx_mempool.go BEHAVIOR SPEC with a Recovery section
describing the reconciliation loop's ticks, reset triggers, concurrency
model, and observability signals. Fix a doc-vs-code inconsistency in
the reconcileLoop docstring: it listed three reset cases, but only two
are implemented. The "chain advanced past highestSent" case is
redundant, because a successfully-sealed wrapper is progressed by the
next legitimate submission, so no wedge needs clearing. Add the two
new CLI flags (tx-reconcile-interval, tx-reconcile-stale-after) to the
README Configuration Flags table.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 README.md                        |  2 ++
 services/requester/tx_mempool.go | 46 +++++++++++++++++++++++---------
 2 files changed, 35 insertions(+), 13 deletions(-)

diff --git a/README.md b/README.md
index 325ff67e..7c26f001 100644
--- a/README.md
+++ b/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
diff --git a/services/requester/tx_mempool.go b/services/requester/tx_mempool.go
index 0bb2d6ae..cfa290e8 100644
--- a/services/requester/tx_mempool.go
+++ b/services/requester/tx_mempool.go
@@ -101,14 +101,32 @@ import (
 //       rejected (it cannot execute until ~595 intervening nonces are filled).
 //
 // Recovery
-//   13. Reconciliation: a background loop (reconcileLoop, tick
-//       TxReconcileInterval) polls the wrapping Cadence tx status for each EOA
-//       with an outstanding in-flight marker. If the wrapper is sealed with an
-//       error (e.g. the run.cdc "nonce too high" assertion after intra-block
-//       reordering) OR it has not sealed within TxReconcileStaleAfter, the
-//       marker is cleared so the next Add() re-classifies against on-chain
-//       state — bounding wedge duration to ~one sealing window instead of the
-//       full idle-eviction retention.
+//   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 re-checks lastFlowTxID still
+//       matches — so a superseding submission is never 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
@@ -1157,17 +1175,19 @@ func (t *TxMemPool) submitTxBatch(ctx context.Context, txs []heldTx) (flow.Ident
 
 // 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. Three cases
+// wrapper is provably not going to advance the on-chain nonce. Two cases
 // trigger a reset:
-//  1. The wrapper is SEALED with a non-zero ErrorCode (it reverted — e.g. the
+//  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.
-//  3. The on-chain nextNonce has advanced past highestSent — the chain moved
-//     on without our help (unlikely with the other two cases, but keeps the
-//     marker precisely in sync).
+//
+// 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()

From 0e925ec8ebb9b01e1200c50e0cb878a1f7bcd558 Mon Sep 17 00:00:00 2001
From: vishal <1117327+vishalchangrani@users.noreply.github.com>
Date: Wed, 29 Jul 2026 11:59:59 -0400
Subject: [PATCH 04/10] fix(requester): reviewer-flagged blockers in
 reconciliation loop
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Address two blockers surfaced during the code-review pass on the initial
implementation (f46029c3):

1. Panic in NewTxMemPool when a programmatic caller (e.g. the e2e test
   config in tests/tx_mempool_test.go) leaves TxReconcileInterval or
   TxReconcileStaleAfter at their zero values: time.NewTicker(0) panics.
   Fix: backfill both fields to their CLI-default values (1s, 30s)
   inside NewTxMemPool before starting the reconcile loop.

2. Behavioral race in reconcileOnce that could clobber a fresh in-flight
   submission. After the snapshot-outside-the-lock pattern, if a
   concurrent background flush called markSubmitting(newHigh) between
   snapshot and reset, the reconciler saw an unchanged lastFlowTxID and
   would reset — clearing submitting for the newer batch. That would let
   a client retry duplicate the in-flight nonce, reintroducing the exact
   duplicate-wrapper failure this loop is meant to eliminate. Fix: after
   the lastFlowTxID re-check under the lock, also require
   q.nonces.inFlight() to be false; a matching lastFlowTxID guarantees
   the previous batch's markSubmitted has returned, so any submitting we
   observe now must belong to a strictly newer batch.

Also enriches the WARN log with elapsed-since-submit duration and, when
known, the wrapper's Cadence error message.

Adds Test_TxMemPool_ReconcileSkipsWhenFreshBatchInFlight; verified to
FAIL without the fix and PASS with it.

Verified: go build/vet clean; unit tests pass under -race; e2e
Test_TxMemPool suite now passes cleanly.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 services/requester/tx_mempool.go      | 48 ++++++++++++++++++++++--
 services/requester/tx_mempool_test.go | 53 +++++++++++++++++++++++++++
 2 files changed, 97 insertions(+), 4 deletions(-)

diff --git a/services/requester/tx_mempool.go b/services/requester/tx_mempool.go
index cfa290e8..fb36b3b4 100644
--- a/services/requester/tx_mempool.go
+++ b/services/requester/tx_mempool.go
@@ -120,8 +120,10 @@ import (
 //       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 re-checks lastFlowTxID still
-//       matches — so a superseding submission is never clobbered.
+//       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
@@ -224,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):
@@ -558,6 +572,15 @@ func NewTxMemPool(
 	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)
 
@@ -1250,16 +1273,33 @@ func (t *TxMemPool) reconcileOnce(ctx context.Context) {
 			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"
 		}
-		t.logger.Warn().
+		elapsed := now.Sub(s.lastSubmittedAt)
+		event := t.logger.Warn().
 			Str("eoa", s.from.Hex()).
 			Str("flow-tx-id", s.flowTxID.Hex()).
 			Str("reason", reason).
-			Msg("reconciliation clearing stuck in-flight marker; subsequent Add() calls will re-classify against chain")
+			Dur("elapsed-since-submit", elapsed)
+		if reverted && result.Error != nil {
+			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{}
diff --git a/services/requester/tx_mempool_test.go b/services/requester/tx_mempool_test.go
index 3cdd23a7..444c634e 100644
--- a/services/requester/tx_mempool_test.go
+++ b/services/requester/tx_mempool_test.go
@@ -1473,6 +1473,59 @@ func Test_TxMemPool_ReconcileSkipsSupersededFlowTxID(t *testing.T) {
 		"the newer lastFlowTxID must survive the reconciler pass")
 }
 
+// If a fresh batch enters flight between the reconciler's snapshot and its
+// reset — such that lastFlowTxID still matches the snapshot but q.nonces.submitting
+// is now set for a newer batch — the reset must be skipped. Otherwise the reset
+// would clobber the newer batch's submitting marker and let a client retry
+// duplicate a nonce that is legitimately in flight, reintroducing the very
+// duplicate-wrapper failure mode this loop exists to prevent.
+func Test_TxMemPool_ReconcileSkipsWhenFreshBatchInFlight(t *testing.T) {
+	key, err := crypto.GenerateKey()
+	require.NoError(t, err)
+
+	clk := newFakeClock(timingClockBase)
+	pool := newTestPool(
+		&fakeNonceProvider{nonce: 3},
+		func(_ context.Context, _ []heldTx) error { return nil },
+		testPoolConfig(),
+	)
+	pool.now = clk.now
+
+	flowTxID := flow.HexToID(
+		"7777777777777777777777777777777777777777777777777777777777777777",
+	)
+	from := primeReconcilePool(t, pool, clk, key, 3, flowTxID)
+
+	// Simulate a concurrent background flush that acquires the lock while the
+	// reconciler is out doing its network call, marks a newer batch in flight,
+	// but does NOT yet update lastFlowTxID (that happens later in
+	// reconcileSubmission on submit success). The reconciler must detect the
+	// in-flight marker and skip.
+	newerInFlightNonce := uint64(4)
+	pool.getTxResult = func(_ context.Context, _ flow.Identifier) (*flow.TransactionResult, error) {
+		pool.queueMux.Lock()
+		pool.queues[from].nonces.markSubmitting(newerInFlightNonce)
+		pool.queueMux.Unlock()
+		return &flow.TransactionResult{
+			Status: flow.TransactionStatusSealed,
+			Error:  errors.New("wrapper reverted"),
+		}, nil
+	}
+
+	pool.reconcileOnce(context.Background())
+
+	q := pool.queues[from]
+	require.NotNil(t, q)
+	assert.True(t, q.nonces.lastConsecutivelySubmitted.set,
+		"snapshot's marker must be preserved; newer batch owns the state now")
+	assert.Equal(t, uint64(3), q.nonces.lastConsecutivelySubmitted.v)
+	assert.True(t, q.nonces.submitting.set,
+		"newer batch's submitting marker must survive the reconciler pass")
+	assert.Equal(t, newerInFlightNonce, q.nonces.submitting.v)
+	assert.Equal(t, flowTxID, q.lastFlowTxID,
+		"lastFlowTxID unchanged (newer batch has not yet ack'd)")
+}
+
 // End-to-end: after reconciliation clears a wedged marker, a client's retry
 // with the same nonce is accepted (fast-paths) rather than being rejected as
 // in flight. This is the operational point of the reconciliation loop.

From 9cc9c74201d1d5cd64a887c1de5c11f5619b1638 Mon Sep 17 00:00:00 2001
From: vishal <1117327+vishalchangrani@users.noreply.github.com>
Date: Wed, 29 Jul 2026 12:11:46 -0400
Subject: [PATCH 05/10] chore(requester): align reconcile-loop with PR #984
 flow_tx_id naming

PR #984's review consolidated the log field name from flow-tx-id to
flow_tx_id (underscores are safer for Grafana filter syntax). The
reconciliation loop's WARN log and its associated behavior-spec comment
still used the old dashed name after rebase; align them so a downstream
grep on 'flow_tx_id' catches both the submission and the reconciler
reset log lines.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 services/requester/tx_mempool.go | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/services/requester/tx_mempool.go b/services/requester/tx_mempool.go
index fb36b3b4..8a93544c 100644
--- a/services/requester/tx_mempool.go
+++ b/services/requester/tx_mempool.go
@@ -125,7 +125,7 @@ import (
 //       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")
+//       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).
@@ -1267,7 +1267,7 @@ func (t *TxMemPool) reconcileOnce(ctx context.Context) {
 
 		t.queueMux.Lock()
 		q, ok := t.queues[s.from]
-		// If the queue was evicted or already advanced (different flow-tx-id
+		// 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()
@@ -1293,7 +1293,7 @@ func (t *TxMemPool) reconcileOnce(ctx context.Context) {
 		elapsed := now.Sub(s.lastSubmittedAt)
 		event := t.logger.Warn().
 			Str("eoa", s.from.Hex()).
-			Str("flow-tx-id", s.flowTxID.Hex()).
+			Str("flow_tx_id", s.flowTxID.Hex()).
 			Str("reason", reason).
 			Dur("elapsed-since-submit", elapsed)
 		if reverted && result.Error != nil {

From 15091f360a83f6214efedbf8c59ccc01575315db Mon Sep 17 00:00:00 2001
From: Ardit Marku <markoupetr@gmail.com>
Date: Fri, 31 Jul 2026 13:42:31 +0300
Subject: [PATCH 06/10] Simplify condition for logging wrapper-error

---
 services/requester/tx_mempool.go | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/services/requester/tx_mempool.go b/services/requester/tx_mempool.go
index 8a93544c..214612f3 100644
--- a/services/requester/tx_mempool.go
+++ b/services/requester/tx_mempool.go
@@ -1296,7 +1296,8 @@ func (t *TxMemPool) reconcileOnce(ctx context.Context) {
 			Str("flow_tx_id", s.flowTxID.Hex()).
 			Str("reason", reason).
 			Dur("elapsed-since-submit", elapsed)
-		if reverted && result.Error != nil {
+
+		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")

From 0b45280d525dd102073cfa2331b05077337dcf1c Mon Sep 17 00:00:00 2001
From: Ardit Marku <markoupetr@gmail.com>
Date: Fri, 31 Jul 2026 13:54:02 +0300
Subject: [PATCH 07/10] Check specifically for tx result status of
 flow.TransactionStatusSealed

---
 services/requester/tx_mempool.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/services/requester/tx_mempool.go b/services/requester/tx_mempool.go
index 214612f3..f2e70bc8 100644
--- a/services/requester/tx_mempool.go
+++ b/services/requester/tx_mempool.go
@@ -1257,7 +1257,7 @@ func (t *TxMemPool) reconcileOnce(ctx context.Context) {
 		// 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
+		sealed := err == nil && result != nil && result.Status == flow.TransactionStatusSealed
 		reverted := sealed && result.Error != nil
 		stale := now.Sub(s.lastSubmittedAt) > t.config.TxReconcileStaleAfter
 

From 4269ef3d9570999d08703aa32ca97ecab4ba9285 Mon Sep 17 00:00:00 2001
From: Ardit Marku <markoupetr@gmail.com>
Date: Fri, 31 Jul 2026 17:38:08 +0300
Subject: [PATCH 08/10] Stale condition should filter out sealed-successful
 wrappers

---
 services/requester/tx_mempool.go      |  2 +-
 services/requester/tx_mempool_test.go | 35 +++++++++++++++++++++++++++
 2 files changed, 36 insertions(+), 1 deletion(-)

diff --git a/services/requester/tx_mempool.go b/services/requester/tx_mempool.go
index f2e70bc8..410ccf0c 100644
--- a/services/requester/tx_mempool.go
+++ b/services/requester/tx_mempool.go
@@ -1259,7 +1259,7 @@ func (t *TxMemPool) reconcileOnce(ctx context.Context) {
 		// staleness threshold is exceeded.
 		sealed := err == nil && result != nil && result.Status == flow.TransactionStatusSealed
 		reverted := sealed && result.Error != nil
-		stale := now.Sub(s.lastSubmittedAt) > t.config.TxReconcileStaleAfter
+		stale := !(sealed && result.Error == nil) && now.Sub(s.lastSubmittedAt) > t.config.TxReconcileStaleAfter
 
 		if !reverted && !stale {
 			continue
diff --git a/services/requester/tx_mempool_test.go b/services/requester/tx_mempool_test.go
index 444c634e..13ee3988 100644
--- a/services/requester/tx_mempool_test.go
+++ b/services/requester/tx_mempool_test.go
@@ -1353,6 +1353,41 @@ func Test_TxMemPool_ReconcileLeavesMarkerWhenWrapperSealedSuccessfully(t *testin
 		"sealed-successful wrapper must preserve lastFlowTxID")
 }
 
+func Test_TxMemPool_ReconcileLeavesMarkerWhenWrapperSealedSuccessfullyPastGracePeriod(t *testing.T) {
+	key, err := crypto.GenerateKey()
+	require.NoError(t, err)
+
+	clk := newFakeClock(timingClockBase)
+	pool := newTestPool(
+		&fakeNonceProvider{nonce: 3},
+		func(_ context.Context, _ []heldTx) error { return nil },
+		testPoolConfig(),
+	)
+	pool.now = clk.now
+
+	flowTxID := flow.HexToID(
+		"3333333333333333333333333333333333333333333333333333333333333333",
+	)
+	from := primeReconcilePool(t, pool, clk, key, 3, flowTxID)
+
+	pool.getTxResult = func(_ context.Context, _ flow.Identifier) (*flow.TransactionResult, error) {
+		return &flow.TransactionResult{Status: flow.TransactionStatusSealed, Error: nil}, nil
+	}
+
+	// Advance by a sufficient amount — well after the stale threshold.
+	clk.advance(pool.config.TxReconcileStaleAfter + 3)
+
+	pool.reconcileOnce(context.Background())
+
+	q := pool.queues[from]
+	require.NotNil(t, q)
+	assert.True(t, q.nonces.lastConsecutivelySubmitted.set,
+		"sealed-successful wrapper must NOT clear the marker")
+	assert.Equal(t, uint64(3), q.nonces.lastConsecutivelySubmitted.v)
+	assert.Equal(t, flowTxID, q.lastFlowTxID,
+		"sealed-successful wrapper must preserve lastFlowTxID")
+}
+
 // A wrapper that is still in flight (not yet sealed) within the stale window
 // is normal steady-state operation. reconcileOnce must not reset in this case;
 // resetting would race the imminent seal and could allow a duplicate submission.

From 5f21e9264712651da684df9d875b32aa8512a48e Mon Sep 17 00:00:00 2001
From: Ardit Marku <markoupetr@gmail.com>
Date: Fri, 31 Jul 2026 17:42:35 +0300
Subject: [PATCH 09/10] Add context.WithTimeout on each GetTransactionResult AN
 call

---
 services/requester/tx_mempool.go | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/services/requester/tx_mempool.go b/services/requester/tx_mempool.go
index 410ccf0c..0caa57b7 100644
--- a/services/requester/tx_mempool.go
+++ b/services/requester/tx_mempool.go
@@ -1252,7 +1252,9 @@ func (t *TxMemPool) reconcileOnce(ctx context.Context) {
 
 	now := t.now()
 	for _, s := range snaps {
-		result, err := t.getTxResult(ctx, s.flowTxID)
+		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

From 2e15b7084e6ebb000da63372041c5747c41e60b8 Mon Sep 17 00:00:00 2001
From: Ardit Marku <markoupetr@gmail.com>
Date: Fri, 31 Jul 2026 17:53:50 +0300
Subject: [PATCH 10/10] Fix linting issue by applying De Morgan's law

---
 services/requester/tx_mempool.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/services/requester/tx_mempool.go b/services/requester/tx_mempool.go
index 0caa57b7..b2e791cb 100644
--- a/services/requester/tx_mempool.go
+++ b/services/requester/tx_mempool.go
@@ -1261,7 +1261,7 @@ func (t *TxMemPool) reconcileOnce(ctx context.Context) {
 		// 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
+		stale := (!sealed || result.Error != nil) && now.Sub(s.lastSubmittedAt) > t.config.TxReconcileStaleAfter
 
 		if !reverted && !stale {
 			continue