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