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,13 @@ const (
 	configKeyBlockchainRpcTLSTimeout   = "blockchain-rpc.tls-timeout"
 	configKeyBlockchainRpcIdleTimeout  = "blockchain-rpc.idle-timeout"
 	configKeyBlockchainRpcKeepalive    = "blockchain-rpc.keepalive"
+
+	// transaction retry
+	optionNameTransactionRetryDelay       = "transaction-retry-delay"
+	optionNameTransactionFeePriority      = "transaction-fee-priority"
+	optionNameTransactionFeePriorityMax   = "transaction-fee-priority-max"
+	optionNameTransactionFeeMaxTxPriceWei = "transaction-fee-max-tx-price-wei"
+	optionNameFeeHistoryBlockCount        = "fee-history-block-count"
 )
 
 var blockchainRpcConfigPairs = []struct{ flat, dotted string }{
@@ -314,6 +323,7 @@ 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().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 +343,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().String(optionNameTransactionFeePriority, "market", "starting fee priority for transaction broadcasts (low, market, aggressive)")
+	cmd.Flags().String(optionNameTransactionFeePriorityMax, "aggressive", "maximum fee priority for transaction escalation (low, market, aggressive)")
+	cmd.Flags().Uint64(optionNameTransactionFeeMaxTxPriceWei, 0, "maximum maxFeePerGas in wei per gas for transactions with retry; 0 means no limit")
 	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 +394,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(optionNameTransactionFeePriority),
+		EndTier:    c.config.GetString(optionNameTransactionFeePriorityMax),
+	}
+	if v := c.config.GetUint64(optionNameTransactionFeeMaxTxPriceWei); 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

@@ -60,6 +60,8 @@ func (c *command) initDeployCmd() error {
 					Keepalive:   c.config.GetDuration(configKeyBlockchainRpcKeepalive),
 				},
 				c.config.GetUint64(optionNameBlockSyncInterval),
+				c.config.GetUint64(optionNameFeeHistoryBlockCount),
+				txRetryConfigFromCommand(c),
 			)
 			if err != nil {
 				return err

cmd/bee/cmd/start.go

@@ -300,6 +300,8 @@ 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),
+		TransactionRetry:              txRetryConfigFromCommand(c),
 		BootnodeMode:                  bootNode,
 		Bootnodes:                     networkConfig.bootNodes,
 		CacheCapacity:                 c.config.GetUint64(optionNameCacheCapacity),

openapi/SwarmCommon.yaml

@@ -1126,7 +1126,9 @@ components:
       schema:
         $ref: "SwarmCommon.yaml#/components/schemas/GasPrice"
       required: false
-      description: "Gas price for transaction"
+      description: >
+        Maximum gas price for the transaction. When set, the node
+        uses the legacy send path, otherwise transaction is sent with retry and automatically escalated fees.
 
     GasLimitParameter:
       in: header

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"
+	FeePriorityHeader = "Fee-Priority"
+	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"`
+				FeePriority string   `map:"Fee-Priority"`
 			}{}
 			if response := s.mapStructure(r.Header, &headers); response != nil {
 				response("invalid header params", logger, w)
@@ -567,6 +569,22 @@ 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)
+			if headers.FeePriority != "" {
+				tier, err := transaction.ParseFeeTier(headers.FeePriority)
+				if err != nil {
+					logger.Debug("invalid fee priority header", "error", err)
+					jsonhttp.BadRequest(w, jsonhttp.StatusResponse{
+						Message: "invalid header params",
+						Code:    http.StatusBadRequest,
+						Reasons: []jsonhttp.Reason{{
+							Field: FeePriorityHeader,
+							Error: err.Error(),
+						}},
+					})
+					return
+				}
+				ctx = sctx.SetFeePriority(ctx, tier.String())
+			}
 
 			h.ServeHTTP(w, r.WithContext(ctx))
 		})
@@ -581,7 +599,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, FeePriorityHeader, ImmutableHeader,
 		SwarmActHeader, SwarmActTimestampHeader, SwarmActPublisherHeader, SwarmActHistoryAddressHeader,
 	}
 	allowedHeadersStr := strings.Join(allowedHeaders, ", ")

pkg/node/chain.go

@@ -64,6 +64,8 @@ func InitChain(
 	fallbackGasLimit uint64,
 	rpcCfg BlockchainRPCConfig,
 	blockSyncInterval uint64,
+	feeHistoryBlockCount uint64,
+	retryCfg transaction.TransactionsRetryConfig,
 ) (transaction.Backend, common.Address, int64, transaction.Monitor, transaction.Service, error) {
 	backend := backendnoop.New(chainID)
 
@@ -98,7 +100,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)
 	}
 
 	backendChainID, err := backend.ChainID(ctx)
@@ -117,7 +119,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,8 @@ type Options struct {
 	BlockProfile                  bool
 	BlockTime                     time.Duration
 	BlockSyncInterval             uint64
+	FeeHistoryBlockCount          uint64
+	TransactionRetry              transaction.TransactionsRetryConfig
 	BootnodeMode                  bool
 	Bootnodes                     []string
 	CacheCapacity                 uint64
@@ -438,6 +440,8 @@ func NewBee(
 			Keepalive:   o.BlockchainRpcKeepalive,
 		},
 		o.BlockSyncInterval,
+		o.FeeHistoryBlockCount,
+		o.TransactionRetry,
 	)
 	if err != nil {
 		return nil, fmt.Errorf("init chain: %w", err)
@@ -1409,6 +1413,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,21 +180,7 @@ func (c *postageContract) sendApproveTransaction(ctx context.Context, amount *bi
 		)
 	}()
 
-	txHash, err := c.transactionService.Send(ctx, request, transaction.DefaultTipBoostPercent)
-	if err != nil {
-		return nil, err
-	}
-
-	receipt, err = c.transactionService.WaitForReceipt(ctx, txHash)
-	if err != nil {
-		return nil, err
-	}
-
-	if receipt.Status == 0 {
-		return nil, transaction.ErrTransactionReverted
-	}
-
-	return receipt, nil
+	return c.sendRequest(ctx, request)
 }
 
 func (c *postageContract) sendTransaction(ctx context.Context, callData []byte, desc string) (receipt *types.Receipt, err error) {
@@ -216,20 +202,32 @@ func (c *postageContract) sendTransaction(ctx context.Context, callData []byte,
 		)
 	}()
 
-	txHash, err := c.transactionService.Send(ctx, request, transaction.DefaultTipBoostPercent)
-	if err != nil {
-		return nil, err
+	return c.sendRequest(ctx, request)
+}
+
+// sendRequest sends a postage transaction. When Gas-Price is set in the request
+// context the legacy Send path is used (client spend ceiling); otherwise
+// SendWithRetry applies automatic fee escalation.
+func (c *postageContract) sendRequest(ctx context.Context, request *transaction.TxRequest) (*types.Receipt, error) {
+	if sctx.GetGasPrice(ctx) != nil {
+		txHash, err := c.transactionService.Send(ctx, request, transaction.DefaultTipBoostPercent)
+		if err != nil {
+			return nil, err
+		}
+		receipt, err := c.transactionService.WaitForReceipt(ctx, txHash)
+		if err != nil {
+			return nil, err
+		}
+		if receipt.Status == 0 {
+			return nil, transaction.ErrTransactionReverted
+		}
+		return receipt, nil
 	}
 
-	receipt, err = c.transactionService.WaitForReceipt(ctx, txHash)
+	_, receipt, err := c.transactionService.SendWithRetry(ctx, request)
 	if err != nil {
 		return nil, err
 	}
-
-	if receipt.Status == 0 {
-		return nil, transaction.ErrTransactionReverted
-	}
-
 	return receipt, nil
 }