database/store.go

   1 package database
   2
   3 import (
   4         "encoding/binary"
   5         "encoding/json"
   6         "time"
   7
   8         log "github.com/sirupsen/logrus"
   9         "github.com/tendermint/tmlibs/common"
  10
  11         dbm "github.com/bytom/bytom/database/leveldb"
  12         "github.com/bytom/bytom/database/storage"
  13         "github.com/bytom/bytom/errors"
  14         "github.com/bytom/bytom/protocol"
  15         "github.com/bytom/bytom/protocol/bc"
  16         "github.com/bytom/bytom/protocol/bc/types"
  17         "github.com/bytom/bytom/protocol/state"
  18 )
  19
  20 const logModule = "leveldb"
  21
  22 var (
  23         // CheckpointPrefix represent the namespace of checkpoints in db
  24         CheckpointPrefix = []byte("CP:")
  25         // BlockStoreKey block store key
  26         BlockStoreKey = []byte("blockStore")
  27         // BlockHeaderIndexPrefix  block header index with height
  28         BlockHeaderIndexPrefix = []byte("BH:")
  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 cache
  49 }
  50
  51 // NewStore creates and returns a new Store object.
  52 func NewStore(db dbm.DB) *Store {
  53         fillBlockHeaderFn := func(hash *bc.Hash) (*types.BlockHeader, error) {
  54                 return GetBlockHeader(db, hash)
  55         }
  56
  57         fillBlockTxsFn := func(hash *bc.Hash) ([]*types.Tx, error) {
  58                 return GetBlockTransactions(db, hash)
  59         }
  60
  61         fillBlockHashesFn := func(height uint64) ([]*bc.Hash, error) {
  62                 return GetBlockHashesByHeight(db, height)
  63         }
  64
  65         cache := newCache(fillBlockHeaderFn, fillBlockTxsFn, fillBlockHashesFn)
  66         return &Store{
  67                 db:    db,
  68                 cache: cache,
  69         }
  70 }
  71
  72 // GetBlockHeader return the BlockHeader by given hash
  73 func (s *Store) GetBlockHeader(hash *bc.Hash) (*types.BlockHeader, error) {
  74         return s.cache.lookupBlockHeader(hash)
  75 }
  76
  77 // GetUtxo will search the utxo in db
  78 func (s *Store) GetUtxo(hash *bc.Hash) (*storage.UtxoEntry, error) {
  79         return getUtxo(s.db, hash)
  80 }
  81
  82 func (s *Store) GetContract(hash [32]byte) ([]byte, error) {
  83         return getContract(s.db, hash)
  84 }
  85
  86 // BlockExist check if the block is stored in disk
  87 func (s *Store) BlockExist(hash *bc.Hash) bool {
  88         _, err := s.cache.lookupBlockHeader(hash)
  89         return err == nil
  90 }
  91
  92 // SaveBlockHeader persists a new block header in the protocol.
  93 func (s *Store) SaveBlockHeader(blockHeader *types.BlockHeader) error {
  94         binaryBlockHeader, err := blockHeader.MarshalText()
  95         if err != nil {
  96                 return errors.Wrap(err, "Marshal block header")
  97         }
  98
  99         blockHash := blockHeader.Hash()
 100         s.db.Set(CalcBlockHeaderKey(&blockHash), binaryBlockHeader)
 101         s.cache.removeBlockHeader(blockHeader)
 102         return nil
 103 }
 104
 105 // GetBlockHashesByHeight return the block hash by the specified height
 106 func (s *Store) GetBlockHashesByHeight(height uint64) ([]*bc.Hash, error) {
 107         return s.cache.lookupBlockHashesByHeight(height)
 108 }
 109
 110 // SaveBlock persists a new block in the protocol.
 111 func (s *Store) SaveBlock(block *types.Block) error {
 112         startTime := time.Now()
 113         binaryBlockHeader, err := block.MarshalTextForBlockHeader()
 114         if err != nil {
 115                 return errors.Wrap(err, "Marshal block header")
 116         }
 117
 118         binaryBlockTxs, err := block.MarshalTextForTransactions()
 119         if err != nil {
 120                 return errors.Wrap(err, "Marshal block transactions")
 121         }
 122
 123         blockHashes := []*bc.Hash{}
 124         hashes, err := s.GetBlockHashesByHeight(block.Height)
 125         if err != nil {
 126                 return err
 127         }
 128
 129         blockHashes = append(blockHashes, hashes...)
 130         blockHash := block.Hash()
 131         blockHashes = append(blockHashes, &blockHash)
 132         binaryBlockHashes, err := json.Marshal(blockHashes)
 133         if err != nil {
 134                 return errors.Wrap(err, "Marshal block hashes")
 135         }
 136
 137         batch := s.db.NewBatch()
 138         batch.Set(CalcBlockHashesKey(block.Height), binaryBlockHashes)
 139         batch.Set(CalcBlockHeaderKey(&blockHash), binaryBlockHeader)
 140         batch.Set(CalcBlockTransactionsKey(&blockHash), binaryBlockTxs)
 141         batch.Set(CalcBlockHeaderIndexKey(block.Height, &blockHash), binaryBlockHeader)
 142         batch.Write()
 143
 144         s.cache.removeBlockHashes(block.Height)
 145         log.WithFields(log.Fields{
 146                 "module":   logModule,
 147                 "height":   block.Height,
 148                 "hash":     blockHash.String(),
 149                 "duration": time.Since(startTime),
 150         }).Info("block saved on disk")
 151         return nil
 152 }
 153
 154 // GetBlockTransactions return the Block transactions by given hash
 155 func (s *Store) GetBlockTransactions(hash *bc.Hash) ([]*types.Tx, error) {
 156         return s.cache.lookupBlockTxs(hash)
 157 }
 158
 159 // GetBlock return the block by given hash
 160 func (s *Store) GetBlock(hash *bc.Hash) (*types.Block, error) {
 161         blockHeader, err := s.GetBlockHeader(hash)
 162         if err != nil {
 163                 return nil, err
 164         }
 165
 166         txs, err := s.GetBlockTransactions(hash)
 167         if err != nil {
 168                 return nil, err
 169         }
 170
 171         return &types.Block{
 172                 BlockHeader:  *blockHeader,
 173                 Transactions: txs,
 174         }, nil
 175 }
 176
 177 // GetTransactionsUtxo will return all the utxo that related to the input txs
 178 func (s *Store) GetTransactionsUtxo(view *state.UtxoViewpoint, txs []*bc.Tx) error {
 179         return getTransactionsUtxo(s.db, view, txs)
 180 }
 181
 182 // GetStoreStatus return the BlockStoreStateJSON
 183 func (s *Store) GetStoreStatus() *protocol.BlockStoreState {
 184         return loadBlockStoreStateJSON(s.db)
 185 }
 186
 187 // LoadBlockIndex loadblockIndex by bestHeight
 188 func (s *Store) LoadBlockIndex(stateBestHeight uint64) (*state.BlockIndex, error) {
 189         startTime := time.Now()
 190         blockIndex := state.NewBlockIndex()
 191         bhIter := s.db.IteratorPrefix(BlockHeaderIndexPrefix)
 192         defer bhIter.Release()
 193
 194         var lastNode *state.BlockNode
 195         for bhIter.Next() {
 196                 bh := &types.BlockHeader{}
 197                 if err := bh.UnmarshalText(bhIter.Value()); err != nil {
 198                         return nil, err
 199                 }
 200
 201                 // If a block with a height greater than the best height of state is added to the index,
 202                 // It may cause a bug that the new block cant not be process properly.
 203                 if bh.Height > stateBestHeight {
 204                         break
 205                 }
 206
 207                 var parent *state.BlockNode
 208                 if lastNode == nil || lastNode.Hash == bh.PreviousBlockHash {
 209                         parent = lastNode
 210                 } else {
 211                         parent = blockIndex.GetNode(&bh.PreviousBlockHash)
 212                 }
 213
 214                 node, err := state.NewBlockNode(bh, parent)
 215                 if err != nil {
 216                         return nil, err
 217                 }
 218
 219                 blockIndex.AddNode(node)
 220                 lastNode = node
 221         }
 222
 223         log.WithFields(log.Fields{
 224                 "module":   logModule,
 225                 "height":   stateBestHeight,
 226                 "duration": time.Since(startTime),
 227         }).Debug("initialize load history block index from database")
 228         return blockIndex, nil
 229 }
 230
 231 // SaveChainStatus save the core's newest status && delete old status
 232 func (s *Store) SaveChainStatus(node *state.BlockNode, view *state.UtxoViewpoint, contractView *state.ContractViewpoint, finalizedHeight uint64, finalizedHash *bc.Hash) error {
 233         batch := s.db.NewBatch()
 234         if err := saveUtxoView(batch, view); err != nil {
 235                 return err
 236         }
 237
 238         if err := deleteContractView(s.db, batch, contractView); err != nil {
 239                 return err
 240         }
 241
 242         if err := saveContractView(s.db, batch, contractView); err != nil {
 243                 return err
 244         }
 245
 246         bytes, err := json.Marshal(protocol.BlockStoreState{Height: node.Height, Hash: &node.Hash, FinalizedHeight: finalizedHeight, FinalizedHash: finalizedHash})
 247         if err != nil {
 248                 return err
 249         }
 250
 251         batch.Set(BlockStoreKey, bytes)
 252         batch.Write()
 253         return nil
 254 }
 255
 256 func calcCheckpointKey(height uint64, hash *bc.Hash) []byte {
 257         buf := make([]byte, 8)
 258         binary.BigEndian.PutUint64(buf, height)
 259         key := append(CheckpointPrefix, buf...)
 260         if hash != nil {
 261                 key = append(key, hash.Bytes()...)
 262         }
 263         return key
 264 }
 265
 266 func (s *Store) GetCheckpoint(hash *bc.Hash) (*state.Checkpoint, error) {
 267         header, err := s.GetBlockHeader(hash)
 268         if err != nil {
 269                 return nil, err
 270         }
 271
 272         data := s.db.Get(calcCheckpointKey(header.Height, hash))
 273         checkpoint := &state.Checkpoint{}
 274         if err := json.Unmarshal(data, checkpoint); err != nil {
 275                 return nil, err
 276         }
 277
 278         return checkpoint, nil
 279 }
 280
 281 // GetCheckpointsByHeight return all checkpoints of specified block height
 282 func (s *Store) GetCheckpointsByHeight(height uint64) ([]*state.Checkpoint, error) {
 283         iter := s.db.IteratorPrefix(calcCheckpointKey(height, nil))
 284         defer iter.Release()
 285         return loadCheckpointsFromIter(iter)
 286 }
 287
 288 // CheckpointsFromNode return all checkpoints from specified block height and hash
 289 func (s *Store) CheckpointsFromNode(height uint64, hash *bc.Hash) ([]*state.Checkpoint, error) {
 290         startKey := calcCheckpointKey(height, hash)
 291         iter := s.db.IteratorPrefixWithStart(CheckpointPrefix, startKey, false)
 292
 293         finalizedCheckpoint := &state.Checkpoint{}
 294         if err := json.Unmarshal(iter.Value(), finalizedCheckpoint); err != nil {
 295                 return nil, err
 296         }
 297
 298         checkpoints := []*state.Checkpoint{finalizedCheckpoint}
 299         subs, err := loadCheckpointsFromIter(iter)
 300         if err != nil {
 301                 return nil, err
 302         }
 303
 304         checkpoints = append(checkpoints, subs...)
 305         return checkpoints, nil
 306 }
 307
 308 func loadCheckpointsFromIter(iter dbm.Iterator) ([]*state.Checkpoint, error) {
 309         var checkpoints []*state.Checkpoint
 310         defer iter.Release()
 311         for iter.Next() {
 312                 checkpoint := &state.Checkpoint{}
 313                 if err := json.Unmarshal(iter.Value(), checkpoint); err != nil {
 314                         return nil, err
 315                 }
 316
 317                 checkpoints = append(checkpoints, checkpoint)
 318         }
 319         return checkpoints, nil
 320 }
 321
 322 // SaveCheckpoints bulk save multiple checkpoint
 323 func (s *Store) SaveCheckpoints(checkpoints ...*state.Checkpoint) error {
 324         batch := s.db.NewBatch()
 325         for _, checkpoint := range checkpoints {
 326                 data, err := json.Marshal(checkpoint)
 327                 if err != nil {
 328                         return err
 329                 }
 330
 331                 batch.Set(calcCheckpointKey(checkpoint.Height, &checkpoint.Hash), data)
 332         }
 333         batch.Write()
 334         return nil
 335 }