Repository: cosmos/relayer
Pull request #813 (2022-06-27 18:43:33)

EventProcessor architecture - wire up with flag
Add `--events`/`-e` flag for relaying using the event processor architecture. Initial block history to query can be configured using `--block-history`/`-h` flag.

Enables event processor architecture relaying in the chain tests.


Patch:
diff --git a/cmd/flags.go b/cmd/flags.go
index 927efa6..5d20cc6 100644
--- a/cmd/flags.go
+++ b/cmd/flags.go
@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/cosmos/relayer/v2/relayer"
+
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 )
@@ -40,6 +42,8 @@ const (
 	flagPageKey                 = "page-key"
 	flagCountTotal              = "count-total"
 	flagReverse                 = "reverse"
+	flagProcessor               = "processor"
+	flagInitialBlockHistory     = "block-history"
 )
 
 const (
@@ -287,6 +291,18 @@ func debugServerFlags(v *viper.Viper, cmd *cobra.Command) *cobra.Command {
 	return cmd
 }
 
+func processorFlags(v *viper.Viper, cmd *cobra.Command) *cobra.Command {
+	cmd.Flags().StringP(flagProcessor, "p", relayer.ProcessorLegacy, "which relayer processor to use")
+	if err := v.BindPFlag(flagProcessor, cmd.Flags().Lookup(flagProcessor)); err != nil {
+		panic(err)
+	}
+	cmd.Flags().Uint64P(flagInitialBlockHistory, "h", 20, "initial block history to query when using 'events' as the processor for relaying")
+	if err := v.BindPFlag(flagInitialBlockHistory, cmd.Flags().Lookup(flagInitialBlockHistory)); err != nil {
+		panic(err)
+	}
+	return cmd
+}
+
 func OverwriteConfigFlag(v *viper.Viper, cmd *cobra.Command) *cobra.Command {
 	cmd.Flags().BoolP(flagOverwriteConfig, "o", false,
 		"overwrite already configured paths - will clear channel filter(s)")
diff --git a/cmd/start.go b/cmd/start.go
index 3f4a5e8..2f690ed 100644
--- a/cmd/start.go
+++ b/cmd/start.go
@@ -43,8 +43,9 @@ func startCmd(a *appState) *cobra.Command {
 		Short:   "Start the listening relayer on a given path",
 		Args:    withUsage(cobra.ExactArgs(1)),
 		Example: strings.TrimSpace(fmt.Sprintf(`
+$ %s start demo-path -p events # to use event processor
 $ %s start demo-path --max-msgs 3
-$ %s start demo-path2 --max-tx-size 10`, appName, appName)),
+$ %s start demo-path2 --max-tx-size 10`, appName, appName, appName)),
 		RunE: func(cmd *cobra.Command, args []string) error {
 			c, src, dst, err := a.Config.ChainsFromPath(args[0])
 			if err != nil {
@@ -81,7 +82,16 @@ $ %s start demo-path2 --max-tx-size 10`, appName, appName)),
 				relaydebug.StartDebugServer(cmd.Context(), log, ln)
 			}
 
-			rlyErrCh := relayer.StartRelayer(cmd.Context(), a.Log, c[src], c[dst], filter, maxTxSize, maxMsgLength)
+			processorType, err := cmd.Flags().GetString(flagProcessor)
+			if err != nil {
+				return err
+			}
+			initialBlockHistory, err := cmd.Flags().GetUint64(flagInitialBlockHistory)
+			if err != nil {
+				return err
+			}
+
+			rlyErrCh := relayer.StartRelayer(cmd.Context(), a.Log, c[src], c[dst], cmd.InOrStdin(), cmd.OutOrStdout(), filter, maxTxSize, maxMsgLength, processorType, initialBlockHistory)
 
 			// NOTE: This block of code is useful for ensuring that the clients tracking each chain do not expire
 			// when there are no packets flowing across the channels. It is currently a source of errors that have been
@@ -138,7 +148,11 @@ $ %s start demo-path2 --max-tx-size 10`, appName, appName)),
 			return nil
 		},
 	}
-	return debugServerFlags(a.Viper, strategyFlag(a.Viper, updateTimeFlags(a.Viper, cmd)))
+	cmd = updateTimeFlags(a.Viper, cmd)
+	cmd = strategyFlag(a.Viper, cmd)
+	cmd = debugServerFlags(a.Viper, cmd)
+	cmd = processorFlags(a.Viper, cmd)
+	return cmd
 }
 
 // UpdateClientsFromChains takes src, dst chains, threshold time and update clients based on expiry time
diff --git a/relayer/chains/cosmos/cosmos_chain_processor.go b/relayer/chains/cosmos/cosmos_chain_processor.go
index 0e542c4..d6487e1 100644
--- a/relayer/chains/cosmos/cosmos_chain_processor.go
+++ b/relayer/chains/cosmos/cosmos_chain_processor.go
@@ -51,7 +51,7 @@ type CosmosChainProcessor struct {
 	channelStateCache processor.ChannelStateCache
 }
 
-func NewCosmosChainProcessor(log *zap.Logger, provider *cosmos.CosmosProvider, rpcAddress string, input io.Reader, output io.Writer, pathProcessors processor.PathProcessors) (*CosmosChainProcessor, error) {
+func NewCosmosChainProcessor(log *zap.Logger, provider *cosmos.CosmosProvider, rpcAddress string, input io.Reader, output io.Writer) (*CosmosChainProcessor, error) {
 	chainID := provider.ChainId()
 	cc, err := getCosmosClient(rpcAddress, chainID, input, output)
 	if err != nil {
@@ -66,7 +66,6 @@ func NewCosmosChainProcessor(log *zap.Logger, provider *cosmos.CosmosProvider, r
 		chainProvider:        provider,
 		cc:                   cc,
 		lightProvider:        lightProvider,
-		pathProcessors:       pathProcessors,
 		latestClientState:    make(latestClientState),
 		connectionStateCache: make(processor.ConnectionStateCache),
 		channelStateCache:    make(processor.ChannelStateCache),
diff --git a/relayer/processor/event_processor.go b/relayer/processor/event_processor.go
index 013b8bd..cc52f33 100644
--- a/relayer/processor/event_processor.go
+++ b/relayer/processor/event_processor.go
@@ -25,25 +25,34 @@ func NewEventProcessor() EventProcessorBuilder {
 	return EventProcessorBuilder{}
 }
 
-// WithChainProcessors will set the ChainProcessors to be used.
+// WithChainProcessors adds to the list of ChainProcessors to be used
+// if the chain name has not already been added.
 func (ep EventProcessorBuilder) WithChainProcessors(chainProcessors ...ChainProcessor) EventProcessorBuilder {
-	ep.chainProcessors = chainProcessors
+ChainProcessorLoop:
+	for _, cp := range chainProcessors {
+		for _, existingCp := range ep.chainProcessors {
+			if existingCp.Provider().ChainName() == cp.Provider().ChainName() {
+				continue ChainProcessorLoop
+			}
+		}
+		ep.chainProcessors = append(ep.chainProcessors, cp)
+	}
 	return ep
 }
 
-// WithInitialBlockHistory will set the initial block history to query for the ChainProcessors.
+// WithInitialBlockHistory sets the initial block history to query for the ChainProcessors.
 func (ep EventProcessorBuilder) WithInitialBlockHistory(initialBlockHistory uint64) EventProcessorBuilder {
 	ep.initialBlockHistory = initialBlockHistory
 	return ep
 }
 
-// WithPathProcessors will set the PathProcessors to be used.
+// WithPathProcessors adds to the list of PathProcessors to be used.
 func (ep EventProcessorBuilder) WithPathProcessors(pathProcessors ...*PathProcessor) EventProcessorBuilder {
-	ep.pathProcessors = pathProcessors
+	ep.pathProcessors = append(ep.pathProcessors, pathProcessors...)
 	return ep
 }
 
-// Build will link the relevant ChainProcessors and PathProcessors, then return an EventProcessor that can be used to run the ChainProcessors and PathProcessors.
+// Build links the relevant ChainProcessors and PathProcessors, then returns an EventProcessor that can be used to run the ChainProcessors and PathProcessors.
 func (ep EventProcessorBuilder) Build() EventProcessor {
 	for _, chainProcessor := range ep.chainProcessors {
 		pathProcessorsForThisChain := PathProcessors{}
diff --git a/relayer/strategies.go b/relayer/strategies.go
index a26c2e6..0f09614 100644
--- a/relayer/strategies.go
+++ b/relayer/strategies.go
@@ -4,12 +4,17 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"strings"
 	"sync"
 	"time"
 
 	"github.com/avast/retry-go/v4"
 	"github.com/cosmos/ibc-go/v3/modules/core/04-channel/types"
+	cosmosprocessor "github.com/cosmos/relayer/v2/relayer/chains/cosmos"
+	"github.com/cosmos/relayer/v2/relayer/processor"
+	"github.com/cosmos/relayer/v2/relayer/provider"
+	cosmosprovider "github.com/cosmos/relayer/v2/relayer/provider/cosmos"
 	"go.uber.org/zap"
 )
 
@@ -19,12 +24,100 @@ type ActiveChannel struct {
 	active  bool
 }
 
+const (
+	ProcessorEvents string = "events"
+	ProcessorLegacy        = "legacy"
+)
+
 // StartRelayer starts the main relaying loop and returns a channel that will contain any control-flow related errors.
-func StartRelayer(ctx context.Context, log *zap.Logger, src, dst *Chain, filter ChannelFilter, maxTxSize, maxMsgLength uint64) chan error {
+func StartRelayer(ctx context.Context, log *zap.Logger, src, dst *Chain, stdin io.Reader, stdout io.Writer, filter ChannelFilter, maxTxSize, maxMsgLength uint64, processorType string, initialBlockHistory uint64) chan error {
 	errorChan := make(chan error, 1)
 
-	go relayerMainLoop(ctx, log, src, dst, filter, maxTxSize, maxMsgLength, errorChan)
-	return errorChan
+	switch processorType {
+	case ProcessorEvents:
+		var filterSrc, filterDst []processor.ChannelKey
+
+		for _, ch := range filter.ChannelList {
+			ruleSrc := processor.ChannelKey{ChannelID: ch}
+			ruleDst := processor.ChannelKey{CounterpartyChannelID: ch}
+			filterSrc = append(filterSrc, ruleSrc)
+			filterDst = append(filterDst, ruleDst)
+		}
+		paths := []path{{
+			src: pathChain{
+				provider:   src.ChainProvider,
+				pathEnd:    processor.NewPathEnd(src.ChainProvider.ChainId(), src.ClientID(), filter.Rule, filterSrc),
+				rpcAddress: src.RPCAddr,
+			},
+			dst: pathChain{
+				provider:   dst.ChainProvider,
+				pathEnd:    processor.NewPathEnd(dst.ChainProvider.ChainId(), dst.ClientID(), filter.Rule, filterDst),
+				rpcAddress: dst.RPCAddr,
+			},
+		}}
+
+		go relayerStartEventProcessor(ctx, log, stdin, stdout, paths, initialBlockHistory, maxTxSize, maxMsgLength, errorChan)
+		return errorChan
+	case ProcessorLegacy:
+		go relayerMainLoop(ctx, log, src, dst, filter, maxTxSize, maxMsgLength, errorChan)
+		return errorChan
+	default:
+		panic(fmt.Errorf("unexpected processor type: %s, supports one of: [%s, %s]", processorType, ProcessorEvents, ProcessorLegacy))
+	}
+}
+
+// TODO: intermediate types. Should combine/replace with the relayer.Chain, relayer.Path, and relayer.PathEnd structs
+// as the stateless and stateful/event-based relaying mechanisms are consolidated.
+type path struct {
+	src pathChain
+	dst pathChain
+}
+
+type pathChain struct {
+	provider   provider.ChainProvider
+	pathEnd    processor.PathEnd
+	rpcAddress string
+}
+
+// chainProcessor returns the corresponding ChainProcessor implementation instance for a pathChain.
+func (chain pathChain) chainProcessor(log *zap.Logger, stdin io.Reader, stdout io.Writer) processor.ChainProcessor {
+	// Handle new ChainProcessor implementations as cases here
+	switch p := chain.provider.(type) {
+	case *cosmosprovider.CosmosProvider:
+		chainProcessor, err := cosmosprocessor.NewCosmosChainProcessor(log, p, chain.rpcAddress, stdin, stdout)
+		if err != nil {
+			panic(err)
+		}
+		return chainProcessor
+	default:
+		panic(fmt.Errorf("unsupported chain provider type: %T", chain.provider))
+	}
+}
+
+// relayerStartEventProcessor is the main relayer process when using the event processor.
+func relayerStartEventProcessor(ctx context.Context, log *zap.Logger, stdin io.Reader, stdout io.Writer, paths []path, initialBlockHistory uint64, maxTxSize, maxMsgLength uint64, errCh chan<- error) {
+	defer close(errCh)
+
+	epb := processor.NewEventProcessor()
+
+	for _, p := range paths {
+		epb = epb.
+			WithChainProcessors(
+				p.src.chainProcessor(log, stdin, stdout),
+				p.dst.chainProcessor(log, stdin, stdout),
+			).
+			WithPathProcessors(processor.NewPathProcessor(
+				log,
+				p.src.pathEnd,
+				p.dst.pathEnd,
+			))
+	}
+
+	ep := epb.
+		WithInitialBlockHistory(initialBlockHistory).
+		Build()
+
+	errCh <- ep.Run(ctx)
 }
 
 // relayerMainLoop is the main loop of the relayer.


Test patch:
diff --git a/_test/relayer_chain_test.go b/_test/relayer_chain_test.go
index ef36f66..6fd0c8d 100644
--- a/_test/relayer_chain_test.go
+++ b/_test/relayer_chain_test.go
@@ -1,8 +1,10 @@
 package test
 
 import (
+	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"testing"
 	"time"
 
@@ -119,7 +121,7 @@ func chainTest(t *testing.T, tcs []testChain) {
 
 	t.Log("Starting relayer")
 	filter := relayer.ChannelFilter{}
-	_ = relayer.StartRelayer(ctx, log, src, dst, filter, 2*cmd.MB, 5)
+	_ = relayer.StartRelayer(ctx, log, src, dst, bytes.NewReader(nil), io.Discard, filter, 2*cmd.MB, 5, relayer.ProcessorEvents, 20)
 
 	t.Log("Waiting for relayer messages to reach both chains")
 	require.NoError(t, src.ChainProvider.WaitForNBlocks(ctx, 2))
@@ -269,7 +271,7 @@ func TestGaiaMisbehaviourMonitoring(t *testing.T) {
 	log := zaptest.NewLogger(t)
 	// start the relayer process in it's own goroutine
 	filter := relayer.ChannelFilter{}
-	_ = relayer.StartRelayer(ctx, log, src, dst, filter, 2*cmd.MB, 5)
+	_ = relayer.StartRelayer(ctx, log, src, dst, bytes.NewReader(nil), io.Discard, filter, 2*cmd.MB, 5, relayer.ProcessorEvents, 20)
 
 	// Wait for relay message inclusion in both chains
 	require.NoError(t, src.ChainProvider.WaitForNBlocks(ctx, 1))
@@ -443,7 +445,7 @@ func TestRelayAllChannelsOnConnection(t *testing.T) {
 
 	t.Log("Starting relayer")
 	filter := relayer.ChannelFilter{}
-	_ = relayer.StartRelayer(ctx, log, src, dst, filter, 2*cmd.MB, 5)
+	_ = relayer.StartRelayer(ctx, log, src, dst, bytes.NewReader(nil), io.Discard, filter, 2*cmd.MB, 5, relayer.ProcessorEvents, 20)
 
 	t.Log("Waiting for relayer message inclusion in both chains")
 	require.NoError(t, src.ChainProvider.WaitForNBlocks(ctx, 1))
@@ -619,7 +621,7 @@ func TestUnorderedChannelBlockHeightTimeout(t *testing.T) {
 
 	// start the relayer process in it's own goroutine
 	filter := relayer.ChannelFilter{}
-	_ = relayer.StartRelayer(ctx, log, src, dst, filter, 2*cmd.MB, 5)
+	_ = relayer.StartRelayer(ctx, log, src, dst, bytes.NewReader(nil), io.Discard, filter, 2*cmd.MB, 5, relayer.ProcessorEvents, 20)
 
 	require.NoError(t, src.ChainProvider.WaitForNBlocks(ctx, 5))
 
@@ -717,7 +719,7 @@ func TestUnorderedChannelTimestampTimeout(t *testing.T) {
 
 	// start the relayer process in it's own goroutine
 	filter := relayer.ChannelFilter{}
-	_ = relayer.StartRelayer(ctx, log, src, dst, filter, 2*cmd.MB, 5)
+	_ = relayer.StartRelayer(ctx, log, src, dst, bytes.NewReader(nil), io.Discard, filter, 2*cmd.MB, 5, relayer.ProcessorEvents, 20)
 
 	time.Sleep(time.Second * 10)
 
diff --git a/relayer/chains/cosmos/msg_handlers_packet_test.go b/relayer/chains/cosmos/msg_handlers_packet_test.go
index 223be42..3ad21d4 100644
--- a/relayer/chains/cosmos/msg_handlers_packet_test.go
+++ b/relayer/chains/cosmos/msg_handlers_packet_test.go
@@ -1,7 +1,8 @@
 package cosmos
 
 import (
-	"os"
+	"bytes"
+	"io"
 	"testing"
 
 	clienttypes "github.com/cosmos/ibc-go/v3/modules/core/02-client/types"
@@ -26,9 +27,11 @@ func mockCosmosChainProcessor(t *testing.T) *CosmosChainProcessor {
 		provider      = cosmos.CosmosProvider{PCfg: cosmos.CosmosProviderConfig{ChainID: chainID1}}
 	)
 
-	ccp, err := NewCosmosChainProcessor(log, &provider, "", os.Stdin, os.Stdout, []*processor.PathProcessor{pathProcessor})
+	ccp, err := NewCosmosChainProcessor(log, &provider, "", bytes.NewReader(nil), io.Discard)
 	require.NoError(t, err, "error constructing cosmos chain processor")
 
+	ccp.SetPathProcessors(processor.PathProcessors{pathProcessor})
+
 	applicable := pathProcessor.SetChainProviderIfApplicable(&provider)
 	require.True(t, applicable, "error setting path processor reference to chain processor")