import (
"database/sql"
"encoding/hex"
+ "encoding/json"
"time"
btmTypes "github.com/bytom/protocol/bc/types"
"github.com/vapor/federation/database"
"github.com/vapor/federation/database/orm"
"github.com/vapor/federation/service"
+ "github.com/vapor/federation/util"
vaporBc "github.com/vapor/protocol/bc"
vaporTypes "github.com/vapor/protocol/bc/types"
)
var collectInterval = 5 * time.Second
+var errUnknownTxType = errors.New("unknown tx type")
+
type warder struct {
db *gorm.DB
assetStore *database.AssetStore
db: db,
assetStore: assetStore,
txCh: make(chan *orm.CrossTransaction),
- fedProg: ParseFedProg(cfg.Warders, cfg.Quorum),
+ fedProg: util.ParseFedProg(cfg.Warders, cfg.Quorum),
position: local.Position,
xpub: local.XPub,
xprv: string2xprv(xprvStr),
if warderCfg.IsLocal {
local = service.NewWarder(&warderCfg)
} else {
- remoteWarder := service.NewWarder(&warderCfg)
- remotes = append(remotes, remoteWarder)
+ remote := service.NewWarder(&warderCfg)
+ remotes = append(remotes, remote)
}
}
}
func (w *warder) Run() {
- go w.collectPendingTx()
- go w.processCrossTxRoutine()
-}
-
-func (w *warder) collectPendingTx() {
ticker := time.NewTicker(collectInterval)
for ; true; <-ticker.C {
txs := []*orm.CrossTransaction{}
if err := w.db.Preload("Chain").Preload("Reqs").
- // do not use "Where(&orm.CrossTransaction{Status: common.CrossTxPendingStatus})" directly,
+ // do not use "Where(&orm.CrossTransaction{Status: common.CrossTxInitiatedStatus})" directly,
// otherwise the field "status" will be ignored
- Model(&orm.CrossTransaction{}).Where("status = ?", common.CrossTxPendingStatus).
+ Model(&orm.CrossTransaction{}).Where("status = ?", common.CrossTxInitiatedStatus).
Find(&txs).Error; err == gorm.ErrRecordNotFound {
continue
} else if err != nil {
}
for _, tx := range txs {
- w.txCh <- tx
+ go w.tryProcessCrossTx(tx)
}
}
}
-func (w *warder) processCrossTxRoutine() {
- for ormTx := range w.txCh {
- if err := w.validateCrossTx(ormTx); err != nil {
- log.Warnln("invalid cross-chain tx", ormTx)
- continue
- }
+func (w *warder) tryProcessCrossTx(ormTx *orm.CrossTransaction) error {
+ dbTx := w.db.Begin()
+ if err := w.processCrossTx(ormTx); err != nil {
+ dbTx.Rollback()
+ return err
+ }
+
+ return dbTx.Commit().Error
+}
+
+func (w *warder) processCrossTx(ormTx *orm.CrossTransaction) error {
+ if err := w.validateCrossTx(ormTx); err != nil {
+ log.Warnln("invalid cross-chain tx", ormTx)
+ return err
+ }
+
+ destTx, destTxID, err := w.proposeDestTx(ormTx)
+ if err != nil {
+ log.WithFields(log.Fields{"err": err, "cross-chain tx": ormTx}).Warnln("proposeDestTx")
+ return err
+ }
+
+ if err := w.initDestTxSigns(destTx, ormTx); err != nil {
+ log.WithFields(log.Fields{"err": err, "cross-chain tx": ormTx}).Warnln("initDestTxSigns")
+ return err
+ }
- destTx, destTxID, err := w.proposeDestTx(ormTx)
+ var signersSigns [][][]byte
+
+ signerSigns, err := w.getSigns(destTx, ormTx)
+ if err != nil {
+ log.WithFields(log.Fields{"err": err, "cross-chain tx": ormTx}).Warnln("getSigns")
+ return err
+ }
+
+ // TODO: pass ref?
+ signersSigns = w.attachSignsForTx( /*destTx,*/ ormTx, w.position, signerSigns)
+
+ for _, remote := range w.remotes {
+ signerSigns, err := remote.RequestSigns(destTx, ormTx)
if err != nil {
- log.WithFields(log.Fields{"err": err, "cross-chain tx": ormTx}).Warnln("proposeDestTx")
- continue
+ log.WithFields(log.Fields{"err": err, "remote": remote, "cross-chain tx": ormTx}).Warnln("RequestSign")
+ return err
}
- if err := w.initDestTxSigns(destTx, ormTx); err != nil {
- log.WithFields(log.Fields{"err": err, "cross-chain tx": ormTx}).Warnln("initDestTxSigns")
- continue
- }
+ // TODO: pass ref?
+ signersSigns = w.attachSignsForTx( /*destTx,*/ ormTx, remote.Position, signerSigns)
+ }
- if err := w.signDestTx(destTx, ormTx); err != nil {
- log.WithFields(log.Fields{"err": err, "cross-chain tx": ormTx}).Warnln("signDestTx")
- continue
+ if w.isTxSignsReachQuorum(signersSigns) && w.isLeader() {
+ // TODO: check err
+ w.finalizeTx(destTx, signersSigns)
+ submittedTxID, err := w.submitTx(destTx)
+ if err != nil {
+ log.WithFields(log.Fields{"err": err, "cross-chain tx": ormTx, "dest tx": destTx}).Warnln("submitTx")
+ return err
}
- for _, remote := range w.remotes {
- signs, err := remote.RequestSign(destTx, ormTx)
- if err != nil {
- log.WithFields(log.Fields{"err": err, "remote": remote, "cross-chain tx": ormTx}).Warnln("RequestSign")
- continue
- }
-
- w.attachSignsForTx(destTx, ormTx, remote.Position, signs)
+ if submittedTxID != destTxID {
+ log.WithFields(log.Fields{"err": err, "cross-chain tx": ormTx, "builtTx ID": destTxID, "submittedTx ID": submittedTxID}).Warnln("submitTx ID mismatch")
+ return err
}
- if w.isTxSignsReachQuorum(destTx) && w.isLeader() {
- submittedTxID, err := w.submitTx(destTx)
- if err != nil {
- log.WithFields(log.Fields{"err": err, "cross-chain tx": ormTx, "dest tx": destTx}).Warnln("submitTx")
- continue
- }
-
- if submittedTxID != destTxID {
- log.WithFields(log.Fields{"err": err, "cross-chain tx": ormTx, "builtTx ID": destTxID, "submittedTx ID": submittedTxID}).Warnln("submitTx ID mismatch")
- continue
- }
-
- if err := w.updateSubmission(ormTx); err != nil {
- log.WithFields(log.Fields{"err": err, "cross-chain tx": ormTx}).Warnln("updateSubmission")
- continue
- }
+ if err := w.updateSubmission(ormTx); err != nil {
+ log.WithFields(log.Fields{"err": err, "cross-chain tx": ormTx}).Warnln("updateSubmission")
+ return err
}
}
+
+ return nil
}
func (w *warder) validateCrossTx(tx *orm.CrossTransaction) error {
return nil, "", errors.Wrap(err, "decode rawDefinitionByte")
}
- input := vaporTypes.NewCrossChainInput(nil, *muxID, *assetID, req.AssetAmount, req.SourcePos, w.fedProg, rawDefinitionByte)
+ issuanceProgramByte, err := hex.DecodeString(asset.IssuanceProgram)
+ if err != nil {
+ return nil, "", errors.Wrap(err, "decode issuanceProgramByte")
+ }
+
+ input := vaporTypes.NewCrossChainInput(nil, *muxID, *assetID, req.AssetAmount, req.SourcePos, 1, rawDefinitionByte, issuanceProgramByte)
destTxData.Inputs = append(destTxData.Inputs, input)
controlProgram, err := hex.DecodeString(req.Script)
destTx := vaporTypes.NewTx(*destTxData)
w.addInputWitness(destTx)
- if err := w.db.Where(ormTx).UpdateColumn(&orm.CrossTransaction{
- DestTxHash: sql.NullString{destTx.ID.String(), true},
- }).Error; err != nil {
+ if err := w.db.Model(&orm.CrossTransaction{}).
+ Where(&orm.CrossTransaction{ID: ormTx.ID}).
+ UpdateColumn(&orm.CrossTransaction{
+ DestTxHash: sql.NullString{destTx.ID.String(), true},
+ }).Error; err != nil {
return nil, "", err
}
// tx is a pointer to types.Tx, so the InputArguments can be set and be valid afterward
func (w *warder) addInputWitness(tx interface{}) {
- witness := [][]byte{w.fedProg}
switch tx := tx.(type) {
case *vaporTypes.Tx:
+ args := [][]byte{w.fedProg}
for i := range tx.Inputs {
- tx.SetInputArguments(uint32(i), witness)
+ tx.SetInputArguments(uint32(i), args)
}
case *btmTypes.Tx:
+ args := [][]byte{util.SegWitWrap(w.fedProg)}
for i := range tx.Inputs {
- tx.SetInputArguments(uint32(i), witness)
+ tx.SetInputArguments(uint32(i), args)
}
}
}
func (w *warder) initDestTxSigns(destTx interface{}, ormTx *orm.CrossTransaction) error {
- crossTxSigns := []*orm.CrossTransactionSign{}
for i := 1; i <= len(w.remotes)+1; i++ {
- crossTxSigns = append(crossTxSigns, &orm.CrossTransactionSign{
+ if err := w.db.Create(&orm.CrossTransactionSign{
CrossTransactionID: ormTx.ID,
WarderID: uint8(i),
Status: common.CrossTxSignPendingStatus,
- })
+ }).Error; err != nil {
+ return err
+ }
}
- return w.db.Create(crossTxSigns).Error
+
+ return w.db.Model(&orm.CrossTransaction{}).
+ Where(&orm.CrossTransaction{ID: ormTx.ID}).
+ UpdateColumn(&orm.CrossTransaction{
+ Status: common.CrossTxPendingStatus,
+ }).Error
}
-// TODO:
-func (w *warder) signDestTx(destTx interface{}, tx *orm.CrossTransaction) error {
- if tx.Status != common.CrossTxPendingStatus || !tx.DestTxHash.Valid {
- return errors.New("cross-chain tx status error")
+func (w *warder) getSignData(destTx interface{}) ([][]byte, error) {
+ var signData [][]byte
+
+ switch destTx := destTx.(type) {
+ case *vaporTypes.Tx:
+ signData = make([][]byte, len(destTx.Inputs))
+ for i := range destTx.Inputs {
+ signHash := destTx.SigHash(uint32(i))
+ signData[i] = signHash.Bytes()
+ }
+
+ case *btmTypes.Tx:
+ signData = make([][]byte, len(destTx.Inputs))
+ for i := range destTx.Inputs {
+ signHash := destTx.SigHash(uint32(i))
+ signData[i] = signHash.Bytes()
+ }
+
+ default:
+ return [][]byte{}, errUnknownTxType
}
- return nil
+ return signData, nil
+}
+
+func (w *warder) getSigns(destTx interface{}, ormTx *orm.CrossTransaction) ([][]byte, error) {
+ if ormTx.Status != common.CrossTxPendingStatus || !ormTx.DestTxHash.Valid {
+ return nil, errors.New("cross-chain tx status error")
+ }
+
+ signData, err := w.getSignData(destTx)
+ if err != nil {
+ return nil, errors.New("getSignData")
+ }
+
+ var signs [][]byte
+ for _, data := range signData {
+ var b [32]byte
+ copy(b[:], data)
+ // vaporBc.Hash & btmBc.Hash are marshaled in the same way
+ msg := vaporBc.NewHash(b)
+ sign := w.xprv.Sign([]byte(msg.String()))
+ signs = append(signs, sign)
+ }
+
+ return signs, nil
}
// TODO:
-func (w *warder) attachSignsForTx(destTx interface{}, ormTx *orm.CrossTransaction, position uint8, signs string) {
+func (w *warder) attachSignsForTx(destTx interface{}, ormTx *orm.CrossTransaction, position uint8, signs []string) error {
+ var inputsLen int
+ switch destTx := destTx.(type) {
+ case *vaporTypes.Tx:
+ inputsLen = len(destTx.Inputs)
+ case *btmTypes.Tx:
+ inputsLen = len(destTx.Inputs)
+ default:
+ return errUnknownTxType
+ }
+
+ // finalize tx?
+
+ signWitness := make([][]string, inputsLen)
+
+ b, err := json.Marshal(signs)
+ if err != nil {
+ return errors.Wrap(err, "marshal signs")
+ }
+
+ return w.db.Model(&orm.CrossTransactionSign{}).
+ Where(&orm.CrossTransactionSign{
+ CrossTransactionID: ormTx.ID,
+ WarderID: w.position,
+ }).
+ UpdateColumn(&orm.CrossTransactionSign{
+ Signatures: string(b),
+ Status: common.CrossTxSignCompletedStatus,
+ }).Error
}
// TODO:
return w.position == 1
}
+func (w *warder) finalizeTx(destTx interface{}, signersSigns [][][]byte) error {
+ return nil
+}
+
func (w *warder) submitTx(destTx interface{}) (string, error) {
switch tx := destTx.(type) {
case *btmTypes.Tx:
case *vaporTypes.Tx:
return w.sidechainNode.SubmitTx(tx)
default:
- return "", errors.New("unknown destTx type")
+ return "", errUnknownTxType
}
}
-func (w *warder) updateSubmission(tx *orm.CrossTransaction) error {
- if err := w.db.Where(tx).UpdateColumn(&orm.CrossTransaction{
- Status: common.CrossTxSubmittedStatus,
- }).Error; err != nil {
+func (w *warder) updateSubmission(ormTx *orm.CrossTransaction) error {
+ if err := w.db.Model(&orm.CrossTransaction{}).
+ Where(&orm.CrossTransaction{ID: ormTx.ID}).
+ UpdateColumn(&orm.CrossTransaction{
+ Status: common.CrossTxSubmittedStatus,
+ }).Error; err != nil {
return err
}
for _, remote := range w.remotes {
- remote.NotifySubmission(tx)
+ remote.NotifySubmission(ormTx)
}
return nil
}
OSDN Git Service
