eth: Create separate pricefeed client unit

This will make the code more testable.

eth/pricefeed.go

@@ -0,0 +1,107 @@
+package eth
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math/big"
+	"regexp"
+	"time"
+
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/ethclient"
+	"github.com/livepeer/go-livepeer/eth/contracts/chainlink"
+)
+
+type PriceData struct {
+	RoundID   int64
+	Price     *big.Rat
+	UpdatedAt time.Time
+}
+
+type PriceFeedEthClient interface {
+	Description() (string, error)
+	FetchPriceData() (PriceData, error)
+}
+
+func NewPriceFeedEthClient(ctx context.Context, rpcUrl, priceFeedAddr string) (PriceFeedEthClient, error) {
+	client, err := ethclient.DialContext(ctx, rpcUrl)
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize client: %w", err)
+	}
+
+	ok := isContractAddress(priceFeedAddr, client)
+	if !ok {
+		return nil, fmt.Errorf("not a contract address: %s", priceFeedAddr)
+	}
+
+	addr := common.HexToAddress(priceFeedAddr)
+	priceFeed, err := chainlink.NewAggregatorV3Interface(addr, client)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create mock aggregator proxy: %w", err)
+	}
+
+	return &priceFeedClient{
+		client:    client,
+		priceFeed: priceFeed,
+	}, nil
+}
+
+type priceFeedClient struct {
+	client    *ethclient.Client
+	priceFeed *chainlink.AggregatorV3Interface
+}
+
+func (c *priceFeedClient) Description() (string, error) {
+	return c.priceFeed.Description(&bind.CallOpts{})
+}
+
+func (c *priceFeedClient) FetchPriceData() (PriceData, error) {
+	data, err := c.priceFeed.LatestRoundData(&bind.CallOpts{})
+	if err != nil {
+		return PriceData{}, errors.New("failed to get latest round data: " + err.Error())
+	}
+
+	decimals, err := c.priceFeed.Decimals(&bind.CallOpts{})
+	if err != nil {
+		return PriceData{}, errors.New("failed to get decimals: " + err.Error())
+	}
+
+	return computePriceData(data.RoundId, data.UpdatedAt, data.Answer, decimals), nil
+}
+
+func computePriceData(roundID, updatedAt, answer *big.Int, decimals uint8) PriceData {
+	// Compute a big.int which is 10^decimals.
+	divisor := new(big.Int).Exp(
+		big.NewInt(10),
+		big.NewInt(int64(decimals)),
+		nil)
+
+	return PriceData{
+		RoundID:   roundID.Int64(),
+		Price:     new(big.Rat).SetFrac(answer, divisor),
+		UpdatedAt: time.Unix(updatedAt.Int64(), 0),
+	}
+}
+
+func isContractAddress(addr string, client *ethclient.Client) bool {
+	if len(addr) == 0 {
+		return false
+	}
+
+	// Ensure it is an Ethereum address: 0x followed by 40 hexadecimal characters.
+	re := regexp.MustCompile("^0x[0-9a-fA-F]{40}$")
+	if !re.MatchString(addr) {
+		return false
+	}
+
+	// Ensure it is a contract address.
+	address := common.HexToAddress(addr)
+	bytecode, err := client.CodeAt(context.Background(), address, nil) // nil is latest block
+	if err != nil {
+		return false
+	}
+	isContract := len(bytecode) > 0
+	return isContract
+}

eth/watchers/pricefeedwatcher.go

@@ -3,16 +3,11 @@ package watchers
 import (
 	"context"
 	"fmt"
-	"math/big"
-	"regexp"
 	"strings"
 	"time"
 
-	"github.com/ethereum/go-ethereum/accounts/abi/bind"
-	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/livepeer/go-livepeer/clog"
-	"github.com/livepeer/go-livepeer/eth/contracts/chainlink"
+	"github.com/livepeer/go-livepeer/eth"
 )
 
 const (
@@ -23,41 +18,25 @@ const (
 type PriceFeedWatcher struct {
 	ctx context.Context
 
-	client *ethclient.Client
-	proxy  *chainlink.AggregatorV3Interface
-
+	updatePeriod                time.Duration
+	priceFeed                   eth.PriceFeedEthClient
 	currencyBase, currencyQuote string
 
-	current      PriceData
-	priceUpdated chan PriceData
-}
-
-type PriceData struct {
-	RoundId   int64
-	Price     *big.Rat
-	UpdatedAt time.Time
+	current      eth.PriceData
+	priceUpdated chan eth.PriceData
 }
 
-func NewPriceFeedWatcher(ctx context.Context, rpcUrl, proxyAddrStr string) (*PriceFeedWatcher, error) {
-	// Initialize client instance using the rpcUrl.
-	client, err := ethclient.DialContext(ctx, rpcUrl)
-	if err != nil {
-		return nil, fmt.Errorf("failed to initialize client: %w", err)
+func NewPriceFeedWatcher(ctx context.Context, rpcUrl, priceFeedAddr string, updatePeriod time.Duration) (*PriceFeedWatcher, error) {
+	if updatePeriod <= 0 {
+		updatePeriod = 1 * time.Hour
 	}
 
-	// Test if it is a contract address.
-	ok := isContractAddress(proxyAddrStr, client)
-	if !ok {
-		return nil, fmt.Errorf("not a contract address: %s", proxyAddrStr)
-	}
-
-	proxyAddr := common.HexToAddress(proxyAddrStr)
-	proxy, err := chainlink.NewAggregatorV3Interface(proxyAddr, client)
+	priceFeed, err := eth.NewPriceFeedEthClient(ctx, rpcUrl, priceFeedAddr)
 	if err != nil {
-		return nil, fmt.Errorf("failed to create mock aggregator proxy: %w", err)
+		return nil, fmt.Errorf("failed to create price feed client: %w", err)
 	}
 
-	description, err := proxy.Description(&bind.CallOpts{})
+	description, err := priceFeed.Description()
 	if err != nil {
 		return nil, fmt.Errorf("failed to get description: %w", err)
 	}
@@ -69,14 +48,14 @@ func NewPriceFeedWatcher(ctx context.Context, rpcUrl, proxyAddrStr string) (*Pri
 
 	w := &PriceFeedWatcher{
 		ctx:           ctx,
-		client:        client,
-		proxy:         proxy,
+		updatePeriod:  updatePeriod,
+		priceFeed:     priceFeed,
 		currencyBase:  currencyFrom,
 		currencyQuote: currencyTo,
-		priceUpdated:  make(chan PriceData, 1),
+		priceUpdated:  make(chan eth.PriceData, 1),
 	}
 
-	err = w.fetchPrice()
+	err = w.updatePrice()
 	if err != nil {
 		return nil, fmt.Errorf("failed to update price: %w", err)
 	}
@@ -91,40 +70,18 @@ func (w *PriceFeedWatcher) Currencies() (base string, quote string) {
 	return w.currencyBase, w.currencyQuote
 }
 
-func (w *PriceFeedWatcher) Current() PriceData {
+func (w *PriceFeedWatcher) Current() eth.PriceData {
 	return w.current
 }
 
-func (w *PriceFeedWatcher) PriceUpdated() <-chan PriceData {
+func (w *PriceFeedWatcher) PriceUpdated() <-chan eth.PriceData {
 	return w.priceUpdated
 }
 
-func (w *PriceFeedWatcher) fetchPrice() error {
-	roundData, err := w.proxy.LatestRoundData(&bind.CallOpts{})
+func (w *PriceFeedWatcher) updatePrice() error {
+	newPrice, err := w.priceFeed.FetchPriceData()
 	if err != nil {
-		return fmt.Errorf("failed to get latest round data: %w", err)
-	}
-
-	decimals, err := w.proxy.Decimals(&bind.CallOpts{})
-	if err != nil {
-		return fmt.Errorf("failed to get decimals: %w", err)
-	}
-
-	w.updatePrice(roundData.Answer, decimals, roundData.RoundId, roundData.UpdatedAt)
-	return nil
-}
-
-func (w *PriceFeedWatcher) updatePrice(current *big.Int, decimals uint8, roundId, updatedAt *big.Int) {
-	// Compute a big.int which is 10^decimals.
-	divisor := new(big.Int).Exp(
-		big.NewInt(10),
-		big.NewInt(int64(decimals)),
-		nil)
-
-	newPrice := PriceData{
-		RoundId:   roundId.Int64(),
-		Price:     new(big.Rat).SetFrac(current, divisor),
-		UpdatedAt: time.Unix(updatedAt.Int64(), 0),
+		return fmt.Errorf("failed to fetch price data: %w", err)
 	}
 
 	if newPrice.UpdatedAt.After(w.current.UpdatedAt) {
@@ -134,54 +91,38 @@ func (w *PriceFeedWatcher) updatePrice(current *big.Int, decimals uint8, roundId
 		default:
 		}
 	}
+
+	return nil
 }
 
 func (w *PriceFeedWatcher) watch() {
 	ctx, cancel := context.WithCancel(w.ctx)
 	defer cancel()
-	ticker := newTruncatedTicker(ctx, 1*time.Hour)
+	ticker := newTruncatedTicker(ctx, w.updatePeriod)
 
 	for {
 		select {
 		case <-w.ctx.Done():
 			return
 		case <-ticker:
-			retryDelay := priceUpdateBaseRetryDelay
-			for attempt := 1; attempt <= priceUpdateMaxRetries; attempt++ {
-				err := w.fetchPrice()
+			attempt, retryDelay := 1, priceUpdateBaseRetryDelay
+			for {
+				err := w.updatePrice()
 				if err == nil {
 					break
+				} else if attempt >= priceUpdateMaxRetries {
+					clog.Errorf(ctx, "Failed to fetch updated price from PriceFeed attempts=%d err=%q", attempt, err)
+					break
 				}
 
-				clog.Warningf(ctx, "Failed to fetch updated price from PriceFeed, retrying after retryDelay=%d attempt=%d err=%v", retryDelay, attempt, err)
+				clog.Warningf(ctx, "Failed to fetch updated price from PriceFeed, retrying after retryDelay=%d attempt=%d err=%q", retryDelay, attempt, err)
 				time.Sleep(retryDelay)
-				retryDelay *= 2
+				attempt, retryDelay = attempt+1, retryDelay*2
 			}
 		}
 	}
 }
 
-func isContractAddress(addr string, client *ethclient.Client) bool {
-	if len(addr) == 0 {
-		return false
-	}
-
-	// Ensure it is an Ethereum address: 0x followed by 40 hexadecimal characters.
-	re := regexp.MustCompile("^0x[0-9a-fA-F]{40}$")
-	if !re.MatchString(addr) {
-		return false
-	}
-
-	// Ensure it is a contract address.
-	address := common.HexToAddress(addr)
-	bytecode, err := client.CodeAt(context.Background(), address, nil) // nil is latest block
-	if err != nil {
-		return false
-	}
-	isContract := len(bytecode) > 0
-	return isContract
-}
-
 func parseCurrencies(description string) (currencyBase string, currencyQuote string, err error) {
 	currencies := strings.Split(description, "/")
 	if len(currencies) != 2 {