fix: if chainlink price feed has 2 events in same block, use last

Author: 5lliot

.github/workflows/go.yml

@@ -26,4 +26,4 @@ jobs:
     - run: go get github.com/google/go-jsonnet/cmd/jsonnet && ./generate-test.sh
       if: steps.cache-packages.outputs.cache-hit != 'true'
     - name: Run Test
-      run: go test ./tests && go test ./services && go test ./ds && go test ./models/credit_manager
+      run: go test ./tests && go test ./services && go test ./ds && go test ./models/...

ds/dc_wrapper/mainnet.go

@@ -29,13 +29,13 @@ func NewMainnetDC(client core.ClientI) *MainnetDC {
 	}
 }
 
-func (m *MainnetDC) setAddr(addr common.Address) {
+func (m *MainnetDC) SetAddr(addr common.Address) {
 	dc, err := mainnet.NewDataCompressor(addr, m.client)
 	log.CheckFatal(err)
 	m.dc = dc
 }
 
-func (repo *MainnetDC) addCreditManagerToFilter(cmAddr, cfAddr string) {
+func (repo *MainnetDC) AddCreditManagerToFilter(cmAddr, cfAddr string) {
 	repo.creditManagerToFilter[common.HexToAddress(cmAddr)] = common.HexToAddress(cfAddr)
 }
 

ds/dc_wrapper/wrapper.go

@@ -200,13 +200,13 @@ func (dcw *DataCompressorWrapper) setv1DC(discoveredAt int64) {
 	dcw.mu.Lock()
 	defer dcw.mu.Unlock()
 	addr := dcw.discoveredAtToAddr[discoveredAt]
-	dcw.v1DC.setAddr(addr)
+	dcw.v1DC.SetAddr(addr)
 }
 
 func (dcw *DataCompressorWrapper) AddCreditManagerToFilter(cmAddr, cfAddr string) {
 	dcw.mu.Lock()
 	defer dcw.mu.Unlock()
-	dcw.v1DC.addCreditManagerToFilter(cmAddr, cfAddr)
+	dcw.v1DC.AddCreditManagerToFilter(cmAddr, cfAddr)
 }
 
 func (dcw *DataCompressorWrapper) setv2DC(discoveredAt int64) {

ds/repo_dummy.go

@@ -58,7 +58,7 @@ func (DummyRepo) GetExecuteParser() ExecuteParserI {
 // price feed/oracle funcs
 func (DummyRepo) DirectlyAddTokenOracle(tokenOracle *schemas.TokenOracle) {
 }
-func (DummyRepo) AddPriceFeed(pf *schemas.PriceFeed) {
+func (*DummyRepo) AddPriceFeed(pf *schemas.PriceFeed) {
 }
 
 // token funcs

models/chainlink_price_feed/on_log.go

@@ -17,20 +17,21 @@ func (mdl *ChainlinkPriceFeed) OnLog(txLog types.Log) {
 }
 func (mdl *ChainlinkPriceFeed) OnLogs(txLogs []types.Log) {
 	var priceFeeds []*schemas.PriceFeed
-	var prevBlock int64
-	for _, txLog := range txLogs {
+	for txLogInd, txLog := range txLogs {
 		var priceFeed *schemas.PriceFeed
 		blockNum := int64(txLog.BlockNumber)
 		switch txLog.Topics[0] {
 		case core.Topic("AnswerUpdated(int256,uint256,uint256)"):
-			// there might be 2 AnswerUpdated events for same block
+			// there might be 2 AnswerUpdated events for same block, use the last one
 			// example
 			// https://goerli.etherscan.io/tx/0x03308a0b6f024e6c35a92e7708ab5a72322f733d22427d51624862d82ca1983a
 			// https://goerli.etherscan.io/tx/0x38e5551ae639d22554072ba1a53e026a0858c2cfedcedb83e5cc63bb1c8b8ea8
-			if prevBlock == blockNum {
+			// on mainnet
+			// https://etherscan.io/tx/0xb3aaa84cac23a30ab20cbd254b2297840f23057faf1f05e7655304be6cffc19e#eventlog
+			// https://etherscan.io/tx/0x3112f0a42f288ca56a2c8f8003355ad20e87e1f23c3ffa991633f6bb25eb8c58#eventlog
+			if txLogInd+1 < len(txLogs) && int64(txLogs[txLogInd+1].BlockNumber) == blockNum {
 				continue
 			}
-			prevBlock = blockNum
 			//
 			roundId, err := strconv.ParseInt(txLog.Topics[2].Hex()[50:], 16, 64)
 			if err != nil {

models/chainlink_price_feed/on_logs_test.go

@@ -0,0 +1,69 @@
+package chainlink_price_feed
+
+import (
+	"math/big"
+	"testing"
+
+	"github.com/Gearbox-protocol/sdk-go/core"
+	"github.com/Gearbox-protocol/sdk-go/core/schemas"
+	"github.com/Gearbox-protocol/sdk-go/utils"
+	"github.com/Gearbox-protocol/third-eye/ds"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+)
+
+type OnLogsChecker struct {
+	ds.DummyRepo
+	pfs []*schemas.PriceFeed
+}
+
+func (x *OnLogsChecker) AddPriceFeed(pf *schemas.PriceFeed) {
+	x.pfs = append(x.pfs, pf)
+}
+func TestOnLogs(t *testing.T) {
+	validPf := &schemas.PriceFeed{
+		Feed:        utils.RandomAddr(),
+		Token:       utils.RandomAddr(),
+		BlockNumber: 1,
+		PriceBI:     (*core.BigInt)(big.NewInt(222)),
+		RoundId:     3,
+	}
+	repo := &OnLogsChecker{}
+	obj := &ChainlinkPriceFeed{SyncAdapter: &ds.SyncAdapter{
+		Repo: repo,
+		SyncAdapterSchema: &schemas.SyncAdapterSchema{
+			Contract: &schemas.Contract{
+				Address: validPf.Feed,
+			},
+			Details: core.Json{"token": validPf.Token},
+		},
+	}}
+	txLogs := []types.Log{
+		{
+			BlockNumber: 1,
+			Index:       1,
+			Topics: []common.Hash{
+				core.Topic("AnswerUpdated(int256,uint256,uint256)"),
+				common.BytesToHash([]byte{1}),
+				common.BytesToHash([]byte{1}),
+			},
+		},
+		{
+			BlockNumber: uint64(validPf.BlockNumber),
+			Index:       2,
+			Topics: []common.Hash{
+				core.Topic("AnswerUpdated(int256,uint256,uint256)"),
+				common.BytesToHash(validPf.PriceBI.Convert().Bytes()),
+				common.BytesToHash([]byte{byte(validPf.RoundId)}),
+			},
+		},
+	}
+	obj.OnLogs(txLogs)
+	if len(repo.pfs) != 1 ||
+		repo.pfs[0].BlockNumber != validPf.BlockNumber ||
+		repo.pfs[0].Feed != validPf.Feed ||
+		repo.pfs[0].RoundId != validPf.RoundId ||
+		repo.pfs[0].PriceBI.Cmp(validPf.PriceBI) != 0 {
+		t.Fatal(utils.ToJson(repo.pfs))
+	}
+}