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