Improve algorithm for sharing old orders

core/ordersync/ordersync.go

@@ -79,7 +79,8 @@ const (
 // Request represents a high-level ordersync request. It abstracts away some
 // of the details of subprotocol negotiation and encoding/decoding.
 type Request struct {
-	Metadata interface{} `json:"metadata"`
+	RequesterID peer.ID     `json:"requesterID"`
+	Metadata    interface{} `json:"metadata"`
 }
 
 // rawRequest contains all the details we need at the lowest level to encode/decode
@@ -93,9 +94,10 @@ type rawRequest struct {
 // Response represents a high-level ordersync response. It abstracts away some
 // of the details of subprotocol negotiation and encoding/decoding.
 type Response struct {
-	Orders   []*zeroex.SignedOrder `json:"orders"`
-	Complete bool                  `json:"complete"`
-	Metadata interface{}           `json:"metadata"`
+	ProviderID peer.ID               `json:"providerID"`
+	Orders     []*zeroex.SignedOrder `json:"orders"`
+	Complete   bool                  `json:"complete"`
+	Metadata   interface{}           `json:"metadata"`
 }
 
 // rawResponse contains all the details we need at the lowest level to encode/decode
@@ -206,7 +208,7 @@ func (s *Service) HandleStream(stream network.Stream) {
 	defer func() {
 		_ = stream.Close()
 	}()
-	remotePeerID := stream.Conn().RemotePeer()
+	requesterID := stream.Conn().RemotePeer()
 
 	for {
 		if err := s.requestRateLimiter.Wait(s.ctx); err != nil {
@@ -225,16 +227,16 @@ func (s *Service) HandleStream(stream network.Stream) {
 		}).Trace("received ordersync request")
 		if rawReq.Type != TypeRequest {
 			log.WithField("gotType", rawReq.Type).Warn("wrong type for Request")
-			s.handlePeerScoreEvent(remotePeerID, psInvalidMessage)
+			s.handlePeerScoreEvent(requesterID, psInvalidMessage)
 			return
 		}
 		subprotocol, err := s.GetMatchingSubprotocol(rawReq)
 		if err != nil {
 			log.WithError(err).Warn("GetMatchingSubprotocol returned error")
-			s.handlePeerScoreEvent(remotePeerID, psSubprotocolNegotiationFailed)
+			s.handlePeerScoreEvent(requesterID, psSubprotocolNegotiationFailed)
 			return
 		}
-		res, err := handleRequestWithSubprotocol(s.ctx, subprotocol, rawReq)
+		res, err := handleRequestWithSubprotocol(s.ctx, subprotocol, requesterID, rawReq)
 		if err != nil {
 			log.WithError(err).Warn("subprotocol returned error")
 			return
@@ -244,7 +246,7 @@ func (s *Service) HandleStream(stream network.Stream) {
 			log.WithError(err).Error("could not encode raw metadata")
 			return
 		}
-		s.handlePeerScoreEvent(remotePeerID, psValidMessage)
+		s.handlePeerScoreEvent(requesterID, psValidMessage)
 		rawRes := rawResponse{
 			Type:        TypeResponse,
 			Subprotocol: subprotocol.Name(),
@@ -255,9 +257,9 @@ func (s *Service) HandleStream(stream network.Stream) {
 		if err := json.NewEncoder(stream).Encode(rawRes); err != nil {
 			log.WithFields(log.Fields{
 				"error":     err.Error(),
-				"requester": remotePeerID.Pretty(),
+				"requester": requesterID.Pretty(),
 			}).Warn("could not encode ordersync response")
-			s.handlePeerScoreEvent(remotePeerID, psUnexpectedDisconnect)
+			s.handlePeerScoreEvent(requesterID, psUnexpectedDisconnect)
 			return
 		}
 		if res.Complete {
@@ -336,33 +338,35 @@ func (s *Service) GetOrders(ctx context.Context, minPeers int) error {
 	return nil
 }
 
-func handleRequestWithSubprotocol(ctx context.Context, subprotocol Subprotocol, rawReq *rawRequest) (*Response, error) {
-	req, err := parseRequestWithSubprotocol(subprotocol, rawReq)
+func handleRequestWithSubprotocol(ctx context.Context, subprotocol Subprotocol, requesterID peer.ID, rawReq *rawRequest) (*Response, error) {
+	req, err := parseRequestWithSubprotocol(subprotocol, requesterID, rawReq)
 	if err != nil {
 		return nil, err
 	}
 	return subprotocol.HandleOrderSyncRequest(ctx, req)
 }
 
-func parseRequestWithSubprotocol(subprotocol Subprotocol, rawReq *rawRequest) (*Request, error) {
+func parseRequestWithSubprotocol(subprotocol Subprotocol, requesterID peer.ID, rawReq *rawRequest) (*Request, error) {
 	metadata, err := subprotocol.ParseRequestMetadata(rawReq.Metadata)
 	if err != nil {
 		return nil, err
 	}
 	return &Request{
-		Metadata: metadata,
+		RequesterID: requesterID,
+		Metadata:    metadata,
 	}, nil
 }
 
-func parseResponseWithSubprotocol(subprotocol Subprotocol, rawRes *rawResponse) (*Response, error) {
+func parseResponseWithSubprotocol(subprotocol Subprotocol, providerID peer.ID, rawRes *rawResponse) (*Response, error) {
 	metadata, err := subprotocol.ParseResponseMetadata(rawRes.Metadata)
 	if err != nil {
 		return nil, err
 	}
 	return &Response{
-		Orders:   rawRes.Orders,
-		Complete: rawRes.Complete,
-		Metadata: metadata,
+		ProviderID: providerID,
+		Orders:     rawRes.Orders,
+		Complete:   rawRes.Complete,
+		Metadata:   metadata,
 	}, nil
 }
 
@@ -422,7 +426,7 @@ func (s *Service) getOrdersFromPeer(ctx context.Context, providerID peer.ID) err
 			return fmt.Errorf("unsupported subprotocol: %s", subprotocol)
 		}
 		selectedSubprotocol = subprotocol
-		res, err := parseResponseWithSubprotocol(subprotocol, rawRes)
+		res, err := parseResponseWithSubprotocol(subprotocol, providerID, rawRes)
 		if err != nil {
 			s.handlePeerScoreEvent(providerID, psInvalidMessage)
 			return err

core/ordersync_subprotocols.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 
 	"github.com/0xProject/0x-mesh/core/ordersync"
+	"github.com/0xProject/0x-mesh/orderfilter"
 	"github.com/0xProject/0x-mesh/zeroex"
 )
 
@@ -17,16 +18,18 @@ var _ ordersync.Subprotocol = (*FilteredPaginationSubProtocol)(nil)
 // paginating through them. It involves sending multiple requests until pagination is
 // finished and all orders have been returned.
 type FilteredPaginationSubProtocol struct {
-	app     *App
-	perPage int
+	app         *App
+	orderFilter *orderfilter.Filter
+	perPage     int
 }
 
 // NewFilteredPaginationSubprotocol creates and returns a new FilteredPaginationSubprotocol
 // which will respond with perPage orders for each individual request/response.
 func NewFilteredPaginationSubprotocol(app *App, perPage int) *FilteredPaginationSubProtocol {
 	return &FilteredPaginationSubProtocol{
-		app:     app,
-		perPage: perPage,
+		app:         app,
+		orderFilter: app.orderFilter,
+		perPage:     perPage,
 	}
 }
 
@@ -68,22 +71,40 @@ func (p *FilteredPaginationSubProtocol) HandleOrderSyncRequest(ctx context.Conte
 			return nil, fmt.Errorf("FilteredPaginationSubProtocol received request with wrong metadata type (got %T)", req.Metadata)
 		}
 	}
-
-	ordersResp, err := p.app.GetOrders(metadata.Page, p.perPage, metadata.SnapshotID)
-	if err != nil {
-		return nil, err
-	}
-	orders := make([]*zeroex.SignedOrder, len(ordersResp.OrdersInfos))
-	for i, orderInfo := range ordersResp.OrdersInfos {
-		orders[i] = orderInfo.SignedOrder
+	// It's possible that none of the orders in the current page match the filter.
+	// We don't want to respond with zero orders, so keep iterating until we find
+	// at least some orders that match the filter.
+	filteredOrders := []*zeroex.SignedOrder{}
+	var snapshotID string
+	var currentPage int
+	for currentPage = metadata.Page; len(filteredOrders) == 0; currentPage += 1 {
+		// Get the orders for this page.
+		ordersResp, err := p.app.GetOrders(currentPage, p.perPage, metadata.SnapshotID)
+		if err != nil {
+			return nil, err
+		}
+		snapshotID = ordersResp.SnapshotID
+		if len(ordersResp.OrdersInfos) == 0 {
+			// No more orders left.
+			break
+		}
+		// Filter the orders for this page. If none of them match the filter, we continue
+		// on to the next page.
+		for _, orderInfo := range ordersResp.OrdersInfos {
+			if matches, err := p.orderFilter.MatchOrder(orderInfo.SignedOrder); err != nil {
+				return nil, err
+			} else if matches {
+				filteredOrders = append(filteredOrders, orderInfo.SignedOrder)
+			}
+		}
 	}
-	// TODO(albrow): Filter orders
+
 	return &ordersync.Response{
-		Orders:   orders,
-		Complete: len(orders) == 0,
+		Orders:   filteredOrders,
+		Complete: len(filteredOrders) == 0,
 		Metadata: &FilteredPaginationResponseMetadata{
-			Page:       metadata.Page,
-			SnapshotID: ordersResp.SnapshotID,
+			Page:       currentPage,
+			SnapshotID: snapshotID,
 		},
 	}, nil
 }
@@ -99,7 +120,16 @@ func (p *FilteredPaginationSubProtocol) HandleOrderSyncResponse(ctx context.Cont
 	if !ok {
 		return nil, fmt.Errorf("FilteredPaginationSubProtocol received response with wrong metadata type (got %T)", res.Metadata)
 	}
-	// TODO(albrow): Check that this order matches our current filter/topic
+	filteredOrders := []*zeroex.SignedOrder{}
+	for _, order := range res.Orders {
+		if matches, err := p.orderFilter.MatchOrder(order); err != nil {
+			return nil, err
+		} else if matches {
+			filteredOrders = append(filteredOrders, order)
+		} else if !matches {
+			p.app.handlePeerScoreEvent(res.ProviderID, psReceivedOrderDoesNotMatchFilter)
+		}
+	}
 	_, err := p.app.orderWatcher.ValidateAndStoreValidOrders(ctx, res.Orders, false, p.app.chainID)
 	if err != nil {
 		return nil, err

core/peer_score.go

@@ -11,6 +11,7 @@ const (
 	psInvalidMessage peerScoreEvent = iota
 	psValidMessage
 	psOrderStored
+	psReceivedOrderDoesNotMatchFilter
 )
 
 func (app *App) handlePeerScoreEvent(id peer.ID, event peerScoreEvent) {
@@ -26,6 +27,8 @@ func (app *App) handlePeerScoreEvent(id peer.ID, event peerScoreEvent) {
 		app.node.SetPeerScore(id, "valid-message", 5)
 	case psOrderStored:
 		app.node.SetPeerScore(id, "order-stored", 10)
+	case psReceivedOrderDoesNotMatchFilter:
+		app.node.SetPeerScore(id, "received-order-does-not-match-filter", -10)
 	default:
 		log.WithField("event", event).Error("unknown peerScoreEvent")
 	}