eth,eth/watcher: Create Chainlink price feed watcher (#2972)

* eth/watchers: Create PriceFeed watcher

Makefile: Use mockgen binary from tool dependencies

eth/contracts: Add chainlink interfaces source

Makefile: Generate Chainlink contracts ABI

tools: Add abigen tool to repo

eth/contracts: Generate chainlink bindings

Makefile: Fix abigen bindings generation

Revert everything abigen

Turns out there's already bindings exported from the Chainlink lib.

go.mod: Add chainlink library

eth/watchers: Add pricefeed watcher

eth/watchers: Clean-up event watching code

eth/watchers: Improve price tracking

Revert "go.mod: Add chainlink library"

This reverts commit ac415bd.

Revert "Revert everything abigen"

This reverts commit b7c40b1.

eth/contracts: Gen bindings for proxy iface

eth/watchers: Use local bindings for contracts

eth/watchers: Simplify event subs logic

eth/watchers: Simplify&optimize truncated ticker

eth/watchers: Update decimals on fetch

eth/watchers: Improve handling of decimals

eth/watchers: Fix price rat creation

eth/watchers: Make sure we use UTC on truncated timer

eth/contracts/chainlink: Generate only V3 contract bindings

eth/watchers: Watch PriceFeed only with polling

eth/watchers: Add a retry logic on price update

eth/watchers: Use clog instead of fmt.Printf

* eth: Create separate pricefeed client unit

This will make the code more testable.

* eth: Add tests for pricefeed client

* eth/watchers: Add tests to the truncated ticker

Gosh that was much harder than I thought

* eth/watchers: Add tests for pricefeedwatcher

* eth: Add comments to the new components

* go fmt

* eth: Address minor review comments

* eth,eth/watchers: Improve pricefeed watcher interface

* eth/watchers: Remove truncated ticker tests

Author: victorges

Makefile

@@ -1,7 +1,10 @@
 SHELL=/bin/bash
 GO_BUILD_DIR?="./"
 
-all: net/lp_rpc.pb.go net/redeemer.pb.go net/redeemer_mock.pb.go core/test_segment.go livepeer livepeer_cli livepeer_router livepeer_bench
+MOCKGEN=go run github.com/golang/mock/mockgen
+ABIGEN=go run github.com/ethereum/go-ethereum/cmd/abigen
+
+all: net/lp_rpc.pb.go net/redeemer.pb.go net/redeemer_mock.pb.go core/test_segment.go eth/contracts/chainlink/AggregatorV3Interface.go livepeer livepeer_cli livepeer_router livepeer_bench
 
 net/lp_rpc.pb.go: net/lp_rpc.proto
 	protoc -I=. --go_out=. --go-grpc_out=. $^
@@ -10,12 +13,21 @@ net/redeemer.pb.go: net/redeemer.proto
 	protoc -I=. --go_out=. --go-grpc_out=. $^
 
 net/redeemer_mock.pb.go net/redeemer_grpc_mock.pb.go: net/redeemer.pb.go net/redeemer_grpc.pb.go
-	@mockgen -source net/redeemer.pb.go -destination net/redeemer_mock.pb.go -package net
-	@mockgen -source net/redeemer_grpc.pb.go -destination net/redeemer_grpc_mock.pb.go -package net
+	@$(MOCKGEN) -source net/redeemer.pb.go -destination net/redeemer_mock.pb.go -package net
+	@$(MOCKGEN) -source net/redeemer_grpc.pb.go -destination net/redeemer_grpc_mock.pb.go -package net
 
 core/test_segment.go:
 	core/test_segment.sh core/test_segment.go
 
+eth/contracts/chainlink/AggregatorV3Interface.go:
+	solc --version | grep 0.7.6+commit.7338295f
+	@set -ex; \
+	for sol_file in eth/contracts/chainlink/*.sol; do \
+		contract_name=$$(basename "$$sol_file" .sol); \
+		solc --abi --optimize --overwrite -o $$(dirname "$$sol_file") $$sol_file; \
+		$(ABIGEN) --abi=$${sol_file%.sol}.abi --pkg=chainlink --type=$$contract_name --out=$${sol_file%.sol}.go; \
+	done
+
 version=$(shell cat VERSION)
 
 ldflags := -X github.com/livepeer/go-livepeer/core.LivepeerVersion=$(shell ./print_version.sh)

eth/contracts/chainlink/AggregatorV3Interface.abi

@@ -0,0 +1 @@
+[{"inputs":[],"name":"decimals","outputs":[{"internalType":"uint8","name":"","type":"uint8"}],"stateMutability":"view","type":"function"},{"inputs":[],"name":"description","outputs":[{"internalType":"string","name":"","type":"string"}],"stateMutability":"view","type":"function"},{"inputs":[{"internalType":"uint80","name":"_roundId","type":"uint80"}],"name":"getRoundData","outputs":[{"internalType":"uint80","name":"roundId","type":"uint80"},{"internalType":"int256","name":"answer","type":"int256"},{"internalType":"uint256","name":"startedAt","type":"uint256"},{"internalType":"uint256","name":"updatedAt","type":"uint256"},{"internalType":"uint80","name":"answeredInRound","type":"uint80"}],"stateMutability":"view","type":"function"},{"inputs":[],"name":"latestRoundData","outputs":[{"internalType":"uint80","name":"roundId","type":"uint80"},{"internalType":"int256","name":"answer","type":"int256"},{"internalType":"uint256","name":"startedAt","type":"uint256"},{"internalType":"uint256","name":"updatedAt","type":"uint256"},{"internalType":"uint80","name":"answeredInRound","type":"uint80"}],"stateMutability":"view","type":"function"},{"inputs":[],"name":"version","outputs":[{"internalType":"uint256","name":"","type":"uint256"}],"stateMutability":"view","type":"function"}]

eth/contracts/chainlink/AggregatorV3Interface.go

@@ -0,0 +1,36 @@
+// SPDX-License-Identifier: MIT
+// https://github.com/smartcontractkit/chainlink/blob/v2.9.1/contracts/src/v0.7/interfaces/AggregatorV3Interface.sol
+pragma solidity ^0.7.0;
+
+interface AggregatorV3Interface {
+    function decimals() external view returns (uint8);
+
+    function description() external view returns (string memory);
+
+    function version() external view returns (uint256);
+
+    // getRoundData and latestRoundData should both raise "No data present"
+    // if they do not have data to report, instead of returning unset values
+    // which could be misinterpreted as actual reported values.
+    function getRoundData(uint80 _roundId)
+        external
+        view
+        returns (
+            uint80 roundId,
+            int256 answer,
+            uint256 startedAt,
+            uint256 updatedAt,
+            uint80 answeredInRound
+        );
+
+    function latestRoundData()
+        external
+        view
+        returns (
+            uint80 roundId,
+            int256 answer,
+            uint256 startedAt,
+            uint256 updatedAt,
+            uint80 answeredInRound
+        );
+}

eth/contracts/chainlink/AggregatorV3Interface.sol

@@ -0,0 +1,78 @@
+package eth
+
+import (
+	"errors"
+	"fmt"
+	"math/big"
+	"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
+}
+
+// PriceFeedEthClient is an interface for fetching price data from a Chainlink
+// PriceFeed contract.
+type PriceFeedEthClient interface {
+	Description() (string, error)
+	FetchPriceData() (PriceData, error)
+}
+
+func NewPriceFeedEthClient(ethClient *ethclient.Client, priceFeedAddr string) (PriceFeedEthClient, error) {
+	addr := common.HexToAddress(priceFeedAddr)
+	priceFeed, err := chainlink.NewAggregatorV3Interface(addr, ethClient)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create aggregator proxy: %w", err)
+	}
+
+	return &priceFeedClient{
+		client:    ethClient,
+		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
+}
+
+// computePriceData transforms the raw data from the PriceFeed into the higher
+// level PriceData struct, more easily usable by the rest of the system.
+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),
+	}
+}

eth/pricefeed.go

@@ -0,0 +1,51 @@
+package eth
+
+import (
+	"math/big"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestComputePriceData(t *testing.T) {
+	assert := assert.New(t)
+
+	t.Run("valid data", func(t *testing.T) {
+		roundID := big.NewInt(1)
+		updatedAt := big.NewInt(1626192000)
+		answer := big.NewInt(420666000)
+		decimals := uint8(6)
+
+		data := computePriceData(roundID, updatedAt, answer, decimals)
+
+		assert.EqualValues(int64(1), data.RoundID, "Round ID didn't match")
+		assert.Equal("210333/500", data.Price.RatString(), "The Price Rat didn't match")
+		assert.Equal("2021-07-13 16:00:00 +0000 UTC", data.UpdatedAt.UTC().String(), "The updated at time did not match")
+	})
+
+	t.Run("zero answer", func(t *testing.T) {
+		roundID := big.NewInt(2)
+		updatedAt := big.NewInt(1626192000)
+		answer := big.NewInt(0)
+		decimals := uint8(18)
+
+		data := computePriceData(roundID, updatedAt, answer, decimals)
+
+		assert.EqualValues(int64(2), data.RoundID, "Round ID didn't match")
+		assert.Equal("0", data.Price.RatString(), "The Price Rat didn't match")
+		assert.Equal("2021-07-13 16:00:00 +0000 UTC", data.UpdatedAt.UTC().String(), "The updated at time did not match")
+	})
+
+	t.Run("zero decimals", func(t *testing.T) {
+		roundID := big.NewInt(3)
+		updatedAt := big.NewInt(1626192000)
+		answer := big.NewInt(13)
+		decimals := uint8(0)
+
+		data := computePriceData(roundID, updatedAt, answer, decimals)
+
+		assert.EqualValues(int64(3), data.RoundID, "Round ID didn't match")
+		assert.Equal("13", data.Price.RatString(), "The Price Rat didn't match")
+		assert.Equal("2021-07-13 16:00:00 +0000 UTC", data.UpdatedAt.UTC().String(), "The updated at time did not match")
+	})
+}

eth/pricefeed_test.go

@@ -0,0 +1,182 @@
+package watchers
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/ethclient"
+	"github.com/ethereum/go-ethereum/event"
+	"github.com/livepeer/go-livepeer/clog"
+	"github.com/livepeer/go-livepeer/eth"
+)
+
+const (
+	priceUpdateMaxRetries     = 5
+	priceUpdateBaseRetryDelay = 30 * time.Second
+	priceUpdatePeriod         = 1 * time.Hour
+)
+
+// PriceFeedWatcher monitors a Chainlink PriceFeed for updated pricing info. It
+// allows fetching the current price as well as listening for updates on the
+// PriceUpdated channel.
+type PriceFeedWatcher struct {
+	baseRetryDelay time.Duration
+
+	priceFeed                   eth.PriceFeedEthClient
+	currencyBase, currencyQuote string
+
+	mu             sync.RWMutex
+	current        eth.PriceData
+	priceEventFeed event.Feed
+}
+
+// NewPriceFeedWatcher creates a new PriceFeedWatcher instance. It will already
+// fetch the current price and start a goroutine to watch for updates.
+func NewPriceFeedWatcher(ethClient *ethclient.Client, priceFeedAddr string) (*PriceFeedWatcher, error) {
+	priceFeed, err := eth.NewPriceFeedEthClient(ethClient, priceFeedAddr)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create price feed client: %w", err)
+	}
+
+	description, err := priceFeed.Description()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get description: %w", err)
+	}
+
+	currencyFrom, currencyTo, err := parseCurrencies(description)
+	if err != nil {
+		return nil, err
+	}
+
+	w := &PriceFeedWatcher{
+		baseRetryDelay: priceUpdateBaseRetryDelay,
+		priceFeed:      priceFeed,
+		currencyBase:   currencyFrom,
+		currencyQuote:  currencyTo,
+	}
+
+	err = w.updatePrice()
+	if err != nil {
+		return nil, fmt.Errorf("failed to update price: %w", err)
+	}
+
+	return w, nil
+}
+
+// Currencies returns the base and quote currencies of the price feed.
+// i.e. base = CurrentPrice() * quote
+func (w *PriceFeedWatcher) Currencies() (base string, quote string) {
+	return w.currencyBase, w.currencyQuote
+}
+
+// Current returns the latest fetched price data.
+func (w *PriceFeedWatcher) Current() eth.PriceData {
+	w.mu.RLock()
+	defer w.mu.RUnlock()
+	return w.current
+}
+
+// Subscribe allows one to subscribe to price updates emitted by the Watcher.
+// To unsubscribe, simply call `Unsubscribe` on the returned subscription.
+// The sink channel should have ample buffer space to avoid blocking other
+// subscribers. Slow subscribers are not dropped.
+func (w *PriceFeedWatcher) Subscribe(sub chan<- eth.PriceData) event.Subscription {
+	return w.priceEventFeed.Subscribe(sub)
+}
+
+func (w *PriceFeedWatcher) updatePrice() error {
+	newPrice, err := w.priceFeed.FetchPriceData()
+	if err != nil {
+		return fmt.Errorf("failed to fetch price data: %w", err)
+	}
+
+	if newPrice.UpdatedAt.After(w.current.UpdatedAt) {
+		w.mu.Lock()
+		w.current = newPrice
+		w.mu.Unlock()
+		w.priceEventFeed.Send(newPrice)
+	}
+
+	return nil
+}
+
+// Watch starts the watch process. It will periodically poll the price feed for
+// price updates until the given context is canceled. Typically, you want to
+// call Watch inside a goroutine.
+func (w *PriceFeedWatcher) Watch(ctx context.Context) {
+	ticker := newTruncatedTicker(ctx, priceUpdatePeriod)
+	w.watchTicker(ctx, ticker)
+}
+
+func (w *PriceFeedWatcher) watchTicker(ctx context.Context, ticker <-chan time.Time) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker:
+			attempt, retryDelay := 1, w.baseRetryDelay
+			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=%q", retryDelay, attempt, err)
+				select {
+				case <-ctx.Done():
+					return
+				case <-time.After(retryDelay):
+				}
+				attempt, retryDelay = attempt+1, retryDelay*2
+			}
+		}
+	}
+}
+
+// parseCurrencies parses the base and quote currencies from a price feed based
+// on Chainlink PriceFeed description pattern "FROM / TO".
+func parseCurrencies(description string) (currencyBase string, currencyQuote string, err error) {
+	currencies := strings.Split(description, "/")
+	if len(currencies) != 2 {
+		return "", "", fmt.Errorf("aggregator description must be in the format 'FROM / TO' but got: %s", description)
+	}
+
+	currencyBase = strings.TrimSpace(currencies[0])
+	currencyQuote = strings.TrimSpace(currencies[1])
+	return
+}
+
+// newTruncatedTicker creates a ticker that ticks at the next time that is a
+// multiple of d, starting from the current time. This is a best-effort approach
+// to ensure that nodes update their prices around the same time to avoid too
+// big price discrepancies.
+func newTruncatedTicker(ctx context.Context, d time.Duration) <-chan time.Time {
+	ch := make(chan time.Time, 1)
+	go func() {
+		defer close(ch)
+
+		nextTick := time.Now().UTC().Truncate(d)
+		for {
+			nextTick = nextTick.Add(d)
+			untilNextTick := nextTick.Sub(time.Now().UTC())
+			if untilNextTick <= 0 {
+				continue
+			}
+
+			select {
+			case <-ctx.Done():
+				return
+			case t := <-time.After(untilNextTick):
+				ch <- t
+			}
+		}
+	}()
+
+	return ch
+}