feat: transactions with retry

cmd/bee/cmd/cmd.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math/big"
 	"os"
 	"path/filepath"
 	"strings"
@@ -17,6 +18,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/log"
 	"github.com/ethersphere/bee/v2/pkg/node"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
+	"github.com/ethersphere/bee/v2/pkg/transaction"
 	p2pforge "github.com/ipshipyard/p2p-forge/client"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
@@ -103,6 +105,14 @@ const (
 	configKeyBlockchainRpcTLSTimeout   = "blockchain-rpc.tls-timeout"
 	configKeyBlockchainRpcIdleTimeout  = "blockchain-rpc.idle-timeout"
 	configKeyBlockchainRpcKeepalive    = "blockchain-rpc.keepalive"
+
+	// transaction retry
+	optionNameTransactionRetryDelay         = "transaction-retry-delay"
+	optionNameTransactionRetryMaxTxPriceWei = "transaction-retry-max-tx-price-wei"
+	optionNameTransactionRetryStartTier     = "transaction-retry-start-tier"
+	optionNameTransactionRetryEndTier       = "transaction-retry-end-tier"
+	optionNameFeeHistoryBlockCount          = "fee-history-block-count"
+	optionNameFeeHistoryRewardPercentiles   = "fee-history-reward-percentiles"
 )
 
 var blockchainRpcConfigPairs = []struct{ flat, dotted string }{
@@ -314,6 +324,8 @@ func (c *command) setAllFlags(cmd *cobra.Command) {
 	cmd.Flags().String(optionNameStakingAddress, "", "staking contract address")
 	cmd.Flags().Uint64(optionNameBlockTime, 5, "chain block time")
 	cmd.Flags().Uint64(optionNameBlockSyncInterval, 10, "block number cache sync interval in blocks")
+	cmd.Flags().Uint64(optionNameFeeHistoryBlockCount, 100, "eth_feeHistory block count for fee hints")
+	cmd.Flags().String(optionNameFeeHistoryRewardPercentiles, "10,50,90", "comma-separated reward percentiles for eth_feeHistory")
 	cmd.Flags().Duration(optionWarmUpTime, time.Minute*5, "maximum node warmup duration; proceeds when stable or after this time")
 	cmd.Flags().Bool(optionNameMainNet, true, "triggers connect to main net bootnodes.")
 	cmd.Flags().Bool(optionNameRetrievalCaching, true, "enable forwarded content caching")
@@ -333,6 +345,10 @@ func (c *command) setAllFlags(cmd *cobra.Command) {
 	cmd.Flags().Bool(optionSkipPostageSnapshot, false, "skip postage snapshot")
 	cmd.Flags().Uint64(optionNameMinimumGasTipCap, 0, "minimum gas tip cap in wei for transactions, 0 means use suggested gas tip cap")
 	cmd.Flags().Uint64(optionNameGasLimitFallback, 500_000, "gas limit fallback when estimation fails for contract transactions")
+	cmd.Flags().Duration(optionNameTransactionRetryDelay, time.Minute, "how long to wait for a receipt before escalating fees in transactions with retry")
+	cmd.Flags().Uint64(optionNameTransactionRetryMaxTxPriceWei, 0, "maximum maxFeePerGas in wei per gas for transactions with retry")
+	cmd.Flags().String(optionNameTransactionRetryStartTier, "market", "starting fee tier for transaction retry escalation (low, market, aggressive)")
+	cmd.Flags().String(optionNameTransactionRetryEndTier, "aggressive", "ending fee tier for transaction retry escalation (low, market, aggressive)")
 	cmd.Flags().Bool(optionNameP2PWSSEnable, false, "Enable Secure WebSocket P2P connections")
 	cmd.Flags().String(optionP2PWSSAddr, ":1635", "p2p wss address")
 	cmd.Flags().String(optionNATWSSAddr, "", "WSS NAT exposed address")
@@ -380,6 +396,18 @@ func (c *command) bindBlockchainRpcConfig(cmd *cobra.Command) {
 	}
 }
 
+func txRetryConfigFromCommand(c *command) transaction.TransactionsRetryConfig {
+	cfg := transaction.TransactionsRetryConfig{
+		RetryDelay: c.config.GetDuration(optionNameTransactionRetryDelay),
+		StartTier:  c.config.GetString(optionNameTransactionRetryStartTier),
+		EndTier:    c.config.GetString(optionNameTransactionRetryEndTier),
+	}
+	if v := c.config.GetUint64(optionNameTransactionRetryMaxTxPriceWei); v != 0 {
+		cfg.MaxTxPrice = new(big.Int).SetUint64(v)
+	}
+	return cfg
+}
+
 func newLogger(cmd *cobra.Command, verbosity string) (log.Logger, error) {
 	var (
 		sink   = cmd.OutOrStdout()

cmd/bee/cmd/deploy.go

@@ -42,6 +42,11 @@ func (c *command) initDeployCmd() error {
 
 			ctx := cmd.Context()
 
+			feeHistoryRewardPerc, err := node.ParseFeeHistoryRewardPercentiles(c.config.GetString(optionNameFeeHistoryRewardPercentiles))
+			if err != nil {
+				return err
+			}
+
 			swapBackend, overlayEthAddress, chainID, transactionMonitor, transactionService, err := node.InitChain(
 				ctx,
 				logger,
@@ -60,6 +65,9 @@ func (c *command) initDeployCmd() error {
 					Keepalive:   c.config.GetDuration(configKeyBlockchainRpcKeepalive),
 				},
 				c.config.GetUint64(optionNameBlockSyncInterval),
+				c.config.GetUint64(optionNameFeeHistoryBlockCount),
+				feeHistoryRewardPerc,
+				txRetryConfigFromCommand(c),
 			)
 			if err != nil {
 				return err

cmd/bee/cmd/start.go

@@ -202,6 +202,11 @@ func buildBeeNode(ctx context.Context, c *command, cmd *cobra.Command, logger lo
 		}
 	}
 
+	feeHistoryRewardPerc, err := node.ParseFeeHistoryRewardPercentiles(c.config.GetString(optionNameFeeHistoryRewardPercentiles))
+	if err != nil {
+		return nil, err
+	}
+
 	signerConfig, err := c.configureSigner(cmd, logger)
 	if err != nil {
 		return nil, fmt.Errorf("configure signer: %w", err)
@@ -300,6 +305,12 @@ func buildBeeNode(ctx context.Context, c *command, cmd *cobra.Command, logger lo
 		BlockProfile:                  c.config.GetBool(optionNamePProfBlock),
 		BlockTime:                     networkConfig.blockTime,
 		BlockSyncInterval:             c.config.GetUint64(optionNameBlockSyncInterval),
+		FeeHistoryBlockCount:          c.config.GetUint64(optionNameFeeHistoryBlockCount),
+		FeeHistoryRewardPercentiles:   feeHistoryRewardPerc,
+		TransactionRetryDelay:         c.config.GetDuration(optionNameTransactionRetryDelay),
+		TransactionRetryMaxTxPriceWei: c.config.GetUint64(optionNameTransactionRetryMaxTxPriceWei),
+		TransactionRetryStartTier:     c.config.GetString(optionNameTransactionRetryStartTier),
+		TransactionRetryEndTier:       c.config.GetString(optionNameTransactionRetryEndTier),
 		BootnodeMode:                  bootNode,
 		Bootnodes:                     networkConfig.bootNodes,
 		CacheCapacity:                 c.config.GetUint64(optionNameCacheCapacity),

pkg/api/api.go

@@ -94,10 +94,11 @@ const (
 	SwarmActPublisherHeader           = "Swarm-Act-Publisher"
 	SwarmActHistoryAddressHeader      = "Swarm-Act-History-Address"
 
-	ImmutableHeader = "Immutable"
-	GasPriceHeader  = "Gas-Price"
-	GasLimitHeader  = "Gas-Limit"
-	ETagHeader      = "ETag"
+	ImmutableHeader    = "Immutable"
+	GasPriceHeader     = "Gas-Price"
+	GasLimitHeader     = "Gas-Limit"
+	DisableRetryHeader = "Disable-Retry"
+	ETagHeader         = "ETag"
 
 	AuthorizationHeader        = "Authorization"
 	AcceptEncodingHeader       = "Accept-Encoding"
@@ -557,8 +558,9 @@ func (s *Service) gasConfigMiddleware(handlerName string) func(h http.Handler) h
 			logger := s.logger.WithName(handlerName).Build()
 
 			headers := struct {
-				GasPrice *big.Int `map:"Gas-Price"`
-				GasLimit uint64   `map:"Gas-Limit"`
+				GasPrice     *big.Int `map:"Gas-Price"`
+				GasLimit     uint64   `map:"Gas-Limit"`
+				DisableRetry bool     `map:"Disable-Retry"`
 			}{}
 			if response := s.mapStructure(r.Header, &headers); response != nil {
 				response("invalid header params", logger, w)
@@ -567,6 +569,7 @@ func (s *Service) gasConfigMiddleware(handlerName string) func(h http.Handler) h
 			ctx := r.Context()
 			ctx = sctx.SetGasPrice(ctx, headers.GasPrice)
 			ctx = sctx.SetGasLimit(ctx, headers.GasLimit)
+			ctx = sctx.SetDisableRetry(ctx, headers.DisableRetry)
 
 			h.ServeHTTP(w, r.WithContext(ctx))
 		})
@@ -581,7 +584,7 @@ func (s *Service) corsHandler(h http.Handler) http.Handler {
 		SwarmTagHeader, SwarmPinHeader, SwarmEncryptHeader, SwarmIndexDocumentHeader, SwarmErrorDocumentHeader, SwarmCollectionHeader,
 		SwarmPostageBatchIdHeader, SwarmPostageStampHeader, SwarmDeferredUploadHeader, SwarmRedundancyLevelHeader,
 		SwarmRedundancyStrategyHeader, SwarmRedundancyFallbackModeHeader, SwarmChunkRetrievalTimeoutHeader, SwarmLookAheadBufferSizeHeader,
-		SwarmFeedIndexHeader, SwarmFeedIndexNextHeader, SwarmSocSignatureHeader, SwarmOnlyRootChunk, GasPriceHeader, GasLimitHeader, ImmutableHeader,
+		SwarmFeedIndexHeader, SwarmFeedIndexNextHeader, SwarmSocSignatureHeader, SwarmOnlyRootChunk, GasPriceHeader, GasLimitHeader, DisableRetryHeader, ImmutableHeader,
 		SwarmActHeader, SwarmActTimestampHeader, SwarmActPublisherHeader, SwarmActHistoryAddressHeader,
 	}
 	allowedHeadersStr := strings.Join(allowedHeaders, ", ")

pkg/node/chain.go

@@ -12,6 +12,7 @@ import (
 	"math/big"
 	"net"
 	"net/http"
+	"strconv"
 	"strings"
 	"time"
 
@@ -41,6 +42,35 @@ const (
 	additionalConfirmations = 2
 )
 
+// ParseFeeHistoryRewardPercentiles parses a comma-separated list of floats for eth_feeHistory
+// rewardPercentiles. Exactly three values in the range [0, 100] are required.
+func ParseFeeHistoryRewardPercentiles(s string) ([]float64, error) {
+	s = strings.TrimSpace(s)
+	if s == "" {
+		return nil, errors.New("fee history reward percentiles: empty string")
+	}
+	parts := strings.Split(s, ",")
+	out := make([]float64, 0, len(parts))
+	for _, p := range parts {
+		p = strings.TrimSpace(p)
+		if p == "" {
+			return nil, errors.New("fee history reward percentiles: empty token")
+		}
+		v, err := strconv.ParseFloat(p, 64)
+		if err != nil {
+			return nil, fmt.Errorf("fee history reward percentiles: parse %q: %w", p, err)
+		}
+		if v < 0 || v > 100 {
+			return nil, fmt.Errorf("fee history reward percentiles: %g out of range [0,100]", v)
+		}
+		out = append(out, v)
+	}
+	if len(out) != 3 {
+		return nil, fmt.Errorf("fee history reward percentiles: exactly 3 values, got %d", len(out))
+	}
+	return out, nil
+}
+
 // BlockchainRPCConfig holds the configuration parameters for the blockchain RPC client transport.
 type BlockchainRPCConfig struct {
 	Endpoint    string
@@ -64,6 +94,9 @@ func InitChain(
 	fallbackGasLimit uint64,
 	rpcCfg BlockchainRPCConfig,
 	blockSyncInterval uint64,
+	feeHistoryBlockCount uint64,
+	feeHistoryRewardPercentiles []float64,
+	retryCfg transaction.TransactionsRetryConfig,
 ) (transaction.Backend, common.Address, int64, transaction.Monitor, transaction.Service, error) {
 	backend := backendnoop.New(chainID)
 
@@ -98,7 +131,7 @@ func InitChain(
 
 		logger.Info("connected to blockchain backend", "version", versionString)
 
-		backend = wrapped.NewBackend(ethclient.NewClient(rpcClient), minimumGasTipCap, pollingInterval, blockSyncInterval)
+		backend = wrapped.NewBackend(ethclient.NewClient(rpcClient), minimumGasTipCap, pollingInterval, blockSyncInterval, feeHistoryBlockCount, feeHistoryRewardPercentiles)
 	}
 
 	backendChainID, err := backend.ChainID(ctx)
@@ -117,7 +150,7 @@ func InitChain(
 
 	transactionMonitor := transaction.NewMonitor(logger, backend, overlayEthAddress, pollingInterval, cancellationDepth)
 
-	transactionService, err := transaction.NewService(logger, overlayEthAddress, backend, signer, stateStore, backendChainID, transactionMonitor, fallbackGasLimit)
+	transactionService, err := transaction.NewService(logger, overlayEthAddress, backend, signer, stateStore, backendChainID, transactionMonitor, fallbackGasLimit, retryCfg)
 	if err != nil {
 		return nil, common.Address{}, 0, nil, nil, fmt.Errorf("transaction service: %w", err)
 	}

pkg/node/node.go

@@ -142,6 +142,12 @@ type Options struct {
 	BlockProfile                  bool
 	BlockTime                     time.Duration
 	BlockSyncInterval             uint64
+	FeeHistoryBlockCount          uint64
+	FeeHistoryRewardPercentiles   []float64
+	TransactionRetryDelay         time.Duration
+	TransactionRetryMaxTxPriceWei uint64
+	TransactionRetryStartTier     string
+	TransactionRetryEndTier       string
 	BootnodeMode                  bool
 	Bootnodes                     []string
 	CacheCapacity                 uint64
@@ -197,6 +203,18 @@ type Options struct {
 	WhitelistedWithdrawalAddress  []string
 }
 
+func txRetryConfigFromOptions(o *Options) transaction.TransactionsRetryConfig {
+	c := transaction.TransactionsRetryConfig{
+		RetryDelay: o.TransactionRetryDelay,
+		StartTier:  o.TransactionRetryStartTier,
+		EndTier:    o.TransactionRetryEndTier,
+	}
+	if o.TransactionRetryMaxTxPriceWei != 0 {
+		c.MaxTxPrice = new(big.Int).SetUint64(o.TransactionRetryMaxTxPriceWei)
+	}
+	return c
+}
+
 const (
 	refreshRate                   = int64(4_500_000)          // accounting units refreshed per second
 	lightFactor                   = 10                        // downscale payment thresholds and their change rate, and refresh rates by this for light nodes
@@ -438,6 +456,9 @@ func NewBee(
 			Keepalive:   o.BlockchainRpcKeepalive,
 		},
 		o.BlockSyncInterval,
+		o.FeeHistoryBlockCount,
+		o.FeeHistoryRewardPercentiles,
+		txRetryConfigFromOptions(o),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("init chain: %w", err)
@@ -1409,6 +1430,9 @@ func NewBee(
 		if swapBackendMetrics, ok := chainBackend.(metrics.Collector); ok {
 			apiService.MustRegisterMetrics(swapBackendMetrics.Metrics()...)
 		}
+		if txMetrics, ok := transactionService.(metrics.Collector); ok {
+			apiService.MustRegisterMetrics(txMetrics.Metrics()...)
+		}
 
 		if l, ok := logger.(metrics.Collector); ok {
 			apiService.MustRegisterMetrics(l.Metrics()...)

pkg/postage/postagecontract/contract.go

@@ -180,6 +180,14 @@ func (c *postageContract) sendApproveTransaction(ctx context.Context, amount *bi
 		)
 	}()
 
+	if !sctx.GetDisableRetry(ctx) {
+		_, receipt, err = c.transactionService.SendWithRetry(ctx, request)
+		if err != nil {
+			return nil, err
+		}
+		return receipt, nil
+	}
+
 	txHash, err := c.transactionService.Send(ctx, request, transaction.DefaultTipBoostPercent)
 	if err != nil {
 		return nil, err
@@ -216,6 +224,14 @@ func (c *postageContract) sendTransaction(ctx context.Context, callData []byte,
 		)
 	}()
 
+	if !sctx.GetDisableRetry(ctx) {
+		_, receipt, err = c.transactionService.SendWithRetry(ctx, request)
+		if err != nil {
+			return nil, err
+		}
+		return receipt, nil
+	}
+
 	txHash, err := c.transactionService.Send(ctx, request, transaction.DefaultTipBoostPercent)
 	if err != nil {
 		return nil, err