orderwatch,core,blockwatch: Improve Order Event subscription stability

CHANGELOG.md

@@ -13,6 +13,12 @@ This changelog is a work in progress and may contain notes for versions which ha
 - Instead of progressing Mesh forward by a single block on every invocation of the `BLOCK_POLLING_INTERVAL`, we now attempt to sync as many blocks as necessary to reach the latest block available. This will reduce the chances of Mesh becoming out-of-sync with it's backing Ethereum node ([#564](https://github.com/0xProject/0x-mesh/pull/564))
 - Added a new environment variable `ENABLE_ETHEREUM_RPC_RATE_LIMITING` and config option `enableEthereumRPCRateLimiting` which can be used to completely disable Mesh's internal Ethereum RPC rate limiting features. By default it is enabled, and disabling can have some consequences depending on your RPC provider. ([#584](https://github.com/0xProject/0x-mesh/pull/584))
 - Added a `SnapshotTimestamp` field to `GetOrdersResponse`, the return type of the `mesh_getOrders` RPC method. This way, the caller can know at what point in time the snapshot had been created. ([#591](https://github.com/0xProject/0x-mesh/pull/591))
+- Improved batching of events emitted from order events subscriptions ([#566](https://github.com/0xProject/0x-mesh/pull/566))
+
+### Bug fixes 🐞
+
+- Fixed an issue where order updates could have been missed if Mesh discovered blocks but didn't have time to process them before getting shut down. Now, blocks are only persisted to the DB once any order updates resulting from it have been processed. ([#566](https://github.com/0xProject/0x-mesh/pull/566)).
+- Fixed a race-condition when adding new orders to Mesh which could result in order-relevant events being missed if they occured very soon after the order was submitted and the order validation RPC call took a long time ([#566](https://github.com/0xProject/0x-mesh/pull/566)).
 
 
 ## v6.1.2-beta

Makefile

@@ -56,7 +56,7 @@ test-go-parallel:
 
 .PHONY: test-go-serial
 test-go-serial:
-	go test ./zeroex/ordervalidator ./zeroex/orderwatch ./core -race -timeout 30s -p=1 --serial
+	go test ./zeroex/ordervalidator ./zeroex/orderwatch ./core -race -timeout 60s -p=1 --serial
 
 
 .PHONY: test-browser-integration

browser/go/main.go

@@ -216,7 +216,7 @@ func (cw *MeshWrapper) AddOrders(rawOrders js.Value, pinned bool) (js.Value, err
 	if err := json.Unmarshal([]byte(encodedOrders), &rawMessages); err != nil {
 		return js.Undefined(), err
 	}
-	results, err := cw.app.AddOrders(rawMessages, pinned)
+	results, err := cw.app.AddOrders(cw.ctx, rawMessages, pinned)
 	if err != nil {
 		return js.Undefined(), err
 	}

cmd/mesh/main.go

@@ -62,7 +62,7 @@ func main() {
 	go func() {
 		defer wg.Done()
 		log.WithField("rpc_addr", config.RPCAddr).Info("starting RPC server")
-		if err := listenRPC(app, config, ctx); err != nil {
+		if err := listenRPC(ctx, app, config); err != nil {
 			rpcErrChan <- err
 		}
 	}()

cmd/mesh/rpc_handler.go

@@ -28,15 +28,17 @@ const orderEventsBufferSize = 8000
 
 type rpcHandler struct {
 	app *core.App
+	ctx context.Context
 }
 
 // listenRPC starts the RPC server and listens on config.RPCAddr. It blocks
 // until there is an error or the RPC server is closed.
-func listenRPC(app *core.App, config standaloneConfig, ctx context.Context) error {
+func listenRPC(ctx context.Context, app *core.App, config standaloneConfig) error {
 	// Initialize the JSON RPC WebSocket server (but don't start it yet).
 	rpcAddr := fmt.Sprintf("%s", config.RPCAddr)
 	rpcHandler := &rpcHandler{
 		app: app,
+		ctx: ctx,
 	}
 	rpcServer, err := rpc.NewServer(rpcAddr, rpcHandler)
 	if err != nil {
@@ -117,7 +119,7 @@ func (handler *rpcHandler) AddOrders(signedOrdersRaw []*json.RawMessage, opts rp
 			err = errors.New("method handler crashed in AddOrders RPC call (check logs for stack trace)")
 		}
 	}()
-	validationResults, err := handler.app.AddOrders(signedOrdersRaw, opts.Pinned)
+	validationResults, err := handler.app.AddOrders(handler.ctx, signedOrdersRaw, opts.Pinned)
 	if err != nil {
 		// We don't want to leak internal error details to the RPC client.
 		log.WithField("error", err.Error()).Error("internal error in AddOrders RPC call")

constants/constants.go

@@ -2,6 +2,7 @@ package constants
 
 import (
 	"errors"
+	"fmt"
 	"math/big"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -27,16 +28,16 @@ var (
 	// GanacheAccount0 is the first account exposed on the Ganache test Ethereum node
 	GanacheAccount0           = common.HexToAddress("0x5409ed021d9299bf6814279a6a1411a7e866a631")
 	ganacheAccount0PrivateKey = common.Hex2Bytes("f2f48ee19680706196e2e339e5da3491186e0c4c5030670656b0e0164837257d")
-	// GanacheAccount1 is the first account exposed on the Ganache test Ethereum node
+	// GanacheAccount1 is the second account exposed on the Ganache test Ethereum node
 	GanacheAccount1           = common.HexToAddress("0x6ecbe1db9ef729cbe972c83fb886247691fb6beb")
 	ganacheAccount1PrivateKey = common.Hex2Bytes("5d862464fe9303452126c8bc94274b8c5f9874cbd219789b3eb2128075a76f72")
-	// GanacheAccount2 is the first account exposed on the Ganache test Ethereum node
+	// GanacheAccount2 is the third account exposed on the Ganache test Ethereum node
 	GanacheAccount2           = common.HexToAddress("0xe36ea790bc9d7ab70c55260c66d52b1eca985f84")
 	ganacheAccount2PrivateKey = common.Hex2Bytes("df02719c4df8b9b8ac7f551fcb5d9ef48fa27eef7a66453879f4d8fdc6e78fb1")
-	// GanacheAccount3 is the first account exposed on the Ganache test Ethereum node
+	// GanacheAccount3 is the fourth account exposed on the Ganache test Ethereum node
 	GanacheAccount3           = common.HexToAddress("0xe834ec434daba538cd1b9fe1582052b880bd7e63")
 	ganacheAccount3PrivateKey = common.Hex2Bytes("ff12e391b79415e941a94de3bf3a9aee577aed0731e297d5cfa0b8a1e02fa1d0")
-	// GanacheAccount4 is the first account exposed on the Ganache test Ethereum node
+	// GanacheAccount4 is the fifth account exposed on the Ganache test Ethereum node
 	GanacheAccount4           = common.HexToAddress("0x78dc5d2d739606d31509c31d654056a45185ecb6")
 	ganacheAccount4PrivateKey = common.Hex2Bytes("752dd9cf65e68cfaba7d60225cbdbc1f4729dd5e5507def72815ed0d8abc6249")
 )
@@ -79,3 +80,6 @@ const MaxOrderSizeInBytes = 8192
 // node stores archive-level state. One cannot make `eth_call` requests specifying blocks earlier than
 // 128 blocks ago on non-archive nodes.
 const MaxBlocksStoredInNonArchiveNode = 128
+
+// ErrMaxMessageSize is the error emitted when a message exceeds it's max size
+var ErrMaxMessageSize = fmt.Errorf("message exceeds maximum size of %d bytes", MaxOrderSizeInBytes)

core/core.go

@@ -14,11 +14,12 @@ import (
 
 	"github.com/0xProject/0x-mesh/constants"
 	"github.com/0xProject/0x-mesh/db"
+	"github.com/0xProject/0x-mesh/encoding"
 	"github.com/0xProject/0x-mesh/ethereum"
 	"github.com/0xProject/0x-mesh/ethereum/blockwatch"
-	"github.com/0xProject/0x-mesh/ethereum/dbstack"
 	"github.com/0xProject/0x-mesh/ethereum/ethrpcclient"
 	"github.com/0xProject/0x-mesh/ethereum/ratelimit"
+	"github.com/0xProject/0x-mesh/ethereum/simplestack"
 	"github.com/0xProject/0x-mesh/expirationwatch"
 	"github.com/0xProject/0x-mesh/keys"
 	"github.com/0xProject/0x-mesh/loghooks"
@@ -250,10 +251,11 @@ func New(config Config) (*App, error) {
 		return nil, err
 	}
 	topics := orderwatch.GetRelevantTopics()
-	stack, err := dbstack.New(meshDB, blockWatcherRetentionLimit)
+	miniHeaders, err := meshDB.FindAllMiniHeadersSortedByNumber()
 	if err != nil {
 		return nil, err
 	}
+	stack := simplestack.New(blockWatcherRetentionLimit, miniHeaders)
 	blockWatcherConfig := blockwatch.Config{
 		Stack:           stack,
 		PollingInterval: config.BlockPollingInterval,
@@ -444,7 +446,7 @@ func (app *App) Start(ctx context.Context) error {
 	}()
 
 	// Note: this is a blocking call so we won't continue set up until its finished.
-	blocksElapsed, err := app.blockWatcher.SyncToLatestBlock(innerCtx)
+	blocksElapsed, err := app.blockWatcher.FastSyncToLatestBlock(innerCtx)
 	if err != nil {
 		return err
 	}
@@ -458,6 +460,13 @@ func (app *App) Start(ctx context.Context) error {
 		blockWatcherErrChan <- app.blockWatcher.Watch(innerCtx)
 	}()
 
+	// Ensure orderWatcher has processed at least one recent block before
+	// starting the P2P node and completing app start, so that Mesh does
+	// not validate any orders at outdated block heights
+	if err := app.orderWatcher.WaitForAtLeastOneBlockToBeProcessed(ctx); err != nil {
+		return err
+	}
+
 	if blocksElapsed >= constants.MaxBlocksStoredInNonArchiveNode {
 		log.WithField("blocksElapsed", blocksElapsed).Info("More than 128 blocks have elapsed since last boot. Re-validating all orders stored (this can take a while)...")
 		// Re-validate all orders since too many blocks have elapsed to fast-sync events
@@ -687,7 +696,7 @@ func (app *App) GetOrders(page, perPage int, snapshotID string) (*rpc.GetOrdersR
 // peers. If pinned is true, the orders will be marked as pinned, which means
 // they will only be removed if they become unfillable and will not be removed
 // due to having a high expiration time or any incentive mechanisms.
-func (app *App) AddOrders(signedOrdersRaw []*json.RawMessage, pinned bool) (*ordervalidator.ValidationResults, error) {
+func (app *App) AddOrders(ctx context.Context, signedOrdersRaw []*json.RawMessage, pinned bool) (*ordervalidator.ValidationResults, error) {
 	<-app.started
 
 	allValidationResults := &ordervalidator.ValidationResults{
@@ -752,39 +761,25 @@ func (app *App) AddOrders(signedOrdersRaw []*json.RawMessage, pinned bool) (*ord
 		orderHashesSeen[orderHash] = struct{}{}
 	}
 
-	validationResults, err := app.validateOrders(schemaValidOrders)
+	validationResults, err := app.orderWatcher.ValidateAndStoreValidOrders(ctx, schemaValidOrders, pinned, app.chainID)
 	if err != nil {
 		return nil, err
 	}
+
 	for _, orderInfo := range validationResults.Accepted {
 		allValidationResults.Accepted = append(allValidationResults.Accepted, orderInfo)
 	}
 	for _, orderInfo := range validationResults.Rejected {
 		allValidationResults.Rejected = append(allValidationResults.Rejected, orderInfo)
 	}
 
-	for i, acceptedOrderInfo := range allValidationResults.Accepted {
+	for _, acceptedOrderInfo := range allValidationResults.Accepted {
 		// If the order isn't new, we don't add to OrderWatcher, log it's receipt
 		// or share the order with peers.
 		if !acceptedOrderInfo.IsNew {
 			continue
 		}
-		// Add the order to the OrderWatcher. This also saves the order in the
-		// database.
-		err = app.orderWatcher.Add(acceptedOrderInfo, pinned)
-		if err != nil {
-			if err == meshdb.ErrDBFilledWithPinnedOrders {
-				allValidationResults.Accepted = append(allValidationResults.Accepted[:i], allValidationResults.Accepted[i+1:]...)
-				allValidationResults.Rejected = append(allValidationResults.Rejected, &ordervalidator.RejectedOrderInfo{
-					OrderHash:   acceptedOrderInfo.OrderHash,
-					SignedOrder: acceptedOrderInfo.SignedOrder,
-					Kind:        ordervalidator.MeshError,
-					Status:      ordervalidator.RODatabaseFullOfOrders,
-				})
-			} else {
-				return nil, err
-			}
-		}
+
 		log.WithFields(log.Fields{
 			"orderHash": acceptedOrderInfo.OrderHash.String(),
 		}).Debug("added new valid order via RPC or browser callback")
@@ -794,14 +789,15 @@ func (app *App) AddOrders(signedOrdersRaw []*json.RawMessage, pinned bool) (*ord
 			return nil, err
 		}
 	}
+
 	return allValidationResults, nil
 }
 
 // shareOrder immediately shares the given order on the GossipSub network.
 func (app *App) shareOrder(order *zeroex.SignedOrder) error {
 	<-app.started
 
-	encoded, err := encodeOrder(order)
+	encoded, err := encoding.OrderToRawMessage(order)
 	if err != nil {
 		return err
 	}
@@ -819,16 +815,13 @@ func (app *App) AddPeer(peerInfo peerstore.PeerInfo) error {
 func (app *App) GetStats() (*rpc.GetStatsResponse, error) {
 	<-app.started
 
-	latestBlockHeader, err := app.blockWatcher.GetLatestBlockProcessed()
+	latestBlockHeader, err := app.db.FindLatestMiniHeader()
 	if err != nil {
 		return nil, err
 	}
-	var latestBlock rpc.LatestBlock
-	if latestBlockHeader != nil {
-		latestBlock = rpc.LatestBlock{
-			Number: int(latestBlockHeader.Number.Int64()),
-			Hash:   latestBlockHeader.Hash,
-		}
+	latestBlock := rpc.LatestBlock{
+		Number: int(latestBlockHeader.Number.Int64()),
+		Hash:   latestBlockHeader.Hash,
 	}
 	notRemovedFilter := app.db.Orders.IsRemovedIndex.ValueFilter([]byte{0})
 	numOrders, err := app.db.Orders.NewQuery(notRemovedFilter).Count()

core/message_handler.go

@@ -1,7 +1,10 @@
 package core
 
 import (
+	"context"
+
 	"github.com/0xProject/0x-mesh/constants"
+	"github.com/0xProject/0x-mesh/encoding"
 	"github.com/0xProject/0x-mesh/meshdb"
 	"github.com/0xProject/0x-mesh/p2p"
 	"github.com/0xProject/0x-mesh/zeroex"
@@ -92,7 +95,7 @@ func (orderSelector *orderSelector) GetMessagesToShare(max int) ([][]byte, error
 		log.WithFields(map[string]interface{}{
 			"order": order,
 		}).Trace("selected order to share")
-		encoded, err := encodeOrder(order.SignedOrder)
+		encoded, err := encoding.OrderToRawMessage(order.SignedOrder)
 		if err != nil {
 			return nil, err
 		}
@@ -101,7 +104,7 @@ func (orderSelector *orderSelector) GetMessagesToShare(max int) ([][]byte, error
 	return messageData, nil
 }
 
-func (app *App) HandleMessages(messages []*p2p.Message) error {
+func (app *App) HandleMessages(ctx context.Context, messages []*p2p.Message) error {
 	// First we validate the messages and decode them into orders.
 	orders := []*zeroex.SignedOrder{}
 	orderHashToMessage := map[common.Hash]*p2p.Message{}
@@ -140,7 +143,7 @@ func (app *App) HandleMessages(messages []*p2p.Message) error {
 			continue
 		}
 
-		order, err := decodeOrder(msg.Data)
+		order, err := encoding.RawMessageToOrder(msg.Data)
 		if err != nil {
 			log.WithFields(map[string]interface{}{
 				"error": err,
@@ -164,20 +167,21 @@ func (app *App) HandleMessages(messages []*p2p.Message) error {
 	}
 
 	// Next, we validate the orders.
-	validationResults, err := app.validateOrders(orders)
+	validationResults, err := app.orderWatcher.ValidateAndStoreValidOrders(ctx, orders, false, app.chainID)
 	if err != nil {
 		return err
 	}
 
 	// Store any valid orders and update the peer scores.
 	for _, acceptedOrderInfo := range validationResults.Accepted {
+		// If the order isn't new, we don't log it's receipt or adjust peer scores
 		if !acceptedOrderInfo.IsNew {
 			continue
 		}
 		msg := orderHashToMessage[acceptedOrderInfo.OrderHash]
-		// If we've reached this point, the message is valid and we were able to
-		// decode it into an order. Append it to the list of orders to validate and
-		// update peer scores accordingly.
+		// If we've reached this point, the message is valid, we were able to
+		// decode it into an order and check that this order is valid. Update
+		// peer scores accordingly.
 		log.WithFields(map[string]interface{}{
 			"orderHash": acceptedOrderInfo.OrderHash.Hex(),
 			"from":      msg.From.String(),
@@ -187,20 +191,6 @@ func (app *App) HandleMessages(messages []*p2p.Message) error {
 			"orderHash": acceptedOrderInfo.OrderHash.Hex(),
 			"from":      msg.From.String(),
 		}).Trace("all fields for new valid order received from peer")
-		// Add stores the message in the database.
-		if err := app.orderWatcher.Add(acceptedOrderInfo, false); err != nil {
-			if err == meshdb.ErrDBFilledWithPinnedOrders {
-				// If the database is full of pinned orders, log and then continue.
-				log.WithFields(map[string]interface{}{
-					"error":     err.Error(),
-					"orderHash": acceptedOrderInfo.OrderHash.Hex(),
-					"from":      msg.From.String(),
-				}).Error("could not store valid order because database is full")
-				continue
-			}
-			// For any other type of error, return it.
-			return err
-		}
 		app.handlePeerScoreEvent(msg.From, psOrderStored)
 	}
 
@@ -213,7 +203,7 @@ func (app *App) HandleMessages(messages []*p2p.Message) error {
 			"from":              msg.From.String(),
 		}).Trace("not storing rejected order received from peer")
 		switch rejectedOrderInfo.Status {
-		case ordervalidator.ROInternalError, ordervalidator.ROEthRPCRequestFailed, ordervalidator.ROCoordinatorRequestFailed:
+		case ordervalidator.ROInternalError, ordervalidator.ROEthRPCRequestFailed, ordervalidator.ROCoordinatorRequestFailed, ordervalidator.RODatabaseFullOfOrders:
 			// Don't incur a negative score for these status types (it might not be
 			// their fault).
 		default:

core/message_handler_test.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/0xProject/0x-mesh/constants"
+	"github.com/0xProject/0x-mesh/encoding"
 	"github.com/0xProject/0x-mesh/ethereum"
 	"github.com/0xProject/0x-mesh/meshdb"
 	"github.com/0xProject/0x-mesh/scenario"
@@ -191,7 +192,7 @@ func verifyRoundRobinSharing(t *testing.T, selector *orderSelector, nextOffset i
 
 	// Calculate the orders that we expect to be shared
 	for i := 0; i < expectedOrdersLength; i++ {
-		encodedOrder, err := encodeOrder(orderList[(nextOffset+i)%count].SignedOrder)
+		encodedOrder, err := encoding.OrderToRawMessage(orderList[(nextOffset+i)%count].SignedOrder)
 		require.NoError(t, err)
 
 		expectedOrders[i] = encodedOrder