diff --git a/cmd/run/cmd.go b/cmd/run/cmd.go index ca425125..92417152 100644 --- a/cmd/run/cmd.go +++ b/cmd/run/cmd.go @@ -323,6 +323,7 @@ func init() { Cmd.Flags().DurationVar(&cfg.TxSubmissionSpacing, "tx-submission-spacing", 1200*time.Millisecond, "Minimum gap between consecutive Cadence submissions for the same EOA in the transaction mempool; also serves as the flush deadline for a continuously-fed collection window. Recommended ~1.5x the block production rate.") 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.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 b37597ba..4206b415 100644 --- a/config/config.go +++ b/config/config.go @@ -147,6 +147,14 @@ type Config struct { // single EVM.batchRun Cadence transaction by the transaction mempool, // bounded by the Cadence transaction computation limit. TxMaxBatchSize int + // TxMaxNonceGap is how far ahead of an EOA's on-chain nonce the transaction + // mempool will accept a nonce. A transaction whose nonce exceeds + // indexedNonce + TxMaxNonceGap is rejected up front with ErrNonceTooHigh, + // giving the client immediate feedback instead of holding an unexecutable tx + // until TTL. 0 means no upper bound (any future nonce is accepted). This + // bounds only the upper end: a nonce below the indexed nonce is always + // rejected with ErrNonceTooLow regardless of this setting. + TxMaxNonceGap uint64 // RpcRequestTimeout is the maximum duration at which JSON-RPC requests should generate // a response, before they timeout. RpcRequestTimeout time.Duration diff --git a/models/errors/errors.go b/models/errors/errors.go index 4438ea22..871e8034 100644 --- a/models/errors/errors.go +++ b/models/errors/errors.go @@ -36,6 +36,13 @@ var ( // already been submitted to the network and is awaiting execution. Letting // it through would burn Flow fees on a guaranteed nonce-mismatch failure. ErrInFlightNonce = fmt.Errorf("%w: %s", ErrInvalid, "transaction with the same nonce already submitted") + // ErrNonceTooLow is returned when a transaction's nonce is below the EOA's + // current on-chain nonce: it has already been used and can never execute. + ErrNonceTooLow = fmt.Errorf("%w: %s", ErrInvalid, "nonce too low") + // ErrNonceTooHigh is returned when a transaction's nonce is more than the + // configured maximum gap ahead of the EOA's on-chain nonce. Such a tx cannot + // execute until the gap fills, so it is rejected up front for fast feedback. + ErrNonceTooHigh = fmt.Errorf("%w: %s", ErrInvalid, "nonce too high") // Storage errors diff --git a/services/requester/nonce_provider.go b/services/requester/nonce_provider.go index 435a55e5..2dafad99 100644 --- a/services/requester/nonce_provider.go +++ b/services/requester/nonce_provider.go @@ -1,6 +1,8 @@ package requester import ( + "sync" + gethCommon "github.com/ethereum/go-ethereum/common" "github.com/onflow/flow-go/fvm/evm" "github.com/onflow/flow-go/fvm/evm/offchain/query" @@ -10,25 +12,49 @@ import ( "github.com/onflow/flow-evm-gateway/storage/pebble" ) -// NonceProvider returns the current nonce of the given EOA address. -// The transaction mempool uses it to determine the expected next nonce. +// NonceView reads EOA nonces at a single, fixed EVM state (one built block +// view). The mempool reads many EOAs' nonces from one view per flush tick +// rather than rebuilding the (expensive) view per address. It is an interface +// so tests can fake it without constructing a real query.View. +type NonceView interface { + // GetNonce returns the EOA's account nonce (the next nonce to use) at this + // view's state. Named GetNonce — not GetNextNonce — because this interface is + // satisfied directly by flow-go's query.View, whose method is GetNonce. + GetNonce(address gethCommon.Address) (uint64, error) +} + +// NonceProvider returns the next nonce of the given EOA address. The transaction +// mempool uses it to determine the expected next nonce. type NonceProvider interface { - // GetNonce returns the current nonce of the given EOA address. + // GetNextNonce returns the account nonce of the given EOA — its transaction + // count, i.e. the next nonce the EOA should use (matches eth_getTransactionCount). // // A non-nil error represents an EXCEPTION, not an expected condition: // the underlying read is a local state-index lookup that should not // fail under normal operation. Callers must therefore treat an error // as a hard failure (reject the transaction / abort the operation) // rather than a routine, recoverable condition to swallow. - GetNonce(address gethCommon.Address) (uint64, error) + GetNextNonce(address gethCommon.Address) (uint64, error) + + // GetBlockView returns a NonceView over the latest indexed EVM state. A + // non-nil error is an EXCEPTION, same contract as GetNextNonce. + GetBlockView() (NonceView, error) } // LocalNonceProvider reads the EOA nonce from the latest height of the -// local state index. +// local state index. It caches the built block view and reuses it while the +// indexed height is unchanged (see GetBlockView). type LocalNonceProvider struct { chainID flowGo.ChainID registerStore *pebble.RegisterStorage blocks storage.BlockIndexer + + // mu guards the cached view below. The cached view is shared across reads; + // callers that read it concurrently must serialize (the mempool does, via + // its queueMux). + mu sync.Mutex + cachedView NonceView + cachedHeight uint64 } var _ NonceProvider = &LocalNonceProvider{} @@ -45,10 +71,23 @@ func NewLocalNonceProvider( } } -func (p *LocalNonceProvider) GetNonce(address gethCommon.Address) (uint64, error) { +// GetBlockView returns a NonceView over the latest indexed EVM height. The view +// is cached and reused while the indexed height is unchanged, so a burst of +// reads within one block — many Add calls, or a collectDueBatches pass — builds +// the (expensive) view only once. It is rebuilt when a new block is indexed. +// Reuse is safe because an EOA's on-chain nonce cannot change without a new +// block being indexed. +func (p *LocalNonceProvider) GetBlockView() (NonceView, error) { height, err := p.blocks.LatestEVMHeight() if err != nil { - return 0, err + return nil, err + } + + p.mu.Lock() + defer p.mu.Unlock() + + if p.cachedView != nil && p.cachedHeight == height { + return p.cachedView, nil } viewProvider := query.NewViewProvider( @@ -60,6 +99,18 @@ func (p *LocalNonceProvider) GetNonce(address gethCommon.Address) (uint64, error ) view, err := viewProvider.GetBlockView(height) + if err != nil { + return nil, err + } + + p.cachedView = view + p.cachedHeight = height + + return view, nil +} + +func (p *LocalNonceProvider) GetNextNonce(address gethCommon.Address) (uint64, error) { + view, err := p.GetBlockView() if err != nil { return 0, err } diff --git a/services/requester/tx_mempool.go b/services/requester/tx_mempool.go index 811c67e9..3ca7b1db 100644 --- a/services/requester/tx_mempool.go +++ b/services/requester/tx_mempool.go @@ -3,6 +3,8 @@ package requester import ( "context" "encoding/hex" + "fmt" + "math" "sort" "sync" "time" @@ -19,6 +21,96 @@ import ( "github.com/onflow/flow-evm-gateway/services/requester/keystore" ) +// This file implements TxMemPool, a nonce-aware transaction mempool that +// decides when and how to submit EVM transactions to Flow. It treats the EOA's +// nonce from the local state index as the on-chain frontier and reasons about +// each incoming nonce relative to that frontier and to what it has already sent. +// +// BEHAVIOR SPEC — every case the pool handles, written as a checklist that maps +// to code and to tests. +// +// (The examples below assume the EOA's next expected nonce is N unless stated.) +// +// Submission paths +// 1. Fast path: a transaction whose nonce is the next expected nonce, arriving +// to an empty queue with nothing in flight and submission spacing +// satisfied, is submitted synchronously inside Add — zero added latency. +// Example: expected nonce 5, an empty queue, tx with nonce 5 arrives → it +// is sent immediately, nothing is queued. +// 2. Burst batching: when the fast path does not apply, transactions queue +// per-EOA. A sliding collection window (TxCollectionWindow, reset on each +// arrival) decides when a burst is complete; a flush deadline anchored at +// the FIRST enqueue (TxSubmissionSpacing) caps how long a continuously +// resetting window can defer the flush. +// Example: nonce 5 fast-paths and is sent; nonces 6,7,8 then arrive a few +// ms apart while the submission-spacing gap since that send is still open, +// so they cannot fast-path and queue instead. The window keeps resetting as +// they arrive, and they flush together as one batch once arrivals pause for +// TxCollectionWindow (or the deadline is hit) AND spacing has elapsed. (A +// burst whose lead nonce is itself the next expected one on an empty, idle +// queue would fast-path that first tx — case 1 — not queue it.) +// 3. Consecutive-prefix flush: on flush, the longest run of consecutive nonces +// starting at the expected nonce is sent as ONE Cadence transaction, capped +// at TxMaxBatchSize. A nonce gap splits the queue: only the prefix before +// the first gap is sent; the post-gap remainder waits for a later tick. +// Example: txs with nonces 1,2,3,5,6,7 arrive (gap at 4); the queue holds +// 1,2,3,5,6,7; the flush sends a batch of 1,2,3 while 5,6,7 wait in the +// queue for nonce 4 to arrive (or to age out via case 7). +// 4. Submission spacing: consecutive Cadence submissions for one EOA are kept +// at least TxSubmissionSpacing apart, so two Flow transactions land in +// different blocks and cannot be reordered by Collection Nodes. +// Example: a batch is sent at t=0; the next batch for the same EOA is held +// until t=TxSubmissionSpacing even if it is already due. +// +// Holding and eventual disposal of out-of-order transactions +// 5. Queue (gap ahead): a future nonce within the accepted window is held +// until the gap fills or it ages out. +// Example: expected nonce 5, tx with nonce 7 arrives (5 and 6 missing) → +// 7 is held, not sent. +// 6. Stale pruning: a held tx whose nonce has fallen below the on-chain +// frontier (e.g. filled via another gateway) can never execute and is +// dropped with a WARN log before it would burn fees. +// Example: nonce 3 sits in the queue; the frontier advances to 5 (3 and 4 +// were filled elsewhere) → 3 is dropped with a WARN, never submitted. +// 7. TTL submit-anyway: a held tx whose head gap never fills within TxPoolTTL +// is submitted ANYWAY (capped at TxMaxBatchSize) rather than dropped, so +// its failure is observable on-chain instead of a silent disappearance. +// Example: nonce 7 is held while 5,6 never arrive; after TxPoolTTL, 7 is +// submitted and fails on-chain with a nonce mismatch (visible on flowscan). +// 8. Idle-queue retention: a queue with no held txs and no activity for +// idleQueueRetention is removed to bound memory. +// +// Rejections (returned synchronously from Add; the tx is never queued) +// 9. Duplicate: same nonce AND same tx hash as one already queued → +// ErrDuplicateTransaction (cheapest check; no index read). +// Example: tx with nonce 5 is queued; the identical tx (same hash) is +// submitted again → rejected. (A same-nonce tx with a DIFFERENT hash is +// not a duplicate — it replaces the queued one, last write wins.) +// 10. In-flight: nonce at or below the highest nonce already sent (still in +// flight or already ack'd) → ErrInFlightNonce; re-accepting would burn +// Flow fees on a guaranteed nonce-mismatch. +// Example: nonces 5,6 were just sent and are in flight; a new tx with +// nonce 6 arrives → rejected. +// 11. Too-low: nonce below the on-chain frontier (already used) → +// ErrNonceTooLow. Always enforced. +// Example: frontier is 5, tx with nonce 4 arrives → rejected. +// 12. Too-high: nonce more than TxMaxNonceGap beyond the frontier → +// ErrNonceTooHigh. Only enforced when TxMaxNonceGap > 0. +// Example: frontier 5, TxMaxNonceGap 500, tx with nonce 600 arrives → +// rejected (it cannot execute until ~595 intervening nonces are filled). +// +// 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 +// stale prune) — never nothing. See logSubmission. The one exception is +// shutdown: held-but-not-yet-submitted txs are discarded without a WARN when +// the pool's context is cancelled (no graceful drain — see processQueues). +// - Failure handling: a failed submission drops the batch (clients resubmit) +// 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. + // heldTx is a transaction held in the mempool, waiting for its // collection window to elapse or its nonce gap to be filled. type heldTx struct { @@ -36,7 +128,7 @@ func selectConsecutivePrefix( expectedNonce uint64, maxBatch int, ) []heldTx { - prefix := make([]heldTx, 0) + prefix := make([]heldTx, 0, maxBatch) for nonce := expectedNonce; len(prefix) < maxBatch; nonce++ { tx, ok := txs[nonce] if !ok { @@ -47,6 +139,22 @@ func selectConsecutivePrefix( return prefix } +// deleteByNonce removes every transaction in batch from txs, keyed by nonce. +func deleteByNonce(txs map[uint64]heldTx, batch []heldTx) { + for _, htx := range batch { + delete(txs, htx.nonce) + } +} + +// txHashHexes returns the hex tx hashes of a batch, for structured log fields. +func txHashHexes(txs []heldTx) []string { + hashes := make([]string, len(txs)) + for i, htx := range txs { + hashes[i] = htx.txHash.Hex() + } + return hashes +} + // selectExpired returns the held transactions older than ttl, sorted by // nonce ascending. func selectExpired( @@ -54,7 +162,7 @@ func selectExpired( now time.Time, ttl time.Duration, ) []heldTx { - expired := make([]heldTx, 0) + expired := make([]heldTx, 0, len(txs)) for _, tx := range txs { if now.Sub(tx.enqueuedAt) > ttl { expired = append(expired, tx) @@ -68,13 +176,239 @@ func selectExpired( // txMemPoolTickInterval is the resolution at which due queues are // scanned and flushed. Deadlines are therefore honored with up to this -// much slack, which is acceptable relative to the 300ms collection window. +// much slack, which is acceptable relative to the collection window. const txMemPoolTickInterval = 50 * time.Millisecond +// fastPathSubmitTimeout bounds how long a single fast-path submission may take. +// The fast path holds the pool-wide queueMux across the whole Flow submission +// (see the note on TxMemPool), so without a bound a hung Access-node call would +// block every other EOA's Add and the background flush loop for as long as the +// caller's context allows — up to RpcRequestTimeout (120s by default). This is a +// liveness safety ceiling, NOT a latency SLA: normal submits complete well +// within it and release the lock immediately; only a genuinely stalled call is +// cut off (and its tx is dropped-and-logged for the client to resubmit). +const fastPathSubmitTimeout = 10 * time.Second + // idleQueueRetention is how long a queue with no held transactions and no // recent activity is kept before being removed, to bound memory usage. const idleQueueRetention = time.Minute +// 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): +// below every real nonce and never "at or above" one, so callers need no set +// checks. +type nonceWrapper struct { + v uint64 + set bool +} + +func toNonceWrapper(v uint64) nonceWrapper { return nonceWrapper{v: v, set: true} } + +// atLeast reports whether this nonce is set and >= n. An unset nonce (-∞) is +// never >= a real nonce, so it returns false. +func (w nonceWrapper) atLeast(n uint64) bool { return w.set && w.v >= n } + +// is reports whether this nonce is set and exactly equals n. +func (w nonceWrapper) is(n uint64) bool { return w.set && w.v == n } + +// max returns the greater of two optional nonces, treating unset as -∞. +func (w nonceWrapper) max(other nonceWrapper) nonceWrapper { + if !w.set { + return other + } + if other.set && other.v > w.v { + return other + } + return w +} + +// nonceVerdict is what classify decides should happen to an incoming nonce. +type nonceVerdict int + +const ( + // nonceNextExpected: nonce == expectedNonce; eligible for immediate submit. + nonceNextExpected nonceVerdict = iota + // nonceInFlight: nonce is at or below one we have already sent (still in + // flight or already ack'd). Re-accepting it would burn Flow fees on a + // guaranteed nonce-mismatch, so it is rejected. + nonceInFlight + // nonceTooLow: nonce is below the on-chain frontier — already used, can + // never execute. Rejected up front. + nonceTooLow + // nonceTooHigh: nonce is more than maxNonceGap beyond the on-chain frontier; + // it cannot execute until the gap fills, so it is rejected up front. + nonceTooHigh + // nonceQueue: a future nonce beyond the expected one (a gap ahead) but within + // the accepted window; hold it. + nonceQueue +) + +// normalizeNonceGap maps the configured max nonce gap to the value stored in a +// nonceTracker. The config value 0 ("no upper bound") becomes math.MaxUint64, so +// classify can use a single distance comparison without a per-call "gap > 0" +// guard. Any non-zero value is used as-is. +func normalizeNonceGap(configGap uint64) uint64 { + if configGap == 0 { + return math.MaxUint64 + } + return configGap +} + +// nonceTracker is the per-EOA submission-state machine. It records the nonce +// facts the mempool reasons about and answers "what should happen to an +// incoming nonce?" (classify) and "what is the next nonce to submit?" +// (expectedNonce), so callers never compare raw fields directly. +// +// The tracker has no lock of its own: all methods assume the pool's queueMux is +// held. Serialized access is guaranteed by the single submit goroutine plus +// Add running under that same lock (see the TxMemPool docstring). +type nonceTracker struct { + // localNextNonce is the EOA's next expected nonce per the local state + // index (the on-chain frontier). A cache of a fact about the chain, refreshed + // from a fresh read via refreshNextNonce — not a record of our own sends. + localNextNonce uint64 + // submitting is the highest consecutive nonce sent to Flow but not yet ack'd. It marks the + // window between detaching a batch and its submit result returning, and + // strictly means "a network call is in flight": it is cleared the moment that + // call returns, by markSubmitted on success or rollbackSubmitting on failure. + // Unset when no submission is outstanding. + // it is also used to filter incoming txs to accept only whose nonce is bigger than this, + // otherwise would be rejected + submitting nonceWrapper + // lastConsecutivelySubmitted is the highest nonce consecutively and + // successfully submitted (ack'd). Only consecutive-prefix batches advance it; + // TTL-expiry (gapped) batches do NOT, so it never includes a nonce past a gap. + // Unset before the EOA's first success. + lastConsecutivelySubmitted nonceWrapper + // maxNonceGap is how far above localNextNonce a nonce may be before it is + // rejected as too-high. math.MaxUint64 means no upper bound: the config value + // TxMaxNonceGap, where 0 = "unbounded", is normalized to it at construction + // (see normalizeNonceGap) so classify needs no per-call "is a gap configured?" + // guard. It does NOT affect the too-low check (a nonce below localNextNonce + // is always rejected). Set once when the queue is created. + maxNonceGap uint64 +} + +// inFlight reports whether a submission is outstanding (sent, not yet ack'd). +func (n *nonceTracker) inFlight() bool { return n.submitting.set } + +// highestSent returns the highest nonce we have already sent — whether still in +// flight or already ack'd (unset if neither). Re-accepting a nonce at or below +// it would burn Flow fees on a guaranteed nonce-mismatch. +func (n *nonceTracker) highestSent() nonceWrapper { + return n.lastConsecutivelySubmitted.max(n.submitting) +} + +// expectedNonce is the next nonce eligible for submission: one past the highest +// nonce already sent, or the on-chain frontier, whichever is higher. +func (n *nonceTracker) expectedNonce() uint64 { + if hi := n.highestSent(); hi.atLeast(n.localNextNonce) { + return hi.v + 1 + } + return n.localNextNonce +} + +// classify returns the verdict for an incoming nonce, refreshing the cached +// on-chain frontier as part of the decision. +// +// A nonce at or below our highest already-sent nonce is an in-flight/duplicate +// retry, decided from local state alone with no index read. This case is checked +// FIRST because reading the frontier builds a block view (the expensive step), +// which the retry path must avoid. +// +// Any other nonce is beyond what we've sent, so the remaining verdicts +// (too-low/too-high/next-expected) need the frontier, which is read and +// refreshed. A read error is an exception (a local state-index read should not +// fail) and is returned so the caller can reject the transaction. Pruning of any +// now-stale queued txs is deferred to collectDueBatches. +// +// Callers must hold queueMux. +func (n *nonceTracker) classify( + nonce uint64, + np NonceProvider, + from gethCommon.Address, +) (nonceVerdict, error) { + // At or below the highest nonce we've already sent: in flight or submitted. + if n.highestSent().atLeast(nonce) { + return nonceInFlight, nil + } + + // Beyond what we've sent: refresh the frontier for the remaining verdicts. + nextNonce, err := np.GetNextNonce(from) + if err != nil { + return 0, fmt.Errorf("reading next nonce for %s: %w", from.Hex(), err) + } + n.refreshNextNonce(nextNonce) + + // Below the on-chain frontier: already used, can never execute. + if nonce < n.localNextNonce { + return nonceTooLow, nil + } + // More than maxNonceGap beyond the frontier: cannot execute until the gap + // fills. A behind (stale) index can only make this over-strict, which is + // acceptable — the gateway is catching up. maxNonceGap is normalized + // (math.MaxUint64 = unbounded), so no "is a gap configured?" guard is needed. + // We compare the distance rather than localNextNonce+maxNonceGap (which + // could overflow): the too-low check above guarantees nonce >= localNextNonce + // so the subtraction never underflows, and the distance is at most MaxUint64 + // so the unbounded case never trips. + if nonce-n.localNextNonce > n.maxNonceGap { + return nonceTooHigh, nil + } + if nonce == n.expectedNonce() { + return nonceNextExpected, nil + } + return nonceQueue, nil +} + +// markSubmitting records that nonces up to highNonce have been sent but not yet +// ack'd. Called when a batch is detached for async submission in +// collectDueBatches. The synchronous fast path in Add skips it: Add holds the +// lock across the whole submit, so there is no concurrency window to guard. +func (n *nonceTracker) markSubmitting(highNonce uint64) { + n.submitting = toNonceWrapper(highNonce) +} + +// clearSubmittingIf retracts the in-flight marker, but only if it still refers +// to highNonce (a newer submission may have replaced it once submissions run +// concurrently). Shared by the success (markSubmitted) and failure +// (rollbackSubmitting) acks — both fire when the network call returns. +func (n *nonceTracker) clearSubmittingIf(highNonce uint64) { + if n.submitting.is(highNonce) { + n.submitting = nonceWrapper{} + } +} + +// markSubmitted acks a successful submission: it advances the consecutively- +// submitted nonce and clears the in-flight marker. It runs the moment the +// submission succeeds, under the lock, rather than waiting for the index to +// confirm — so `submitting` strictly means "a network call is outstanding" and +// clears as soon as the call returns. The cost is one quick lock per successful +// submission, accepted in exchange for a state machine that is easy to reason +// about. +// +// Both updates are guarded so a stale ack cannot corrupt newer state: +// lastConsecutivelySubmitted only advances (never regresses), and submitting is +// cleared only if it still refers to this batch. +func (n *nonceTracker) markSubmitted(highNonce uint64) { + n.lastConsecutivelySubmitted = n.lastConsecutivelySubmitted.max(toNonceWrapper(highNonce)) + n.clearSubmittingIf(highNonce) +} + +// rollbackSubmitting clears the in-flight marker after a failed submission. +// Because a failure never advances lastConsecutivelySubmitted, there is nothing +// else to undo: the next flush recomputes expectedNonce from the unchanged +// frontier. +func (n *nonceTracker) rollbackSubmitting(highNonce uint64) { + n.clearSubmittingIf(highNonce) +} + +// refreshNextNonce updates the cached on-chain frontier from a fresh index read. +func (n *nonceTracker) refreshNextNonce(nextNonce uint64) { + n.localNextNonce = nextNonce +} + // 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 @@ -85,24 +419,20 @@ type eoaQueue struct { // not flushed until the current time has passed this instant: it marks the // end of the sliding collection window, NOT a deadline to act before. collectionWindowEndsAt time.Time - // flushDeadline is firstEnqueue + TxSubmissionSpacing. It caps how long - // a continuously-resetting collection window can defer a flush. There is - // deliberately no separate "hard cap" knob: TxSubmissionSpacing serves - // both purposes (see PR #965 discussion). + // flushDeadline is firstEnqueue + TxSubmissionSpacing. It caps how long a + // continuously-resetting collection window can defer a flush. There is + // deliberately no separate "hard cap" knob: TxSubmissionSpacing serves both + // purposes. flushDeadline time.Time - // lastSubmittedAt is when the last Cadence tx for this EOA was submitted. + // lastSubmittedAt is when the last Cadence tx for this EOA was submitted + // (used for submission spacing). lastSubmittedAt time.Time - // lastSubmittedNonce is the highest nonce included in the last submission. - // Only meaningful while hasInFlight is true. (Kept next to lastSubmittedAt: - // "submitted" and "sent" mean the same action here.) - lastSubmittedNonce uint64 // lastActivity is when this EOA was last touched — a transaction received // (Add) or a batch flushed (collectDueBatches). It bounds memory: a queue // with no held txs and no activity past idleQueueRetention is removed. lastActivity time.Time - // hasInFlight reports whether a submission exists that the local index - // has not yet confirmed (index nonce <= lastSubmittedNonce). - hasInFlight bool + // nonces is the submission-state machine for this EOA. + nonces nonceTracker } // isEmpty reports whether the queue holds no transactions. Callers must hold @@ -117,40 +447,22 @@ func (q *eoaQueue) size() int { return len(q.txs) } -// TxMemPool is a `TxPool` implementation that uses the EOA nonce from -// the local state index to decide when and how to submit transactions to the -// Flow network. +// TxMemPool is a `TxPool` implementation that uses the EOA nonce from the local +// state index to decide when and how to submit transactions to the Flow +// network. The full behavior — fast path, burst batching, gap handling, TTL +// submit-anyway, and the rejection rules — is enumerated in the behavior spec +// at the top of this file. // -// Fast path: a transaction carrying the expected next nonce, with an empty -// queue, nothing in flight, and submission spacing satisfied, is submitted -// IMMEDIATELY — zero added latency for the common case. +// `TxSubmissionSpacing` deliberately serves two roles at once: (a) the minimum +// gap between consecutive Cadence submissions for one EOA (so two Flow +// transactions land in different blocks and cannot be reordered by Collection +// Nodes) and (b) the flush deadline anchored at first enqueue. There is +// intentionally NO separate hard-cap knob. // -// Otherwise transactions queue per-EOA. A sliding collection window -// (`TxCollectionWindow`, reset on each arrival) decides when a burst is -// complete. `TxSubmissionSpacing` is BOTH (a) the minimum gap between -// consecutive Cadence submissions for the same EOA (so two Flow transactions -// land in different blocks and cannot be reordered by Collection Nodes) and -// (b) the flush deadline anchored at first enqueue (caps a -// continuously-resetting window). There is deliberately NO separate hard-cap -// knob. -// -// On flush, the longest consecutive nonce prefix starting at the expected -// nonce (from the local index, advanced past any in-flight submission) is -// submitted, capped at `TxMaxBatchSize`. -// -// Out-of-order transactions are held until the gap fills, the local index -// advances past them (then they are stale and pruned), or `TxPoolTTL` -// expires — on expiry they are submitted anyway so the failure is observable -// on-chain rather than a silent drop. -// -// A nonce already submitted and still in flight is rejected with -// `ErrInFlightNonce`, since a duplicate would burn Flow fees on a guaranteed -// nonce-mismatch failure. -// -// Note on locking: fast-path submissions hold the pool-wide queue lock for -// the duration of one Flow submission, trading cross-EOA throughput for the -// simplicity of atomic state updates; a per-EOA lock is the known upgrade -// path if contention shows up. +// Note on locking: fast-path submissions hold the pool-wide queue lock for the +// duration of one Flow submission, trading cross-EOA throughput for the +// simplicity of atomic state updates; a per-EOA lock is the known upgrade path +// if contention shows up. type TxMemPool struct { *SingleTxPool nonceProvider NonceProvider @@ -159,6 +471,11 @@ type TxMemPool struct { // submitBatch performs the actual Flow submission. It defaults to // submitTxBatch and exists as a field so tests can inject a fake. submitBatch func(ctx context.Context, txs []heldTx) 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 + // rather than wall-clock sleeps. + now func() time.Time } var _ TxPool = &TxMemPool{} @@ -184,6 +501,7 @@ func NewTxMemPool( SingleTxPool: singleTxPool, nonceProvider: nonceProvider, queues: make(map[gethCommon.Address]*eoaQueue), + now: time.Now, } pool.submitBatch = pool.submitTxBatch @@ -220,124 +538,132 @@ func (t *TxMemPool) Add( q, ok := t.queues[from] if !ok { - q = &eoaQueue{txs: make(map[uint64]heldTx)} + // A fresh queue's other fields are intentionally left at their zero + // values, each safe to read as-is: + // - the nonceTracker's nonceWrapper fields read as "unset", so nonce 0 + // is never mistaken for a real submission (see nonceWrapper); + // - collectionWindowEndsAt and flushDeadline are read only once the + // queue is non-empty (collectDueBatches skips empty queues), and the + // first enqueue below sets them before that — their zero value is + // never observed; + // - lastSubmittedAt MAY be read while still zero (the fast path checks + // spacing immediately after creation), but spacingElapsed treats zero + // as "never submitted" → spacing satisfied, which is exactly right; + // - lastActivity is set unconditionally on the next line. + // Only maxNonceGap needs seeding from config. + q = &eoaQueue{ + txs: make(map[uint64]heldTx), + nonces: nonceTracker{maxNonceGap: normalizeNonceGap(t.config.TxMaxNonceGap)}, + } t.queues[from] = q } - now := time.Now() + now := t.now() + // The EOA was "touched" even if this turns out to be a duplicate, so record + // activity here to keep the idle-queue retention clock accurate. q.lastActivity = now - userTx := heldTx{ - txPayload: hexEncodedTx, - txHash: tx.Hash(), - nonce: tx.Nonce(), - enqueuedAt: now, - } - - // Reject obvious cases before reading the index nonce — each read builds a - // full block view, so when we already know we will reject the transaction - // we must not pay that cost. - - // Reject an exact duplicate of a transaction already in the queue. + // Reject an exact duplicate of a transaction already in the queue (cheapest + // check; needs no index read). if existing, ok := q.txs[tx.Nonce()]; ok && existing.txHash == tx.Hash() { return errs.ErrDuplicateTransaction } - // Reject a nonce that has been submitted and is still in flight: it - // would burn Flow fees on a guaranteed nonce-mismatch failure. A nonce at - // or below lastSubmittedNonce while hasInFlight is inherently in flight, so - // rejecting it here is correct. - // - // Note (zhangchiqing): ErrInFlightNonce covers a nonce at/below the last - // in-flight nonce. A nonce strictly below the indexed (already-used) nonce - // is NOT separately distinguished here — doing so would require an extra - // index read on every Add. Such a transaction is instead pruned by - // pruneStaleTxs on the background loop, or fails observably on-chain. - if q.hasInFlight && tx.Nonce() <= q.lastSubmittedNonce { - return errs.ErrInFlightNonce + // Classify the nonce, reading the on-chain frontier only when needed (see + // classify). A read error is an exception — reject rather than routing + // through the queue path. + verdict, err := q.nonces.classify(tx.Nonce(), t.nonceProvider, from) + if err != nil { + return err } - // Read the index nonce — an expensive operation that builds a full block - // view — at most once per Add, and only when it can change the decision: - // to clear a stale in-flight marker, and/or to evaluate the fast path for - // an empty, spacing-satisfied queue. - if q.hasInFlight || (q.isEmpty() && t.spacingElapsed(q, now)) { - indexNonce, nonceErr := q.queryAndRefreshInFlight(t.nonceProvider, from) - if nonceErr != nil { - // A nonce lookup failure is an exception, not an expected - // condition: this is a local state-index read that should not - // fail under normal operation. The gateway is in an unknown - // state, so reject the transaction rather than silently routing - // it through the queue path. - return nonceErr + // Handle every verdict. The rejections return up front, before allocating any + // held state (a transaction we are about to reject must not cost a heldTx). + // The two keep verdicts build the heldTx and either fast-path or enqueue it. + // An unhandled verdict is a programming error, so panic. + switch verdict { + case nonceInFlight: + return errs.ErrInFlightNonce + case nonceTooLow: + return errs.ErrNonceTooLow + case nonceTooHigh: + return errs.ErrNonceTooHigh + case nonceQueue: + // A gap ahead within the accepted window — hold it for the background loop. + held := heldTx{ + txPayload: hexEncodedTx, + txHash: tx.Hash(), + nonce: tx.Nonce(), + enqueuedAt: now, } - - // We deliberately do NOT prune stale txs (nonce < indexNonce) here, - // even though the index may have just advanced: pruning is deferred to - // collectDueBatches, which walks every queued tx anyway, so repeating - // it per-Add would be redundant work. - - // Fast path: the queue is empty, nothing is in flight (the marker may - // have just been cleared above), spacing is satisfied and this tx is - // exactly the next expected nonce. Submit right away — zero added - // latency for the common case. - if q.isEmpty() && !q.hasInFlight && - t.spacingElapsed(q, now) && tx.Nonce() == indexNonce { - if submitErr := t.submitBatch(ctx, []heldTx{userTx}); submitErr != nil { - // Submission failed: leave queue state untouched so the EOA - // is neither marked in flight nor rate-limited behind a tx - // that never landed. + t.enqueue(q, held, now) + return nil + case nonceNextExpected: + held := heldTx{ + txPayload: hexEncodedTx, + txHash: tx.Hash(), + nonce: tx.Nonce(), + enqueuedAt: now, + } + // Fast path: an empty queue with nothing in flight and spacing satisfied + // submits immediately — zero added latency. The lock is held across the + // whole submit, so there is no concurrency window: no need to mark + // "submitting" first; on success we record the ack, on failure we leave + // the EOA untouched. Otherwise fall back to enqueue for the background loop. + if q.isEmpty() && !q.nonces.inFlight() && t.spacingElapsed(q, now) { + batch := []heldTx{held} + // Bound the submit so a hung call cannot pin queueMux indefinitely + // (see fastPathSubmitTimeout). + submitCtx, cancel := context.WithTimeout(ctx, fastPathSubmitTimeout) + submitErr := t.submitBatch(submitCtx, batch) + cancel() + t.logSubmission(from, batch, flushReasonFastPath, q.nonces.localNextNonce, submitErr) + if submitErr != nil { return submitErr } - q.lastSubmittedAt = time.Now() - q.lastSubmittedNonce = tx.Nonce() - q.hasInFlight = true + q.nonces.markSubmitted(tx.Nonce()) + q.lastSubmittedAt = t.now() return nil } - // On an unexpected nonce, fall through to the queue path. + t.enqueue(q, held, now) + return nil + default: + panic(fmt.Sprintf("unhandled nonce verdict: %d", verdict)) } +} - // Enqueue. A same-nonce, different-payload resubmission replaces the - // queued transaction (last write wins), matching mempool semantics. +// enqueue holds a transaction for the background flush loop, (re)arming the +// collection window. A same-nonce, different-payload resubmission replaces the +// queued transaction (last write wins), matching mempool semantics. Callers must +// hold queueMux. +func (t *TxMemPool) enqueue(q *eoaQueue, held heldTx, now time.Time) { wasEmpty := q.isEmpty() - q.txs[tx.Nonce()] = userTx + q.txs[held.nonce] = held q.collectionWindowEndsAt = now.Add(t.config.TxCollectionWindow) // Anchor the flush deadline at the FIRST enqueue only. Re-arming it on a - // same-nonce replacement would let a client defer the flush indefinitely - // by resubmitting one held transaction before each deadline. + // same-nonce replacement would let a client defer the flush indefinitely by + // resubmitting one held transaction before each deadline. if wasEmpty { q.flushDeadline = now.Add(t.config.TxSubmissionSpacing) } - - return nil } -// refreshInFlight clears the in-flight marker once the local index has -// advanced past the last submitted nonce. Callers must hold the pool's -// queueMux. -func (q *eoaQueue) refreshInFlight(indexNonce uint64) { - if q.hasInFlight && indexNonce > q.lastSubmittedNonce { - q.hasInFlight = false - } -} - -// queryAndRefreshInFlight reads the EOA's current nonce from the local index -// and clears the in-flight marker if the index has advanced past the last -// submitted nonce, returning the index nonce. Callers must hold queueMux. -func (q *eoaQueue) queryAndRefreshInFlight( - np NonceProvider, - from gethCommon.Address, -) (uint64, error) { - indexNonce, err := np.GetNonce(from) - if err != nil { - return 0, err - } - q.refreshInFlight(indexNonce) - return indexNonce, nil -} - -// spacingElapsed reports whether enough time has passed since the last -// Cadence submission for this EOA. Callers must hold queueMux. +// spacingElapsed reports whether enough time has passed since the last Cadence +// submission for this EOA. Callers must hold queueMux. +// +// Timing caveat: for background batches, lastSubmittedAt is stamped at COLLECTION +// time (collectPrefix/collectExpired), not when the detached submit actually +// returns. Since one tick's batches are submitted sequentially afterward, a +// later EOA's stamp can predate its real submit by the cumulative submit latency +// ahead of it, so its NEXT batch's effective spacing can be short by that skew. +// The skew is bounded by (concurrently-due EOAs ahead) x (per-submit latency), +// which is small for the current few-EOA deployment and accepted. The +// collection-time stamp is also load-bearing the other way: it is what gates a +// follow-up batch from being collected while the previous one is still in flight +// (the spacing check runs before prefix selection, and the in-flight marker does +// not by itself stop the next consecutive prefix from being selected). A precise +// fix would keep this gate and additionally re-stamp on submit completion in +// reconcileSubmission. func (t *TxMemPool) spacingElapsed(q *eoaQueue, now time.Time) bool { return q.lastSubmittedAt.IsZero() || now.Sub(q.lastSubmittedAt) >= t.config.TxSubmissionSpacing @@ -348,14 +674,30 @@ func (t *TxMemPool) spacingElapsed(q *eoaQueue, now time.Time) bool { type flushWork struct { from gethCommon.Address txs []heldTx - // inFlight is true for consecutive-prefix batches, which optimistically - // advance lastSubmittedNonce/hasInFlight on the queue and must therefore - // roll those back if the submission fails (see rollbackFailedSubmission). - // TTL-expiry batches are not marked in flight and must never clear the - // marker. - inFlight bool + // needsReconcile is true when reconcileSubmission must update the tracker for + // this batch. Consecutive-prefix batches set it: they mark the nonceTracker + // "submitting" and must reconcile it once the submit returns (markSubmitted on + // success, rollbackSubmitting on failure — see reconcileSubmission). TTL-expiry + // batches leave it false: they never mark the tracker, so there is nothing to + // reconcile. + needsReconcile bool + // reason is why this batch was flushed, recorded purely for the submission + // log (see logSubmission): flushReasonPrefix for a consecutive-prefix flush, + // flushReasonTTL for a TTL-expiry submit-anyway. + reason string + // localNextNonce is the on-chain next expected nonce observed when the batch was + // collected. It is logged on drop so a "lost transaction" report can be + // debugged against where the chain actually was. + localNextNonce uint64 } +// flushReason values label why a batch was submitted, for the submission log. +const ( + flushReasonFastPath = "fast-path" + flushReasonPrefix = "consecutive-prefix" + flushReasonTTL = "ttl-expiry" +) + func (t *TxMemPool) processQueues(ctx context.Context) { ticker := time.NewTicker(txMemPoolTickInterval) defer ticker.Stop() @@ -363,6 +705,14 @@ func (t *TxMemPool) processQueues(ctx context.Context) { for { select { case <-ctx.Done(): + // Shutdown is NOT a graceful drain: any transactions still held in a + // queue (waiting on their collection window, gap fill, or TTL) are + // dropped without a WARN, which is the one gap in the no-silent-drops + // invariant — it holds during steady-state operation, not across a + // shutdown/restart. This is acceptable because clients resubmit, and + // a held tx has not been sent to Flow (nothing on-chain to reconcile + // against). If this ever needs to change, drain due+held batches here + // before returning rather than relying on resubmission. return case <-ticker.C: for _, w := range t.collectDueBatches() { @@ -377,36 +727,114 @@ func (t *TxMemPool) processQueues(ctx context.Context) { } } -// submitWork submits one detached batch. On failure it rolls back the state -// the queue committed optimistically when the batch was collected; on success -// there is nothing to do — that optimistic state already reflects the -// submission. +// submitWork submits one detached batch, records its fate (logSubmission), and +// reconciles the queue's nonce state once the network call returns. func (t *TxMemPool) submitWork(ctx context.Context, w flushWork) error { err := t.submitBatch(ctx, w.txs) - if err != nil { - t.rollbackFailedSubmission(w) - } + t.logSubmission(w.from, w.txs, w.reason, w.localNextNonce, err) + t.reconcileSubmission(w, err) return err } -// rollbackFailedSubmission re-opens an EOA after a failed flush. -// collectDueBatches optimistically marks a consecutive-prefix batch in flight -// and advances lastSubmittedNonce BEFORE the network call. If that call fails, -// the batch's transactions are dropped (already counted and logged by -// submitBatch) and never reach the chain — so without this rollback every -// resubmission of those nonces would be rejected with ErrInFlightNonce, and the -// index would never advance past them to clear the marker: the EOA would be -// permanently wedged. +// batchLogFields attaches to e the structured fields common to every per-batch +// lifecycle log — submission failure, stale prune, and TTL submit-anyway — so +// the no-silent-drops fields stay consistent and greppable across all three +// sites (eoa, tx hashes, nonce range, batch size, the on-chain frontier). The +// caller creates e at the desired level, adds any site-specific fields (reason, +// error, expected-nonce, ...), and calls Msg. The nonce range is computed as the +// true min/max so it is correct even when txs is not sorted (e.g. a stale-prune +// batch collected in map order). +func batchLogFields( + e *zerolog.Event, + from gethCommon.Address, + txs []heldTx, + nextNonce uint64, +) *zerolog.Event { + var lowNonce, highNonce uint64 + for i, htx := range txs { + if i == 0 || htx.nonce < lowNonce { + lowNonce = htx.nonce + } + if i == 0 || htx.nonce > highNonce { + highNonce = htx.nonce + } + } + return e. + Str("eoa", from.Hex()). + Strs("tx-hashes", txHashHexes(txs)). + Uint64("low-nonce", lowNonce). + Uint64("high-nonce", highNonce). + Int("batch-size", len(txs)). + Uint64("local-next-nonce", nextNonce) +} + +// logSubmission records the fate of a submitted batch so a transaction is never +// silently lost. This is the observability half of the no-silent-drops +// invariant: for any tx id you can either find it on-chain (sent) OR find a WARN +// log here (dropped) — never nothing. +// +// - On a Flow submit FAILURE the batch's EVM transactions are dropped (we do +// not retry — clients resubmit), so we WARN with the full batch context +// (batchLogFields) plus the flush reason and the error. +// - On SUCCESS we emit a lighter DEBUG line (eoa + nonce range) so a sent +// batch is traceable without the noise of a warning. +// +// txs is assumed nonce-ascending (selectConsecutivePrefix / selectExpired and +// the single-tx fast path all satisfy this), so txs[0] is the low nonce. +func (t *TxMemPool) logSubmission( + from gethCommon.Address, + txs []heldTx, + reason string, + localNextNonce uint64, + submitErr error, +) { + if len(txs) == 0 { + return + } + + if submitErr != nil { + batchLogFields(t.logger.Warn(), from, txs, localNextNonce). + Str("reason", reason). + Err(submitErr). + Msg("Flow submission failed, EVM transactions dropped") + return + } + + t.logger.Debug(). + Str("eoa", from.Hex()). + Uint64("low-nonce", txs[0].nonce). + Uint64("high-nonce", txs[len(txs)-1].nonce). + Int("batch-size", len(txs)). + Str("reason", reason). + Msg("submitted EVM transactions to Flow") +} + +// reconcileSubmission updates the EOA's nonceTracker after a detached +// consecutive-prefix submission returns. collectDueBatches marked the batch +// "submitting" (under the lock) before the network call; this records the +// outcome: // -// The marker is cleared only when it still belongs to the failed batch — a -// newer submission may have replaced it while the failed one was on the wire. -// Only in-flight (prefix) batches are rolled back; TTL-expiry batches never set -// the marker. +// - On SUCCESS it advances the consecutively-submitted nonce now +// (markSubmitted), under the lock, rather than waiting for the index to +// confirm — keeping `submitting` meaning strictly "a network call is +// outstanding". // -// lastSubmittedAt is deliberately NOT restored: a brief, self-correcting -// spacing delay after a rare submission failure is harmless and not worth the -// extra bookkeeping. -func (t *TxMemPool) rollbackFailedSubmission(w flushWork) { +// - On FAILURE the batch's transactions are dropped (already counted and +// logged) and never reach the chain, so the "submitting" marker is cleared +// (rollbackSubmitting). Without this, resubmissions of those nonces would be +// rejected as in-flight forever — the index never advances to clear the +// marker, so the EOA would be permanently wedged. rollbackSubmitting clears +// the marker only if it still refers to this batch (a newer submission may +// have replaced it). lastSubmittedAt is deliberately NOT restored: a brief, +// self-correcting spacing delay after a rare failure is harmless. +// +// 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) { + if !w.needsReconcile { + return + } + t.queueMux.Lock() defer t.queueMux.Unlock() @@ -414,155 +842,187 @@ func (t *TxMemPool) rollbackFailedSubmission(w flushWork) { if !ok { return } - if w.inFlight && q.hasInFlight && q.lastSubmittedNonce == w.txs[len(w.txs)-1].nonce { - q.hasInFlight = false + + highNonce := w.txs[len(w.txs)-1].nonce + if submitErr != nil { + q.nonces.rollbackSubmitting(highNonce) + return } + q.nonces.markSubmitted(highNonce) } -// collectDueBatches selects, under the queue lock, every batch that is due -// for submission, updates the queue state optimistically, and returns the -// detached work items. +// collectDueBatches selects, under the queue lock, every batch that is due for +// submission, updates the queue state optimistically, and returns the detached +// work items. The per-EOA decision lives in collectQueue. +// +// Each due EOA's nonce is read via GetNextNonce; the provider caches the block view +// by indexed height, so all reads in this pass (and across ticks at the same +// height) reuse one built view rather than rebuilding per address. func (t *TxMemPool) collectDueBatches() []flushWork { t.queueMux.Lock() defer t.queueMux.Unlock() - now := time.Now() - work := make([]flushWork, 0) - + now := t.now() + work := make([]flushWork, 0, len(t.queues)) for from, q := range t.queues { - if q.isEmpty() { - // Bound memory: drop queues with no held txs and no activity past - // the retention period. Any in-flight submission has long since - // resolved on-chain after this window, so discarding a lingering - // in-flight marker here is safe — a later transaction for the EOA - // creates a fresh queue and re-reads the index nonce. - if now.Sub(q.lastActivity) > idleQueueRetention { - delete(t.queues, from) - } - continue + if w, ok := t.collectQueue(from, q, now); ok { + work = append(work, w) } + } + t.reportSize() + return work +} - // Not due yet: both the sliding window and the flush deadline are - // still in the future. - if now.Before(q.collectionWindowEndsAt) && now.Before(q.flushDeadline) { - continue +// collectQueue evaluates one EOA's queue and returns its due batch, if any. It +// applies the gating checks (idle eviction, due-time, submission spacing, a +// fresh frontier read + stale prune) and then collects at most one batch: +// the consecutive prefix takes precedence, and only when a gap at the head +// blocks it does the TTL-expiry path run. The post-gap / over-cap remainder is +// left in the queue for a later tick — never merged with this batch, since a +// head gap would fail the whole Flow transaction. Callers must hold queueMux; +// it may evict an idle empty queue as a side effect. +func (t *TxMemPool) collectQueue( + from gethCommon.Address, + q *eoaQueue, + now time.Time, +) (flushWork, bool) { + if q.isEmpty() { + // Bound memory: drop queues with no held txs and no activity past the + // retention period. Any in-flight submission has long since resolved + // on-chain after this window, so discarding a lingering in-flight marker + // here is safe — a later transaction for the EOA creates a fresh queue + // and re-reads the index nonce. + if now.Sub(q.lastActivity) > idleQueueRetention { + delete(t.queues, from) } + return flushWork{}, false + } - // Safety gap since the previous submission not yet elapsed. - if !t.spacingElapsed(q, now) { - continue - } + // Not due yet: both the sliding window and the flush deadline are still in + // the future. + if now.Before(q.collectionWindowEndsAt) && now.Before(q.flushDeadline) { + return flushWork{}, false + } - indexNonce, err := t.nonceProvider.GetNonce(from) - if err != nil { - // Exception: a local state-index nonce read should not fail - // under normal operation. This is a background loop with no - // caller to reject the tx to, so skip this EOA for the current - // tick (its batch is deferred until the read succeeds) without - // aborting the whole flush for other EOAs. - t.logger.Error().Err(err).Str("eoa", from.Hex()). - Msg("unexpected failure reading nonce from local index, skipping EOA this tick") - continue - } + // Safety gap since the previous submission not yet elapsed. + if !t.spacingElapsed(q, now) { + return flushWork{}, false + } - q.refreshInFlight(indexNonce) + nextNonce, err := t.nonceProvider.GetNextNonce(from) + if err != nil { + // Exception: a local state-index nonce read should not fail under normal + // operation. This is a background loop with no caller to reject the tx + // to, so skip this EOA for the current tick (its batch is deferred until + // the read succeeds) without aborting the whole flush for other EOAs. + t.logger.Error().Err(err).Str("eoa", from.Hex()). + Msg("unexpected failure reading nonce from local index, skipping EOA this tick") + return flushWork{}, false + } - // Prune transactions that can never execute: their nonce is already - // used on-chain (e.g. filled via another gateway). They would only - // burn fees at TTL expiry. - t.pruneStaleTxs(q, from, indexNonce) + q.nonces.refreshNextNonce(nextNonce) - expected := indexNonce - if q.hasInFlight && q.lastSubmittedNonce+1 > expected { - expected = q.lastSubmittedNonce + 1 - } + // Prune transactions that can never execute: their nonce is already used + // on-chain (e.g. filled via another gateway). They would only burn fees at + // TTL expiry. + t.pruneStaleTxs(q, from, nextNonce) - // At most one batch is collected per EOA per tick. The consecutive - // prefix below takes precedence and `continue`s; only when there is no - // eligible prefix (a gap at the head) do we consider the TTL-expiry - // path. The post-gap / over-cap remainder is left in the queue and - // drained on a later tick, gated by submission spacing — it is never - // merged with this batch, since a head gap would make the whole Flow - // transaction fail. - prefix := selectConsecutivePrefix(q.txs, expected, t.config.TxMaxBatchSize) - if len(prefix) > 0 { - for _, htx := range prefix { - delete(q.txs, htx.nonce) - } - q.lastSubmittedNonce = prefix[len(prefix)-1].nonce - q.lastSubmittedAt = now - q.lastActivity = now - q.hasInFlight = true - if !q.isEmpty() { - // Re-arm for the remaining (post-gap or over-cap) txs. - q.collectionWindowEndsAt = now.Add(t.config.TxCollectionWindow) - q.flushDeadline = now.Add(t.config.TxSubmissionSpacing) - } - work = append(work, flushWork{ - from: from, - txs: prefix, - inFlight: true, - }) - continue - } + if w, ok := t.collectPrefix(from, q, now, nextNonce); ok { + return w, true + } + return t.collectExpired(from, q, now, nextNonce) +} - // No eligible prefix (gap at the head). Submit transactions held - // past their TTL anyway instead of dropping them. - // - // Rationale (no silent drops): submitting an unexecutable transaction - // produces a real, observable on-chain failure (operators can see the - // failed Flow transaction and its nonce-mismatch), whereas silently - // dropping it leaves no trace. The no-silent-drop requirement is the - // whole reason this pool exists (see PR #965 / DFNS), so an observable - // failure is strictly preferable to an invisible drop. - // - // Known tradeoff (zhangchiqing): a transaction whose nonce is far - // ahead of the index will still be submitted at TTL and burn fees on a - // guaranteed failure. Rejecting far-ahead nonces in Add is deliberately - // NOT done — it remains an open design question (how far ahead is "too - // far", and whether to add a knob for it), and is intentionally left - // out of this change rather than guessed at. - // - // Cap the batch at TxMaxBatchSize so a long-lived gap cannot produce an - // unbounded Flow transaction; the remainder drains on later ticks, - // gated by submission spacing. - expired := selectExpired(q.txs, now, t.config.TxPoolTTL) - if len(expired) > t.config.TxMaxBatchSize { - expired = expired[:t.config.TxMaxBatchSize] - } - if len(expired) > 0 { - for _, htx := range expired { - delete(q.txs, htx.nonce) - } - // Deliberately do NOT set hasInFlight/lastSubmittedNonce here: these - // nonces are out of order, and marking them in flight would - // corrupt the expected-nonce computation for future flushes. - q.lastSubmittedAt = now - q.lastActivity = now - txHashes := make([]string, len(expired)) - for i, htx := range expired { - txHashes[i] = htx.txHash.Hex() - } - t.logger.Warn().Strs("tx-hashes", txHashes).Str("eoa", from.Hex()). - Msg("nonce gap never filled within TTL, submitting held transactions anyway") - work = append(work, flushWork{ - from: from, - txs: expired, - }) - } +// collectPrefix detaches the longest consecutive nonce run starting at the +// expected nonce (capped at TxMaxBatchSize), marks it in flight, and re-arms the +// queue for any remainder. Returns false when a gap at the head leaves no +// eligible prefix. Callers must hold queueMux. +func (t *TxMemPool) collectPrefix( + from gethCommon.Address, + q *eoaQueue, + now time.Time, + nextNonce uint64, +) (flushWork, bool) { + prefix := selectConsecutivePrefix(q.txs, q.nonces.expectedNonce(), t.config.TxMaxBatchSize) + if len(prefix) == 0 { + return flushWork{}, false } - // Report the pool's memory footprint while still holding queueMux: the - // number of per-EOA queues and the total number of held transactions. - // Counting must happen under the lock since t.queues is mutated - // concurrently (and idle queues are pruned above). + deleteByNonce(q.txs, prefix) + // Optimistically mark the batch submitting; reconcileSubmission advances + // submitted on success or clears it on failure. + q.nonces.markSubmitting(prefix[len(prefix)-1].nonce) + q.lastSubmittedAt = now + q.lastActivity = now + if !q.isEmpty() { + // Re-arm for the remaining (post-gap or over-cap) txs. + q.collectionWindowEndsAt = now.Add(t.config.TxCollectionWindow) + q.flushDeadline = now.Add(t.config.TxSubmissionSpacing) + } + return flushWork{ + from: from, + txs: prefix, + needsReconcile: true, + reason: flushReasonPrefix, + localNextNonce: nextNonce, + }, true +} + +// collectExpired detaches transactions held past their TTL when a head gap +// blocks the prefix path, submitting them anyway (capped at TxMaxBatchSize) +// rather than dropping them. Returns false when nothing has expired. Callers +// must hold queueMux. +// +// Rationale (no silent drops): submitting an unexecutable transaction produces a +// real, observable on-chain failure (operators can see the failed Flow +// transaction and its nonce-mismatch), whereas silently dropping it leaves no +// trace. Avoiding silent drops is the whole reason this pool exists, so an +// observable failure is strictly preferable to an invisible drop. The batch is +// deliberately NOT marked in flight: these nonces are out of order (past a gap), +// and marking them would corrupt the expected-nonce computation for future +// flushes. +// +// Tradeoff: a tx whose nonce is far ahead of the index is still submitted at TTL +// and burns fees on a guaranteed failure. How far ahead a nonce may be is +// bounded instead at Add time by TxMaxNonceGap (see classify); when that is +// unset (0), far-ahead nonces reach here. +func (t *TxMemPool) collectExpired( + from gethCommon.Address, + q *eoaQueue, + now time.Time, + nextNonce uint64, +) (flushWork, bool) { + expired := selectExpired(q.txs, now, t.config.TxPoolTTL) + if len(expired) > t.config.TxMaxBatchSize { + expired = expired[:t.config.TxMaxBatchSize] + } + if len(expired) == 0 { + return flushWork{}, false + } + + deleteByNonce(q.txs, expired) + q.lastSubmittedAt = now + q.lastActivity = now + batchLogFields(t.logger.Warn(), from, expired, nextNonce). + Uint64("expected-nonce", q.nonces.expectedNonce()). + Msg("nonce gap never filled within TTL, submitting held transactions anyway") + return flushWork{ + from: from, + txs: expired, + reason: flushReasonTTL, + localNextNonce: nextNonce, + }, true +} + +// reportSize emits the pool's memory footprint: the number of per-EOA queues and +// the total number of held transactions. Callers must hold queueMux, since it +// reads t.queues (mutated concurrently, and pruned during collection). +func (t *TxMemPool) reportSize() { queuedTxs := 0 for _, q := range t.queues { queuedTxs += q.size() } t.collector.TxPoolSize(len(t.queues), queuedTxs) - - return work } // pruneStaleTxs removes queued transactions whose nonce is below the current @@ -571,18 +1031,18 @@ func (t *TxMemPool) collectDueBatches() []flushWork { func (t *TxMemPool) pruneStaleTxs( q *eoaQueue, from gethCommon.Address, - indexNonce uint64, + nextNonce uint64, ) { - stale := make([]string, 0) + var stale []heldTx for nonce, htx := range q.txs { - if nonce < indexNonce { - stale = append(stale, htx.txHash.Hex()) - delete(q.txs, nonce) + if nonce < nextNonce { + stale = append(stale, htx) } } if len(stale) > 0 { - t.logger.Warn().Strs("tx-hashes", stale).Str("eoa", from.Hex()). - Msg("dropping stale transactions with nonce below indexed state") + deleteByNonce(q.txs, stale) + batchLogFields(t.logger.Warn(), from, stale, nextNonce). + Msg("dropping stale transactions with nonce below the on-chain frontier") } } @@ -601,6 +1061,11 @@ func (t *TxMemPool) submitTxBatch(ctx context.Context, txs []heldTx) error { } script := replaceAddresses(runTxScript, t.config.FlowNetworkID) + // On a build/send failure the batch's EVM transactions are dropped; count + // them here (the metric is reserved for Cadence build/submission errors) and + // return the error. The observable WARN drop log — with eoa, nonce range and + // flush reason — is emitted by logSubmission at the call site, which has that + // context; this keeps submitTxBatch a pure submission primitive. flowTx, err := t.buildTransaction( ctx, t.getReferenceBlock(), @@ -610,23 +1075,13 @@ func (t *TxMemPool) submitTxBatch(ctx context.Context, txs []heldTx) error { ) if err != nil { t.collector.TransactionsDropped(len(txs)) - t.logTxsDropped(txs, err, "failed to build Flow transaction, EVM transactions dropped") - return err + return fmt.Errorf("building Flow transaction: %w", err) } if err := t.client.SendTransaction(ctx, *flowTx); err != nil { t.collector.TransactionsDropped(len(txs)) - t.logTxsDropped(txs, err, "failed to send Flow transaction, EVM transactions dropped") - return err + return fmt.Errorf("sending Flow transaction: %w", err) } return nil } - -func (t *TxMemPool) logTxsDropped(txs []heldTx, err error, msg string) { - txHashes := make([]string, len(txs)) - for i, htx := range txs { - txHashes[i] = htx.txHash.Hex() - } - t.logger.Error().Err(err).Strs("tx-hashes", txHashes).Msg(msg) -} diff --git a/services/requester/tx_mempool_test.go b/services/requester/tx_mempool_test.go index c4bd83a1..6ee0f2e9 100644 --- a/services/requester/tx_mempool_test.go +++ b/services/requester/tx_mempool_test.go @@ -1,10 +1,13 @@ package requester import ( + "bytes" "context" "crypto/ecdsa" "errors" + "math" "math/big" + "sync" "testing" "time" @@ -107,10 +110,23 @@ type fakeNonceProvider struct { err error } +// GetNextNonce satisfies NonceProvider; GetNonce satisfies the NonceView that +// GetBlockView returns. Both read every EOA's nonce as f.nonce/f.err. +func (f *fakeNonceProvider) GetNextNonce(_ gethCommon.Address) (uint64, error) { + return f.nonce, f.err +} + func (f *fakeNonceProvider) GetNonce(_ gethCommon.Address) (uint64, error) { return f.nonce, f.err } +// GetBlockView returns the fake itself as the NonceView: a single fake reads +// every EOA's nonce as f.nonce/f.err, mirroring the production single-view +// read path. +func (f *fakeNonceProvider) GetBlockView() (NonceView, error) { + return f, nil +} + func newTestPool( np NonceProvider, submit func(context.Context, []heldTx) error, @@ -125,11 +141,34 @@ func newTestPool( }, nonceProvider: np, queues: make(map[gethCommon.Address]*eoaQueue), + now: time.Now, } pool.submitBatch = submit return pool } +// fakeClock is a controllable clock for driving the pool's time-based behavior +// in tests without wall-clock sleeps. It is safe for the single-goroutine unit +// tests here (Add and collectDueBatches are called synchronously). +type fakeClock struct { + mu sync.Mutex + t time.Time +} + +func newFakeClock(t time.Time) *fakeClock { return &fakeClock{t: t} } + +func (c *fakeClock) now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.t +} + +func (c *fakeClock) advance(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.t = c.t.Add(d) +} + func testPoolConfig() config.Config { return config.Config{ TxCollectionWindow: 100 * time.Millisecond, @@ -191,8 +230,38 @@ func Test_TxMemPool_FastPathSubmitsImmediately(t *testing.T) { q := pool.queues[from] require.NotNil(t, q) assert.Empty(t, q.txs) - assert.True(t, q.hasInFlight) - assert.Equal(t, uint64(0), q.lastSubmittedNonce) + // Explicit success update: submitting cleared, submitted advanced to 0. + assert.False(t, q.nonces.inFlight()) + assert.Equal(t, toNonceWrapper(0), q.nonces.lastConsecutivelySubmitted) +} + +// The fast-path submit must run under a bounded-deadline context so a hung +// Flow call cannot pin the pool-wide queueMux indefinitely (audit liveness +// finding). Before the bound, Add forwarded its raw context (here Background, +// with no deadline) straight into the submit. +func Test_TxMemPool_FastPathSubmitHasBoundedTimeout(t *testing.T) { + key, err := crypto.GenerateKey() + require.NoError(t, err) + + var deadline time.Time + var hasDeadline bool + pool := newTestPool( + &fakeNonceProvider{nonce: 0}, + func(ctx context.Context, _ []heldTx) error { + deadline, hasDeadline = ctx.Deadline() + return nil + }, + testPoolConfig(), + ) + + before := time.Now() + require.NoError(t, pool.Add(context.Background(), signedTestTx(t, key, 0, 1))) + after := time.Now() + + require.True(t, hasDeadline, "fast-path submit context must carry a deadline") + assert.Greater(t, deadline, before) + assert.LessOrEqual(t, deadline, after.Add(fastPathSubmitTimeout), + "deadline must be bounded by fastPathSubmitTimeout") } func Test_TxMemPool_UnexpectedNonceEnqueues(t *testing.T) { @@ -219,7 +288,7 @@ func Test_TxMemPool_UnexpectedNonceEnqueues(t *testing.T) { held, ok := q.txs[5] require.True(t, ok) assert.Equal(t, tx.Hash(), held.txHash) - assert.False(t, q.hasInFlight) + assert.False(t, q.nonces.inFlight()) } func Test_TxMemPool_NonceReadErrorRejectsTx(t *testing.T) { @@ -248,7 +317,7 @@ func Test_TxMemPool_NonceReadErrorRejectsTx(t *testing.T) { q := pool.queues[from] require.NotNil(t, q) assert.Empty(t, q.txs) - assert.False(t, q.hasInFlight) + assert.False(t, q.nonces.inFlight()) } func Test_TxMemPool_InFlightDuplicateRejected(t *testing.T) { @@ -270,6 +339,39 @@ func Test_TxMemPool_InFlightDuplicateRejected(t *testing.T) { assert.ErrorIs(t, err, errs.ErrInFlightNonce) } +// Case 9 (duplicate): re-adding the IDENTICAL transaction (same nonce AND same +// hash) of one already HELD in the queue is rejected with +// ErrDuplicateTransaction, while a same-nonce tx with a DIFFERENT hash replaces +// the queued one (last write wins). This exercises the duplicate check itself, +// distinct from the in-flight rejection above (which removes the tx from the +// queue before the second Add). +func Test_TxMemPool_DuplicateQueuedTxRejected(t *testing.T) { + key, err := crypto.GenerateKey() + require.NoError(t, err) + from := crypto.PubkeyToAddress(key.PublicKey) + + pool := newTestPool( + // Frontier 0, so a nonce-5 tx is out of order and is HELD (not submitted). + &fakeNonceProvider{nonce: 0}, + func(_ context.Context, _ []heldTx) error { return nil }, + testPoolConfig(), + ) + + tx := signedTestTx(t, key, 5, 1) + require.NoError(t, pool.Add(context.Background(), tx)) + + // Re-adding the identical tx (same hash) is a duplicate. + err = pool.Add(context.Background(), tx) + assert.ErrorIs(t, err, errs.ErrDuplicateTransaction) + + // A same-nonce, different-payload (different hash) tx is NOT a duplicate; it + // replaces the held one. + replacement := signedTestTx(t, key, 5, 2) + require.NotEqual(t, tx.Hash(), replacement.Hash()) + require.NoError(t, pool.Add(context.Background(), replacement)) + assert.Equal(t, replacement.Hash(), pool.queues[from].txs[5].txHash) +} + func Test_TxMemPool_FailedFlushDoesNotWedgeEOA(t *testing.T) { key, err := crypto.GenerateKey() require.NoError(t, err) @@ -296,25 +398,25 @@ func Test_TxMemPool_FailedFlushDoesNotWedgeEOA(t *testing.T) { work := pool.collectDueBatches() require.Len(t, work, 1) - assert.True(t, work[0].inFlight) + assert.True(t, work[0].needsReconcile) require.Len(t, work[0].txs, 2) // State was committed optimistically under the lock. q := pool.queues[from] - require.True(t, q.hasInFlight) - assert.Equal(t, uint64(1), q.lastSubmittedNonce) + require.True(t, q.nonces.inFlight()) + assert.Equal(t, toNonceWrapper(1), q.nonces.submitting) - // The submission fails; submitWork must reconcile the in-flight marker. + // The submission fails; submitWork must clear the in-flight marker. err = pool.submitWork(context.Background(), work[0]) require.ErrorIs(t, err, submitErr) - assert.False(t, q.hasInFlight) + assert.False(t, q.nonces.inFlight()) // A resubmission of the failed nonce must NOT be rejected as in flight. err = pool.Add(context.Background(), signedTestTx(t, key, 0, 2)) assert.NotErrorIs(t, err, errs.ErrInFlightNonce) } -func Test_RollbackFailedSubmission_OnlyClearsMatchingBatch(t *testing.T) { +func Test_ReconcileSubmission_OnlyReconcilesMatchingInFlightBatch(t *testing.T) { pool := newTestPool( &fakeNonceProvider{nonce: 0}, func(_ context.Context, _ []heldTx) error { return nil }, @@ -322,39 +424,42 @@ func Test_RollbackFailedSubmission_OnlyClearsMatchingBatch(t *testing.T) { ) from := gethCommon.HexToAddress("0xabc") pool.queues[from] = &eoaQueue{ - txs: map[uint64]heldTx{}, - hasInFlight: true, - lastSubmittedNonce: 7, + txs: map[uint64]heldTx{}, + nonces: nonceTracker{submitting: toNonceWrapper(7)}, } + submitErr := errors.New("network down") - // A different (newer) batch owns the marker: not cleared. - pool.rollbackFailedSubmission( - flushWork{from: from, txs: []heldTx{makeHeldTx(5, time.Time{})}, inFlight: true}, + // A different (newer) in-flight nonce owns the marker: not cleared. + pool.reconcileSubmission( + flushWork{from: from, txs: []heldTx{makeHeldTx(5, time.Time{})}, needsReconcile: true}, + submitErr, ) - assert.True(t, pool.queues[from].hasInFlight) + assert.True(t, pool.queues[from].nonces.inFlight()) - // A TTL-expiry batch (inFlight false) never clears the marker, even if its - // last nonce coincides with lastSubmittedNonce. - pool.rollbackFailedSubmission( - flushWork{from: from, txs: []heldTx{makeHeldTx(7, time.Time{})}, inFlight: false}, + // A TTL-expiry batch (needsReconcile false) never touches the tracker. + pool.reconcileSubmission( + flushWork{from: from, txs: []heldTx{makeHeldTx(7, time.Time{})}, needsReconcile: false}, + submitErr, ) - assert.True(t, pool.queues[from].hasInFlight) + assert.True(t, pool.queues[from].nonces.inFlight()) // The failed in-flight batch still owns the marker: cleared. - pool.rollbackFailedSubmission( - flushWork{from: from, txs: []heldTx{makeHeldTx(7, time.Time{})}, inFlight: true}, + pool.reconcileSubmission( + flushWork{from: from, txs: []heldTx{makeHeldTx(7, time.Time{})}, needsReconcile: true}, + submitErr, ) - assert.False(t, pool.queues[from].hasInFlight) + assert.False(t, pool.queues[from].nonces.inFlight()) // Unknown EOA: no panic. - pool.rollbackFailedSubmission( - flushWork{from: gethCommon.HexToAddress("0xdef"), txs: []heldTx{makeHeldTx(7, time.Time{})}, inFlight: true}, + pool.reconcileSubmission( + flushWork{from: gethCommon.HexToAddress("0xdef"), txs: []heldTx{makeHeldTx(7, time.Time{})}, needsReconcile: true}, + submitErr, ) } -// A successful submission leaves the optimistically-committed in-flight state -// intact — there is no success-path reconciliation to undo it. -func Test_TxMemPool_SuccessfulFlushKeepsInFlight(t *testing.T) { +// A successful submission advances the consecutively-submitted nonce and clears +// the in-flight marker (the explicit success update). +func Test_TxMemPool_SuccessfulFlushMarksSubmitted(t *testing.T) { key, err := crypto.GenerateKey() require.NoError(t, err) from := crypto.PubkeyToAddress(key.PublicKey) @@ -381,12 +486,94 @@ func Test_TxMemPool_SuccessfulFlushKeepsInFlight(t *testing.T) { require.NoError(t, pool.submitWork(context.Background(), work[0])) q := pool.queues[from] - assert.True(t, q.hasInFlight) - assert.Equal(t, uint64(1), q.lastSubmittedNonce) + assert.False(t, q.nonces.inFlight()) + assert.Equal(t, toNonceWrapper(1), q.nonces.lastConsecutivelySubmitted) } -// Fix 1: a failed fast-path submission must not rate-limit the EOA via -// lastSubmittedAt, and must leave nothing in flight. +// No silent drops: a failed flush submission must produce an observable WARN log +// carrying the dropped tx hashes and enough context to debug a "lost +// transaction" report (eoa, nonce range, next nonce, batch size, reason, +// error). +func Test_TxMemPool_FailedSubmitLogsDropWarning(t *testing.T) { + key, err := crypto.GenerateKey() + require.NoError(t, err) + from := crypto.PubkeyToAddress(key.PublicKey) + + var logBuf bytes.Buffer + submitErr := errors.New("network down") + pool := newTestPool( + &fakeNonceProvider{nonce: 0}, + func(_ context.Context, _ []heldTx) error { return submitErr }, + testPoolConfig(), + ) + pool.logger = zerolog.New(&logBuf) + + tx0 := signedTestTx(t, key, 0, 1) + tx1 := signedTestTx(t, key, 1, 1) + past := time.Now().Add(-time.Second) + pool.queues[from] = &eoaQueue{ + txs: map[uint64]heldTx{ + 0: {txHash: tx0.Hash(), nonce: 0, enqueuedAt: past}, + 1: {txHash: tx1.Hash(), nonce: 1, enqueuedAt: past}, + }, + collectionWindowEndsAt: past, + flushDeadline: past, + } + + work := pool.collectDueBatches() + require.Len(t, work, 1) + require.Error(t, pool.submitWork(context.Background(), work[0])) + + out := logBuf.String() + assert.Contains(t, out, `"level":"warn"`, "drop must be observable at WARN level") + assert.Contains(t, out, tx0.Hash().Hex(), "dropped tx hash must be logged") + assert.Contains(t, out, tx1.Hash().Hex(), "dropped tx hash must be logged") + assert.Contains(t, out, from.Hex(), "eoa must be logged") + assert.Contains(t, out, `"local-next-nonce":0`, "next nonce must be logged") + assert.Contains(t, out, `"batch-size":2`, "batch size must be logged") + assert.Contains(t, out, "network down", "submit error must be logged") +} + +// A successful submission is traceable via a DEBUG log carrying the eoa and +// nonce range, so a sent batch can be found in logs without a warning. +func Test_TxMemPool_SuccessfulSubmitLogsDebug(t *testing.T) { + key, err := crypto.GenerateKey() + require.NoError(t, err) + from := crypto.PubkeyToAddress(key.PublicKey) + + var logBuf bytes.Buffer + pool := newTestPool( + &fakeNonceProvider{nonce: 0}, + func(_ context.Context, _ []heldTx) error { return nil }, + testPoolConfig(), + ) + pool.logger = zerolog.New(&logBuf) + + tx0 := signedTestTx(t, key, 0, 1) + tx1 := signedTestTx(t, key, 1, 1) + past := time.Now().Add(-time.Second) + pool.queues[from] = &eoaQueue{ + txs: map[uint64]heldTx{ + 0: {txHash: tx0.Hash(), nonce: 0, enqueuedAt: past}, + 1: {txHash: tx1.Hash(), nonce: 1, enqueuedAt: past}, + }, + collectionWindowEndsAt: past, + flushDeadline: past, + } + + work := pool.collectDueBatches() + require.Len(t, work, 1) + require.NoError(t, pool.submitWork(context.Background(), work[0])) + + out := logBuf.String() + assert.Contains(t, out, `"level":"debug"`, "successful send must be traceable at DEBUG level") + assert.Contains(t, out, from.Hex(), "eoa must be logged") + assert.Contains(t, out, `"low-nonce":0`) + assert.Contains(t, out, `"high-nonce":1`) +} + +// A failed fast-path submission must not rate-limit the EOA via lastSubmittedAt, +// and must leave nothing in flight. func Test_TxMemPool_FailedFastPathLeavesNoState(t *testing.T) { key, err := crypto.GenerateKey() require.NoError(t, err) @@ -404,12 +591,13 @@ func Test_TxMemPool_FailedFastPathLeavesNoState(t *testing.T) { q := pool.queues[from] require.NotNil(t, q) - assert.False(t, q.hasInFlight) + assert.False(t, q.nonces.inFlight()) + assert.False(t, q.nonces.lastConsecutivelySubmitted.set, "failed submission must not advance submitted") assert.True(t, q.lastSubmittedAt.IsZero(), "failed submission must not stamp lastSubmittedAt") } -// Fix 2: resubmitting a single held tx with the same nonce must not re-arm the -// flush deadline anchored at first enqueue. +// Resubmitting a single held tx with the same nonce must not re-arm the flush +// deadline anchored at first enqueue. func Test_TxMemPool_SameNonceReplacementKeepsFlushDeadline(t *testing.T) { key, err := crypto.GenerateKey() require.NoError(t, err) @@ -430,7 +618,7 @@ func Test_TxMemPool_SameNonceReplacementKeepsFlushDeadline(t *testing.T) { assert.Equal(t, firstDeadline, pool.queues[from].flushDeadline) } -// Fix 3: TTL-expiry flushes are capped at TxMaxBatchSize. +// TTL-expiry flushes are capped at TxMaxBatchSize. func Test_TxMemPool_TTLFlushCappedAtMaxBatch(t *testing.T) { key, err := crypto.GenerateKey() require.NoError(t, err) @@ -458,14 +646,14 @@ func Test_TxMemPool_TTLFlushCappedAtMaxBatch(t *testing.T) { work := pool.collectDueBatches() require.Len(t, work, 1) assert.Len(t, work[0].txs, 3, "TTL flush must be capped at TxMaxBatchSize") - assert.False(t, work[0].inFlight) + assert.False(t, work[0].needsReconcile) // Lowest nonces drained first; remainder stays queued. assert.Equal(t, uint64(10), work[0].txs[0].nonce) assert.Len(t, pool.queues[from].txs, 4) } -// Fix 4: a queue emptied without ever submitting (e.g. all txs pruned) must -// still age out via lastActivity rather than leaking forever. +// A queue emptied without ever submitting (e.g. all txs pruned) must still age +// out via lastActivity rather than leaking forever. func Test_TxMemPool_EmptyQueueAgesOut(t *testing.T) { pool := newTestPool( &fakeNonceProvider{nonce: 0}, @@ -474,11 +662,11 @@ func Test_TxMemPool_EmptyQueueAgesOut(t *testing.T) { ) from := gethCommon.HexToAddress("0xabc") - // Empty queue, never submitted (lastSubmittedAt zero), but in flight and + // Empty queue, never submitted, but with a lingering in-flight marker and // last active beyond the retention window: must be removed. pool.queues[from] = &eoaQueue{ txs: map[uint64]heldTx{}, - hasInFlight: true, + nonces: nonceTracker{submitting: toNonceWrapper(5)}, lastActivity: time.Now().Add(-2 * idleQueueRetention), } pool.collectDueBatches() @@ -512,7 +700,7 @@ func (c *countingCollector) TxPoolSize(queues int, queued int) { c.txPoolSizeSet = true } -// Stale txs (nonce below the indexed nonce) must be pruned from the queue, but +// Stale txs (nonce below the next nonce) must be pruned from the queue, but // pruning must NOT increment the TransactionsDropped metric — that metric is // reserved for build/submission errors of the Cadence transaction to Flow. func Test_TxMemPool_PruneStaleDoesNotIncrementDropped(t *testing.T) { @@ -574,18 +762,342 @@ func Test_TxMemPool_CollectDueBatchesReportsSize(t *testing.T) { assert.Equal(t, 3, collector.txPoolQueued) } -func Test_RefreshInFlight(t *testing.T) { - q := &eoaQueue{hasInFlight: true, lastSubmittedNonce: 3} +func Test_NonceTracker_Classify(t *testing.T) { + // classify reads the on-chain frontier from the provider, so the frontier + // is supplied via `frontier` (not set directly on the tracker). In-flight + // cases return before the read, so their frontier value is irrelevant. + tests := []struct { + name string + tracker nonceTracker + frontier uint64 + nonce uint64 + want nonceVerdict + }{ + {"nonce at the frontier is next-expected", nonceTracker{}, 5, 5, nonceNextExpected}, + {"gap ahead queues", nonceTracker{}, 5, 7, nonceQueue}, + {"below index is too low (no gap configured)", nonceTracker{}, 5, 3, nonceTooLow}, + {"nonce 0 is next-expected on a zero tracker", nonceTracker{}, 0, 0, nonceNextExpected}, + {"at submitted is in-flight", nonceTracker{lastConsecutivelySubmitted: toNonceWrapper(6)}, 5, 6, nonceInFlight}, + {"below submitted is in-flight", nonceTracker{lastConsecutivelySubmitted: toNonceWrapper(6)}, 5, 4, nonceInFlight}, + {"next after submitted is expected", nonceTracker{lastConsecutivelySubmitted: toNonceWrapper(6)}, 5, 7, nonceNextExpected}, + {"at submitting is in-flight", nonceTracker{submitting: toNonceWrapper(8)}, 5, 8, nonceInFlight}, + {"next after submitting is expected", nonceTracker{submitting: toNonceWrapper(8)}, 5, 9, nonceNextExpected}, + {"index ahead of submitted: expected follows index", nonceTracker{lastConsecutivelySubmitted: toNonceWrapper(6)}, 10, 10, nonceNextExpected}, + {"index ahead of submitted: below index is too low", nonceTracker{lastConsecutivelySubmitted: toNonceWrapper(6)}, 10, 8, nonceTooLow}, + // Range checks (maxNonceGap > 0). + {"gap: index nonce is next-expected", nonceTracker{maxNonceGap: 50}, 5, 5, nonceNextExpected}, + {"gap: below index is too low", nonceTracker{maxNonceGap: 50}, 5, 4, nonceTooLow}, + {"gap: at the upper bound is accepted (queued)", nonceTracker{maxNonceGap: 50}, 5, 55, nonceQueue}, + {"gap: beyond the upper bound is too high", nonceTracker{maxNonceGap: 50}, 5, 56, nonceTooHigh}, + {"gap: in-flight takes precedence over too-low", nonceTracker{maxNonceGap: 50, lastConsecutivelySubmitted: toNonceWrapper(6)}, 5, 3, nonceInFlight}, + {"no gap: far-ahead nonce queues (no upper bound)", nonceTracker{}, 5, 100_000, nonceQueue}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + tracker := tc.tracker + // classify expects a normalized gap (0 in the table means "unbounded", + // stored as math.MaxUint64), matching how queues are constructed. + tracker.maxNonceGap = normalizeNonceGap(tracker.maxNonceGap) + got, err := tracker.classify( + tc.nonce, + &fakeNonceProvider{nonce: tc.frontier}, + gethCommon.HexToAddress("0xabc"), + ) + require.NoError(t, err) + assert.Equal(t, tc.want, got) + }) + } +} + +// normalizeNonceGap maps the config "0 = no upper bound" to MaxUint64 (so +// classify needs no per-call gap>0 guard) and passes other values through. +func Test_NormalizeNonceGap(t *testing.T) { + assert.Equal(t, uint64(math.MaxUint64), normalizeNonceGap(0), "0 = unbounded maps to MaxUint64") + assert.Equal(t, uint64(500), normalizeNonceGap(500)) +} + +// With a configured max gap, Add rejects out-of-range nonces up front. +func Test_TxMemPool_RejectsNonceOutOfRange(t *testing.T) { + key, err := crypto.GenerateKey() + require.NoError(t, err) + + cfg := testPoolConfig() + cfg.TxMaxNonceGap = 50 + + submitCalls := 0 + pool := newTestPool( + &fakeNonceProvider{nonce: 100}, // on-chain frontier at nonce 100 + func(_ context.Context, _ []heldTx) error { submitCalls++; return nil }, + cfg, + ) + ctx := context.Background() + + // Below the frontier: already used. + require.ErrorIs(t, pool.Add(ctx, signedTestTx(t, key, 99, 1)), errs.ErrNonceTooLow) + + // More than maxNonceGap (50) ahead of the frontier. + require.ErrorIs(t, pool.Add(ctx, signedTestTx(t, key, 200, 1)), errs.ErrNonceTooHigh) + + // Within the accepted window (ahead of expected, but <= frontier+gap): held. + require.NoError(t, pool.Add(ctx, signedTestTx(t, key, 120, 1))) + + assert.Zero(t, submitCalls, "out-of-order/rejected txs are never fast-path submitted") +} + +func Test_NonceTracker_ExpectedNonce(t *testing.T) { + assert.Equal(t, uint64(5), (&nonceTracker{localNextNonce: 5}).expectedNonce()) + assert.Equal(t, uint64(7), + (&nonceTracker{localNextNonce: 5, lastConsecutivelySubmitted: toNonceWrapper(6)}).expectedNonce()) + assert.Equal(t, uint64(9), + (&nonceTracker{localNextNonce: 5, submitting: toNonceWrapper(8)}).expectedNonce()) + // The on-chain frontier wins when it is ahead of our own sends. + assert.Equal(t, uint64(10), + (&nonceTracker{localNextNonce: 10, lastConsecutivelySubmitted: toNonceWrapper(6)}).expectedNonce()) +} + +func Test_NonceTracker_Transitions(t *testing.T) { + n := &nonceTracker{localNextNonce: 5} + + // markSubmitting sets the in-flight marker. + n.markSubmitting(7) + assert.True(t, n.inFlight()) + assert.Equal(t, toNonceWrapper(7), n.submitting) + + // markSubmitted advances submitted and clears submitting. + n.markSubmitted(7) + assert.False(t, n.inFlight()) + assert.Equal(t, toNonceWrapper(7), n.lastConsecutivelySubmitted) + + // rollbackSubmitting only clears a matching in-flight nonce. + n.markSubmitting(9) + n.rollbackSubmitting(8) // non-matching: no-op + assert.True(t, n.inFlight()) + n.rollbackSubmitting(9) // matching: cleared + assert.False(t, n.inFlight()) + // A rollback never disturbs the consecutively-submitted nonce. + assert.Equal(t, toNonceWrapper(7), n.lastConsecutivelySubmitted) + + // refreshNextNonce updates the cached frontier. + n.refreshNextNonce(12) + assert.Equal(t, uint64(12), n.localNextNonce) +} + +// A stale success ack (e.g. once submissions run concurrently and a newer batch +// has already been marked in flight) must not clear the newer in-flight marker +// or regress the consecutively-submitted nonce. +func Test_NonceTracker_StaleMarkSubmittedDoesNotClobber(t *testing.T) { + n := &nonceTracker{} + + n.markSubmitting(6) // batch A {.,6} in flight + n.markSubmitting(8) // batch B {7,8} replaces it as the newer in-flight batch + + // Batch A's (stale) success ack arrives: it must not clear B's marker. + n.markSubmitted(6) + assert.True(t, n.inFlight(), "newer in-flight marker must survive a stale ack") + assert.Equal(t, toNonceWrapper(8), n.submitting) + assert.Equal(t, toNonceWrapper(6), n.lastConsecutivelySubmitted) + + // Batch B's ack then clears the marker and advances submitted to 8. + n.markSubmitted(8) + assert.False(t, n.inFlight()) + assert.Equal(t, toNonceWrapper(8), n.lastConsecutivelySubmitted) + + // An out-of-order ack for an older nonce must not regress submitted. + n.markSubmitted(7) + assert.Equal(t, toNonceWrapper(8), n.lastConsecutivelySubmitted, "submitted must never regress") +} - // Index has not advanced past the sent nonce: stays in flight. - q.refreshInFlight(3) - assert.True(t, q.hasInFlight) +// --- Clock-driven timing tests ------------------------------------------- +// These drive the collection window, flush deadline, submission spacing, TTL +// expiry and idle-queue retention through the real Add -> collectDueBatches +// path using an injected clock, rather than hand-building deadline timestamps. + +var timingClockBase = time.Unix(1_700_000_000, 0) + +// The sliding collection window is re-armed on every arrival. +func Test_TxMemPool_CollectionWindowResetsOnEachArrival(t *testing.T) { + key, err := crypto.GenerateKey() + require.NoError(t, err) + from := crypto.PubkeyToAddress(key.PublicKey) + + clk := newFakeClock(timingClockBase) + pool := newTestPool( + &fakeNonceProvider{nonce: 0}, + func(_ context.Context, _ []heldTx) error { return nil }, + testPoolConfig(), + ) + pool.now = clk.now + + // A gapped nonce (frontier 0, nonce 5) queues rather than fast-pathing. + require.NoError(t, pool.Add(context.Background(), signedTestTx(t, key, 5, 1))) + q := pool.queues[from] + require.NotNil(t, q) + assert.Equal(t, timingClockBase.Add(testPoolConfig().TxCollectionWindow), q.collectionWindowEndsAt) + firstDeadline := q.flushDeadline + + // A later arrival re-arms the window relative to its own arrival time, but + // must NOT re-arm the first-enqueue flush deadline. + clk.advance(40 * time.Millisecond) + require.NoError(t, pool.Add(context.Background(), signedTestTx(t, key, 6, 1))) + assert.Equal(t, clk.now().Add(testPoolConfig().TxCollectionWindow), q.collectionWindowEndsAt) + assert.Equal(t, firstDeadline, q.flushDeadline, "flush deadline stays anchored at first enqueue") +} - // Index advanced past the sent nonce: cleared. - q.refreshInFlight(4) - assert.False(t, q.hasInFlight) +// Submission spacing gates the background flush: a due batch is held until +// TxSubmissionSpacing has elapsed since the previous submission, then flushed. +func Test_TxMemPool_SpacingGateDefersThenFlushes(t *testing.T) { + key, err := crypto.GenerateKey() + require.NoError(t, err) + from := crypto.PubkeyToAddress(key.PublicKey) + + clk := newFakeClock(timingClockBase) + var submitted [][]heldTx + pool := newTestPool( + &fakeNonceProvider{nonce: 0}, + func(_ context.Context, txs []heldTx) error { + submitted = append(submitted, txs) + return nil + }, + testPoolConfig(), + ) + pool.now = clk.now + cfg := testPoolConfig() + + // nonce 0 fast-paths immediately and stamps lastSubmittedAt = base. + require.NoError(t, pool.Add(context.Background(), signedTestTx(t, key, 0, 1))) + require.Len(t, submitted, 1) + // nonces 1,2 arrive while spacing has not elapsed, so they queue. + require.NoError(t, pool.Add(context.Background(), signedTestTx(t, key, 1, 1))) + require.NoError(t, pool.Add(context.Background(), signedTestTx(t, key, 2, 1))) + + // Past the collection window but within the spacing gap: NOT flushed yet. + clk.advance(cfg.TxCollectionWindow + 50*time.Millisecond) + pool.collectDueBatches() + assert.Len(t, submitted, 1, "spacing gate must defer the flush") + assert.Len(t, pool.queues[from].txs, 2) + + // Once spacing has elapsed, the consecutive prefix {1,2} flushes. + clk.advance(cfg.TxSubmissionSpacing) + for _, w := range pool.collectDueBatches() { + require.NoError(t, pool.submitWork(context.Background(), w)) + } + require.Len(t, submitted, 2) + assert.Equal(t, []uint64{1, 2}, noncesOf(submitted[1])) + assert.Empty(t, pool.queues[from].txs) +} + +// The first-enqueue flush deadline forces a flush even while the (continuously +// re-armed) collection window is still in the future. +func Test_TxMemPool_FlushDeadlineForcesFlushWhileWindowOpen(t *testing.T) { + key, err := crypto.GenerateKey() + require.NoError(t, err) + from := crypto.PubkeyToAddress(key.PublicKey) - // No-op when nothing is in flight. - q.refreshInFlight(100) - assert.False(t, q.hasInFlight) + clk := newFakeClock(timingClockBase) + var submitted [][]heldTx + pool := newTestPool( + &fakeNonceProvider{nonce: 0}, + func(_ context.Context, txs []heldTx) error { + submitted = append(submitted, txs) + return nil + }, + testPoolConfig(), + ) + pool.now = clk.now + cfg := testPoolConfig() + + require.NoError(t, pool.Add(context.Background(), signedTestTx(t, key, 0, 1))) // fast-path + require.NoError(t, pool.Add(context.Background(), signedTestTx(t, key, 1, 1))) // queues; deadline = base + spacing + + // Just before the deadline, re-arm the window with a late arrival so the + // window remains in the future at flush time. + clk.advance(cfg.TxSubmissionSpacing - 50*time.Millisecond) + require.NoError(t, pool.Add(context.Background(), signedTestTx(t, key, 2, 1))) + q := pool.queues[from] + require.True(t, clk.now().Before(q.collectionWindowEndsAt), "window must still be open") + + // Cross the deadline (still before the window end): the deadline forces it. + clk.advance(60 * time.Millisecond) + require.True(t, clk.now().Before(q.collectionWindowEndsAt), "window still open at flush time") + require.False(t, clk.now().Before(q.flushDeadline), "deadline has passed") + for _, w := range pool.collectDueBatches() { + require.NoError(t, pool.submitWork(context.Background(), w)) + } + require.Len(t, submitted, 2) + assert.Equal(t, []uint64{1, 2}, noncesOf(submitted[1])) +} + +// A nonce stuck behind a permanent head gap is submitted anyway once TxPoolTTL +// elapses (driven by the clock via enqueuedAt), not silently dropped. +func Test_TxMemPool_TTLExpiryViaClock(t *testing.T) { + key, err := crypto.GenerateKey() + require.NoError(t, err) + from := crypto.PubkeyToAddress(key.PublicKey) + + clk := newFakeClock(timingClockBase) + var submitted []flushWork + pool := newTestPool( + &fakeNonceProvider{nonce: 0}, + func(_ context.Context, _ []heldTx) error { return nil }, + testPoolConfig(), + ) + pool.now = clk.now + cfg := testPoolConfig() + + // Frontier 0, nonce 10 — a permanent head gap, so the only exit is TTL. + require.NoError(t, pool.Add(context.Background(), signedTestTx(t, key, 10, 1))) + + // Before TTL: held, not submitted. + clk.advance(cfg.TxCollectionWindow + 50*time.Millisecond) + assert.Empty(t, pool.collectDueBatches()) + assert.Len(t, pool.queues[from].txs, 1) + + // Past TTL: submitted anyway, as a non-in-flight TTL batch. + clk.advance(cfg.TxPoolTTL) + work := pool.collectDueBatches() + require.Len(t, work, 1) + submitted = work + assert.False(t, submitted[0].needsReconcile) + assert.Equal(t, flushReasonTTL, submitted[0].reason) + assert.Equal(t, []uint64{10}, noncesOf(submitted[0].txs)) + assert.Empty(t, pool.queues[from].txs) +} + +// An empty queue with no activity past idleQueueRetention is evicted. +func Test_TxMemPool_IdleQueueEvictedViaClock(t *testing.T) { + key, err := crypto.GenerateKey() + require.NoError(t, err) + from := crypto.PubkeyToAddress(key.PublicKey) + + clk := newFakeClock(timingClockBase) + pool := newTestPool( + &fakeNonceProvider{nonce: 0}, + func(_ context.Context, _ []heldTx) error { return nil }, + testPoolConfig(), + ) + pool.now = clk.now + + // nonce 0 fast-paths and leaves an empty queue with lastActivity = base. + require.NoError(t, pool.Add(context.Background(), signedTestTx(t, key, 0, 1))) + require.NotNil(t, pool.queues[from]) + + // Within retention: kept. + clk.advance(idleQueueRetention / 2) + pool.collectDueBatches() + assert.NotNil(t, pool.queues[from]) + + // Past retention: evicted. + clk.advance(idleQueueRetention) + pool.collectDueBatches() + _, ok := pool.queues[from] + assert.False(t, ok, "idle empty queue must be evicted past retention") +} + +// noncesOf extracts the nonces of a batch in order, for concise assertions. +func noncesOf(txs []heldTx) []uint64 { + ns := make([]uint64, len(txs)) + for i, htx := range txs { + ns[i] = htx.nonce + } + return ns }