Repository: cosmos/relayer
Pull request #667 (2022-04-04 15:50:28)

Add return type to (*RelayMsgs).Send()
This adds a new return type indicating the accumulation of errors
encountered during message sending and the count of successful batches
sent. The return value is not yet used anywhere and existing behavior is
preserved so far.

There are two questionable existing behaviors:

1. On the first part of the batching, RelayMsgs.Success is &&-ed with
   the newly received success, and when sending the "leftover" messages,
   the Success field is overwritten to the final value. This means we
   could report success if all early batches failed but only the last
   batch to the destination chain succeeded.

   I intend to address this in a following change by adding an
   equivalent Success() method to the SendMsgsResult type which properly
   reports if all sent batches succeeded.

   I am not sure how this will differ from existing behavior in the
   wild. I assume we will see more failures than before.

2. It is unclear to me, when there are multiple batches to be sent, and
   one batch fails, is it safe to send a following batch, or should the
   entire send operation abort? I don't yet have a thorough
   understanding of what will be sent here to judge for myself which is
   more appropriate.

/cc @jackzampolin and @jtieri for those behavior questions.

mark-rushakoff: I'm going to merge this PR now since it doesn't change any existing behavior, but still +1 to hear from @jackzampolin on the questions.

Patch:
diff --git a/relayer/relayMsgs.go b/relayer/relayMsgs.go
index c0259df..74e97f1 100644
--- a/relayer/relayMsgs.go
+++ b/relayer/relayMsgs.go
@@ -5,6 +5,7 @@ import (
 
 	"github.com/cosmos/relayer/v2/relayer/provider"
 	"github.com/cosmos/relayer/v2/relayer/provider/cosmos"
+	"go.uber.org/multierr"
 	"go.uber.org/zap"
 )
 
@@ -107,11 +108,25 @@ func AsRelayMsgSender(c *Chain) RelayMsgSender {
 	}
 }
 
-func (r *RelayMsgs) Send(ctx context.Context, log *zap.Logger, src, dst RelayMsgSender) {
+type SendMsgsResult struct {
+	// Count of successfully sent batches,
+	// where "successful" means there was no error in sending the batch across the network,
+	// and the remote end sent a response indicating success.
+	SuccessfulSrcBatches, SuccessfulDstBatches int
+
+	// Accumulation of errors encountered when sending to source or destination.
+	// If multiple errors occurred, these will be multierr errors
+	// which are displayed nicely through zap logging.
+	SrcSendError, DstSendError error
+}
+
+func (r *RelayMsgs) Send(ctx context.Context, log *zap.Logger, src, dst RelayMsgSender) SendMsgsResult {
 	//nolint:prealloc // can not be pre allocated
 	var (
 		msgLen, txSize uint64
 		msgs           []provider.RelayerMessage
+
+		result SendMsgsResult
 	)
 
 	r.Succeeded = true
@@ -129,10 +144,15 @@ func (r *RelayMsgs) Send(ctx context.Context, log *zap.Logger, src, dst RelayMsg
 
 			if r.IsMaxTx(msgLen, txSize) {
 				// Submit the transactions to src chain and update its status
-				res, success, err := src.SendMessages(ctx, msgs)
+				resp, success, err := src.SendMessages(ctx, msgs)
 				if err != nil {
-					logFailedTx(log, src.ChainID, res, err, msgs)
+					logFailedTx(log, src.ChainID, resp, err, msgs)
+					multierr.AppendInto(&result.SrcSendError, err)
+				}
+				if success {
+					result.SuccessfulSrcBatches++
 				}
+
 				r.Succeeded = r.Succeeded && success
 
 				// clear the current batch and reset variables
@@ -145,9 +165,13 @@ func (r *RelayMsgs) Send(ctx context.Context, log *zap.Logger, src, dst RelayMsg
 
 	// submit leftover msgs
 	if len(msgs) > 0 {
-		res, success, err := src.SendMessages(ctx, msgs)
+		resp, success, err := src.SendMessages(ctx, msgs)
 		if err != nil {
-			logFailedTx(log, src.ChainID, res, err, msgs)
+			logFailedTx(log, src.ChainID, resp, err, msgs)
+			multierr.AppendInto(&result.SrcSendError, err)
+		}
+		if success {
+			result.SuccessfulSrcBatches++
 		}
 
 		r.Succeeded = success
@@ -169,9 +193,13 @@ func (r *RelayMsgs) Send(ctx context.Context, log *zap.Logger, src, dst RelayMsg
 
 			if r.IsMaxTx(msgLen, txSize) {
 				// Submit the transaction to dst chain and update its status
-				res, success, err := dst.SendMessages(ctx, msgs)
+				resp, success, err := dst.SendMessages(ctx, msgs)
 				if err != nil {
-					logFailedTx(log, dst.ChainID, res, err, msgs)
+					logFailedTx(log, dst.ChainID, resp, err, msgs)
+					multierr.AppendInto(&result.DstSendError, err)
+				}
+				if success {
+					result.SuccessfulDstBatches++
 				}
 
 				r.Succeeded = r.Succeeded && success
@@ -186,11 +214,17 @@ func (r *RelayMsgs) Send(ctx context.Context, log *zap.Logger, src, dst RelayMsg
 
 	// submit leftover msgs
 	if len(msgs) > 0 {
-		res, success, err := dst.SendMessages(ctx, msgs)
+		resp, success, err := dst.SendMessages(ctx, msgs)
 		if err != nil {
-			logFailedTx(log, dst.ChainID, res, err, msgs)
+			logFailedTx(log, dst.ChainID, resp, err, msgs)
+			multierr.AppendInto(&result.DstSendError, err)
+		}
+		if success {
+			result.SuccessfulDstBatches++
 		}
 
 		r.Succeeded = success
 	}
+
+	return result
 }


Test patch:
diff --git a/relayer/relaymsgs_test.go b/relayer/relaymsgs_test.go
index e635934..1739c8a 100644
--- a/relayer/relaymsgs_test.go
+++ b/relayer/relaymsgs_test.go
@@ -2,6 +2,7 @@ package relayer_test
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/cosmos/relayer/v2/relayer"
@@ -55,7 +56,7 @@ func (m fakeRelayerMessage) MsgBytes() ([]byte, error) {
 	return []byte(m.b), nil
 }
 
-func TestRelayMsgs_Send(t *testing.T) {
+func TestRelayMsgs_Send_Success(t *testing.T) {
 	// Fixtures for test.
 	// src appends to srcSent and dst appends to dstSent.
 	var srcSent []provider.RelayerMessage
@@ -63,7 +64,7 @@ func TestRelayMsgs_Send(t *testing.T) {
 		ChainID: "src",
 		SendMessages: func(ctx context.Context, msgs []provider.RelayerMessage) (*provider.RelayerTxResponse, bool, error) {
 			srcSent = append(srcSent, msgs...)
-			return nil, false, nil
+			return nil, true, nil
 		},
 	}
 
@@ -72,7 +73,7 @@ func TestRelayMsgs_Send(t *testing.T) {
 		ChainID: "dst",
 		SendMessages: func(ctx context.Context, msgs []provider.RelayerMessage) (*provider.RelayerTxResponse, bool, error) {
 			dstSent = append(dstSent, msgs...)
-			return nil, false, nil
+			return nil, true, nil
 		},
 	}
 
@@ -89,7 +90,11 @@ func TestRelayMsgs_Send(t *testing.T) {
 			Dst: []provider.RelayerMessage{dstMsg},
 		}
 
-		rm.Send(context.Background(), zaptest.NewLogger(t), src, dst)
+		result := rm.Send(context.Background(), zaptest.NewLogger(t), src, dst)
+		require.Equal(t, result.SuccessfulSrcBatches, 1)
+		require.Equal(t, result.SuccessfulDstBatches, 1)
+		require.NoError(t, result.SrcSendError)
+		require.NoError(t, result.DstSendError)
 
 		require.Equal(t, []provider.RelayerMessage{srcMsg}, srcSent)
 		require.Equal(t, []provider.RelayerMessage{dstMsg}, dstSent)
@@ -107,7 +112,11 @@ func TestRelayMsgs_Send(t *testing.T) {
 			MaxMsgLength: 2,
 		}
 
-		rm.Send(context.Background(), zaptest.NewLogger(t), src, dst)
+		result := rm.Send(context.Background(), zaptest.NewLogger(t), src, dst)
+		require.Equal(t, result.SuccessfulSrcBatches, 2)
+		require.Equal(t, result.SuccessfulDstBatches, 2)
+		require.NoError(t, result.SrcSendError)
+		require.NoError(t, result.DstSendError)
 
 		require.Equal(t, []provider.RelayerMessage{srcMsg, srcMsg, srcMsg}, srcSent)
 		require.Equal(t, []provider.RelayerMessage{dstMsg, dstMsg, dstMsg}, dstSent)
@@ -125,9 +134,153 @@ func TestRelayMsgs_Send(t *testing.T) {
 			MaxMsgLength: 2,
 		}
 
-		rm.Send(context.Background(), zaptest.NewLogger(t), src, dst)
+		result := rm.Send(context.Background(), zaptest.NewLogger(t), src, dst)
+		require.Equal(t, result.SuccessfulSrcBatches, 2)
+		require.Equal(t, result.SuccessfulDstBatches, 2)
+		require.NoError(t, result.SrcSendError)
+		require.NoError(t, result.DstSendError)
 
 		require.Equal(t, []provider.RelayerMessage{srcMsg, srcMsg, srcMsg}, srcSent)
 		require.Equal(t, []provider.RelayerMessage{dstMsg, dstMsg, dstMsg}, dstSent)
 	})
 }
+
+func TestRelayMsgs_Send_Errors(t *testing.T) {
+	t.Run("one batch and one error", func(t *testing.T) {
+		srcErr := fmt.Errorf("source error")
+		src := relayer.RelayMsgSender{
+			ChainID: "src",
+			SendMessages: func(ctx context.Context, msgs []provider.RelayerMessage) (*provider.RelayerTxResponse, bool, error) {
+				return nil, false, srcErr
+			},
+		}
+
+		dstErr := fmt.Errorf("dest error")
+		dst := relayer.RelayMsgSender{
+			ChainID: "dst",
+			SendMessages: func(ctx context.Context, msgs []provider.RelayerMessage) (*provider.RelayerTxResponse, bool, error) {
+				return nil, false, dstErr
+			},
+		}
+
+		srcMsg := fakeRelayerMessage{t: "srctype", b: "srcdata"}
+		dstMsg := fakeRelayerMessage{t: "dsttype", b: "dstdata"}
+
+		rm := relayer.RelayMsgs{
+			Src: []provider.RelayerMessage{srcMsg},
+			Dst: []provider.RelayerMessage{dstMsg},
+		}
+
+		result := rm.Send(context.Background(), zaptest.NewLogger(t), src, dst)
+		require.Equal(t, result.SuccessfulSrcBatches, 0)
+		require.Equal(t, result.SuccessfulDstBatches, 0)
+		require.ErrorIs(t, result.SrcSendError, srcErr)
+		require.ErrorIs(t, result.DstSendError, dstErr)
+	})
+
+	t.Run("multiple batches and all errors", func(t *testing.T) {
+		srcErr1, srcErr2 := fmt.Errorf("source error 1"), fmt.Errorf("source error 2")
+		var srcCalls int
+		src := relayer.RelayMsgSender{
+			ChainID: "src",
+			SendMessages: func(ctx context.Context, msgs []provider.RelayerMessage) (*provider.RelayerTxResponse, bool, error) {
+				srcCalls++
+				switch srcCalls {
+				case 1:
+					return nil, false, srcErr1
+				case 2:
+					return nil, false, srcErr2
+				default:
+					panic("src.SendMessages called too many times")
+				}
+			},
+		}
+
+		dstErr1, dstErr2 := fmt.Errorf("dest error 1"), fmt.Errorf("dest error 2")
+		var dstCalls int
+		dst := relayer.RelayMsgSender{
+			ChainID: "dst",
+			SendMessages: func(ctx context.Context, msgs []provider.RelayerMessage) (*provider.RelayerTxResponse, bool, error) {
+				dstCalls++
+				switch dstCalls {
+				case 1:
+					return nil, false, dstErr1
+				case 2:
+					return nil, false, dstErr2
+				default:
+					panic("dst.SendMessages called too many times")
+				}
+			},
+		}
+
+		srcMsg := fakeRelayerMessage{t: "srctype", b: "srcdata"}
+		dstMsg := fakeRelayerMessage{t: "dsttype", b: "dstdata"}
+
+		rm := relayer.RelayMsgs{
+			Src: []provider.RelayerMessage{srcMsg, srcMsg, srcMsg},
+			Dst: []provider.RelayerMessage{dstMsg, dstMsg, dstMsg},
+
+			MaxMsgLength: 2,
+		}
+
+		result := rm.Send(context.Background(), zaptest.NewLogger(t), src, dst)
+		require.Equal(t, result.SuccessfulSrcBatches, 0)
+		require.Equal(t, result.SuccessfulDstBatches, 0)
+		require.ErrorIs(t, result.SrcSendError, srcErr1)
+		require.ErrorIs(t, result.SrcSendError, srcErr2)
+		require.ErrorIs(t, result.DstSendError, dstErr1)
+		require.ErrorIs(t, result.DstSendError, dstErr2)
+	})
+
+	t.Run("two batches with success then error", func(t *testing.T) {
+		srcErr := fmt.Errorf("source error")
+		var srcCalls int
+		src := relayer.RelayMsgSender{
+			ChainID: "src",
+			SendMessages: func(ctx context.Context, msgs []provider.RelayerMessage) (*provider.RelayerTxResponse, bool, error) {
+				srcCalls++
+				switch srcCalls {
+				case 1:
+					return nil, true, nil
+				case 2:
+					return nil, false, srcErr
+				default:
+					panic("src.SendMessages called too many times")
+				}
+			},
+		}
+
+		dstErr := fmt.Errorf("dest error")
+		var dstCalls int
+		dst := relayer.RelayMsgSender{
+			ChainID: "dst",
+			SendMessages: func(ctx context.Context, msgs []provider.RelayerMessage) (*provider.RelayerTxResponse, bool, error) {
+				dstCalls++
+				switch dstCalls {
+				case 1:
+					return nil, true, nil
+				case 2:
+					return nil, false, dstErr
+				default:
+					panic("dst.SendMessages called too many times")
+				}
+			},
+		}
+
+		srcMsg := fakeRelayerMessage{t: "srctype", b: "srcdata"}
+		dstMsg := fakeRelayerMessage{t: "dsttype", b: "dstdata"}
+
+		rm := relayer.RelayMsgs{
+			Src: []provider.RelayerMessage{srcMsg, srcMsg, srcMsg},
+			Dst: []provider.RelayerMessage{dstMsg, dstMsg, dstMsg},
+
+			MaxMsgLength: 2,
+		}
+
+		result := rm.Send(context.Background(), zaptest.NewLogger(t), src, dst)
+		require.Equal(t, result.SuccessfulSrcBatches, 1)
+		require.Equal(t, result.SuccessfulDstBatches, 1)
+		require.ErrorIs(t, result.SrcSendError, srcErr)
+		require.ErrorIs(t, result.DstSendError, dstErr)
+	})
+}