database/store.go

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