Resolve deadlock caused by fetching snapshot during update

streaming/streaming.go

@@ -304,57 +304,74 @@ func (c *Conn) receivedOrderBook(ob orderBook) error {
 }
 
 func (c *Conn) receivedUpdate(u Update) error {
+	valid, err := c.processUpdate(u)
+	if err != nil {
+		return err
+	}
+
+	// If update is not valid, ignore
+	if !valid {
+		return nil
+	}
+
+	if c.updateCallback != nil {
+		c.updateCallback(u)
+	}
+
+	return nil
+}
+
+// Validate and process update into orderbook.
+// Return bool indicating if update is valid.
+func (c *Conn) processUpdate(u Update) (bool, error) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
 	if c.seq == 0 {
 		// State not initialized so we can't update it.
-		return nil
+		return false, nil
 	}
 
 	if u.Sequence <= c.seq {
 		// Old update. We can just discard it.
-		return nil
+		return false, nil
 	}
+
 	if u.Sequence != c.seq+1 {
-		return errors.New("streaming: update received out of sequence")
+		return false, errors.New("streaming: update received out of sequence")
 	}
 
 	// Process trades
 	for _, t := range u.TradeUpdates {
 		if err := c.processTrade(*t); err != nil {
-			return err
+			return false, err
 		}
 	}
 
 	// Process create
 	if u.CreateUpdate != nil {
 		if err := c.processCreate(*u.CreateUpdate); err != nil {
-			return err
+			return false, err
 		}
 	}
 
 	// Process delete
 	if u.DeleteUpdate != nil {
 		if err := c.processDelete(*u.DeleteUpdate); err != nil {
-			return err
+			return false, err
 		}
 	}
 
 	// Process status
 	if u.StatusUpdate != nil {
 		if err := c.processStatus(*u.StatusUpdate); err != nil {
-			return err
+			return false, err
 		}
 	}
 
 	c.seq = u.Sequence
 
-	if c.updateCallback != nil {
-		c.updateCallback(u)
-	}
-
-	return nil
+	return true, nil
 }
 
 func decTrade(m map[string]order, id string, base decimal.Decimal) (
@@ -483,7 +500,7 @@ func (c *Conn) Snapshot() Snapshot {
 	}
 }
 
-// Status returns the currenct status of the streaming connection.
+// Status returns the current status of the streaming connection.
 func (c *Conn) Status() luno.Status {
 	c.mu.RLock()
 	defer c.mu.RUnlock()

streaming/streaming_internal_test.go

@@ -2,6 +2,7 @@ package streaming
 
 import (
 	"testing"
+	"time"
 
 	"github.com/luno/luno-go"
 	"github.com/luno/luno-go/decimal"
@@ -583,3 +584,52 @@ func compareOrderBookEntry(t *testing.T, want, got luno.OrderBookEntry) {
 		t.Errorf("got = %v, want %v", got, want)
 	}
 }
+
+func TestReceiveUpdateSnapshot(t *testing.T) {
+	chDone := make(chan struct{})
+
+	c := &Conn{
+		asks:   asksMap(),
+		bids:   bidsMap(),
+		seq:    1,
+		status: luno.StatusActive,
+	}
+
+	onUpdate := func(up Update) {
+		// Get snapshot to confirm mutex does not create deadlock
+		_ = c.Snapshot()
+		chDone <- struct{}{}
+	}
+
+	c.updateCallback = onUpdate
+
+	tu := []*TradeUpdate{
+		{
+			Sequence:     2,
+			Base:         decimal.NewFromFloat64(0.02, 2),
+			Counter:      decimal.NewFromFloat64(0.002, 2),
+			MakerOrderID: "1",
+			TakerOrderID: "32",
+		},
+		{
+			Sequence:     3,
+			Base:         decimal.NewFromFloat64(0.01, 2),
+			Counter:      decimal.NewFromFloat64(0.001, 2),
+			MakerOrderID: "1",
+			TakerOrderID: "34",
+		},
+	}
+
+	go func() {
+		err := c.receivedUpdate(Update{Sequence: 2, TradeUpdates: tu})
+		if err != nil {
+			t.Errorf("Expected success got: %v", err)
+		}
+	}()
+
+	select {
+	case <-chDone:
+	case <-time.After(time.Second):
+		t.Errorf("timeout trying to retrieve snapshot on update")
+	}
+}