Repository: cosmos/relayer
Pull request #1103 (2023-02-21 04:53:01)

Flush command and PathProcessor periodic flush
Consolidates `rly tx relay-packets` and `rly tx relay-acknowledgements` into a single `rly tx flush` command.

Main reason for this: simplify the operator experience of flushing old messages. The `tx relay-packets` and `tx relay-acknowledgements` commands require both the specific `pathName` and the `channelID` as arguments. This means if you don't have them memorized, you have to reference the config and do channel queries to determine the args.

The new `flush` command still allows passing `pathName` and `channelID` arguments to isolate the scope of the flush, but it also allows passing the following number of args:
- one arg - only `pathName`. It will flush all channels allowed by the filter rule for the specific path.
 - no args. It will flush all paths in the config on the channels allowed by the filter rule for each path.

This means the most simple operator experience is now `rly tx flush` without any arguments, and all paths in the config will be flushed. This was the main use case I wanted to cover.

This new command utilizes the `PathProcessor` by queueing up pending messages that may have occurred before the start query height of the `ChainProcessor`s, then allowing the `PathProcessor` correlation framework to handle anything that needs to be relayed and then shut down.

This means that if a packet not yet been relayed, both the `MsgRecvPacket` to the destination chain and `MsgAcknowledgement` back to the source chain will be relayed in the flush. For packets that have been relayed to the destination chain with `MsgRecvPacket` but not yet acked back to the source chain, the `MsgAcknowledgement` will be relayed to the source chain during the flush.

The `PathProcessor` will terminate once all IBC message caches are clear. Messages are cleared from the caches when they are either successful, or given up on (after 5 retries). Additionally, the command is bounded by a 10 minute timeout.

Logs are printed for each channel that is flushed.

Since optimized flush capabilities were added to the `PathProcessor` for this, I added a periodic flush. A flush will happen when the relayer first starts up after both chains are in sync, and again on the flush interval (default of `5m`, configurable with the `--flush-interval`/`-i` flag. This reduces the need for the `--block-history` flag for the chain processors, and may be worth removing in a follow up PR.




Patch:
diff --git a/cmd/flags.go b/cmd/flags.go
index 54d7342..7f6e5eb 100644
--- a/cmd/flags.go
+++ b/cmd/flags.go
@@ -44,6 +44,7 @@ const (
 	flagReverse                 = "reverse"
 	flagProcessor               = "processor"
 	flagInitialBlockHistory     = "block-history"
+	flagFlushInterval           = "flush-interval"
 	flagMemo                    = "memo"
 	flagFilterRule              = "filter-rule"
 	flagFilterChannels          = "filter-channels"
@@ -355,6 +356,14 @@ func initBlockFlag(v *viper.Viper, cmd *cobra.Command) *cobra.Command {
 	return cmd
 }
 
+func flushIntervalFlag(v *viper.Viper, cmd *cobra.Command) *cobra.Command {
+	cmd.Flags().DurationP(flagFlushInterval, "i", relayer.DefaultFlushInterval, "how frequently should a flush routine be run")
+	if err := v.BindPFlag(flagFlushInterval, cmd.Flags().Lookup(flagFlushInterval)); err != nil {
+		panic(err)
+	}
+	return cmd
+}
+
 func memoFlag(v *viper.Viper, cmd *cobra.Command) *cobra.Command {
 	cmd.Flags().String(flagMemo, "", "a memo to include in relayed packets")
 	if err := v.BindPFlag(flagMemo, cmd.Flags().Lookup(flagMemo)); err != nil {
diff --git a/cmd/start.go b/cmd/start.go
index 8137eb2..e88dcd7 100644
--- a/cmd/start.go
+++ b/cmd/start.go
@@ -132,6 +132,11 @@ $ %s start demo-path2 --max-tx-size 10`, appName, appName, appName, appName)),
 				return err
 			}
 
+			flushInterval, err := cmd.Flags().GetDuration(flagFlushInterval)
+			if err != nil {
+				return err
+			}
+
 			rlyErrCh := relayer.StartRelayer(
 				cmd.Context(),
 				a.Log,
@@ -140,7 +145,10 @@ $ %s start demo-path2 --max-tx-size 10`, appName, appName, appName, appName)),
 				maxTxSize, maxMsgLength,
 				a.Config.memo(cmd),
 				clientUpdateThresholdTime,
-				processorType, initialBlockHistory,
+				flushInterval,
+				nil,
+				processorType,
+				initialBlockHistory,
 				prometheusMetrics,
 			)
 
@@ -163,6 +171,7 @@ $ %s start demo-path2 --max-tx-size 10`, appName, appName, appName, appName)),
 	cmd = debugServerFlags(a.Viper, cmd)
 	cmd = processorFlag(a.Viper, cmd)
 	cmd = initBlockFlag(a.Viper, cmd)
+	cmd = flushIntervalFlag(a.Viper, cmd)
 	cmd = memoFlag(a.Viper, cmd)
 	return cmd
 }
diff --git a/cmd/tx.go b/cmd/tx.go
index 8cf21b6..437edbc 100644
--- a/cmd/tx.go
+++ b/cmd/tx.go
@@ -1,6 +1,8 @@
 package cmd
 
 import (
+	"context"
+	"errors"
 	"fmt"
 	"strings"
 	"time"
@@ -9,11 +11,14 @@ import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	chantypes "github.com/cosmos/ibc-go/v7/modules/core/04-channel/types"
 	"github.com/cosmos/relayer/v2/relayer"
+	"github.com/cosmos/relayer/v2/relayer/processor"
 	"github.com/cosmos/relayer/v2/relayer/provider"
 	"github.com/spf13/cobra"
 	"go.uber.org/zap"
 )
 
+const flushTimeout = 10 * time.Minute
+
 // transactionCmd returns a parent transaction command handler, where all child
 // commands can submit transactions on IBC-connected networks.
 func transactionCmd(a *appState) *cobra.Command {
@@ -31,6 +36,7 @@ Most of these commands take a [path] argument. Make sure:
 	cmd.AddCommand(
 		linkCmd(a),
 		linkThenStartCmd(a),
+		flushCmd(a),
 		relayMsgsCmd(a),
 		relayAcksCmd(a),
 		xfersend(a),
@@ -719,24 +725,58 @@ $ %s tx link-then-start demo-path --timeout 5s`, appName, appName)),
 	return cmd
 }
 
-func relayMsgsCmd(a *appState) *cobra.Command {
+func flushCmd(a *appState) *cobra.Command {
 	cmd := &cobra.Command{
-		Use:     "relay-packets path_name src_channel_id",
+		Use:     "flush [path_name]? [src_channel_id]?",
 		Aliases: []string{"relay-pkts"},
-		Short:   "relay any remaining non-relayed packets on a given path, in both directions",
-		Args:    withUsage(cobra.ExactArgs(2)),
+		Short:   "flush any pending MsgRecvPacket and MsgAcknowledgement messages on a given path, in both directions",
+		Args:    withUsage(cobra.RangeArgs(0, 2)),
 		Example: strings.TrimSpace(fmt.Sprintf(`
-$ %s transact relay-packets demo-path channel-0
-$ %s tx relay-pkts demo-path channel-0`,
-			appName, appName,
+$ %s tx flush
+$ %s tx flush demo-path
+$ %s tx flush demo-path channel-0`,
+			appName, appName, appName,
 		)),
 		RunE: func(cmd *cobra.Command, args []string) error {
-			c, src, dst, err := a.Config.ChainsFromPath(args[0])
+			chains := make(map[string]*relayer.Chain)
+			var paths []relayer.NamedPath
+
+			if len(args) > 0 {
+				pathName := args[0]
+				path := a.Config.Paths.MustGet(pathName)
+				paths = append(paths, relayer.NamedPath{
+					Name: pathName,
+					Path: path,
+				})
+
+				// collect unique chain IDs
+				chains[path.Src.ChainID] = nil
+				chains[path.Dst.ChainID] = nil
+			} else {
+				for n, path := range a.Config.Paths {
+					paths = append(paths, relayer.NamedPath{
+						Name: n,
+						Path: path,
+					})
+
+					// collect unique chain IDs
+					chains[path.Src.ChainID] = nil
+					chains[path.Dst.ChainID] = nil
+				}
+			}
+
+			chainIDs := make([]string, 0, len(chains))
+			for chainID := range chains {
+				chainIDs = append(chainIDs, chainID)
+			}
+
+			// get chain configurations
+			chains, err := a.Config.Chains.Gets(chainIDs...)
 			if err != nil {
 				return err
 			}
 
-			if err = ensureKeysExist(c); err != nil {
+			if err := ensureKeysExist(chains); err != nil {
 				return err
 			}
 
@@ -745,18 +785,43 @@ $ %s tx relay-pkts demo-path channel-0`,
 				return err
 			}
 
-			channelID := args[1]
-			channel, err := relayer.QueryChannel(cmd.Context(), c[src], channelID)
-			if err != nil {
-				return err
+			if len(args) == 2 {
+				// Only allow specific channel
+				paths[0].Path.Filter = relayer.ChannelFilter{
+					Rule:        processor.RuleAllowList,
+					ChannelList: []string{args[1]},
+				}
 			}
 
-			sp := relayer.UnrelayedSequences(cmd.Context(), c[src], c[dst], channel)
-
-			if err = relayer.RelayPackets(cmd.Context(), a.Log, c[src], c[dst], sp, maxTxSize, maxMsgLength, a.Config.memo(cmd), channel); err != nil {
+			ctx, cancel := context.WithTimeout(cmd.Context(), flushTimeout)
+			defer cancel()
+
+			rlyErrCh := relayer.StartRelayer(
+				ctx,
+				a.Log,
+				chains,
+				paths,
+				maxTxSize, maxMsgLength,
+				a.Config.memo(cmd),
+				0,
+				0,
+				&processor.FlushLifecycle{},
+				relayer.ProcessorEvents,
+				0,
+				nil,
+			)
+
+			// Block until the error channel sends a message.
+			// The context being canceled will cause the relayer to stop,
+			// so we don't want to separately monitor the ctx.Done channel,
+			// because we would risk returning before the relayer cleans up.
+			if err := <-rlyErrCh; err != nil && !errors.Is(err, context.Canceled) {
+				a.Log.Warn(
+					"Relayer start error",
+					zap.Error(err),
+				)
 				return err
 			}
-
 			return nil
 		},
 	}
@@ -766,6 +831,28 @@ $ %s tx relay-pkts demo-path channel-0`,
 	return cmd
 }
 
+func relayMsgsCmd(a *appState) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:     "relay-packets path_name src_channel_id",
+		Aliases: []string{"relay-pkts"},
+		Short:   "relay any remaining non-relayed packets on a given path, in both directions",
+		Args:    withUsage(cobra.ExactArgs(2)),
+		Example: strings.TrimSpace(fmt.Sprintf(`
+$ %s transact relay-packets demo-path channel-0
+$ %s tx relay-pkts demo-path channel-0`,
+			appName, appName,
+		)),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			a.Log.Warn("This command is deprecated. Please use 'tx flush' command instead")
+			return flushCmd(a).RunE(cmd, args)
+		},
+	}
+
+	cmd = strategyFlag(a.Viper, cmd)
+	cmd = memoFlag(a.Viper, cmd)
+	return cmd
+}
+
 func relayAcksCmd(a *appState) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:     "relay-acknowledgements path_name src_channel_id",
@@ -778,35 +865,8 @@ $ %s tx relay-acks demo-path channel-0 -l 3 -s 6`,
 			appName, appName,
 		)),
 		RunE: func(cmd *cobra.Command, args []string) error {
-			c, src, dst, err := a.Config.ChainsFromPath(args[0])
-			if err != nil {
-				return err
-			}
-
-			if err = ensureKeysExist(c); err != nil {
-				return err
-			}
-
-			maxTxSize, maxMsgLength, err := GetStartOptions(cmd)
-			if err != nil {
-				return err
-			}
-
-			channelID := args[1]
-			channel, err := relayer.QueryChannel(cmd.Context(), c[src], channelID)
-			if err != nil {
-				return err
-			}
-
-			// sp.Src contains all sequences acked on SRC but acknowledgement not processed on DST
-			// sp.Dst contains all sequences acked on DST but acknowledgement not processed on SRC
-			sp := relayer.UnrelayedAcknowledgements(cmd.Context(), c[src], c[dst], channel)
-
-			if err = relayer.RelayAcknowledgements(cmd.Context(), a.Log, c[src], c[dst], sp, maxTxSize, maxMsgLength, a.Config.memo(cmd), channel); err != nil {
-				return err
-			}
-
-			return nil
+			a.Log.Warn("This command is deprecated. Please use 'tx flush' command instead")
+			return flushCmd(a).RunE(cmd, args)
 		},
 	}
 
diff --git a/relayer/channel.go b/relayer/channel.go
index 37055be..f5aa9dc 100644
--- a/relayer/channel.go
+++ b/relayer/channel.go
@@ -58,6 +58,7 @@ func (c *Chain) CreateOpenChannels(
 		nil,
 		memo,
 		DefaultClientUpdateThreshold,
+		DefaultFlushInterval,
 	)
 
 	c.log.Info("Starting event processor for channel handshake",
@@ -128,6 +129,7 @@ func (c *Chain) CloseChannel(
 			nil,
 			memo,
 			DefaultClientUpdateThreshold,
+			DefaultFlushInterval,
 		)).
 		WithInitialBlockHistory(0).
 		WithMessageLifecycle(&processor.ChannelMessageLifecycle{
diff --git a/relayer/connection.go b/relayer/connection.go
index 8d6f8b0..1f818ef 100644
--- a/relayer/connection.go
+++ b/relayer/connection.go
@@ -39,6 +39,7 @@ func (c *Chain) CreateOpenConnections(
 		nil,
 		memo,
 		DefaultClientUpdateThreshold,
+		DefaultFlushInterval,
 	)
 
 	var connectionSrc, connectionDst string
diff --git a/relayer/processor/event_processor.go b/relayer/processor/event_processor.go
index 825e7c9..1b577a3 100644
--- a/relayer/processor/event_processor.go
+++ b/relayer/processor/event_processor.go
@@ -72,6 +72,9 @@ func (ep EventProcessorBuilder) Build() EventProcessor {
 		}
 		chainProcessor.SetPathProcessors(pathProcessorsForThisChain)
 	}
+	for _, pathProcessor := range ep.pathProcessors {
+		pathProcessor.SetMessageLifecycle(ep.messageLifecycle)
+	}
 
 	return EventProcessor(ep)
 }
@@ -85,7 +88,7 @@ func (ep EventProcessor) Run(ctx context.Context) error {
 	for _, pathProcessor := range ep.pathProcessors {
 		pathProcessor := pathProcessor
 		eg.Go(func() error {
-			pathProcessor.Run(runCtx, runCtxCancel, ep.messageLifecycle)
+			pathProcessor.Run(runCtx, runCtxCancel)
 			return nil
 		})
 	}
diff --git a/relayer/processor/path_processor.go b/relayer/processor/path_processor.go
index c21af22..67cf8b2 100644
--- a/relayer/processor/path_processor.go
+++ b/relayer/processor/path_processor.go
@@ -60,6 +60,12 @@ type PathProcessor struct {
 
 	clientUpdateThresholdTime time.Duration
 
+	messageLifecycle MessageLifecycle
+
+	initialFlushComplete bool
+	flushTicker          *time.Ticker
+	flushInterval        time.Duration
+
 	// Signals to retry.
 	retryProcess chan struct{}
 
@@ -87,7 +93,12 @@ func NewPathProcessor(
 	metrics *PrometheusMetrics,
 	memo string,
 	clientUpdateThresholdTime time.Duration,
+	flushInterval time.Duration,
 ) *PathProcessor {
+	if flushInterval == 0 {
+		// "disable" periodic flushing by using a large value.
+		flushInterval = 200 * 24 * 365 * time.Hour
+	}
 	return &PathProcessor{
 		log:                       log,
 		pathEnd1:                  newPathEndRuntime(log, pathEnd1, metrics),
@@ -95,10 +106,15 @@ func NewPathProcessor(
 		retryProcess:              make(chan struct{}, 2),
 		memo:                      memo,
 		clientUpdateThresholdTime: clientUpdateThresholdTime,
+		flushInterval:             flushInterval,
 		metrics:                   metrics,
 	}
 }
 
+func (pp *PathProcessor) SetMessageLifecycle(messageLifecycle MessageLifecycle) {
+	pp.messageLifecycle = messageLifecycle
+}
+
 // TEST USE ONLY
 func (pp *PathProcessor) PathEnd1Messages(channelKey ChannelKey, message string) PacketSequenceCache {
 	return pp.pathEnd1.messageCache.PacketFlow[channelKey][message]
@@ -231,7 +247,7 @@ func (pp *PathProcessor) HandleNewData(chainID string, cacheData ChainProcessorC
 
 // processAvailableSignals will block if signals are not yet available, otherwise it will process one of the available signals.
 // It returns whether or not the pathProcessor should quit.
-func (pp *PathProcessor) processAvailableSignals(ctx context.Context, cancel func(), messageLifecycle MessageLifecycle) bool {
+func (pp *PathProcessor) processAvailableSignals(ctx context.Context, cancel func()) bool {
 	select {
 	case <-ctx.Done():
 		pp.log.Debug("Context done, quitting PathProcessor",
@@ -244,30 +260,36 @@ func (pp *PathProcessor) processAvailableSignals(ctx context.Context, cancel fun
 		return true
 	case d := <-pp.pathEnd1.incomingCacheData:
 		// we have new data from ChainProcessor for pathEnd1
-		pp.pathEnd1.mergeCacheData(ctx, cancel, d, pp.pathEnd2.info.ChainID, pp.pathEnd2.inSync, messageLifecycle, pp.pathEnd2)
+		pp.pathEnd1.mergeCacheData(ctx, cancel, d, pp.pathEnd2.info.ChainID, pp.pathEnd2.inSync, pp.messageLifecycle, pp.pathEnd2)
 
 	case d := <-pp.pathEnd2.incomingCacheData:
 		// we have new data from ChainProcessor for pathEnd2
-		pp.pathEnd2.mergeCacheData(ctx, cancel, d, pp.pathEnd1.info.ChainID, pp.pathEnd1.inSync, messageLifecycle, pp.pathEnd1)
+		pp.pathEnd2.mergeCacheData(ctx, cancel, d, pp.pathEnd1.info.ChainID, pp.pathEnd1.inSync, pp.messageLifecycle, pp.pathEnd1)
 
 	case <-pp.retryProcess:
 		// No new data to merge in, just retry handling.
+	case <-pp.flushTicker.C:
+		// Periodic flush to clear out any old packets
+		pp.flush(ctx)
 	}
 	return false
 }
 
 // Run executes the main path process.
-func (pp *PathProcessor) Run(ctx context.Context, cancel func(), messageLifecycle MessageLifecycle) {
+func (pp *PathProcessor) Run(ctx context.Context, cancel func()) {
 	var retryTimer *time.Timer
+
+	pp.flushTicker = time.NewTicker(pp.flushInterval)
+
 	for {
 		// block until we have any signals to process
-		if pp.processAvailableSignals(ctx, cancel, messageLifecycle) {
+		if pp.processAvailableSignals(ctx, cancel) {
 			return
 		}
 
 		for len(pp.pathEnd1.incomingCacheData) > 0 || len(pp.pathEnd2.incomingCacheData) > 0 || len(pp.retryProcess) > 0 {
 			// signals are available, so this will not need to block.
-			if pp.processAvailableSignals(ctx, cancel, messageLifecycle) {
+			if pp.processAvailableSignals(ctx, cancel) {
 				return
 			}
 		}
@@ -276,8 +298,16 @@ func (pp *PathProcessor) Run(ctx context.Context, cancel func(), messageLifecycl
 			continue
 		}
 
+		if !pp.initialFlushComplete {
+			pp.flush(ctx)
+			pp.initialFlushComplete = true
+		} else if pp.shouldTerminateForFlushComplete(ctx, cancel) {
+			cancel()
+			return
+		}
+
 		// process latest message cache state from both pathEnds
-		if err := pp.processLatestMessages(ctx, messageLifecycle); err != nil {
+		if err := pp.processLatestMessages(ctx); err != nil {
 			// in case of IBC message send errors, schedule retry after durationErrorRetry
 			if retryTimer != nil {
 				retryTimer.Stop()
diff --git a/relayer/processor/path_processor_internal.go b/relayer/processor/path_processor_internal.go
index 1c0d528..f588a44 100644
--- a/relayer/processor/path_processor_internal.go
+++ b/relayer/processor/path_processor_internal.go
@@ -5,6 +5,7 @@ import (
 	"context"
 	"errors"
 	"sort"
+	"sync"
 
 	conntypes "github.com/cosmos/ibc-go/v7/modules/core/03-connection/types"
 	chantypes "github.com/cosmos/ibc-go/v7/modules/core/04-channel/types"
@@ -433,12 +434,12 @@ func (pp *PathProcessor) updateClientTrustedState(src *pathEndRuntime, dst *path
 	}
 }
 
-func (pp *PathProcessor) appendInitialMessageIfNecessary(msg MessageLifecycle, pathEnd1Messages, pathEnd2Messages *pathEndMessages) {
-	if msg == nil || pp.sentInitialMsg {
+func (pp *PathProcessor) appendInitialMessageIfNecessary(pathEnd1Messages, pathEnd2Messages *pathEndMessages) {
+	if pp.messageLifecycle == nil || pp.sentInitialMsg {
 		return
 	}
 	pp.sentInitialMsg = true
-	switch m := msg.(type) {
+	switch m := pp.messageLifecycle.(type) {
 	case *PacketMessageLifecycle:
 		if m.Initial == nil {
 			return
@@ -506,7 +507,7 @@ func (pp *PathProcessor) appendInitialMessageIfNecessary(msg MessageLifecycle, p
 }
 
 // messages from both pathEnds are needed in order to determine what needs to be relayed for a single pathEnd
-func (pp *PathProcessor) processLatestMessages(ctx context.Context, messageLifecycle MessageLifecycle) error {
+func (pp *PathProcessor) processLatestMessages(ctx context.Context) error {
 	// Update trusted client state for both pathends
 	pp.updateClientTrustedState(pp.pathEnd1, pp.pathEnd2)
 	pp.updateClientTrustedState(pp.pathEnd2, pp.pathEnd1)
@@ -647,7 +648,7 @@ func (pp *PathProcessor) processLatestMessages(ctx context.Context, messageLifec
 		clientICQMessages:  pathEnd2ClientICQMessages,
 	}
 
-	pp.appendInitialMessageIfNecessary(messageLifecycle, &pathEnd1Messages, &pathEnd2Messages)
+	pp.appendInitialMessageIfNecessary(&pathEnd1Messages, &pathEnd2Messages)
 
 	// now assemble and send messages in parallel
 	// if sending messages fails to one pathEnd, we don't need to halt sending to the other pathEnd.
@@ -759,3 +760,214 @@ func (pp *PathProcessor) packetMessagesToSend(
 
 	return pathEnd1PacketMessages, pathEnd2PacketMessages, pathEnd1ChannelMessage, pathEnd2ChannelMessage
 }
+
+func queryPacketCommitments(
+	ctx context.Context,
+	pathEnd *pathEndRuntime,
+	k ChannelKey,
+	commitments map[ChannelKey][]uint64,
+	mu *sync.Mutex,
+) func() error {
+	return func() error {
+		pathEnd.log.Debug("Flushing", zap.String("channel", k.ChannelID), zap.String("port", k.PortID))
+
+		c, err := pathEnd.chainProvider.QueryPacketCommitments(ctx, pathEnd.latestBlock.Height, k.ChannelID, k.PortID)
+		if err != nil {
+			return err
+		}
+		mu.Lock()
+		defer mu.Unlock()
+		commitments[k] = make([]uint64, len(c.Commitments))
+		for i, p := range c.Commitments {
+			commitments[k][i] = p.Sequence
+		}
+		return nil
+	}
+}
+
+func queuePendingRecvAndAcks(
+	ctx context.Context,
+	src, dst *pathEndRuntime,
+	k ChannelKey,
+	seqs []uint64,
+	srcCache ChannelPacketMessagesCache,
+	dstCache ChannelPacketMessagesCache,
+	srcMu *sync.Mutex,
+	dstMu *sync.Mutex,
+) func() error {
+	return func() error {
+		if len(seqs) == 0 {
+			src.log.Debug("Nothing to flush", zap.String("channel", k.ChannelID), zap.String("port", k.PortID))
+			return nil
+		}
+
+		unrecv, err := dst.chainProvider.QueryUnreceivedPackets(ctx, dst.latestBlock.Height, k.CounterpartyChannelID, k.CounterpartyPortID, seqs)
+		if err != nil {
+			return err
+		}
+
+		if len(unrecv) > 0 {
+			src.log.Debug("Will flush MsgRecvPacket", zap.String("channel", k.ChannelID), zap.String("port", k.PortID), zap.Uint64s("sequences", unrecv))
+		} else {
+			src.log.Debug("No MsgRecvPacket to flush", zap.String("channel", k.ChannelID), zap.String("port", k.PortID))
+		}
+
+		for _, seq := range unrecv {
+			sendPacket, err := src.chainProvider.QuerySendPacket(ctx, k.ChannelID, k.PortID, seq)
+			if err != nil {
+				return err
+			}
+			srcMu.Lock()
+			if _, ok := srcCache[k]; !ok {
+				srcCache[k] = make(PacketMessagesCache)
+			}
+			if _, ok := srcCache[k][chantypes.EventTypeSendPacket]; !ok {
+				srcCache[k][chantypes.EventTypeSendPacket] = make(PacketSequenceCache)
+			}
+			srcCache[k][chantypes.EventTypeSendPacket][seq] = sendPacket
+			srcMu.Unlock()
+		}
+
+		var unacked []uint64
+
+	SeqLoop:
+		for _, seq := range seqs {
+			for _, unrecvSeq := range unrecv {
+				if seq == unrecvSeq {
+					continue SeqLoop
+				}
+			}
+			// does not exist in unrecv, so this is an ack that must be written
+			unacked = append(unacked, seq)
+		}
+
+		if len(unacked) > 0 {
+			src.log.Debug("Will flush MsgAcknowledgement", zap.String("channel", k.ChannelID), zap.String("port", k.PortID), zap.Uint64s("sequences", unrecv))
+		} else {
+			src.log.Debug("No MsgAcknowledgement to flush", zap.String("channel", k.ChannelID), zap.String("port", k.PortID))
+		}
+
+		for _, seq := range unacked {
+			recvPacket, err := dst.chainProvider.QueryRecvPacket(ctx, k.CounterpartyChannelID, k.CounterpartyPortID, seq)
+			if err != nil {
+				return err
+			}
+			srcMu.Lock()
+			if _, ok := srcCache[k]; !ok {
+				srcCache[k] = make(PacketMessagesCache)
+			}
+			if _, ok := srcCache[k][chantypes.EventTypeSendPacket]; !ok {
+				srcCache[k][chantypes.EventTypeSendPacket] = make(PacketSequenceCache)
+			}
+			srcCache[k][chantypes.EventTypeSendPacket][seq] = recvPacket
+			srcMu.Unlock()
+
+			dstMu.Lock()
+			if _, ok := dstCache[k]; !ok {
+				dstCache[k] = make(PacketMessagesCache)
+			}
+			if _, ok := dstCache[k][chantypes.EventTypeRecvPacket]; !ok {
+				dstCache[k][chantypes.EventTypeRecvPacket] = make(PacketSequenceCache)
+			}
+			if _, ok := dstCache[k][chantypes.EventTypeWriteAck]; !ok {
+				dstCache[k][chantypes.EventTypeWriteAck] = make(PacketSequenceCache)
+			}
+			dstCache[k][chantypes.EventTypeRecvPacket][seq] = recvPacket
+			dstCache[k][chantypes.EventTypeWriteAck][seq] = recvPacket
+			dstMu.Unlock()
+		}
+		return nil
+	}
+}
+
+// flush runs queries to relay any pending messages which may have been
+// in blocks before the height that the chain processors started querying.
+func (pp *PathProcessor) flush(ctx context.Context) {
+	var (
+		commitments1                   = make(map[ChannelKey][]uint64)
+		commitments2                   = make(map[ChannelKey][]uint64)
+		commitments1Mu, commitments2Mu sync.Mutex
+
+		pathEnd1Cache                    = NewIBCMessagesCache()
+		pathEnd2Cache                    = NewIBCMessagesCache()
+		pathEnd1CacheMu, pathEnd2CacheMu sync.Mutex
+	)
+
+	// Query remaining packet commitments on both chains
+	var eg errgroup.Group
+	for k := range pp.pathEnd1.channelStateCache {
+		eg.Go(queryPacketCommitments(ctx, pp.pathEnd1, k, commitments1, &commitments1Mu))
+	}
+	for k := range pp.pathEnd2.channelStateCache {
+		eg.Go(queryPacketCommitments(ctx, pp.pathEnd2, k, commitments2, &commitments2Mu))
+	}
+
+	if err := eg.Wait(); err != nil {
+		pp.log.Error("Failed to query packet commitments", zap.Error(err))
+	}
+
+	// From remaining packet commitments, determine if:
+	// 1. Packet commitment is on source, but MsgRecvPacket has not yet been relayed to destination
+	// 2. Packet commitment is on source, and MsgRecvPacket has been relayed to destination, but MsgAcknowledgement has not been written to source to clear the packet commitment.
+	// Based on above conditions, enqueue MsgRecvPacket and MsgAcknowledgement messages
+	for k, seqs := range commitments1 {
+		eg.Go(queuePendingRecvAndAcks(ctx, pp.pathEnd1, pp.pathEnd2, k, seqs, pathEnd1Cache.PacketFlow, pathEnd2Cache.PacketFlow, &pathEnd1CacheMu, &pathEnd2CacheMu))
+	}
+
+	for k, seqs := range commitments2 {
+		eg.Go(queuePendingRecvAndAcks(ctx, pp.pathEnd2, pp.pathEnd1, k, seqs, pathEnd2Cache.PacketFlow, pathEnd1Cache.PacketFlow, &pathEnd2CacheMu, &pathEnd1CacheMu))
+	}
+
+	if err := eg.Wait(); err != nil {
+		pp.log.Error("Failed to enqueue pending messages for flush", zap.Error(err))
+	}
+
+	pp.pathEnd1.mergeMessageCache(pathEnd1Cache, pp.pathEnd2.info.ChainID, pp.pathEnd2.inSync)
+	pp.pathEnd2.mergeMessageCache(pathEnd2Cache, pp.pathEnd1.info.ChainID, pp.pathEnd1.inSync)
+}
+
+// shouldTerminateForFlushComplete will determine if the relayer should exit
+// when FlushLifecycle is used. It will exit when all of the message caches are cleared.
+func (pp *PathProcessor) shouldTerminateForFlushComplete(
+	ctx context.Context, cancel func(),
+) bool {
+	if _, ok := pp.messageLifecycle.(*FlushLifecycle); !ok {
+		return false
+	}
+	for _, packetMessagesCache := range pp.pathEnd1.messageCache.PacketFlow {
+		for _, c := range packetMessagesCache {
+			if len(c) > 0 {
+				return false
+			}
+		}
+	}
+	for _, c := range pp.pathEnd1.messageCache.ChannelHandshake {
+		if len(c) > 0 {
+			return false
+		}
+	}
+	for _, c := range pp.pathEnd1.messageCache.ConnectionHandshake {
+		if len(c) > 0 {
+			return false
+		}
+	}
+	for _, packetMessagesCache := range pp.pathEnd2.messageCache.PacketFlow {
+		for _, c := range packetMessagesCache {
+			if len(c) > 0 {
+				return false
+			}
+		}
+	}
+	for _, c := range pp.pathEnd2.messageCache.ChannelHandshake {
+		if len(c) > 0 {
+			return false
+		}
+	}
+	for _, c := range pp.pathEnd2.messageCache.ConnectionHandshake {
+		if len(c) > 0 {
+			return false
+		}
+	}
+	pp.log.Info("Found termination condition for flush, all caches cleared")
+	return true
+}
diff --git a/relayer/processor/types.go b/relayer/processor/types.go
index fd5287c..ac2dced 100644
--- a/relayer/processor/types.go
+++ b/relayer/processor/types.go
@@ -18,6 +18,12 @@ type MessageLifecycle interface {
 	messageLifecycler() //noop
 }
 
+// Flush lifecycle informs the PathProcessor to terminate once
+// all pending messages have been flushed.
+type FlushLifecycle struct{}
+
+func (t *FlushLifecycle) messageLifecycler() {}
+
 type PacketMessage struct {
 	ChainID   string
 	EventType string
diff --git a/relayer/strategies.go b/relayer/strategies.go
index ab458e4..dd82beb 100644
--- a/relayer/strategies.go
+++ b/relayer/strategies.go
@@ -25,6 +25,7 @@ const (
 	ProcessorEvents              string = "events"
 	ProcessorLegacy                     = "legacy"
 	DefaultClientUpdateThreshold        = 0 * time.Millisecond
+	DefaultFlushInterval                = 5 * time.Minute
 )
 
 // StartRelayer starts the main relaying loop and returns a channel that will contain any control-flow related errors.
@@ -36,6 +37,8 @@ func StartRelayer(
 	maxTxSize, maxMsgLength uint64,
 	memo string,
 	clientUpdateThresholdTime time.Duration,
+	flushInterval time.Duration,
+	messageLifecycle processor.MessageLifecycle,
 	processorType string,
 	initialBlockHistory uint64,
 	metrics *processor.PrometheusMetrics,
@@ -70,7 +73,21 @@ func StartRelayer(
 			}
 		}
 
-		go relayerStartEventProcessor(ctx, log, chainProcessors, ePaths, initialBlockHistory, maxTxSize, maxMsgLength, memo, clientUpdateThresholdTime, errorChan, metrics)
+		go relayerStartEventProcessor(
+			ctx,
+			log,
+			chainProcessors,
+			ePaths,
+			initialBlockHistory,
+			maxTxSize,
+			maxMsgLength,
+			memo,
+			messageLifecycle,
+			clientUpdateThresholdTime,
+			flushInterval,
+			errorChan,
+			metrics,
+		)
 		return errorChan
 	case ProcessorLegacy:
 		if len(paths) != 1 {
@@ -115,7 +132,9 @@ func relayerStartEventProcessor(
 	maxTxSize,
 	maxMsgLength uint64,
 	memo string,
+	messageLifecycle processor.MessageLifecycle,
 	clientUpdateThresholdTime time.Duration,
+	flushInterval time.Duration,
 	errCh chan<- error,
 	metrics *processor.PrometheusMetrics,
 ) {
@@ -132,9 +151,14 @@ func relayerStartEventProcessor(
 				metrics,
 				memo,
 				clientUpdateThresholdTime,
+				flushInterval,
 			))
 	}
 
+	if messageLifecycle != nil {
+		epb = epb.WithMessageLifecycle(messageLifecycle)
+	}
+
 	ep := epb.
 		WithInitialBlockHistory(initialBlockHistory).
 		Build()


Test patch:
diff --git a/interchaintest/go.mod b/interchaintest/go.mod
index 0b8e570..17735fb 100644
--- a/interchaintest/go.mod
+++ b/interchaintest/go.mod
@@ -9,7 +9,7 @@ require (
 	github.com/docker/docker v20.10.19+incompatible
 	github.com/icza/dyno v0.0.0-20220812133438-f0b6f8a18845
 	github.com/moby/moby v20.10.22+incompatible
-	github.com/strangelove-ventures/interchaintest/v7 v7.0.0-20230201060741-e0e4e0f9283b
+	github.com/strangelove-ventures/interchaintest/v7 v7.0.0-20230221024505-c2607f279410
 	github.com/stretchr/testify v1.8.1
 	github.com/tendermint/tendermint v0.37.0-rc2
 	go.uber.org/zap v1.24.0
diff --git a/interchaintest/go.sum b/interchaintest/go.sum
index d80f80e..f35e99e 100644
--- a/interchaintest/go.sum
+++ b/interchaintest/go.sum
@@ -1368,6 +1368,8 @@ github.com/strangelove-ventures/go-subkey v1.0.7 h1:cOP/Lajg3uxV/tvspu0m6+0Cu+DJ
 github.com/strangelove-ventures/go-subkey v1.0.7/go.mod h1:E34izOIEm+sZ1YmYawYRquqBQWeZBjVB4pF7bMuhc1c=
 github.com/strangelove-ventures/interchaintest/v7 v7.0.0-20230201060741-e0e4e0f9283b h1:TPt4lyXM0XJUrgaO5ufBsAx9jbrNuwvSZvnIBL/5ui0=
 github.com/strangelove-ventures/interchaintest/v7 v7.0.0-20230201060741-e0e4e0f9283b/go.mod h1:xABocejOFzpH8VR9Se92KE02IqR3cLhzGnl4SHV36Qw=
+github.com/strangelove-ventures/interchaintest/v7 v7.0.0-20230221024505-c2607f279410 h1:tcrXeN0vQdH5nUOGlvgLHu4mxdB4VPVdEsM9GLeWRxs=
+github.com/strangelove-ventures/interchaintest/v7 v7.0.0-20230221024505-c2607f279410/go.mod h1:xABocejOFzpH8VR9Se92KE02IqR3cLhzGnl4SHV36Qw=
 github.com/streadway/amqp v0.0.0-20190404075320-75d898a42a94/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
 github.com/streadway/amqp v0.0.0-20190827072141-edfb9018d271/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
 github.com/streadway/handy v0.0.0-20190108123426-d5acb3125c2a/go.mod h1:qNTQ5P5JnDBl6z3cMAg/SywNDC5ABu5ApDIw6lUbRmI=
diff --git a/interchaintest/ibc_test.go b/interchaintest/ibc_test.go
index 642c021..6ee7221 100644
--- a/interchaintest/ibc_test.go
+++ b/interchaintest/ibc_test.go
@@ -85,7 +85,7 @@ func TestRelayerEventProcessor(t *testing.T) {
 
 	interchaintestConformance(t, relayerinterchaintest.NewRelayerFactory(relayerinterchaintest.RelayerConfig{
 		Processor:           relayer.ProcessorEvents,
-		InitialBlockHistory: 100,
+		InitialBlockHistory: 0,
 	}))
 }
 
diff --git a/interchaintest/relayer.go b/interchaintest/relayer.go
index cb93ca9..7c5a0e8 100644
--- a/interchaintest/relayer.go
+++ b/interchaintest/relayer.go
@@ -316,16 +316,15 @@ func (r *Relayer) Exec(ctx context.Context, _ ibc.RelayerExecReporter, cmd, env
 	}
 }
 
-func (r *Relayer) FlushAcknowledgements(ctx context.Context, _ ibc.RelayerExecReporter, pathName string, channelID string) error {
-	res := r.sys().RunC(ctx, r.log(), "tx", "relay-acks", pathName, channelID)
-	if res.Err != nil {
-		return res.Err
+func (r *Relayer) Flush(ctx context.Context, _ ibc.RelayerExecReporter, pathName string, channelID string) error {
+	cmd := []string{"tx", "flush"}
+	if pathName != "" {
+		cmd = append(cmd, pathName)
+		if channelID != "" {
+			cmd = append(cmd, channelID)
+		}
 	}
-	return nil
-}
-
-func (r *Relayer) FlushPackets(ctx context.Context, _ ibc.RelayerExecReporter, pathName string, channelID string) error {
-	res := r.sys().RunC(ctx, r.log(), "tx", "relay-pkts", pathName, channelID)
+	res := r.sys().RunC(ctx, r.log(), cmd...)
 	if res.Err != nil {
 		return res.Err
 	}
diff --git a/relayer/chains/mock/mock_chain_processor_test.go b/relayer/chains/mock/mock_chain_processor_test.go
index 67d2a72..505fa87 100644
--- a/relayer/chains/mock/mock_chain_processor_test.go
+++ b/relayer/chains/mock/mock_chain_processor_test.go
@@ -59,8 +59,9 @@ func TestMockChainAndPathProcessors(t *testing.T) {
 	metrics := processor.NewPrometheusMetrics()
 
 	clientUpdateThresholdTime := 6 * time.Hour
+	flushInterval := 6 * time.Hour
 
-	pathProcessor := processor.NewPathProcessor(log, pathEnd1, pathEnd2, metrics, "", clientUpdateThresholdTime)
+	pathProcessor := processor.NewPathProcessor(log, pathEnd1, pathEnd2, metrics, "", clientUpdateThresholdTime, flushInterval)
 
 	eventProcessor := processor.NewEventProcessor().
 		WithChainProcessors(