Repository: cosmos/relayer
Pull request #819 (2022-06-29 05:22:10)

Use provider for all CosmosChainProcessor queries
All `CosmosChainProcessor` queries now go through the provider:
- Removes the cosmos client.Context and the tendermint light provider from the CosmosChainProcessor. 
- `CosmosChainProcessor` no longer needs to know RPC address, only the provider needs to know.
- `sdk_helpers.go` removed from the CosmosChainProcessor
- minimizes method signatures and returns


Patch:
diff --git a/cmd/flags.go b/cmd/flags.go
index 5d20cc6..9551aab 100644
--- a/cmd/flags.go
+++ b/cmd/flags.go
@@ -296,7 +296,7 @@ func processorFlags(v *viper.Viper, cmd *cobra.Command) *cobra.Command {
 	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")
+	cmd.Flags().Uint64P(flagInitialBlockHistory, "b", 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)
 	}
diff --git a/cmd/start.go b/cmd/start.go
index 2f690ed..c56a46a 100644
--- a/cmd/start.go
+++ b/cmd/start.go
@@ -91,7 +91,7 @@ $ %s start demo-path2 --max-tx-size 10`, appName, appName, appName)),
 				return err
 			}
 
-			rlyErrCh := relayer.StartRelayer(cmd.Context(), a.Log, c[src], c[dst], cmd.InOrStdin(), cmd.OutOrStdout(), filter, maxTxSize, maxMsgLength, processorType, initialBlockHistory)
+			rlyErrCh := relayer.StartRelayer(cmd.Context(), a.Log, c[src], c[dst], 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
diff --git a/relayer/chains/cosmos/cosmos_chain_processor.go b/relayer/chains/cosmos/cosmos_chain_processor.go
index d6487e1..986e17a 100644
--- a/relayer/chains/cosmos/cosmos_chain_processor.go
+++ b/relayer/chains/cosmos/cosmos_chain_processor.go
@@ -4,20 +4,15 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"time"
 
 	"github.com/avast/retry-go/v4"
-	cosmosClient "github.com/cosmos/cosmos-sdk/client"
 	clienttypes "github.com/cosmos/ibc-go/v3/modules/core/02-client/types"
 	chantypes "github.com/cosmos/ibc-go/v3/modules/core/04-channel/types"
 	"github.com/cosmos/relayer/v2/relayer/processor"
 	"github.com/cosmos/relayer/v2/relayer/provider"
 	"github.com/cosmos/relayer/v2/relayer/provider/cosmos"
-	provtypes "github.com/tendermint/tendermint/light/provider"
-	prov "github.com/tendermint/tendermint/light/provider/http"
 	ctypes "github.com/tendermint/tendermint/rpc/core/types"
-	tmtypes "github.com/tendermint/tendermint/types"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 )
@@ -27,12 +22,6 @@ type CosmosChainProcessor struct {
 
 	chainProvider *cosmos.CosmosProvider
 
-	// sdk context
-	cc *cosmosClient.Context
-
-	// for fetching light blocks from tendermint
-	lightProvider provtypes.Provider
-
 	pathProcessors processor.PathProcessors
 
 	// indicates whether queries are in sync with latest height of the chain
@@ -51,25 +40,14 @@ type CosmosChainProcessor struct {
 	channelStateCache processor.ChannelStateCache
 }
 
-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 {
-		return nil, fmt.Errorf("error getting cosmos client: %w", err)
-	}
-	lightProvider, err := prov.New(chainID, rpcAddress)
-	if err != nil {
-		return nil, fmt.Errorf("error getting light provider: %w", err)
-	}
+func NewCosmosChainProcessor(log *zap.Logger, provider *cosmos.CosmosProvider) *CosmosChainProcessor {
 	return &CosmosChainProcessor{
-		log:                  log.With(zap.String("chain_name", provider.ChainName()), zap.String("chain_id", chainID)),
+		log:                  log.With(zap.String("chain_name", provider.ChainName()), zap.String("chain_id", provider.ChainId())),
 		chainProvider:        provider,
-		cc:                   cc,
-		lightProvider:        lightProvider,
 		latestClientState:    make(latestClientState),
 		connectionStateCache: make(processor.ConnectionStateCache),
 		channelStateCache:    make(processor.ChannelStateCache),
-	}, nil
+	}
 }
 
 const (
@@ -292,17 +270,17 @@ func (ccp *CosmosChainProcessor) queryCycle(ctx context.Context, persistence *qu
 	for i := persistence.latestQueriedBlock + 1; i <= persistence.latestHeight; i++ {
 		var eg errgroup.Group
 		var blockRes *ctypes.ResultBlockResults
-		var lightBlock *tmtypes.LightBlock
+		var ibcHeader provider.IBCHeader
 		eg.Go(func() (err error) {
 			queryCtx, cancelQueryCtx := context.WithTimeout(ctx, queryTimeout)
 			defer cancelQueryCtx()
-			blockRes, err = ccp.cc.Client.BlockResults(queryCtx, &i)
+			blockRes, err = ccp.chainProvider.RPCClient.BlockResults(queryCtx, &i)
 			return err
 		})
 		eg.Go(func() (err error) {
 			queryCtx, cancelQueryCtx := context.WithTimeout(ctx, queryTimeout)
 			defer cancelQueryCtx()
-			lightBlock, err = ccp.lightProvider.LightBlock(queryCtx, i)
+			ibcHeader, err = ccp.chainProvider.IBCHeaderAtHeight(queryCtx, i)
 			return err
 		})
 
@@ -311,14 +289,11 @@ func (ccp *CosmosChainProcessor) queryCycle(ctx context.Context, persistence *qu
 			return nil
 		}
 
-		ccp.latestBlock = provider.LatestBlock{
-			Height: uint64(lightBlock.Height),
-			Time:   lightBlock.Header.Time,
-		}
+		latestHeader = ibcHeader.(cosmos.CosmosIBCHeader)
 
-		latestHeader = cosmos.CosmosIBCHeader{
-			SignedHeader: lightBlock.SignedHeader,
-			ValidatorSet: lightBlock.ValidatorSet,
+		ccp.latestBlock = provider.LatestBlock{
+			Height: uint64(latestHeader.SignedHeader.Height),
+			Time:   latestHeader.SignedHeader.Time,
 		}
 
 		ibcHeaderCache[uint64(i)] = latestHeader
@@ -331,8 +306,6 @@ func (ccp *CosmosChainProcessor) queryCycle(ctx context.Context, persistence *qu
 			}
 			messages := ccp.ibcMessagesFromTransaction(tx)
 
-			ccp.log.Debug("Parsed IBC messages", zap.Any("messages", messages))
-
 			for _, m := range messages {
 				handler, ok := messageHandlers[m.messageType]
 				if !ok {
diff --git a/relayer/chains/cosmos/sdk_helper.go b/relayer/chains/cosmos/sdk_helper.go
deleted file mode 100644
index d3423b2..0000000
--- a/relayer/chains/cosmos/sdk_helper.go
+++ /dev/null
@@ -1,90 +0,0 @@
-package cosmos
-
-import (
-	"io"
-	"time"
-
-	"github.com/cosmos/cosmos-sdk/client"
-	"github.com/cosmos/cosmos-sdk/codec"
-	"github.com/cosmos/cosmos-sdk/codec/types"
-	"github.com/cosmos/cosmos-sdk/std"
-	"github.com/cosmos/cosmos-sdk/types/module"
-	"github.com/cosmos/cosmos-sdk/x/auth/tx"
-	transfertypes "github.com/cosmos/ibc-go/v3/modules/apps/transfer/types"
-	clienttypes "github.com/cosmos/ibc-go/v3/modules/core/02-client/types"
-	conntypes "github.com/cosmos/ibc-go/v3/modules/core/03-connection/types"
-	chantypes "github.com/cosmos/ibc-go/v3/modules/core/04-channel/types"
-	commitmenttypes "github.com/cosmos/ibc-go/v3/modules/core/23-commitment/types"
-	ibctypes "github.com/cosmos/ibc-go/v3/modules/core/types"
-	rpcclient "github.com/tendermint/tendermint/rpc/client"
-	rpchttp "github.com/tendermint/tendermint/rpc/client/http"
-	libclient "github.com/tendermint/tendermint/rpc/jsonrpc/client"
-)
-
-type Codec struct {
-	InterfaceRegistry types.InterfaceRegistry
-	Marshaler         codec.Codec
-	TxConfig          client.TxConfig
-	Amino             *codec.LegacyAmino
-}
-
-func makeCodecConfig() Codec {
-	interfaceRegistry := types.NewInterfaceRegistry()
-	marshaler := codec.NewProtoCodec(interfaceRegistry)
-	chantypes.RegisterInterfaces(interfaceRegistry)
-	clienttypes.RegisterInterfaces(interfaceRegistry)
-	conntypes.RegisterInterfaces(interfaceRegistry)
-	ibctypes.RegisterInterfaces(interfaceRegistry)
-	transfertypes.RegisterInterfaces(interfaceRegistry)
-	commitmenttypes.RegisterInterfaces(interfaceRegistry)
-
-	return Codec{
-		InterfaceRegistry: interfaceRegistry,
-		Marshaler:         marshaler,
-		TxConfig:          tx.NewTxConfig(marshaler, tx.DefaultSignModes),
-		Amino:             codec.NewLegacyAmino(),
-	}
-}
-
-func makeCodec(moduleBasics []module.AppModuleBasic) Codec {
-	modBasic := module.NewBasicManager(moduleBasics...)
-	encodingConfig := makeCodecConfig()
-	std.RegisterLegacyAminoCodec(encodingConfig.Amino)
-	std.RegisterInterfaces(encodingConfig.InterfaceRegistry)
-	modBasic.RegisterLegacyAminoCodec(encodingConfig.Amino)
-	modBasic.RegisterInterfaces(encodingConfig.InterfaceRegistry)
-	return encodingConfig
-}
-
-func newClient(addr string) (rpcclient.Client, error) {
-	httpClient, err := libclient.DefaultHTTPClient(addr)
-	if err != nil {
-		return nil, err
-	}
-
-	httpClient.Timeout = 10 * time.Second
-	rpcClient, err := rpchttp.NewWithClient(addr, "/websocket", httpClient)
-	if err != nil {
-		return nil, err
-	}
-
-	return rpcClient, nil
-}
-
-func getCosmosClient(rpcAddress string, chainID string, input io.Reader, output io.Writer) (*client.Context, error) {
-	c, err := newClient(rpcAddress)
-	if err != nil {
-		return nil, err
-	}
-	cdc := makeCodec([]module.AppModuleBasic{})
-	return &client.Context{
-		Client:            c,
-		ChainID:           chainID,
-		Input:             input,
-		Output:            output,
-		TxConfig:          cdc.TxConfig,
-		Codec:             cdc.Marshaler,
-		InterfaceRegistry: cdc.InterfaceRegistry,
-		OutputFormat:      "json",
-	}, nil
-}
diff --git a/relayer/processor/path_processor.go b/relayer/processor/path_processor.go
index e7c6d60..d877fa1 100644
--- a/relayer/processor/path_processor.go
+++ b/relayer/processor/path_processor.go
@@ -204,7 +204,7 @@ func (pp *PathProcessor) appendPacketOrTimeout(ctx context.Context, src, dst *pa
 		// if timeouts were detected, need to generate msgs for them for src
 		switch {
 		case errors.As(err, &timeoutHeightErr) || errors.As(err, &timeoutTimestampErr):
-			if err := pp.assemblePacketIBCMessage(ctx, dst, src, channelKey, MsgTimeout, sequence, msgRecvPacket, dst.chainProvider.MsgTimeout, &res.SrcMessages); err != nil {
+			if err := pp.assemblePacketIBCMessage(ctx, dst, src, channelKey.Counterparty(), MsgTimeout, sequence, msgRecvPacket, dst.chainProvider.MsgTimeout, &res.SrcMessages); err != nil {
 				pp.log.Error("Error assembling MsgTimeout",
 					zap.Uint64("sequence", sequence),
 					zap.String("chain_id", src.info.ChainID),
@@ -212,7 +212,7 @@ func (pp *PathProcessor) appendPacketOrTimeout(ctx context.Context, src, dst *pa
 				)
 			}
 		case errors.As(err, &timeoutOnCloseErr):
-			if err := pp.assemblePacketIBCMessage(ctx, dst, src, channelKey, MsgTimeoutOnClose, sequence, msgRecvPacket, dst.chainProvider.MsgTimeoutOnClose, &res.SrcMessages); err != nil {
+			if err := pp.assemblePacketIBCMessage(ctx, dst, src, channelKey.Counterparty(), MsgTimeoutOnClose, sequence, msgRecvPacket, dst.chainProvider.MsgTimeoutOnClose, &res.SrcMessages); err != nil {
 				pp.log.Error("Error assembling MsgTimeoutOnClose",
 					zap.Uint64("sequence", sequence),
 					zap.String("chain_id", src.info.ChainID),
@@ -237,8 +237,8 @@ func (pp *PathProcessor) appendPacketOrTimeout(ctx context.Context, src, dst *pa
 	}
 }
 
-func (pp *PathProcessor) appendAcknowledgement(ctx context.Context, src, dst *pathEndRuntime, channelKey ChannelKey, sequence uint64, msgAcknowledgement provider.RelayerMessage, res *pathEndPacketFlowResponse) {
-	if err := pp.assemblePacketIBCMessage(ctx, src, dst, channelKey, MsgAcknowledgement, sequence, msgAcknowledgement, src.chainProvider.MsgAcknowledgement, &res.DstMessages); err != nil {
+func (pp *PathProcessor) appendAcknowledgement(ctx context.Context, src, dst *pathEndRuntime, channelKey ChannelKey, sequence uint64, msgAcknowledgement provider.RelayerMessage, messages *[]packetIBCMessage) {
+	if err := pp.assemblePacketIBCMessage(ctx, src, dst, channelKey, MsgAcknowledgement, sequence, msgAcknowledgement, src.chainProvider.MsgAcknowledgement, messages); err != nil {
 		pp.log.Error("Error assembling MsgAcknowledgement",
 			zap.Uint64("sequence", sequence),
 			zap.String("src_chain_id", src.info.ChainID),
@@ -288,7 +288,7 @@ MsgTransferLoop:
 		for msgRecvSeq, msgAcknowledgement := range pathEndPacketFlowMessages.DstMsgRecvPacket {
 			if transferSeq == msgRecvSeq {
 				// msg is received by dst chain, but no ack yet. Need to relay ack from dst to src!
-				pp.appendAcknowledgement(ctx, pathEndPacketFlowMessages.Dst, pathEndPacketFlowMessages.Src, pathEndPacketFlowMessages.ChannelKey, msgRecvSeq, msgAcknowledgement, res)
+				pp.appendAcknowledgement(ctx, pathEndPacketFlowMessages.Dst, pathEndPacketFlowMessages.Src, pathEndPacketFlowMessages.ChannelKey, msgRecvSeq, msgAcknowledgement, &res.SrcMessages)
 				continue MsgTransferLoop
 			}
 		}
@@ -461,6 +461,12 @@ func (pp *PathProcessor) assembleMsgUpdateClient(ctx context.Context, src, dst *
 		trustedConsensusHeight = clientConsensusHeight
 	}
 
+	if src.latestHeader.Height() == trustedConsensusHeight.RevisionHeight {
+		return nil, fmt.Errorf("latest header height is equal to the client trusted height: %d, "+
+			"need to wait for next block's header before we can assemble and send a new MsgUpdateClient",
+			trustedConsensusHeight.RevisionHeight)
+	}
+
 	msgUpdateClientHeader, err := src.chainProvider.MsgUpdateClientHeader(src.latestHeader, trustedConsensusHeight, dst.clientTrustedState.IBCHeader)
 	if err != nil {
 		return nil, fmt.Errorf("error assembling new client header: %w", err)
diff --git a/relayer/provider/cosmos/provider.go b/relayer/provider/cosmos/provider.go
index b90442b..cb811ba 100644
--- a/relayer/provider/cosmos/provider.go
+++ b/relayer/provider/cosmos/provider.go
@@ -101,6 +101,10 @@ type CosmosIBCHeader struct {
 // noop to implement processor.IBCHeader
 func (h CosmosIBCHeader) IBCHeaderIndicator() {}
 
+func (h CosmosIBCHeader) Height() uint64 {
+	return uint64(h.SignedHeader.Height)
+}
+
 func (cc *CosmosProvider) ProviderConfig() provider.ProviderConfig {
 	return cc.PCfg
 }
diff --git a/relayer/provider/provider.go b/relayer/provider/provider.go
index 85b66a7..ef5edcb 100644
--- a/relayer/provider/provider.go
+++ b/relayer/provider/provider.go
@@ -46,6 +46,7 @@ type LatestBlock struct {
 
 type IBCHeader interface {
 	IBCHeaderIndicator()
+	Height() uint64
 }
 
 // ClientState holds the current state of a client from a single chain's perspective
diff --git a/relayer/strategies.go b/relayer/strategies.go
index 0f09614..c15535b 100644
--- a/relayer/strategies.go
+++ b/relayer/strategies.go
@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"strings"
 	"sync"
 	"time"
@@ -30,7 +29,15 @@ const (
 )
 
 // 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, stdin io.Reader, stdout io.Writer, filter ChannelFilter, maxTxSize, maxMsgLength uint64, processorType string, initialBlockHistory uint64) chan error {
+func StartRelayer(
+	ctx context.Context,
+	log *zap.Logger,
+	src, dst *Chain,
+	filter ChannelFilter,
+	maxTxSize, maxMsgLength uint64,
+	processorType string,
+	initialBlockHistory uint64,
+) chan error {
 	errorChan := make(chan error, 1)
 
 	switch processorType {
@@ -45,18 +52,16 @@ func StartRelayer(ctx context.Context, log *zap.Logger, src, dst *Chain, stdin i
 		}
 		paths := []path{{
 			src: pathChain{
-				provider:   src.ChainProvider,
-				pathEnd:    processor.NewPathEnd(src.ChainProvider.ChainId(), src.ClientID(), filter.Rule, filterSrc),
-				rpcAddress: src.RPCAddr,
+				provider: src.ChainProvider,
+				pathEnd:  processor.NewPathEnd(src.ChainProvider.ChainId(), src.ClientID(), filter.Rule, filterSrc),
 			},
 			dst: pathChain{
-				provider:   dst.ChainProvider,
-				pathEnd:    processor.NewPathEnd(dst.ChainProvider.ChainId(), dst.ClientID(), filter.Rule, filterDst),
-				rpcAddress: dst.RPCAddr,
+				provider: dst.ChainProvider,
+				pathEnd:  processor.NewPathEnd(dst.ChainProvider.ChainId(), dst.ClientID(), filter.Rule, filterDst),
 			},
 		}}
 
-		go relayerStartEventProcessor(ctx, log, stdin, stdout, paths, initialBlockHistory, maxTxSize, maxMsgLength, errorChan)
+		go relayerStartEventProcessor(ctx, log, paths, initialBlockHistory, maxTxSize, maxMsgLength, errorChan)
 		return errorChan
 	case ProcessorLegacy:
 		go relayerMainLoop(ctx, log, src, dst, filter, maxTxSize, maxMsgLength, errorChan)
@@ -74,28 +79,23 @@ type path struct {
 }
 
 type pathChain struct {
-	provider   provider.ChainProvider
-	pathEnd    processor.PathEnd
-	rpcAddress string
+	provider provider.ChainProvider
+	pathEnd  processor.PathEnd
 }
 
 // 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 {
+func (chain pathChain) chainProcessor(log *zap.Logger) 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
+		return cosmosprocessor.NewCosmosChainProcessor(log, p)
 	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) {
+func relayerStartEventProcessor(ctx context.Context, log *zap.Logger, paths []path, initialBlockHistory uint64, maxTxSize, maxMsgLength uint64, errCh chan<- error) {
 	defer close(errCh)
 
 	epb := processor.NewEventProcessor()
@@ -103,8 +103,8 @@ func relayerStartEventProcessor(ctx context.Context, log *zap.Logger, stdin io.R
 	for _, p := range paths {
 		epb = epb.
 			WithChainProcessors(
-				p.src.chainProcessor(log, stdin, stdout),
-				p.dst.chainProcessor(log, stdin, stdout),
+				p.src.chainProcessor(log),
+				p.dst.chainProcessor(log),
 			).
 			WithPathProcessors(processor.NewPathProcessor(
 				log,


Test patch:
diff --git a/_test/relayer_chain_test.go b/_test/relayer_chain_test.go
index 6fd0c8d..ae33276 100644
--- a/_test/relayer_chain_test.go
+++ b/_test/relayer_chain_test.go
@@ -1,10 +1,8 @@
 package test
 
 import (
-	"bytes"
 	"context"
 	"fmt"
-	"io"
 	"testing"
 	"time"
 
@@ -121,7 +119,7 @@ func chainTest(t *testing.T, tcs []testChain) {
 
 	t.Log("Starting relayer")
 	filter := relayer.ChannelFilter{}
-	_ = relayer.StartRelayer(ctx, log, src, dst, bytes.NewReader(nil), io.Discard, filter, 2*cmd.MB, 5, relayer.ProcessorEvents, 20)
+	_ = relayer.StartRelayer(ctx, log, src, dst, 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))
@@ -271,7 +269,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, bytes.NewReader(nil), io.Discard, filter, 2*cmd.MB, 5, relayer.ProcessorEvents, 20)
+	_ = relayer.StartRelayer(ctx, log, src, dst, filter, 2*cmd.MB, 5, relayer.ProcessorEvents, 20)
 
 	// Wait for relay message inclusion in both chains
 	require.NoError(t, src.ChainProvider.WaitForNBlocks(ctx, 1))
@@ -445,7 +443,7 @@ func TestRelayAllChannelsOnConnection(t *testing.T) {
 
 	t.Log("Starting relayer")
 	filter := relayer.ChannelFilter{}
-	_ = relayer.StartRelayer(ctx, log, src, dst, bytes.NewReader(nil), io.Discard, filter, 2*cmd.MB, 5, relayer.ProcessorEvents, 20)
+	_ = relayer.StartRelayer(ctx, log, src, dst, 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))
@@ -621,7 +619,7 @@ func TestUnorderedChannelBlockHeightTimeout(t *testing.T) {
 
 	// start the relayer process in it's own goroutine
 	filter := relayer.ChannelFilter{}
-	_ = relayer.StartRelayer(ctx, log, src, dst, bytes.NewReader(nil), io.Discard, filter, 2*cmd.MB, 5, relayer.ProcessorEvents, 20)
+	_ = relayer.StartRelayer(ctx, log, src, dst, filter, 2*cmd.MB, 5, relayer.ProcessorEvents, 20)
 
 	require.NoError(t, src.ChainProvider.WaitForNBlocks(ctx, 5))
 
@@ -719,7 +717,7 @@ func TestUnorderedChannelTimestampTimeout(t *testing.T) {
 
 	// start the relayer process in it's own goroutine
 	filter := relayer.ChannelFilter{}
-	_ = relayer.StartRelayer(ctx, log, src, dst, bytes.NewReader(nil), io.Discard, filter, 2*cmd.MB, 5, relayer.ProcessorEvents, 20)
+	_ = relayer.StartRelayer(ctx, log, src, dst, 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 3ad21d4..f8b3da3 100644
--- a/relayer/chains/cosmos/msg_handlers_packet_test.go
+++ b/relayer/chains/cosmos/msg_handlers_packet_test.go
@@ -1,8 +1,6 @@
 package cosmos
 
 import (
-	"bytes"
-	"io"
 	"testing"
 
 	clienttypes "github.com/cosmos/ibc-go/v3/modules/core/02-client/types"
@@ -27,9 +25,7 @@ func mockCosmosChainProcessor(t *testing.T) *CosmosChainProcessor {
 		provider      = cosmos.CosmosProvider{PCfg: cosmos.CosmosProviderConfig{ChainID: chainID1}}
 	)
 
-	ccp, err := NewCosmosChainProcessor(log, &provider, "", bytes.NewReader(nil), io.Discard)
-	require.NoError(t, err, "error constructing cosmos chain processor")
-
+	ccp := NewCosmosChainProcessor(log, &provider)
 	ccp.SetPathProcessors(processor.PathProcessors{pathProcessor})
 
 	applicable := pathProcessor.SetChainProviderIfApplicable(&provider)
diff --git a/relayer/processor/types_test.go b/relayer/processor/types_test.go
index 3cf6048..12fc8b0 100644
--- a/relayer/processor/types_test.go
+++ b/relayer/processor/types_test.go
@@ -11,6 +11,7 @@ import (
 type mockIBCHeader struct{}
 
 func (h mockIBCHeader) IBCHeaderIndicator() {}
+func (h mockIBCHeader) Height() uint64      { return 0 }
 
 func TestIBCHeaderCachePrune(t *testing.T) {
 	cache := make(processor.IBCHeaderCache)