feat(oracle-feeder): initialise core logic

CHANGELOG.md

@@ -108,7 +108,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * [#1012](https://github.com/NibiruChain/nibiru/pull/1012) - test(vpool): make vpool simulation with random parameters
 
 ### Features
-
+* [1010](https://github.com/NibiruChain/nibiru/pull/1010) - feeder: initialize oracle feeder core logic
 * [#966](https://github.com/NibiruChain/nibiru/pull/966) - collections: add indexed map
 * [#852](https://github.com/NibiruChain/nibiru/pull/852) - feat(genesis): add cli command to add pairs at genesis
 * [#861](https://github.com/NibiruChain/nibiru/pull/861) - feat: query cumulative funding payments

feeder/Makefile

@@ -0,0 +1,4 @@
+generate:
+	go generate ./...
+
+mocks: generate

feeder/feeder.go

@@ -0,0 +1,146 @@
+package feeder
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/rs/zerolog"
+
+	"github.com/NibiruChain/nibiru/feeder/types"
+)
+
+var (
+	InitTimeout = 5 * time.Second
+)
+
+// votingPeriodContext keeps track of the current voting period.
+type votingPeriodContext struct {
+	height                     uint64
+	votingPeriodDurationBlocks uint64
+	cancelSendPrices           func() // signals cancel send prices for types.PricePoster
+	pricePosterCtx             context.Context
+	sendPricesDone             chan struct{} // response given by types.PricePoster to signal if prices were sent correctly.
+}
+
+// Feeder is the price feeder.
+type Feeder struct {
+	log zerolog.Logger
+
+	stop chan struct{}
+	done chan struct{}
+
+	params types.Params
+
+	votingPeriodContext *votingPeriodContext
+
+	eventsStream  types.EventsStream
+	pricePoster   types.PricePoster
+	priceProvider types.PriceProvider
+}
+
+// Run instantiates a new Feeder instance.
+func Run(stream types.EventsStream, poster types.PricePoster, provider types.PriceProvider, log zerolog.Logger) *Feeder {
+	f := &Feeder{
+		log:                 log,
+		stop:                make(chan struct{}),
+		done:                make(chan struct{}),
+		params:              types.Params{},
+		votingPeriodContext: nil,
+		eventsStream:        stream,
+		pricePoster:         poster,
+		priceProvider:       provider,
+	}
+
+	// init params
+	select {
+	case initParams := <-stream.ParamsUpdate():
+		f.handleParamsUpdate(initParams)
+	case <-time.After(InitTimeout):
+		panic("init timeout deadline exceeded")
+	}
+
+	go f.loop()
+
+	return f
+}
+
+func (f *Feeder) loop() {
+	defer close(f.done)
+	defer f.eventsStream.Close()
+	defer f.pricePoster.Close()
+	defer f.priceProvider.Close()
+	defer f.endLastVotingPeriod()
+	for {
+		select {
+		case <-f.stop:
+			return
+		case params := <-f.eventsStream.ParamsUpdate():
+			f.log.Info().Interface("changes", params).Msg("params changed")
+			f.handleParamsUpdate(params)
+		case vp := <-f.eventsStream.VotingPeriodStarted():
+			f.log.Info().Interface("voting-period", vp).Msg("new voting period")
+			f.handleVotingPeriod(vp)
+		}
+	}
+}
+
+func (f *Feeder) handleParamsUpdate(params types.Params) {
+	f.params = params
+}
+
+func (f *Feeder) handleVotingPeriod(vp types.VotingPeriod) {
+	f.endLastVotingPeriod()
+	f.startNewVotingPeriod(vp)
+}
+
+func (f *Feeder) endLastVotingPeriod() {
+	if f.votingPeriodContext == nil {
+		return
+	}
+
+	// cancel the last voting period send prices operation
+	f.votingPeriodContext.cancelSendPrices()
+	// check if it prices sends went fine
+	select {
+	case <-f.votingPeriodContext.sendPricesDone:
+	default:
+		// TODO(testinginprod): from coverage we can see this was called. Maybe in the future
+		// we can add something extra to assert this was hit in testing.
+		f.log.Err(fmt.Errorf("vote period missed")).
+			Uint64("voting-period-start-height", f.votingPeriodContext.height).
+			Uint64("voting-period-block-duration", f.votingPeriodContext.votingPeriodDurationBlocks)
+	}
+	// reset voting period context
+	f.votingPeriodContext = nil
+}
+
+func (f *Feeder) startNewVotingPeriod(vp types.VotingPeriod) {
+	// gather prices
+	prices := make([]types.Price, len(f.params.Pairs))
+	for i, p := range f.params.Pairs {
+		price := f.priceProvider.GetPrice(p)
+		if !price.Valid {
+			f.log.Err(fmt.Errorf("no valid price")).Str("asset", p.String()).Str("source", price.Source)
+			price.Price = 0
+		}
+		prices[i] = price
+	}
+
+	// send prices asynchronously and non-blocking
+	ctx, cancel := context.WithCancel(context.Background())
+	asyncSendPriceResponse := f.pricePoster.SendPrices(ctx, prices)
+
+	f.votingPeriodContext = &votingPeriodContext{
+		height:                     vp.Height,
+		votingPeriodDurationBlocks: f.params.VotePeriodBlocks,
+		cancelSendPrices:           cancel,
+		pricePosterCtx:             ctx,
+		sendPricesDone:             asyncSendPriceResponse,
+	}
+}
+
+func (f *Feeder) Close() {
+	close(f.stop)
+	<-f.done
+}

feeder/feeder_test.go

@@ -0,0 +1,113 @@
+package feeder
+
+import (
+	"io"
+	"testing"
+	"time"
+
+	"github.com/golang/mock/gomock"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/require"
+
+	mock_feeder "github.com/NibiruChain/nibiru/feeder/mocks/feeder/types"
+	"github.com/NibiruChain/nibiru/feeder/types"
+	"github.com/NibiruChain/nibiru/x/common"
+)
+
+func TestRun(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	t.Run("events stream params timeout", func(t *testing.T) {
+		ps := mock_feeder.NewMockPricePoster(ctrl)
+		pp := mock_feeder.NewMockPriceProvider(ctrl)
+
+		es := mock_feeder.NewMockEventsStream(ctrl)
+
+		es.EXPECT().ParamsUpdate().
+			Return(make(chan types.Params))
+
+		require.Panics(t, func() {
+			Run(es, ps, pp, zerolog.New(io.Discard))
+		})
+	})
+}
+
+func TestParamsUpdate(t *testing.T) {
+	tf := initFeeder(t)
+	defer tf.close()
+	p := types.Params{
+		Pairs:            []common.AssetPair{common.Pair_NIBI_NUSD},
+		VotePeriodBlocks: 50,
+	}
+
+	tf.paramsUpdate <- p
+	time.Sleep(10 * time.Millisecond)
+	require.Equal(t, tf.f.params, p)
+}
+
+func TestVotingPeriod(t *testing.T) {
+	tf := initFeeder(t)
+	defer tf.close()
+
+	validPrice := types.Price{
+		Pair:   common.Pair_BTC_NUSD,
+		Price:  100_000.8,
+		Source: "mock-source",
+		Valid:  true,
+	}
+
+	invalidPrice := types.Price{
+		Pair:   common.Pair_ETH_NUSD,
+		Price:  7000.11,
+		Source: "mock-source",
+		Valid:  false,
+	}
+
+	abstainPrice := invalidPrice
+	abstainPrice.Price = 0.0
+
+	tf.mockPriceProvider.EXPECT().GetPrice(common.Pair_BTC_NUSD).Return(validPrice)
+	tf.mockPriceProvider.EXPECT().GetPrice(common.Pair_ETH_NUSD).Return(invalidPrice)
+	tf.mockPricePoster.EXPECT().SendPrices(gomock.Any(), []types.Price{validPrice, abstainPrice})
+	// trigger voting period.
+	tf.newVotingPeriod <- types.VotingPeriod{Height: 100}
+	time.Sleep(10 * time.Millisecond)
+}
+
+type testFeeder struct {
+	f                 *Feeder
+	mockPriceProvider *mock_feeder.MockPriceProvider
+	mockEventsStream  *mock_feeder.MockEventsStream
+	mockPricePoster   *mock_feeder.MockPricePoster
+	newVotingPeriod   chan types.VotingPeriod
+	paramsUpdate      chan types.Params
+	close             func()
+}
+
+func initFeeder(t *testing.T) testFeeder {
+	ctrl := gomock.NewController(t)
+	ps := mock_feeder.NewMockPricePoster(ctrl)
+	pp := mock_feeder.NewMockPriceProvider(ctrl)
+	es := mock_feeder.NewMockEventsStream(ctrl)
+	params := make(chan types.Params, 1)
+	es.EXPECT().ParamsUpdate().AnyTimes().Return(params)
+	nvp := make(chan types.VotingPeriod, 1)
+	es.EXPECT().VotingPeriodStarted().AnyTimes().Return(nvp)
+
+	params <- types.Params{Pairs: []common.AssetPair{common.Pair_BTC_NUSD, common.Pair_ETH_NUSD}}
+	f := Run(es, ps, pp, zerolog.New(io.Discard))
+	es.EXPECT().Close()
+	pp.EXPECT().Close()
+	ps.EXPECT().Close()
+
+	return testFeeder{
+		f:                 f,
+		mockPriceProvider: pp,
+		mockEventsStream:  es,
+		mockPricePoster:   ps,
+		newVotingPeriod:   nvp,
+		paramsUpdate:      params,
+		close: func() {
+			f.Close()
+		},
+	}
+}

feeder/go.mod

@@ -0,0 +1,99 @@
+module github.com/NibiruChain/nibiru/feeder
+
+go 1.18
+
+// latest grpc doesn't work with with our modified proto compiler, so we need to enforce
+// the following version across all dependencies.
+replace google.golang.org/grpc => google.golang.org/grpc v1.33.2
+
+replace github.com/gogo/protobuf => github.com/regen-network/protobuf v1.3.3-alpha.regen.1
+
+// Fix upstream GHSA-h395-qcrw-5vmq vulnerability.
+// TODO Remove it: https://github.com/cosmos/cosmos-sdk/issues/10409
+replace github.com/gin-gonic/gin => github.com/gin-gonic/gin v1.7.0
+
+replace github.com/confio/ics23/go => github.com/cosmos/cosmos-sdk/ics23/go v0.8.0
+
+replace github.com/NibiruChain/nibiru => ../
+
+require (
+	github.com/NibiruChain/nibiru v0.0.0-00010101000000-000000000000
+	github.com/cosmos/cosmos-sdk v0.45.9
+	github.com/golang/mock v1.6.0
+	github.com/rs/zerolog v1.28.0
+	github.com/stretchr/testify v1.8.0
+)
+
+require (
+	filippo.io/edwards25519 v1.0.0-beta.2 // indirect
+	github.com/armon/go-metrics v0.3.10 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/btcsuite/btcd v0.22.1 // indirect
+	github.com/cespare/xxhash v1.1.0 // indirect
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/confio/ics23/go v0.7.0 // indirect
+	github.com/cosmos/btcutil v1.0.4 // indirect
+	github.com/cosmos/gorocksdb v1.2.0 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/dgraph-io/badger/v2 v2.2007.4 // indirect
+	github.com/dgraph-io/ristretto v0.1.0 // indirect
+	github.com/dgryski/go-farm v0.0.0-20200201041132-a6ae2369ad13 // indirect
+	github.com/dustin/go-humanize v1.0.1-0.20200219035652-afde56e7acac // indirect
+	github.com/fsnotify/fsnotify v1.5.4 // indirect
+	github.com/go-kit/kit v0.12.0 // indirect
+	github.com/go-kit/log v0.2.1 // indirect
+	github.com/go-logfmt/logfmt v0.5.1 // indirect
+	github.com/gogo/protobuf v1.3.3 // indirect
+	github.com/golang/glog v1.0.0 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
+	github.com/google/btree v1.0.1 // indirect
+	github.com/gtank/merlin v0.1.1 // indirect
+	github.com/hashicorp/go-immutable-radix v1.3.1 // indirect
+	github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d // indirect
+	github.com/hashicorp/hcl v1.0.0 // indirect
+	github.com/hdevalence/ed25519consensus v0.0.0-20210204194344-59a8610d2b87 // indirect
+	github.com/inconshreveable/mousetrap v1.0.1 // indirect
+	github.com/jmhodges/levigo v1.0.0 // indirect
+	github.com/klauspost/compress v1.15.9 // indirect
+	github.com/libp2p/go-buffer-pool v0.1.0 // indirect
+	github.com/magiconair/properties v1.8.6 // indirect
+	github.com/mattn/go-colorable v0.1.12 // indirect
+	github.com/mattn/go-isatty v0.0.16 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
+	github.com/mimoo/StrobeGo v0.0.0-20181016162300-f8f6d4d2b643 // indirect
+	github.com/mitchellh/mapstructure v1.5.0 // indirect
+	github.com/pelletier/go-toml v1.9.5 // indirect
+	github.com/pelletier/go-toml/v2 v2.0.5 // indirect
+	github.com/petermattis/goid v0.0.0-20180202154549-b0b1615b78e5 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_golang v1.12.2 // indirect
+	github.com/prometheus/client_model v0.2.0 // indirect
+	github.com/prometheus/common v0.34.0 // indirect
+	github.com/prometheus/procfs v0.7.3 // indirect
+	github.com/sasha-s/go-deadlock v0.2.1-0.20190427202633-1595213edefa // indirect
+	github.com/spf13/afero v1.8.2 // indirect
+	github.com/spf13/cast v1.5.0 // indirect
+	github.com/spf13/cobra v1.6.0 // indirect
+	github.com/spf13/jwalterweatherman v1.1.0 // indirect
+	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/spf13/viper v1.13.0 // indirect
+	github.com/subosito/gotenv v1.4.1 // indirect
+	github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7 // indirect
+	github.com/tendermint/go-amino v0.16.0 // indirect
+	github.com/tendermint/tendermint v0.34.21 // indirect
+	github.com/tendermint/tm-db v0.6.7 // indirect
+	go.etcd.io/bbolt v1.3.6 // indirect
+	golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e // indirect
+	golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e // indirect
+	golang.org/x/net v0.0.0-20220726230323-06994584191e // indirect
+	golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab // indirect
+	golang.org/x/text v0.3.7 // indirect
+	google.golang.org/genproto v0.0.0-20220815135757-37a418bb8959 // indirect
+	google.golang.org/grpc v1.49.0 // indirect
+	google.golang.org/protobuf v1.28.1 // indirect
+	gopkg.in/ini.v1 v1.67.0 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)