forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-17 02:10:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

all: remove mongodb support in XDCx (#1679)

Browse source 
* all: remove SDK node

* cmd: remove XDCXDBEngineFlag

* cmd: remove XDCXDBConnectionUrlFlag

* cmd, XDCx: remove XDCXDBReplicaSetNameFlag

* XDCx: remove ConnectionUrl

* all: remove mongodb support

* cmd: remove XDCXEnabledFlag
This commit is contained in:
Daniel Liu 2025-11-15 19:20:02 +08:00 committed by GitHub
parent 0c96e76992
commit 71e8e27f84
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 15 additions and 2097 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

328
XDCx/XDCx.go
View file

@ -5,7 +5,6 @@ import (
"fmt" "fmt"
"math/big" "math/big"
"strconv" "strconv"
"time"
"github.com/XinFinOrg/XDPoSChain/XDCx/tradingstate" "github.com/XinFinOrg/XDPoSChain/XDCx/tradingstate"
"github.com/XinFinOrg/XDPoSChain/XDCxDAO" "github.com/XinFinOrg/XDPoSChain/XDCxDAO"
@ -30,11 +29,8 @@ var (
) )
type Config struct { type Config struct {
DataDir string `toml:",omitempty"` DataDir string `toml:",omitempty"`
DBEngine string `toml:",omitempty"` DBName string `toml:",omitempty"`
DBName string `toml:",omitempty"`
ConnectionUrl string `toml:",omitempty"`
ReplicaSetName string `toml:",omitempty"`
} }
// DefaultConfig represents (shocker!) the default configuration. // DefaultConfig represents (shocker!) the default configuration.
@ -45,7 +41,6 @@ var DefaultConfig = Config{
type XDCX struct { type XDCX struct {
// Order related // Order related
db XDCxDAO.XDCXDAO db XDCxDAO.XDCXDAO
mongodb XDCxDAO.XDCXDAO
Triegc *prque.Prque[int64, common.Hash] // Priority queue mapping block numbers to tries to gc Triegc *prque.Prque[int64, common.Hash] // Priority queue mapping block numbers to tries to gc
StateCache tradingstate.Database // State database to reuse between imports (contains state cache) *XDCx_state.TradingStateDB StateCache tradingstate.Database // State database to reuse between imports (contains state cache) *XDCx_state.TradingStateDB
@ -60,16 +55,6 @@ func NewLDBEngine(cfg *Config) *XDCxDAO.BatchDatabase {
return batchDB return batchDB
} }
func NewMongoDBEngine(cfg *Config) *XDCxDAO.MongoDatabase {
mongoDB, err := XDCxDAO.NewMongoDatabase(nil, cfg.DBName, cfg.ConnectionUrl, cfg.ReplicaSetName, 0)
if err != nil {
log.Crit("Failed to init mongodb engine", "err", err)
}
return mongoDB
}
func New(stack *node.Node, cfg *Config) *XDCX { func New(stack *node.Node, cfg *Config) *XDCX {
XDCX := &XDCX{ XDCX := &XDCX{
Triegc: prque.New[int64, common.Hash](nil), Triegc: prque.New[int64, common.Hash](nil),
@ -79,30 +64,16 @@ func New(stack *node.Node, cfg *Config) *XDCX {
// default DBEngine: levelDB // default DBEngine: levelDB
XDCX.db = NewLDBEngine(cfg) XDCX.db = NewLDBEngine(cfg)
XDCX.sdkNode = false
if cfg.DBEngine == "mongodb" { // this is an add-on DBEngine for SDK nodes
XDCX.mongodb = NewMongoDBEngine(cfg)
XDCX.sdkNode = true
}
XDCX.StateCache = tradingstate.NewDatabase(XDCX.db) XDCX.StateCache = tradingstate.NewDatabase(XDCX.db)
return XDCX return XDCX
} }
func (XDCx *XDCX) IsSDKNode() bool {
return XDCx.sdkNode
}
func (XDCx *XDCX) GetLevelDB() XDCxDAO.XDCXDAO { func (XDCx *XDCX) GetLevelDB() XDCxDAO.XDCXDAO {
return XDCx.db return XDCx.db
} }
func (XDCx *XDCX) GetMongoDB() XDCxDAO.XDCXDAO {
return XDCx.mongodb
}
func (XDCx *XDCX) ProcessOrderPending(header *types.Header, coinbase common.Address, chain consensus.ChainContext, pending map[common.Address]types.OrderTransactions, statedb *state.StateDB, XDCXstatedb *tradingstate.TradingStateDB) ([]tradingstate.TxDataMatch, map[common.Hash]tradingstate.MatchingResult) { func (XDCx *XDCX) ProcessOrderPending(header *types.Header, coinbase common.Address, chain consensus.ChainContext, pending map[common.Address]types.OrderTransactions, statedb *state.StateDB, XDCXstatedb *tradingstate.TradingStateDB) ([]tradingstate.TxDataMatch, map[common.Hash]tradingstate.MatchingResult) {
txMatches := []tradingstate.TxDataMatch{} txMatches := []tradingstate.TxDataMatch{}
matchingResults := map[common.Hash]tradingstate.MatchingResult{} matchingResults := map[common.Hash]tradingstate.MatchingResult{}
@ -250,259 +221,6 @@ func (XDCx *XDCX) ConvertXDCToToken(chain consensus.ChainContext, statedb *state
return tokenQuantity, tokenPriceInXDC, nil return tokenQuantity, tokenPriceInXDC, nil
} }
// there are 3 tasks need to complete to update data in SDK nodes after matching
// 1. txMatchData.Order: order has been processed. This order should be put to `orders` collection with status sdktypes.OrderStatusOpen
// 2. txMatchData.Trades: includes information of matched orders.
// a. PutObject them to `trades` collection
// b. Update status of regrading orders to sdktypes.OrderStatusFilled
func (XDCx *XDCX) SyncDataToSDKNode(takerOrderInTx *tradingstate.OrderItem, txHash common.Hash, txMatchTime time.Time, statedb *state.StateDB, trades []map[string]string, rejectedOrders []*tradingstate.OrderItem, dirtyOrderCount *uint64) error {
var (
// originTakerOrder: order get from db, nil if it doesn't exist
// takerOrderInTx: order decoded from txdata
// updatedTakerOrder: order with new status, filledAmount, CreatedAt, UpdatedAt. This will be inserted to db
originTakerOrder, updatedTakerOrder *tradingstate.OrderItem
makerDirtyHashes []string
makerDirtyFilledAmount map[string]*big.Int
err error
)
db := XDCx.GetMongoDB()
db.InitBulk()
if takerOrderInTx.Status == tradingstate.OrderStatusCancelled && len(rejectedOrders) > 0 {
// cancel order is rejected -> nothing change
log.Debug("Cancel order is rejected", "order", tradingstate.ToJSON(takerOrderInTx))
return nil
}
// 1. put processed takerOrderInTx to db
lastState := tradingstate.OrderHistoryItem{}
val, err := db.GetObject(takerOrderInTx.Hash, &tradingstate.OrderItem{})
if err == nil && val != nil {
originTakerOrder = val.(*tradingstate.OrderItem)
lastState = tradingstate.OrderHistoryItem{
TxHash: originTakerOrder.TxHash,
FilledAmount: tradingstate.CloneBigInt(originTakerOrder.FilledAmount),
Status: originTakerOrder.Status,
UpdatedAt: originTakerOrder.UpdatedAt,
}
}
if originTakerOrder != nil {
updatedTakerOrder = originTakerOrder
} else {
updatedTakerOrder = takerOrderInTx
updatedTakerOrder.FilledAmount = new(big.Int)
}
if takerOrderInTx.Status != tradingstate.OrderStatusCancelled {
updatedTakerOrder.Status = tradingstate.OrderStatusOpen
} else {
updatedTakerOrder.Status = tradingstate.OrderStatusCancelled
updatedTakerOrder.ExtraData = takerOrderInTx.ExtraData
}
updatedTakerOrder.TxHash = txHash
if updatedTakerOrder.CreatedAt.IsZero() {
updatedTakerOrder.CreatedAt = txMatchTime
}
if txMatchTime.Before(updatedTakerOrder.UpdatedAt) || (txMatchTime.Equal(updatedTakerOrder.UpdatedAt) && *dirtyOrderCount == 0) {
log.Debug("Ignore old orders/trades taker", "txHash", txHash.Hex(), "txTime", txMatchTime.UnixNano(), "updatedAt", updatedTakerOrder.UpdatedAt.UnixNano())
return nil
}
*dirtyOrderCount++
XDCx.UpdateOrderCache(updatedTakerOrder.BaseToken, updatedTakerOrder.QuoteToken, updatedTakerOrder.Hash, txHash, lastState)
updatedTakerOrder.UpdatedAt = txMatchTime
// 2. put trades to db and update status to FILLED
log.Debug("Got trades", "number", len(trades), "txhash", txHash.Hex())
makerDirtyFilledAmount = make(map[string]*big.Int)
for _, trade := range trades {
// 2.a. put to trades
if trade == nil {
continue
}
tradeRecord := &tradingstate.Trade{}
quantity := tradingstate.ToBigInt(trade[tradingstate.TradeQuantity])
price := tradingstate.ToBigInt(trade[tradingstate.TradePrice])
if price.Sign() <= 0 || quantity.Sign() <= 0 {
return fmt.Errorf("trade misses important information. tradedPrice %v, tradedQuantity %v", price, quantity)
}
tradeRecord.Amount = quantity
tradeRecord.PricePoint = price
tradeRecord.BaseToken = updatedTakerOrder.BaseToken
tradeRecord.QuoteToken = updatedTakerOrder.QuoteToken
tradeRecord.Status = tradingstate.TradeStatusSuccess
tradeRecord.Taker = updatedTakerOrder.UserAddress
tradeRecord.Maker = common.HexToAddress(trade[tradingstate.TradeMaker])
tradeRecord.TakerOrderHash = updatedTakerOrder.Hash
tradeRecord.MakerOrderHash = common.HexToHash(trade[tradingstate.TradeMakerOrderHash])
tradeRecord.TxHash = txHash
tradeRecord.TakerOrderSide = updatedTakerOrder.Side
tradeRecord.TakerExchange = updatedTakerOrder.ExchangeAddress
tradeRecord.MakerExchange = common.HexToAddress(trade[tradingstate.TradeMakerExchange])
tradeRecord.MakeFee, _ = new(big.Int).SetString(trade[tradingstate.MakerFee], 10)
tradeRecord.TakeFee, _ = new(big.Int).SetString(trade[tradingstate.TakerFee], 10)
// set makerOrderType, takerOrderType
tradeRecord.MakerOrderType = trade[tradingstate.MakerOrderType]
tradeRecord.TakerOrderType = updatedTakerOrder.Type
if tradeRecord.CreatedAt.IsZero() {
tradeRecord.CreatedAt = txMatchTime
}
tradeRecord.UpdatedAt = txMatchTime
tradeRecord.Hash = tradeRecord.ComputeHash()
log.Debug("TRADE history", "amount", tradeRecord.Amount, "pricepoint", tradeRecord.PricePoint,
"taker", tradeRecord.Taker.Hex(), "maker", tradeRecord.Maker.Hex(), "takerOrder", tradeRecord.TakerOrderHash.Hex(), "makerOrder", tradeRecord.MakerOrderHash.Hex(),
"takerFee", tradeRecord.TakeFee, "makerFee", tradeRecord.MakeFee)
if err := db.PutObject(tradeRecord.Hash, tradeRecord); err != nil {
return fmt.Errorf("SDKNode: failed to store tradeRecord %s", err.Error())
}
// 2.b. update status and filledAmount
filledAmount := quantity
// maker dirty order
makerFilledAmount := big.NewInt(0)
if amount, ok := makerDirtyFilledAmount[trade[tradingstate.TradeMakerOrderHash]]; ok {
makerFilledAmount = tradingstate.CloneBigInt(amount)
}
makerFilledAmount = new(big.Int).Add(makerFilledAmount, filledAmount)
makerDirtyFilledAmount[trade[tradingstate.TradeMakerOrderHash]] = makerFilledAmount
makerDirtyHashes = append(makerDirtyHashes, trade[tradingstate.TradeMakerOrderHash])
//updatedTakerOrder = XDCx.updateMatchedOrder(updatedTakerOrder, filledAmount, txMatchTime, txHash)
// update filledAmount, status of takerOrder
updatedTakerOrder.FilledAmount = new(big.Int).Add(updatedTakerOrder.FilledAmount, filledAmount)
if updatedTakerOrder.FilledAmount.Cmp(updatedTakerOrder.Quantity) < 0 && updatedTakerOrder.Type == tradingstate.Limit {
updatedTakerOrder.Status = tradingstate.OrderStatusPartialFilled
} else {
updatedTakerOrder.Status = tradingstate.OrderStatusFilled
}
}
// for Market orders
// filledAmount > 0 : FILLED
// otherwise: REJECTED
if updatedTakerOrder.Type == tradingstate.Market {
if updatedTakerOrder.FilledAmount.Sign() > 0 {
updatedTakerOrder.Status = tradingstate.OrderStatusFilled
} else {
updatedTakerOrder.Status = tradingstate.OrderStatusRejected
}
}
log.Debug("PutObject processed takerOrder",
"userAddr", updatedTakerOrder.UserAddress.Hex(), "side", updatedTakerOrder.Side,
"price", updatedTakerOrder.Price, "quantity", updatedTakerOrder.Quantity, "filledAmount", updatedTakerOrder.FilledAmount, "status", updatedTakerOrder.Status,
"hash", updatedTakerOrder.Hash.Hex(), "txHash", updatedTakerOrder.TxHash.Hex())
if err := db.PutObject(updatedTakerOrder.Hash, updatedTakerOrder); err != nil {
return fmt.Errorf("SDKNode: failed to put processed takerOrder. Hash: %s Error: %s", updatedTakerOrder.Hash.Hex(), err.Error())
}
items := db.GetListItemByHashes(makerDirtyHashes, &tradingstate.OrderItem{})
if items != nil {
makerOrders := items.([]*tradingstate.OrderItem)
log.Debug("Maker dirty orders", "len", len(makerOrders), "txhash", txHash.Hex())
for _, o := range makerOrders {
if txMatchTime.Before(o.UpdatedAt) {
log.Debug("Ignore old orders/trades maker", "txHash", txHash.Hex(), "txTime", txMatchTime.UnixNano(), "updatedAt", updatedTakerOrder.UpdatedAt.UnixNano())
continue
}
lastState = tradingstate.OrderHistoryItem{
TxHash: o.TxHash,
FilledAmount: tradingstate.CloneBigInt(o.FilledAmount),
Status: o.Status,
UpdatedAt: o.UpdatedAt,
}
XDCx.UpdateOrderCache(o.BaseToken, o.QuoteToken, o.Hash, txHash, lastState)
o.TxHash = txHash
o.UpdatedAt = txMatchTime
o.FilledAmount = new(big.Int).Add(o.FilledAmount, makerDirtyFilledAmount[o.Hash.Hex()])
if o.FilledAmount.Cmp(o.Quantity) < 0 {
o.Status = tradingstate.OrderStatusPartialFilled
} else {
o.Status = tradingstate.OrderStatusFilled
}
log.Debug("PutObject processed makerOrder",
"userAddr", o.UserAddress.Hex(), "side", o.Side,
"price", o.Price, "quantity", o.Quantity, "filledAmount", o.FilledAmount, "status", o.Status,
"hash", o.Hash.Hex(), "txHash", o.TxHash.Hex())
if err := db.PutObject(o.Hash, o); err != nil {
return fmt.Errorf("SDKNode: failed to put processed makerOrder. Hash: %s Error: %s", o.Hash.Hex(), err.Error())
}
}
}
// 3. put rejected orders to db and update status REJECTED
log.Debug("Got rejected orders", "number", len(rejectedOrders), "rejectedOrders", rejectedOrders)
if len(rejectedOrders) > 0 {
var rejectedHashes []string
// updateRejectedOrders
for _, rejectedOrder := range rejectedOrders {
rejectedHashes = append(rejectedHashes, rejectedOrder.Hash.Hex())
if updatedTakerOrder.Hash == rejectedOrder.Hash && !txMatchTime.Before(updatedTakerOrder.UpdatedAt) {
// cache order history for handling reorg
orderHistoryRecord := tradingstate.OrderHistoryItem{
TxHash: updatedTakerOrder.TxHash,
FilledAmount: tradingstate.CloneBigInt(updatedTakerOrder.FilledAmount),
Status: updatedTakerOrder.Status,
UpdatedAt: updatedTakerOrder.UpdatedAt,
}
XDCx.UpdateOrderCache(updatedTakerOrder.BaseToken, updatedTakerOrder.QuoteToken, updatedTakerOrder.Hash, txHash, orderHistoryRecord)
// if whole order is rejected, status = REJECTED
// otherwise, status = FILLED
if updatedTakerOrder.FilledAmount.Sign() > 0 {
updatedTakerOrder.Status = tradingstate.OrderStatusFilled
} else {
updatedTakerOrder.Status = tradingstate.OrderStatusRejected
}
updatedTakerOrder.TxHash = txHash
updatedTakerOrder.UpdatedAt = txMatchTime
if err := db.PutObject(updatedTakerOrder.Hash, updatedTakerOrder); err != nil {
return fmt.Errorf("SDKNode: failed to reject takerOrder. Hash: %s Error: %s", updatedTakerOrder.Hash.Hex(), err.Error())
}
}
}
items := db.GetListItemByHashes(rejectedHashes, &tradingstate.OrderItem{})
if items != nil {
dirtyRejectedOrders := items.([]*tradingstate.OrderItem)
for _, order := range dirtyRejectedOrders {
if txMatchTime.Before(order.UpdatedAt) {
log.Debug("Ignore old orders/trades reject", "txHash", txHash.Hex(), "txTime", txMatchTime.UnixNano(), "updatedAt", updatedTakerOrder.UpdatedAt.UnixNano())
continue
}
// cache order history for handling reorg
orderHistoryRecord := tradingstate.OrderHistoryItem{
TxHash: order.TxHash,
FilledAmount: tradingstate.CloneBigInt(order.FilledAmount),
Status: order.Status,
UpdatedAt: order.UpdatedAt,
}
XDCx.UpdateOrderCache(order.BaseToken, order.QuoteToken, order.Hash, txHash, orderHistoryRecord)
dirtyFilledAmount, ok := makerDirtyFilledAmount[order.Hash.Hex()]
if ok && dirtyFilledAmount != nil {
order.FilledAmount = new(big.Int).Add(order.FilledAmount, dirtyFilledAmount)
}
// if whole order is rejected, status = REJECTED
// otherwise, status = FILLED
if order.FilledAmount.Sign() > 0 {
order.Status = tradingstate.OrderStatusFilled
} else {
order.Status = tradingstate.OrderStatusRejected
}
order.TxHash = txHash
order.UpdatedAt = txMatchTime
if err = db.PutObject(order.Hash, order); err != nil {
return fmt.Errorf("SDKNode: failed to update rejectedOder to sdkNode %s", err.Error())
}
}
}
}
if err := db.CommitBulk(); err != nil {
return fmt.Errorf("SDKNode fail to commit bulk update orders, trades at txhash %s . Error: %s", txHash.Hex(), err.Error())
}
return nil
}
func (XDCx *XDCX) GetTradingState(block *types.Block, author common.Address) (*tradingstate.TradingStateDB, error) { func (XDCx *XDCX) GetTradingState(block *types.Block, author common.Address) (*tradingstate.TradingStateDB, error) {
root, err := XDCx.GetTradingStateRoot(block, author) root, err := XDCx.GetTradingStateRoot(block, author)
if err != nil { if err != nil {
@ -559,45 +277,3 @@ func (XDCx *XDCX) UpdateOrderCache(baseToken, quoteToken common.Address, orderHa
} }
XDCx.orderCache.Add(txhash, orderCacheAtTxHash) XDCx.orderCache.Add(txhash, orderCacheAtTxHash)
} }
func (XDCx *XDCX) RollbackReorgTxMatch(txhash common.Hash) error {
db := XDCx.GetMongoDB()
db.InitBulk()
items := db.GetListItemByTxHash(txhash, &tradingstate.OrderItem{})
if items != nil {
for _, order := range items.([]*tradingstate.OrderItem) {
orderCacheAtTxHash, ok := XDCx.orderCache.Get(txhash)
log.Debug("XDCx reorg: rollback order", "txhash", txhash.Hex(), "order", tradingstate.ToJSON(order), "orderHistoryItem", orderCacheAtTxHash)
if !ok || orderCacheAtTxHash == nil {
log.Debug("XDCx reorg: remove order due to no orderCache", "order", tradingstate.ToJSON(order))
if err := db.DeleteObject(order.Hash, &tradingstate.OrderItem{}); err != nil {
log.Crit("SDKNode: failed to remove reorg order", "err", err.Error(), "order", tradingstate.ToJSON(order))
}
continue
}
orderHistoryItem := orderCacheAtTxHash[tradingstate.GetOrderHistoryKey(order.BaseToken, order.QuoteToken, order.Hash)]
if (orderHistoryItem == tradingstate.OrderHistoryItem{}) {
log.Debug("XDCx reorg: remove order due to empty orderHistory", "order", tradingstate.ToJSON(order))
if err := db.DeleteObject(order.Hash, &tradingstate.OrderItem{}); err != nil {
log.Crit("SDKNode: failed to remove reorg order", "err", err.Error(), "order", tradingstate.ToJSON(order))
}
continue
}
order.TxHash = orderHistoryItem.TxHash
order.Status = orderHistoryItem.Status
order.FilledAmount = tradingstate.CloneBigInt(orderHistoryItem.FilledAmount)
order.UpdatedAt = orderHistoryItem.UpdatedAt
log.Debug("XDCx reorg: update order to the last orderHistoryItem", "order", tradingstate.ToJSON(order), "orderHistoryItem", orderHistoryItem)
if err := db.PutObject(order.Hash, order); err != nil {
log.Crit("SDKNode: failed to update reorg order", "err", err.Error(), "order", tradingstate.ToJSON(order))
}
}
}
log.Debug("XDCx reorg: DeleteTradeByTxHash", "txhash", txhash.Hex())
db.DeleteItemByTxHash(txhash, &tradingstate.Trade{})
if err := db.CommitBulk(); err != nil {
return fmt.Errorf("failed to RollbackTradingData. %v", err)
}
return nil
}

32
XDCx/order_processor.go
View file

@ -784,37 +784,5 @@ func (XDCx *XDCX) UpdateMediumPriceBeforeEpoch(epochNumber uint64, tradingStateD
} }
tradingStateDB.SetMediumPrice(orderbook, tradingstate.Zero, tradingstate.Zero) tradingStateDB.SetMediumPrice(orderbook, tradingstate.Zero, tradingstate.Zero)
} }
if XDCx.IsSDKNode() {
if err := XDCx.LogEpochPrice(epochNumber, epochPriceResult); err != nil {
log.Error("failed to update epochPrice", "err", err)
}
}
return nil
}
// put average price of epoch to mongodb for tracking liquidation trades
// epochPriceResult: a map of epoch average price, key is orderbook hash , value is epoch average price
// orderbook hash genereted from baseToken, quoteToken at XDPoSChain/XDCx/tradingstate/common.go:214
func (XDCx *XDCX) LogEpochPrice(epochNumber uint64, epochPriceResult map[common.Hash]*big.Int) error {
db := XDCx.GetMongoDB()
db.InitBulk()
for orderbook, price := range epochPriceResult {
if price.Sign() <= 0 {
continue
}
epochPriceItem := &tradingstate.EpochPriceItem{
Epoch: epochNumber,
Orderbook: orderbook,
Price: price,
}
epochPriceItem.Hash = epochPriceItem.ComputeHash()
if err := db.PutObject(epochPriceItem.Hash, epochPriceItem); err != nil {
return err
}
}
if err := db.CommitBulk(); err != nil {
return err
}
return nil return nil
} }

9
XDCxDAO/interfaces.go
View file

@ -2,6 +2,7 @@
// This file is part of the Core XDPoSChain infrastructure // This file is part of the Core XDPoSChain infrastructure
// https://XDPoSChain.com // https://XDPoSChain.com
// Package XDCxDAO provides an interface to work with XDCx database, including leveldb for masternode and mongodb for SDK node // Package XDCxDAO provides an interface to work with XDCx database, including leveldb for masternode and mongodb for SDK node
package XDCxDAO package XDCxDAO
import ( import (
@ -15,15 +16,7 @@ type XDCXDAO interface {
// for both leveldb and mongodb // for both leveldb and mongodb
IsEmptyKey(key []byte) bool IsEmptyKey(key []byte) bool
Close() error Close() error
// mongodb methods
HasObject(hash common.Hash, val interface{}) (bool, error)
GetObject(hash common.Hash, val interface{}) (interface{}, error) GetObject(hash common.Hash, val interface{}) (interface{}, error)
PutObject(hash common.Hash, val interface{}) error
DeleteObject(hash common.Hash, val interface{}) error // won't return error if key not found
GetListItemByTxHash(txhash common.Hash, val interface{}) interface{}
GetListItemByHashes(hashes []string, val interface{}) interface{}
DeleteItemByTxHash(txhash common.Hash, val interface{})
// basic XDCx // basic XDCx
InitBulk() InitBulk()

16
XDCxDAO/leveldb.go
View file

@ -52,26 +52,10 @@ func (db *BatchDatabase) IsEmptyKey(key []byte) bool {
return len(key) == 0 || bytes.Equal(key, db.emptyKey) return len(key) == 0 || bytes.Equal(key, db.emptyKey)
} }
func (db *BatchDatabase) HasObject(hash common.Hash, val interface{}) (bool, error) {
// for mongodb only
return false, nil
}
func (db *BatchDatabase) GetObject(hash common.Hash, val interface{}) (interface{}, error) { func (db *BatchDatabase) GetObject(hash common.Hash, val interface{}) (interface{}, error) {
// for mongodb only
return nil, nil return nil, nil
} }
func (db *BatchDatabase) PutObject(hash common.Hash, val interface{}) error {
// for mongodb only
return nil
}
func (db *BatchDatabase) DeleteObject(hash common.Hash, val interface{}) error {
// for mongodb only
return nil
}
func (db *BatchDatabase) Put(key []byte, val []byte) error { func (db *BatchDatabase) Put(key []byte, val []byte) error {
return db.db.Put(key, val) return db.db.Put(key, val)
} }

920
XDCxDAO/mongodb.go
View file

@ -1,920 +0,0 @@
package XDCxDAO
import (
"bytes"
"encoding/hex"
"fmt"
"strings"
"time"
"github.com/XinFinOrg/XDPoSChain/XDCx/tradingstate"
"github.com/XinFinOrg/XDPoSChain/XDCxlending/lendingstate"
"github.com/XinFinOrg/XDPoSChain/common"
"github.com/XinFinOrg/XDPoSChain/common/lru"
"github.com/XinFinOrg/XDPoSChain/ethdb"
"github.com/XinFinOrg/XDPoSChain/log"
"github.com/globalsign/mgo"
"github.com/globalsign/mgo/bson"
)
const (
ordersCollection = "orders"
tradesCollection = "trades"
lendingItemsCollection = "lending_items"
lendingTradesCollection = "lending_trades"
lendingTopUpCollection = "lending_topups"
lendingRepayCollection = "lending_repays"
lendingRecallCollection = "lending_recalls"
epochPriceCollection = "epoch_prices"
)
type MongoDatabase struct {
Session *mgo.Session
dbName string
emptyKey []byte
cacheItems *lru.Cache[string, interface{}] // Cache for reading
orderBulk *mgo.Bulk
tradeBulk *mgo.Bulk
epochPriceBulk *mgo.Bulk
lendingItemBulk *mgo.Bulk
topUpBulk *mgo.Bulk
recallBulk *mgo.Bulk
repayBulk *mgo.Bulk
lendingTradeBulk *mgo.Bulk
}
// InitSession initializes a new session with mongodb
func NewMongoDatabase(session *mgo.Session, dbName string, mongoURL string, replicaSetName string, cacheLimit int) (*MongoDatabase, error) {
if session == nil {
// in case of multiple database instances
hosts := strings.Split(mongoURL, ",")
dbInfo := &mgo.DialInfo{
Addrs: hosts,
Database: dbName,
ReplicaSetName: replicaSetName,
Timeout: 30 * time.Second,
}
ns, err := mgo.DialWithInfo(dbInfo)
if err != nil {
return nil, err
}
session = ns
}
itemCacheLimit := defaultCacheLimit
if cacheLimit > 0 {
itemCacheLimit = cacheLimit
}
db := &MongoDatabase{
Session: session,
dbName: dbName,
cacheItems: lru.NewCache[string, interface{}](itemCacheLimit),
}
if err := db.EnsureIndexes(); err != nil {
return nil, err
}
return db, nil
}
func (db *MongoDatabase) IsEmptyKey(key []byte) bool {
return len(key) == 0 || bytes.Equal(key, db.emptyKey)
}
func (db *MongoDatabase) getCacheKey(key []byte) string {
return hex.EncodeToString(key)
}
func (db *MongoDatabase) HasObject(hash common.Hash, val interface{}) (bool, error) {
if db.IsEmptyKey(hash.Bytes()) {
return false, nil
}
cacheKey := db.getCacheKey(hash.Bytes())
if db.cacheItems.Contains(cacheKey) {
return true, nil
}
sc := db.Session.Copy()
defer sc.Close()
var (
count int
err error
)
query := bson.M{"hash": hash.Hex()}
switch item := val.(type) {
case *tradingstate.OrderItem:
// Find key in ordersCollection collection
count, err = sc.DB(db.dbName).C(ordersCollection).Find(query).Limit(1).Count()
if err != nil {
return false, err
}
if count == 1 {
return true, nil
}
case *tradingstate.Trade:
// Find key in tradesCollection collection
count, err = sc.DB(db.dbName).C(tradesCollection).Find(query).Limit(1).Count()
if err != nil {
return false, err
}
if count == 1 {
return true, nil
}
case *lendingstate.LendingItem:
// Find key in lendingItemsCollection collection
switch item.Type {
case lendingstate.Repay:
count, err = sc.DB(db.dbName).C(lendingRepayCollection).Find(query).Limit(1).Count()
case lendingstate.TopUp:
count, err = sc.DB(db.dbName).C(lendingTopUpCollection).Find(query).Limit(1).Count()
case lendingstate.Recall:
count, err = sc.DB(db.dbName).C(lendingRecallCollection).Find(query).Limit(1).Count()
default:
count, err = sc.DB(db.dbName).C(lendingItemsCollection).Find(query).Limit(1).Count()
}
if err != nil {
return false, err
}
if count == 1 {
return true, nil
}
case *lendingstate.LendingTrade:
// Find key in lendingTradesCollection collection
count, err = sc.DB(db.dbName).C(lendingTradesCollection).Find(query).Limit(1).Count()
if err != nil {
return false, err
}
if count == 1 {
return true, nil
}
}
return false, nil
}
func (db *MongoDatabase) GetObject(hash common.Hash, val interface{}) (interface{}, error) {
if db.IsEmptyKey(hash.Bytes()) {
return nil, nil
}
cacheKey := db.getCacheKey(hash.Bytes())
if cached, ok := db.cacheItems.Get(cacheKey); ok {
return cached, nil
} else {
sc := db.Session.Copy()
defer sc.Close()
query := bson.M{"hash": hash.Hex()}
switch item := val.(type) {
case *tradingstate.OrderItem:
var oi *tradingstate.OrderItem
err := sc.DB(db.dbName).C(ordersCollection).Find(query).One(&oi)
if err != nil {
return nil, err
}
db.cacheItems.Add(cacheKey, oi)
return oi, nil
case *tradingstate.Trade:
var t *tradingstate.Trade
err := sc.DB(db.dbName).C(tradesCollection).Find(query).One(&t)
if err != nil {
return nil, err
}
db.cacheItems.Add(cacheKey, t)
return t, nil
case *lendingstate.LendingItem:
var li *lendingstate.LendingItem
var err error
switch item.Type {
case lendingstate.Repay:
err = sc.DB(db.dbName).C(lendingRepayCollection).Find(query).One(&li)
case lendingstate.TopUp:
err = sc.DB(db.dbName).C(lendingTopUpCollection).Find(query).One(&li)
case lendingstate.Recall:
err = sc.DB(db.dbName).C(lendingRecallCollection).Find(query).One(&li)
default:
err = sc.DB(db.dbName).C(lendingItemsCollection).Find(query).One(&li)
}
if err != nil {
return nil, err
}
db.cacheItems.Add(cacheKey, li)
return li, nil
case *lendingstate.LendingTrade:
var t *lendingstate.LendingTrade
err := sc.DB(db.dbName).C(lendingTradesCollection).Find(query).One(&t)
if err != nil {
return nil, err
}
db.cacheItems.Add(cacheKey, t)
return t, nil
default:
return nil, nil
}
}
}
func (db *MongoDatabase) PutObject(hash common.Hash, val interface{}) error {
cacheKey := db.getCacheKey(hash.Bytes())
db.cacheItems.Add(cacheKey, val)
switch item := val.(type) {
case *tradingstate.Trade:
// PutObject trade into tradesCollection collection
db.tradeBulk.Insert(item)
case *tradingstate.OrderItem:
// PutObject order into ordersCollection collection
if item.Status == tradingstate.OrderStatusOpen {
db.orderBulk.Insert(item)
} else {
query := bson.M{"hash": item.Hash.Hex()}
db.orderBulk.Upsert(query, item)
}
return nil
case *tradingstate.EpochPriceItem:
query := bson.M{"hash": item.Hash.Hex()}
db.epochPriceBulk.Upsert(query, item)
return nil
case *lendingstate.LendingTrade:
// PutObject LendingTrade into tradesCollection collection
if existed, err := db.HasObject(hash, val); err == nil && existed {
query := bson.M{"hash": item.Hash.Hex()}
db.lendingTradeBulk.Upsert(query, item)
} else {
db.lendingTradeBulk.Insert(item)
}
case *lendingstate.LendingItem:
// PutObject order into ordersCollection collection
switch item.Type {
case lendingstate.Repay:
if item.Status != lendingstate.LendingStatusReject {
item.Status = lendingstate.Repay
}
db.repayBulk.Insert(item)
return nil
case lendingstate.TopUp:
if item.Status != lendingstate.LendingStatusReject {
item.Status = lendingstate.TopUp
}
db.topUpBulk.Insert(item)
return nil
case lendingstate.Recall:
if item.Status != lendingstate.LendingStatusReject {
item.Status = lendingstate.Recall
}
db.recallBulk.Insert(item)
return nil
default:
if item.Status == lendingstate.LendingStatusOpen {
db.lendingItemBulk.Insert(item)
} else {
query := bson.M{"hash": item.Hash.Hex()}
db.lendingItemBulk.Upsert(query, item)
}
return nil
}
default:
log.Error("PutObject: unknown type of object", "val", val)
}
return nil
}
func (db *MongoDatabase) DeleteObject(hash common.Hash, val interface{}) error {
cacheKey := db.getCacheKey(hash.Bytes())
db.cacheItems.Remove(cacheKey)
sc := db.Session.Copy()
defer sc.Close()
query := bson.M{"hash": hash.Hex()}
found, err := db.HasObject(hash, val)
if err != nil {
return err
}
if found {
var err error
switch item := val.(type) {
case *tradingstate.OrderItem:
err = sc.DB(db.dbName).C(ordersCollection).Remove(query)
if err != nil && err != mgo.ErrNotFound {
return fmt.Errorf("failed to delete orderItem. Err: %v", err)
}
case *tradingstate.Trade:
err = sc.DB(db.dbName).C(tradesCollection).Remove(query)
if err != nil && err != mgo.ErrNotFound {
return fmt.Errorf("failed to delete XDCx trade. Err: %v", err)
}
case *lendingstate.LendingItem:
switch item.Type {
case lendingstate.Repay:
err = sc.DB(db.dbName).C(lendingRepayCollection).Remove(query)
case lendingstate.TopUp:
err = sc.DB(db.dbName).C(lendingTopUpCollection).Remove(query)
case lendingstate.Recall:
err = sc.DB(db.dbName).C(lendingRecallCollection).Remove(query)
default:
err = sc.DB(db.dbName).C(lendingItemsCollection).Remove(query)
}
if err != nil && err != mgo.ErrNotFound {
return fmt.Errorf("failed to delete lendingItem. Err: %v", err)
}
case *lendingstate.LendingTrade:
err = sc.DB(db.dbName).C(lendingTradesCollection).Remove(query)
if err != nil && err != mgo.ErrNotFound {
return fmt.Errorf("failed to delete lendingTrade. Err: %v", err)
}
}
}
return nil
}
func (db *MongoDatabase) InitBulk() {
sc := db.Session
db.orderBulk = sc.DB(db.dbName).C(ordersCollection).Bulk()
db.tradeBulk = sc.DB(db.dbName).C(tradesCollection).Bulk()
db.epochPriceBulk = sc.DB(db.dbName).C(epochPriceCollection).Bulk()
}
func (db *MongoDatabase) InitLendingBulk() {
sc := db.Session
db.lendingItemBulk = sc.DB(db.dbName).C(lendingItemsCollection).Bulk()
db.lendingTradeBulk = sc.DB(db.dbName).C(lendingTradesCollection).Bulk()
db.topUpBulk = sc.DB(db.dbName).C(lendingTopUpCollection).Bulk()
db.repayBulk = sc.DB(db.dbName).C(lendingRepayCollection).Bulk()
db.recallBulk = sc.DB(db.dbName).C(lendingRecallCollection).Bulk()
}
func (db *MongoDatabase) CommitBulk() error {
if _, err := db.orderBulk.Run(); err != nil && !mgo.IsDup(err) {
return err
}
if _, err := db.tradeBulk.Run(); err != nil && !mgo.IsDup(err) {
return err
}
if _, err := db.epochPriceBulk.Run(); err != nil && !mgo.IsDup(err) {
return err
}
return nil
}
func (db *MongoDatabase) CommitLendingBulk() error {
if _, err := db.lendingItemBulk.Run(); err != nil && !mgo.IsDup(err) {
return err
}
if _, err := db.lendingTradeBulk.Run(); err != nil && !mgo.IsDup(err) {
return err
}
if _, err := db.topUpBulk.Run(); err != nil && !mgo.IsDup(err) {
return err
}
if _, err := db.repayBulk.Run(); err != nil && !mgo.IsDup(err) {
return err
}
if _, err := db.recallBulk.Run(); err != nil && !mgo.IsDup(err) {
return err
}
return nil
}
func (db *MongoDatabase) Put(key []byte, val []byte) error {
// for levelDB only
return nil
}
func (db *MongoDatabase) Delete(key []byte) error {
// for levelDB only
return nil
}
func (db *MongoDatabase) Has(key []byte) (bool, error) {
// for levelDB only
return false, nil
}
func (db *MongoDatabase) Get(key []byte) ([]byte, error) {
// for levelDB only
return nil, nil
}
func (db *MongoDatabase) DeleteItemByTxHash(txhash common.Hash, val interface{}) {
sc := db.Session.Copy()
defer sc.Close()
query := bson.M{"txHash": txhash.Hex()}
switch item := val.(type) {
case *tradingstate.OrderItem:
if err := sc.DB(db.dbName).C(ordersCollection).Remove(query); err != nil && err != mgo.ErrNotFound {
log.Error("DeleteItemByTxHash: failed to delete order", "txhash", txhash, "err", err)
}
case *tradingstate.Trade:
if err := sc.DB(db.dbName).C(tradesCollection).Remove(query); err != nil && err != mgo.ErrNotFound {
log.Error("DeleteItemByTxHash: failed to delete trade", "txhash", txhash, "err", err)
}
case *lendingstate.LendingItem:
switch item.Type {
case lendingstate.Repay:
if err := sc.DB(db.dbName).C(lendingRepayCollection).Remove(query); err != nil && err != mgo.ErrNotFound {
log.Error("DeleteItemByTxHash: failed to delete repayItem", "txhash", txhash, "err", err)
}
return
case lendingstate.TopUp:
if err := sc.DB(db.dbName).C(lendingTopUpCollection).Remove(query); err != nil && err != mgo.ErrNotFound {
log.Error("DeleteItemByTxHash: failed to delete topupItem", "txhash", txhash, "err", err)
}
return
case lendingstate.Recall:
if err := sc.DB(db.dbName).C(lendingRecallCollection).Remove(query); err != nil && err != mgo.ErrNotFound {
log.Error("DeleteItemByTxHash: failed to delete recallItem", "txhash", txhash, "err", err)
}
return
default:
if err := sc.DB(db.dbName).C(lendingItemsCollection).Remove(query); err != nil && err != mgo.ErrNotFound {
log.Error("DeleteItemByTxHash: failed to delete lendingItem", "txhash", txhash, "err", err)
}
return
}
case *lendingstate.LendingTrade:
if err := sc.DB(db.dbName).C(lendingTradesCollection).Remove(query); err != nil && err != mgo.ErrNotFound {
log.Error("DeleteItemByTxHash: failed to delete lendingTrade", "txhash", txhash, "err", err)
}
default:
log.Error("DeleteItemByTxHash: Unknown object type", "txhash", txhash, "object", val)
}
}
func (db *MongoDatabase) GetListItemByTxHash(txhash common.Hash, val interface{}) interface{} {
sc := db.Session.Copy()
defer sc.Close()
query := bson.M{"txHash": txhash.Hex()}
switch item := val.(type) {
case *tradingstate.OrderItem:
result := []*tradingstate.OrderItem{}
if err := sc.DB(db.dbName).C(ordersCollection).Find(query).All(&result); err != nil && err != mgo.ErrNotFound {
log.Error("failed to GetListItemByTxHash (orders)", "err", err, "Txhash", txhash)
}
return result
case *tradingstate.Trade:
result := []*tradingstate.Trade{}
if err := sc.DB(db.dbName).C(tradesCollection).Find(query).All(&result); err != nil && err != mgo.ErrNotFound {
log.Error("failed to GetListItemByTxHash (trades)", "err", err, "Txhash", txhash)
}
return result
case *lendingstate.LendingItem:
result := []*lendingstate.LendingItem{}
switch item.Type {
case lendingstate.Repay:
if err := sc.DB(db.dbName).C(lendingRepayCollection).Find(query).All(&result); err != nil && err != mgo.ErrNotFound {
log.Error("failed to GetListItemByTxHash (repayItems)", "err", err, "txhash", txhash)
}
return result
case lendingstate.TopUp:
if err := sc.DB(db.dbName).C(lendingTopUpCollection).Find(query).All(&result); err != nil && err != mgo.ErrNotFound {
log.Error("failed to GetListItemByTxHash (topupItems)", "err", err, "txhash", txhash)
}
return result
case lendingstate.Recall:
if err := sc.DB(db.dbName).C(lendingRecallCollection).Find(query).All(&result); err != nil && err != mgo.ErrNotFound {
log.Error("failed to GetListItemByTxHash (recallItems)", "err", err, "txhash", txhash)
}
return result
default:
if err := sc.DB(db.dbName).C(lendingItemsCollection).Find(query).All(&result); err != nil && err != mgo.ErrNotFound {
log.Error("failed to GetListItemByTxHash (lendingItems)", "err", err, "txhash", txhash)
}
return result
}
case *lendingstate.LendingTrade:
result := []*lendingstate.LendingTrade{}
if err := sc.DB(db.dbName).C(lendingTradesCollection).Find(query).All(&result); err != nil && err != mgo.ErrNotFound {
log.Error("failed to GetListItemByTxHash (lendingTrades)", "err", err, "Txhash", txhash)
}
return result
default:
log.Error("GetListItemByTxHash: Unknown object type", "txhash", txhash, "object", val)
}
return nil
}
func (db *MongoDatabase) GetListItemByHashes(hashes []string, val interface{}) interface{} {
sc := db.Session.Copy()
defer sc.Close()
query := bson.M{"hash": bson.M{"$in": hashes}}
switch item := val.(type) {
case *tradingstate.OrderItem:
result := []*tradingstate.OrderItem{}
if err := sc.DB(db.dbName).C(ordersCollection).Find(query).All(&result); err != nil && err != mgo.ErrNotFound {
log.Error("failed to GetListItemByHashes (orders)", "err", err, "hashes", hashes)
}
return result
case *tradingstate.Trade:
result := []*tradingstate.Trade{}
if err := sc.DB(db.dbName).C(tradesCollection).Find(query).All(&result); err != nil && err != mgo.ErrNotFound {
log.Error("failed to GetListItemByHashes (trades)", "err", err, "hashes", hashes)
}
return result
case *lendingstate.LendingItem:
result := []*lendingstate.LendingItem{}
switch item.Type {
case lendingstate.Repay:
if err := sc.DB(db.dbName).C(lendingRepayCollection).Find(query).All(&result); err != nil && err != mgo.ErrNotFound {
log.Error("failed to GetListItemByHashes (repayItems)", "err", err, "hashes", hashes)
}
return result
case lendingstate.TopUp:
if err := sc.DB(db.dbName).C(lendingTopUpCollection).Find(query).All(&result); err != nil && err != mgo.ErrNotFound {
log.Error("failed to GetListItemByHashes (topupItems)", "err", err, "hashes", hashes)
}
return result
case lendingstate.Recall:
if err := sc.DB(db.dbName).C(lendingRecallCollection).Find(query).All(&result); err != nil && err != mgo.ErrNotFound {
log.Error("failed to GetListItemByHashes (recallItems)", "err", err, "hashes", hashes)
}
return result
default:
if err := sc.DB(db.dbName).C(lendingItemsCollection).Find(query).All(&result); err != nil && err != mgo.ErrNotFound {
log.Error("failed to GetListItemByHashes (lendingItems)", "err", err, "hashes", hashes)
}
return result
}
case *lendingstate.LendingTrade:
result := []*lendingstate.LendingTrade{}
if err := sc.DB(db.dbName).C(lendingTradesCollection).Find(query).All(&result); err != nil && err != mgo.ErrNotFound {
log.Error("failed to GetListItemByHashes (lendingTrades)", "err", err, "hashes", hashes)
}
return result
default:
log.Error("GetListItemByHashes: Unknown object type", "hashes", hashes, "object", val)
}
return nil
}
func (db *MongoDatabase) EnsureIndexes() error {
orderHashIndex := mgo.Index{
Key: []string{"hash"},
Unique: true,
DropDups: true,
Background: true,
Sparse: true,
Name: "index_order_hash",
}
orderTxHashIndex := mgo.Index{
Key: []string{"txHash"},
DropDups: true,
Background: true,
Sparse: true,
Name: "index_order_tx_hash",
}
tradeHashIndex := mgo.Index{
Key: []string{"hash"},
Unique: true,
DropDups: true,
Background: true,
Sparse: true,
Name: "index_trade_hash",
}
tradeTxHashIndex := mgo.Index{
Key: []string{"txHash"},
DropDups: true,
Background: true,
Sparse: true,
Name: "index_trade_tx_hash",
}
lendingItemHashIndex := mgo.Index{
Key: []string{"hash"},
Unique: true,
DropDups: true,
Background: true,
Sparse: true,
Name: "index_lending_item_hash",
}
lendingItemTxHashIndex := mgo.Index{
Key: []string{"txHash"},
DropDups: true,
Background: true,
Sparse: true,
Name: "index_lending_item_tx_hash",
}
lendingTradeHashIndex := mgo.Index{
Key: []string{"hash"},
Unique: true,
DropDups: true,
Background: true,
Sparse: true,
Name: "index_lending_trade_hash",
}
lendingTradeTxHashIndex := mgo.Index{
Key: []string{"txHash"},
DropDups: true,
Background: true,
Sparse: true,
Name: "index_lending_trade_tx_hash",
}
repayHashIndex := mgo.Index{
Key: []string{"hash"},
DropDups: true,
Background: true,
Sparse: true,
Name: "index_lending_repay_hash",
}
repayTxHashIndex := mgo.Index{
Key: []string{"txHash"},
DropDups: true,
Background: true,
Sparse: true,
Name: "index_lending_repay_tx_hash",
}
repayUniqueIndex := mgo.Index{
Key: []string{"txHash", "hash"},
Unique: true,
DropDups: true,
Background: true,
Sparse: true,
Name: "index_lending_repay_unique",
}
recallHashIndex := mgo.Index{
Key: []string{"hash"},
DropDups: true,
Background: true,
Sparse: true,
Name: "index_lending_recall_hash",
}
recallTxHashIndex := mgo.Index{
Key: []string{"txHash"},
DropDups: true,
Background: true,
Sparse: true,
Name: "index_lending_recall_tx_hash",
}
recallUniqueIndex := mgo.Index{
Key: []string{"txHash", "hash"},
Unique: true,
DropDups: true,
Background: true,
Sparse: true,
Name: "index_lending_recall_unique",
}
topupHashIndex := mgo.Index{
Key: []string{"hash"},
DropDups: true,
Background: true,
Sparse: true,
Name: "index_lending_topup_hash",
}
topupTxHashIndex := mgo.Index{
Key: []string{"txHash"},
DropDups: true,
Background: true,
Sparse: true,
Name: "index_lending_topup_tx_hash",
}
topUpUniqueIndex := mgo.Index{
Key: []string{"txHash", "hash"},
Unique: true,
DropDups: true,
Background: true,
Sparse: true,
Name: "index_lending_topup_unique",
}
epochPriceIndex := mgo.Index{
Key: []string{"hash"},
Unique: true,
DropDups: true,
Background: true,
Sparse: true,
Name: "index_epoch_price",
}
sc := db.Session.Copy()
defer sc.Close()
indexes, _ := sc.DB(db.dbName).C(ordersCollection).Indexes()
if !existingIndex(orderHashIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(ordersCollection).EnsureIndex(orderHashIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", orderHashIndex.Name, err)
}
}
if !existingIndex(orderTxHashIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(ordersCollection).EnsureIndex(orderTxHashIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", orderTxHashIndex.Name, err)
}
}
indexes, _ = sc.DB(db.dbName).C(tradesCollection).Indexes()
if !existingIndex(tradeHashIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(tradesCollection).EnsureIndex(tradeHashIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", tradeHashIndex.Name, err)
}
}
if !existingIndex(tradeTxHashIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(tradesCollection).EnsureIndex(tradeTxHashIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", tradeTxHashIndex.Name, err)
}
}
indexes, _ = sc.DB(db.dbName).C(lendingItemsCollection).Indexes()
if !existingIndex(lendingItemHashIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(lendingItemsCollection).EnsureIndex(lendingItemHashIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", lendingItemHashIndex.Name, err)
}
}
if !existingIndex(lendingItemTxHashIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(lendingItemsCollection).EnsureIndex(lendingItemTxHashIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", lendingItemTxHashIndex.Name, err)
}
}
indexes, _ = sc.DB(db.dbName).C(lendingTradesCollection).Indexes()
if !existingIndex(lendingTradeHashIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(lendingTradesCollection).EnsureIndex(lendingTradeHashIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", lendingTradeHashIndex.Name, err)
}
}
if !existingIndex(lendingTradeTxHashIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(lendingTradesCollection).EnsureIndex(lendingTradeTxHashIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", lendingTradeTxHashIndex.Name, err)
}
}
indexes, _ = sc.DB(db.dbName).C(lendingRepayCollection).Indexes()
if !existingIndex(repayHashIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(lendingRepayCollection).EnsureIndex(repayHashIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", repayHashIndex.Name, err)
}
}
if !existingIndex(repayTxHashIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(lendingRepayCollection).EnsureIndex(repayTxHashIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", repayTxHashIndex.Name, err)
}
}
if !existingIndex(repayUniqueIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(lendingRepayCollection).EnsureIndex(repayUniqueIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", repayUniqueIndex.Name, err)
}
}
indexes, _ = sc.DB(db.dbName).C(lendingRecallCollection).Indexes()
if !existingIndex(recallHashIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(lendingRecallCollection).EnsureIndex(recallHashIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", recallHashIndex.Name, err)
}
}
if !existingIndex(recallTxHashIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(lendingRecallCollection).EnsureIndex(recallTxHashIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", recallTxHashIndex.Name, err)
}
}
if !existingIndex(recallUniqueIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(lendingRecallCollection).EnsureIndex(repayUniqueIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", repayUniqueIndex.Name, err)
}
}
indexes, _ = sc.DB(db.dbName).C(lendingTopUpCollection).Indexes()
if !existingIndex(topupHashIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(lendingTopUpCollection).EnsureIndex(topupHashIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", topupHashIndex.Name, err)
}
}
if !existingIndex(topupTxHashIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(lendingTopUpCollection).EnsureIndex(topupTxHashIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", topupTxHashIndex.Name, err)
}
}
if !existingIndex(topUpUniqueIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(lendingTopUpCollection).EnsureIndex(repayUniqueIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", repayUniqueIndex.Name, err)
}
}
indexes, _ = sc.DB(db.dbName).C(epochPriceCollection).Indexes()
if !existingIndex(epochPriceIndex.Name, indexes) {
if err := sc.DB(db.dbName).C(epochPriceCollection).EnsureIndex(epochPriceIndex); err != nil {
return fmt.Errorf("failed to create index %s . Err: %v", epochPriceIndex.Name, err)
}
}
return nil
}
func (db *MongoDatabase) Close() error {
if db.Session != nil {
db.Session.Close()
}
return nil
}
// HasAncient returns an error as we don't have a backing chain freezer.
func (db *MongoDatabase) HasAncient(kind string, number uint64) (bool, error) {
return false, errNotSupported
}
// Ancient returns an error as we don't have a backing chain freezer.
func (db *MongoDatabase) Ancient(kind string, number uint64) ([]byte, error) {
return nil, errNotSupported
}
// Ancients returns an error as we don't have a backing chain freezer.
func (db *MongoDatabase) Ancients() (uint64, error) {
return 0, errNotSupported
}
// AncientSize returns an error as we don't have a backing chain freezer.
func (db *MongoDatabase) AncientSize(kind string) (uint64, error) {
return 0, errNotSupported
}
// AppendAncient returns an error as we don't have a backing chain freezer.
func (db *MongoDatabase) AppendAncient(number uint64, hash, header, body, receipts, td []byte) error {
return errNotSupported
}
// TruncateAncients returns an error as we don't have a backing chain freezer.
func (db *MongoDatabase) TruncateAncients(items uint64) error {
return errNotSupported
}
// Sync returns an error as we don't have a backing chain freezer.
func (db *MongoDatabase) Sync() error {
return errNotSupported
}
func (db *MongoDatabase) NewIterator(prefix []byte, start []byte) ethdb.Iterator {
panic("NewIterator from XDCxDAO mongodb is not supported")
}
func (db *MongoDatabase) Stat(property string) (string, error) {
return "", errNotSupported
}
func (db *MongoDatabase) Compact(start []byte, limit []byte) error {
return errNotSupported
}
func (db *MongoDatabase) NewBatch() ethdb.Batch {
// for levelDB only
return nil
}
func (db *MongoDatabase) NewBatchWithSize(size int) ethdb.Batch {
return nil
}
type Batch struct {
collection string
}
func (b *Batch) Put(key, value []byte) error {
// for levelDB only
return nil
}
func (b *Batch) Write() error {
// for levelDB only
return nil
}
func (b *Batch) ValueSize() int {
// for levelDB only
return int(0)
}
func (b *Batch) Reset() {
// for levelDB only
}
func existingIndex(indexName string, indexes []mgo.Index) bool {
if len(indexes) == 0 {
return false
}
for _, index := range indexes {
if index.Name == indexName {
return true
}
}
return false
}

554
XDCxlending/XDCxlending.go
View file

@ -3,10 +3,8 @@ package XDCxlending
import ( import (
"encoding/json" "encoding/json"
"errors" "errors"
"fmt"
"math/big" "math/big"
"strconv" "strconv"
"time"
"github.com/XinFinOrg/XDPoSChain/XDCx" "github.com/XinFinOrg/XDPoSChain/XDCx"
"github.com/XinFinOrg/XDPoSChain/XDCx/tradingstate" "github.com/XinFinOrg/XDPoSChain/XDCx/tradingstate"
@ -56,10 +54,6 @@ func (l *Lending) GetLevelDB() XDCxDAO.XDCXDAO {
return l.XDCx.GetLevelDB() return l.XDCx.GetLevelDB()
} }
func (l *Lending) GetMongoDB() XDCxDAO.XDCXDAO {
return l.XDCx.GetMongoDB()
}
func (l *Lending) ProcessOrderPending(header *types.Header, coinbase common.Address, chain consensus.ChainContext, pending map[common.Address]types.LendingTransactions, statedb *state.StateDB, lendingStatedb *lendingstate.LendingStateDB, tradingStateDb *tradingstate.TradingStateDB) ([]*lendingstate.LendingItem, map[common.Hash]lendingstate.MatchingResult) { func (l *Lending) ProcessOrderPending(header *types.Header, coinbase common.Address, chain consensus.ChainContext, pending map[common.Address]types.LendingTransactions, statedb *state.StateDB, lendingStatedb *lendingstate.LendingStateDB, tradingStateDb *tradingstate.TradingStateDB) ([]*lendingstate.LendingItem, map[common.Hash]lendingstate.MatchingResult) {
lendingItems := []*lendingstate.LendingItem{} lendingItems := []*lendingstate.LendingItem{}
matchingResults := map[common.Hash]lendingstate.MatchingResult{} matchingResults := map[common.Hash]lendingstate.MatchingResult{}
@ -151,474 +145,6 @@ func (l *Lending) ProcessOrderPending(header *types.Header, coinbase common.Addr
return lendingItems, matchingResults return lendingItems, matchingResults
} }
// there are 3 tasks need to complete (for SDK nodes) after matching
// 1. Put takerLendingItem to database
// 2.a Update status, filledAmount of makerLendingItem
// 2.b. Put lendingTrade to database
// 3. Update status of rejected items
func (l *Lending) SyncDataToSDKNode(chain consensus.ChainContext, statedb *state.StateDB, block *types.Block, takerLendingItem *lendingstate.LendingItem, txHash common.Hash, txMatchTime time.Time, trades []*lendingstate.LendingTrade, rejectedItems []*lendingstate.LendingItem, dirtyOrderCount *uint64) error {
var (
// originTakerLendingItem: item getting from database
originTakerLendingItem, updatedTakerLendingItem *lendingstate.LendingItem
makerDirtyHashes []string
makerDirtyFilledAmount map[string]*big.Int
err error
)
db := l.GetMongoDB()
db.InitLendingBulk()
if takerLendingItem.Status == lendingstate.LendingStatusCancelled && len(rejectedItems) > 0 {
// cancel order is rejected -> nothing change
log.Debug("Cancel order is rejected", "order", lendingstate.ToJSON(takerLendingItem))
return nil
}
// 1. put processed takerLendingItem to database
lastState := lendingstate.LendingItemHistoryItem{}
// Typically, takerItem has never existed in database
// except cancel case: in this case, item existed in database with status = OPEN, then use send another lendingItem to cancel it
val, err := db.GetObject(takerLendingItem.Hash, &lendingstate.LendingItem{Type: takerLendingItem.Type})
if err == nil && val != nil {
originTakerLendingItem = val.(*lendingstate.LendingItem)
lastState = lendingstate.LendingItemHistoryItem{
TxHash: originTakerLendingItem.TxHash,
FilledAmount: lendingstate.CloneBigInt(originTakerLendingItem.FilledAmount),
Status: originTakerLendingItem.Status,
UpdatedAt: originTakerLendingItem.UpdatedAt,
}
}
if originTakerLendingItem != nil {
updatedTakerLendingItem = originTakerLendingItem
} else {
updatedTakerLendingItem = takerLendingItem
updatedTakerLendingItem.FilledAmount = new(big.Int)
}
if takerLendingItem.Status == lendingstate.LendingStatusNew {
updatedTakerLendingItem.Status = lendingstate.LendingStatusOpen
} else if takerLendingItem.Status == lendingstate.LendingStatusCancelled {
updatedTakerLendingItem.Status = lendingstate.LendingStatusCancelled
updatedTakerLendingItem.ExtraData = takerLendingItem.ExtraData
}
updatedTakerLendingItem.TxHash = txHash
if updatedTakerLendingItem.CreatedAt.IsZero() {
updatedTakerLendingItem.CreatedAt = txMatchTime
}
if txMatchTime.Before(updatedTakerLendingItem.UpdatedAt) || (txMatchTime.Equal(updatedTakerLendingItem.UpdatedAt) && *dirtyOrderCount == 0) {
log.Debug("Ignore old lendingItem/lendingTrades taker", "txHash", txHash.Hex(), "txTime", txMatchTime.UnixNano(), "updatedAt", updatedTakerLendingItem.UpdatedAt.UnixNano())
return nil
}
*dirtyOrderCount++
l.UpdateLendingItemCache(updatedTakerLendingItem.LendingToken, updatedTakerLendingItem.CollateralToken, updatedTakerLendingItem.Hash, txHash, lastState)
updatedTakerLendingItem.UpdatedAt = txMatchTime
// 2. put trades to database and update status
log.Debug("Got lendingTrades", "number", len(trades), "txhash", txHash.Hex())
makerDirtyFilledAmount = make(map[string]*big.Int)
tradeList := map[common.Hash]*lendingstate.LendingTrade{}
for _, tradeRecord := range trades {
// 2.a. put to trades
if tradeRecord == nil {
continue
}
if updatedTakerLendingItem.Type == lendingstate.Repay || updatedTakerLendingItem.Type == lendingstate.TopUp || updatedTakerLendingItem.Type == lendingstate.Recall {
// repay, topup: assign hash = trade.hash
updatedTakerLendingItem.Hash = tradeRecord.Hash
updatedTakerLendingItem.CollateralToken = tradeRecord.CollateralToken
updatedTakerLendingItem.FilledAmount = updatedTakerLendingItem.Quantity
updatedTakerLendingItem.Interest = new(big.Int).SetUint64(tradeRecord.Interest)
switch updatedTakerLendingItem.Type {
case lendingstate.TopUp:
updatedTakerLendingItem.Status = lendingstate.TopUp
extraData, _ := json.Marshal(struct {
Price *big.Int
}{
Price: new(big.Int).Div(new(big.Int).Mul(tradeRecord.LiquidationPrice, tradeRecord.DepositRate), tradeRecord.LiquidationRate),
})
updatedTakerLendingItem.ExtraData = string(extraData)
// manual topUp item
updatedTakerLendingItem.AutoTopUp = false
case lendingstate.Repay:
updatedTakerLendingItem.Status = lendingstate.Repay
paymentBalance := lendingstate.CalculateTotalRepayValue(block.Time(), tradeRecord.LiquidationTime, tradeRecord.Term, tradeRecord.Interest, tradeRecord.Amount)
updatedTakerLendingItem.Quantity = paymentBalance
updatedTakerLendingItem.FilledAmount = paymentBalance
// manual repay item
updatedTakerLendingItem.AutoTopUp = false
case lendingstate.Recall:
updatedTakerLendingItem.Status = lendingstate.Recall
// manual recall item
updatedTakerLendingItem.AutoTopUp = false
}
log.Debug("UpdateLendingTrade:", "type", updatedTakerLendingItem.Type, "hash", tradeRecord.Hash.Hex(), "status", tradeRecord.Status, "tradeId", tradeRecord.TradeId)
tradeList[tradeRecord.Hash] = tradeRecord
continue
}
if tradeRecord.CreatedAt.IsZero() {
tradeRecord.CreatedAt = txMatchTime
}
tradeRecord.UpdatedAt = txMatchTime
tradeRecord.TxHash = txHash
tradeRecord.Hash = tradeRecord.ComputeHash()
tradeList[tradeRecord.Hash] = tradeRecord
// 2.b. update status and filledAmount
filledAmount := new(big.Int)
if tradeRecord.Amount != nil {
filledAmount = lendingstate.CloneBigInt(tradeRecord.Amount)
}
// maker dirty order
makerFilledAmount := big.NewInt(0)
var makerOrderHash common.Hash
if updatedTakerLendingItem.Side == lendingstate.Borrowing {
makerOrderHash = tradeRecord.InvestingOrderHash
} else {
makerOrderHash = tradeRecord.BorrowingOrderHash
}
if amount, ok := makerDirtyFilledAmount[makerOrderHash.Hex()]; ok {
makerFilledAmount = lendingstate.CloneBigInt(amount)
}
makerFilledAmount = new(big.Int).Add(makerFilledAmount, filledAmount)
makerDirtyFilledAmount[makerOrderHash.Hex()] = makerFilledAmount
makerDirtyHashes = append(makerDirtyHashes, makerOrderHash.Hex())
if updatedTakerLendingItem.Type == lendingstate.Limit || updatedTakerLendingItem.Type == lendingstate.Market {
//updatedTakerOrder = l.updateMatchedOrder(updatedTakerOrder, filledAmount, txMatchTime, txHash)
// update filledAmount, status of takerOrder
updatedTakerLendingItem.FilledAmount = new(big.Int).Add(updatedTakerLendingItem.FilledAmount, filledAmount)
if updatedTakerLendingItem.FilledAmount.Cmp(updatedTakerLendingItem.Quantity) < 0 && updatedTakerLendingItem.Type == lendingstate.Limit {
updatedTakerLendingItem.Status = lendingstate.LendingStatusPartialFilled
} else {
updatedTakerLendingItem.Status = lendingstate.LendingStatusFilled
}
}
}
if err := l.UpdateLendingTrade(tradeList, txHash, txMatchTime); err != nil {
return err
}
// for Market orders
// filledAmount > 0 : FILLED
// otherwise: REJECTED
if updatedTakerLendingItem.Type == lendingstate.Market {
if updatedTakerLendingItem.FilledAmount.Sign() > 0 {
updatedTakerLendingItem.Status = lendingstate.LendingStatusFilled
} else {
updatedTakerLendingItem.Status = lendingstate.LendingStatusReject
}
}
log.Debug("PutObject processed takerLendingItem",
"term", updatedTakerLendingItem.Term, "userAddr", updatedTakerLendingItem.UserAddress.Hex(), "side", updatedTakerLendingItem.Side,
"Interest", updatedTakerLendingItem.Interest, "quantity", updatedTakerLendingItem.Quantity, "filledAmount", updatedTakerLendingItem.FilledAmount, "status", updatedTakerLendingItem.Status,
"hash", updatedTakerLendingItem.Hash.Hex(), "txHash", updatedTakerLendingItem.TxHash.Hex())
if !(updatedTakerLendingItem.Type == lendingstate.Repay || updatedTakerLendingItem.Type == lendingstate.TopUp || updatedTakerLendingItem.Type == lendingstate.Recall) || updatedTakerLendingItem.Status != lendingstate.LendingStatusOpen {
if err := db.PutObject(updatedTakerLendingItem.Hash, updatedTakerLendingItem); err != nil {
return fmt.Errorf("SDKNode: failed to put processed takerOrder. Hash: %s Error: %s", updatedTakerLendingItem.Hash.Hex(), err.Error())
}
}
items := db.GetListItemByHashes(makerDirtyHashes, &lendingstate.LendingItem{})
if items != nil {
makerItems := items.([]*lendingstate.LendingItem)
log.Debug("Maker dirty lendingItem", "len", len(makerItems), "txhash", txHash.Hex())
for _, m := range makerItems {
if txMatchTime.Before(m.UpdatedAt) {
log.Debug("Ignore old lendingItem/lendingTrades maker", "txHash", txHash.Hex(), "txTime", txMatchTime.UnixNano(), "updatedAt", m.UpdatedAt.UnixNano())
continue
}
lastState = lendingstate.LendingItemHistoryItem{
TxHash: m.TxHash,
FilledAmount: lendingstate.CloneBigInt(m.FilledAmount),
Status: m.Status,
UpdatedAt: m.UpdatedAt,
}
l.UpdateLendingItemCache(m.LendingToken, m.CollateralToken, m.Hash, txHash, lastState)
m.TxHash = txHash
m.UpdatedAt = txMatchTime
m.FilledAmount = new(big.Int).Add(m.FilledAmount, makerDirtyFilledAmount[m.Hash.Hex()])
if m.FilledAmount.Cmp(m.Quantity) < 0 {
m.Status = lendingstate.LendingStatusPartialFilled
} else {
m.Status = lendingstate.LendingStatusFilled
}
log.Debug("PutObject processed makerLendingItem",
"term", m.Term, "userAddr", m.UserAddress.Hex(), "side", m.Side,
"Interest", m.Interest, "quantity", m.Quantity, "filledAmount", m.FilledAmount, "status", m.Status,
"hash", m.Hash.Hex(), "txHash", m.TxHash.Hex())
if err := db.PutObject(m.Hash, m); err != nil {
return fmt.Errorf("SDKNode: failed to put processed makerOrder. Hash: %s Error: %s", m.Hash.Hex(), err.Error())
}
}
}
// 3. put rejected orders to leveldb and update status REJECTED
log.Debug("Got rejected lendingItems", "number", len(rejectedItems), "rejectedLendingItems", rejectedItems)
if len(rejectedItems) > 0 {
var rejectedHashes []string
// updateRejectedOrders
for _, r := range rejectedItems {
rejectedHashes = append(rejectedHashes, r.Hash.Hex())
if updatedTakerLendingItem.Hash == r.Hash && !txMatchTime.Before(r.UpdatedAt) {
// cache r history for handling reorg
historyRecord := lendingstate.LendingItemHistoryItem{
TxHash: updatedTakerLendingItem.TxHash,
FilledAmount: lendingstate.CloneBigInt(updatedTakerLendingItem.FilledAmount),
Status: updatedTakerLendingItem.Status,
UpdatedAt: updatedTakerLendingItem.UpdatedAt,
}
l.UpdateLendingItemCache(updatedTakerLendingItem.LendingToken, updatedTakerLendingItem.CollateralToken, updatedTakerLendingItem.Hash, txHash, historyRecord)
// if whole order is rejected, status = REJECTED
// otherwise, status = FILLED
if updatedTakerLendingItem.FilledAmount.Sign() > 0 {
updatedTakerLendingItem.Status = lendingstate.LendingStatusFilled
} else {
updatedTakerLendingItem.Status = lendingstate.LendingStatusReject
}
updatedTakerLendingItem.TxHash = txHash
updatedTakerLendingItem.UpdatedAt = txMatchTime
if err := db.PutObject(updatedTakerLendingItem.Hash, updatedTakerLendingItem); err != nil {
return fmt.Errorf("SDKNode: failed to reject takerOrder. Hash: %s Error: %s", updatedTakerLendingItem.Hash.Hex(), err.Error())
}
}
}
items := db.GetListItemByHashes(rejectedHashes, &lendingstate.LendingItem{})
if items != nil {
dirtyRejectedItems := items.([]*lendingstate.LendingItem)
for _, r := range dirtyRejectedItems {
if txMatchTime.Before(r.UpdatedAt) {
log.Debug("Ignore old orders/trades reject", "txHash", txHash.Hex(), "txTime", txMatchTime.UnixNano(), "updatedAt", updatedTakerLendingItem.UpdatedAt.UnixNano())
continue
}
// cache lendingItem for handling reorg
historyRecord := lendingstate.LendingItemHistoryItem{
TxHash: r.TxHash,
FilledAmount: lendingstate.CloneBigInt(r.FilledAmount),
Status: r.Status,
UpdatedAt: r.UpdatedAt,
}
l.UpdateLendingItemCache(r.LendingToken, r.CollateralToken, r.Hash, txHash, historyRecord)
dirtyFilledAmount, ok := makerDirtyFilledAmount[r.Hash.Hex()]
if ok && dirtyFilledAmount != nil {
r.FilledAmount = new(big.Int).Add(r.FilledAmount, dirtyFilledAmount)
}
// if whole order is rejected, status = REJECTED
// otherwise, status = FILLED
if r.FilledAmount.Sign() > 0 {
r.Status = lendingstate.LendingStatusFilled
} else {
r.Status = lendingstate.LendingStatusReject
}
r.TxHash = txHash
r.UpdatedAt = txMatchTime
if err = db.PutObject(r.Hash, r); err != nil {
return fmt.Errorf("SDKNode: failed to update rejectedOder to sdkNode %s", err.Error())
}
}
}
}
if err := db.CommitLendingBulk(); err != nil {
return fmt.Errorf("SDKNode fail to commit bulk update lendingItem/lendingTrades at txhash %s . Error: %s", txHash.Hex(), err.Error())
}
return nil
}
func (l *Lending) UpdateLiquidatedTrade(blockTime uint64, result lendingstate.FinalizedResult, trades map[common.Hash]*lendingstate.LendingTrade) error {
db := l.GetMongoDB()
db.InitLendingBulk()
txhash := result.TxHash
txTime := time.Unix(int64(blockTime), 0).UTC()
if err := l.UpdateLendingTrade(trades, txhash, txTime); err != nil {
return err
}
// adding auto repay transaction
if len(result.AutoRepay) > 0 {
for _, hash := range result.AutoRepay {
trade := trades[hash]
if trade == nil {
continue
}
paymentBalance := lendingstate.CalculateTotalRepayValue(blockTime, trade.LiquidationTime, trade.Term, trade.Interest, trade.Amount)
repayItem := &lendingstate.LendingItem{
Quantity: paymentBalance,
Interest: big.NewInt(int64(trade.Interest)),
Side: "",
Type: lendingstate.Repay,
LendingToken: trade.LendingToken,
CollateralToken: trade.CollateralToken,
FilledAmount: paymentBalance,
Status: lendingstate.Repay,
Relayer: trade.BorrowingRelayer,
Term: trade.Term,
UserAddress: trade.Borrower,
Signature: nil,
Hash: trade.Hash,
TxHash: txhash,
Nonce: nil,
CreatedAt: txTime,
UpdatedAt: txTime,
LendingId: 0,
LendingTradeId: trade.TradeId,
AutoTopUp: true, // auto repay
ExtraData: "",
}
if err := db.PutObject(repayItem.Hash, repayItem); err != nil {
return err
}
}
}
// adding auto topup transaction
if len(result.AutoTopUp) > 0 {
oldTradeHashes := []string{}
for _, hash := range result.AutoTopUp {
oldTradeHashes = append(oldTradeHashes, hash.Hex())
}
items := db.GetListItemByHashes(oldTradeHashes, &lendingstate.LendingTrade{})
if items != nil && len(items.([]*lendingstate.LendingTrade)) > 0 {
for _, oldTrade := range items.([]*lendingstate.LendingTrade) {
newTrade := trades[oldTrade.Hash]
topUpAmount := new(big.Int).Sub(newTrade.CollateralLockedAmount, oldTrade.CollateralLockedAmount)
extraData, _ := json.Marshal(struct {
Price *big.Int
}{
Price: new(big.Int).Div(new(big.Int).Mul(newTrade.LiquidationPrice, common.BaseTopUp), common.RateTopUp),
})
topUpItem := &lendingstate.LendingItem{
Quantity: topUpAmount,
Interest: big.NewInt(int64(oldTrade.Interest)),
Side: "",
Type: lendingstate.TopUp,
LendingToken: oldTrade.LendingToken,
CollateralToken: oldTrade.CollateralToken,
FilledAmount: topUpAmount,
Status: lendingstate.TopUp,
AutoTopUp: true, // auto topup
Relayer: oldTrade.BorrowingRelayer,
Term: oldTrade.Term,
UserAddress: oldTrade.Borrower,
Signature: nil,
Hash: oldTrade.Hash,
TxHash: txhash,
Nonce: nil,
CreatedAt: txTime,
UpdatedAt: txTime,
LendingId: 0,
LendingTradeId: oldTrade.TradeId,
ExtraData: string(extraData),
}
if err := db.PutObject(topUpItem.Hash, topUpItem); err != nil {
return err
}
}
}
}
// adding auto recall transaction
if len(result.AutoRecall) > 0 {
oldTradeHashes := []string{}
for _, hash := range result.AutoRecall {
oldTradeHashes = append(oldTradeHashes, hash.Hex())
}
items := db.GetListItemByHashes(oldTradeHashes, &lendingstate.LendingTrade{})
if items != nil && len(items.([]*lendingstate.LendingTrade)) > 0 {
for _, oldTrade := range items.([]*lendingstate.LendingTrade) {
newTrade := trades[oldTrade.Hash]
recallAmount := new(big.Int).Sub(oldTrade.CollateralLockedAmount, newTrade.CollateralLockedAmount)
extraData, _ := json.Marshal(struct {
Price *big.Int
}{
Price: new(big.Int).Div(new(big.Int).Mul(newTrade.LiquidationPrice, oldTrade.DepositRate), oldTrade.LiquidationRate),
})
topUpItem := &lendingstate.LendingItem{
Quantity: recallAmount,
Interest: big.NewInt(int64(oldTrade.Interest)),
Side: "",
Type: lendingstate.Recall,
LendingToken: oldTrade.LendingToken,
CollateralToken: oldTrade.CollateralToken,
FilledAmount: recallAmount,
Status: lendingstate.Recall,
AutoTopUp: true, // auto recall
Relayer: oldTrade.BorrowingRelayer,
Term: oldTrade.Term,
UserAddress: oldTrade.Borrower,
Signature: nil,
Hash: oldTrade.Hash,
TxHash: txhash,
Nonce: nil,
CreatedAt: txTime,
UpdatedAt: txTime,
LendingId: 0,
LendingTradeId: oldTrade.TradeId,
ExtraData: string(extraData),
}
if err := db.PutObject(topUpItem.Hash, topUpItem); err != nil {
return err
}
}
}
}
if err := db.CommitLendingBulk(); err != nil {
return fmt.Errorf("failed to updateLendingTrade . Err: %v", err)
}
return nil
}
func (l *Lending) UpdateLendingTrade(trades map[common.Hash]*lendingstate.LendingTrade, txhash common.Hash, txTime time.Time) error {
db := l.GetMongoDB()
hashQuery := []string{}
if len(trades) == 0 {
return nil
}
for _, trade := range trades {
hashQuery = append(hashQuery, trade.Hash.Hex())
}
items := db.GetListItemByHashes(hashQuery, &lendingstate.LendingTrade{})
if items != nil && len(items.([]*lendingstate.LendingTrade)) > 0 {
for _, trade := range items.([]*lendingstate.LendingTrade) {
history := lendingstate.LendingTradeHistoryItem{
TxHash: trade.TxHash,
CollateralLockedAmount: trade.CollateralLockedAmount,
LiquidationPrice: trade.LiquidationPrice,
Status: trade.Status,
UpdatedAt: trade.UpdatedAt,
}
l.UpdateLendingTradeCache(trade.Hash, txhash, history)
trade.TxHash = txhash
trade.UpdatedAt = txTime
newTrade := trades[trade.Hash]
trade.CollateralLockedAmount = newTrade.CollateralLockedAmount
trade.Status = newTrade.Status
trade.LiquidationPrice = newTrade.LiquidationPrice
trade.ExtraData = newTrade.ExtraData
if err := db.PutObject(trade.Hash, trade); err != nil {
return err
}
}
log.Debug("UpdateLendingTrade successfully", "txhash", txhash, "hash", hashQuery)
} else {
// not update, just upsert
for _, trade := range trades {
if err := db.PutObject(trade.Hash, trade); err != nil {
return err
}
}
}
return nil
}
func (l *Lending) GetLendingState(block *types.Block, author common.Address) (*lendingstate.LendingStateDB, error) { func (l *Lending) GetLendingState(block *types.Block, author common.Address) (*lendingstate.LendingStateDB, error) {
root, err := l.GetLendingStateRoot(block, author) root, err := l.GetLendingStateRoot(block, author)
if err != nil { if err != nil {
@ -690,86 +216,6 @@ func (l *Lending) UpdateLendingTradeCache(hash common.Hash, txhash common.Hash,
l.lendingTradeHistory.Add(txhash, lendingCacheAtTxHash) l.lendingTradeHistory.Add(txhash, lendingCacheAtTxHash)
} }
func (l *Lending) RollbackLendingData(txhash common.Hash) error {
db := l.GetMongoDB()
db.InitLendingBulk()
// rollback lendingItem
items := db.GetListItemByTxHash(txhash, &lendingstate.LendingItem{})
if items != nil {
for _, item := range items.([]*lendingstate.LendingItem) {
cacheAtTxHash, ok := l.lendingItemHistory.Get(txhash)
log.Debug("XDCxlending reorg: rollback lendingItem", "txhash", txhash.Hex(), "item", lendingstate.ToJSON(item), "lendingItemHistory", cacheAtTxHash)
if !ok || cacheAtTxHash == nil {
log.Debug("XDCxlending reorg: remove item due to no lendingItemHistory", "item", lendingstate.ToJSON(item))
if err := db.DeleteObject(item.Hash, &lendingstate.LendingItem{}); err != nil {
return fmt.Errorf("failed to remove reorg LendingItem. Err: %v . Item: %s", err.Error(), lendingstate.ToJSON(item))
}
continue
}
lendingItemHistory := cacheAtTxHash[lendingstate.GetLendingItemHistoryKey(item.LendingToken, item.CollateralToken, item.Hash)]
if (lendingItemHistory == lendingstate.LendingItemHistoryItem{}) {
log.Debug("XDCxlending reorg: remove item due to empty lendingItemHistory", "item", lendingstate.ToJSON(item))
if err := db.DeleteObject(item.Hash, &lendingstate.LendingItem{}); err != nil {
return fmt.Errorf("failed to remove reorg LendingItem. Err: %v . Item: %s", err.Error(), lendingstate.ToJSON(item))
}
continue
}
item.TxHash = lendingItemHistory.TxHash
item.Status = lendingItemHistory.Status
item.FilledAmount = lendingstate.CloneBigInt(lendingItemHistory.FilledAmount)
item.UpdatedAt = lendingItemHistory.UpdatedAt
log.Debug("XDCxlending reorg: update item to the last lendingItemHistory", "item", lendingstate.ToJSON(item), "lendingItemHistory", lendingItemHistory)
if err := db.PutObject(item.Hash, item); err != nil {
return fmt.Errorf("failed to update reorg LendingItem. Err: %v . Item: %s", err.Error(), lendingstate.ToJSON(item))
}
}
}
// rollback lendingTrade
items = db.GetListItemByTxHash(txhash, &lendingstate.LendingTrade{})
if items != nil {
for _, trade := range items.([]*lendingstate.LendingTrade) {
cacheAtTxHash, ok := l.lendingTradeHistory.Get(txhash)
log.Debug("XDCxlending reorg: rollback LendingTrade", "txhash", txhash.Hex(), "trade", lendingstate.ToJSON(trade), "LendingTradeHistory", cacheAtTxHash)
if !ok || cacheAtTxHash == nil {
log.Debug("XDCxlending reorg: remove trade due to no LendingTradeHistory", "trade", lendingstate.ToJSON(trade))
if err := db.DeleteObject(trade.Hash, &lendingstate.LendingTrade{}); err != nil {
return fmt.Errorf("failed to remove reorg LendingTrade. Err: %v . Trade: %s", err.Error(), lendingstate.ToJSON(trade))
}
continue
}
lendingTradeHistoryItem := cacheAtTxHash[trade.Hash]
if (lendingTradeHistoryItem == lendingstate.LendingTradeHistoryItem{}) {
log.Debug("XDCxlending reorg: remove trade due to empty LendingTradeHistory", "trade", lendingstate.ToJSON(trade))
if err := db.DeleteObject(trade.Hash, &lendingstate.LendingTrade{}); err != nil {
return fmt.Errorf("failed to remove reorg LendingTrade. Err: %v . Trade: %s", err.Error(), lendingstate.ToJSON(trade))
}
continue
}
trade.TxHash = lendingTradeHistoryItem.TxHash
trade.Status = lendingTradeHistoryItem.Status
trade.CollateralLockedAmount = lendingstate.CloneBigInt(lendingTradeHistoryItem.CollateralLockedAmount)
trade.LiquidationPrice = lendingstate.CloneBigInt(lendingTradeHistoryItem.LiquidationPrice)
trade.UpdatedAt = lendingTradeHistoryItem.UpdatedAt
log.Debug("XDCxlending reorg: update trade to the last lendingTradeHistoryItem", "trade", lendingstate.ToJSON(trade), "lendingTradeHistoryItem", lendingTradeHistoryItem)
if err := db.PutObject(trade.Hash, trade); err != nil {
return fmt.Errorf("failed to update reorg LendingTrade. Err: %v . Trade: %s", err.Error(), lendingstate.ToJSON(trade))
}
}
}
// remove repay/topup/recall history
db.DeleteItemByTxHash(txhash, &lendingstate.LendingItem{Type: lendingstate.Repay})
db.DeleteItemByTxHash(txhash, &lendingstate.LendingItem{Type: lendingstate.TopUp})
db.DeleteItemByTxHash(txhash, &lendingstate.LendingItem{Type: lendingstate.Recall})
if err := db.CommitLendingBulk(); err != nil {
return fmt.Errorf("failed to RollbackLendingData. %v", err)
}
return nil
}
func (l *Lending) ProcessLiquidationData(header *types.Header, chain consensus.ChainContext, statedb *state.StateDB, tradingState *tradingstate.TradingStateDB, lendingState *lendingstate.LendingStateDB) (updatedTrades map[common.Hash]*lendingstate.LendingTrade, liquidatedTrades, autoRepayTrades, autoTopUpTrades, autoRecallTrades []*lendingstate.LendingTrade, err error) { func (l *Lending) ProcessLiquidationData(header *types.Header, chain consensus.ChainContext, statedb *state.StateDB, tradingState *tradingstate.TradingStateDB, lendingState *lendingstate.LendingStateDB) (updatedTrades map[common.Hash]*lendingstate.LendingTrade, liquidatedTrades, autoRepayTrades, autoTopUpTrades, autoRecallTrades []*lendingstate.LendingTrade, err error) {
time := new(big.Int).SetUint64(header.Time) time := new(big.Int).SetUint64(header.Time)
updatedTrades = map[common.Hash]*lendingstate.LendingTrade{} // sum of liquidatedTrades, autoRepayTrades, autoTopUpTrades, autoRecallTrades updatedTrades = map[common.Hash]*lendingstate.LendingTrade{} // sum of liquidatedTrades, autoRepayTrades, autoTopUpTrades, autoRecallTrades

4
cmd/XDC/main.go
View file

@ -71,10 +71,6 @@ var (
utils.USBFlag, utils.USBFlag,
utils.SmartCardDaemonPathFlag, utils.SmartCardDaemonPathFlag,
utils.EnablePersonal, utils.EnablePersonal,
utils.XDCXEnabledFlag,
utils.XDCXDBEngineFlag,
utils.XDCXDBConnectionUrlFlag,
utils.XDCXDBReplicaSetNameFlag,
utils.XDCXDBNameFlag, utils.XDCXDBNameFlag,
utils.TxPoolNoLocalsFlag, utils.TxPoolNoLocalsFlag,
utils.TxPoolJournalFlag, utils.TxPoolJournalFlag,

38
cmd/utils/flags.go
View file

@ -845,18 +845,6 @@ var (
} }
// XDCX settings // XDCX settings
XDCXEnabledFlag = &cli.BoolFlag{
Name: "XDCx",
Usage: "Enable the XDCX protocol",
Category: flags.XdcxCategory,
}
XDCXDBEngineFlag = &cli.StringFlag{
Name: "XDCx-dbengine",
Aliases: []string{"XDCx.dbengine"},
Usage: "Database engine for XDCX (leveldb, mongodb)",
Value: "leveldb",
Category: flags.XdcxCategory,
}
XDCXDBNameFlag = &cli.StringFlag{ XDCXDBNameFlag = &cli.StringFlag{
Name: "XDCx-dbName", Name: "XDCx-dbName",
Aliases: []string{"XDCx.dbName"}, Aliases: []string{"XDCx.dbName"},
@ -864,19 +852,6 @@ var (
Value: "XDCdex", Value: "XDCdex",
Category: flags.XdcxCategory, Category: flags.XdcxCategory,
} }
XDCXDBConnectionUrlFlag = &cli.StringFlag{
Name: "XDCx-dbConnectionUrl",
Aliases: []string{"XDCx.dbConnectionUrl"},
Usage: "ConnectionUrl to database if dbEngine is mongodb. Host:port. If there are multiple instances, separated by comma. Eg: localhost:27017,localhost:27018",
Value: "localhost:27017",
Category: flags.XdcxCategory,
}
XDCXDBReplicaSetNameFlag = &cli.StringFlag{
Name: "XDCx-dbReplicaSetName",
Aliases: []string{"XDCx.dbReplicaSetName"},
Usage: "ReplicaSetName if Master-Slave is setup",
Category: flags.XdcxCategory,
}
) )
var ( var (
@ -1475,24 +1450,11 @@ func SetXDCXConfig(ctx *cli.Context, cfg *XDCx.Config, XDCDataDir string) {
// XDCx datadir: XDCDataDir/XDCx // XDCx datadir: XDCDataDir/XDCx
cfg.DataDir = filepath.Join(XDCDataDir, "XDCx") cfg.DataDir = filepath.Join(XDCDataDir, "XDCx")
log.Info("XDCX datadir", "path", cfg.DataDir) log.Info("XDCX datadir", "path", cfg.DataDir)
if ctx.IsSet(XDCXDBEngineFlag.Name) {
cfg.DBEngine = ctx.String(XDCXDBEngineFlag.Name)
} else {
cfg.DBEngine = XDCXDBEngineFlag.Value
}
if ctx.IsSet(XDCXDBNameFlag.Name) { if ctx.IsSet(XDCXDBNameFlag.Name) {
cfg.DBName = ctx.String(XDCXDBNameFlag.Name) cfg.DBName = ctx.String(XDCXDBNameFlag.Name)
} else { } else {
cfg.DBName = XDCXDBNameFlag.Value cfg.DBName = XDCXDBNameFlag.Value
} }
if ctx.IsSet(XDCXDBConnectionUrlFlag.Name) {
cfg.ConnectionUrl = ctx.String(XDCXDBConnectionUrlFlag.Name)
} else {
cfg.ConnectionUrl = XDCXDBConnectionUrlFlag.Value
}
if ctx.IsSet(XDCXDBReplicaSetNameFlag.Name) {
cfg.ReplicaSetName = ctx.String(XDCXDBReplicaSetNameFlag.Name)
}
} }
// SetEthConfig applies eth-related command line flags to the config. // SetEthConfig applies eth-related command line flags to the config.

7
consensus/XDPoS/utils/types.go
View file

@ -2,7 +2,6 @@ package utils
import ( import (
"math/big" "math/big"
"time"
"github.com/XinFinOrg/XDPoSChain/XDCx/tradingstate" "github.com/XinFinOrg/XDPoSChain/XDCx/tradingstate"
"github.com/XinFinOrg/XDPoSChain/XDCxlending/lendingstate" "github.com/XinFinOrg/XDPoSChain/XDCxlending/lendingstate"
@ -29,9 +28,6 @@ type TradingService interface {
GetTriegc() *prque.Prque[int64, common.Hash] GetTriegc() *prque.Prque[int64, common.Hash]
ApplyOrder(header *types.Header, coinbase common.Address, chain consensus.ChainContext, statedb *state.StateDB, XDCXstatedb *tradingstate.TradingStateDB, orderBook common.Hash, order *tradingstate.OrderItem) ([]map[string]string, []*tradingstate.OrderItem, error) ApplyOrder(header *types.Header, coinbase common.Address, chain consensus.ChainContext, statedb *state.StateDB, XDCXstatedb *tradingstate.TradingStateDB, orderBook common.Hash, order *tradingstate.OrderItem) ([]map[string]string, []*tradingstate.OrderItem, error)
UpdateMediumPriceBeforeEpoch(epochNumber uint64, tradingStateDB *tradingstate.TradingStateDB, statedb *state.StateDB) error UpdateMediumPriceBeforeEpoch(epochNumber uint64, tradingStateDB *tradingstate.TradingStateDB, statedb *state.StateDB) error
IsSDKNode() bool
SyncDataToSDKNode(takerOrder *tradingstate.OrderItem, txHash common.Hash, txMatchTime time.Time, statedb *state.StateDB, trades []map[string]string, rejectedOrders []*tradingstate.OrderItem, dirtyOrderCount *uint64) error
RollbackReorgTxMatch(txhash common.Hash) error
GetTokenDecimal(chain consensus.ChainContext, statedb *state.StateDB, tokenAddr common.Address) (*big.Int, error) GetTokenDecimal(chain consensus.ChainContext, statedb *state.StateDB, tokenAddr common.Address) (*big.Int, error)
} }
@ -45,9 +41,6 @@ type LendingService interface {
GetCollateralPrices(header *types.Header, chain consensus.ChainContext, statedb *state.StateDB, tradingStateDb *tradingstate.TradingStateDB, collateralToken common.Address, lendingToken common.Address) (*big.Int, *big.Int, error) GetCollateralPrices(header *types.Header, chain consensus.ChainContext, statedb *state.StateDB, tradingStateDb *tradingstate.TradingStateDB, collateralToken common.Address, lendingToken common.Address) (*big.Int, *big.Int, error)
GetMediumTradePriceBeforeEpoch(chain consensus.ChainContext, statedb *state.StateDB, tradingStateDb *tradingstate.TradingStateDB, baseToken common.Address, quoteToken common.Address) (*big.Int, error) GetMediumTradePriceBeforeEpoch(chain consensus.ChainContext, statedb *state.StateDB, tradingStateDb *tradingstate.TradingStateDB, baseToken common.Address, quoteToken common.Address) (*big.Int, error)
ProcessLiquidationData(header *types.Header, chain consensus.ChainContext, statedb *state.StateDB, tradingState *tradingstate.TradingStateDB, lendingState *lendingstate.LendingStateDB) (updatedTrades map[common.Hash]*lendingstate.LendingTrade, liquidatedTrades, autoRepayTrades, autoTopUpTrades, autoRecallTrades []*lendingstate.LendingTrade, err error) ProcessLiquidationData(header *types.Header, chain consensus.ChainContext, statedb *state.StateDB, tradingState *tradingstate.TradingStateDB, lendingState *lendingstate.LendingStateDB) (updatedTrades map[common.Hash]*lendingstate.LendingTrade, liquidatedTrades, autoRepayTrades, autoTopUpTrades, autoRecallTrades []*lendingstate.LendingTrade, err error)
SyncDataToSDKNode(chain consensus.ChainContext, state *state.StateDB, block *types.Block, takerOrderInTx *lendingstate.LendingItem, txHash common.Hash, txMatchTime time.Time, trades []*lendingstate.LendingTrade, rejectedOrders []*lendingstate.LendingItem, dirtyOrderCount *uint64) error
UpdateLiquidatedTrade(blockTime uint64, result lendingstate.FinalizedResult, trades map[common.Hash]*lendingstate.LendingTrade) error
RollbackLendingData(txhash common.Hash) error
} }
type PublicApiSnapshot struct { type PublicApiSnapshot struct {

6
core/block_validator.go
View file

@ -138,9 +138,6 @@ func (v *BlockValidator) ValidateTradingOrder(statedb *state.StateDB, XDCxStated
Rejects: newRejectedOrders, Rejects: newRejectedOrders,
} }
} }
if XDCXService.IsSDKNode() {
v.bc.AddMatchingResult(txMatchBatch.TxHash, tradingResult)
}
return nil return nil
} }
@ -173,9 +170,6 @@ func (v *BlockValidator) ValidateLendingOrder(statedb *state.StateDB, lendingSta
Rejects: newRejectedOrders, Rejects: newRejectedOrders,
} }
} }
if XDCXService.IsSDKNode() {
v.bc.AddLendingResult(batch.TxHash, lendingResult)
}
return nil return nil
} }

157
core/blockchain.go
View file

@ -1824,10 +1824,6 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, []
// Only count canonical blocks for GC processing time // Only count canonical blocks for GC processing time
bc.gcproc += res.procTime bc.gcproc += res.procTime
bc.UpdateBlocksHashCache(block) bc.UpdateBlocksHashCache(block)
if bc.chainConfig.IsTIPXDCX(block.Number()) && bc.chainConfig.XDPoS != nil && block.NumberU64() > bc.chainConfig.XDPoS.Epoch {
bc.logExchangeData(block)
bc.logLendingData(block)
}
case SideStatTy: case SideStatTy:
log.Debug("Inserted forked block from downloader", "number", block.Number(), "hash", block.Hash(), "diff", block.Difficulty(), "elapsed", log.Debug("Inserted forked block from downloader", "number", block.Number(), "hash", block.Hash(), "diff", block.Difficulty(), "elapsed",
common.PrettyDuration(time.Since(start)), "txs", len(block.Transactions()), "gas", block.GasUsed(), "uncles", len(block.Uncles())) common.PrettyDuration(time.Since(start)), "txs", len(block.Transactions()), "gas", block.GasUsed(), "uncles", len(block.Uncles()))
@ -2282,10 +2278,6 @@ func (bc *BlockChain) insertBlock(block *types.Block) ([]interface{}, []*types.L
// Only count canonical blocks for GC processing time // Only count canonical blocks for GC processing time
bc.gcproc += result.proctime bc.gcproc += result.proctime
bc.UpdateBlocksHashCache(block) bc.UpdateBlocksHashCache(block)
if bc.chainConfig.IsTIPXDCXReceiver(block.Number()) && bc.chainConfig.XDPoS != nil && block.NumberU64() > bc.chainConfig.XDPoS.Epoch {
bc.logExchangeData(block)
bc.logLendingData(block)
}
case SideStatTy: case SideStatTy:
log.Debug("Inserted forked block from fetcher", "number", block.Number(), "hash", block.Hash(), "diff", block.Difficulty(), "elapsed", log.Debug("Inserted forked block from fetcher", "number", block.Number(), "hash", block.Hash(), "diff", block.Difficulty(), "elapsed",
common.PrettyDuration(time.Since(block.ReceivedAt)), "txs", len(block.Transactions()), "gas", block.GasUsed(), "uncles", len(block.Uncles())) common.PrettyDuration(time.Since(block.ReceivedAt)), "txs", len(block.Transactions()), "gas", block.GasUsed(), "uncles", len(block.Uncles()))
@ -2830,146 +2822,6 @@ func (bc *BlockChain) UpdateM1() error {
return nil return nil
} }
func (bc *BlockChain) logExchangeData(block *types.Block) {
engine, ok := bc.Engine().(*XDPoS.XDPoS)
if !ok || engine == nil {
return
}
XDCXService := engine.GetXDCXService()
if XDCXService == nil || !XDCXService.IsSDKNode() {
return
}
txMatchBatchData, err := ExtractTradingTransactions(block.Transactions())
if err != nil {
log.Error("failed to extract matching transaction", "err", err)
return
}
if len(txMatchBatchData) == 0 {
return
}
currentState, err := bc.State()
if err != nil {
log.Crit("logExchangeData: failed to get current state", "err", err)
return
}
start := time.Now()
defer func() {
//The deferred call's arguments are evaluated immediately, but the function call is not executed until the surrounding function returns
// That's why we should put this log statement in an anonymous function
log.Debug("logExchangeData takes", "time", common.PrettyDuration(time.Since(start)), "blockNumber", block.NumberU64())
}()
for _, txMatchBatch := range txMatchBatchData {
dirtyOrderCount := uint64(0)
for _, txMatch := range txMatchBatch.Data {
var (
takerOrderInTx *tradingstate.OrderItem
trades []map[string]string
rejectedOrders []*tradingstate.OrderItem
)
if takerOrderInTx, err = txMatch.DecodeOrder(); err != nil {
log.Error("SDK node decode takerOrderInTx failed", "txDataMatch", txMatch)
return
}
cacheKey := crypto.Keccak256Hash(txMatchBatch.TxHash.Bytes(), tradingstate.GetMatchingResultCacheKey(takerOrderInTx).Bytes())
// getTrades from cache
resultTrades, ok := bc.resultTrade.Get(cacheKey)
if ok && resultTrades != nil {
trades = resultTrades.([]map[string]string)
}
// getRejectedOrder from cache
rejected, ok := bc.rejectedOrders.Get(cacheKey)
if ok && rejected != nil {
rejectedOrders = rejected.([]*tradingstate.OrderItem)
}
txMatchTime := time.Unix(int64(block.Header().Time), 0).UTC()
if err := XDCXService.SyncDataToSDKNode(takerOrderInTx, txMatchBatch.TxHash, txMatchTime, currentState, trades, rejectedOrders, &dirtyOrderCount); err != nil {
log.Error("failed to SyncDataToSDKNode ", "blockNumber", block.Number(), "err", err)
return
}
}
}
}
func (bc *BlockChain) logLendingData(block *types.Block) {
engine, ok := bc.Engine().(*XDPoS.XDPoS)
if !ok || engine == nil {
return
}
XDCXService := engine.GetXDCXService()
if XDCXService == nil || !XDCXService.IsSDKNode() {
return
}
lendingService := engine.GetLendingService()
if lendingService == nil {
return
}
batches, err := ExtractLendingTransactions(block.Transactions())
if err != nil {
log.Error("failed to extract lending transaction", "err", err)
return
}
start := time.Now()
defer func() {
//The deferred call's arguments are evaluated immediately, but the function call is not executed until the surrounding function returns
// That's why we should put this log statement in an anonymous function
log.Debug("logLendingData takes", "time", common.PrettyDuration(time.Since(start)), "blockNumber", block.NumberU64())
}()
for _, batch := range batches {
dirtyOrderCount := uint64(0)
for _, item := range batch.Data {
var (
trades []*lendingstate.LendingTrade
rejectedOrders []*lendingstate.LendingItem
)
// getTrades from cache
resultLendingTrades, ok := bc.resultLendingTrade.Get(crypto.Keccak256Hash(batch.TxHash.Bytes(), lendingstate.GetLendingCacheKey(item).Bytes()))
if ok && resultLendingTrades != nil {
trades = resultLendingTrades.([]*lendingstate.LendingTrade)
}
// getRejectedOrder from cache
rejected, ok := bc.rejectedLendingItem.Get(crypto.Keccak256Hash(batch.TxHash.Bytes(), lendingstate.GetLendingCacheKey(item).Bytes()))
if ok && rejected != nil {
rejectedOrders = rejected.([]*lendingstate.LendingItem)
}
txMatchTime := time.Unix(int64(block.Header().Time), 0).UTC()
statedb, _ := bc.State()
if err := lendingService.SyncDataToSDKNode(bc, statedb.Copy(), block, item, batch.TxHash, txMatchTime, trades, rejectedOrders, &dirtyOrderCount); err != nil {
log.Error("lending: failed to SyncDataToSDKNode ", "blockNumber", block.Number(), "err", err)
return
}
}
}
// update finalizedTrades
if block.Number().Uint64()%bc.chainConfig.XDPoS.Epoch == common.LiquidateLendingTradeBlock {
finalizedTx, err := ExtractLendingFinalizedTradeTransactions(block.Transactions())
if err != nil {
log.Error("failed to extract finalizedTrades transaction", "err", err)
return
}
finalizedTrades := map[common.Hash]*lendingstate.LendingTrade{}
finalizedData, ok := bc.finalizedTrade.Get(finalizedTx.TxHash)
if ok && finalizedData != nil {
finalizedTrades = finalizedData.(map[common.Hash]*lendingstate.LendingTrade)
}
if len(finalizedTrades) > 0 {
if err := lendingService.UpdateLiquidatedTrade(block.Time(), finalizedTx, finalizedTrades); err != nil {
log.Error("lending: failed to UpdateLiquidatedTrade ", "blockNumber", block.Number(), "err", err)
return
}
}
}
}
func (bc *BlockChain) AddMatchingResult(txHash common.Hash, matchingResults map[common.Hash]tradingstate.MatchingResult) { func (bc *BlockChain) AddMatchingResult(txHash common.Hash, matchingResults map[common.Hash]tradingstate.MatchingResult) {
for hash, result := range matchingResults { for hash, result := range matchingResults {
cacheKey := crypto.Keccak256Hash(txHash.Bytes(), hash.Bytes()) cacheKey := crypto.Keccak256Hash(txHash.Bytes(), hash.Bytes())
@ -3078,17 +2930,10 @@ func (bc *BlockChain) processTradingAndLendingStates(isValidBlockNumber bool, bl
} }
// liquidate / finalize open lendingTrades // liquidate / finalize open lendingTrades
if block.Number().Uint64()%bc.chainConfig.XDPoS.Epoch == common.LiquidateLendingTradeBlock { if block.Number().Uint64()%bc.chainConfig.XDPoS.Epoch == common.LiquidateLendingTradeBlock {
finalizedTrades, _, _, _, _, err := lendingService.ProcessLiquidationData(block.Header(), bc, statedb, tradingState, lendingState) _, _, _, _, _, err := lendingService.ProcessLiquidationData(block.Header(), bc, statedb, tradingState, lendingState)
if err != nil { if err != nil {
return tradingState, lendingState, fmt.Errorf("failed to ProcessLiquidationData. Err: %v", err) return tradingState, lendingState, fmt.Errorf("failed to ProcessLiquidationData. Err: %v", err)
} }
if tradingService.IsSDKNode() {
finalizedTx, err := ExtractLendingFinalizedTradeTransactions(block.Transactions())
if err != nil {
return tradingState, lendingState, err
}
bc.AddFinalizedTrades(finalizedTx.TxHash, finalizedTrades)
}
} }
} }

41
miner/worker.go
View file

@ -752,15 +752,11 @@ func (w *worker) commitNewWork() {
var ( var (
txs *types.TransactionsByPriceAndNonce txs *types.TransactionsByPriceAndNonce
specialTxs types.Transactions specialTxs types.Transactions
tradingTransaction *types.Transaction
lendingTransaction *types.Transaction
tradingTxMatches []tradingstate.TxDataMatch tradingTxMatches []tradingstate.TxDataMatch
tradingMatchingResults map[common.Hash]tradingstate.MatchingResult
lendingMatchingResults map[common.Hash]lendingstate.MatchingResult lendingMatchingResults map[common.Hash]lendingstate.MatchingResult
lendingInput []*lendingstate.LendingItem lendingInput []*lendingstate.LendingItem
updatedTrades map[common.Hash]*lendingstate.LendingTrade updatedTrades map[common.Hash]*lendingstate.LendingTrade
liquidatedTrades, autoRepayTrades, autoTopUpTrades, autoRecallTrades []*lendingstate.LendingTrade liquidatedTrades, autoRepayTrades, autoTopUpTrades, autoRecallTrades []*lendingstate.LendingTrade
lendingFinalizedTradeTransaction *types.Transaction
) )
feeCapacity := state.GetTRC21FeeCapacityFromStateWithCache(parent.Root(), work.state) feeCapacity := state.GetTRC21FeeCapacityFromStateWithCache(parent.Root(), work.state)
if w.config.XDPoS != nil { if w.config.XDPoS != nil {
@ -800,7 +796,7 @@ func (w *worker) commitNewWork() {
log.Debug("Start processing order pending") log.Debug("Start processing order pending")
tradingOrderPending, _ := w.eth.OrderPool().Pending() tradingOrderPending, _ := w.eth.OrderPool().Pending()
log.Debug("Start processing order pending", "len", len(tradingOrderPending)) log.Debug("Start processing order pending", "len", len(tradingOrderPending))
tradingTxMatches, tradingMatchingResults = XDCX.ProcessOrderPending(header, w.coinbase, w.chain, tradingOrderPending, work.state, work.tradingState) tradingTxMatches, _ = XDCX.ProcessOrderPending(header, w.coinbase, w.chain, tradingOrderPending, work.state, work.tradingState)
log.Debug("trading transaction matches found", "tradingTxMatches", len(tradingTxMatches)) log.Debug("trading transaction matches found", "tradingTxMatches", len(tradingTxMatches))
lendingOrderPending, _ := w.eth.LendingPool().Pending() lendingOrderPending, _ := w.eth.LendingPool().Pending()
@ -832,15 +828,10 @@ func (w *worker) commitNewWork() {
if err != nil { if err != nil {
log.Error("Fail to create tx matches", "error", err) log.Error("Fail to create tx matches", "error", err)
return return
} else { }
tradingTransaction = txM // force adding trading, lending transaction to this block
if XDCX.IsSDKNode() { if txM != nil {
w.chain.AddMatchingResult(tradingTransaction.Hash(), tradingMatchingResults) specialTxs = append(specialTxs, txM)
}
// force adding trading, lending transaction to this block
if tradingTransaction != nil {
specialTxs = append(specialTxs, tradingTransaction)
}
} }
} }
@ -862,14 +853,9 @@ func (w *worker) commitNewWork() {
if err != nil { if err != nil {
log.Error("Fail to create lending tx", "error", err) log.Error("Fail to create lending tx", "error", err)
return return
} else { }
lendingTransaction = signedLendingTx if signedLendingTx != nil {
if XDCX.IsSDKNode() { specialTxs = append(specialTxs, signedLendingTx)
w.chain.AddLendingResult(lendingTransaction.Hash(), lendingMatchingResults)
}
if lendingTransaction != nil {
specialTxs = append(specialTxs, lendingTransaction)
}
} }
} }
@ -886,14 +872,9 @@ func (w *worker) commitNewWork() {
if err != nil { if err != nil {
log.Error("Fail to create lending tx", "error", err) log.Error("Fail to create lending tx", "error", err)
return return
} else { }
lendingFinalizedTradeTransaction = signedFinalizedTx if signedFinalizedTx != nil {
if XDCX.IsSDKNode() { specialTxs = append(specialTxs, signedFinalizedTx)
w.chain.AddFinalizedTrades(lendingFinalizedTradeTransaction.Hash(), updatedTrades)
}
if lendingFinalizedTradeTransaction != nil {
specialTxs = append(specialTxs, lendingFinalizedTradeTransaction)
}
} }
} }
} }