Merge pull request #552 from 0xProject/fix/always-unsubscribe-order-feed

Always unsubscribe from order events in SetupOrderStream

Author: albrow

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 This changelog is a work in progress and may contain notes for versions which have not actually been released. Check the [Releases](https://github.com/0xProject/0x-mesh/releases) page to see full release notes and more information about the latest released versions.
 
+## v6.1.1-beta
+
+### Bug fixes 🐞
+
+- Fixed a bug where the internal order event feed could become blocked, rendering Mesh unable to receive any new orders or update existing ones ([#552](https://github.com/0xProject/0x-mesh/pull/552)).
+
+
 ## v6.1.0-beta
 
 ### Features ✅

cmd/mesh/rpc_handler.go

@@ -22,6 +22,10 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
+// orderEventsBufferSize is the buffer size for the orderEvents channel. If
+// the buffer is full, any additional events won't be processed.
+const orderEventsBufferSize = 8000
+
 type rpcHandler struct {
 	app *core.App
 }
@@ -209,8 +213,9 @@ func SetupOrderStream(ctx context.Context, app *core.App) (*ethrpc.Subscription,
 	rpcSub := notifier.CreateSubscription()
 
 	go func() {
-		orderEventsChan := make(chan []*zeroex.OrderEvent)
+		orderEventsChan := make(chan []*zeroex.OrderEvent, orderEventsBufferSize)
 		orderWatcherSub := app.SubscribeToOrderEvents(orderEventsChan)
+		defer orderWatcherSub.Unsubscribe()
 
 		for {
 			select {
@@ -237,7 +242,6 @@ func SetupOrderStream(ctx context.Context, app *core.App) (*ethrpc.Subscription,
 					// error.
 					if _, ok := err.(*net.OpError); ok {
 						logEntry.Trace(message)
-						orderWatcherSub.Unsubscribe()
 						return
 					}
 					if strings.Contains(err.Error(), "write: broken pipe") {
@@ -249,13 +253,11 @@ func SetupOrderStream(ctx context.Context, app *core.App) (*ethrpc.Subscription,
 			case err := <-rpcSub.Err():
 				if err != nil {
 					log.WithField("err", err).Error("rpcSub returned an error")
-					orderWatcherSub.Unsubscribe()
 				} else {
 					log.Debug("rpcSub was closed without error")
 				}
 				return
 			case <-notifier.Closed():
-				orderWatcherSub.Unsubscribe()
 				return
 			}
 		}

rpc/service.go

@@ -114,7 +114,6 @@ func SetupHeartbeat(ctx context.Context) (*ethrpc.Subscription, error) {
 
 			// Wait MinCleanupInterval before emitting the next heartbeat.
 			time.Sleep(minHeartbeatInterval - time.Since(start))
-
 		}
 	}()