database/store.go

   1 package database
   2
   3 import (
   4         "encoding/binary"
   5         "encoding/json"
   6         "time"
   7
   8         "github.com/golang/protobuf/proto"
   9         log "github.com/sirupsen/logrus"
  10         "github.com/tendermint/tmlibs/common"
  11
  12         dbm "github.com/vapor/database/leveldb"
  13         "github.com/vapor/database/storage"
  14         "github.com/vapor/errors"
  15         "github.com/vapor/protocol"
  16         "github.com/vapor/protocol/bc"
  17         "github.com/vapor/protocol/bc/types"
  18         "github.com/vapor/protocol/state"
  19 )
  20
  21 const logModule = "leveldb"
  22
  23 var (
  24         blockStoreKey     = []byte("blockStore")
  25         blockPrefix       = []byte("B:")
  26         blockHeaderPrefix = []byte("BH:")
  27         txStatusPrefix    = []byte("BTS:")
  28         voteResultPrefix  = []byte("VR:")
  29 )
  30
  31 func loadBlockStoreStateJSON(db dbm.DB) *protocol.BlockStoreState {
  32         bytes := db.Get(blockStoreKey)
  33         if bytes == nil {
  34                 return nil
  35         }
  36         bsj := &protocol.BlockStoreState{}
  37         if err := json.Unmarshal(bytes, bsj); err != nil {
  38                 common.PanicCrisis(common.Fmt("Could not unmarshal bytes: %X", bytes))
  39         }
  40         return bsj
  41 }
  42
  43 // A Store encapsulates storage for blockchain validation.
  44 // It satisfies the interface protocol.Store, and provides additional
  45 // methods for querying current data.
  46 type Store struct {
  47         db    dbm.DB
  48         cache blockCache
  49 }
  50
  51 func calcBlockKey(hash *bc.Hash) []byte {
  52         return append(blockPrefix, hash.Bytes()...)
  53 }
  54
  55 func calcBlockHeaderKey(height uint64, hash *bc.Hash) []byte {
  56         buf := [8]byte{}
  57         binary.BigEndian.PutUint64(buf[:], height)
  58         key := append(blockHeaderPrefix, buf[:]...)
  59         return append(key, hash.Bytes()...)
  60 }
  61
  62 func calcTxStatusKey(hash *bc.Hash) []byte {
  63         return append(txStatusPrefix, hash.Bytes()...)
  64 }
  65
  66 func calcVoteResultKey(seq uint64) []byte {
  67         buf := [8]byte{}
  68         binary.BigEndian.PutUint64(buf[:], seq)
  69         return append(voteResultPrefix, buf[:]...)
  70 }
  71
  72 // GetBlock return the block by given hash
  73 func GetBlock(db dbm.DB, hash *bc.Hash) (*types.Block, error) {
  74         bytez := db.Get(calcBlockKey(hash))
  75         if bytez == nil {
  76                 return nil, nil
  77         }
  78
  79         block := &types.Block{}
  80         err := block.UnmarshalText(bytez)
  81         return block, err
  82 }
  83
  84 // NewStore creates and returns a new Store object.
  85 func NewStore(db dbm.DB) *Store {
  86         cache := newBlockCache(func(hash *bc.Hash) (*types.Block, error) {
  87                 return GetBlock(db, hash)
  88         })
  89         return &Store{
  90                 db:    db,
  91                 cache: cache,
  92         }
  93 }
  94
  95 // GetUtxo will search the utxo in db
  96 func (s *Store) GetUtxo(hash *bc.Hash) (*storage.UtxoEntry, error) {
  97         return getUtxo(s.db, hash)
  98 }
  99
 100 // BlockExist check if the block is stored in disk
 101 func (s *Store) BlockExist(hash *bc.Hash) bool {
 102         block, err := s.cache.lookup(hash)
 103         return err == nil && block != nil
 104 }
 105
 106 // GetBlock return the block by given hash
 107 func (s *Store) GetBlock(hash *bc.Hash) (*types.Block, error) {
 108         return s.cache.lookup(hash)
 109 }
 110
 111 // GetTransactionsUtxo will return all the utxo that related to the input txs
 112 func (s *Store) GetTransactionsUtxo(view *state.UtxoViewpoint, txs []*bc.Tx) error {
 113         return getTransactionsUtxo(s.db, view, txs)
 114 }
 115
 116 // GetTransactionStatus will return the utxo that related to the block hash
 117 func (s *Store) GetTransactionStatus(hash *bc.Hash) (*bc.TransactionStatus, error) {
 118         data := s.db.Get(calcTxStatusKey(hash))
 119         if data == nil {
 120                 return nil, errors.New("can't find the transaction status by given hash")
 121         }
 122
 123         ts := &bc.TransactionStatus{}
 124         if err := proto.Unmarshal(data, ts); err != nil {
 125                 return nil, errors.Wrap(err, "unmarshaling transaction status")
 126         }
 127         return ts, nil
 128 }
 129
 130 // GetStoreStatus return the BlockStoreStateJSON
 131 func (s *Store) GetStoreStatus() *protocol.BlockStoreState {
 132         return loadBlockStoreStateJSON(s.db)
 133 }
 134
 135 // GetVoteResult retrive the voting result in specified vote sequence
 136 func (s *Store) GetVoteResult(seq uint64) (*state.VoteResult, error) {
 137         data := s.db.Get(calcVoteResultKey(seq))
 138         if data == nil {
 139                 return nil, protocol.ErrNotFoundVoteResult
 140         }
 141
 142         vr := &state.VoteResult{}
 143         if err := json.Unmarshal(data, vr); err != nil {
 144                 return nil, errors.Wrap(err, "unmarshaling vote result")
 145         }
 146         return vr, nil
 147 }
 148
 149 func (s *Store) LoadBlockIndex(stateBestHeight uint64) (*state.BlockIndex, error) {
 150         startTime := time.Now()
 151         blockIndex := state.NewBlockIndex()
 152         bhIter := s.db.IteratorPrefix(blockHeaderPrefix)
 153         defer bhIter.Release()
 154
 155         var lastNode *state.BlockNode
 156         for bhIter.Next() {
 157                 bh := &types.BlockHeader{}
 158                 if err := bh.UnmarshalText(bhIter.Value()); err != nil {
 159                         return nil, err
 160                 }
 161
 162                 // If a block with a height greater than the best height of state is added to the index,
 163                 // It may cause a bug that the new block cant not be process properly.
 164                 if bh.Height > stateBestHeight {
 165                         break
 166                 }
 167
 168                 var parent *state.BlockNode
 169                 if lastNode == nil || lastNode.Hash == bh.PreviousBlockHash {
 170                         parent = lastNode
 171                 } else {
 172                         parent = blockIndex.GetNode(&bh.PreviousBlockHash)
 173                 }
 174
 175                 node, err := state.NewBlockNode(bh, parent)
 176                 if err != nil {
 177                         return nil, err
 178                 }
 179
 180                 blockIndex.AddNode(node)
 181                 lastNode = node
 182         }
 183
 184         log.WithFields(log.Fields{
 185                 "module":   logModule,
 186                 "height":   stateBestHeight,
 187                 "duration": time.Since(startTime),
 188         }).Debug("initialize load history block index from database")
 189         return blockIndex, nil
 190 }
 191
 192 // SaveBlock persists a new block in the protocol.
 193 func (s *Store) SaveBlock(block *types.Block, ts *bc.TransactionStatus) error {
 194         startTime := time.Now()
 195         binaryBlock, err := block.MarshalText()
 196         if err != nil {
 197                 return errors.Wrap(err, "Marshal block meta")
 198         }
 199
 200         binaryBlockHeader, err := block.BlockHeader.MarshalText()
 201         if err != nil {
 202                 return errors.Wrap(err, "Marshal block header")
 203         }
 204
 205         binaryTxStatus, err := proto.Marshal(ts)
 206         if err != nil {
 207                 return errors.Wrap(err, "marshal block transaction status")
 208         }
 209
 210         blockHash := block.Hash()
 211         batch := s.db.NewBatch()
 212         batch.Set(calcBlockKey(&blockHash), binaryBlock)
 213         batch.Set(calcBlockHeaderKey(block.Height, &blockHash), binaryBlockHeader)
 214         batch.Set(calcTxStatusKey(&blockHash), binaryTxStatus)
 215         batch.Write()
 216
 217         log.WithFields(log.Fields{
 218                 "module":   logModule,
 219                 "height":   block.Height,
 220                 "hash":     blockHash.String(),
 221                 "duration": time.Since(startTime),
 222         }).Info("block saved on disk")
 223         return nil
 224 }
 225
 226 // SaveChainStatus save the core's newest status && delete old status
 227 func (s *Store) SaveChainStatus(node, irreversibleNode *state.BlockNode, view *state.UtxoViewpoint, voteMap map[uint64]*state.VoteResult) error {
 228         batch := s.db.NewBatch()
 229         if err := saveUtxoView(batch, view); err != nil {
 230                 return err
 231         }
 232
 233         if err := saveVoteResult(batch, voteMap); err != nil {
 234                 return err
 235         }
 236
 237         bytes, err := json.Marshal(protocol.BlockStoreState{
 238                 Height:             node.Height,
 239                 Hash:               &node.Hash,
 240                 IrreversibleHeight: irreversibleNode.Height,
 241                 IrreversibleHash:   &irreversibleNode.Hash,
 242         })
 243         if err != nil {
 244                 return err
 245         }
 246
 247         batch.Set(blockStoreKey, bytes)
 248         batch.Write()
 249         return nil
 250 }
 251
 252 // SaveChainNodeStatus update the best node and irreversible node
 253 func (s *Store) SaveChainNodeStatus(bestNode, irreversibleNode *state.BlockNode) error {
 254         bytes, err := json.Marshal(protocol.BlockStoreState{
 255                 Height:             bestNode.Height,
 256                 Hash:               &bestNode.Hash,
 257                 IrreversibleHeight: irreversibleNode.Height,
 258                 IrreversibleHash:   &irreversibleNode.Hash,
 259         })
 260         if err != nil {
 261                 return err
 262         }
 263
 264         s.db.Set(blockStoreKey, bytes)
 265         return nil
 266 }
 267
 268 // saveVoteResult update the voting results generated by each irreversible block
 269 func saveVoteResult(batch dbm.Batch, voteMap map[uint64]*state.VoteResult) error {
 270         for _, vote := range voteMap {
 271                 bytes, err := json.Marshal(vote)
 272                 if err != nil {
 273                         return err
 274                 }
 275
 276                 batch.Set(calcVoteResultKey(vote.Seq), bytes)
 277         }
 278         return nil
 279 }