wallet/unconfirmed.go

   1 package wallet
   2
   3 import (
   4         "encoding/json"
   5         "fmt"
   6         "sort"
   7         "time"
   8
   9         log "github.com/sirupsen/logrus"
  10
  11         "github.com/vapor/account"
  12         "github.com/vapor/blockchain/query"
  13         "github.com/vapor/crypto/sha3pool"
  14         "github.com/vapor/protocol"
  15         "github.com/vapor/protocol/bc/types"
  16 )
  17
  18 const (
  19         //UnconfirmedTxPrefix is txpool unconfirmed transactions prefix
  20         UnconfirmedTxPrefix      = "UTXS:"
  21         UnconfirmedTxCheckPeriod = 30 * time.Minute
  22         MaxUnconfirmedTxDuration = 24 * time.Hour
  23 )
  24
  25 func calcUnconfirmedTxKey(formatKey string) []byte {
  26         return []byte(UnconfirmedTxPrefix + formatKey)
  27 }
  28
  29 // SortByTimestamp implements sort.Interface for AnnotatedTx slices
  30 type SortByTimestamp []*query.AnnotatedTx
  31
  32 func (a SortByTimestamp) Len() int           { return len(a) }
  33 func (a SortByTimestamp) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
  34 func (a SortByTimestamp) Less(i, j int) bool { return a[i].Timestamp > a[j].Timestamp }
  35
  36 // AddUnconfirmedTx handle wallet status update when tx add into txpool
  37 func (w *Wallet) AddUnconfirmedTx(txD *protocol.TxDesc) {
  38         if err := w.saveUnconfirmedTx(txD.Tx); err != nil {
  39                 log.WithFields(log.Fields{"module": logModule, "err": err}).Error("wallet fail on saveUnconfirmedTx")
  40         }
  41
  42         utxos := txOutToUtxos(txD.Tx, txD.StatusFail, 0)
  43         utxos = w.filterAccountUtxo(utxos)
  44         w.AccountMgr.AddUnconfirmedUtxo(utxos)
  45 }
  46
  47 // GetUnconfirmedTxs get account unconfirmed transactions, filter transactions by accountID when accountID is not empty
  48 func (w *Wallet) GetUnconfirmedTxs(accountID string) ([]*query.AnnotatedTx, error) {
  49         annotatedTxs := []*query.AnnotatedTx{}
  50         txIter := w.DB.IteratorPrefix([]byte(UnconfirmedTxPrefix))
  51         defer txIter.Release()
  52
  53         // replace with GetAllUnconfirmedTxs
  54         for txIter.Next() {
  55                 annotatedTx := &query.AnnotatedTx{}
  56                 if err := json.Unmarshal(txIter.Value(), &annotatedTx); err != nil {
  57                         return nil, err
  58                 }
  59
  60                 if accountID == "" || findTransactionsByAccount(annotatedTx, accountID) {
  61                         annotateTxsAsset(w, []*query.AnnotatedTx{annotatedTx})
  62                         annotatedTxs = append([]*query.AnnotatedTx{annotatedTx}, annotatedTxs...)
  63                 }
  64         }
  65
  66         sort.Sort(SortByTimestamp(annotatedTxs))
  67         return annotatedTxs, nil
  68 }
  69
  70 // GetUnconfirmedTxByTxID get unconfirmed transaction by txID
  71 func (w *Wallet) GetUnconfirmedTxByTxID(txID string) (*query.AnnotatedTx, error) {
  72         annotatedTx := &query.AnnotatedTx{}
  73         txInfo := w.DB.Get(calcUnconfirmedTxKey(txID))
  74         if txInfo == nil {
  75                 return nil, fmt.Errorf("No transaction(tx_id=%s) from txpool", txID)
  76         }
  77
  78         if err := json.Unmarshal(txInfo, annotatedTx); err != nil {
  79                 return nil, err
  80         }
  81
  82         annotateTxsAsset(w, []*query.AnnotatedTx{annotatedTx})
  83         return annotatedTx, nil
  84 }
  85
  86 // RemoveUnconfirmedTx handle wallet status update when tx removed from txpool
  87 func (w *Wallet) RemoveUnconfirmedTx(txD *protocol.TxDesc) {
  88         if !w.checkRelatedTransaction(txD.Tx) {
  89                 return
  90         }
  91         w.DB.Delete(calcUnconfirmedTxKey(txD.Tx.ID.String()))
  92         w.AccountMgr.RemoveUnconfirmedUtxo(txD.Tx.ResultIds)
  93 }
  94
  95 func (w *Wallet) buildAnnotatedUnconfirmedTx(tx *types.Tx) *query.AnnotatedTx {
  96         annotatedTx := &query.AnnotatedTx{
  97                 ID:        tx.ID,
  98                 Timestamp: uint64(time.Now().UnixNano() / int64(time.Millisecond)),
  99                 Inputs:    make([]*query.AnnotatedInput, 0, len(tx.Inputs)),
 100                 Outputs:   make([]*query.AnnotatedOutput, 0, len(tx.Outputs)),
 101                 Size:      tx.SerializedSize,
 102         }
 103
 104         for i := range tx.Inputs {
 105                 annotatedTx.Inputs = append(annotatedTx.Inputs, w.BuildAnnotatedInput(tx, uint32(i)))
 106         }
 107         for i := range tx.Outputs {
 108                 annotatedTx.Outputs = append(annotatedTx.Outputs, w.BuildAnnotatedOutput(tx, i))
 109         }
 110         return annotatedTx
 111 }
 112
 113 // checkRelatedTransaction check related unconfirmed transaction.
 114 func (w *Wallet) checkRelatedTransaction(tx *types.Tx) bool {
 115         for _, v := range tx.Outputs {
 116                 var hash [32]byte
 117                 sha3pool.Sum256(hash[:], v.ControlProgram())
 118                 if bytes := w.DB.Get(account.ContractKey(hash)); bytes != nil {
 119                         return true
 120                 }
 121         }
 122
 123         for _, v := range tx.Inputs {
 124                 outid, err := v.SpentOutputID()
 125                 if err != nil {
 126                         continue
 127                 }
 128                 if bytes := w.DB.Get(account.StandardUTXOKey(outid)); bytes != nil {
 129                         return true
 130                 }
 131         }
 132         return false
 133 }
 134
 135 // SaveUnconfirmedTx save unconfirmed annotated transaction to the database
 136 func (w *Wallet) saveUnconfirmedTx(tx *types.Tx) error {
 137         if !w.checkRelatedTransaction(tx) {
 138                 return nil
 139         }
 140
 141         // annotate account and asset
 142         annotatedTx := w.buildAnnotatedUnconfirmedTx(tx)
 143         annotatedTxs := []*query.AnnotatedTx{}
 144         annotatedTxs = append(annotatedTxs, annotatedTx)
 145         annotateTxsAccount(annotatedTxs, w.DB)
 146
 147         rawTx, err := json.Marshal(annotatedTxs[0])
 148         if err != nil {
 149                 return err
 150         }
 151
 152         w.DB.Set(calcUnconfirmedTxKey(tx.ID.String()), rawTx)
 153         return nil
 154 }
 155
 156 func (w *Wallet) delExpiredTxs() error {
 157         AnnotatedTx, err := w.GetUnconfirmedTxs("")
 158         if err != nil {
 159                 return err
 160         }
 161         for _, tx := range AnnotatedTx {
 162                 if time.Now().After(time.Unix(int64(tx.Timestamp), 0).Add(MaxUnconfirmedTxDuration)) {
 163                         w.DB.Delete(calcUnconfirmedTxKey(tx.ID.String()))
 164                 }
 165         }
 166         return nil
 167 }
 168
 169 //delUnconfirmedTx periodically delete locally stored timeout did not confirm txs
 170 func (w *Wallet) delUnconfirmedTx() {
 171         if err := w.delExpiredTxs(); err != nil {
 172                 log.WithFields(log.Fields{"module": logModule, "err": err}).Error("wallet fail on delUnconfirmedTx")
 173                 return
 174         }
 175         ticker := time.NewTicker(UnconfirmedTxCheckPeriod)
 176         defer ticker.Stop()
 177         for {
 178                 <-ticker.C
 179                 if err := w.delExpiredTxs(); err != nil {
 180                         log.WithFields(log.Fields{"module": logModule, "err": err}).Error("wallet fail on delUnconfirmedTx")
 181                 }
 182         }
 183 }