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         "github.com/vapor/errors"
  17         "github.com/vapor/federation/common"
  18         "github.com/vapor/federation/config"
  19         "github.com/vapor/federation/database"
  20         "github.com/vapor/federation/database/orm"
  21         "github.com/vapor/federation/service"
  22         "github.com/vapor/federation/util"
  23         "github.com/vapor/protocol/bc"
  24 )
  25
  26 type mainchainKeeper struct {
  27         cfg        *config.Chain
  28         db         *gorm.DB
  29         node       *service.Node
  30         chainName  string
  31         assetStore *database.AssetStore
  32         fedProg    []byte
  33 }
  34
  35 func NewMainchainKeeper(db *gorm.DB, assetStore *database.AssetStore, cfg *config.Config) *mainchainKeeper {
  36         return &mainchainKeeper{
  37                 cfg:        &cfg.Mainchain,
  38                 db:         db,
  39                 node:       service.NewNode(cfg.Mainchain.Upstream),
  40                 chainName:  cfg.Mainchain.Name,
  41                 assetStore: assetStore,
  42                 fedProg:    util.SegWitWrap(util.ParseFedProg(cfg.Warders, cfg.Quorum)),
  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         dbTx := m.db.Begin()
 107         if err := m.processBlock(chain, block, txStatus); err != nil {
 108                 dbTx.Rollback()
 109                 return err
 110         }
 111
 112         return dbTx.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, m.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(), m.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         res0ID := tx.ResultIds[0]
 160         switch res := tx.Entries[*res0ID].(type) {
 161         case *btmBc.Output:
 162                 muxID = *res.Source.Ref
 163         case *btmBc.Retirement:
 164                 muxID = *res.Source.Ref
 165         default:
 166                 return ErrOutputType
 167         }
 168
 169         rawTx, err := tx.MarshalText()
 170         if err != nil {
 171                 return err
 172         }
 173
 174         ormTx := &orm.CrossTransaction{
 175                 ChainID:              chain.ID,
 176                 SourceBlockHeight:    block.Height,
 177                 SourceBlockTimestamp: block.Timestamp,
 178                 SourceBlockHash:      blockHash.String(),
 179                 SourceTxIndex:        txIndex,
 180                 SourceMuxID:          muxID.String(),
 181                 SourceTxHash:         tx.ID.String(),
 182                 SourceRawTransaction: string(rawTx),
 183                 DestBlockHeight:      sql.NullInt64{Valid: false},
 184                 DestBlockTimestamp:   sql.NullInt64{Valid: false},
 185                 DestBlockHash:        sql.NullString{Valid: false},
 186                 DestTxIndex:          sql.NullInt64{Valid: false},
 187                 DestTxHash:           sql.NullString{Valid: false},
 188                 Status:               common.CrossTxPendingStatus,
 189         }
 190         if err := m.db.Create(ormTx).Error; err != nil {
 191                 return errors.Wrap(err, fmt.Sprintf("create mainchain DepositTx %s", tx.ID.String()))
 192         }
 193
 194         statusFail := txStatus.VerifyStatus[txIndex].StatusFail
 195         crossChainInputs, err := m.getCrossChainReqs(ormTx.ID, tx, statusFail)
 196         if err != nil {
 197                 return err
 198         }
 199
 200         for _, input := range crossChainInputs {
 201                 if err := m.db.Create(input).Error; err != nil {
 202                         return errors.Wrap(err, fmt.Sprintf("create DepositFromMainchain input: txid(%s), pos(%d)", tx.ID.String(), input.SourcePos))
 203                 }
 204         }
 205
 206         return nil
 207 }
 208
 209 func (m *mainchainKeeper) getCrossChainReqs(crossTransactionID uint64, tx *types.Tx, statusFail bool) ([]*orm.CrossTransactionReq, error) {
 210         // assume inputs are from an identical owner
 211         script := hex.EncodeToString(tx.Inputs[0].ControlProgram())
 212         reqs := []*orm.CrossTransactionReq{}
 213         for i, rawOutput := range tx.Outputs {
 214                 // check valid deposit
 215                 if !bytes.Equal(rawOutput.OutputCommitment.ControlProgram, m.fedProg) {
 216                         continue
 217                 }
 218
 219                 if statusFail && *rawOutput.OutputCommitment.AssetAmount.AssetId != *consensus.BTMAssetID {
 220                         continue
 221                 }
 222
 223                 asset, err := m.assetStore.GetByAssetID(rawOutput.OutputCommitment.AssetAmount.AssetId.String())
 224                 if err != nil {
 225                         return nil, err
 226                 }
 227
 228                 req := &orm.CrossTransactionReq{
 229                         CrossTransactionID: crossTransactionID,
 230                         SourcePos:          uint64(i),
 231                         AssetID:            asset.ID,
 232                         AssetAmount:        rawOutput.OutputCommitment.AssetAmount.Amount,
 233                         Script:             script,
 234                 }
 235                 reqs = append(reqs, req)
 236         }
 237         return reqs, nil
 238 }
 239
 240 func (m *mainchainKeeper) processWithdrawalTx(chain *orm.Chain, block *types.Block, txIndex uint64, tx *types.Tx) error {
 241         blockHash := block.Hash()
 242         stmt := m.db.Model(&orm.CrossTransaction{}).Where("chain_id != ?", chain.ID).
 243                 Where(&orm.CrossTransaction{
 244                         DestTxHash: sql.NullString{tx.ID.String(), true},
 245                         Status:     common.CrossTxPendingStatus,
 246                 }).UpdateColumn(&orm.CrossTransaction{
 247                 DestBlockHeight:    sql.NullInt64{int64(block.Height), true},
 248                 DestBlockTimestamp: sql.NullInt64{int64(block.Timestamp), true},
 249                 DestBlockHash:      sql.NullString{blockHash.String(), true},
 250                 DestTxIndex:        sql.NullInt64{int64(txIndex), true},
 251                 Status:             common.CrossTxCompletedStatus,
 252         })
 253         if stmt.Error != nil {
 254                 return stmt.Error
 255         }
 256
 257         if stmt.RowsAffected != 1 {
 258                 log.Warnf("mainchainKeeper.processWithdrawalTx(%v): rows affected != 1", tx.ID.String())
 259         }
 260         return nil
 261 }
 262
 263 func (m *mainchainKeeper) processChainInfo(chain *orm.Chain, block *types.Block) error {
 264         blockHash := block.Hash()
 265         chain.BlockHash = blockHash.String()
 266         chain.BlockHeight = block.Height
 267         res := m.db.Model(chain).Where("block_hash = ?", block.PreviousBlockHash.String()).Updates(chain)
 268         if err := res.Error; err != nil {
 269                 return err
 270         }
 271
 272         if res.RowsAffected != 1 {
 273                 return ErrInconsistentDB
 274         }
 275
 276         return nil
 277 }
 278
 279 func (m *mainchainKeeper) processIssuing(txs []*types.Tx) error {
 280         for _, tx := range txs {
 281                 for _, input := range tx.Inputs {
 282                         switch inp := input.TypedInput.(type) {
 283                         case *types.IssuanceInput:
 284                                 assetID := inp.AssetID()
 285                                 if _, err := m.assetStore.GetByAssetID(assetID.String()); err == nil {
 286                                         continue
 287                                 }
 288
 289                                 m.assetStore.Add(&orm.Asset{
 290                                         AssetID:           assetID.String(),
 291                                         IssuanceProgram:   hex.EncodeToString(inp.IssuanceProgram),
 292                                         VMVersion:         inp.VMVersion,
 293                                         RawDefinitionByte: hex.EncodeToString(inp.AssetDefinition),
 294                                 })
 295                         }
 296                 }
 297         }
 298
 299         return nil
 300 }