Support for parallel processing

relayer/build_processors.go

@@ -3,6 +3,7 @@ package relayer
 import (
 	"context"
 	"math/big"
+	"sync"
 
 	"github.com/VolumeFi/whoops"
 	"github.com/ethereum/go-ethereum/common"
@@ -12,28 +13,60 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-func (r *Relayer) buildProcessors(ctx context.Context) ([]chain.Processor, error) {
-	chainsInfos, err := r.palomaClient.QueryGetEVMChainInfos(ctx)
+func (r *Relayer) buildProcessors(ctx context.Context, locker sync.Locker) error {
+	locker.Lock()
+	defer locker.Unlock()
+	queriedChainsInfos, err := r.palomaClient.QueryGetEVMChainInfos(ctx)
 	if err != nil {
-		return nil, err
+		return err
 	}
-	log.WithField("chains-infos", chainsInfos).Trace("got chain infos")
+	log.WithField("chains-infos", queriedChainsInfos).Trace("got chain infos")
 
-	processors := []chain.Processor{}
-	for _, chainInfo := range chainsInfos {
+	// See if we need to update
+	if (r.processors != nil) && (r.chainsInfos != nil) && (len(r.chainsInfos) == len(queriedChainsInfos)) {
+		chainsChanged := false
+		for k, c := range r.chainsInfos {
+			if c.Id != queriedChainsInfos[k].Id ||
+				c.ChainReferenceID != queriedChainsInfos[k].ChainReferenceID ||
+				c.ChainID != queriedChainsInfos[k].ChainID ||
+				string(c.SmartContractUniqueID) != string(queriedChainsInfos[k].SmartContractUniqueID) ||
+				c.SmartContractAddr != queriedChainsInfos[k].SmartContractAddr ||
+				c.ReferenceBlockHeight != queriedChainsInfos[k].ReferenceBlockHeight ||
+				c.ReferenceBlockHash != queriedChainsInfos[k].ReferenceBlockHash ||
+				c.Abi != queriedChainsInfos[k].Abi ||
+				string(c.Bytecode) != string(queriedChainsInfos[k].Bytecode) ||
+				string(c.ConstructorInput) != string(queriedChainsInfos[k].ConstructorInput) ||
+				c.Status != queriedChainsInfos[k].Status ||
+				c.ActiveSmartContractID != queriedChainsInfos[k].ActiveSmartContractID ||
+				c.MinOnChainBalance != queriedChainsInfos[k].MinOnChainBalance {
+				chainsChanged = true
+			}
+		}
+		if !chainsChanged {
+			log.Debug("chain infos unchanged since last tick")
+			return nil
+		}
+	}
+
+	log.Debug("chain infos changed.  building processors")
+
+	r.processors = []chain.Processor{}
+	r.chainsInfos = []evmtypes.ChainInfo{}
+	for _, chainInfo := range queriedChainsInfos {
 		processor, err := r.processorFactory(chainInfo)
 		if errors.IsUnrecoverable(err) {
-			return nil, err
+			return err
 		}
 
 		if err := processor.IsRightChain(ctx); err != nil {
-			return nil, err
+			return err
 		}
 
-		processors = append(processors, processor)
+		r.processors = append(r.processors, processor)
+		r.chainsInfos = append(r.chainsInfos, *chainInfo)
 	}
 
-	return processors, nil
+	return nil
 }
 
 func (r *Relayer) processorFactory(chainInfo *evmtypes.ChainInfo) (chain.Processor, error) {

relayer/build_processors_test.go

@@ -0,0 +1,311 @@
+package relayer
+
+import (
+	"context"
+	"testing"
+
+	"github.com/palomachain/paloma/x/evm/types"
+	"github.com/palomachain/pigeon/chain"
+	chainmocks "github.com/palomachain/pigeon/chain/mocks"
+	"github.com/palomachain/pigeon/config"
+	"github.com/palomachain/pigeon/relayer/mocks"
+	"github.com/palomachain/pigeon/testutil"
+	timemocks "github.com/palomachain/pigeon/util/time/mocks"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+func TestBuildProcessors(t *testing.T) {
+	testcases := []struct {
+		name        string
+		setup       func(t *testing.T) (*Relayer, []chain.Processor, []types.ChainInfo)
+		expectedErr error
+	}{
+		{
+			name: "when there are no processors on relayer yet it builds processors",
+			setup: func(t *testing.T) (*Relayer, []chain.Processor, []types.ChainInfo) {
+				chain1Info := types.ChainInfo{
+					Id:                1,
+					ChainReferenceID:  "chain-1",
+					MinOnChainBalance: "5",
+				}
+				pc := mocks.NewPalomaClienter(t)
+				pc.On(
+					"QueryGetEVMChainInfos",
+					mock.Anything,
+					mock.Anything,
+				).Return(
+					[]*types.ChainInfo{
+						&chain1Info,
+					},
+					nil,
+				)
+
+				processorMock := chainmocks.NewProcessor(t)
+				processorMock.On("IsRightChain", mock.Anything).Return(nil)
+
+				evmFactoryMock := mocks.NewEvmFactorier(t)
+				evmFactoryMock.On("Build", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(processorMock, nil)
+
+				r := New(
+					config.Root{
+						EVM: map[string]config.EVM{
+							"chain-1": {},
+						},
+					},
+					pc,
+					evmFactoryMock,
+					timemocks.NewTime(t),
+					Config{},
+				)
+				r.chainsInfos = []types.ChainInfo{
+					chain1Info,
+				}
+
+				return r,
+					[]chain.Processor{
+						processorMock,
+					},
+					[]types.ChainInfo{
+						chain1Info,
+					}
+			},
+		},
+		{
+			name: "when there are no chainsInfos on relayer yet it builds processors",
+			setup: func(t *testing.T) (*Relayer, []chain.Processor, []types.ChainInfo) {
+				chain1Info := types.ChainInfo{
+					Id:                1,
+					ChainReferenceID:  "chain-1",
+					MinOnChainBalance: "5",
+				}
+				pc := mocks.NewPalomaClienter(t)
+				pc.On(
+					"QueryGetEVMChainInfos",
+					mock.Anything,
+					mock.Anything,
+				).Return(
+					[]*types.ChainInfo{
+						&chain1Info,
+					},
+					nil,
+				)
+
+				processorMock := chainmocks.NewProcessor(t)
+				processorMock.On("IsRightChain", mock.Anything).Return(nil)
+
+				evmFactoryMock := mocks.NewEvmFactorier(t)
+				evmFactoryMock.On("Build", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(processorMock, nil)
+
+				r := New(
+					config.Root{
+						EVM: map[string]config.EVM{
+							"chain-1": {},
+						},
+					},
+					pc,
+					evmFactoryMock,
+					timemocks.NewTime(t),
+					Config{},
+				)
+				r.processors = []chain.Processor{
+					chainmocks.NewProcessor(t),
+				}
+
+				return r,
+					[]chain.Processor{
+						processorMock,
+					},
+					[]types.ChainInfo{
+						chain1Info,
+					}
+			},
+		},
+		{
+			name: "when the chains lengths are different it builds processors",
+			setup: func(t *testing.T) (*Relayer, []chain.Processor, []types.ChainInfo) {
+				chain1Info := types.ChainInfo{
+					Id:                1,
+					ChainReferenceID:  "chain-1",
+					MinOnChainBalance: "5",
+				}
+				chain2Info := types.ChainInfo{
+					Id:                2,
+					ChainReferenceID:  "chain-2",
+					MinOnChainBalance: "5",
+				}
+
+				pc := mocks.NewPalomaClienter(t)
+				pc.On(
+					"QueryGetEVMChainInfos",
+					mock.Anything,
+					mock.Anything,
+				).Return(
+					[]*types.ChainInfo{
+						&chain1Info,
+					},
+					nil,
+				)
+
+				processorMock := chainmocks.NewProcessor(t)
+				processorMock.On("IsRightChain", mock.Anything).Return(nil)
+
+				evmFactoryMock := mocks.NewEvmFactorier(t)
+				evmFactoryMock.On("Build", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(processorMock, nil)
+
+				r := New(
+					config.Root{
+						EVM: map[string]config.EVM{
+							"chain-1": {},
+						},
+					},
+					pc,
+					evmFactoryMock,
+					timemocks.NewTime(t),
+					Config{},
+				)
+				r.processors = []chain.Processor{
+					chainmocks.NewProcessor(t),
+				}
+
+				r.chainsInfos = []types.ChainInfo{
+					chain1Info,
+					chain2Info,
+				}
+
+				return r,
+					[]chain.Processor{
+						processorMock,
+					},
+					[]types.ChainInfo{
+						chain1Info,
+					}
+			},
+		},
+		{
+			name: "when there is a difference in the chain data it builds processors",
+			setup: func(t *testing.T) (*Relayer, []chain.Processor, []types.ChainInfo) {
+				chain1Info := types.ChainInfo{
+					Id:                1,
+					ChainReferenceID:  "chain-1",
+					MinOnChainBalance: "5",
+				}
+
+				chain1NewInfo := types.ChainInfo{
+					Id:                1,
+					ChainReferenceID:  "chain-1",
+					MinOnChainBalance: "50",
+				}
+				pc := mocks.NewPalomaClienter(t)
+				pc.On(
+					"QueryGetEVMChainInfos",
+					mock.Anything,
+					mock.Anything,
+				).Return(
+					[]*types.ChainInfo{
+						&chain1NewInfo,
+					},
+					nil,
+				)
+
+				processorMock := chainmocks.NewProcessor(t)
+				processorMock.On("IsRightChain", mock.Anything).Return(nil)
+
+				evmFactoryMock := mocks.NewEvmFactorier(t)
+				evmFactoryMock.On("Build", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(processorMock, nil)
+
+				r := New(
+					config.Root{
+						EVM: map[string]config.EVM{
+							"chain-1": {},
+						},
+					},
+					pc,
+					evmFactoryMock,
+					timemocks.NewTime(t),
+					Config{},
+				)
+				r.processors = []chain.Processor{
+					chainmocks.NewProcessor(t),
+				}
+
+				r.chainsInfos = []types.ChainInfo{
+					chain1Info,
+				}
+
+				return r,
+					[]chain.Processor{
+						processorMock,
+					},
+					[]types.ChainInfo{
+						chain1NewInfo,
+					}
+			},
+		},
+		{
+			name: "when the chains are the same it doesn't build processors",
+			setup: func(t *testing.T) (*Relayer, []chain.Processor, []types.ChainInfo) {
+				chain1Info := types.ChainInfo{
+					Id:                1,
+					ChainReferenceID:  "chain-1",
+					MinOnChainBalance: "5",
+				}
+
+				pc := mocks.NewPalomaClienter(t)
+				pc.On(
+					"QueryGetEVMChainInfos",
+					mock.Anything,
+					mock.Anything,
+				).Return(
+					[]*types.ChainInfo{
+						&chain1Info,
+					},
+					nil,
+				)
+
+				r := New(
+					config.Root{
+						EVM: map[string]config.EVM{
+							"chain-1": {},
+						},
+					},
+					pc,
+					mocks.NewEvmFactorier(t),
+					timemocks.NewTime(t),
+					Config{},
+				)
+
+				origProcessor := chainmocks.NewProcessor(t)
+				r.processors = []chain.Processor{
+					origProcessor,
+				}
+
+				r.chainsInfos = []types.ChainInfo{
+					chain1Info,
+				}
+
+				return r,
+					[]chain.Processor{
+						origProcessor,
+					},
+					[]types.ChainInfo{
+						chain1Info,
+					}
+			},
+		},
+	}
+
+	asserter := assert.New(t)
+	ctx := context.Background()
+	for _, tt := range testcases {
+		t.Run(tt.name, func(t *testing.T) {
+			relayer, expectedProcessors, expectedChainsInfos := tt.setup(t)
+			var locker testutil.FakeMutex
+
+			actualErr := relayer.buildProcessors(ctx, locker)
+			asserter.Equal(tt.expectedErr, actualErr)
+			asserter.Equal(expectedProcessors, relayer.processors)
+			asserter.Equal(expectedChainsInfos, relayer.chainsInfos)
+		})
+	}
+}