Repository: cosmos/relayer
Pull request #754 (2022-05-25 05:13:09)

Slim down path processor, mock chain processor
Slimmed down version of #743 to highlight the concurrent interactions between the chain processors and the path processor.

This has been designed with the intention of supporting multiple path processors and multiple chain processors, with multiple chain processors connected to multiple path processors and vise versa, to support relaying multiple different paths, but only needing a single `ChainProcessor` to be running per involved chain. For example, if relaying `cosmoshub` to `osmosis` and `cosmoshub` to `juno`, there would be two `PathProcessor`s and three `ChainProcessor`s.

Includes a mock `ChainProcessor` with test `TestMockChainAndPathProcessors` to observe the behavior.

Much of the `PathProcessor` is still there to illustrate the core logic and dependencies between the chain and path processors with regards to concurrent access to the `messages` field of `PathEndRuntime`.

Here was the desired behavior with the `PathProcessor` while initially building this out:
- When the `ChainProcessor` publishes relevant messages to the `PathProcessor`, it should execute the process cycle right away if it is not currently running (if both `ChainProcessor`s are in sync).
- If the `PathProcessor` `process` is in progress while messages are published, schedule the `process` to run again as soon as it completes the current cycle.
- If any messages fail to send, schedule the `process` to retry after a delay.
- The `PathProcessor` `process` should not need to run outside of the above conditions. Being triggered by the `ChainProcessor`s and/or a retry should be enough.


agouin: > I do see multiple mutexes mixed in with waitgroups. In my experience, that is typically a code smell because it's higher risk for deadlocks.

The order is important, that's why I made sure the waitgroup is only marked done after any locks are unlocked. Also the locks are used in a very isolated scope, locking to mutate or read and unlocking immediately after, avoiding any nested locks which would introduce the possibility of deadlock if different threads did nested locks in a different order. 
agouin: > Here is a skeleton example of a PathProcessor that has a single background goroutine handling work without mutating state crossing many goroutine boundaries

I like this for it's simplicity and lighter state, but I am not sure if it is keeping track of enough things to stop message from falling through the cracks.

With the map message cache, we know exactly what state the packet flow (for unfinished packet flow sequences) is in for each sequence, so when going through the process loop again with new messages, we can update the state for those packet flows, handle anything new that needs to be relayed or retried, and clear them out if complete. 

The main difference I see is that the map message cache uses confirmation from the chain (in subsequent blocks) that the packets have been handled, and the channel pattern depends on always getting proper error messages from the chain nodes from the send messages call.

The channel pattern also doesn't seem to cover scenarios such as when we observe an ack from a different relayer for a mesgrecv that was already added to the channel previously. This would lead to an unnecessary spend from the relayer.

Patch:
diff --git a/relayer/chains/mock/message_handlers.go b/relayer/chains/mock/message_handlers.go
new file mode 100644
index 0000000..cc5b46f
--- /dev/null
+++ b/relayer/chains/mock/message_handlers.go
@@ -0,0 +1,90 @@
+package mock
+
+import (
+	"fmt"
+
+	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"
+	"go.uber.org/zap"
+)
+
+type MsgHandlerParams struct {
+	mcp           *MockChainProcessor
+	PacketInfo    *chantypes.Packet
+	FoundMessages processor.ChannelMessageCache
+}
+
+var messageHandlers = map[string]func(MsgHandlerParams){
+	processor.MsgTransfer:        handleMsgTransfer,
+	processor.MsgRecvPacket:      handleMsgRecvPacket,
+	processor.MsgAcknowledgement: handleMsgAcknowlegement,
+
+	// TODO handlers for packet timeout, client, channel, and connection messages
+}
+
+func retainMessage(p MsgHandlerParams, channelKey processor.ChannelKey, message string, sequence uint64, ibcMessage provider.RelayerMessage) {
+	if _, ok := p.FoundMessages[channelKey]; !ok {
+		p.FoundMessages[channelKey] = make(processor.MessageCache)
+	}
+	if _, ok := p.FoundMessages[channelKey][message]; !ok {
+		p.FoundMessages[channelKey][message] = make(processor.SequenceCache)
+	}
+	p.FoundMessages[channelKey][message][sequence] = ibcMessage
+}
+
+func handleMsgTransfer(p MsgHandlerParams) {
+	channelKey := processor.ChannelKey{
+		ChannelID:             p.PacketInfo.SourceChannel,
+		PortID:                p.PacketInfo.SourcePort,
+		CounterpartyChannelID: p.PacketInfo.DestinationChannel,
+		CounterpartyPortID:    p.PacketInfo.DestinationPort,
+	}
+	retainMessage(p, channelKey, processor.MsgTransfer, p.PacketInfo.Sequence, cosmos.NewCosmosMessage(&chantypes.MsgRecvPacket{Packet: *p.PacketInfo}))
+	p.mcp.log.Debug("observed MsgTransfer",
+		zap.String("chain_id", p.mcp.chainID),
+		zap.Uint64("sequence", p.PacketInfo.Sequence),
+		zap.String("src_channel", p.PacketInfo.SourceChannel),
+		zap.String("src_port", p.PacketInfo.SourcePort),
+		zap.String("timeout_height", fmt.Sprintf("%d-%d", p.PacketInfo.TimeoutHeight.RevisionNumber, p.PacketInfo.TimeoutHeight.RevisionHeight)),
+		zap.Uint64("timeout_timestamp", p.PacketInfo.TimeoutTimestamp),
+	)
+}
+
+func handleMsgRecvPacket(p MsgHandlerParams) {
+	channelKey := processor.ChannelKey{
+		ChannelID:             p.PacketInfo.DestinationChannel,
+		PortID:                p.PacketInfo.DestinationPort,
+		CounterpartyChannelID: p.PacketInfo.SourceChannel,
+		CounterpartyPortID:    p.PacketInfo.SourcePort,
+	}
+	retainMessage(p, channelKey, processor.MsgRecvPacket, p.PacketInfo.Sequence, cosmos.NewCosmosMessage(&chantypes.MsgAcknowledgement{Packet: *p.PacketInfo}))
+	p.mcp.log.Debug("observed MsgRecvPacket",
+		zap.String("chain_id", p.mcp.chainID),
+		zap.Uint64("sequence", p.PacketInfo.Sequence),
+		zap.String("src_channel", p.PacketInfo.SourceChannel),
+		zap.String("src_port", p.PacketInfo.SourcePort),
+		zap.String("dst_channel", p.PacketInfo.DestinationChannel),
+		zap.String("dst_port", p.PacketInfo.DestinationPort),
+	)
+}
+
+func handleMsgAcknowlegement(p MsgHandlerParams) {
+	channelKey := processor.ChannelKey{
+		ChannelID:             p.PacketInfo.SourceChannel,
+		PortID:                p.PacketInfo.SourcePort,
+		CounterpartyChannelID: p.PacketInfo.DestinationChannel,
+		CounterpartyPortID:    p.PacketInfo.DestinationPort,
+	}
+	retainMessage(p, channelKey, processor.MsgAcknowledgement, p.PacketInfo.Sequence, nil)
+
+	p.mcp.log.Debug("observed MsgAcknowledgement",
+		zap.String("chain_id", p.mcp.chainID),
+		zap.Uint64("sequence", p.PacketInfo.Sequence),
+		zap.String("src_channel", p.PacketInfo.SourceChannel),
+		zap.String("src_port", p.PacketInfo.SourcePort),
+		zap.String("dst_channel", p.PacketInfo.DestinationChannel),
+		zap.String("dst_port", p.PacketInfo.DestinationPort),
+	)
+}
diff --git a/relayer/chains/mock/mock_chain_processor.go b/relayer/chains/mock/mock_chain_processor.go
new file mode 100644
index 0000000..0afa40f
--- /dev/null
+++ b/relayer/chains/mock/mock_chain_processor.go
@@ -0,0 +1,170 @@
+package mock
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/cosmos/relayer/v2/relayer/processor"
+
+	chantypes "github.com/cosmos/ibc-go/v3/modules/core/04-channel/types"
+	"go.uber.org/zap"
+)
+
+const (
+	minQueryLoopDuration = 1 * time.Second
+)
+
+type MockChainProcessor struct {
+	log *zap.Logger
+
+	chainID string
+
+	// subscribers to this chain processor, where relevant IBC messages will be published
+	pathProcessors []*processor.PathProcessor
+
+	// cached latest height of the chain
+	latestHeight uint64
+
+	// is the query loop up to date with the latest blocks of the chain
+	inSync     bool
+	inSyncLock sync.RWMutex
+
+	getMockMessages func() []TransactionMessage
+}
+
+// types used for parsing IBC messages from transactions, then passed to message handlers for mutating the MockChainProcessor state if necessary and retaining applicable messages for sending to the Path Processors
+type TransactionMessage struct {
+	Action     string
+	PacketInfo *chantypes.Packet
+}
+
+func NewMockChainProcessor(log *zap.Logger, chainID string, getMockMessages func() []TransactionMessage) *MockChainProcessor {
+	return &MockChainProcessor{
+		log:             log,
+		chainID:         chainID,
+		getMockMessages: getMockMessages,
+	}
+}
+
+func (mcp *MockChainProcessor) SetPathProcessors(pathProcessors processor.PathProcessors) {
+	mcp.pathProcessors = pathProcessors
+}
+
+func (mcp *MockChainProcessor) InSync() bool {
+	mcp.inSyncLock.RLock()
+	defer mcp.inSyncLock.RUnlock()
+	return mcp.inSync
+}
+
+func (mcp *MockChainProcessor) ChainID() string {
+	return mcp.chainID
+}
+
+func (mcp *MockChainProcessor) Run(ctx context.Context, initialBlockHistory uint64) error {
+
+	// would be query of latest height, mocking 20
+	mcp.latestHeight = 20
+
+	// this will make initial QueryLoop iteration look back initialBlockHistory blocks in history
+	latestQueriedBlockInt64 := int64(mcp.latestHeight) - int64(initialBlockHistory)
+
+	var latestQueriedBlock uint64
+	if latestQueriedBlockInt64 < 0 {
+		latestQueriedBlock = 0
+	} else {
+		latestQueriedBlock = uint64(latestQueriedBlockInt64)
+	}
+
+	mcp.log.Info("entering main query loop", zap.String("chain_id", mcp.chainID))
+
+	// QueryLoop:
+	for {
+		if ctx.Err() != nil {
+			return nil
+		}
+		cycleTimeStart := time.Now()
+		doneWithThisCycle := func() {
+			queryDuration := time.Since(cycleTimeStart)
+			if queryDuration < minQueryLoopDuration {
+				time.Sleep(minQueryLoopDuration - queryDuration)
+			}
+		}
+		// would be query of latest height
+		mcp.latestHeight++
+		// until in sync, determine if our latest queries are up to date with the current chain height
+		// this will cause the PathProcessors to start processing the backlog of message state (once both chainprocessors are in sync)
+		mcp.inSyncLock.RLock()
+		inSync := mcp.inSync
+		mcp.inSyncLock.RUnlock()
+
+		if !inSync {
+			if (mcp.latestHeight - latestQueriedBlock) < 2 {
+				mcp.inSyncLock.Lock()
+				mcp.inSync = true
+				mcp.inSyncLock.Unlock()
+				mcp.log.Info("chain is in sync", zap.String("chain_id", mcp.chainID))
+			} else {
+				mcp.log.Warn("chain is not yet in sync",
+					zap.String("chain_id", mcp.chainID),
+					zap.Uint64("latest_queried_block", latestQueriedBlock),
+					zap.Uint64("latest_height", mcp.latestHeight),
+				)
+			}
+		}
+
+		mcp.log.Debug("queried latest height",
+			zap.String("chain_id", mcp.chainID),
+			zap.Uint64("latest_height", mcp.latestHeight),
+		)
+
+		for i := latestQueriedBlock + 1; i <= mcp.latestHeight; i++ {
+			// fetch light block
+
+			// store light block's latest signed header and validatorset on chainProcessor if i == latestHeight, needed for constructing MsgUpdateClient on counterparty chain
+			// cache last n validatorsets also since needed for constructing MsgUpdateClient for this chain
+
+			// fetch block
+
+			// used for collecting IBC messages that will be sent to the Path Processors
+			foundMessages := make(processor.ChannelMessageCache)
+
+			// iterate through transactions
+			// iterate through messages in transactions
+			// get slice of all IBC messages in those transactions
+
+			// for _, tx := range blockRes.TxsResults {
+			//   if tx.Code != 0 {
+			//     // tx was not successful
+			//     continue
+			//   }
+			messages := mcp.getMockMessages()
+
+			// iterate through ibc messages and call specific handler for each
+			// will do things like mutate chainprocessor state and add relevant messages to foundMessages
+			// this can be parralelized also
+			for _, m := range messages {
+				if handler, ok := messageHandlers[m.Action]; ok {
+					handler(MsgHandlerParams{
+						mcp:           mcp,
+						PacketInfo:    m.PacketInfo,
+						FoundMessages: foundMessages,
+					})
+				}
+			}
+			// }
+
+			// now pass foundMessages to the path processors
+			for channelKey, messages := range foundMessages {
+				// TODO do not relay on closed channels
+				for _, pp := range mcp.pathProcessors {
+					mcp.log.Info("sending messages to path processor", zap.String("chain_id", mcp.chainID))
+					pp.HandleNewMessages(mcp.chainID, channelKey, messages)
+				}
+			}
+			latestQueriedBlock = i
+		}
+
+		doneWithThisCycle()
+	}
+}
diff --git a/relayer/processor/chain_processor.go b/relayer/processor/chain_processor.go
new file mode 100644
index 0000000..c1003ab
--- /dev/null
+++ b/relayer/processor/chain_processor.go
@@ -0,0 +1,28 @@
+package processor
+
+import (
+	"context"
+)
+
+// The ChainProcessor interface is reponsible for polling blocks and emitting IBC message events to the PathProcessors.
+// It is also responsible for tracking open channels and not sending messages to the PathProcessors for closed channels.
+type ChainProcessor interface {
+	// Run starts the query loop for the chain which will gather applicable ibc messages and push events out to the relevant PathProcessors.
+	// The initialBlockHistory parameter determines how many historical blocks should be fetched and processed before continuing with current blocks.
+	// ChainProcessors should obey the context and return upon context cancellation.
+	Run(ctx context.Context, initialBlockHistory uint64) error
+
+	// InSync indicates whether queries are in sync with latest height of the chain.
+	// The PathProcessors use this as a signal for determining if the backlog of messaged is ready to be processed and relayed.
+	InSync() bool
+
+	// ChainID returns the identifier of the chain
+	ChainID() string
+
+	// Set the PathProcessors that this ChainProcessor should publish relevant IBC events to.
+	// ChainProcessors need reference to their PathProcessors and vice-versa, handled by EventProcessorBuilder.Build().
+	SetPathProcessors(pathProcessors PathProcessors)
+}
+
+// ChainProcessors is a slice of ChainProcessor instances.
+type ChainProcessors []ChainProcessor
diff --git a/relayer/processor/event_processor.go b/relayer/processor/event_processor.go
new file mode 100644
index 0000000..c0eb79e
--- /dev/null
+++ b/relayer/processor/event_processor.go
@@ -0,0 +1,90 @@
+package processor
+
+import (
+	"context"
+
+	"golang.org/x/sync/errgroup"
+)
+
+// EventProcessorBuilder is a configuration type with .With functions used for building an EventProcessor.
+type EventProcessorBuilder struct {
+	chainProcessors     ChainProcessors
+	initialBlockHistory uint64
+	pathProcessors      PathProcessors
+}
+
+// EventProcessor is a built instance that is ready to be executed with Run(ctx).
+type EventProcessor struct {
+	chainProcessors     ChainProcessors
+	initialBlockHistory uint64
+	pathProcessors      PathProcessors
+}
+
+// NewEventProcessor creates a builder than can be used to construct a multi-ChainProcessor, multi-PathProcessor topology for the relayer.
+func NewEventProcessor() EventProcessorBuilder {
+	return EventProcessorBuilder{}
+}
+
+// WithChainProcessors will set the ChainProcessors to be used.
+func (ep EventProcessorBuilder) WithChainProcessors(chainProcessors ...ChainProcessor) EventProcessorBuilder {
+	ep.chainProcessors = chainProcessors
+	return ep
+}
+
+// WithInitialBlockHistory will set 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.
+func (ep EventProcessorBuilder) WithPathProcessors(pathProcessors ...*PathProcessor) EventProcessorBuilder {
+	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.
+func (ep EventProcessorBuilder) Build() EventProcessor {
+	for _, chainProcessor := range ep.chainProcessors {
+		pathProcessorsForThisChain := PathProcessors{}
+		for _, pathProcessor := range ep.pathProcessors {
+			if pathProcessor.SetChainProcessorIfApplicable(chainProcessor.ChainID(), chainProcessor) {
+				pathProcessorsForThisChain = append(pathProcessorsForThisChain, pathProcessor)
+			}
+		}
+		chainProcessor.SetPathProcessors(pathProcessorsForThisChain)
+	}
+
+	return EventProcessor{
+		chainProcessors:     ep.chainProcessors,
+		initialBlockHistory: ep.initialBlockHistory,
+		pathProcessors:      ep.pathProcessors,
+	}
+}
+
+// Run is a blocking call that launches all provided PathProcessors and ChainProcessors in parallel.
+// It will return once all PathProcessors and ChainProcessors have stopped running due to context cancellation,
+// or if a critical error has occurred within one of the ChainProcessors.
+func (ep EventProcessor) Run(ctx context.Context) error {
+	var eg errgroup.Group
+	runCtx, runCtxCancel := context.WithCancel(ctx)
+	for _, pathProcessor := range ep.pathProcessors {
+		pathProcessor := pathProcessor
+		eg.Go(func() error {
+			pathProcessor.Run(runCtx)
+			return nil
+		})
+	}
+	for _, chainProcessor := range ep.chainProcessors {
+		chainProcessor := chainProcessor
+		eg.Go(func() error {
+			err := chainProcessor.Run(runCtx, ep.initialBlockHistory)
+			// Signal the other chain processors to exit.
+			runCtxCancel()
+			return err
+		})
+	}
+	err := eg.Wait()
+	runCtxCancel()
+	return err
+}
diff --git a/relayer/processor/path_end.go b/relayer/processor/path_end.go
new file mode 100644
index 0000000..e64fa7b
--- /dev/null
+++ b/relayer/processor/path_end.go
@@ -0,0 +1,9 @@
+package processor
+
+type PathEnd struct {
+	ChainID string
+
+	// TODO clientID, connectionID, and channel filter stuff (allowlist, denylist)
+}
+
+// TODO methods for determining if a given channel, client, or connection is relevant for this pathEnd
diff --git a/relayer/processor/path_processor.go b/relayer/processor/path_processor.go
new file mode 100644
index 0000000..3e86778
--- /dev/null
+++ b/relayer/processor/path_processor.go
@@ -0,0 +1,265 @@
+package processor
+
+import (
+	"context"
+
+	"sync"
+	"time"
+
+	"github.com/cosmos/relayer/v2/relayer/provider"
+	"go.uber.org/zap"
+	"golang.org/x/sync/errgroup"
+)
+
+const (
+	// DurationErrorRetry determines how long
+	DurationErrorRetry = 5 * time.Second
+)
+
+// PathProcessor is a process that handles incoming IBC messages from a pair of chains.
+// It determines what messages need to be relayed, and sends them.
+type PathProcessor struct {
+	log *zap.Logger
+
+	pathEnd1 *PathEndRuntime
+	pathEnd2 *PathEndRuntime
+
+	// Signals to retry.
+	retryProcess chan struct{}
+}
+
+// PathProcessors is a slice of PathProcessor instances
+type PathProcessors []*PathProcessor
+
+// PathEndRuntime is used at runtime for each chain involved in the path.
+// It holds a channel for incoming messages from the ChainProcessors, which will
+// be processed during Run(ctx).
+type PathEndRuntime struct {
+	info PathEnd
+
+	chainProcessor ChainProcessor
+
+	messageCache MessageCache
+
+	// New messages arriving from the handleNewMessagesForPathEnd method.
+	incomingMessages chan MessageCache
+}
+
+// IBCMessageWithSequence holds a packet's sequence along with it,
+// useful for sending packets around internal to the PathProcessor.
+type IBCMessageWithSequence struct {
+	Sequence uint64
+	Message  provider.RelayerMessage
+}
+
+func NewPathProcessor(log *zap.Logger, pathEnd1 PathEnd, pathEnd2 PathEnd) *PathProcessor {
+	return &PathProcessor{
+		log: log,
+		pathEnd1: &PathEndRuntime{
+			info:             pathEnd1,
+			incomingMessages: make(chan MessageCache, 100),
+			messageCache:     make(MessageCache),
+		},
+		pathEnd2: &PathEndRuntime{
+			info:             pathEnd2,
+			incomingMessages: make(chan MessageCache, 100),
+			messageCache:     make(MessageCache),
+		},
+		retryProcess: make(chan struct{}, 8),
+	}
+}
+
+// TEST USE ONLY
+func (pp *PathProcessor) PathEnd1Messages(message string) SequenceCache {
+	return pp.pathEnd1.messageCache[message]
+}
+
+// TEST USE ONLY
+func (pp *PathProcessor) PathEnd2Messages(message string) SequenceCache {
+	return pp.pathEnd2.messageCache[message]
+}
+
+// Path Processors are constructed before ChainProcessors, so reference needs to be added afterwards
+// This can be done inside the ChainProcessor constructor for simplification
+func (pp *PathProcessor) SetChainProcessorIfApplicable(chainID string, chainProcessor ChainProcessor) bool {
+	if pp.pathEnd1.info.ChainID == chainID {
+		pp.pathEnd1.chainProcessor = chainProcessor
+		return true
+	} else if pp.pathEnd2.info.ChainID == chainID {
+		pp.pathEnd2.chainProcessor = chainProcessor
+		return true
+	}
+	return false
+}
+
+// ChainProcessors call this method when they have new IBC messages
+func (pp *PathProcessor) HandleNewMessages(chainID string, channelKey ChannelKey, messages MessageCache) {
+	if pp.pathEnd1.info.ChainID == chainID {
+		// TODO make sure passes channel filter for pathEnd1 before calling this
+		pp.handleNewMessagesForPathEnd(pp.pathEnd1, messages)
+	} else if pp.pathEnd2.info.ChainID == chainID {
+		// TODO make sure passes channel filter for pathEnd2 before calling this
+		pp.handleNewMessagesForPathEnd(pp.pathEnd2, messages)
+	}
+}
+
+func (pp *PathProcessor) handleNewMessagesForPathEnd(
+	p *PathEndRuntime,
+	newMessages MessageCache,
+) {
+	p.incomingMessages <- newMessages
+}
+
+// this contains MsgRecvPacket from same chain
+// needs to be transformed into PathEndPacketFlowMessages once counterparty info is available to complete packet flow state for pathEnd
+type PathEndMessages struct {
+	MsgTransfer        SequenceCache
+	MsgRecvPacket      SequenceCache
+	MsgAcknowledgement SequenceCache
+}
+
+// contains MsgRecvPacket from counterparty
+// entire packet flow
+type PathEndPacketFlowMessages struct {
+	SrcMsgTransfer        SequenceCache
+	DstMsgRecvPacket      SequenceCache
+	SrcMsgAcknowledgement SequenceCache
+	// TODO SrcTimeout and SrcTimeoutOnClose
+}
+
+type PathEndProcessedResponse struct {
+	UnrelayedPackets          []IBCMessageWithSequence
+	UnrelayedAcknowledgements []IBCMessageWithSequence
+
+	ToDeleteSrc map[string][]uint64
+	ToDeleteDst map[string][]uint64
+}
+
+func (m PathEndProcessedResponse) appendPacket(sequence uint64, msgRecvPacket provider.RelayerMessage) {
+	m.UnrelayedPackets = append(m.UnrelayedPackets, IBCMessageWithSequence{Sequence: sequence, Message: msgRecvPacket})
+}
+
+func (m PathEndProcessedResponse) appendAcknowledgement(sequence uint64, msgAcknowledgement provider.RelayerMessage) {
+	m.UnrelayedAcknowledgements = append(m.UnrelayedAcknowledgements, IBCMessageWithSequence{Sequence: sequence, Message: msgAcknowledgement})
+}
+
+func (pp *PathProcessor) getUnrelayedPacketsAndAcksAndToDelete(pathEndPacketFlowMessages PathEndPacketFlowMessages, wg *sync.WaitGroup, res *PathEndProcessedResponse) {
+	defer wg.Done()
+	res.UnrelayedPackets = nil
+	res.UnrelayedAcknowledgements = nil
+	res.ToDeleteSrc = make(map[string][]uint64)
+	res.ToDeleteDst = make(map[string][]uint64)
+
+MsgTransferLoop:
+	for transferSeq, msgTransfer := range pathEndPacketFlowMessages.SrcMsgTransfer {
+		for ackSeq := range pathEndPacketFlowMessages.SrcMsgAcknowledgement {
+			if transferSeq == ackSeq {
+				// we have an ack for this packet, so packet flow is complete
+				// remove all retention of this sequence number
+				res.ToDeleteSrc[MsgTransfer] = append(res.ToDeleteSrc[MsgTransfer], transferSeq)
+				res.ToDeleteDst[MsgRecvPacket] = append(res.ToDeleteDst[MsgRecvPacket], transferSeq)
+				res.ToDeleteSrc[MsgAcknowledgement] = append(res.ToDeleteSrc[MsgAcknowledgement], transferSeq)
+				continue MsgTransferLoop
+			}
+		}
+
+		// would iterate timeout messages here also
+
+		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!
+				res.appendAcknowledgement(msgRecvSeq, msgAcknowledgement)
+				continue MsgTransferLoop
+			}
+		}
+		// Packet is not yet relayed! need to relay from src to dst
+		res.appendPacket(transferSeq, msgTransfer)
+	}
+
+	// now iterate through packet-flow-complete messages and remove any leftover messages if the MsgTransfer or MsgRecvPacket was in a previous block that we did not query
+	for ackSeq := range pathEndPacketFlowMessages.SrcMsgAcknowledgement {
+		res.ToDeleteSrc[MsgTransfer] = append(res.ToDeleteSrc[MsgTransfer], ackSeq)
+		res.ToDeleteDst[MsgRecvPacket] = append(res.ToDeleteDst[MsgRecvPacket], ackSeq)
+		res.ToDeleteSrc[MsgAcknowledgement] = append(res.ToDeleteSrc[MsgAcknowledgement], ackSeq)
+	}
+
+	// would iterate timeout messages here also
+}
+
+func (pp *PathProcessor) sendMessages(pathEnd *PathEndRuntime, messages []IBCMessageWithSequence) error {
+	if len(messages) == 0 {
+		return nil
+	}
+	// TODO construct MsgUpdateClient for this pathEnd, using the latest trusted IBC header from other pathEnd, prepend messages with the MsgUpdateClient, then send the messages to this pathEnd
+
+	pp.log.Debug("will send", zap.Any("messages", messages))
+
+	return nil
+}
+
+// messages from both pathEnds are needed in order to determine what needs to be relayed for a single pathEnd
+func (pp *PathProcessor) processLatestMessages() error {
+	pathEnd1PacketFlowMessages := PathEndPacketFlowMessages{
+		SrcMsgTransfer:        pp.pathEnd1.messageCache[MsgTransfer],
+		DstMsgRecvPacket:      pp.pathEnd2.messageCache[MsgRecvPacket],
+		SrcMsgAcknowledgement: pp.pathEnd1.messageCache[MsgAcknowledgement],
+	}
+	pathEnd2PacketFlowMessages := PathEndPacketFlowMessages{
+		SrcMsgTransfer:        pp.pathEnd2.messageCache[MsgTransfer],
+		DstMsgRecvPacket:      pp.pathEnd1.messageCache[MsgRecvPacket],
+		SrcMsgAcknowledgement: pp.pathEnd2.messageCache[MsgAcknowledgement],
+	}
+
+	// process the packet flows for both packends to determine what needs to be relayed
+	var pathEnd1ProcessRes, pathEnd2ProcessRes PathEndProcessedResponse
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go pp.getUnrelayedPacketsAndAcksAndToDelete(pathEnd1PacketFlowMessages, &wg, &pathEnd1ProcessRes)
+	go pp.getUnrelayedPacketsAndAcksAndToDelete(pathEnd2PacketFlowMessages, &wg, &pathEnd2ProcessRes)
+	wg.Wait()
+
+	// concatenate applicable messages for pathend
+	pathEnd1Messages := append(pathEnd2ProcessRes.UnrelayedPackets, pathEnd1ProcessRes.UnrelayedAcknowledgements...)
+	pathEnd2Messages := append(pathEnd1ProcessRes.UnrelayedPackets, pathEnd2ProcessRes.UnrelayedAcknowledgements...)
+
+	pp.pathEnd1.messageCache.DeleteCachedMessages(pathEnd1ProcessRes.ToDeleteSrc, pathEnd2ProcessRes.ToDeleteDst)
+	pp.pathEnd2.messageCache.DeleteCachedMessages(pathEnd2ProcessRes.ToDeleteSrc, pathEnd1ProcessRes.ToDeleteDst)
+
+	// now send messages in parallel
+	var eg errgroup.Group
+	eg.Go(func() error { return pp.sendMessages(pp.pathEnd1, pathEnd1Messages) })
+	eg.Go(func() error { return pp.sendMessages(pp.pathEnd2, pathEnd2Messages) })
+	return eg.Wait()
+}
+
+// Run executes the main path process.
+func (pp *PathProcessor) Run(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+
+		case m := <-pp.pathEnd1.incomingMessages:
+			// if new messages are available from pathEnd1, run processLatestMessages
+			pp.pathEnd1.messageCache.Merge(m) // Merge incoming messages into the backlog of IBC messages for pathEnd1
+
+		case m := <-pp.pathEnd2.incomingMessages:
+			// if new messages are available from pathEnd2, run processLatestMessages
+			pp.pathEnd2.messageCache.Merge(m) // Merge incoming messages into the backlog of IBC messages for pathEnd2
+
+		case <-pp.retryProcess:
+			// No new messages to merge in, just retry handling.
+		}
+
+		if !pp.pathEnd1.chainProcessor.InSync() || !pp.pathEnd2.chainProcessor.InSync() {
+			continue
+		}
+		// process latest message cache state from both pathEnds
+		if err := pp.processLatestMessages(); err != nil {
+			// in case of IBC message send errors, schedule retry after DurationErrorRetry
+			time.AfterFunc(DurationErrorRetry, func() {
+				pp.retryProcess <- struct{}{}
+			})
+		}
+	}
+}
diff --git a/relayer/processor/types.go b/relayer/processor/types.go
new file mode 100644
index 0000000..2f437c8
--- /dev/null
+++ b/relayer/processor/types.go
@@ -0,0 +1,57 @@
+package processor
+
+import "github.com/cosmos/relayer/v2/relayer/provider"
+
+// These are IBC message types used as shared message keys between ChainProcessors and PathProcessors.
+const (
+	MsgTransfer        = "/ibc.applications.transfer.v1.MsgTransfer"
+	MsgRecvPacket      = "/ibc.core.channel.v1.MsgRecvPacket"
+	MsgAcknowledgement = "/ibc.core.channel.v1.MsgAcknowledgement"
+)
+
+// SequenceCache is used for caching an IBC message for a given packet sequence.
+type SequenceCache map[uint64]provider.RelayerMessage
+
+// MessageCache is used for caching a SequenceCache for a given IBC message type.
+type MessageCache map[string]SequenceCache
+
+// ChannelMessageCache is used for caching a MessageCache for a given IBC channel.
+type ChannelMessageCache map[ChannelKey]MessageCache
+
+// ChannelKey is the key used between ChainProcessor and PathProcessor.
+type ChannelKey struct {
+	ChannelID             string
+	PortID                string
+	CounterpartyChannelID string
+	CounterpartyPortID    string
+}
+
+// Merge will merge another MessageCache into this one.
+func (c MessageCache) Merge(other MessageCache) {
+	for k, v := range other {
+		c[k] = v
+	}
+}
+
+// Clone will create a deep copy of a MessageCache.
+func (c MessageCache) Clone() MessageCache {
+	newMessageCache := make(MessageCache)
+	for mk, mv := range c {
+		newSequenceCache := make(SequenceCache)
+		for sk, sv := range mv {
+			newSequenceCache[sk] = sv
+		}
+		newMessageCache[mk] = newSequenceCache
+	}
+	return newMessageCache
+}
+
+func (c MessageCache) DeleteCachedMessages(toDelete ...map[string][]uint64) {
+	for _, toDeleteMap := range toDelete {
+		for message, toDeleteMessages := range toDeleteMap {
+			for _, sequence := range toDeleteMessages {
+				delete(c[message], sequence)
+			}
+		}
+	}
+}


Test patch:
diff --git a/relayer/chains/mock/mock_chain_processor_test.go b/relayer/chains/mock/mock_chain_processor_test.go
new file mode 100644
index 0000000..72a6469
--- /dev/null
+++ b/relayer/chains/mock/mock_chain_processor_test.go
@@ -0,0 +1,117 @@
+package mock_test
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	chantypes "github.com/cosmos/ibc-go/v3/modules/core/04-channel/types"
+	"github.com/cosmos/relayer/v2/relayer/chains/mock"
+	"github.com/cosmos/relayer/v2/relayer/processor"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest"
+)
+
+func TestMockChainAndPathProcessors(t *testing.T) {
+	if testing.Short() {
+		t.Skip()
+	}
+
+	mockChainID1 := "mock-chain-1"
+	mockChainID2 := "mock-chain-2"
+
+	pathEnd1 := processor.PathEnd{ChainID: mockChainID1}
+	pathEnd2 := processor.PathEnd{ChainID: mockChainID2}
+
+	mockSequence1 := uint64(0)
+	mockSequence2 := uint64(0)
+	lastSentMockMsgRecvSequence1 := uint64(0)
+	lastSentMockMsgRecvSequence2 := uint64(0)
+	var mockLock sync.Mutex
+
+	getMockMessages1 := func() []mock.TransactionMessage {
+		return getMockMessages(&mockSequence1, &mockSequence2, &lastSentMockMsgRecvSequence2, &mockLock)
+	}
+	getMockMessages2 := func() []mock.TransactionMessage {
+		return getMockMessages(&mockSequence2, &mockSequence1, &lastSentMockMsgRecvSequence1, &mockLock)
+	}
+
+	log := zaptest.NewLogger(t)
+
+	ctx, ctxCancel := context.WithTimeout(context.Background(), time.Second*20)
+	defer ctxCancel()
+
+	pathProcessor := processor.NewPathProcessor(log, pathEnd1, pathEnd2)
+
+	eventProcessor := processor.NewEventProcessor().
+		WithChainProcessors(
+			mock.NewMockChainProcessor(log, mockChainID1, getMockMessages1),
+			mock.NewMockChainProcessor(log, mockChainID2, getMockMessages2),
+		).
+		WithInitialBlockHistory(100).
+		WithPathProcessors(pathProcessor).
+		Build()
+
+	err := eventProcessor.Run(ctx)
+	require.NoError(t, err, "error running event processor")
+
+	pathEnd1LeftoverMsgTransfer := pathProcessor.PathEnd1Messages(processor.MsgTransfer)
+	pathEnd1LeftoverMsgRecvPacket := pathProcessor.PathEnd1Messages(processor.MsgRecvPacket)
+	pathEnd1LeftoverMsgAcknowledgement := pathProcessor.PathEnd1Messages(processor.MsgAcknowledgement)
+
+	pathEnd2LeftoverMsgTransfer := pathProcessor.PathEnd2Messages(processor.MsgTransfer)
+	pathEnd2LeftoverMsgRecvPacket := pathProcessor.PathEnd2Messages(processor.MsgRecvPacket)
+	pathEnd2LeftoverMsgAcknowledgement := pathProcessor.PathEnd2Messages(processor.MsgAcknowledgement)
+
+	log.Debug("leftover",
+		zap.Int("pathEnd1MsgTransfer", len(pathEnd1LeftoverMsgTransfer)),
+		zap.Int("pathEnd1MsgRecvPacket", len(pathEnd1LeftoverMsgRecvPacket)),
+		zap.Int("pathEnd1MsgAcknowledgement", len(pathEnd1LeftoverMsgAcknowledgement)),
+		zap.Int("pathEnd2MsgTransfer", len(pathEnd2LeftoverMsgTransfer)),
+		zap.Int("pathEnd2MsgRecvPacket", len(pathEnd2LeftoverMsgRecvPacket)),
+		zap.Int("pathEnd2MsgAcknowledgement", len(pathEnd2LeftoverMsgAcknowledgement)),
+	)
+
+	// at most 3 msg transfer could still be stuck in queue since chain processor was shut down, so msgrecvpacket would never be "received" by counterparty
+	require.LessOrEqual(t, len(pathEnd1LeftoverMsgTransfer), 3)
+	// at most 2 msgrecvpacket could still be stuck in the queue
+	require.LessOrEqual(t, len(pathEnd1LeftoverMsgRecvPacket), 2)
+	// at most 1 msgAcknowledgement could still be stuck in the queue
+	require.LessOrEqual(t, len(pathEnd1LeftoverMsgAcknowledgement), 1)
+
+	require.LessOrEqual(t, len(pathEnd2LeftoverMsgTransfer), 3)
+	require.LessOrEqual(t, len(pathEnd2LeftoverMsgRecvPacket), 2)
+	require.LessOrEqual(t, len(pathEnd2LeftoverMsgAcknowledgement), 1)
+}
+
+// will send cycles of:
+// MsgTransfer
+// MsgRecvPacket for counterparty
+// MsgAcknowledgement
+func getMockMessages(mockSequence, mockSequenceCounterparty, lastSentMockMsgRecvCounterparty *uint64, lock *sync.Mutex) []mock.TransactionMessage {
+	lock.Lock()
+	defer lock.Unlock()
+	*mockSequence++
+	mockMessages := []mock.TransactionMessage{
+		{
+			Action:     processor.MsgTransfer,
+			PacketInfo: &chantypes.Packet{Sequence: *mockSequence},
+		},
+	}
+	if *mockSequenceCounterparty > 1 && *lastSentMockMsgRecvCounterparty != *mockSequenceCounterparty {
+		*lastSentMockMsgRecvCounterparty = *mockSequenceCounterparty
+		mockMessages = append(mockMessages, mock.TransactionMessage{
+			Action:     processor.MsgRecvPacket,
+			PacketInfo: &chantypes.Packet{Sequence: *mockSequenceCounterparty - 1},
+		})
+	}
+	if *mockSequence > 2 {
+		mockMessages = append(mockMessages, mock.TransactionMessage{
+			Action:     processor.MsgAcknowledgement,
+			PacketInfo: &chantypes.Packet{Sequence: *mockSequence - 2},
+		})
+	}
+	return mockMessages
+}