multi: handle trade suspension messages.

client/core/core.go

@@ -76,6 +76,48 @@ type dexConnection struct {
 
 	regConfMtx  sync.RWMutex
 	regConfirms uint32
+
+	pendingSuspendsMtx sync.Mutex
+	pendingSuspends    map[string]*time.Timer
+}
+
+// suspended returns the suspended status of the provided market.
+func (dc *dexConnection) suspended(mkt string) bool {
+	dc.marketMtx.Lock()
+	defer dc.marketMtx.Unlock()
+	market, ok := dc.marketMap[mkt]
+	if !ok {
+		return false
+	}
+	return market.Suspended
+}
+
+// suspend halts trading for the provided market.
+func (dc *dexConnection) suspend(mkt string) error {
+	dc.marketMtx.Lock()
+	defer dc.marketMtx.Unlock()
+
+	market, ok := dc.marketMap[mkt]
+	if !ok {
+		return fmt.Errorf("no market found with ID %s", mkt)
+	}
+
+	market.Suspended = true
+	return nil
+}
+
+// resume commences trading for the provided market.
+func (dc *dexConnection) resume(mkt string) error {
+	dc.marketMtx.Lock()
+	defer dc.marketMtx.Unlock()
+
+	market, ok := dc.marketMap[mkt]
+	if !ok {
+		return fmt.Errorf("no market found with ID %s", mkt)
+	}
+
+	market.Suspended = false
+	return nil
 }
 
 // refreshMarkets rebuilds, saves, and returns the market map. The map itself
@@ -98,6 +140,7 @@ func (dc *dexConnection) refreshMarkets() map[string]*Market {
 			EpochLen:        mkt.EpochLen,
 			StartEpoch:      mkt.StartEpoch,
 			MarketBuyBuffer: mkt.MarketBuyBuffer,
+			Suspended:       dc.suspended(mkt.Name),
 		}
 		mid := market.marketName()
 		dc.tradeMtx.RLock()
@@ -1467,6 +1510,12 @@ func (c *Core) Trade(pw []byte, form *TradeForm) (*Order, error) {
 		return nil, fmt.Errorf("order placed for unknown market")
 	}
 
+	// Proceed with the order if there is no trade suspension
+	// scheduled for the market.
+	if dc.suspended(mktID) {
+		return nil, fmt.Errorf("suspended market")
+	}
+
 	rate, qty := form.Rate, form.Qty
 	if form.IsLimit && rate == 0 {
 		return nil, fmt.Errorf("zero-rate order not allowed")
@@ -2310,17 +2359,18 @@ func (c *Core) connectDEX(acctInfo *db.AccountInfo) (*dexConnection, error) {
 
 	// Create the dexConnection and listen for incoming messages.
 	dc := &dexConnection{
-		WsConn:     conn,
-		connMaster: connMaster,
-		assets:     assets,
-		cfg:        dexCfg,
-		books:      make(map[string]*bookie),
-		acct:       newDEXAccount(acctInfo),
-		marketMap:  marketMap,
-		trades:     make(map[order.OrderID]*trackedTrade),
-		notify:     c.notify,
-		epoch:      epochMap,
-		connected:  true,
+		WsConn:          conn,
+		connMaster:      connMaster,
+		assets:          assets,
+		cfg:             dexCfg,
+		books:           make(map[string]*bookie),
+		acct:            newDEXAccount(acctInfo),
+		marketMap:       marketMap,
+		trades:          make(map[order.OrderID]*trackedTrade),
+		notify:          c.notify,
+		epoch:           epochMap,
+		connected:       true,
+		pendingSuspends: make(map[string]*time.Timer),
 	}
 
 	dc.refreshMarkets()
@@ -2422,6 +2472,92 @@ func handleRevokeMatchMsg(_ *Core, dc *dexConnection, msg *msgjson.Message) erro
 	return tracker.db.UpdateOrder(metaOrder)
 }
 
+// handleTradeSuspensionMsg is called when a trade suspension notification is received.
+func handleTradeSuspensionMsg(c *Core, dc *dexConnection, msg *msgjson.Message) error {
+	var sp msgjson.TradeSuspension
+	err := msg.Unmarshal(&sp)
+	if err != nil {
+		return fmt.Errorf("trade suspension unmarshal error: %v", err)
+	}
+
+	dc.pendingSuspendsMtx.Lock()
+	defer dc.pendingSuspendsMtx.Unlock()
+
+	// Ensure the provided market exists for the dex.
+	dc.marketMtx.RLock()
+	mkt, ok := dc.marketMap[sp.MarketID]
+	dc.marketMtx.RUnlock()
+	if !ok {
+		return fmt.Errorf("no market found with ID %s", sp.MarketID)
+	}
+
+	// Ensure the market is not already suspended.
+	if mkt.Suspended {
+		return fmt.Errorf("market %s for dex %s is already suspended",
+			sp.MarketID, dc.acct.host)
+	}
+
+	// Remove pending suspends for the market.
+	if sched := dc.pendingSuspends[sp.MarketID]; sched != nil {
+		if !sched.Stop() {
+			// TODO: too late, timer already fired. Need to request the
+			// current configuration for the market at this point.
+			return fmt.Errorf("unable to stop previously scheduled "+
+				"suspend for market %s on dex %s", sp.MarketID, dc.acct.host)
+		}
+	}
+
+	// Set the new scheduled suspend.
+	duration := time.Until(encode.UnixTimeMilli(int64(sp.SuspendTime)))
+	dc.pendingSuspends[sp.MarketID] = time.AfterFunc(duration, func() {
+		// Update the market as suspended.
+		err := dc.suspend(sp.MarketID)
+		if err != nil {
+			log.Error(err)
+			return
+		}
+
+		// Clear the bookie associated with the suspended market.
+		dc.booksMtx.Lock()
+		if bookie := dc.books[sp.MarketID]; bookie != nil {
+			dc.books[sp.MarketID] = newBookie(func() { c.unsub(dc, sp.MarketID) })
+		}
+		dc.booksMtx.Unlock()
+
+		if !sp.Persist {
+			// Revoke all active orders of the suspended market for the dex.
+			dc.tradeMtx.RLock()
+			for _, tracker := range dc.trades {
+				if tracker.Order.Base() == mkt.BaseID &&
+					tracker.Order.Quote() == mkt.QuoteID &&
+					tracker.metaData.Host == dc.acct.host &&
+					(tracker.metaData.Status == order.OrderStatusEpoch ||
+						tracker.metaData.Status == order.OrderStatusBooked) {
+					md := tracker.metaData
+					md.Status = order.OrderStatusRevoked
+					metaOrder := &db.MetaOrder{
+						MetaData: md,
+						Order:    tracker.Order,
+					}
+
+					err = tracker.db.UpdateOrder(metaOrder)
+					if err != nil {
+						log.Errorf("unable to update order: %v", err)
+					}
+				}
+			}
+			dc.tradeMtx.RUnlock()
+		}
+
+		// Remove suspend after execution.
+		dc.pendingSuspendsMtx.Lock()
+		delete(dc.pendingSuspends, sp.MarketID)
+		dc.pendingSuspendsMtx.Unlock()
+	})
+
+	return nil
+}
+
 // routeHandler is a handler for a message from the DEX.
 type routeHandler func(*Core, *dexConnection, *msgjson.Message) error
 
@@ -2431,7 +2567,6 @@ var reqHandlers = map[string]routeHandler{
 	msgjson.AuditRoute:       handleAuditRoute,
 	msgjson.RedemptionRoute:  handleRedemptionRoute,
 	msgjson.RevokeMatchRoute: handleRevokeMatchMsg,
-	msgjson.SuspensionRoute:  nil,
 }
 
 var noteHandlers = map[string]routeHandler{
@@ -2440,6 +2575,7 @@ var noteHandlers = map[string]routeHandler{
 	msgjson.EpochOrderRoute:      handleEpochOrderMsg,
 	msgjson.UnbookOrderRoute:     handleUnbookOrderMsg,
 	msgjson.UpdateRemainingRoute: handleUpdateRemainingMsg,
+	msgjson.SuspensionRoute:      handleTradeSuspensionMsg,
 }
 
 // listen monitors the DEX websocket connection for server requests and

client/core/core_test.go

@@ -150,6 +150,7 @@ func testDexConnection() (*dexConnection, *TWebsocket, *dexAccount) {
 		QuoteSymbol:     tBTC.Symbol,
 		EpochLen:        60000,
 		MarketBuyBuffer: 1.1,
+		Suspended:       false,
 	}
 	return &dexConnection{
 		WsConn:     conn,
@@ -178,10 +179,11 @@ func testDexConnection() (*dexConnection, *TWebsocket, *dexAccount) {
 			},
 			Fee: tFee,
 		},
-		notify:    func(Notification) {},
-		marketMap: map[string]*Market{tDcrBtcMktName: mkt},
-		trades:    make(map[order.OrderID]*trackedTrade),
-		epoch:     map[string]uint64{tDcrBtcMktName: 0},
+		notify:          func(Notification) {},
+		marketMap:       map[string]*Market{tDcrBtcMktName: mkt},
+		trades:          make(map[order.OrderID]*trackedTrade),
+		epoch:           map[string]uint64{tDcrBtcMktName: 0},
+		pendingSuspends: make(map[string]*time.Timer),
 	}, conn, acct
 }
 
@@ -2942,3 +2944,73 @@ func TestAssetCounter(t *testing.T) {
 		t.Fatalf("absorbed counts not combined correctly")
 	}
 }
+
+func TestHandleTradeSuspensionMsg(t *testing.T) {
+	rig := newTestRig()
+
+	// Ensure a non-existent market cannot be suspended.
+	payload := &msgjson.TradeSuspension{
+		MarketID:    "dcr_dcr",
+		FinalEpoch:  100,
+		SuspendTime: encode.UnixMilliU(time.Now().Add(time.Second * 2)),
+		Persist:     true,
+	}
+
+	req, _ := msgjson.NewRequest(rig.dc.NextID(), msgjson.SuspensionRoute, payload)
+	err := handleTradeSuspensionMsg(rig.core, rig.dc, req)
+	if err == nil {
+		t.Fatal("[handleTradeSuspensionMsg] expected a market " +
+			"ID not found error: %v")
+	}
+
+	mkt := "dcr_btc"
+
+	// Ensure an already suspended market cannot be suspended again.
+	err = rig.dc.suspend(mkt)
+	if err != nil {
+		t.Fatalf("[handleTradeSuspensionMsg] unexpected error: %v", err)
+	}
+
+	payload = &msgjson.TradeSuspension{
+		MarketID:    mkt,
+		FinalEpoch:  100,
+		SuspendTime: encode.UnixMilliU(time.Now().Add(time.Second * 2)),
+		Persist:     true,
+	}
+
+	req, _ = msgjson.NewRequest(rig.dc.NextID(), msgjson.SuspensionRoute, payload)
+	err = handleTradeSuspensionMsg(rig.core, rig.dc, req)
+	if err == nil {
+		t.Fatal("[handleTradeSuspensionMsg] expected a market " +
+			"suspended error: %v")
+	}
+
+	// Suspend a market.
+	err = rig.dc.resume(mkt)
+	if err != nil {
+		t.Fatalf("[handleTradeSuspensionMsg] unexpected error: %v", err)
+	}
+
+	req, _ = msgjson.NewRequest(rig.dc.NextID(), msgjson.SuspensionRoute, payload)
+	err = handleTradeSuspensionMsg(rig.core, rig.dc, req)
+	if err != nil {
+		t.Fatalf("[handleTradeSuspensionMsg] unexpected error: %v", err)
+	}
+
+	// Ensure trades for a suspended market generate an error.
+	form := &TradeForm{
+		Host:    tDexHost,
+		IsLimit: true,
+		Sell:    true,
+		Base:    tDCR.ID,
+		Quote:   tBTC.ID,
+		Qty:     tDCR.LotSize * 10,
+		Rate:    tBTC.RateStep * 1000,
+		TifNow:  false,
+	}
+
+	_, err = rig.core.Trade(tPW, form)
+	if err == nil {
+		t.Fatalf("expected a trade suspension set error")
+	}
+}

client/core/types.go

@@ -158,6 +158,7 @@ type Market struct {
 	StartEpoch      uint64   `json:"startepoch"`
 	MarketBuyBuffer float64  `json:"buybuffer"`
 	Orders          []*Order `json:"orders"`
+	Suspended       bool     `json:"suspended"`
 }
 
 // Display returns an ID string suitable for displaying in a UI.

client/order/bookside.go

@@ -131,6 +131,15 @@ func (d *bookSide) orders() []*Order {
 	return orders
 }
 
+// Reset clears out the book side.
+func (d *bookSide) Reset() {
+	d.mtx.Lock()
+	defer d.mtx.Unlock()
+
+	d.bins = make(map[uint64][]*Order)
+	d.rateIndex = NewRateIndex()
+}
+
 // BestNOrders returns the best N orders of the book side.
 func (d *bookSide) BestNOrders(n int) ([]*Order, bool) {
 	d.mtx.RLock()