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