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