Merge PR #98: Refactor relayer loops and prep for packet ack

Packet Acknowledgements

cmd/query.go

@@ -634,12 +634,12 @@ func queryUnrelayed() *cobra.Command {
 				return err
 			}
 
-			hs, err := relayer.UpdatesWithHeaders(c[src], c[dst])
+			sh, err := relayer.NewSyncHeaders(c[src], c[dst])
 			if err != nil {
 				return err
 			}
 
-			sp, err := relayer.UnrelayedSequences(c[src], c[dst], hs[src].Height, hs[dst].Height)
+			sp, err := relayer.UnrelayedSequences(c[src], c[dst], sh)
 			if err != nil {
 				return err
 			}

cmd/tx.go

@@ -188,7 +188,17 @@ func relayMsgsCmd() *cobra.Command {
 				return err
 			}
 
-			if err = relayer.RelayUnRelayedPacketsOrderedChan(c[src], c[dst]); err != nil {
+			sh, err := relayer.NewSyncHeaders(c[src], c[dst])
+			if err != nil {
+				return err
+			}
+
+			sp, err := relayer.UnrelayedSequences(c[src], c[dst], sh)
+			if err != nil {
+				return err
+			}
+
+			if err = relayer.RelayPacketsOrderedChan(c[src], c[dst], sh, sp); err != nil {
 				return err
 			}
 

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/btcsuite/btcd v0.0.0-20190807005414-4063feeff79a // indirect
 	github.com/cenkalti/backoff/v3 v3.2.2 // indirect
 	github.com/containerd/continuity v0.0.0-20200228182428-0f16d7a0959c // indirect
-	github.com/cosmos/cosmos-sdk v0.34.4-0.20200417201027-11528d39594c
+	github.com/cosmos/cosmos-sdk v0.34.4-0.20200419154345-84774907316c
 	github.com/cosmos/go-bip39 v0.0.0-20180819234021-555e2067c45d
 	github.com/gorilla/mux v1.7.4
 	github.com/ory/dockertest/v3 v3.5.5

go.sum

@@ -92,6 +92,8 @@ github.com/coreos/pkg v0.0.0-20160727233714-3ac0863d7acf/go.mod h1:E3G3o1h8I7cfc
 github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
 github.com/cosmos/cosmos-sdk v0.34.4-0.20200417201027-11528d39594c h1:Yd7DVqfImC4YZi6oRtGaTMaufgzdaYgD2ydkMhz3/qQ=
 github.com/cosmos/cosmos-sdk v0.34.4-0.20200417201027-11528d39594c/go.mod h1:jngw1LJfwEAU+xc/tZwUGIXBAJHapckYGtXycsq8D+U=
+github.com/cosmos/cosmos-sdk v0.34.4-0.20200419154345-84774907316c h1:QZgeN76JbDNEOko3EoI8FeN2IwPS7dPiap18E7O5Q+U=
+github.com/cosmos/cosmos-sdk v0.34.4-0.20200419154345-84774907316c/go.mod h1:jngw1LJfwEAU+xc/tZwUGIXBAJHapckYGtXycsq8D+U=
 github.com/cosmos/go-bip39 v0.0.0-20180819234021-555e2067c45d h1:49RLWk1j44Xu4fjHb6JFYmeUnDORVwHNkDxaQ0ctCVU=
 github.com/cosmos/go-bip39 v0.0.0-20180819234021-555e2067c45d/go.mod h1:tSxLoYXyBmiFeKpvmq4dzayMdCjCnu8uqmCysIGBT2Y=
 github.com/cosmos/ledger-cosmos-go v0.11.1 h1:9JIYsGnXP613pb2vPjFeMMjBI5lEDsEaF6oYorTy6J4=

relayer/channel-tx.go

@@ -19,6 +19,7 @@ func (src *Chain) CreateChannel(dst *Chain, ordered bool, to time.Duration) erro
 	}
 
 	ticker := time.NewTicker(to)
+	failures := 0
 	for ; true; <-ticker.C {
 		chanSteps, err := src.CreateChannelStep(dst, order)
 		if err != nil {
@@ -29,7 +30,12 @@ func (src *Chain) CreateChannel(dst *Chain, ordered bool, to time.Duration) erro
 			break
 		}
 
-		if chanSteps.Send(src, dst); chanSteps.success && chanSteps.last {
+		chanSteps.Send(src, dst)
+
+		switch {
+		// In the case of success and this being the last transaction
+		// debug logging, log created connection and break
+		case chanSteps.success && chanSteps.last:
 			chans, err := QueryChannelPair(src, dst, 0, 0)
 			if err != nil {
 				return err
@@ -41,6 +47,19 @@ func (src *Chain) CreateChannel(dst *Chain, ordered bool, to time.Duration) erro
 				src.ChainID, src.PathEnd.ChannelID, src.PathEnd.PortID,
 				dst.ChainID, dst.PathEnd.ChannelID, dst.PathEnd.PortID))
 			break
+		// In the case of success, reset the failures counter
+		case chanSteps.success:
+			failures = 0
+			continue
+		// In the case of failure, increment the failures counter and exit if this is the 3rd failure
+		case !chanSteps.success:
+			failures++
+			if failures > 2 {
+				src.Error(fmt.Errorf("! Channel failed: [%s]chan{%s}port{%s} -> [%s]chan{%s}port{%s}",
+					src.ChainID, src.PathEnd.ChannelID, src.PathEnd.PortID,
+					dst.ChainID, dst.PathEnd.ChannelID, dst.PathEnd.PortID))
+				break
+			}
 		}
 	}
 

relayer/connection-tx.go

@@ -12,6 +12,7 @@ import (
 // TODO: add max retries or something to this function
 func (src *Chain) CreateConnection(dst *Chain, to time.Duration) error {
 	ticker := time.NewTicker(to)
+	failed := 0
 	for ; true; <-ticker.C {
 		connSteps, err := src.CreateConnectionStep(dst)
 		if err != nil {
@@ -22,20 +23,37 @@ func (src *Chain) CreateConnection(dst *Chain, to time.Duration) error {
 			break
 		}
 
-		if connSteps.Send(src, dst); connSteps.success && connSteps.last {
-			conns, err := QueryConnectionPair(src, dst, 0, 0)
-			if err != nil {
-				return err
-			}
+		connSteps.Send(src, dst)
 
+		switch {
+		// In the case of success and this being the last transaction
+		// debug logging, log created connection and break
+		case connSteps.success && connSteps.last:
 			if src.debug {
+				conns, err := QueryConnectionPair(src, dst, 0, 0)
+				if err != nil {
+					return err
+				}
 				logConnectionStates(src, dst, conns)
 			}
 
 			src.Log(fmt.Sprintf("★ Connection created: [%s]client{%s}conn{%s} -> [%s]client{%s}conn{%s}",
 				src.ChainID, src.PathEnd.ClientID, src.PathEnd.ConnectionID,
 				dst.ChainID, dst.PathEnd.ClientID, dst.PathEnd.ConnectionID))
 			break
+		// In the case of success, reset the failures counter
+		case connSteps.success:
+			failed = 0
+			continue
+		// In the case of failure, increment the failures counter and exit if this is the 3rd failure
+		case !connSteps.success:
+			failed++
+			if failed > 2 {
+				src.Error(fmt.Errorf("! Connection failed: [%s]client{%s}conn{%s} -> [%s]client{%s}conn{%s}",
+					src.ChainID, src.PathEnd.ClientID, src.PathEnd.ConnectionID,
+					dst.ChainID, dst.PathEnd.ClientID, dst.PathEnd.ConnectionID))
+				break
+			}
 		}
 	}
 

relayer/contextual.go

@@ -1,14 +1,10 @@
 package relayer
 
 import (
-	"sync"
-
 	"github.com/cosmos/cosmos-sdk/codec"
 	stdcodec "github.com/cosmos/cosmos-sdk/codec/std"
 )
 
-var globalMutex sync.Mutex
-
 type contextualStdCodec struct {
 	*stdcodec.Codec
 	useContext func() func()

relayer/headers.go

@@ -0,0 +1,51 @@
+package relayer
+
+import (
+	"sync"
+
+	tmclient "github.com/cosmos/cosmos-sdk/x/ibc/07-tendermint/types"
+)
+
+// NewSyncHeaders returns a new instance of map[string]*tmclient.Header that can be easily
+// kept "reasonably up to date"
+func NewSyncHeaders(chains ...*Chain) (*SyncHeaders, error) {
+	mp, err := UpdatesWithHeaders(chains...)
+	if err != nil {
+		return nil, err
+	}
+	return &SyncHeaders{hds: mp}, nil
+}
+
+// SyncHeaders is an instance of map[string]*tmclient.Header
+// that can be kept "reasonably up to date" using it's Update method
+type SyncHeaders struct {
+	sync.Mutex
+
+	hds map[string]*tmclient.Header
+}
+
+// Update the header for a given chain
+func (uh *SyncHeaders) Update(c *Chain) error {
+	hd, err := c.UpdateLiteWithHeader()
+	if err != nil {
+		return err
+	}
+	uh.Lock()
+	defer uh.Unlock()
+	uh.hds[c.ChainID] = hd
+	return nil
+}
+
+// GetHeader returns the latest header for a given chainID
+func (uh *SyncHeaders) GetHeader(chainID string) *tmclient.Header {
+	uh.Lock()
+	defer uh.Unlock()
+	return uh.hds[chainID]
+}
+
+// GetHeight returns the latest height for a given chainID
+func (uh *SyncHeaders) GetHeight(chainID string) uint64 {
+	uh.Lock()
+	defer uh.Unlock()
+	return uh.hds[chainID].GetHeight()
+}

relayer/log-tx.go

@@ -2,6 +2,8 @@ package relayer
 
 import (
 	"fmt"
+	"strconv"
+	"strings"
 
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	connTypes "github.com/cosmos/cosmos-sdk/x/ibc/03-connection/types"
@@ -75,8 +77,24 @@ func (c *Chain) logCreateClient(dst *Chain, dstH uint64) {
 func (c *Chain) logTx(events map[string][]string) {
 	c.Log(fmt.Sprintf("• [%s]@{%d} - actions(%s) hash(%s)",
 		c.ChainID,
-		getEventHeight(events),
-		actions(events["message.action"]),
+		getTxEventHeight(events),
+		getTxActions(events["message.action"]),
 		events["tx.hash"][0]),
 	)
 }
+
+func getTxEventHeight(events map[string][]string) int64 {
+	if val, ok := events["tx.height"]; ok {
+		out, _ := strconv.ParseInt(val[0], 10, 64)
+		return out
+	}
+	return -1
+}
+
+func getTxActions(act []string) string {
+	out := ""
+	for i, a := range act {
+		out += fmt.Sprintf("%d:%s,", i, a)
+	}
+	return strings.TrimSuffix(out, ",")
+}