Merge pull request #1667 from OpenBazaar/mg-resend-order-msg-spike-2

Re-send missed ORDER messages on processing failure

Author: cpacia

Dockerfile.dev

@@ -30,6 +30,11 @@ RUN go get -u github.com/gogo/protobuf/proto \
               github.com/derekparker/delve/cmd/dlv \
               github.com/tools/godep
 
+RUN cd /go/src/github.com/golang/protobuf && \
+    git checkout tags/v1.2.0 -b v1.2.0 && \
+    cd protoc-gen-go && \
+    go install
+
 RUN curl -sfL https://install.goreleaser.com/github.com/golangci/golangci-lint.sh | bash -s -- -b $GOPATH/bin v1.15.0
 
 WORKDIR /go/src/github.com/OpenBazaar/openbazaar-go

api/jsonapi.go

@@ -4251,47 +4251,26 @@ func (i *jsonAPIHandler) POSTResendOrderMessage(w http.ResponseWriter, r *http.R
 		return
 	}
 
-	if args.OrderID == "" {
-		ErrorResponse(w, http.StatusBadRequest, core.ErrOrderNotFound.Error())
+	if args.MessageType == "" {
+		ErrorResponse(w, http.StatusBadRequest, fmt.Sprintf("missing messageType argument"))
 		return
 	}
-
-	var msgType int32
-	var ok bool
-
-	if msgType, ok = pb.Message_MessageType_value[args.MessageType]; !ok {
-		ErrorResponse(w, http.StatusBadRequest, "invalid order message type")
+	if args.OrderID == "" {
+		ErrorResponse(w, http.StatusBadRequest, fmt.Sprintf("missing orderID argument"))
 		return
 	}
 
-	msg, peerID, err := i.node.Datastore.Messages().
-		GetByOrderIDType(args.OrderID, pb.Message_MessageType(msgType))
-	if err != nil || msg == nil || msg.Msg.GetPayload() == nil {
-		ErrorResponse(w, http.StatusBadRequest, "order message not found")
+	msgInt, ok := pb.Message_MessageType_value[strings.ToUpper(args.MessageType)]
+	if !ok {
+		ErrorResponse(w, http.StatusBadRequest, fmt.Sprintf("unknown messageType (%s)", args.MessageType))
 		return
 	}
 
-	p, err := peer.IDB58Decode(peerID)
-	if err != nil {
-		ErrorResponse(w, http.StatusBadRequest, "invalid peer id")
+	if err := i.node.ResendCachedOrderMessage(args.OrderID, pb.Message_MessageType(msgInt)); err != nil {
+		ErrorResponse(w, http.StatusInternalServerError, err.Error())
 		return
 	}
 
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	err = i.node.Service.SendMessage(ctx, p, &msg.Msg)
-	if err != nil {
-		// If send message failed, try sending offline message
-		log.Warningf("resending message failed: %v", err)
-		err = i.node.SendOfflineMessage(p, nil, &msg.Msg)
-		if err != nil {
-			log.Errorf("resending offline message failed: %v", err)
-			ErrorResponse(w, http.StatusBadRequest, "order message not sent")
-			return
-		}
-	}
-
 	SanitizedResponse(w, `{}`)
 }
 

api/jsonapi_test.go

@@ -847,6 +847,19 @@ func TestNotificationsAreReturnedInExpectedOrder(t *testing.T) {
 	}, dbSetup, dbTeardown)
 }
 
+func TestResendOrderMessage(t *testing.T) {
+	runAPITests(t, apiTests{
+		// supports missing messageType
+		{"POST", "/ob/resendordermessage", `{"orderID":"123"}`, http.StatusBadRequest, errorResponseJSON(fmt.Errorf("missing messageType argument"))},
+		// supports missing order ID
+		{"POST", "/ob/resendordermessage", `{"messageType":"nonexistant"}`, http.StatusBadRequest, errorResponseJSON(fmt.Errorf("missing orderID argument"))},
+		// supports nonexistant message types
+		{"POST", "/ob/resendordermessage", `{"orderID":"123","messageType":"nonexistant"}`, http.StatusBadRequest, errorResponseJSON(fmt.Errorf("unknown messageType (nonexistant)"))},
+		// supports downcased message types, expected not to find order ID
+		{"POST", "/ob/resendordermessage", `{"orderID":"123","messageType":"order"}`, http.StatusInternalServerError, errorResponseJSON(fmt.Errorf("unable to find message for order ID (123) and message type (ORDER)"))},
+	})
+}
+
 // TODO: Make NewDisputeCaseRecord return a valid fixture for this valid case to work
 //func TestCloseDisputeReturnsOK(t *testing.T) {
 //dbSetup := func(testRepo *test.Repository) error {

core/disputes.go

@@ -305,9 +305,12 @@ func (n *OpenBazaarNode) ProcessDisputeOpen(rc *pb.RicardianContract, peerID str
 		DisputerHandle = contract.BuyerOrder.BuyerID.Handle
 		DisputeeID = contract.VendorListings[0].VendorID.PeerID
 		DisputeeHandle = contract.VendorListings[0].VendorID.Handle
-		// Load out version of the contract from the db
+		// Load our version of the contract from the db
 		myContract, state, _, records, _, _, err := n.Datastore.Sales().GetByOrderId(orderID)
 		if err != nil {
+			if err := n.SendProcessingError(DisputerID, orderID, pb.Message_DISPUTE_OPEN, nil); err != nil {
+				log.Errorf("failed sending ORDER_PROCESSING_FAILURE to peer (%s): %s", DisputerID, err)
+			}
 			return net.OutOfOrderMessage
 		}
 		// Check this order is currently in a state which can be disputed
@@ -365,6 +368,9 @@ func (n *OpenBazaarNode) ProcessDisputeOpen(rc *pb.RicardianContract, peerID str
 			return err
 		}
 		if state == pb.OrderState_AWAITING_PAYMENT || state == pb.OrderState_AWAITING_FULFILLMENT || state == pb.OrderState_PARTIALLY_FULFILLED || state == pb.OrderState_PENDING {
+			if err := n.SendProcessingError(DisputerID, orderID, pb.Message_DISPUTE_OPEN, myContract); err != nil {
+				log.Errorf("failed sending ORDER_PROCESSING_FAILURE to peer (%s): %s", DisputerID, err)
+			}
 			return net.OutOfOrderMessage
 		}
 		// Check this order is currently in a state which can be disputed

core/errors.go

@@ -1,10 +1,16 @@
 package core
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"strconv"
+
+	peer "gx/ipfs/QmYVXrKrKHDC9FobgmcmshCDyWwdrfwfanNQN4oxJ9Fk3h/go-libp2p-peer"
+
+	"github.com/OpenBazaar/openbazaar-go/pb"
+	"github.com/golang/protobuf/ptypes"
 )
 
 var (
@@ -121,3 +127,35 @@ func (e ErrMarketPriceListingIllegalField) Error() string {
 func illegalFieldString(objectType string, field string) string {
 	return fmt.Sprintf("Illegal %s field: %s", objectType, field)
 }
+
+// SendProcessingError will encapsulate the failing state in a message to be sent back to pid
+// When pid receives the OrderProcessingError, it will analyze the contract and send the messages
+// that this node is missing to resynchronize the order
+func (n *OpenBazaarNode) SendProcessingError(pid, oid string, attemptedMessage pb.Message_MessageType, latestContract *pb.RicardianContract) error {
+	log.Debugf("sending ORDER_PROCESSING_ERROR to peer (%s)", pid)
+	var (
+		procErrMsg = &pb.OrderProcessingFailure{
+			OrderID:              oid,
+			AttemptedMessageType: attemptedMessage,
+			Contract:             latestContract,
+		}
+		procErrBytes, mErr = ptypes.MarshalAny(procErrMsg)
+		errMsg             = &pb.Message{
+			MessageType: pb.Message_ORDER_PROCESSING_FAILURE,
+			Payload:     procErrBytes,
+		}
+		p, pErr = peer.IDB58Decode(pid)
+	)
+	if mErr != nil {
+		log.Errorf("failed marshaling OrderProcessingFailure message for order (%s): %s", oid, mErr)
+		return mErr
+	}
+	if pErr != nil {
+		log.Errorf("failed decoding peer ID (%s): %s", pid, pErr)
+		return pErr
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), n.OfflineMessageFailoverTimeout)
+	defer cancel()
+
+	return n.Service.SendMessage(ctx, p, errMsg)
+}

core/net.go

@@ -263,6 +263,36 @@ func (n *OpenBazaarNode) Unfollow(peerID string) error {
 	return nil
 }
 
+// ResendCachedOrderMessage will retrieve the ORDER message from the datastore and resend it to the peerID
+// for which it was originally intended
+func (n *OpenBazaarNode) ResendCachedOrderMessage(orderID string, msgType pb.Message_MessageType) error {
+	if _, ok := pb.Message_MessageType_name[int32(msgType)]; !ok {
+		return fmt.Errorf("invalid order message type (%d)", int(msgType))
+	}
+
+	msg, peerID, err := n.Datastore.Messages().GetByOrderIDType(orderID, msgType)
+	if err != nil || msg == nil || msg.Msg.GetPayload() == nil {
+		return fmt.Errorf("unable to find message for order ID (%s) and message type (%s)", orderID, msgType.String())
+	}
+
+	p, err := peer.IDB58Decode(peerID)
+	if err != nil {
+		return fmt.Errorf("unable to decode invalid peer ID for order (%s) and message type (%s)", orderID, msgType.String())
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), n.OfflineMessageFailoverTimeout)
+	defer cancel()
+
+	if err = n.Service.SendMessage(ctx, p, &msg.Msg); err != nil {
+		go func() {
+			if err := n.SendOfflineMessage(p, nil, &msg.Msg); err != nil {
+				log.Errorf("error resending offline message for order id (%s) and message type (%+v): %s", orderID, msgType, err.Error())
+			}
+		}()
+	}
+	return nil
+}
+
 // SendOrder - send order created msg to peer
 func (n *OpenBazaarNode) SendOrder(peerID string, contract *pb.RicardianContract) (resp *pb.Message, err error) {
 	p, err := peer.IDB58Decode(peerID)

net/retriever/retriever_test.go

@@ -17,11 +17,12 @@ func TestEnsureNoOmissionsInMessageProcessingOrder(t *testing.T) {
 
 	// Add deliberate omissions to this list
 	blackList := map[pb.Message_MessageType]struct{}{
-		pb.Message_PING:          {},
-		pb.Message_OFFLINE_RELAY: {},
-		pb.Message_STORE:         {},
-		pb.Message_BLOCK:         {},
-		pb.Message_ERROR:         {},
+		pb.Message_PING:                     {},
+		pb.Message_OFFLINE_RELAY:            {},
+		pb.Message_STORE:                    {},
+		pb.Message_BLOCK:                    {},
+		pb.Message_ORDER_PROCESSING_FAILURE: {},
+		pb.Message_ERROR:                    {},
 	}
 
 	// Inclusion check