Match orders in same block as they are placed (#185)

* Match orders in same block as they are placed

* Add metrics

* Changes for latest avalanche cli

* Review fixes

* Fix compile error

.avalanche-cli.json

@@ -1,5 +1,6 @@
 {
   "node-config": {
     "log-level": "info"
-  }
-}
+  },
+  "SingleNodeEnabled": false
+}

core/types/transaction.go

@@ -639,6 +639,22 @@ func (t *TransactionsByPriceAndNonce) Pop() {
 	heap.Pop(&t.heads)
 }
 
+func (t *TransactionsByPriceAndNonce) Copy() *TransactionsByPriceAndNonce {
+	txs := make(map[common.Address]Transactions, len(t.txs))
+	for acc, accTxs := range t.txs {
+		txs[acc] = make(Transactions, len(accTxs))
+		copy(txs[acc], accTxs)
+	}
+	heads := make(TxByPriceAndTime, len(t.heads))
+	copy(heads, t.heads)
+	return &TransactionsByPriceAndNonce{
+		txs:     txs,
+		heads:   heads,
+		signer:  t.signer,
+		baseFee: big.NewInt(0).Set(t.baseFee),
+	}
+}
+
 // copyAddressPtr copies an address.
 func copyAddressPtr(a *common.Address) *common.Address {
 	if a == nil {

eth/backend.go

@@ -335,6 +335,11 @@ func (s *Ethereum) SetEtherbase(etherbase common.Address) {
 	s.miner.SetEtherbase(etherbase)
 }
 
+func (s *Ethereum) SetOrderbookChecker(orderBookChecker miner.OrderbookChecker) {
+	s.miner.SetOrderbookChecker(orderBookChecker)
+
+}
+
 func (s *Ethereum) Miner() *miner.Miner { return s.miner }
 
 func (s *Ethereum) AccountManager() *accounts.Manager { return s.accountManager }

miner/miner.go

@@ -64,6 +64,10 @@ func (miner *Miner) SetEtherbase(addr common.Address) {
 	miner.worker.setEtherbase(addr)
 }
 
+func (miner *Miner) SetOrderbookChecker(orderBookChecker OrderbookChecker) {
+	miner.worker.setOrderbookChecker(orderBookChecker)
+}
+
 func (miner *Miner) GenerateBlock(predicateContext *precompileconfig.PredicateContext) (*types.Block, error) {
 	return miner.worker.commitNewWork(predicateContext)
 }

miner/orderbook_checker.go

@@ -0,0 +1,14 @@
+package miner
+
+import (
+	"math/big"
+
+	"github.com/ava-labs/subnet-evm/core/state"
+	"github.com/ava-labs/subnet-evm/core/types"
+	"github.com/ethereum/go-ethereum/common"
+)
+
+type OrderbookChecker interface {
+	GetMatchingTxs(tx *types.Transaction, stateDB *state.StateDB, blockNumber *big.Int) map[common.Address]types.Transactions
+	ResetMemoryDB()
+}

miner/worker.go

@@ -99,6 +99,8 @@ type worker struct {
 	mu       sync.RWMutex   // The lock used to protect the coinbase and extra fields
 	coinbase common.Address
 	clock    *mockable.Clock // Allows us mock the clock for testing
+
+	orderbookChecker OrderbookChecker
 }
 
 func newWorker(config *Config, chainConfig *params.ChainConfig, engine consensus.Engine, eth Backend, mux *event.TypeMux, clock *mockable.Clock) *worker {
@@ -123,6 +125,10 @@ func (w *worker) setEtherbase(addr common.Address) {
 	w.coinbase = addr
 }
 
+func (w *worker) setOrderbookChecker(orderBookChecker OrderbookChecker) {
+	w.orderbookChecker = orderBookChecker
+}
+
 // commitNewWork generates several new sealing tasks based on the parent block.
 func (w *worker) commitNewWork(predicateContext *precompileconfig.PredicateContext) (*types.Block, error) {
 	w.mu.RLock()
@@ -229,16 +235,38 @@ func (w *worker) commitNewWork(predicateContext *precompileconfig.PredicateConte
 	}
 	if len(localTxs) > 0 {
 		txs := types.NewTransactionsByPriceAndNonce(env.signer, localTxs, header.BaseFee)
+		txsCopy := txs.Copy()
 		w.commitTransactions(env, txs, header.Coinbase)
+		w.commitOrderbookTxs(env, txsCopy, header)
 	}
 	if len(remoteTxs) > 0 {
 		txs := types.NewTransactionsByPriceAndNonce(env.signer, remoteTxs, header.BaseFee)
+		txsCopy := txs.Copy()
 		w.commitTransactions(env, txs, header.Coinbase)
+		w.commitOrderbookTxs(env, txsCopy, header)
 	}
 
+	w.orderbookChecker.ResetMemoryDB()
+
 	return w.commit(env)
 }
 
+func (w *worker) commitOrderbookTxs(env *environment, transactions *types.TransactionsByPriceAndNonce, header *types.Header) {
+	for {
+		tx := transactions.Peek()
+		if tx == nil {
+			break
+		}
+		transactions.Pop()
+
+		orderbookTxs := w.orderbookChecker.GetMatchingTxs(tx, env.state, header.Number)
+		if orderbookTxs != nil {
+			txsByPrice := types.NewTransactionsByPriceAndNonce(env.signer, orderbookTxs, header.BaseFee)
+			w.commitTransactions(env, txsByPrice, header.Coinbase)
+		}
+	}
+}
+
 func (w *worker) createCurrentEnvironment(predicateContext *precompileconfig.PredicateContext, parent *types.Header, header *types.Header, tstart time.Time) (*environment, error) {
 	state, err := w.chain.StateAt(parent.Root)
 	if err != nil {

plugin/evm/limit_order.go

@@ -38,6 +38,8 @@ type LimitOrderProcesser interface {
 	GetTestingAPI() *orderbook.TestingAPI
 	GetTradingAPI() *orderbook.TradingAPI
 	RunMatchingPipeline()
+	GetMemoryDB() orderbook.LimitOrderDatabase
+	GetLimitOrderTxProcessor() orderbook.LimitOrderTxProcessor
 }
 
 type limitOrderProcesser struct {
@@ -206,6 +208,14 @@ func (lop *limitOrderProcesser) GetTestingAPI() *orderbook.TestingAPI {
 	return orderbook.NewTestingAPI(lop.memoryDb, lop.backend, lop.configService, lop.hubbleDB)
 }
 
+func (lop *limitOrderProcesser) GetMemoryDB() orderbook.LimitOrderDatabase {
+	return lop.memoryDb
+}
+
+func (lop *limitOrderProcesser) GetLimitOrderTxProcessor() orderbook.LimitOrderTxProcessor {
+	return lop.limitOrderTxProcessor
+}
+
 func (lop *limitOrderProcesser) listenAndStoreLimitOrderTransactions() {
 	logsCh := make(chan []*types.Log)
 	logsSubscription := lop.backend.SubscribeHubbleLogsEvent(logsCh)

plugin/evm/orderbook/config_service.go

@@ -39,6 +39,7 @@ type IConfigService interface {
 
 type ConfigService struct {
 	blockChain *core.BlockChain
+	stateDB    *state.StateDB
 }
 
 func NewConfigService(blockChain *core.BlockChain) IConfigService {
@@ -47,6 +48,20 @@ func NewConfigService(blockChain *core.BlockChain) IConfigService {
 	}
 }
 
+func NewConfigServiceFromStateDB(stateDB *state.StateDB) IConfigService {
+	return &ConfigService{
+		stateDB: stateDB,
+	}
+}
+
+func (cs *ConfigService) getStateAtCurrentBlock() *state.StateDB {
+	if cs.stateDB != nil {
+		return cs.stateDB
+	}
+	stateDB, _ := cs.blockChain.StateAt(cs.blockChain.CurrentBlock().Root)
+	return stateDB
+}
+
 func (cs *ConfigService) GetAcceptableBounds(market Market) (*big.Int, *big.Int) {
 	return bibliophile.GetAcceptableBounds(cs.getStateAtCurrentBlock(), int64(market))
 }
@@ -79,11 +94,6 @@ func (cs *ConfigService) GetPriceMultiplier(market Market) *big.Int {
 	return bibliophile.GetMultiplier(cs.getStateAtCurrentBlock(), int64(market))
 }
 
-func (cs *ConfigService) getStateAtCurrentBlock() *state.StateDB {
-	stateDB, _ := cs.blockChain.StateAt(cs.blockChain.CurrentBlock().Root)
-	return stateDB
-}
-
 func (cs *ConfigService) GetActiveMarketsCount() int64 {
 	return bibliophile.GetActiveMarketsCount(cs.getStateAtCurrentBlock())
 }

plugin/evm/orderbook/matching_pipeline.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/ava-labs/subnet-evm/core/types"
 	hu "github.com/ava-labs/subnet-evm/plugin/evm/orderbook/hubbleutils"
 	"github.com/ava-labs/subnet-evm/utils"
 	"github.com/ethereum/go-ethereum/common"
@@ -42,6 +43,18 @@ func NewMatchingPipeline(
 	}
 }
 
+func NewTemporaryMatchingPipeline(
+	db LimitOrderDatabase,
+	lotp LimitOrderTxProcessor,
+	configService IConfigService) *MatchingPipeline {
+
+	return &MatchingPipeline{
+		db:            db,
+		lotp:          lotp,
+		configService: configService,
+	}
+}
+
 func (pipeline *MatchingPipeline) RunSanitization() {
 	pipeline.db.RemoveExpiredSignedOrders()
 }
@@ -106,6 +119,47 @@ func (pipeline *MatchingPipeline) Run(blockNumber *big.Int) bool {
 	return false
 }
 
+func (pipeline *MatchingPipeline) GetOrderMatchingTransactions(blockNumber *big.Int, markets []Market) map[common.Address]types.Transactions {
+	pipeline.mu.Lock()
+	defer pipeline.mu.Unlock()
+
+	activeMarkets := pipeline.GetActiveMarkets()
+	log.Info("MatchingPipeline:GetOrderMatchingTransactions")
+
+	if len(activeMarkets) == 0 {
+		return nil
+	}
+
+	// start fresh and purge all local transactions
+	pipeline.lotp.PurgeOrderBookTxs()
+
+	// fetch various hubble market params and run the matching engine
+	hState := hu.GetHubbleState()
+	hState.OraclePrices = hu.ArrayToMap(pipeline.configService.GetUnderlyingPrices())
+
+	marginMap := make(map[common.Address]*big.Int)
+	for addr, trader := range pipeline.db.GetAllTraders() {
+		userState := &hu.UserState{
+			Positions:      translatePositions(trader.Positions),
+			Margins:        getMargins(&trader, len(hState.Assets)),
+			PendingFunding: getTotalFunding(&trader, hState.ActiveMarkets),
+			ReservedMargin: new(big.Int).Set(trader.Margin.Reserved),
+			// this is the only leveldb read, others above are in-memory reads
+			ReduceOnlyAmounts: pipeline.configService.GetReduceOnlyAmounts(addr),
+		}
+		marginMap[addr] = hu.GetAvailableMargin(hState, userState)
+	}
+	for _, market := range markets {
+		orders := pipeline.fetchOrders(market, hState.OraclePrices[market], map[common.Hash]struct{}{}, blockNumber)
+		upperBound, _ := pipeline.configService.GetAcceptableBounds(market)
+		pipeline.runMatchingEngine(pipeline.lotp, orders.longOrders, orders.shortOrders, marginMap, hState.MinAllowableMargin, hState.TakerFee, upperBound)
+	}
+
+	orderbookTxs := pipeline.lotp.GetOrderBookTxs()
+	pipeline.lotp.PurgeOrderBookTxs()
+	return orderbookTxs
+}
+
 type Orders struct {
 	longOrders  []Order
 	shortOrders []Order

plugin/evm/orderbook/mocks.go

@@ -200,6 +200,10 @@ func (lotp *MockLimitOrderTxProcessor) GetOrderBookTxsCount() uint64 {
 	return uint64(args.Int(0))
 }
 
+func (lotp *MockLimitOrderTxProcessor) GetOrderBookTxs() map[common.Address]types.Transactions {
+	return nil
+}
+
 func (lotp *MockLimitOrderTxProcessor) ExecuteFundingPaymentTx() error {
 	return nil
 }