Merge pull request #1585 from Roasbeef/retry-n-block

 chain_bridge+fn: add retry logic for RPC calls to improve reliability in load-balanced environments

Author: Roasbeef

chain_bridge.go

@@ -11,6 +11,7 @@ import (
 	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/neutrino/cache/lru"
 	"github.com/lightninglabs/taproot-assets/asset"
+	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightninglabs/taproot-assets/rfq"
 	"github.com/lightninglabs/taproot-assets/tapchannel"
@@ -49,6 +50,7 @@ type LndRpcChainBridge struct {
 	lnd *lndclient.LndServices
 
 	blockTimestampCache *lru.Cache[uint32, cacheableTimestamp]
+	retryConfig         fn.RetryConfig
 
 	assetStore *tapdb.AssetStore
 }
@@ -63,7 +65,8 @@ func NewLndRpcChainBridge(lnd *lndclient.LndServices,
 		blockTimestampCache: lru.NewCache[uint32, cacheableTimestamp](
 			maxNumBlocksInCache,
 		),
-		assetStore: assetStore,
+		retryConfig: fn.DefaultRetryConfig(),
+		assetStore:  assetStore,
 	}
 }
 
@@ -112,39 +115,56 @@ func (l *LndRpcChainBridge) RegisterBlockEpochNtfn(
 func (l *LndRpcChainBridge) GetBlock(ctx context.Context,
 	hash chainhash.Hash) (*wire.MsgBlock, error) {
 
-	block, err := l.lnd.ChainKit.GetBlock(ctx, hash)
-	if err != nil {
-		return nil, fmt.Errorf("unable to retrieve block: %w", err)
-	}
-
-	return block, nil
+	return fn.RetryFuncN(
+		ctx, l.retryConfig, func() (*wire.MsgBlock, error) {
+			block, err := l.lnd.ChainKit.GetBlock(ctx, hash)
+			if err != nil {
+				return nil, fmt.Errorf(
+					"unable to retrieve block: %w", err,
+				)
+			}
+			return block, nil
+		},
+	)
 }
 
 // GetBlockHeader returns a block header given its hash.
 func (l *LndRpcChainBridge) GetBlockHeader(ctx context.Context,
 	hash chainhash.Hash) (*wire.BlockHeader, error) {
 
-	header, err := l.lnd.ChainKit.GetBlockHeader(ctx, hash)
-	if err != nil {
-		return nil, fmt.Errorf("unable to retrieve block header: %w",
-			err)
-	}
-
-	return header, nil
+	return fn.RetryFuncN(
+		ctx, l.retryConfig, func() (*wire.BlockHeader, error) {
+			header, err := l.lnd.ChainKit.GetBlockHeader(ctx, hash)
+			if err != nil {
+				return nil, fmt.Errorf(
+					"unable to retrieve block "+
+						"header: %w", err,
+				)
+			}
+			return header, nil
+		},
+	)
 }
 
 // GetBlockHash returns the hash of the block in the best blockchain at the
 // given height.
 func (l *LndRpcChainBridge) GetBlockHash(ctx context.Context,
 	blockHeight int64) (chainhash.Hash, error) {
 
-	blockHash, err := l.lnd.ChainKit.GetBlockHash(ctx, blockHeight)
-	if err != nil {
-		return chainhash.Hash{}, fmt.Errorf("unable to retrieve "+
-			"block hash: %w", err)
-	}
-
-	return blockHash, nil
+	return fn.RetryFuncN(
+		ctx, l.retryConfig, func() (chainhash.Hash, error) {
+			blockHash, err := l.lnd.ChainKit.GetBlockHash(
+				ctx, blockHeight,
+			)
+			if err != nil {
+				return chainhash.Hash{}, fmt.Errorf(
+					"unable to retrieve block hash: %w",
+					err,
+				)
+			}
+			return blockHash, nil
+		},
+	)
 }
 
 // VerifyBlock returns an error if a block (with given header and height) is not
@@ -184,12 +204,17 @@ func (l *LndRpcChainBridge) VerifyBlock(ctx context.Context,
 
 // CurrentHeight return the current height of the main chain.
 func (l *LndRpcChainBridge) CurrentHeight(ctx context.Context) (uint32, error) {
-	_, bestHeight, err := l.lnd.ChainKit.GetBestBlock(ctx)
-	if err != nil {
-		return 0, fmt.Errorf("unable to grab block height: %w", err)
-	}
-
-	return uint32(bestHeight), nil
+	return fn.RetryFuncN(
+		ctx, l.retryConfig, func() (uint32, error) {
+			_, bestHeight, err := l.lnd.ChainKit.GetBestBlock(ctx)
+			if err != nil {
+				return 0, fmt.Errorf(
+					"unable to grab block height: %w", err,
+				)
+			}
+			return uint32(bestHeight), nil
+		},
+	)
 }
 
 // GetBlockTimestamp returns the timestamp of the block at the given height.
@@ -207,7 +232,11 @@ func (l *LndRpcChainBridge) GetBlockTimestamp(ctx context.Context,
 		return int64(cacheTS)
 	}
 
-	hash, err := l.lnd.ChainKit.GetBlockHash(ctx, int64(height))
+	hash, err := fn.RetryFuncN(
+		ctx, l.retryConfig, func() (chainhash.Hash, error) {
+			return l.lnd.ChainKit.GetBlockHash(ctx, int64(height))
+		},
+	)
 	if err != nil {
 		return 0
 	}
@@ -229,14 +258,27 @@ func (l *LndRpcChainBridge) GetBlockTimestamp(ctx context.Context,
 func (l *LndRpcChainBridge) PublishTransaction(ctx context.Context,
 	tx *wire.MsgTx, label string) error {
 
-	return l.lnd.WalletKit.PublishTransaction(ctx, tx, label)
+	_, err := fn.RetryFuncN(
+		ctx, l.retryConfig, func() (struct{}, error) {
+			return struct{}{}, l.lnd.WalletKit.PublishTransaction(
+				ctx, tx, label,
+			)
+		},
+	)
+	return err
 }
 
 // EstimateFee returns a fee estimate for the confirmation target.
 func (l *LndRpcChainBridge) EstimateFee(ctx context.Context,
 	confTarget uint32) (chainfee.SatPerKWeight, error) {
 
-	return l.lnd.WalletKit.EstimateFeeRate(ctx, int32(confTarget))
+	return fn.RetryFuncN(
+		ctx, l.retryConfig, func() (chainfee.SatPerKWeight, error) {
+			return l.lnd.WalletKit.EstimateFeeRate(
+				ctx, int32(confTarget),
+			)
+		},
+	)
 }
 
 // GenFileChainLookup generates a chain lookup interface for the given

fn/retry.go

@@ -0,0 +1,86 @@
+package fn
+
+import (
+	"context"
+	"time"
+)
+
+// RetryConfig defines the parameters for exponential backoff retry behavior.
+type RetryConfig struct {
+	// MaxRetries specifies how many times to retry after the initial
+	// attempt fails.
+	MaxRetries int
+
+	// InitialBackoff sets the delay before the first retry attempt.
+	InitialBackoff time.Duration
+
+	// BackoffMultiplier determines the exponential growth rate of the
+	// backoff duration between successive retries.
+	BackoffMultiplier float64
+
+	// MaxBackoff caps the delay between retries to prevent excessive
+	// wait times.
+	MaxBackoff time.Duration
+}
+
+// DefaultRetryConfig provides sensible defaults for retrying RPC calls in
+// load-balanced environments where transient failures are expected.
+func DefaultRetryConfig() RetryConfig {
+	return RetryConfig{
+		MaxRetries:        10,
+		InitialBackoff:    100 * time.Millisecond,
+		BackoffMultiplier: 2.0,
+		MaxBackoff:        5 * time.Second,
+	}
+}
+
+// RetryFuncN executes the provided function with exponential backoff retry
+// logic. This is particularly useful for RPC calls in load-balanced
+// environments where nodes may temporarily return inconsistent results. The
+// function respects context cancellation and returns immediately if the context
+// is cancelled.
+func RetryFuncN[T any](ctx context.Context,
+	config RetryConfig, fn func() (T, error)) (T, error) {
+
+	var (
+		result T
+		err    error
+	)
+
+	backoff := config.InitialBackoff
+
+	// We'll retry the function up to MaxRetries times, backing off each
+	// time until it succeeds.
+	for attempt := 0; attempt <= config.MaxRetries; attempt++ {
+		result, err = fn()
+		if err == nil {
+			return result, nil
+		}
+
+		if attempt == config.MaxRetries {
+			return result, err
+		}
+
+		// Cap the backoff at the configured maximum to prevent
+		// excessive delays.
+		if backoff > config.MaxBackoff {
+			backoff = config.MaxBackoff
+		}
+
+		// Wait for the backoff duration or until the context is
+		// cancelled, whichever comes first.
+		select {
+		case <-ctx.Done():
+			return result, ctx.Err()
+
+		case <-time.After(backoff):
+			// Apply the multiplier to implement exponential
+			// backoff.
+			backoff = time.Duration(
+				float64(backoff) * config.BackoffMultiplier,
+			)
+		}
+	}
+
+	return result, err
+}