database/store.go

   1 package database
   2
   3 import (
   4         "encoding/binary"
   5         "encoding/json"
   6         "fmt"
   7         "time"
   8
   9         "github.com/golang/protobuf/proto"
  10         log "github.com/sirupsen/logrus"
  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         blockHeaderPrefix      = []byte("BH:")
  26         blockTransactonsPrefix = []byte("BTXS:")
  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
  37         bsj := &protocol.BlockStoreState{}
  38         if err := json.Unmarshal(bytes, bsj); err != nil {
  39                 log.WithField("err", err).Panic("fail on unmarshal BlockStoreStateJSON")
  40         }
  41         return bsj
  42 }
  43
  44 // A Store encapsulates storage for blockchain validation.
  45 // It satisfies the interface protocol.Store, and provides additional
  46 // methods for querying current data.
  47 type Store struct {
  48         db    dbm.DB
  49         cache cache
  50 }
  51
  52 func calcBlockHeaderKey(height uint64, hash *bc.Hash) []byte {
  53         buf := [8]byte{}
  54         binary.BigEndian.PutUint64(buf[:], height)
  55         key := append(blockHeaderPrefix, buf[:]...)
  56         return append(key, hash.Bytes()...)
  57 }
  58
  59 func calcBlockTransactionsKey(hash *bc.Hash) []byte {
  60         return append(blockTransactonsPrefix, hash.Bytes()...)
  61 }
  62
  63 func calcTxStatusKey(hash *bc.Hash) []byte {
  64         return append(txStatusPrefix, hash.Bytes()...)
  65 }
  66
  67 func calcVoteResultKey(seq uint64) []byte {
  68         buf := [8]byte{}
  69         binary.BigEndian.PutUint64(buf[:], seq)
  70         return append(voteResultPrefix, buf[:]...)
  71 }
  72
  73 // GetBlockHeader return the block header by given hash and height
  74 func GetBlockHeader(db dbm.DB, hash *bc.Hash, height uint64) (*types.BlockHeader, error) {
  75         binaryBlockHeader := db.Get(calcBlockHeaderKey(height, hash))
  76         if binaryBlockHeader == nil {
  77                 return nil, fmt.Errorf("There are no blockHeader with given hash %s", hash.String())
  78         }
  79
  80         block := &types.Block{}
  81         if err := block.UnmarshalText(binaryBlockHeader); err != nil {
  82                 return nil, err
  83         }
  84         return &block.BlockHeader, nil
  85 }
  86
  87 // GetBlockTransactions return the block transactions by given hash
  88 func GetBlockTransactions(db dbm.DB, hash *bc.Hash) ([]*types.Tx, error) {
  89         binaryBlockTxs := db.Get(calcBlockTransactionsKey(hash))
  90         if binaryBlockTxs == nil {
  91                 return nil, fmt.Errorf("There are no block transactions with given hash %s", hash.String())
  92         }
  93
  94         block := &types.Block{}
  95         if err := block.UnmarshalText(binaryBlockTxs); err != nil {
  96                 return nil, err
  97         }
  98         return block.Transactions, nil
  99 }
 100
 101 // GetVoteResult return the vote result by given sequence
 102 func GetVoteResult(db dbm.DB, seq uint64) (*state.VoteResult, error) {
 103         data := db.Get(calcVoteResultKey(seq))
 104         if data == nil {
 105                 return nil, protocol.ErrNotFoundVoteResult
 106         }
 107
 108         voteResult := new(state.VoteResult)
 109         if err := json.Unmarshal(data, voteResult); err != nil {
 110                 return nil, errors.Wrap(err, "unmarshaling vote result")
 111         }
 112         return voteResult, nil
 113 }
 114
 115 // NewStore creates and returns a new Store object.
 116 func NewStore(db dbm.DB) *Store {
 117         fillBlockHeaderFn := func(hash *bc.Hash, height uint64) (*types.BlockHeader, error) {
 118                 return GetBlockHeader(db, hash, height)
 119         }
 120         fillBlockTxsFn := func(hash *bc.Hash) ([]*types.Tx, error) {
 121                 return GetBlockTransactions(db, hash)
 122         }
 123         fillVoteResultFn := func(seq uint64) (*state.VoteResult, error) {
 124                 return GetVoteResult(db, seq)
 125         }
 126         bc := newCache(fillBlockHeaderFn, fillBlockTxsFn, fillVoteResultFn)
 127         return &Store{
 128                 db:    db,
 129                 cache: bc,
 130         }
 131 }
 132
 133 // BlockExist check if the block is stored in disk
 134 func (s *Store) BlockExist(hash *bc.Hash, height uint64) bool {
 135         _, err := s.cache.lookupBlockHeader(hash, height)
 136         return err == nil
 137 }
 138
 139 // GetBlock return the block by given hash
 140 func (s *Store) GetBlock(hash *bc.Hash, height uint64) (*types.Block, error) {
 141         blockHeader, err := s.GetBlockHeader(hash, height)
 142         if err != nil {
 143                 return nil, err
 144         }
 145
 146         txs, err := s.GetBlockTransactions(hash)
 147         if err != nil {
 148                 return nil, err
 149         }
 150
 151         return &types.Block{
 152                 BlockHeader:  *blockHeader,
 153                 Transactions: txs,
 154         }, nil
 155 }
 156
 157 // GetBlockHeader return the BlockHeader by given hash
 158 func (s *Store) GetBlockHeader(hash *bc.Hash, height uint64) (*types.BlockHeader, error) {
 159         return s.cache.lookupBlockHeader(hash, height)
 160 }
 161
 162 // GetBlockTransactions return the Block transactions by given hash
 163 func (s *Store) GetBlockTransactions(hash *bc.Hash) ([]*types.Tx, error) {
 164         return s.cache.lookupBlockTxs(hash)
 165 }
 166
 167 // GetStoreStatus return the BlockStoreStateJSON
 168 func (s *Store) GetStoreStatus() *protocol.BlockStoreState {
 169         return loadBlockStoreStateJSON(s.db)
 170 }
 171
 172 // GetTransactionsUtxo will return all the utxo that related to the input txs
 173 func (s *Store) GetTransactionsUtxo(view *state.UtxoViewpoint, txs []*bc.Tx) error {
 174         return getTransactionsUtxo(s.db, view, txs)
 175 }
 176
 177 // GetTransactionStatus will return the utxo that related to the block hash
 178 func (s *Store) GetTransactionStatus(hash *bc.Hash) (*bc.TransactionStatus, error) {
 179         data := s.db.Get(calcTxStatusKey(hash))
 180         if data == nil {
 181                 return nil, errors.New("can't find the transaction status by given hash")
 182         }
 183
 184         ts := &bc.TransactionStatus{}
 185         if err := proto.Unmarshal(data, ts); err != nil {
 186                 return nil, errors.Wrap(err, "unmarshaling transaction status")
 187         }
 188         return ts, nil
 189 }
 190
 191 // GetUtxo will search the utxo in db
 192 func (s *Store) GetUtxo(hash *bc.Hash) (*storage.UtxoEntry, error) {
 193         return getUtxo(s.db, hash)
 194 }
 195
 196 // GetVoteResult retrive the voting result in specified vote sequence
 197 func (s *Store) GetVoteResult(seq uint64) (*state.VoteResult, error) {
 198         return s.cache.lookupVoteResult(seq)
 199 }
 200
 201 func (s *Store) LoadBlockIndex(stateBestHeight uint64) (*state.BlockIndex, error) {
 202         startTime := time.Now()
 203         blockIndex := state.NewBlockIndex()
 204         bhIter := s.db.IteratorPrefix(blockHeaderPrefix)
 205         defer bhIter.Release()
 206
 207         var lastNode *state.BlockNode
 208         for bhIter.Next() {
 209                 bh := &types.BlockHeader{}
 210                 if err := bh.UnmarshalText(bhIter.Value()); err != nil {
 211                         return nil, err
 212                 }
 213
 214                 // If a block with a height greater than the best height of state is added to the index,
 215                 // It may cause a bug that the new block cant not be process properly.
 216                 if bh.Height > stateBestHeight {
 217                         break
 218                 }
 219
 220                 var parent *state.BlockNode
 221                 if lastNode == nil || lastNode.Hash == bh.PreviousBlockHash {
 222                         parent = lastNode
 223                 } else {
 224                         parent = blockIndex.GetNode(&bh.PreviousBlockHash)
 225                 }
 226
 227                 node, err := state.NewBlockNode(bh, parent)
 228                 if err != nil {
 229                         return nil, err
 230                 }
 231
 232                 blockIndex.AddNode(node)
 233                 lastNode = node
 234         }
 235
 236         log.WithFields(log.Fields{
 237                 "module":   logModule,
 238                 "height":   stateBestHeight,
 239                 "duration": time.Since(startTime),
 240         }).Debug("initialize load history block index from database")
 241         return blockIndex, nil
 242 }
 243
 244 // SaveBlock persists a new block in the protocol.
 245 func (s *Store) SaveBlock(block *types.Block, ts *bc.TransactionStatus) error {
 246         startTime := time.Now()
 247         binaryBlockHeader, err := block.MarshalTextForBlockHeader()
 248         if err != nil {
 249                 return errors.Wrap(err, "Marshal block header")
 250         }
 251
 252         binaryBlockTxs, err := block.MarshalTextForTransactions()
 253         if err != nil {
 254                 return errors.Wrap(err, "Marshal block transactions")
 255         }
 256
 257         binaryTxStatus, err := proto.Marshal(ts)
 258         if err != nil {
 259                 return errors.Wrap(err, "marshal block transaction status")
 260         }
 261
 262         blockHash := block.Hash()
 263         batch := s.db.NewBatch()
 264         batch.Set(calcBlockHeaderKey(block.Height, &blockHash), binaryBlockHeader)
 265         batch.Set(calcBlockTransactionsKey(&blockHash), binaryBlockTxs)
 266         batch.Set(calcTxStatusKey(&blockHash), binaryTxStatus)
 267         batch.Write()
 268
 269         log.WithFields(log.Fields{
 270                 "module":   logModule,
 271                 "height":   block.Height,
 272                 "hash":     blockHash.String(),
 273                 "duration": time.Since(startTime),
 274         }).Info("block saved on disk")
 275         return nil
 276 }
 277
 278 // SaveBlockHeader persists a new block header in the protocol.
 279 func (s *Store) SaveBlockHeader(blockHeader *types.BlockHeader) error {
 280         binaryBlockHeader, err := blockHeader.MarshalText()
 281         if err != nil {
 282                 return errors.Wrap(err, "Marshal block header")
 283         }
 284
 285         blockHash := blockHeader.Hash()
 286         s.db.Set(calcBlockHeaderKey(blockHeader.Height, &blockHash), binaryBlockHeader)
 287         s.cache.removeBlockHeader(blockHeader)
 288         return nil
 289 }
 290
 291 // SaveChainStatus save the core's newest status && delete old status
 292 func (s *Store) SaveChainStatus(node, irreversibleNode *state.BlockNode, view *state.UtxoViewpoint, voteResults []*state.VoteResult) error {
 293         batch := s.db.NewBatch()
 294         if err := saveUtxoView(batch, view); err != nil {
 295                 return err
 296         }
 297
 298         for _, vote := range voteResults {
 299                 bytes, err := json.Marshal(vote)
 300                 if err != nil {
 301                         return err
 302                 }
 303
 304                 batch.Set(calcVoteResultKey(vote.Seq), bytes)
 305                 s.cache.removeVoteResult(vote)
 306         }
 307
 308         bytes, err := json.Marshal(protocol.BlockStoreState{
 309                 Height:             node.Height,
 310                 Hash:               &node.Hash,
 311                 IrreversibleHeight: irreversibleNode.Height,
 312                 IrreversibleHash:   &irreversibleNode.Hash,
 313         })
 314         if err != nil {
 315                 return err
 316         }
 317
 318         batch.Set(blockStoreKey, bytes)
 319         batch.Write()
 320         return nil
 321 }