From b76eb6906c6d3ddec1c11d7f65eb3f9d0a61aa1b Mon Sep 17 00:00:00 2001 From: Callum Waters Date: Fri, 18 Nov 2022 16:11:35 +0100 Subject: [PATCH] add hash and state meta tracking --- abci/example/orderbook/app.go | 421 ++++++++++++++++++----------- abci/example/orderbook/app_test.go | 3 +- abci/example/orderbook/market.go | 2 +- abci/example/orderbook/types.go | 8 +- abci/example/orderbook/wire.proto | 1 + 5 files changed, 271 insertions(+), 164 deletions(-) diff --git a/abci/example/orderbook/app.go b/abci/example/orderbook/app.go index cd71d89d8..c2664a65c 100644 --- a/abci/example/orderbook/app.go +++ b/abci/example/orderbook/app.go @@ -1,14 +1,15 @@ package orderbook import ( + "bytes" "encoding/binary" "github.com/cosmos/gogoproto/proto" dbm "github.com/tendermint/tm-db" "github.com/tendermint/tendermint/abci/types" - "github.com/tendermint/tendermint/crypto/tmhash" "github.com/tendermint/tendermint/crypto/ed25519" + "github.com/tendermint/tendermint/crypto/tmhash" ) var _ types.Application = (*StateMachine)(nil) @@ -31,25 +32,32 @@ const ( ) var ( + stateKey = []byte("state") accountKey = []byte("account") - pairKey = []byte("pair") + pairKey = []byte("pair") ) +// StateMachine is the main struct that encompasses the logic of the orderbook type StateMachine struct { + // inherit all the abci methods so we don't have to implement everything + types.BaseApplication + // persisted state which is a key value store containing: // accountID -> account // pairID -> pair db dbm.DB - // ephemeral state (not used for the app hash) but for - // convienience - accounts map[uint64]*Account - pairs map[string]struct{} // lookup pairs + // in-memory state + lastHeight int64 // the last height that was persisted + lastHash []byte // the last hash that was persisted + // list of accounts (this is used for the app hash) + accounts []*Account + pairs map[string]*Pair // lookup pairs commodities map[string]struct{} // lookup commodities publicKeys map[string]struct{} // lookup existence of an account // a list of transactions that have been modified by the most recent block // and will need to result in an update to the db - touchedAccounts map[uint64]struct{} + touchedAccounts map[uint64]struct{} // new pairs added in this block which will needed to be added to the // db on "Commit" newPairs []*Pair @@ -58,31 +66,94 @@ type StateMachine struct { // this takes ask and bid transactions from `CheckTx` // and matches them as a "MatchedOrder" which is // then proposed in a block - // + // // it's important to note that there is no garbage collection // here. Bids and asks, potentially even invalid, will // continue to stay here until matched markets map[string]*Market // i.e. ATOM/USDC } -func New(db dbm.DB) *StateMachine { - // execute a database call that fetches the data for accounts - StateMachine := StateMachine{ - accounts: make(map[uint64]*Account), - pairs: make(map[string]struct{}), - commodities: make(map[string]struct{}), - publicKeys: make(map[string]struct{}), - markets: make(map[string]*Market), - db: db, +// New creates a StateMachine from a given database. If the database is +// empty a fresh instance is created else the accounts, pairs and +// state are loaded into memory. +func New(db dbm.DB) (*StateMachine, error) { + // iterate over all the account keys + iter, err := db.Iterator(nil, nil) + if err != nil { + return nil, err } - return &StateMachine + defer iter.Close() + + var ( + accounts = make([]*Account, 0) + publicKeys = make(map[string]struct{}) + commodities = make(map[string]struct{}) + pairs = make(map[string]*Pair) + lastHeight uint64 + lastHash []byte + ) + + for ; iter.Valid(); iter.Next() { + if bytes.HasPrefix(iter.Key(), pairKey) { + var pair *Pair + if err := proto.Unmarshal(iter.Value(), pair); err != nil { + return nil, err + } + pairs[pair.String()] = pair + commodities[pair.BuyersDenomination] = struct{}{} + + } + + if bytes.HasPrefix(iter.Key(), accountKey) { + var acc *Account + if err := proto.Unmarshal(iter.Value(), acc); err != nil { + return nil, err + } + + accounts = append(accounts, acc) + publicKeys[string(acc.PublicKey)] = struct{}{} + } + + if bytes.HasPrefix(iter.Key(), stateKey) { + state := iter.Value() + lastHeight = binary.BigEndian.Uint64(state[:4]) + lastHash = state[4:] + } + + var acc *Account + if err := proto.Unmarshal(iter.Value(), acc); err != nil { + return nil, err + } + + accounts = append(accounts, acc) + } + + return &StateMachine{ + accounts: accounts, + pairs: pairs, + commodities: commodities, + publicKeys: publicKeys, + markets: make(map[string]*Market), + lastHeight: int64(lastHeight), + lastHash: lastHash, + db: db, + }, nil } +// Info is used by Tendermint to understand the state of the application. +// This is useful for replay and syncing modes. func (sm *StateMachine) Info(req types.RequestInfo) types.ResponseInfo { - return types.ResponseInfo{} + return types.ResponseInfo{ + AppVersion: Version, + LastBlockHeight: sm.lastHeight, + LastBlockAppHash: sm.lastHash, + } } -// CheckTx indicates which transactions +// CheckTx indicates which transactions should be accepted in the mempool. It is +// not a perfect validity check because we're unsure of the state that the transaction +// will be executed against. We should treat this as a gatekeeper to the mempool. +// Apart from adding transactions to the app-side mempool, this check is stateless. func (sm *StateMachine) CheckTx(req types.RequestCheckTx) types.ResponseCheckTx { var msg = new(Msg) @@ -114,6 +185,7 @@ func (sm *StateMachine) CheckTx(req types.RequestCheckTx) types.ResponseCheckTx return types.ResponseCheckTx{Code: StatusOK} } +// ValidateTx validates the transactions against state. func (sm *StateMachine) ValidateTx(msg *Msg) uint32 { if err := msg.ValidateBasic(); err != nil { return StatusErrValidateBasic @@ -144,7 +216,7 @@ func (sm *StateMachine) ValidateTx(msg *Msg) uint32 { for _, commodity := range m.MsgCreateAccount.Commodities { if _, exists := sm.commodities[commodity.Denom]; !exists { return StatusErrNoCommodity - } + } } case *Msg_MsgTradeSet: @@ -167,138 +239,9 @@ func (sm *StateMachine) ValidateTx(msg *Msg) uint32 { return StatusOK } -func (sm *StateMachine) Commit() types.ResponseCommit { - batch := sm.db.NewBatch() - - // write to accounts that were modified by the last block - for accountID := range sm.touchedAccounts { - value, err := proto.Marshal(sm.accounts[accountID]) - if err != nil { - panic(err) - } - var key []byte - copy(key, accountKey) - binary.BigEndian.PutUint64(key, accountID) - - batch.Set(key, value) - } - - // write the new pairs that were added by the last block - pairID := len(sm.pairs) - len(sm.newPairs) - for id, pair := range sm.newPairs { - value, err := proto.Marshal(pair) - if err != nil { - panic(err) - } - var key []byte - copy(key, pairKey) - binary.BigEndian.PutUint64(key, uint64(pairID + id)) - batch.Set(key, value) - } - - batch.WriteSync() - - return types.ResponseCommit{Data: sm.hash()} -} - -func (sm *StateMachine) hash() []byte { - return tmhash.Sum([]byte("hash")) -} - -func (sm *StateMachine) Query(req types.RequestQuery) types.ResponseQuery { - return types.ResponseQuery{Code: 0} -} - -func (sm *StateMachine) InitChain(req types.RequestInitChain) types.ResponseInitChain { - return types.ResponseInitChain{} -} - -func (sm *StateMachine) BeginBlock(req types.RequestBeginBlock) types.ResponseBeginBlock { - // reset the new pairs - sm.newPairs = make([]*Pair, 0) - return types.ResponseBeginBlock{} -} - -func (sm *StateMachine) DeliverTx(req types.RequestDeliverTx) types.ResponseDeliverTx { - var msg = new(Msg) - - err := proto.Unmarshal(req.Tx, msg) - if err != nil { - return types.ResponseDeliverTx{Code: StatusErrDecoding, Log: err.Error()} // decoding error - } - - if status := sm.ValidateTx(msg); status != StatusOK { - return types.ResponseDeliverTx{Code: status} - } - - switch m := msg.Sum.(type) { - case *Msg_MsgRegisterPair: - sm.markets[m.MsgRegisterPair.Pair.String()] = NewMarket(m.MsgRegisterPair.Pair) - sm.pairs[m.MsgRegisterPair.Pair.String()] = struct{}{} - sm.commodities[m.MsgRegisterPair.Pair.BuyersDenomination] = struct{}{} - sm.commodities[m.MsgRegisterPair.Pair.SellersDenomination] = struct{}{} - sm.newPairs = append(sm.newPairs, m.MsgRegisterPair.Pair) - - case *Msg_MsgCreateAccount: - nextAccountID := uint64(len(sm.accounts)) - sm.accounts[nextAccountID] = &Account{ - Index: nextAccountID, - PublicKey: m.MsgCreateAccount.PublicKey, - Commodities: m.MsgCreateAccount.Commodities, - } - sm.touchedAccounts[nextAccountID] = struct{}{} - sm.publicKeys[string(m.MsgCreateAccount.PublicKey)] = struct{}{} - - case *Msg_MsgTradeSet: - pair := m.MsgTradeSet.TradeSet.Pair - for _, order := range m.MsgTradeSet.TradeSet.MatchedOrders { - buyer := sm.accounts[order.OrderBid.OwnerId] - seller := sm.accounts[order.OrderAsk.OwnerId] - - // the buyer gets quantity of the asset that the seller was selling - buyer.AddCommodity(NewCommodity(pair.SellersDenomination, order.OrderAsk.Quantity)) - // the buyer gives up quantity * ask price of the buyers denomination - buyer.SubtractCommodity(NewCommodity(pair.BuyersDenomination, order.OrderAsk.Quantity * order.OrderAsk.AskPrice)) - - // the seller gets quantity * ask price of the asset that the buyer was paying with - seller.AddCommodity(NewCommodity(pair.BuyersDenomination, order.OrderAsk.Quantity * order.OrderAsk.AskPrice)) - // the seller gives up quantity of the commodity they were selling - seller.SubtractCommodity(NewCommodity(pair.SellersDenomination, order.OrderAsk.Quantity)) - - // mark that these account have been touched - sm.touchedAccounts[order.OrderBid.OwnerId] = struct{}{} - sm.touchedAccounts[order.OrderAsk.OwnerId] = struct{}{} - } - - default: - return types.ResponseDeliverTx{Code: StatusErrUnknownMessage} - } - - return types.ResponseDeliverTx{Code: 0} -} - -// EndBlock is used to update consensus params and the validator set. For the orderbook, -// we keep both the same for thw -func (sm *StateMachine) EndBlock(req types.RequestEndBlock) types.ResponseEndBlock { - return types.ResponseEndBlock{} -} - -func (sm *StateMachine) ListSnapshots(req types.RequestListSnapshots) types.ResponseListSnapshots { - return types.ResponseListSnapshots{} -} - -func (sm *StateMachine) OfferSnapshot(req types.RequestOfferSnapshot) types.ResponseOfferSnapshot { - return types.ResponseOfferSnapshot{} -} - -func (sm *StateMachine) LoadSnapshotChunk(req types.RequestLoadSnapshotChunk) types.ResponseLoadSnapshotChunk { - return types.ResponseLoadSnapshotChunk{} -} - -func (sm *StateMachine) ApplySnapshotChunk(req types.RequestApplySnapshotChunk) types.ResponseApplySnapshotChunk { - return types.ResponseApplySnapshotChunk{} -} - +// PrepareProposal is called whenever the validator is the proposer for that round. First, it adds the non order +// transactions provided by tendermint. The orderbook then loops through each market and tries to match as many +// transactions as possible. For each new transaction it checks that the max bytes has not been exceeded. func (sm *StateMachine) PrepareProposal(req types.RequestPrepareProposal) types.ResponsePrepareProposal { // declare transaction with the size of 0 txs := make([][]byte, 0) @@ -322,7 +265,7 @@ func (sm *StateMachine) PrepareProposal(req types.RequestPrepareProposal) types. // make sure we're proposing valid transactions if status := sm.ValidateTx(msg); status != StatusOK { continue - } + } if len(txs)+len(tx) > int(req.MaxTxBytes) { return types.ResponsePrepareProposal{Txs: txs} @@ -361,6 +304,9 @@ func (sm *StateMachine) PrepareProposal(req types.RequestPrepareProposal) types. } // Process Proposal either rejects or accepts transactions +// +// It uses the same validity function for prepare proposal. This ensures the coherence property +// is adhered to i.e. all honest validators must accept a proposal by an honest proposer func (sm *StateMachine) ProcessProposal(req types.RequestProcessProposal) types.ResponseProcessProposal { for _, tx := range req.Txs { var msg = new(Msg) @@ -377,6 +323,150 @@ func (sm *StateMachine) ProcessProposal(req types.RequestProcessProposal) types. return acceptProposal() } +func (sm *StateMachine) BeginBlock(req types.RequestBeginBlock) types.ResponseBeginBlock { + // reset the new pairs + sm.newPairs = make([]*Pair, 0) + return types.ResponseBeginBlock{} +} + +// DeliverTx is called for each tx in a block once it has been finalized. This is where the +// execution code lives. Most importantly it's where we update the user accounts following +// a successful order. +func (sm *StateMachine) DeliverTx(req types.RequestDeliverTx) types.ResponseDeliverTx { + var msg = new(Msg) + + err := proto.Unmarshal(req.Tx, msg) + if err != nil { + return types.ResponseDeliverTx{Code: StatusErrDecoding, Log: err.Error()} // decoding error + } + + if status := sm.ValidateTx(msg); status != StatusOK { + return types.ResponseDeliverTx{Code: status} + } + + switch m := msg.Sum.(type) { + case *Msg_MsgRegisterPair: + sm.markets[m.MsgRegisterPair.Pair.String()] = NewMarket(m.MsgRegisterPair.Pair) + sm.pairs[m.MsgRegisterPair.Pair.String()] = m.MsgRegisterPair.Pair + sm.commodities[m.MsgRegisterPair.Pair.BuyersDenomination] = struct{}{} + sm.commodities[m.MsgRegisterPair.Pair.SellersDenomination] = struct{}{} + sm.newPairs = append(sm.newPairs, m.MsgRegisterPair.Pair) + + case *Msg_MsgCreateAccount: + nextAccountID := uint64(len(sm.accounts)) + sm.accounts[nextAccountID] = &Account{ + Index: nextAccountID, + PublicKey: m.MsgCreateAccount.PublicKey, + Commodities: m.MsgCreateAccount.Commodities, + } + sm.touchedAccounts[nextAccountID] = struct{}{} + sm.publicKeys[string(m.MsgCreateAccount.PublicKey)] = struct{}{} + + case *Msg_MsgTradeSet: + pair := m.MsgTradeSet.TradeSet.Pair + for _, order := range m.MsgTradeSet.TradeSet.MatchedOrders { + buyer := sm.accounts[order.OrderBid.OwnerId] + seller := sm.accounts[order.OrderAsk.OwnerId] + + // the buyer gets quantity of the asset that the seller was selling + buyer.AddCommodity(NewCommodity(pair.SellersDenomination, order.OrderAsk.Quantity)) + // the buyer gives up quantity * ask price of the buyers denomination + buyer.SubtractCommodity(NewCommodity(pair.BuyersDenomination, order.OrderAsk.Quantity*order.OrderAsk.AskPrice)) + + // the seller gets quantity * ask price of the asset that the buyer was paying with + seller.AddCommodity(NewCommodity(pair.BuyersDenomination, order.OrderAsk.Quantity*order.OrderAsk.AskPrice)) + // the seller gives up quantity of the commodity they were selling + seller.SubtractCommodity(NewCommodity(pair.SellersDenomination, order.OrderAsk.Quantity)) + + // mark that these account have been touched + sm.touchedAccounts[order.OrderBid.OwnerId] = struct{}{} + sm.touchedAccounts[order.OrderAsk.OwnerId] = struct{}{} + } + + default: + return types.ResponseDeliverTx{Code: StatusErrUnknownMessage} + } + + return types.ResponseDeliverTx{Code: 0} +} + +// EndBlock is used to update consensus params and the validator set. For the orderbook, +// we keep both the same for thw +func (sm *StateMachine) EndBlock(req types.RequestEndBlock) types.ResponseEndBlock { + return types.ResponseEndBlock{} +} + +// Commit is called to tell the app it is safe to persist state to disk. +// We now take the in-memory representation and update the parts that have +// changed on to disk. +func (sm *StateMachine) Commit() types.ResponseCommit { + batch := sm.db.NewBatch() + + // write to accounts that were modified by the last block + for accountID := range sm.touchedAccounts { + value, err := proto.Marshal(sm.accounts[accountID]) + if err != nil { + panic(err) + } + var key []byte + copy(key, accountKey) + binary.BigEndian.PutUint64(key, accountID) + + if err := batch.Set(key, value); err != nil { + panic(err) + } + } + + // write the new pairs that were added by the last block + pairID := len(sm.pairs) - len(sm.newPairs) + for id, pair := range sm.newPairs { + value, err := proto.Marshal(pair) + if err != nil { + panic(err) + } + var key []byte + copy(key, pairKey) + binary.BigEndian.PutUint64(key, uint64(pairID+id)) + if err := batch.Set(key, value); err != nil { + panic(err) + } + } + + hash := sm.hash() + err := sm.updateState(batch, sm.lastHeight+1, hash) + if err != nil { + panic(err) + } + err = batch.WriteSync() + if err != nil { + panic(err) + } + + return types.ResponseCommit{Data: hash} +} + +// hash is just the the sha256 of the byte representation of all accounts. +// remember that this needs to be deterministic for all state machines +func (sm *StateMachine) hash() []byte { + digest := bytes.NewBuffer(nil) + for _, account := range sm.accounts { + bz, err := proto.Marshal(account) + if err != nil { + panic(err) + } + digest.Write(bz) + } + return tmhash.Sum(digest.Bytes()) +} + +func (sm *StateMachine) updateState(batch dbm.Batch, height int64, hash []byte) error { + sm.lastHash = hash + sm.lastHeight = height + var heightBytes []byte + binary.BigEndian.PutUint64(heightBytes, uint64(height)) + return batch.Set(stateKey, append(heightBytes, hash...)) +} + func (sm *StateMachine) validateTradeSetAgainstState(tradeSet *TradeSet) *TradeSet { output := &TradeSet{Pair: tradeSet.Pair} @@ -392,13 +482,28 @@ func (sm *StateMachine) validateTradeSetAgainstState(tradeSet *TradeSet) *TradeS return output } +// isMatchedOrderValid is a check against current state to ensure that the order +// is valid and can execute. +// +// This method is also called when preparing a proposal since `CheckTx` doesn't have +// strict validity guarantees and there could be invalid transactions within the mempool +// +// Note: if one of the two orders are invalid we discard both. In the future we could +// improve this by adding back the part of the order that might still be valid. func (sm *StateMachine) isMatchedOrderValid(order *MatchedOrder, pair *Pair) bool { - bidOwner, exists := sm.accounts[order.OrderBid.OwnerId] - if !exists { + if int(order.OrderBid.OwnerId) >= len(sm.accounts) { return false } - askOwner, exists := sm.accounts[order.OrderAsk.OwnerId] - if !exists { + bidOwner := sm.accounts[order.OrderBid.OwnerId] + if bidOwner == nil { + return false + } + + if int(order.OrderAsk.OwnerId) >= len(sm.accounts) { + return false + } + askOwner := sm.accounts[order.OrderAsk.OwnerId] + if askOwner == nil { return false } diff --git a/abci/example/orderbook/app_test.go b/abci/example/orderbook/app_test.go index e813cde5f..03562c944 100644 --- a/abci/example/orderbook/app_test.go +++ b/abci/example/orderbook/app_test.go @@ -12,7 +12,8 @@ import ( ) func TestCheckTx(t *testing.T) { - app := orderbook.New(dbm.NewMemDB()) + app, err := orderbook.New(dbm.NewMemDB()) + require.NoError(t, err) testCases := []struct { name string diff --git a/abci/example/orderbook/market.go b/abci/example/orderbook/market.go index 348be0ec4..a5d17fc4f 100644 --- a/abci/example/orderbook/market.go +++ b/abci/example/orderbook/market.go @@ -5,7 +5,7 @@ import ( ) type Market struct { - pair *Pair // i.e. EUR/USD (a market is bidirectional) + pair *Pair // i.e. EUR/USD (a market is bidirectional) askOrders *AskOrders // i.e. buying EUR for USD lowestAsk float64 bidOrders *BidOrders // i.e. selling EUR for USD or buying USD for EUR diff --git a/abci/example/orderbook/types.go b/abci/example/orderbook/types.go index d9cb65462..b49680d02 100644 --- a/abci/example/orderbook/types.go +++ b/abci/example/orderbook/types.go @@ -117,7 +117,7 @@ func (msg *MsgRegisterPair) ValidateBasic() error { func NewCommodity(denom string, quantity float64) *Commodity { return &Commodity{ - Denom: denom, + Denom: denom, Quantity: quantity, } } @@ -260,7 +260,7 @@ func (a *Account) SubtractCommodity(c *Commodity) { curr.Quantity -= c.Quantity } -func (msg *Msg) ValidateBasic() error { +func (msg *Msg) ValidateBasic() error { switch m := msg.Sum.(type) { case *Msg_MsgRegisterPair: if err := m.MsgRegisterPair.ValidateBasic(); err != nil { @@ -286,10 +286,10 @@ func (msg *Msg) ValidateBasic() error { if err := m.MsgTradeSet.TradeSet.ValidateBasic(); err != nil { return err } - + default: return errors.New("unknown tx") } return nil -} \ No newline at end of file +} diff --git a/abci/example/orderbook/wire.proto b/abci/example/orderbook/wire.proto index babc7ad5a..78019f968 100644 --- a/abci/example/orderbook/wire.proto +++ b/abci/example/orderbook/wire.proto @@ -30,6 +30,7 @@ message Commodity { double quantity = 2; } +// Accounts is the atomic piece of information that is persisted to disk. message Account { uint64 index = 1; bytes public_key = 2;