From 40c694962577254af81dbb8954accd1098296709 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?colin=20axn=C3=A9r?= <25233464+colin-axner@users.noreply.github.com> Date: Tue, 9 Mar 2021 15:55:46 +0100 Subject: [PATCH] refactor state based relaying (fixes update client bug on retries) (#435) * bump sdk version to v0.41.3 (#430) * bump sdk version * bump SDK to v0.41.3 * inital work for refactoring state based relaying * Modify relayPacketFromSequence * update tendermint client to not prune light blocks (#437) * Address comments and fix lint issues * Fix lint issues * Remove onRtyErr (lint issue) * typo fix (#438) * disable tm pruning (#441) * update release naming (#442) * Implement swagger docs and fix path validation (#434) * Add swagger setup * Add some routes docs and swagger ui * Add few more route docs * Add swagger docs for remaining routes * Fix golint issues * Fix unused lint issues * check chain-id in AddChain * add a light client database lock (#447) Add a lock to prevent multiple processes from attempting to access the light client database at the same time. This typically resulted in unnecessary errors or even panics * Close database connection even if second error triggers (#449) Co-authored-by: Mark * address comments Co-authored-by: akhilkumarpilli Co-authored-by: Afanti Co-authored-by: Akhil Kumar P <36399231+akhilkumarpilli@users.noreply.github.com> Co-authored-by: Mark | Microtick <409166+mark-microtick@users.noreply.github.com> Co-authored-by: Mark --- relayer/msgs.go | 193 ++++++++++++++++++++++++++++++++++++++ relayer/naive-strategy.go | 110 +++++++++------------- 2 files changed, 239 insertions(+), 64 deletions(-) diff --git a/relayer/msgs.go b/relayer/msgs.go index cd2c61274..a6a7297bd 100644 --- a/relayer/msgs.go +++ b/relayer/msgs.go @@ -1,6 +1,9 @@ package relayer import ( + "fmt" + + retry "github.com/avast/retry-go" sdk "github.com/cosmos/cosmos-sdk/types" transfertypes "github.com/cosmos/cosmos-sdk/x/ibc/applications/transfer/types" clienttypes "github.com/cosmos/cosmos-sdk/x/ibc/core/02-client/types" @@ -320,3 +323,193 @@ func (c *Chain) MsgTransfer(dst *PathEnd, amount sdk.Coin, dstAddr string, timeoutTimestamp, ) } + +// MsgRelayRecvPacket constructs the MsgRecvPacket which is to be sent to the receiving chain. +// The counterparty represents the sending chain where the packet commitment would be stored. +func (c *Chain) MsgRelayRecvPacket(counterparty *Chain, packet *relayMsgRecvPacket) (msgs []sdk.Msg, err error) { + var comRes *chantypes.QueryPacketCommitmentResponse + if err = retry.Do(func() (err error) { + // NOTE: the proof height uses - 1 due to tendermint's delayed execution model + comRes, err = counterparty.QueryPacketCommitment(int64(counterparty.MustGetLatestLightHeight())-1, packet.seq) + if err != nil { + return err + } + + if comRes.Proof == nil || comRes.Commitment == nil { + return fmt.Errorf("recv packet commitment query seq(%d) is nil", packet.seq) + } + + return nil + }, rtyAtt, rtyDel, rtyErr, retry.OnRetry(func(n uint, _ error) { + // clear messages + msgs = []sdk.Msg{} + + // OnRetry we want to update the light clients and then debug log + updateMsg, err := c.UpdateClient(counterparty) + if err != nil { + return + } + + msgs = append(msgs, updateMsg) + + if counterparty.debug { + counterparty.Log(fmt.Sprintf("- [%s]@{%d} - try(%d/%d) query packet commitment: %s", + counterparty.ChainID, counterparty.MustGetLatestLightHeight()-1, n+1, rtyAttNum, err)) + } + + })); err != nil { + counterparty.Error(err) + return + } + + if comRes == nil { + return nil, fmt.Errorf("receive packet [%s]seq{%d} has no associated proofs", c.ChainID, packet.seq) + } + + version := clienttypes.ParseChainID(c.ChainID) + msg := chantypes.NewMsgRecvPacket( + chantypes.NewPacket( + packet.packetData, + packet.seq, + counterparty.PathEnd.PortID, + counterparty.PathEnd.ChannelID, + c.PathEnd.PortID, + c.PathEnd.ChannelID, + clienttypes.NewHeight(version, packet.timeout), + packet.timeoutStamp, + ), + comRes.Proof, + comRes.ProofHeight, + c.MustGetAddress(), + ) + + return append(msgs, msg), nil +} + +// MsgRelayAcknowledgement constructs the MsgAcknowledgement which is to be sent to the sending chain. +// The counterparty represents the receiving chain where the acknowledgement would be stored. +func (c *Chain) MsgRelayAcknowledgement(counterparty *Chain, packet *relayMsgPacketAck) (msgs []sdk.Msg, err error) { + var ackRes *chantypes.QueryPacketAcknowledgementResponse + if err = retry.Do(func() (err error) { + // NOTE: the proof height uses - 1 due to tendermint's delayed execution model + ackRes, err = counterparty.QueryPacketAcknowledgement(int64(counterparty.MustGetLatestLightHeight())-1, packet.seq) + if err != nil { + return err + } + + if ackRes.Proof == nil || ackRes.Acknowledgement == nil { + return fmt.Errorf("ack packet acknowledgement query seq(%d) is nil", packet.seq) + } + + return nil + }, rtyAtt, rtyDel, rtyErr, retry.OnRetry(func(n uint, _ error) { + // clear messages + msgs = []sdk.Msg{} + + // OnRetry we want to update the light clients and then debug log + updateMsg, err := c.UpdateClient(counterparty) + if err != nil { + return + } + + msgs = append(msgs, updateMsg) + + if counterparty.debug { + counterparty.Log(fmt.Sprintf("- [%s]@{%d} - try(%d/%d) query packet acknowledgement: %s", + counterparty.ChainID, counterparty.MustGetLatestLightHeight()-1, n+1, rtyAttNum, err)) + } + + })); err != nil { + counterparty.Error(err) + return + } + + if ackRes == nil { + return nil, fmt.Errorf("ack packet [%s]seq{%d} has no associated proofs", counterparty.ChainID, packet.seq) + } + + version := clienttypes.ParseChainID(counterparty.ChainID) + msg := chantypes.NewMsgAcknowledgement( + chantypes.NewPacket( + packet.packetData, + packet.seq, + c.PathEnd.PortID, + c.PathEnd.ChannelID, + counterparty.PathEnd.PortID, + counterparty.PathEnd.ChannelID, + clienttypes.NewHeight(version, packet.timeout), + packet.timeoutStamp, + ), + packet.ack, + ackRes.Proof, + ackRes.ProofHeight, + c.MustGetAddress(), + ) + + return append(msgs, msg), nil +} + +// MsgRelayTimeout constructs the MsgTimeout which is to be sent to the sending chain. +// The counterparty represents the receiving chain where the receipts would have been +// stored. +func (c *Chain) MsgRelayTimeout(counterparty *Chain, packet *relayMsgTimeout) (msgs []sdk.Msg, err error) { + var recvRes *chantypes.QueryPacketReceiptResponse + if err = retry.Do(func() (err error) { + // NOTE: Timeouts currently only work with ORDERED channels for nwo + // NOTE: the proof height uses - 1 due to tendermint's delayed execution model + recvRes, err = counterparty.QueryPacketReceipt(int64(counterparty.MustGetLatestLightHeight())-1, packet.seq) + if err != nil { + return err + } + + if recvRes.Proof == nil { + return fmt.Errorf("timeout packet receipt proof seq(%d) is nil", packet.seq) + } + + return nil + }, rtyAtt, rtyDel, rtyErr, retry.OnRetry(func(n uint, _ error) { + // clear messages + msgs = []sdk.Msg{} + + // OnRetry we want to update the light clients and then debug log + updateMsg, err := c.UpdateClient(counterparty) + if err != nil { + return + } + + msgs = append(msgs, updateMsg) + + if counterparty.debug { + counterparty.Log(fmt.Sprintf("- [%s]@{%d} - try(%d/%d) query packet receipt: %s", + counterparty.ChainID, counterparty.MustGetLatestLightHeight()-1, n+1, rtyAttNum, err)) + } + + })); err != nil { + counterparty.Error(err) + return + } + + if recvRes == nil { + return nil, fmt.Errorf("timeout packet [%s]seq{%d} has no associated proofs", c.ChainID, packet.seq) + } + + version := clienttypes.ParseChainID(counterparty.ChainID) + msg := chantypes.NewMsgTimeout( + chantypes.NewPacket( + packet.packetData, + packet.seq, + c.PathEnd.PortID, + c.PathEnd.ChannelID, + counterparty.PathEnd.PortID, + counterparty.PathEnd.ChannelID, + clienttypes.NewHeight(version, packet.timeout), + packet.timeoutStamp, + ), + packet.seq, + recvRes.Proof, + recvRes.ProofHeight, + c.MustGetAddress(), + ) + + return append(msgs, msg), nil +} diff --git a/relayer/naive-strategy.go b/relayer/naive-strategy.go index 59372d5bb..b33106154 100644 --- a/relayer/naive-strategy.go +++ b/relayer/naive-strategy.go @@ -420,31 +420,23 @@ func (nrs *NaiveStrategy) RelayAcknowledgements(src, dst *Chain, sp *RelaySequen // add messages for sequences on src for _, seq := range sp.Src { // SRC wrote ack, so we query packet and send to DST - pkt, err := acknowledgementFromSequence(src, dst, seq) + relayAckMsgs, err := acknowledgementFromSequence(src, dst, seq) if err != nil { return err } - msg, err := pkt.Msg(dst, src) - if err != nil { - return err - } - msgs.Dst = append(msgs.Dst, msg) + msgs.Dst = append(msgs.Dst, relayAckMsgs...) } // add messages for sequences on dst for _, seq := range sp.Dst { // DST wrote ack, so we query packet and send to SRC - pkt, err := acknowledgementFromSequence(dst, src, seq) + relayAckMsgs, err := acknowledgementFromSequence(dst, src, seq) if err != nil { return err } - msg, err := pkt.Msg(src, dst) - if err != nil { - return err - } - msgs.Src = append(msgs.Src, msg) + msgs.Src = append(msgs.Src, relayAckMsgs...) } if !msgs.Ready() { @@ -492,56 +484,38 @@ func (nrs *NaiveStrategy) RelayPackets(src, dst *Chain, sp *RelaySequences) erro // add messages for sequences on src for _, seq := range sp.Src { // Query src for the sequence number to get type of packet - pkt, err := relayPacketFromSequence(src, dst, seq) + recvMsgs, timeoutMsgs, err := relayPacketFromSequence(src, dst, seq) if err != nil { return err } // depending on the type of message to be relayed, we need to // send to different chains - switch pkt.(type) { - case *relayMsgRecvPacket: - msg, err := pkt.Msg(dst, src) - if err != nil { - return err - } - msgs.Dst = append(msgs.Dst, msg) - case *relayMsgTimeout: - msg, err := pkt.Msg(src, dst) - if err != nil { - return err - } - msgs.Src = append(msgs.Src, msg) - default: - return fmt.Errorf("%T packet types not supported", pkt) + if recvMsgs != nil { + msgs.Dst = append(msgs.Dst, recvMsgs...) + } + + if timeoutMsgs != nil { + msgs.Src = append(msgs.Src, timeoutMsgs...) } } // add messages for sequences on dst for _, seq := range sp.Dst { // Query dst for the sequence number to get type of packet - pkt, err := relayPacketFromSequence(dst, src, seq) + recvMsgs, timeoutMsgs, err := relayPacketFromSequence(dst, src, seq) if err != nil { return err } // depending on the type of message to be relayed, we need to // send to different chains - switch pkt.(type) { - case *relayMsgRecvPacket: - msg, err := pkt.Msg(src, dst) - if err != nil { - return err - } - msgs.Src = append(msgs.Src, msg) - case *relayMsgTimeout: - msg, err := pkt.Msg(dst, src) - if err != nil { - return err - } - msgs.Dst = append(msgs.Dst, msg) - default: - return fmt.Errorf("%T packet types not supported", pkt) + if recvMsgs != nil { + msgs.Src = append(msgs.Src, recvMsgs...) + } + + if timeoutMsgs != nil { + msgs.Dst = append(msgs.Dst, timeoutMsgs...) } } @@ -583,54 +557,60 @@ func (nrs *NaiveStrategy) RelayPackets(src, dst *Chain, sp *RelaySequences) erro return nil } -// relayPacketFromSequence returns a sdk.Msg to relay a packet with a given seq on src -func relayPacketFromSequence(src, dst *Chain, seq uint64) (relayPacket, error) { +// relayPacketFromSequence relays a packet with a given seq on src +// and returns recvPacket msgs, timeoutPacketmsgs and error +func relayPacketFromSequence(src, dst *Chain, seq uint64) ([]sdk.Msg, []sdk.Msg, error) { txs, err := src.QueryTxs(src.MustGetLatestLightHeight(), 1, 1000, rcvPacketQuery(src.PathEnd.ChannelID, int(seq))) switch { case err != nil: - return nil, err + return nil, nil, err case len(txs) == 0: - return nil, fmt.Errorf("no transactions returned with query") + return nil, nil, fmt.Errorf("no transactions returned with query") case len(txs) > 1: - return nil, fmt.Errorf("more than one transaction returned with query") + return nil, nil, fmt.Errorf("more than one transaction returned with query") } rcvPackets, timeoutPackets, err := relayPacketsFromResultTx(src, dst, txs[0]) switch { case err != nil: - return nil, err + return nil, nil, err case len(rcvPackets) == 0 && len(timeoutPackets) == 0: - return nil, fmt.Errorf("no relay msgs created from query response") + return nil, nil, fmt.Errorf("no relay msgs created from query response") case len(rcvPackets)+len(timeoutPackets) > 1: - return nil, fmt.Errorf("more than one relay msg found in tx query") + return nil, nil, fmt.Errorf("more than one relay msg found in tx query") } if len(rcvPackets) == 1 { pkt := rcvPackets[0] if seq != pkt.Seq() { - return nil, fmt.Errorf("wrong sequence: expected(%d) got(%d)", seq, pkt.Seq()) + return nil, nil, fmt.Errorf("wrong sequence: expected(%d) got(%d)", seq, pkt.Seq()) } - if err = pkt.FetchCommitResponse(dst, src); err != nil { - return nil, err + + msgs, err := dst.MsgRelayRecvPacket(src, pkt.(*relayMsgRecvPacket)) + if err != nil { + return nil, nil, err } - return pkt, nil + return msgs, nil, nil } if len(timeoutPackets) == 1 { pkt := timeoutPackets[0] if seq != pkt.Seq() { - return nil, fmt.Errorf("wrong sequence: expected(%d) got(%d)", seq, pkt.Seq()) + return nil, nil, fmt.Errorf("wrong sequence: expected(%d) got(%d)", seq, pkt.Seq()) } - if err = pkt.FetchCommitResponse(src, dst); err != nil { - return nil, err + + msgs, err := src.MsgRelayTimeout(dst, pkt.(*relayMsgTimeout)) + if err != nil { + return nil, nil, err } - return pkt, nil + return nil, msgs, nil } - return nil, fmt.Errorf("should have errored before here") + return nil, nil, fmt.Errorf("should have errored before here") } -func acknowledgementFromSequence(src, dst *Chain, seq uint64) (relayPacket, error) { +// source is the sending chain, destination is the receiving chain +func acknowledgementFromSequence(src, dst *Chain, seq uint64) ([]sdk.Msg, error) { txs, err := src.QueryTxs(src.MustGetLatestLightHeight(), 1, 1000, ackPacketQuery(src.PathEnd.ChannelID, int(seq))) switch { case err != nil: @@ -655,10 +635,12 @@ func acknowledgementFromSequence(src, dst *Chain, seq uint64) (relayPacket, erro if seq != pkt.Seq() { return nil, fmt.Errorf("wrong sequence: expected(%d) got(%d)", seq, pkt.Seq()) } - if err = pkt.FetchCommitResponse(dst, src); err != nil { + + msgs, err := src.MsgRelayAcknowledgement(dst, pkt) + if err != nil { return nil, err } - return pkt, nil + return msgs, nil } // relayPacketsFromResultTx looks through the events in a *ctypes.ResultTx