feat: add configurable retry for transient rpc failures

README.md

@@ -7,6 +7,7 @@ batch-export is a tool to retrieve Ethereum event logs for specific contracts, p
 - Retrieve event logs for a specified contract address and block range.
 - Handles large block ranges by querying in smaller chunks.
 - Supports rate limiting for RPC requests.
+- Retries requests that fail with a transient network error (timeouts, dropped connections, rate limiting) using an exponential backoff.
 - Saves retrieved logs to a specified output file (default: `export.ndjson`) in NDJSON format.
 - Graceful shutdown on interrupt signals (Ctrl+C).
 
@@ -49,6 +50,8 @@ The primary command is export.
   -h, --help                       help for export
   -m, --max-request int            Max RPC requests/sec (default 15)
   -o, --output string              Output file path (NDJSON) (default "export.ndjson")
+      --retry-delay duration       Delay before the first retry, doubling per retry up to 30s (default 1s)
+      --retry-max int              Max retries per RPC request on transient network errors (0 disables retrying) (default 5)
       --start uint                 Start block (optional, uses contract start block if 0) (default 31306381)
   -v, --verbosity string           Log verbosity (silent, error, warn, info, debug) (default "info")
 ```

cmd/export.go

@@ -25,21 +25,35 @@ func (c *command) initExportCmd() (err error) {
 		blockRangeLimit uint32
 		outputFile      string
 		compress        bool
+		retryMax        int
+		retryDelay      time.Duration
 	)
 
 	cmd := &cobra.Command{
 		Use:   "export",
 		Short: "Export Swarm Postage Stamp contract event logs within a block range.",
 		Long: `Exports event logs for the Swarm Postage Stamp contract from a specified Ethereum RPC endpoint
 within a given block range (--start to --end). It handles large ranges by querying in chunks (--block-range-limit)
-and respects RPC rate limits (--max-request).
+and respects RPC rate limits (--max-request). Requests failing with a transient network error are retried
+with an exponential backoff (--retry-max, --retry-delay).
 
 The retrieved logs are saved to the specified output file (default: 'export.ndjson') in NDJSON format.
 The process can be interrupted at any time (Ctrl+C), and it will attempt to save already retrieved logs before exiting.`,
 		RunE: func(cmd *cobra.Command, args []string) (err error) {
 			ctx := cmd.Context()
 
-			ec, err := ethclient.NewClient(ctx, rpcEndpoint, ethclient.WithRateLimit(maxRequest), ethclient.WithLogger(c.log))
+			if retryMax < 0 {
+				return fmt.Errorf("invalid --retry-max %d: must not be negative", retryMax)
+			}
+			if retryDelay <= 0 {
+				return fmt.Errorf("invalid --retry-delay %s: must be greater than zero", retryDelay)
+			}
+
+			ec, err := ethclient.NewClient(ctx, rpcEndpoint,
+				ethclient.WithRateLimit(maxRequest),
+				ethclient.WithLogger(c.log),
+				ethclient.WithRetry(retryMax, retryDelay),
+			)
 			if err != nil {
 				return fmt.Errorf("failed to connect to the Ethereum client: %w", err)
 			}
@@ -139,6 +153,8 @@ The process can be interrupted at any time (Ctrl+C), and it will attempt to save
 	cmd.Flags().Uint32VarP(&blockRangeLimit, "block-range-limit", "b", 5, "Max blocks per log query")
 	cmd.Flags().StringVarP(&outputFile, "output", "o", "export.ndjson", "Output file path (NDJSON)")
 	cmd.Flags().BoolVarP(&compress, "compress", "c", false, "Compress to GZIP")
+	cmd.Flags().IntVarP(&retryMax, "retry-max", "", 5, "Max retries per RPC request on transient network errors (0 disables retrying)")
+	cmd.Flags().DurationVarP(&retryDelay, "retry-delay", "", ethclient.DefaultRetryDelay, "Delay before the first retry, doubling per retry up to 30s")
 
 	c.root.AddCommand(cmd)
 

pkg/ethclientwrapper/ethclientwrapper.go

@@ -2,7 +2,9 @@ package ethclientwrapper
 
 import (
 	"context"
+	"math/big"
 	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -15,6 +17,7 @@ type Client struct {
 	*ethclient.Client
 	limiter *rate.Limiter
 	logger  log.Logger
+	retry   retryConfig
 	rawURL  string
 	mu      sync.Mutex
 }
@@ -35,6 +38,16 @@ func WithLogger(logger log.Logger) ClientOption {
 	}
 }
 
+// WithRetry retries requests that fail with a transient error, such as a
+// timed out TLS handshake. maxRetries is the number of retries attempted after
+// the initial request, 0 disables retrying. baseDelay is the delay before the
+// first retry, doubling for every further retry.
+func WithRetry(maxRetries int, baseDelay time.Duration) ClientOption {
+	return func(c *Client) {
+		c.retry = newRetryConfig(maxRetries, baseDelay)
+	}
+}
+
 // NewClient creates a new Ethereum client with possible rate limiting.
 func NewClient(ctx context.Context, rawURL string, opts ...ClientOption) (*Client, error) {
 	ethclient, err := ethclient.DialContext(ctx, rawURL)
@@ -62,14 +75,52 @@ func (c *Client) Close() {
 }
 
 func (c *Client) FilterLogs(ctx context.Context, q ethereum.FilterQuery) ([]types.Log, error) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
+	return retryCall(ctx, c.retryConfigFor("FilterLogs"), func() ([]types.Log, error) {
+		c.mu.Lock()
+		defer c.mu.Unlock()
 
-	if err := c.applyRateLimit(ctx); err != nil {
-		return nil, err
-	}
+		if err := c.applyRateLimit(ctx); err != nil {
+			return nil, err
+		}
+
+		return c.Client.FilterLogs(ctx, q)
+	})
+}
+
+func (c *Client) BlockNumber(ctx context.Context) (uint64, error) {
+	return retryCall(ctx, c.retryConfigFor("BlockNumber"), func() (uint64, error) {
+		c.mu.Lock()
+		defer c.mu.Unlock()
 
-	return c.Client.FilterLogs(ctx, q)
+		if err := c.applyRateLimit(ctx); err != nil {
+			return 0, err
+		}
+
+		return c.Client.BlockNumber(ctx)
+	})
+}
+
+func (c *Client) ChainID(ctx context.Context) (*big.Int, error) {
+	return retryCall(ctx, c.retryConfigFor("ChainID"), func() (*big.Int, error) {
+		c.mu.Lock()
+		defer c.mu.Unlock()
+
+		if err := c.applyRateLimit(ctx); err != nil {
+			return nil, err
+		}
+
+		return c.Client.ChainID(ctx)
+	})
+}
+
+// retryConfigFor returns the retry config for the named call, reporting every
+// retry through the client logger.
+func (c *Client) retryConfigFor(call string) retryConfig {
+	cfg := c.retry
+	cfg.onRetry = func(attempt int, delay time.Duration, err error) {
+		c.logger.Warning("retrying rpc call", "call", call, "attempt", attempt, "retry_in", delay, "error", err)
+	}
+	return cfg
 }
 
 // applyRateLimit checks if the limiter is set and applies the rate limit.

pkg/ethclientwrapper/retry.go

@@ -0,0 +1,179 @@
+package ethclientwrapper
+
+import (
+	"context"
+	"errors"
+	"io"
+	"math/rand/v2"
+	"net"
+	"syscall"
+	"time"
+
+	"github.com/ethereum/go-ethereum/rpc"
+)
+
+const (
+	// DefaultRetryDelay is the base delay applied before the first retry.
+	DefaultRetryDelay = time.Second
+
+	// maxRetryDelay caps the exponential backoff between retries.
+	maxRetryDelay = 30 * time.Second
+
+	// rpcErrCodeLimitExceeded is the JSON-RPC error code endpoints return when
+	// the client is being rate limited. Unlike other server-side errors, it is
+	// worth retrying.
+	rpcErrCodeLimitExceeded = -32005
+)
+
+// retryConfig describes how transient RPC failures are retried. Its zero value
+// performs no retries.
+type retryConfig struct {
+	// maxRetries is the number of retries attempted after the initial call.
+	maxRetries int
+	baseDelay  time.Duration
+	maxDelay   time.Duration
+
+	// sleep, jitter and onRetry are injectable to keep the backoff testable.
+	sleep   func(context.Context, time.Duration) error
+	jitter  func(time.Duration) time.Duration
+	onRetry func(attempt int, delay time.Duration, err error)
+}
+
+// newRetryConfig returns a config retrying up to maxRetries times, starting
+// with baseDelay and doubling it up to maxRetryDelay.
+func newRetryConfig(maxRetries int, baseDelay time.Duration) retryConfig {
+	return retryConfig{
+		maxRetries: maxRetries,
+		baseDelay:  baseDelay,
+		maxDelay:   maxRetryDelay,
+	}
+}
+
+// withDefaults fills in the fields left unset by the caller.
+func (c retryConfig) withDefaults() retryConfig {
+	if c.maxRetries < 0 {
+		c.maxRetries = 0
+	}
+	if c.baseDelay <= 0 {
+		c.baseDelay = DefaultRetryDelay
+	}
+	if c.maxDelay < c.baseDelay {
+		c.maxDelay = c.baseDelay
+	}
+	if c.sleep == nil {
+		c.sleep = sleepCtx
+	}
+	if c.jitter == nil {
+		c.jitter = jitter
+	}
+	if c.onRetry == nil {
+		c.onRetry = func(int, time.Duration, error) {}
+	}
+	return c
+}
+
+// retryCall calls fn, retrying transient failures with an exponential backoff.
+// It gives up as soon as the error is not transient, the retries are exhausted
+// or ctx is done, and then returns the error of the last attempt.
+func retryCall[T any](ctx context.Context, cfg retryConfig, fn func() (T, error)) (T, error) {
+	cfg = cfg.withDefaults()
+
+	for attempt := 0; ; attempt++ {
+		result, err := fn()
+		if err == nil {
+			return result, nil
+		}
+
+		if attempt >= cfg.maxRetries || !isRetryable(err) || ctx.Err() != nil {
+			return result, err
+		}
+
+		delay := cfg.jitter(backoffDelay(attempt+1, cfg.baseDelay, cfg.maxDelay))
+		cfg.onRetry(attempt+1, delay, err)
+
+		if sleepErr := cfg.sleep(ctx, delay); sleepErr != nil {
+			return result, err
+		}
+	}
+}
+
+// backoffDelay returns the delay before the given 1-based retry attempt,
+// doubling base per attempt without ever exceeding max.
+func backoffDelay(attempt int, base, maxDelay time.Duration) time.Duration {
+	delay := base
+	for i := 1; i < attempt; i++ {
+		delay *= 2
+		// A delay that is no longer positive means the doubling overflowed.
+		if delay <= 0 || delay >= maxDelay {
+			return maxDelay
+		}
+	}
+	return delay
+}
+
+// jitter spreads the delay over the second half of the backoff window so that
+// repeated retries do not hit the endpoint in lockstep.
+func jitter(d time.Duration) time.Duration {
+	if d <= 0 {
+		return 0
+	}
+	return d/2 + time.Duration(rand.Int64N(int64(d/2)+1))
+}
+
+// sleepCtx waits for d, returning early if ctx is done.
+func sleepCtx(ctx context.Context, d time.Duration) error {
+	if d <= 0 {
+		return ctx.Err()
+	}
+
+	timer := time.NewTimer(d)
+	defer timer.Stop()
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return nil
+	}
+}
+
+// isRetryable reports whether err is a transient failure that a later attempt
+// may recover from, such as a dropped connection or a timed out TLS handshake.
+func isRetryable(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+		return false
+	}
+
+	// The endpoint answered with an HTTP error: retry only if it is temporary.
+	var httpErr rpc.HTTPError
+	if errors.As(err, &httpErr) {
+		return httpErr.StatusCode == 429 || httpErr.StatusCode >= 500
+	}
+
+	// The endpoint answered with a JSON-RPC error, so the request reached it and
+	// replaying it would fail the same way, unless we are being rate limited.
+	var rpcErr rpc.Error
+	if errors.As(err, &rpcErr) {
+		return rpcErr.ErrorCode() == rpcErrCodeLimitExceeded
+	}
+
+	// Transport level failures: timeouts, TLS handshake failures, DNS errors.
+	var netErr net.Error
+	if errors.As(err, &netErr) {
+		return true
+	}
+
+	if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
+		return true
+	}
+
+	return errors.Is(err, syscall.ECONNRESET) ||
+		errors.Is(err, syscall.ECONNREFUSED) ||
+		errors.Is(err, syscall.ECONNABORTED) ||
+		errors.Is(err, syscall.EPIPE) ||
+		errors.Is(err, syscall.ETIMEDOUT)
+}