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         for txIter.Next() {
  54                 annotatedTx := &query.AnnotatedTx{}
  55                 if err := json.Unmarshal(txIter.Value(), &annotatedTx); err != nil {
  56                         return nil, err
  57                 }
  58
  59                 if accountID == "" || findTransactionsByAccount(annotatedTx, accountID) {
  60                         annotateTxsAsset(w, []*query.AnnotatedTx{annotatedTx})
  61                         annotatedTxs = append([]*query.AnnotatedTx{annotatedTx}, annotatedTxs...)
  62                 }
  63         }
  64
  65         sort.Sort(SortByTimestamp(annotatedTxs))
  66         return annotatedTxs, nil
  67 }
  68
  69 // GetUnconfirmedTxByTxID get unconfirmed transaction by txID
  70 func (w *Wallet) GetUnconfirmedTxByTxID(txID string) (*query.AnnotatedTx, error) {
  71         annotatedTx := &query.AnnotatedTx{}
  72         txInfo := w.DB.Get(calcUnconfirmedTxKey(txID))
  73         if txInfo == nil {
  74                 return nil, fmt.Errorf("No transaction(tx_id=%s) from txpool", txID)
  75         }
  76
  77         if err := json.Unmarshal(txInfo, annotatedTx); err != nil {
  78                 return nil, err
  79         }
  80
  81         annotateTxsAsset(w, []*query.AnnotatedTx{annotatedTx})
  82         return annotatedTx, nil
  83 }
  84
  85 // RemoveUnconfirmedTx handle wallet status update when tx removed from txpool
  86 func (w *Wallet) RemoveUnconfirmedTx(txD *protocol.TxDesc) {
  87         if !w.checkRelatedTransaction(txD.Tx) {
  88                 return
  89         }
  90         w.DB.Delete(calcUnconfirmedTxKey(txD.Tx.ID.String()))
  91         w.AccountMgr.RemoveUnconfirmedUtxo(txD.Tx.ResultIds)
  92 }
  93
  94 func (w *Wallet) buildAnnotatedUnconfirmedTx(tx *types.Tx) *query.AnnotatedTx {
  95         annotatedTx := &query.AnnotatedTx{
  96                 ID:        tx.ID,
  97                 Timestamp: uint64(time.Now().Unix()),
  98                 Inputs:    make([]*query.AnnotatedInput, 0, len(tx.Inputs)),
  99                 Outputs:   make([]*query.AnnotatedOutput, 0, len(tx.Outputs)),
 100                 Size:      tx.SerializedSize,
 101         }
 102
 103         for i := range tx.Inputs {
 104                 annotatedTx.Inputs = append(annotatedTx.Inputs, w.BuildAnnotatedInput(tx, uint32(i)))
 105         }
 106         for i := range tx.Outputs {
 107                 annotatedTx.Outputs = append(annotatedTx.Outputs, w.BuildAnnotatedOutput(tx, i))
 108         }
 109         return annotatedTx
 110 }
 111
 112 // checkRelatedTransaction check related unconfirmed transaction.
 113 func (w *Wallet) checkRelatedTransaction(tx *types.Tx) bool {
 114         for _, v := range tx.Outputs {
 115                 var hash [32]byte
 116                 sha3pool.Sum256(hash[:], v.ControlProgram())
 117                 if bytes := w.DB.Get(account.ContractKey(hash)); bytes != nil {
 118                         return true
 119                 }
 120         }
 121
 122         for _, v := range tx.Inputs {
 123                 outid, err := v.SpentOutputID()
 124                 if err != nil {
 125                         continue
 126                 }
 127                 if bytes := w.DB.Get(account.StandardUTXOKey(outid)); bytes != nil {
 128                         return true
 129                 }
 130         }
 131         return false
 132 }
 133
 134 // SaveUnconfirmedTx save unconfirmed annotated transaction to the database
 135 func (w *Wallet) saveUnconfirmedTx(tx *types.Tx) error {
 136         if !w.checkRelatedTransaction(tx) {
 137                 return nil
 138         }
 139
 140         // annotate account and asset
 141         annotatedTx := w.buildAnnotatedUnconfirmedTx(tx)
 142         annotatedTxs := []*query.AnnotatedTx{}
 143         annotatedTxs = append(annotatedTxs, annotatedTx)
 144         annotateTxsAccount(annotatedTxs, w.DB)
 145
 146         rawTx, err := json.Marshal(annotatedTxs[0])
 147         if err != nil {
 148                 return err
 149         }
 150
 151         w.DB.Set(calcUnconfirmedTxKey(tx.ID.String()), rawTx)
 152         return nil
 153 }
 154
 155 func (w *Wallet) delExpiredTxs() error {
 156         AnnotatedTx, err := w.GetUnconfirmedTxs("")
 157         if err != nil {
 158                 return err
 159         }
 160         for _, tx := range AnnotatedTx {
 161                 if time.Now().After(time.Unix(int64(tx.Timestamp), 0).Add(MaxUnconfirmedTxDuration)) {
 162                         w.DB.Delete(calcUnconfirmedTxKey(tx.ID.String()))
 163                 }
 164         }
 165         return nil
 166 }
 167
 168 //delUnconfirmedTx periodically delete locally stored timeout did not confirm txs
 169 func (w *Wallet) delUnconfirmedTx() {
 170         if err := w.delExpiredTxs(); err != nil {
 171                 log.WithFields(log.Fields{"module": logModule, "err": err}).Error("wallet fail on delUnconfirmedTx")
 172                 return
 173         }
 174         ticker := time.NewTicker(UnconfirmedTxCheckPeriod)
 175         defer ticker.Stop()
 176         for {
 177                 <-ticker.C
 178                 if err := w.delExpiredTxs(); err != nil {
 179                         log.WithFields(log.Fields{"module": logModule, "err": err}).Error("wallet fail on delUnconfirmedTx")
 180                 }
 181         }
 182 }