expire orders by price level

fix & test

format

extract preference size as param

format:

refactor updator

fix format

rename arg

refactor

app/app.go

@@ -263,6 +263,7 @@ func SetUpgradeConfig(upgradeConfig *config.UpgradeConfig) {
 	upgrade.Mgr.AddUpgradeHeight(upgrade.LotSizeOptimization, upgradeConfig.LotSizeUpgradeHeight)
 	upgrade.Mgr.AddUpgradeHeight(upgrade.ListingRuleUpgrade, upgradeConfig.ListingRuleUpgradeHeight)
 	upgrade.Mgr.AddUpgradeHeight(upgrade.FixZeroBalance, upgradeConfig.FixZeroBalanceHeight)
+	upgrade.Mgr.AddUpgradeHeight(upgrade.BEP67, upgradeConfig.BEP67Height)
 
 	// register store keys of upgrade
 	upgrade.Mgr.RegisterStoreKeys(upgrade.BEP9, common.TimeLockStoreKey.Name())

app/config/config.go

@@ -67,6 +67,8 @@ LotSizeUpgradeHeight = {{ .UpgradeConfig.LotSizeUpgradeHeight }}
 ListingRuleUpgradeHeight = {{ .UpgradeConfig.ListingRuleUpgradeHeight }}
 # Block height of FixZeroBalanceHeight upgrade
 FixZeroBalanceHeight = {{ .UpgradeConfig.FixZeroBalanceHeight }}
+# Block height of BEP67 upgrade
+BEP67Height = {{ .UpgradeConfig.BEP67Height }}
 
 [query]
 # ABCI query interface black list, suggested value: ["custom/gov/proposals", "custom/timelock/timelocks", "custom/atomicSwap/swapcreator", "custom/atomicSwap/swaprecipient"]
@@ -364,6 +366,8 @@ type UpgradeConfig struct {
 	LotSizeUpgradeHeight       int64 `mapstructure:"LotSizeUpgradeHeight"`
 	ListingRuleUpgradeHeight   int64 `mapstructure:"ListingRuleUpgradeHeight"`
 	FixZeroBalanceHeight       int64 `mapstructure:"FixZeroBalanceHeight"`
+
+	BEP67Height int64 `mapstructure:"BEP67Height"`
 }
 
 func defaultUpgradeConfig() *UpgradeConfig {
@@ -379,6 +383,7 @@ func defaultUpgradeConfig() *UpgradeConfig {
 		LotSizeUpgradeHeight:       math.MaxInt64,
 		ListingRuleUpgradeHeight:   math.MaxInt64,
 		FixZeroBalanceHeight:       math.MaxInt64,
+		BEP67Height:                1,
 	}
 }
 

common/upgrade/upgrade.go

@@ -22,6 +22,8 @@ const (
 	LotSizeOptimization  = "LotSizeOptimization"
 	ListingRuleUpgrade   = "ListingRuleUpgrade" // Remove restriction that only the owner of base asset can list trading pair
 	FixZeroBalance       = "FixZeroBalance"
+
+	BEP67 = "BEP67" // https://github.com/binance-chain/BEPs/pull/67
 )
 
 func UpgradeBEP10(before func(), after func()) {

plugins/dex/matcheng/orderbook.go

@@ -20,7 +20,8 @@ type OrderBookInterface interface {
 	GetOrder(id string, side int8, price int64) (OrderPart, error)
 	RemoveOrder(id string, side int8, price int64) (OrderPart, error)
 	RemoveOrders(beforeTime int64, side int8, cb func(OrderPart)) error
-	UpdateForEachPriceLevel(side int8, updater func(*PriceLevel))
+	RemoveOrdersBasedOnPriceLevel(expireTime int64, forceExpireTime int64, priceLevelsToReserve int, side int8, removeCallback func(ord OrderPart)) error
+	UpdateForEachPriceLevel(side int8, updater LevelIter)
 	GetPriceLevel(price int64, side int8) *PriceLevel
 	RemovePriceLevel(price int64, side int8) int
 	ShowDepth(maxLevels int, iterBuy LevelIter, iterSell LevelIter)
@@ -130,14 +131,26 @@ func (ob *OrderBookOnULList) RemoveOrder(id string, side int8, price int64) (Ord
 }
 
 func (ob *OrderBookOnULList) RemoveOrders(beforeTime int64, side int8, cb func(OrderPart)) error {
-	ob.UpdateForEachPriceLevel(side, func(pl *PriceLevel) {
+	ob.UpdateForEachPriceLevel(side, func(pl *PriceLevel, levelIndex int) {
 		pl.removeOrders(beforeTime, cb)
 	})
 
 	return nil
 }
 
-func (ob *OrderBookOnULList) UpdateForEachPriceLevel(side int8, updater func(*PriceLevel)) {
+// order beyond priceLevelsToReserve will be expired if it's placed before expireTime. All orders will be expired if they are placed before forceExpireTime
+func (ob *OrderBookOnULList) RemoveOrdersBasedOnPriceLevel(expireTime int64, forceExpireTime int64, priceLevelsToReserve int, side int8, removeCallback func(ord OrderPart)) error {
+	ob.UpdateForEachPriceLevel(side, func(pl *PriceLevel, levelIndex int) {
+		if levelIndex < priceLevelsToReserve {
+			pl.removeOrders(forceExpireTime, removeCallback)
+		} else {
+			pl.removeOrders(expireTime, removeCallback)
+		}
+	})
+	return nil
+}
+
+func (ob *OrderBookOnULList) UpdateForEachPriceLevel(side int8, updater LevelIter) {
 	q := ob.getSideQueue(side)
 	q.UpdateForEach(updater)
 }
@@ -174,11 +187,11 @@ func (ob *OrderBookOnULList) GetAllLevels() ([]PriceLevel, []PriceLevel) {
 	buys := make([]PriceLevel, 0, ob.buyQueue.capacity)
 	sells := make([]PriceLevel, 0, ob.sellQueue.capacity)
 	ob.buyQueue.Iterate(ob.buyQueue.capacity,
-		func(p *PriceLevel) {
+		func(p *PriceLevel, levelIndex int) {
 			buys = append(buys, *p)
 		})
 	ob.sellQueue.Iterate(ob.sellQueue.capacity,
-		func(p *PriceLevel) {
+		func(p *PriceLevel, levelIndex int) {
 			sells = append(sells, *p)
 		})
 	return buys, sells

plugins/dex/matcheng/orderbook_test.go

@@ -558,6 +558,32 @@ func TestOrderBookOnULList_RemoveOrders(t *testing.T) {
 	require.Len(t, sells, 0)
 }
 
+func TestOrderBookOnULList_RemoveOrdersBiasedly(t *testing.T) {
+	book := NewOrderBookOnULList(16, 4)
+	book.InsertOrder("1", BUYSIDE, 10000, 1000, 10000)
+	book.InsertOrder("2", BUYSIDE, 10001, 1000, 10000)
+	book.InsertOrder("3", BUYSIDE, 10002, 1000, 10000)
+	err := book.RemoveOrdersBasedOnPriceLevel(9999, 10001, 10, BUYSIDE, nil)
+	require.NoError(t, err)
+	buys, sells := book.GetAllLevels()
+	//t.Log(buys)
+	//t.Log(sells)
+	require.Len(t, buys, 1)
+	require.Len(t, buys[0].Orders, 2)
+	require.Len(t, sells, 0)
+	err = book.RemoveOrdersBasedOnPriceLevel(10003, 10001, 10, BUYSIDE, nil)
+	require.NoError(t, err)
+	buys, sells = book.GetAllLevels()
+	require.Len(t, buys, 1)
+	require.Len(t, buys[0].Orders, 2)
+	require.Len(t, sells, 0)
+	err = book.RemoveOrdersBasedOnPriceLevel(10003, 10003, 10, BUYSIDE, nil)
+	require.NoError(t, err)
+	buys, sells = book.GetAllLevels()
+	require.Len(t, buys, 0)
+	require.Len(t, sells, 0)
+}
+
 func TestOrderBookOnULList_GetOverlappedRange(t *testing.T) {
 	overlap := make([]OverLappedLevel, 4)
 	buyBuf := make([]PriceLevel, 16)

plugins/dex/matcheng/types.go

@@ -190,7 +190,7 @@ func (overlapped *OverLappedLevel) HasSellTaker() bool {
 	return overlapped.SellTakerStartIdx < len(overlapped.SellOrders)
 }
 
-type LevelIter func(*PriceLevel)
+type LevelIter func(priceLevel *PriceLevel, levelIndex int)
 
 type MergedPriceLevel struct {
 	price    int64

plugins/dex/matcheng/unrolledlinkedlist.go

@@ -332,7 +332,7 @@ func (ull *ULList) Iterate(levelNum int, iter LevelIter) {
 			if curLevel >= levelNum {
 				return
 			}
-			iter(&b.elements[i])
+			iter(&b.elements[i], curLevel)
 			curLevel += 1
 		}
 	}
@@ -395,6 +395,7 @@ func (ull *ULList) GetPriceLevel(p int64) *PriceLevel {
 func (ull *ULList) UpdateForEach(updater LevelIter) {
 	b := ull.begin
 	var last *bucket
+	levelIndex := 0
 	for b != ull.dend {
 		for i := 0; ; {
 			k := len(b.elements)
@@ -404,7 +405,8 @@ func (ull *ULList) UpdateForEach(updater LevelIter) {
 				break
 			}
 			pl := &b.elements[i]
-			updater(pl)
+			updater(pl, levelIndex)
+			levelIndex++
 			if len(pl.Orders) == 0 {
 				b.deleteElement(i)
 			} else {

plugins/dex/matcheng/unrolledlinkedlist_test.go

@@ -453,7 +453,7 @@ func TestULList_UpdateForEach(t *testing.T) {
 	l.AddPriceLevel(&PriceLevel{Price: 1002, Orders: []OrderPart{{Id: "2", Time: 10000}}})
 	l.AddPriceLevel(&PriceLevel{Price: 1003, Orders: []OrderPart{{Id: "3", Time: 10000}}})
 	l.AddPriceLevel(&PriceLevel{Price: 1001, Orders: []OrderPart{{Id: "4", Time: 10000}}})
-	l.UpdateForEach(func(pl *PriceLevel) {
+	l.UpdateForEach(func(pl *PriceLevel, levelIndex int) {
 		if pl.Price <= 1003 {
 			pl.Orders = pl.Orders[:0]
 		}
@@ -467,7 +467,7 @@ func TestULList_UpdateForEach(t *testing.T) {
 	l.AddPriceLevel(&PriceLevel{Price: 1002, Orders: []OrderPart{{Id: "2", Time: 10000}}})
 	l.AddPriceLevel(&PriceLevel{Price: 1003, Orders: []OrderPart{{Id: "3", Time: 10000}}})
 	l.AddPriceLevel(&PriceLevel{Price: 1001, Orders: []OrderPart{{Id: "4", Time: 10000}}})
-	l.UpdateForEach(func(pl *PriceLevel) {
+	l.UpdateForEach(func(pl *PriceLevel, levelIndex int) {
 		if pl.Price <= 1006 {
 			pl.Orders = pl.Orders[:0]
 		}
@@ -536,7 +536,7 @@ func TestULList_Iterate(t *testing.T) {
 				allBuckets: tt.fields.allBuckets,
 			}
 			result := make([]PriceLevel, 0)
-			fillRes := func(p *PriceLevel) {
+			fillRes := func(p *PriceLevel, levelIndex int) {
 				result = append(result, *p)
 			}
 			if ull.Iterate(tt.args.maxLevel, fillRes); !reflect.DeepEqual(result, tt.want) {
@@ -546,3 +546,62 @@ func TestULList_Iterate(t *testing.T) {
 		})
 	}
 }
+
+func TestULList_UpdateForEachBiasedly(t *testing.T) {
+	l := NewULList(5, 2, compareBuy)
+	l.AddPriceLevel(&PriceLevel{Price: 1006, Orders: []OrderPart{{Id: "1", Time: 10000}}})
+	l.AddPriceLevel(&PriceLevel{Price: 1002, Orders: []OrderPart{{Id: "2", Time: 10000}}})
+	l.AddPriceLevel(&PriceLevel{Price: 1003, Orders: []OrderPart{{Id: "3", Time: 10000}}})
+	l.AddPriceLevel(&PriceLevel{Price: 1001, Orders: []OrderPart{{Id: "4", Time: 10000}}})
+	l.UpdateForEach(func(pl *PriceLevel, levelIndex int) {
+		t.Logf("Iterate preferred item: %v", pl)
+		if pl.Price <= 1003 {
+			pl.Orders = pl.Orders[:0]
+		}
+	})
+	require.Len(t, l.begin.elements, 1)
+	require.Equal(t, l.dend, l.begin.next)
+	require.Equal(t, int64(1006), l.begin.elements[0].Price)
+
+	l = NewULList(5, 2, compareBuy)
+	l.AddPriceLevel(&PriceLevel{Price: 1006, Orders: []OrderPart{{Id: "1", Time: 10000}}})
+	l.AddPriceLevel(&PriceLevel{Price: 1002, Orders: []OrderPart{{Id: "2", Time: 10000}}})
+	l.AddPriceLevel(&PriceLevel{Price: 1003, Orders: []OrderPart{{Id: "3", Time: 10000}}})
+	l.AddPriceLevel(&PriceLevel{Price: 1001, Orders: []OrderPart{{Id: "4", Time: 10000}}})
+	l.UpdateForEach(func(pl *PriceLevel, levelIndex int) {
+		if levelIndex < 2 {
+			if pl.Price <= 1002 {
+				pl.Orders = pl.Orders[:0]
+			}
+		} else {
+			if pl.Price <= 1003 {
+				pl.Orders = pl.Orders[:0]
+			}
+		}
+	})
+	require.Len(t, l.begin.elements, 1)
+	require.Equal(t, l.dend, l.begin.next.next)
+	require.Equal(t, int64(1006), l.begin.elements[0].Price)
+	require.Equal(t, int64(1003), l.begin.next.elements[0].Price)
+
+	l = NewULList(5, 2, compareBuy)
+	l.AddPriceLevel(&PriceLevel{Price: 1006, Orders: []OrderPart{{Id: "1", Time: 10000}, {Id: "2", Time: 10000}}})
+	l.AddPriceLevel(&PriceLevel{Price: 1003, Orders: []OrderPart{{Id: "3", Time: 10000}, {Id: "4", Time: 10000}}})
+	l.AddPriceLevel(&PriceLevel{Price: 1002, Orders: []OrderPart{{Id: "5", Time: 10000}}})
+	l.AddPriceLevel(&PriceLevel{Price: 1001, Orders: []OrderPart{{Id: "6", Time: 10000}, {Id: "7", Time: 10000}, {Id: "8", Time: 10000}}})
+	l.UpdateForEach(func(pl *PriceLevel, levelIndex int) {
+		if levelIndex < 2 {
+			if len(pl.Orders) > 1 {
+				pl.Orders = pl.Orders[:1]
+			}
+		} else {
+			if len(pl.Orders) > 2 {
+				pl.Orders = pl.Orders[:2]
+			}
+		}
+	})
+	require.Len(t, l.begin.elements[0].Orders, 1)
+	require.Len(t, l.begin.elements[1].Orders, 1)
+	require.Len(t, l.begin.next.elements[0].Orders, 1)
+	require.Len(t, l.begin.next.elements[1].Orders, 2)
+}

plugins/dex/order/keeper.go

@@ -30,9 +30,10 @@ import (
 )
 
 const (
-	numPricesStored  = 2000
-	pricesStoreEvery = 1000
-	minimalNumPrices = 500
+	numPricesStored      = 2000
+	pricesStoreEvery     = 1000
+	minimalNumPrices     = 500
+	preferencePriceLevel = 500
 )
 
 type FeeHandler func(map[string]*types.Fee)
@@ -457,12 +458,12 @@ func (kp *Keeper) GetOrderBookLevels(pair string, maxLevels int) []store.OrderBo
 
 	if eng, ok := kp.engines[pair]; ok {
 		// TODO: check considered bucket splitting?
-		eng.Book.ShowDepth(maxLevels, func(p *me.PriceLevel) {
+		eng.Book.ShowDepth(maxLevels, func(p *me.PriceLevel, levelIndex int) {
 			orderbook[i].BuyPrice = utils.Fixed8(p.Price)
 			orderbook[i].BuyQty = utils.Fixed8(p.TotalLeavesQty())
 			i++
 		},
-			func(p *me.PriceLevel) {
+			func(p *me.PriceLevel, levelIndex int) {
 				orderbook[j].SellPrice = utils.Fixed8(p.Price)
 				orderbook[j].SellQty = utils.Fixed8(p.TotalLeavesQty())
 				j++
@@ -503,9 +504,9 @@ func (kp *Keeper) GetOrderBooks(maxLevels int) ChangedPriceLevelsMap {
 		res[pair] = ChangedPriceLevelsPerSymbol{buys, sells}
 
 		// TODO: check considered bucket splitting?
-		eng.Book.ShowDepth(maxLevels, func(p *me.PriceLevel) {
+		eng.Book.ShowDepth(maxLevels, func(p *me.PriceLevel, levelIndex int) {
 			buys[p.Price] = p.TotalLeavesQty()
-		}, func(p *me.PriceLevel) {
+		}, func(p *me.PriceLevel, levelIndex int) {
 			sells[p.Price] = p.TotalLeavesQty()
 		})
 	}
@@ -822,12 +823,8 @@ func (kp *Keeper) expireOrders(ctx sdk.Context, blockTime time.Time) []chan Tran
 		return nil
 	}
 
-	// TODO: make effectiveDays configurable
-	const effectiveDays = 3
-	expireHeight, err := kp.GetBreatheBlockHeight(ctx, blockTime, effectiveDays)
+	expireHeight, forceExpireHeight, err := kp.getExpireHeight(ctx, blockTime)
 	if err != nil {
-		// breathe block not found, that should only happens in in the first three days, just log it and ignore.
-		kp.logger.Info(err.Error())
 		return nil
 	}
 
@@ -844,7 +841,7 @@ func (kp *Keeper) expireOrders(ctx sdk.Context, blockTime time.Time) []chan Tran
 	}
 
 	expire := func(orders map[string]*OrderInfo, engine *me.MatchEng, side int8) {
-		engine.Book.RemoveOrders(expireHeight, side, func(ord me.OrderPart) {
+		removeCallback := func(ord me.OrderPart) {
 			// gen transfer
 			if ordMsg, ok := orders[ord.Id]; ok && ordMsg != nil {
 				h := channelHash(ordMsg.Sender, concurrency)
@@ -854,7 +851,12 @@ func (kp *Keeper) expireOrders(ctx sdk.Context, blockTime time.Time) []chan Tran
 			} else {
 				kp.logger.Error("failed to locate order to remove in order book", "oid", ord.Id)
 			}
-		})
+		}
+		if !sdk.IsUpgrade(upgrade.BEP67) {
+			engine.Book.RemoveOrders(expireHeight, side, removeCallback)
+		} else {
+			engine.Book.RemoveOrdersBasedOnPriceLevel(expireHeight, forceExpireHeight, preferencePriceLevel, side, removeCallback)
+		}
 	}
 
 	symbolCh := make(chan string, concurrency)
@@ -880,6 +882,30 @@ func (kp *Keeper) expireOrders(ctx sdk.Context, blockTime time.Time) []chan Tran
 	return transferChs
 }
 
+func (kp *Keeper) getExpireHeight(ctx sdk.Context, blockTime time.Time) (expireHeight, forceExpireHeight int64, noBreatheBlock error) {
+	const effectiveDays = 3
+	expireHeight, noBreatheBlock = kp.GetBreatheBlockHeight(ctx, blockTime, effectiveDays)
+	if noBreatheBlock != nil {
+		// breathe block not found, that should only happens in in the first three days, just log it and ignore.
+		kp.logger.Error(noBreatheBlock.Error())
+		return -1, -1, noBreatheBlock
+	}
+
+	if sdk.IsUpgrade(upgrade.BEP67) {
+		const forceExpireDays = 30
+		var err error
+		forceExpireHeight, err = kp.GetBreatheBlockHeight(ctx, blockTime, forceExpireDays)
+		if err != nil {
+			//if breathe block of 30 days ago not found, the breathe block of 3 days ago still can be processed, so return err=nil
+			kp.logger.Error(err.Error())
+			return expireHeight, -1, nil
+		}
+	} else {
+		forceExpireHeight = -1
+	}
+	return expireHeight, forceExpireHeight, nil
+}
+
 func (kp *Keeper) ExpireOrders(
 	ctx sdk.Context,
 	blockTime time.Time,