Prevent nonce gaps for untracked transactions (#646)

* Prevent nonce gaps for untracked transactions

* Update nonce management logic

* Update tests

---------

Co-authored-by: Augustus <[email protected]>

Author: dimriou

relayer/pkg/chainlink/txm/txm.go

@@ -23,7 +23,8 @@ import (
 )
 
 const (
-	MaxQueueLen = 1000
+	MaxQueueLen           = 1000
+	ConfirmationThreshold = 4
 )
 
 type TxManager interface {
@@ -188,7 +189,7 @@ func (txm *starktxm) broadcast(ctx context.Context, publicKey *felt.Felt, accoun
 		if accountNonceErr != nil {
 			return txhash, fmt.Errorf("failed to check account nonce during TxStore creation: %+w", accountNonceErr)
 		}
-		newTxStore, createErr := txm.accountStore.CreateTxStore(accountAddress, initialNonce)
+		newTxStore, createErr := txm.accountStore.CreateTxStore(accountAddress, initialNonce, txm.lggr)
 		if createErr != nil {
 			return txhash, fmt.Errorf("failed to create TxStore: %+w", createErr)
 		}
@@ -273,7 +274,7 @@ func (txm *starktxm) broadcast(ctx context.Context, publicKey *felt.Felt, accoun
 	broadcastTxnV3.InvokeTxnV3.ResourceBounds.L1Gas.MaxPricePerUnit = txm.updateMaxPriceUnitBounds(L1GasPrice, 150)
 	broadcastTxnV3.InvokeTxnV3.ResourceBounds.L2Gas.MaxPricePerUnit = txm.updateMaxPriceUnitBounds(L2GasPrice, 150)
 
-	txm.lggr.Infow("Set resource bounds", "L1MaxAmount", tx.ResourceBounds.L1Gas.MaxAmount, "L1MaxPricePerUnit", tx.ResourceBounds.L1Gas.MaxPricePerUnit)
+	txm.lggr.Infow("Set resource bounds", "L1MaxAmount", tx.ResourceBounds.L1Gas.MaxAmount, "L1MaxPricePerUnit", tx.ResourceBounds.L1Gas.MaxPricePerUnit, "FinalNonce", nonce)
 
 	L1DataGasConsumed := friEstimate.L1DataGasConsumed.BigInt(new(big.Int))
 	L1DataGasPrice := friEstimate.L1DataGasPrice.BigInt(new(big.Int))
@@ -293,17 +294,8 @@ func (txm *starktxm) broadcast(ctx context.Context, publicKey *felt.Felt, accoun
 	// finally, transmit the invoke
 	res, err := account.Provider.AddInvokeTransaction(execCtx, &broadcastTxnV3)
 	if err != nil {
-		// TODO: handle initial broadcast errors - what kind of errors occur?
-		var dataErr *starknetrpc.RPCError
-		var dataStr string
-		if !errors.As(err, &dataErr) {
-			return txhash, fmt.Errorf("failed to read EstimateFee error: %T %+v", err, err)
-		}
-		data := dataErr.Data
-		dataStr = fmt.Sprintf("%+v", data)
-		txm.lggr.Errorw("failed to invoke tx", "accountAddress", accountAddress, "error", err, "data", dataStr)
-
-		if strings.Contains(dataStr, RPCNonceErrMsg) {
+		txm.lggr.Errorw("failed to invoke tx", "accountAddress", accountAddress, "error", err)
+		if strings.Contains(err.Error(), RPCNonceErrMsg) {
 			// if we see an invalid nonce error at the broadcast stage, that means that we are out of sync.
 			// see the comment at resyncNonce for more details.
 			if resyncErr := txm.resyncNonce(ctx, client, accountAddress); resyncErr != nil {
@@ -362,56 +354,34 @@ func (txm *starktxm) confirmLoop() {
 				break
 			}
 
-			allUnconfirmedTxs := txm.accountStore.GetAllUnconfirmed()
-			for accountAddressStr, unconfirmedTxs := range allUnconfirmedTxs {
+			for _, accountAddressStr := range txm.accountStore.Accounts() {
 				accountAddress, err := new(felt.Felt).SetString(accountAddressStr)
 				// this should never occur because the acccount address string key was created from the account address felt.
 				if err != nil {
 					txm.lggr.Errorw("could not recreate account address felt", "accountAddress", accountAddressStr)
 					continue
 				}
-				for _, unconfirmedTx := range unconfirmedTxs {
-					hash := unconfirmedTx.Hash
-					f, err := starknetutils.HexToFelt(hash)
-					if err != nil {
-						txm.lggr.Errorw("invalid felt value", "hash", hash)
-						continue
-					}
-					response, err := client.Provider.GetTransactionStatus(ctx, f)
-
-					// tx can be rejected due to a nonce error. but we cannot know from the Starknet RPC directly  so we have to wait for
-					// a broadcasted tx to fail in order to fix the nonce errors
-
-					if err != nil {
-						txm.lggr.Errorw("failed to fetch transaction status", "hash", hash, "nonce", unconfirmedTx.Nonce, "error", err)
-						continue
-					}
-
-					finalityStatus := response.FinalityStatus
-					executionStatus := response.ExecutionStatus
-
-					// any finalityStatus other than received
-					if finalityStatus == starknetrpc.TxnStatus_Accepted_On_L1 || finalityStatus == starknetrpc.TxnStatus_Accepted_On_L2 || finalityStatus == starknetrpc.TxnStatus_Rejected {
-						txm.lggr.Debugw(fmt.Sprintf("tx confirmed: %s", finalityStatus), "hash", hash, "nonce", unconfirmedTx.Nonce, "finalityStatus", finalityStatus)
-						if err := txm.accountStore.GetTxStore(accountAddress).Confirm(unconfirmedTx.Nonce, hash); err != nil {
-							txm.lggr.Errorw("failed to confirm tx in TxStore", "hash", hash, "accountAddress", accountAddress, "error", err)
-						}
-					}
-
-					// currently, feeder client is only way to get rejected reason
-					if finalityStatus == starknetrpc.TxnStatus_Rejected {
-						// we assume that all rejected transactions results in a unused rejected nonce, so
-						// resync. see the comment at resyncNonce for more details.
-						if resyncErr := txm.resyncNonce(ctx, client, accountAddress); resyncErr != nil {
-							txm.lggr.Errorw("resync failed for rejected tx", "error", resyncErr)
-						}
-
-						go txm.logFeederError(ctx, hash, f)
-					}
-
-					if executionStatus == starknetrpc.TxnExecutionStatusREVERTED {
-						// TODO: get revert reason?
-						txm.lggr.Errorw("transaction reverted", "hash", hash)
+				nonce, err := client.AccountNonceLatest(ctx, accountAddress)
+				if err != nil {
+					txm.lggr.Errorf("failed to fetch latest nonce for account %v, err: %v", accountAddress, err)
+					continue
+				}
+				// Confirm all transactions with nonce lower than the latest.
+				confirmed, highestUnconfirmed := txm.accountStore.GetTxStore(accountAddress).Confirm(nonce)
+				txm.lggr.Infow("Confirmation loop", "accountAddress", accountAddress, "latestNonce", nonce,
+					"transactionsConfirmed", confirmed, "highestUnconfirmed", highestUnconfirmed)
+
+				// We add a maximum threshold between latest nonce and highest unconfirmed. This prevents the TXM from sending a very large
+				// number of unconfirmed transactions in the mempool and triggers a resync to prevent nonce gaps since the RPC responses are unreliable.
+				// The nonce stored here won't necessarily be picked up by the next transaction since there is a fast-forward functionality in broadcasting.
+				// But it ensures that if for whatever reason the diff between mined and uncofirmed transactions starts to grow, the TXM will be able to
+				// go back on the nonce and fill any nonce gaps.
+				hu := highestUnconfirmed.BigInt(new(big.Int))
+				n := nonce.BigInt(new(big.Int))
+				threshold := big.NewInt(ConfirmationThreshold)
+				if new(big.Int).Sub(hu, n).Cmp(threshold) == 1 {
+					if resyncErr := txm.resyncNonce(ctx, client, accountAddress); resyncErr != nil {
+						txm.lggr.Errorw("resync failed for rejected tx", "error", resyncErr)
 					}
 				}
 			}
@@ -424,22 +394,6 @@ func (txm *starktxm) confirmLoop() {
 	}
 }
 
-func (txm *starktxm) logFeederError(ctx context.Context, hash string, f *felt.Felt) {
-	feederClient, err := txm.feederClient.Get()
-	if err != nil {
-		txm.lggr.Errorw("failed to load feeder client", "error", err)
-		return
-	}
-
-	rejectedTx, err := feederClient.TransactionFailure(ctx, f)
-	if err != nil {
-		txm.lggr.Errorw("failed to fetch reason for transaction failure", "hash", hash, "error", err)
-		return
-	}
-
-	txm.lggr.Errorw("feeder rejected reason", "hash", hash, "errorMessage", rejectedTx.ErrorMessage)
-}
-
 func (txm *starktxm) resyncNonce(ctx context.Context, client *starknet.Client, accountAddress *felt.Felt) error {
 	/*
 	   the follow errors indicate that there could be a problem with our locally tracked nonce value:

relayer/pkg/chainlink/txm/txm_test.go

@@ -9,6 +9,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/NethermindEth/juno/core/felt"
 	"github.com/NethermindEth/starknet.go/curve"
 	"github.com/NethermindEth/starknet.go/devnet"
 	starknetrpc "github.com/NethermindEth/starknet.go/rpc"
@@ -27,7 +28,7 @@ import (
 
 func TestIntegration_Txm(t *testing.T) {
 	ctx := t.Context()
-	n := 2 // number of txs per key
+	var nTransactions uint64 = 2 // Number of txs per key. If you increase that you might have to increase the confirmation timeout
 	// url := SetupLocalStarknetNode(t)
 	url := "http://127.0.0.1:5050"
 	devnet := devnet.NewDevNet(url)
@@ -59,7 +60,7 @@ func TestIntegration_Txm(t *testing.T) {
 	})
 	ksAdapter := NewKeystoreAdapter(looppKs)
 
-	lggr, observer := logger.TestObserved(t, zapcore.DebugLevel)
+	lggr, _ := logger.TestObserved(t, zapcore.DebugLevel)
 	timeout := 10 * time.Second
 	client, err := starknet.NewClient("SN_SEPOLIA", url+"/rpc", "", lggr, &timeout)
 	require.NoError(t, err)
@@ -86,47 +87,47 @@ func TestIntegration_Txm(t *testing.T) {
 	// start txm + checks
 	require.NoError(t, txm.Start(context.Background()))
 	require.NoError(t, txm.Ready())
-	fmt.Println("sss")
 
+	accountAddresses := make(map[*felt.Felt]*felt.Felt) // address -> latestNonce
 	for publicKeyStr := range localKeys {
 		publicKey, err := starknetutils.HexToFelt(publicKeyStr)
 		require.NoError(t, err)
 
 		accountAddress, err := starknetutils.HexToFelt(localKeys[publicKeyStr].Account)
 		require.NoError(t, err)
 
+		c, err := getClient()
+		require.NoError(t, err)
+		latestNonce, err := c.AccountNonceLatest(ctx, accountAddress)
+		require.NoError(t, err)
+		accountAddresses[accountAddress] = latestNonce
+
 		contractAddress, err := starknetutils.HexToFelt("0x49D36570D4E46F48E99674BD3FCC84644DDD6B96F7C741B1562B82F9E004DC7")
 		require.NoError(t, err)
 
 		selector := starknetutils.GetSelectorFromNameFelt("totalSupply")
 
-		for i := 0; i < n; i++ {
+		for range nTransactions {
 			require.NoError(t, txm.Enqueue(ctx, accountAddress, publicKey, starknetrpc.FunctionCall{
 				ContractAddress:    contractAddress, // send to ETH token contract
 				EntryPointSelector: selector,
 			}))
 		}
 	}
-	var empty bool
-	for i := 0; i < 30; i++ {
-		time.Sleep(500 * time.Millisecond)
-		queued, unconfirmed := txm.InflightCount()
-		accepted := len(observer.FilterMessageSnippet("ACCEPTED_ON_L2").All())
-		t.Logf("inflight count: queued (%d), unconfirmed (%d), accepted (%d)", queued, unconfirmed, accepted)
-
-		// check queue + tx store counts are 0, accepted txs == total txs broadcast
-		if queued == 0 && unconfirmed == 0 && n*len(localKeys) == accepted {
-			empty = true
-			break
-		}
-	}
 
-	// stop txm
-	assert.True(t, empty, "txm timed out while trying to confirm transactions")
+	assert.Eventually(t, func() bool {
+		queued, unconfirmed := txm.InflightCount()
+		return queued == 0 && unconfirmed == 0
+	}, 15*time.Second, 500*time.Millisecond)
 	require.NoError(t, txm.Close())
-	require.Error(t, txm.Ready())
-	assert.Equal(t, 0, observer.FilterLevelExact(zapcore.ErrorLevel).Len())                       // assert no error logs
-	assert.Equal(t, n*len(localKeys), len(observer.FilterMessageSnippet("ACCEPTED_ON_L2").All())) // validate txs were successfully included on chain
+	// Ensure all transactions are confirmed via nonce
+	for accountAddress, initialNonce := range accountAddresses {
+		c, err := getClient()
+		require.NoError(t, err)
+		latestNonce, err := c.AccountNonceLatest(ctx, accountAddress)
+		require.NoError(t, err)
+		require.Equal(t, int(0), new(felt.Felt).Add(initialNonce, new(felt.Felt).SetUint64(nTransactions)).Cmp(latestNonce))
+	}
 }
 
 // LooppKeystore implements [loop.Keystore] interface and the requirements

relayer/pkg/chainlink/txm/txstore.go

@@ -8,6 +8,8 @@ import (
 	"github.com/NethermindEth/juno/core/felt"
 	starknetrpc "github.com/NethermindEth/starknet.go/rpc"
 	"golang.org/x/exp/maps"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 )
 
 type UnconfirmedTx struct {
@@ -20,15 +22,17 @@ type UnconfirmedTx struct {
 // TxStore tracks broadcast & unconfirmed txs per account address per chain id
 type TxStore struct {
 	lock sync.RWMutex
+	lggr logger.Logger
 
 	nextNonce         *felt.Felt
 	unconfirmedNonces map[string]*UnconfirmedTx
 }
 
-func NewTxStore(initialNonce *felt.Felt) *TxStore {
+func NewTxStore(initialNonce *felt.Felt, lggr logger.Logger) *TxStore {
 	return &TxStore{
 		nextNonce:         new(felt.Felt).Set(initialNonce),
 		unconfirmedNonces: map[string]*UnconfirmedTx{},
+		lggr:              lggr,
 	}
 }
 
@@ -75,7 +79,7 @@ func (s *TxStore) AddUnconfirmed(nonce *felt.Felt, hash string, call starknetrpc
 
 	nonceStr := nonce.String()
 	if h, exists := s.unconfirmedNonces[nonceStr]; exists {
-		return fmt.Errorf("nonce used: tried to use nonce (%s) for tx (%s), already used by (%s)", nonce, h.Hash, h)
+		s.lggr.Warnf("nonce used: replacing tx (hash: %s) with nonce (%s) for tx with hash (%s)", h.Hash, nonce, hash)
 	}
 
 	s.unconfirmedNonces[nonceStr] = &UnconfirmedTx{
@@ -89,21 +93,24 @@ func (s *TxStore) AddUnconfirmed(nonce *felt.Felt, hash string, call starknetrpc
 	return nil
 }
 
-func (s *TxStore) Confirm(nonce *felt.Felt, hash string) error {
+func (s *TxStore) Confirm(latestNonce *felt.Felt) (int, *felt.Felt) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
-	nonceStr := nonce.String()
-	unconfirmed, exists := s.unconfirmedNonces[nonceStr]
-	if !exists {
-		return fmt.Errorf("no such unconfirmed nonce: %s", nonce)
-	}
-	// sanity check that the hash matches
-	if unconfirmed.Hash != hash {
-		return fmt.Errorf("unexpected tx hash: expected %s, got %s", unconfirmed.Hash, hash)
+	// confirm all transactions with a nonce lower than the latest nonce
+	confirmed := 0
+	highestUnconfirmed := new(felt.Felt).SetUint64(0)
+	for nonceStr, tx := range s.unconfirmedNonces {
+		if tx.Nonce.Cmp(latestNonce) < 0 {
+			confirmed++
+			delete(s.unconfirmedNonces, nonceStr)
+		}
+		if highestUnconfirmed.Cmp(tx.Nonce) < 0 {
+			highestUnconfirmed = tx.Nonce
+		}
 	}
-	delete(s.unconfirmedNonces, nonceStr)
-	return nil
+
+	return confirmed, highestUnconfirmed
 }
 
 func (s *TxStore) GetUnconfirmed() []*UnconfirmedTx {
@@ -137,15 +144,22 @@ func NewAccountStore() *AccountStore {
 	}
 }
 
-func (c *AccountStore) CreateTxStore(accountAddress *felt.Felt, initialNonce *felt.Felt) (*TxStore, error) {
+func (c *AccountStore) Accounts() []string {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	return maps.Keys(c.store)
+}
+
+func (c *AccountStore) CreateTxStore(accountAddress *felt.Felt, initialNonce *felt.Felt, lggr logger.Logger) (*TxStore, error) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 	addressStr := accountAddress.String()
 	_, ok := c.store[addressStr]
 	if ok {
 		return nil, fmt.Errorf("TxStore already exists: %s", accountAddress)
 	}
-	store := NewTxStore(initialNonce)
+	store := NewTxStore(initialNonce, logger.Named(lggr, "TxStore"))
 	c.store[addressStr] = store
 	return store, nil
 }