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/bytom/vapor/consensus"
  18         "github.com/bytom/vapor/errors"
  19         "github.com/bytom/vapor/protocol/bc"
  20         "github.com/bytom/vapor/toolbar/federation/common"
  21         "github.com/bytom/vapor/toolbar/federation/config"
  22         "github.com/bytom/vapor/toolbar/federation/database"
  23         "github.com/bytom/vapor/toolbar/federation/database/orm"
  24         "github.com/bytom/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         sourceTxHash := locateSideChainTx(tx.Outputs[len(tx.Outputs)-1])
 129         return sourceTxHash != ""
 130 }
 131
 132 func locateSideChainTx(output *types.TxOutput) string {
 133         insts, err := vm.ParseProgram(output.OutputCommitment.ControlProgram)
 134         if err != nil {
 135                 return ""
 136         }
 137
 138         if len(insts) != 2 {
 139                 return ""
 140         }
 141
 142         if insts[0].Op != vm.OP_FAIL {
 143                 return ""
 144         }
 145
 146         sourceTxHash := string(insts[1].Data)
 147         if _, err = hex.DecodeString(sourceTxHash); err == nil && len(sourceTxHash) == 64 {
 148                 return sourceTxHash
 149         }
 150         return ""
 151 }
 152
 153 func (m *mainchainKeeper) processBlock(db *gorm.DB, block *types.Block, txStatus *bc.TransactionStatus) error {
 154         for i, tx := range block.Transactions {
 155                 if err := m.processIssuance(tx); err != nil {
 156                         return err
 157                 }
 158
 159                 if m.isDepositTx(tx) {
 160                         if err := m.processDepositTx(db, block, txStatus, i); err != nil {
 161                                 return err
 162                         }
 163                 }
 164
 165                 if m.isWithdrawalTx(tx) {
 166                         if err := m.processWithdrawalTx(db, block, i); err != nil {
 167                                 return err
 168                         }
 169                 }
 170         }
 171
 172         return nil
 173 }
 174
 175 func (m *mainchainKeeper) processChainInfo(db *gorm.DB, block *types.Block) error {
 176         blockHash := block.Hash()
 177         res := db.Model(&orm.Chain{}).Where("block_hash = ?", block.PreviousBlockHash.String()).Updates(&orm.Chain{
 178                 BlockHash:   blockHash.String(),
 179                 BlockHeight: block.Height,
 180         })
 181         if err := res.Error; err != nil {
 182                 return err
 183         }
 184
 185         if res.RowsAffected != 1 {
 186                 return ErrInconsistentDB
 187         }
 188         return nil
 189 }
 190
 191 func (m *mainchainKeeper) processDepositTx(db *gorm.DB, block *types.Block, txStatus *bc.TransactionStatus, txIndex int) error {
 192         tx := block.Transactions[txIndex]
 193         var muxID btmBc.Hash
 194         switch res := tx.Entries[*tx.ResultIds[0]].(type) {
 195         case *btmBc.Output:
 196                 muxID = *res.Source.Ref
 197         case *btmBc.Retirement:
 198                 muxID = *res.Source.Ref
 199         default:
 200                 return ErrOutputType
 201         }
 202
 203         rawTx, err := tx.MarshalText()
 204         if err != nil {
 205                 return err
 206         }
 207
 208         blockHash := block.Hash()
 209         ormTx := &orm.CrossTransaction{
 210                 ChainID:              m.chainID,
 211                 SourceBlockHeight:    block.Height,
 212                 SourceBlockTimestamp: block.Timestamp,
 213                 SourceBlockHash:      blockHash.String(),
 214                 SourceTxIndex:        uint64(txIndex),
 215                 SourceMuxID:          muxID.String(),
 216                 SourceTxHash:         tx.ID.String(),
 217                 SourceRawTransaction: string(rawTx),
 218                 DestBlockHeight:      sql.NullInt64{Valid: false},
 219                 DestBlockTimestamp:   sql.NullInt64{Valid: false},
 220                 DestBlockHash:        sql.NullString{Valid: false},
 221                 DestTxIndex:          sql.NullInt64{Valid: false},
 222                 DestTxHash:           sql.NullString{Valid: false},
 223                 Status:               common.CrossTxPendingStatus,
 224         }
 225         if err := db.Create(ormTx).Error; err != nil {
 226                 return errors.Wrap(err, fmt.Sprintf("create mainchain DepositTx %s", tx.ID.String()))
 227         }
 228
 229         return m.createCrossChainReqs(db, ormTx.ID, tx, txStatus.VerifyStatus[txIndex].StatusFail)
 230 }
 231
 232 func (m *mainchainKeeper) processIssuance(tx *types.Tx) error {
 233         for _, input := range tx.Inputs {
 234                 if input.InputType() != types.IssuanceInputType {
 235                         continue
 236                 }
 237
 238                 issuance := input.TypedInput.(*types.IssuanceInput)
 239                 assetID := issuance.AssetID()
 240                 if _, err := m.assetStore.GetByAssetID(assetID.String()); err == nil {
 241                         continue
 242                 }
 243
 244                 asset := &orm.Asset{
 245                         AssetID:         assetID.String(),
 246                         IssuanceProgram: hex.EncodeToString(issuance.IssuanceProgram),
 247                         VMVersion:       issuance.VMVersion,
 248                         Definition:      string(issuance.AssetDefinition),
 249                 }
 250
 251                 if err := m.db.Create(asset).Error; err != nil {
 252                         return err
 253                 }
 254         }
 255         return nil
 256 }
 257
 258 func (m *mainchainKeeper) processWithdrawalTx(db *gorm.DB, block *types.Block, txIndex int) error {
 259         blockHash := block.Hash()
 260         tx := block.Transactions[txIndex]
 261
 262         crossTx := &orm.CrossTransaction{
 263                 SourceTxHash: locateSideChainTx(tx.Outputs[len(tx.Outputs)-1]),
 264                 Status:       common.CrossTxPendingStatus,
 265         }
 266         stmt := db.Model(&orm.CrossTransaction{}).Where(crossTx).UpdateColumn(&orm.CrossTransaction{
 267                 DestBlockHeight:    sql.NullInt64{int64(block.Height), true},
 268                 DestBlockTimestamp: sql.NullInt64{int64(block.Timestamp), true},
 269                 DestBlockHash:      sql.NullString{blockHash.String(), true},
 270                 DestTxIndex:        sql.NullInt64{int64(txIndex), true},
 271                 DestTxHash:         sql.NullString{tx.ID.String(), true},
 272                 Status:             common.CrossTxCompletedStatus,
 273         })
 274         if stmt.Error != nil {
 275                 return stmt.Error
 276         }
 277
 278         if stmt.RowsAffected != 1 {
 279                 log.WithFields(log.Fields{"sourceTxHash": crossTx.SourceTxHash, "destTxHash": tx.ID.String()}).Error("fail to update withdrawal transaction")
 280         }
 281         return nil
 282 }
 283
 284 func (m *mainchainKeeper) syncBlock() (bool, error) {
 285         chain := &orm.Chain{ID: m.chainID}
 286         if err := m.db.First(chain).Error; err != nil {
 287                 return false, errors.Wrap(err, "query chain")
 288         }
 289
 290         height, err := m.node.GetBlockCount()
 291         if err != nil {
 292                 return false, err
 293         }
 294
 295         if height <= chain.BlockHeight+m.cfg.Confirmations {
 296                 return false, nil
 297         }
 298
 299         nextBlockStr, txStatus, err := m.node.GetBlockByHeight(chain.BlockHeight + 1)
 300         if err != nil {
 301                 return false, err
 302         }
 303
 304         nextBlock := &types.Block{}
 305         if err := nextBlock.UnmarshalText([]byte(nextBlockStr)); err != nil {
 306                 return false, errors.New("Unmarshal nextBlock")
 307         }
 308
 309         if nextBlock.PreviousBlockHash.String() != chain.BlockHash {
 310                 log.WithFields(log.Fields{
 311                         "remote previous_block_Hash": nextBlock.PreviousBlockHash.String(),
 312                         "db block_hash":              chain.BlockHash,
 313                 }).Fatal("fail on block hash mismatch")
 314         }
 315
 316         if err := m.tryAttachBlock(nextBlock, txStatus); err != nil {
 317                 return false, err
 318         }
 319
 320         return true, nil
 321 }
 322
 323 func (m *mainchainKeeper) tryAttachBlock(block *types.Block, txStatus *bc.TransactionStatus) error {
 324         blockHash := block.Hash()
 325         log.WithFields(log.Fields{"block_height": block.Height, "block_hash": blockHash.String()}).Info("start to attachBlock")
 326
 327         dbTx := m.db.Begin()
 328         if err := m.processBlock(dbTx, block, txStatus); err != nil {
 329                 dbTx.Rollback()
 330                 return err
 331         }
 332
 333         if err := m.processChainInfo(dbTx, block); err != nil {
 334                 dbTx.Rollback()
 335                 return err
 336         }
 337
 338         return dbTx.Commit().Error
 339 }