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         isMuxIDFound := false
 160         for _, resOutID := range tx.ResultIds {
 161                 resOut, ok := tx.Entries[*resOutID].(*btmBc.Output)
 162                 if ok {
 163                         muxID = *resOut.Source.Ref
 164                         isMuxIDFound = true
 165                         break
 166                 }
 167         }
 168         if !isMuxIDFound {
 169                 return errors.New("fail to get mux id")
 170         }
 171
 172         rawTx, err := tx.MarshalText()
 173         if err != nil {
 174                 return err
 175         }
 176
 177         ormTx := &orm.CrossTransaction{
 178                 ChainID:              chain.ID,
 179                 SourceBlockHeight:    block.Height,
 180                 SourceBlockHash:      blockHash.String(),
 181                 SourceTxIndex:        txIndex,
 182                 SourceMuxID:          muxID.String(),
 183                 SourceTxHash:         tx.ID.String(),
 184                 SourceRawTransaction: string(rawTx),
 185                 DestBlockHeight:      sql.NullInt64{Valid: false},
 186                 DestBlockHash:        sql.NullString{Valid: false},
 187                 DestTxIndex:          sql.NullInt64{Valid: false},
 188                 DestTxHash:           sql.NullString{Valid: false},
 189                 Status:               common.CrossTxPendingStatus,
 190         }
 191         if err := m.db.Create(ormTx).Error; err != nil {
 192                 return errors.Wrap(err, fmt.Sprintf("create mainchain DepositTx %s", tx.ID.String()))
 193         }
 194
 195         statusFail := txStatus.VerifyStatus[txIndex].StatusFail
 196         crossChainInputs, err := m.getCrossChainInputs(ormTx.ID, tx, statusFail)
 197         if err != nil {
 198                 return err
 199         }
 200
 201         for _, input := range crossChainInputs {
 202                 if err := m.db.Create(input).Error; err != nil {
 203                         return errors.Wrap(err, fmt.Sprintf("create DepositFromMainchain input: txid(%s), pos(%d)", tx.ID.String(), input.SourcePos))
 204                 }
 205         }
 206
 207         return nil
 208 }
 209
 210 func (m *mainchainKeeper) getCrossChainInputs(crossTransactionID uint64, tx *types.Tx, statusFail bool) ([]*orm.CrossTransactionReq, error) {
 211         // assume inputs are from an identical owner
 212         script := hex.EncodeToString(tx.Inputs[0].ControlProgram())
 213         inputs := []*orm.CrossTransactionReq{}
 214         for i, rawOutput := range tx.Outputs {
 215                 // check valid deposit
 216                 if !bytes.Equal(rawOutput.OutputCommitment.ControlProgram, fedProg) {
 217                         continue
 218                 }
 219
 220                 if statusFail && *rawOutput.OutputCommitment.AssetAmount.AssetId != *consensus.BTMAssetID {
 221                         continue
 222                 }
 223
 224                 asset, err := m.getAsset(rawOutput.OutputCommitment.AssetAmount.AssetId.String())
 225                 if err != nil {
 226                         return nil, err
 227                 }
 228
 229                 input := &orm.CrossTransactionReq{
 230                         CrossTransactionID: crossTransactionID,
 231                         SourcePos:          uint64(i),
 232                         AssetID:            asset.ID,
 233                         AssetAmount:        rawOutput.OutputCommitment.AssetAmount.Amount,
 234                         Script:             script,
 235                 }
 236                 inputs = append(inputs, input)
 237         }
 238         return inputs, nil
 239 }
 240
 241 func (m *mainchainKeeper) processWithdrawalTx(chain *orm.Chain, block *types.Block, txIndex uint64, tx *types.Tx) error {
 242         blockHash := block.Hash()
 243         return m.db.Model(&orm.CrossTransaction{}).Where("chain_id != ?", chain.ID).
 244                 Where(&orm.CrossTransaction{
 245                         DestTxHash: sql.NullString{tx.ID.String(), true},
 246                         Status:     common.CrossTxSubmittedStatus,
 247                 }).UpdateColumn(&orm.CrossTransaction{
 248                 DestBlockHeight: sql.NullInt64{int64(block.Height), true},
 249                 DestBlockHash:   sql.NullString{blockHash.String(), true},
 250                 DestTxIndex:     sql.NullInt64{int64(txIndex), true},
 251                 Status:          common.CrossTxCompletedStatus,
 252         }).Error
 253 }
 254
 255 func (m *mainchainKeeper) processChainInfo(chain *orm.Chain, block *types.Block) error {
 256         blockHash := block.Hash()
 257         chain.BlockHash = blockHash.String()
 258         chain.BlockHeight = block.Height
 259         res := m.db.Model(chain).Where("block_hash = ?", block.PreviousBlockHash.String()).Updates(chain)
 260         if err := res.Error; err != nil {
 261                 return err
 262         }
 263
 264         if res.RowsAffected != 1 {
 265                 return ErrInconsistentDB
 266         }
 267
 268         return nil
 269 }
 270
 271 func (m *mainchainKeeper) processIssuing(txs []*types.Tx) error {
 272         for _, tx := range txs {
 273                 for _, input := range tx.Inputs {
 274                         switch inp := input.TypedInput.(type) {
 275                         case *types.IssuanceInput:
 276                                 assetID := inp.AssetID()
 277                                 if _, err := m.getAsset(assetID.String()); err == nil {
 278                                         continue
 279                                 }
 280
 281                                 asset := &orm.Asset{
 282                                         AssetID:           assetID.String(),
 283                                         IssuanceProgram:   hex.EncodeToString(inp.IssuanceProgram),
 284                                         VMVersion:         inp.VMVersion,
 285                                         RawDefinitionByte: hex.EncodeToString(inp.AssetDefinition),
 286                                 }
 287                                 if err := m.db.Create(asset).Error; err != nil {
 288                                         return err
 289                                 }
 290
 291                                 m.assetCache.Add(asset.AssetID, asset)
 292                         }
 293                 }
 294         }
 295
 296         return nil
 297 }
 298
 299 func (m *mainchainKeeper) getAsset(assetID string) (*orm.Asset, error) {
 300         if asset := m.assetCache.Get(assetID); asset != nil {
 301                 return asset, nil
 302         }
 303
 304         asset := &orm.Asset{AssetID: assetID}
 305         if err := m.db.Where(asset).First(asset).Error; err != nil {
 306                 return nil, errors.Wrap(err, "asset not found in memory and mysql")
 307         }
 308
 309         m.assetCache.Add(assetID, asset)
 310         return asset, nil
 311 }