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         "github.com/bytom/bytom/consensus"
  12         dbm "github.com/bytom/bytom/database/leveldb"
  13         "github.com/bytom/bytom/database/storage"
  14         "github.com/bytom/bytom/errors"
  15         "github.com/bytom/bytom/protocol"
  16         "github.com/bytom/bytom/protocol/bc"
  17         "github.com/bytom/bytom/protocol/bc/types"
  18         "github.com/bytom/bytom/protocol/state"
  19 )
  20
  21 const logModule = "leveldb"
  22
  23 var (
  24         // CheckpointPrefix represent the namespace of checkpoints in db
  25         CheckpointPrefix = []byte("CP:")
  26         // BlockStoreKey block store key
  27         BlockStoreKey = []byte("blockStore")
  28         // BlockHeaderIndexPrefix  block header index with height
  29         BlockHeaderIndexPrefix = []byte("BH:")
  30 )
  31
  32 func loadBlockStoreStateJSON(db dbm.DB) *protocol.BlockStoreState {
  33         bytes := db.Get(BlockStoreKey)
  34         if bytes == nil {
  35                 return nil
  36         }
  37         bsj := &protocol.BlockStoreState{}
  38         if err := json.Unmarshal(bytes, bsj); err != nil {
  39                 common.PanicCrisis(common.Fmt("Could not unmarshal bytes: %X", bytes))
  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 // NewStore creates and returns a new Store object.
  53 func NewStore(db dbm.DB) *Store {
  54         fillBlockHeaderFn := func(hash *bc.Hash) (*types.BlockHeader, error) {
  55                 return GetBlockHeader(db, hash)
  56         }
  57
  58         fillBlockTxsFn := func(hash *bc.Hash) ([]*types.Tx, error) {
  59                 return GetBlockTransactions(db, hash)
  60         }
  61
  62         fillBlockHashesFn := func(height uint64) ([]*bc.Hash, error) {
  63                 return GetBlockHashesByHeight(db, height)
  64         }
  65
  66         fillMainChainHashFn := func(height uint64) (*bc.Hash, error) {
  67                 return GetMainChainHash(db, height)
  68         }
  69
  70         cache := newCache(fillBlockHeaderFn, fillBlockTxsFn, fillBlockHashesFn, fillMainChainHashFn)
  71         return &Store{
  72                 db:    db,
  73                 cache: cache,
  74         }
  75 }
  76
  77 // GetBlockHeader return the BlockHeader by given hash
  78 func (s *Store) GetBlockHeader(hash *bc.Hash) (*types.BlockHeader, error) {
  79         return s.cache.lookupBlockHeader(hash)
  80 }
  81
  82 // GetUtxo will search the utxo in db
  83 func (s *Store) GetUtxo(hash *bc.Hash) (*storage.UtxoEntry, error) {
  84         return getUtxo(s.db, hash)
  85 }
  86
  87 func (s *Store) GetContract(hash [32]byte) ([]byte, error) {
  88         return getContract(s.db, hash)
  89 }
  90
  91 // BlockExist check if the block is stored in disk
  92 func (s *Store) BlockExist(hash *bc.Hash) bool {
  93         _, err := s.cache.lookupBlockHeader(hash)
  94         return err == nil
  95 }
  96
  97 // SaveBlockHeader persists a new block header in the protocol.
  98 func (s *Store) SaveBlockHeader(blockHeader *types.BlockHeader) error {
  99         binaryBlockHeader, err := blockHeader.MarshalText()
 100         if err != nil {
 101                 return errors.Wrap(err, "Marshal block header")
 102         }
 103
 104         blockHash := blockHeader.Hash()
 105         s.db.Set(CalcBlockHeaderKey(&blockHash), binaryBlockHeader)
 106         s.cache.removeBlockHeader(blockHeader)
 107         return nil
 108 }
 109
 110 // GetBlockHashesByHeight return the block hash by the specified height
 111 func (s *Store) GetBlockHashesByHeight(height uint64) ([]*bc.Hash, error) {
 112         return s.cache.lookupBlockHashesByHeight(height)
 113 }
 114
 115 // GetMainChainHash return the block hash by the specified height
 116 func (s *Store) GetMainChainHash(height uint64) (*bc.Hash, error) {
 117         return s.cache.lookupMainChainHash(height)
 118 }
 119
 120 // SaveBlock persists a new block in the protocol.
 121 func (s *Store) SaveBlock(block *types.Block) error {
 122         startTime := time.Now()
 123         binaryBlockHeader, err := block.MarshalTextForBlockHeader()
 124         if err != nil {
 125                 return errors.Wrap(err, "Marshal block header")
 126         }
 127
 128         binaryBlockTxs, err := block.MarshalTextForTransactions()
 129         if err != nil {
 130                 return errors.Wrap(err, "Marshal block transactions")
 131         }
 132
 133         blockHashes := []*bc.Hash{}
 134         hashes, err := s.GetBlockHashesByHeight(block.Height)
 135         if err != nil {
 136                 return err
 137         }
 138
 139         blockHashes = append(blockHashes, hashes...)
 140         blockHash := block.Hash()
 141         blockHashes = append(blockHashes, &blockHash)
 142         binaryBlockHashes, err := json.Marshal(blockHashes)
 143         if err != nil {
 144                 return errors.Wrap(err, "Marshal block hashes")
 145         }
 146
 147         batch := s.db.NewBatch()
 148         batch.Set(CalcBlockHashesKey(block.Height), binaryBlockHashes)
 149         batch.Set(CalcBlockHeaderKey(&blockHash), binaryBlockHeader)
 150         batch.Set(CalcBlockTransactionsKey(&blockHash), binaryBlockTxs)
 151         batch.Set(CalcBlockHeaderIndexKey(block.Height, &blockHash), binaryBlockHeader)
 152         batch.Write()
 153
 154         s.cache.removeBlockHashes(block.Height)
 155         log.WithFields(log.Fields{
 156                 "module":   logModule,
 157                 "height":   block.Height,
 158                 "hash":     blockHash.String(),
 159                 "duration": time.Since(startTime),
 160         }).Info("block saved on disk")
 161         return nil
 162 }
 163
 164 // GetBlockTransactions return the Block transactions by given hash
 165 func (s *Store) GetBlockTransactions(hash *bc.Hash) ([]*types.Tx, error) {
 166         return s.cache.lookupBlockTxs(hash)
 167 }
 168
 169 // GetBlock return the block by given hash
 170 func (s *Store) GetBlock(hash *bc.Hash) (*types.Block, error) {
 171         blockHeader, err := s.GetBlockHeader(hash)
 172         if err != nil {
 173                 return nil, err
 174         }
 175
 176         txs, err := s.GetBlockTransactions(hash)
 177         if err != nil {
 178                 return nil, err
 179         }
 180
 181         return &types.Block{
 182                 BlockHeader:  *blockHeader,
 183                 Transactions: txs,
 184         }, nil
 185 }
 186
 187 // GetTransactionsUtxo will return all the utxo that related to the input txs
 188 func (s *Store) GetTransactionsUtxo(view *state.UtxoViewpoint, txs []*bc.Tx) error {
 189         return getTransactionsUtxo(s.db, view, txs)
 190 }
 191
 192 // GetStoreStatus return the BlockStoreStateJSON
 193 func (s *Store) GetStoreStatus() *protocol.BlockStoreState {
 194         return loadBlockStoreStateJSON(s.db)
 195 }
 196
 197 // SaveChainStatus save the core's newest status && delete old status
 198 func (s *Store) SaveChainStatus(blockHeader *types.BlockHeader, mainBlockHeaders []*types.BlockHeader, view *state.UtxoViewpoint, contractView *state.ContractViewpoint, finalizedHeight uint64, finalizedHash *bc.Hash) error {
 199         batch := s.db.NewBatch()
 200         if err := saveUtxoView(batch, view); err != nil {
 201                 return err
 202         }
 203
 204         if err := deleteContractView(s.db, batch, contractView); err != nil {
 205                 return err
 206         }
 207
 208         if err := saveContractView(s.db, batch, contractView); err != nil {
 209                 return err
 210         }
 211
 212         blockHeaderHash := blockHeader.Hash()
 213         bytes, err := json.Marshal(
 214                 protocol.BlockStoreState{
 215                         Height:          blockHeader.Height,
 216                         Hash:            &blockHeaderHash,
 217                         FinalizedHeight: finalizedHeight,
 218                         FinalizedHash:   finalizedHash,
 219                 })
 220         if err != nil {
 221                 return err
 222         }
 223
 224         batch.Set(BlockStoreKey, bytes)
 225
 226         var clearCacheFuncs []func()
 227         // save main chain blockHeaders
 228         for _, blockHeader := range mainBlockHeaders {
 229                 bh := blockHeader
 230                 blockHash := bh.Hash()
 231                 binaryBlockHash, err := blockHash.MarshalText()
 232                 if err != nil {
 233                         return errors.Wrap(err, "Marshal block hash")
 234                 }
 235
 236                 batch.Set(calcMainChainIndexPrefix(bh.Height), binaryBlockHash)
 237                 clearCacheFuncs = append(clearCacheFuncs, func() {
 238                         s.cache.removeMainChainHash(bh.Height)
 239                 })
 240         }
 241         batch.Write()
 242         for _, clearCacheFunc := range clearCacheFuncs {
 243                 clearCacheFunc()
 244         }
 245
 246         return nil
 247 }
 248
 249 func calcCheckpointKey(height uint64, hash *bc.Hash) []byte {
 250         buf := make([]byte, 8)
 251         binary.BigEndian.PutUint64(buf, height)
 252         key := append(CheckpointPrefix, buf...)
 253         if hash != nil {
 254                 key = append(key, hash.Bytes()...)
 255         }
 256         return key
 257 }
 258
 259 func (s *Store) GetCheckpoint(hash *bc.Hash) (*state.Checkpoint, error) {
 260         header, err := s.GetBlockHeader(hash)
 261         if err != nil {
 262                 return nil, err
 263         }
 264
 265         data := s.db.Get(calcCheckpointKey(header.Height, hash))
 266         checkpoint := &state.Checkpoint{}
 267         if err := json.Unmarshal(data, checkpoint); err != nil {
 268                 return nil, err
 269         }
 270
 271         checkpoint.SupLinks = append(checkpoint.SupLinks, header.SupLinks...)
 272         return checkpoint, nil
 273 }
 274
 275 // GetCheckpointsByHeight return all checkpoints of specified block height
 276 func (s *Store) GetCheckpointsByHeight(height uint64) ([]*state.Checkpoint, error) {
 277         iter := s.db.IteratorPrefix(calcCheckpointKey(height, nil))
 278         defer iter.Release()
 279         return s.loadCheckpointsFromIter(iter)
 280 }
 281
 282 // CheckpointsFromNode return all checkpoints from specified block height and hash
 283 func (s *Store) CheckpointsFromNode(height uint64, hash *bc.Hash) ([]*state.Checkpoint, error) {
 284         startKey := calcCheckpointKey(height, hash)
 285         iter := s.db.IteratorPrefixWithStart(CheckpointPrefix, startKey, false)
 286
 287         firstCheckpoint := &state.Checkpoint{}
 288         if err := json.Unmarshal(iter.Value(), firstCheckpoint); err != nil {
 289                 return nil, err
 290         }
 291
 292         checkpoints := []*state.Checkpoint{firstCheckpoint}
 293         subs, err := s.loadCheckpointsFromIter(iter)
 294         if err != nil {
 295                 return nil, err
 296         }
 297
 298         checkpoints = append(checkpoints, subs...)
 299         return checkpoints, nil
 300 }
 301
 302 func (s *Store) loadCheckpointsFromIter(iter dbm.Iterator) ([]*state.Checkpoint, error) {
 303         var checkpoints []*state.Checkpoint
 304         defer iter.Release()
 305         for iter.Next() {
 306                 checkpoint := &state.Checkpoint{}
 307                 if err := json.Unmarshal(iter.Value(), checkpoint); err != nil {
 308                         return nil, err
 309                 }
 310
 311                 header, err := s.GetBlockHeader(&checkpoint.Hash)
 312                 if err != nil {
 313                         return nil, err
 314                 }
 315
 316                 checkpoint.SupLinks = append(checkpoint.SupLinks, header.SupLinks...)
 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
 326         if err := s.saveCheckpoints(batch, checkpoints); err != nil {
 327                 return err
 328         }
 329
 330         batch.Write()
 331         return nil
 332 }
 333
 334 func (s *Store) saveCheckpoints(batch dbm.Batch, checkpoints []*state.Checkpoint) error {
 335         for _, checkpoint := range checkpoints {
 336                 startTime := time.Now()
 337                 data, err := json.Marshal(checkpoint)
 338                 if err != nil {
 339                         return err
 340                 }
 341
 342                 if checkpoint.Height%consensus.ActiveNetParams.BlocksOfEpoch != 1 {
 343                         header, err := s.GetBlockHeader(&checkpoint.Hash)
 344                         if err != nil {
 345                                 return err
 346                         }
 347
 348                         batch.Delete(calcCheckpointKey(header.Height-1, &header.PreviousBlockHash))
 349                 }
 350
 351                 batch.Set(calcCheckpointKey(checkpoint.Height, &checkpoint.Hash), data)
 352                 log.WithFields(log.Fields{
 353                         "module":   logModule,
 354                         "height":   checkpoint.Height,
 355                         "hash":     checkpoint.Hash.String(),
 356                         "status":   checkpoint.Status,
 357                         "duration": time.Since(startTime),
 358                 }).Info("checkpoint saved on disk")
 359         }
 360         return nil
 361 }