add rate limiting when fetching headers, store failed event decoding information, various smaller fixes

Author: Tofel

seth/client.go

@@ -532,8 +532,6 @@ func (m *Client) TransferETHFromKey(ctx context.Context, fromKeyNum int, to stri
 			fromKeyNum, m.Addresses[fromKeyNum].Hex(), err)
 	}
 
-	ctx, sendCancel := context.WithTimeout(ctx, m.Cfg.Network.TxnTimeout.Duration())
-	defer sendCancel()
 	err = m.Client.SendTransaction(ctx, signedTx)
 	if err != nil {
 		return fmt.Errorf("failed to send transaction to network: %w\n"+
@@ -1345,12 +1343,13 @@ func (t TransactionLog) GetData() []byte {
 	return t.Data
 }
 
-func (m *Client) decodeContractLogs(l zerolog.Logger, logs []types.Log, allABIs []*abi.ABI) ([]DecodedTransactionLog, error) {
+func (m *Client) decodeContractLogs(l zerolog.Logger, logs []types.Log, allABIs []*abi.ABI) ([]DecodedTransactionLog, []EventDecodingError, error) {
 	l.Trace().
 		Msg("Decoding events")
 	sigMap := buildEventSignatureMap(allABIs)
 
 	var eventsParsed []DecodedTransactionLog
+	var decodeErrors []EventDecodingError
 	for _, lo := range logs {
 		if len(lo.Topics) == 0 {
 			l.Debug().
@@ -1384,6 +1383,7 @@ func (m *Client) decodeContractLogs(l zerolog.Logger, logs []types.Log, allABIs
 
 		// Iterate over possible events with the same signature
 		matched := false
+		var decodeAttempts []ABIDecodingError
 		for _, evWithABI := range possibleEvents {
 			evSpec := evWithABI.EventSpec
 			contractABI := evWithABI.ContractABI
@@ -1421,19 +1421,28 @@ func (m *Client) decodeContractLogs(l zerolog.Logger, logs []types.Log, allABIs
 			}
 
 			// Proceed to decode the event
+			// Find ABI name for this contract ABI
+			abiName := m.findABIName(contractABI)
 			d := TransactionLog{lo.Topics, lo.Data}
 			l.Trace().
 				Str("Name", evSpec.RawName).
 				Str("Signature", evSpec.Sig).
+				Str("ABI", abiName).
 				Msg("Unpacking event")
 
 			eventsMap, topicsMap, err := decodeEventFromLog(l, *contractABI, *evSpec, d)
 			if err != nil {
-				l.Error().
+				l.Debug().
 					Err(err).
 					Str("Event", evSpec.Name).
-					Msg("Failed to decode event; skipping")
-				continue // Skip this event instead of returning an error
+					Str("ABI", abiName).
+					Msg("Failed to decode event; trying next ABI")
+				decodeAttempts = append(decodeAttempts, ABIDecodingError{
+					ABIName:   abiName,
+					EventName: evSpec.Name,
+					Error:     err.Error(),
+				})
+				continue // Try next ABI instead of giving up
 			}
 
 			parsedEvent := decodedLogFromMaps(&DecodedTransactionLog{}, eventsMap, topicsMap)
@@ -1455,12 +1464,36 @@ func (m *Client) decodeContractLogs(l zerolog.Logger, logs []types.Log, allABIs
 		}
 
 		if !matched {
+			// Record the decode failure
+			topics := make([]string, len(lo.Topics))
+			for i, topic := range lo.Topics {
+				topics[i] = topic.Hex()
+			}
+			
+			decodeError := EventDecodingError{
+				Signature: eventSig,
+				LogIndex:  lo.Index,
+				Address:   lo.Address.Hex(),
+				Topics:    topics,
+				Errors:    decodeAttempts,
+			}
+			decodeErrors = append(decodeErrors, decodeError)
+			
+			abiNames := make([]string, len(decodeAttempts))
+			for i, attempt := range decodeAttempts {
+				abiNames[i] = attempt.ABIName
+			}
+			
 			l.Warn().
 				Str("Signature", eventSig).
-				Msg("No matching event with valid indexed parameter count found for log")
+				Uint("LogIndex", lo.Index).
+				Str("Address", lo.Address.Hex()).
+				Strs("AttemptedABIs", abiNames).
+				Int("FailedAttempts", len(decodeAttempts)).
+				Msg("Failed to decode event log")
 		}
 	}
-	return eventsParsed, nil
+	return eventsParsed, decodeErrors, nil
 }
 
 type eventWithABI struct {
@@ -1484,6 +1517,21 @@ func buildEventSignatureMap(allABIs []*abi.ABI) map[string][]*eventWithABI {
 	return sigMap
 }
 
+// findABIName finds the name of the ABI in the ContractStore, returns "unknown" if not found
+func (m *Client) findABIName(targetABI *abi.ABI) string {
+	if m.ContractStore == nil {
+		return "unknown"
+	}
+	
+	for name, storedABI := range m.ContractStore.ABIs {
+		if reflect.DeepEqual(storedABI, *targetABI) {
+			return strings.TrimSuffix(name, ".abi")
+		}
+	}
+	
+	return "unknown"
+}
+
 // WaitUntilNoPendingTxForRootKey waits until there's no pending transaction for root key. If after timeout there are still pending transactions, it returns error.
 func (m *Client) WaitUntilNoPendingTxForRootKey(timeout time.Duration) error {
 	return m.WaitUntilNoPendingTx(m.MustGetRootKeyAddress(), timeout)

seth/client_builder.go

@@ -32,7 +32,7 @@ func NewClientBuilder() *ClientBuilder {
 		DialTimeout:                    MustMakeDuration(DefaultDialTimeout),
 		TransferGasFee:                 DefaultTransferGasFee,
 		GasPriceEstimationEnabled:      true,
-		GasPriceEstimationBlocks:       200,
+		GasPriceEstimationBlocks:       20,
 		GasPriceEstimationTxPriority:   Priority_Standard,
 		GasPrice:                       DefaultGasPrice,
 		GasFeeCap:                      DefaultGasFeeCap,

seth/decode.go

@@ -36,12 +36,29 @@ const (
 // DecodedTransaction decoded transaction
 type DecodedTransaction struct {
 	CommonData
-	Index       uint                    `json:"index"`
-	Hash        string                  `json:"hash,omitempty"`
-	Protected   bool                    `json:"protected,omitempty"`
-	Transaction *types.Transaction      `json:"transaction,omitempty"`
-	Receipt     *types.Receipt          `json:"receipt,omitempty"`
-	Events      []DecodedTransactionLog `json:"events,omitempty"`
+	Index              uint                    `json:"index"`
+	Hash               string                  `json:"hash,omitempty"`
+	Protected          bool                    `json:"protected,omitempty"`
+	Transaction        *types.Transaction      `json:"transaction,omitempty"`
+	Receipt            *types.Receipt          `json:"receipt,omitempty"`
+	Events             []DecodedTransactionLog `json:"events,omitempty"`
+	EventDecodingErrors []EventDecodingError    `json:"event_decoding_errors,omitempty"`
+}
+
+// EventDecodingError represents a failed event decode attempt
+type EventDecodingError struct {
+	Signature    string                 `json:"signature"`
+	LogIndex     uint                   `json:"log_index"`
+	Address      string                 `json:"address"`
+	Topics       []string               `json:"topics,omitempty"`
+	Errors       []ABIDecodingError     `json:"errors,omitempty"`
+}
+
+// ABIDecodingError represents a single ABI decode attempt failure
+type ABIDecodingError struct {
+	ABIName    string `json:"abi_name"`
+	EventName  string `json:"event_name"`
+	Error      string `json:"error"`
 }
 
 type CommonData struct {
@@ -448,6 +465,7 @@ func (m *Client) decodeTransaction(l zerolog.Logger, tx *types.Transaction, rece
 	}
 
 	var txIndex uint
+	var decodeErrors []EventDecodingError
 
 	if receipt != nil {
 		l.Trace().Interface("Receipt", receipt).Msg("TX receipt")
@@ -463,7 +481,8 @@ func (m *Client) decodeTransaction(l zerolog.Logger, tx *types.Transaction, rece
 			allABIs = m.ContractStore.GetAllABIs()
 		}
 
-		txEvents, err = m.decodeContractLogs(l, logsValues, allABIs)
+		var err error
+		txEvents, decodeErrors, err = m.decodeContractLogs(l, logsValues, allABIs)
 		if err != nil {
 			return defaultTxn, err
 		}
@@ -475,12 +494,13 @@ func (m *Client) decodeTransaction(l zerolog.Logger, tx *types.Transaction, rece
 			Method:    abiResult.Method.Sig,
 			Input:     txInput,
 		},
-		Index:       txIndex,
-		Receipt:     receipt,
-		Transaction: tx,
-		Protected:   tx.Protected(),
-		Hash:        tx.Hash().String(),
-		Events:      txEvents,
+		Index:               txIndex,
+		Receipt:             receipt,
+		Transaction:         tx,
+		Protected:           tx.Protected(),
+		Hash:                tx.Hash().String(),
+		Events:              txEvents,
+		EventDecodingErrors: decodeErrors,
 	}
 
 	return ptx, nil
@@ -499,7 +519,34 @@ func (m *Client) printDecodedTXData(l zerolog.Logger, ptx *DecodedTransaction) {
 	for _, e := range ptx.Events {
 		l.Debug().
 			Str("Signature", e.Signature).
-			Interface("Log", e.EventData).Send()
+			Str("Address", e.Address.Hex()).
+			Interface("Topics", e.Topics).
+			Interface("Data", e.EventData).
+			Msg("Event emitted")
+	}
+	
+	// Print event decoding errors separately
+	if len(ptx.EventDecodingErrors) > 0 {
+		l.Warn().
+			Int("Failed event decodes", len(ptx.EventDecodingErrors)).
+			Msg("Some events could not be decoded")
+		
+		for _, decodeErr := range ptx.EventDecodingErrors {
+			abiNames := make([]string, len(decodeErr.Errors))
+			errorMsgs := make([]string, len(decodeErr.Errors))
+			for i, abiErr := range decodeErr.Errors {
+				abiNames[i] = abiErr.ABIName
+				errorMsgs[i] = fmt.Sprintf("%s.%s: %s", abiErr.ABIName, abiErr.EventName, abiErr.Error)
+			}
+			
+			l.Warn().
+				Str("Signature", decodeErr.Signature).
+				Uint("LogIndex", decodeErr.LogIndex).
+				Str("Address", decodeErr.Address).
+				Strs("AttemptedABIs", abiNames).
+				Strs("Errors", errorMsgs).
+				Msg("Failed to decode event log")
+		}
 	}
 }
 

seth/gas_adjuster.go

@@ -2,6 +2,7 @@ package seth
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math"
 	"math/big"
@@ -12,6 +13,7 @@ import (
 
 	"github.com/avast/retry-go"
 	"github.com/ethereum/go-ethereum/core/types"
+	"go.uber.org/ratelimit"
 )
 
 const (
@@ -56,7 +58,7 @@ func (m *Client) CalculateNetworkCongestionMetric(blocksNumber uint64, strategy
 			return cachedHeader, nil
 		}
 
-		timeout := blocksNumber / 100
+		timeout := blocksNumber / 10
 		if timeout < 3 {
 			timeout = 3
 		} else if timeout > 6 {
@@ -94,7 +96,9 @@ func (m *Client) CalculateNetworkCongestionMetric(blocksNumber uint64, strategy
 	var wg sync.WaitGroup
 	dataCh := make(chan *types.Header)
 
+	limit := ratelimit.New(4) // 4 concurrent requests
 	go func() {
+		limit.Take()
 		for header := range dataCh {
 			headers = append(headers, header)
 			// placed here, because we want to wait for all headers to be received and added to slice before continuing
@@ -262,6 +266,7 @@ func (m *Client) GetSuggestedEIP1559Fees(ctx context.Context, priority string) (
 			Float64("BaseFee", baseFee64).
 			Int64("SuggestedTip", currentGasTip.Int64()).
 			Msgf("Incorrect gas data received from node: base fee was 0. Skipping gas estimation")
+		err = errors.New("incorrect gas data received from node: base fee was 0. Skipping gas estimation")
 		return
 	}
 
@@ -734,16 +739,18 @@ func calculateGasUsedRatio(headers []*types.Header) float64 {
 		return 0
 	}
 
-	var totalRatio float64
+	var totalGasUsedRatio float64
+	validHeaders := 0
 	for _, header := range headers {
-		if header.GasLimit == 0 {
-			continue
+		if header.GasLimit > 0 {
+			totalGasUsedRatio += float64(header.GasUsed) / float64(header.GasLimit)
+			validHeaders++
 		}
-		ratio := float64(header.GasUsed) / float64(header.GasLimit)
-		totalRatio += ratio
 	}
-	averageRatio := totalRatio / float64(len(headers))
-	return averageRatio
+	if validHeaders == 0 {
+		return 0.0
+	}
+	return totalGasUsedRatio / float64(validHeaders)
 }
 
 func calculateMagnitudeDifference(first, second *big.Float) (int, string) {

seth/nonce.go

@@ -107,7 +107,7 @@ func (m *NonceManager) UpdateNonces() error {
 	for addr := range m.Nonces {
 		nonce, err := m.Client.Client.NonceAt(context.Background(), addr, nil)
 		if err != nil {
-			return err
+			return fmt.Errorf("failed to updated nonces for address '%s': %w", addr, err)
 		}
 		m.Nonces[addr] = mustSafeInt64(nonce)
 	}