federation/synchron/mainchain_keeper.go

   1 package synchron
   2
   3 import (
   4         "bytes"
   5         "database/sql"
   6         "encoding/hex"
   7         "fmt"
   8         "time"
   9
  10         "github.com/bytom/consensus"
  11         btmBc "github.com/bytom/protocol/bc"
  12         "github.com/bytom/protocol/bc/types"
  13         "github.com/jinzhu/gorm"
  14         log "github.com/sirupsen/logrus"
  15
  16         vaporCfg "github.com/vapor/config"
  17         "github.com/vapor/errors"
  18         "github.com/vapor/federation/common"
  19         "github.com/vapor/federation/config"
  20         "github.com/vapor/federation/database"
  21         "github.com/vapor/federation/database/orm"
  22         "github.com/vapor/federation/service"
  23         "github.com/vapor/protocol/bc"
  24 )
  25
  26 var fedProg = vaporCfg.FederationProgrom(vaporCfg.CommonConfig)
  27
  28 type mainchainKeeper struct {
  29         cfg        *config.Chain
  30         db         *gorm.DB
  31         node       *service.Node
  32         chainName  string
  33         assetCache *database.AssetCache
  34 }
  35
  36 func NewMainchainKeeper(db *gorm.DB, chainCfg *config.Chain) *mainchainKeeper {
  37         return &mainchainKeeper{
  38                 cfg:        chainCfg,
  39                 db:         db,
  40                 node:       service.NewNode(chainCfg.Upstream),
  41                 chainName:  chainCfg.Name,
  42                 assetCache: database.NewAssetCache(),
  43         }
  44 }
  45
  46 func (m *mainchainKeeper) Run() {
  47         ticker := time.NewTicker(time.Duration(m.cfg.SyncSeconds) * time.Second)
  48         for ; true; <-ticker.C {
  49                 for {
  50                         isUpdate, err := m.syncBlock()
  51                         if err != nil {
  52                                 log.WithField("error", err).Errorln("blockKeeper fail on process block")
  53                                 break
  54                         }
  55
  56                         if !isUpdate {
  57                                 break
  58                         }
  59                 }
  60         }
  61 }
  62
  63 func (m *mainchainKeeper) syncBlock() (bool, error) {
  64         chain := &orm.Chain{Name: m.chainName}
  65         if err := m.db.Where(chain).First(chain).Error; err != nil {
  66                 return false, errors.Wrap(err, "query chain")
  67         }
  68
  69         height, err := m.node.GetBlockCount()
  70         if err != nil {
  71                 return false, err
  72         }
  73
  74         if height <= chain.BlockHeight+m.cfg.Confirmations {
  75                 return false, nil
  76         }
  77
  78         nextBlockStr, txStatus, err := m.node.GetBlockByHeight(chain.BlockHeight + 1)
  79         if err != nil {
  80                 return false, err
  81         }
  82
  83         nextBlock := &types.Block{}
  84         if err := nextBlock.UnmarshalText([]byte(nextBlockStr)); err != nil {
  85                 return false, errors.New("Unmarshal nextBlock")
  86         }
  87
  88         if nextBlock.PreviousBlockHash.String() != chain.BlockHash {
  89                 log.WithFields(log.Fields{
  90                         "remote PreviousBlockHash": nextBlock.PreviousBlockHash.String(),
  91                         "db block_hash":            chain.BlockHash,
  92                 }).Fatal("BlockHash mismatch")
  93                 return false, ErrInconsistentDB
  94         }
  95
  96         if err := m.tryAttachBlock(chain, nextBlock, txStatus); err != nil {
  97                 return false, err
  98         }
  99
 100         return true, nil
 101 }
 102
 103 func (m *mainchainKeeper) tryAttachBlock(chain *orm.Chain, block *types.Block, txStatus *bc.TransactionStatus) error {
 104         blockHash := block.Hash()
 105         log.WithFields(log.Fields{"block_height": block.Height, "block_hash": blockHash.String()}).Info("start to attachBlock")
 106         m.db.Begin()
 107         if err := m.processBlock(chain, block, txStatus); err != nil {
 108                 m.db.Rollback()
 109                 return err
 110         }
 111
 112         return m.db.Commit().Error
 113 }
 114
 115 func (m *mainchainKeeper) processBlock(chain *orm.Chain, block *types.Block, txStatus *bc.TransactionStatus) error {
 116         if err := m.processIssuing(block.Transactions); err != nil {
 117                 return err
 118         }
 119
 120         for i, tx := range block.Transactions {
 121                 if m.isDepositTx(tx) {
 122                         if err := m.processDepositTx(chain, block, txStatus, uint64(i), tx); err != nil {
 123                                 return err
 124                         }
 125                 }
 126
 127                 if m.isWithdrawalTx(tx) {
 128                         if err := m.processWithdrawalTx(chain, block, uint64(i), tx); err != nil {
 129                                 return err
 130                         }
 131                 }
 132         }
 133
 134         return m.processChainInfo(chain, block)
 135 }
 136
 137 func (m *mainchainKeeper) isDepositTx(tx *types.Tx) bool {
 138         for _, output := range tx.Outputs {
 139                 if bytes.Equal(output.OutputCommitment.ControlProgram, fedProg) {
 140                         return true
 141                 }
 142         }
 143         return false
 144 }
 145
 146 func (m *mainchainKeeper) isWithdrawalTx(tx *types.Tx) bool {
 147         for _, input := range tx.Inputs {
 148                 if bytes.Equal(input.ControlProgram(), fedProg) {
 149                         return true
 150                 }
 151         }
 152         return false
 153 }
 154
 155 func (m *mainchainKeeper) processDepositTx(chain *orm.Chain, block *types.Block, txStatus *bc.TransactionStatus, txIndex uint64, tx *types.Tx) error {
 156         blockHash := block.Hash()
 157
 158         var muxID btmBc.Hash
 159         res0ID := tx.ResultIds[0]
 160         switch res := tx.Entries[*res0ID].(type) {
 161         case *btmBc.Output:
 162                 muxID = *res.Source.Ref
 163         case *btmBc.Retirement:
 164                 muxID = *res.Source.Ref
 165         default:
 166                 return ErrOutputType
 167         }
 168
 169         rawTx, err := tx.MarshalText()
 170         if err != nil {
 171                 return err
 172         }
 173
 174         ormTx := &orm.CrossTransaction{
 175                 ChainID:              chain.ID,
 176                 SourceBlockHeight:    block.Height,
 177                 SourceBlockHash:      blockHash.String(),
 178                 SourceTxIndex:        txIndex,
 179                 SourceMuxID:          muxID.String(),
 180                 SourceTxHash:         tx.ID.String(),
 181                 SourceRawTransaction: string(rawTx),
 182                 DestBlockHeight:      sql.NullInt64{Valid: false},
 183                 DestBlockHash:        sql.NullString{Valid: false},
 184                 DestTxIndex:          sql.NullInt64{Valid: false},
 185                 DestTxHash:           sql.NullString{Valid: false},
 186                 Status:               common.CrossTxPendingStatus,
 187         }
 188         if err := m.db.Create(ormTx).Error; err != nil {
 189                 return errors.Wrap(err, fmt.Sprintf("create mainchain DepositTx %s", tx.ID.String()))
 190         }
 191
 192         statusFail := txStatus.VerifyStatus[txIndex].StatusFail
 193         crossChainInputs, err := m.getCrossChainReqs(ormTx.ID, tx, statusFail)
 194         if err != nil {
 195                 return err
 196         }
 197
 198         for _, input := range crossChainInputs {
 199                 if err := m.db.Create(input).Error; err != nil {
 200                         return errors.Wrap(err, fmt.Sprintf("create DepositFromMainchain input: txid(%s), pos(%d)", tx.ID.String(), input.SourcePos))
 201                 }
 202         }
 203
 204         return nil
 205 }
 206
 207 func (m *mainchainKeeper) getCrossChainReqs(crossTransactionID uint64, tx *types.Tx, statusFail bool) ([]*orm.CrossTransactionReq, error) {
 208         // assume inputs are from an identical owner
 209         script := hex.EncodeToString(tx.Inputs[0].ControlProgram())
 210         inputs := []*orm.CrossTransactionReq{}
 211         for i, rawOutput := range tx.Outputs {
 212                 // check valid deposit
 213                 if !bytes.Equal(rawOutput.OutputCommitment.ControlProgram, fedProg) {
 214                         continue
 215                 }
 216
 217                 if statusFail && *rawOutput.OutputCommitment.AssetAmount.AssetId != *consensus.BTMAssetID {
 218                         continue
 219                 }
 220
 221                 asset, err := m.getAsset(rawOutput.OutputCommitment.AssetAmount.AssetId.String())
 222                 if err != nil {
 223                         return nil, err
 224                 }
 225
 226                 input := &orm.CrossTransactionReq{
 227                         CrossTransactionID: crossTransactionID,
 228                         SourcePos:          uint64(i),
 229                         AssetID:            asset.ID,
 230                         AssetAmount:        rawOutput.OutputCommitment.AssetAmount.Amount,
 231                         Script:             script,
 232                 }
 233                 inputs = append(inputs, input)
 234         }
 235         return inputs, nil
 236 }
 237
 238 func (m *mainchainKeeper) processWithdrawalTx(chain *orm.Chain, block *types.Block, txIndex uint64, tx *types.Tx) error {
 239         blockHash := block.Hash()
 240         stmt := m.db.Model(&orm.CrossTransaction{}).Where("chain_id != ?", chain.ID).
 241                 Where(&orm.CrossTransaction{
 242                         DestTxHash: sql.NullString{tx.ID.String(), true},
 243                         Status:     common.CrossTxSubmittedStatus,
 244                 }).UpdateColumn(&orm.CrossTransaction{
 245                 DestBlockHeight: sql.NullInt64{int64(block.Height), true},
 246                 DestBlockHash:   sql.NullString{blockHash.String(), true},
 247                 DestTxIndex:     sql.NullInt64{int64(txIndex), true},
 248                 Status:          common.CrossTxCompletedStatus,
 249         })
 250         if stmt.Error != nil {
 251                 return stmt.Error
 252         }
 253
 254         if stmt.RowsAffected != 1 {
 255                 log.Warnf("mainchainKeeper.processWithdrawalTx(%v): rows affected != 1", tx.ID.String())
 256         }
 257         return nil
 258 }
 259
 260 func (m *mainchainKeeper) processChainInfo(chain *orm.Chain, block *types.Block) error {
 261         blockHash := block.Hash()
 262         chain.BlockHash = blockHash.String()
 263         chain.BlockHeight = block.Height
 264         res := m.db.Model(chain).Where("block_hash = ?", block.PreviousBlockHash.String()).Updates(chain)
 265         if err := res.Error; err != nil {
 266                 return err
 267         }
 268
 269         if res.RowsAffected != 1 {
 270                 return ErrInconsistentDB
 271         }
 272
 273         return nil
 274 }
 275
 276 func (m *mainchainKeeper) processIssuing(txs []*types.Tx) error {
 277         for _, tx := range txs {
 278                 for _, input := range tx.Inputs {
 279                         switch inp := input.TypedInput.(type) {
 280                         case *types.IssuanceInput:
 281                                 assetID := inp.AssetID()
 282                                 if _, err := m.getAsset(assetID.String()); err == nil {
 283                                         continue
 284                                 }
 285
 286                                 asset := &orm.Asset{
 287                                         AssetID:           assetID.String(),
 288                                         IssuanceProgram:   hex.EncodeToString(inp.IssuanceProgram),
 289                                         VMVersion:         inp.VMVersion,
 290                                         RawDefinitionByte: hex.EncodeToString(inp.AssetDefinition),
 291                                 }
 292                                 if err := m.db.Create(asset).Error; err != nil {
 293                                         return err
 294                                 }
 295
 296                                 m.assetCache.Add(asset.AssetID, asset)
 297                         }
 298                 }
 299         }
 300
 301         return nil
 302 }
 303
 304 func (m *mainchainKeeper) getAsset(assetID string) (*orm.Asset, error) {
 305         if asset := m.assetCache.Get(assetID); asset != nil {
 306                 return asset, nil
 307         }
 308
 309         asset := &orm.Asset{AssetID: assetID}
 310         if err := m.db.Where(asset).First(asset).Error; err != nil {
 311                 return nil, errors.Wrap(err, "asset not found in memory and mysql")
 312         }
 313
 314         m.assetCache.Add(assetID, asset)
 315         return asset, nil
 316 }