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                 if bytes := w.store.GetRawProgram(hash); bytes != nil {
 111                         return true
 112                 }
 113         }
 114
 115         for _, v := range tx.Inputs {
 116                 outid, err := v.SpentOutputID()
 117                 if err != nil {
 118                         continue
 119                 }
 120                 if bytes := w.store.GetStandardUTXO(outid); bytes != nil {
 121                         return true
 122                 }
 123         }
 124         return false
 125 }
 126
 127 // SaveUnconfirmedTx save unconfirmed annotated transaction to the database
 128 func (w *Wallet) saveUnconfirmedTx(tx *types.Tx) error {
 129         if !w.checkRelatedTransaction(tx) {
 130                 return nil
 131         }
 132
 133         // annotate account and asset
 134         annotatedTx := w.buildAnnotatedUnconfirmedTx(tx)
 135         annotatedTxs := []*query.AnnotatedTx{}
 136         annotatedTxs = append(annotatedTxs, annotatedTx)
 137         annotateTxsAccount(annotatedTxs, w.store)
 138
 139         rawTx, err := json.Marshal(annotatedTxs[0])
 140         if err != nil {
 141                 return err
 142         }
 143
 144         w.store.SetUnconfirmedTransaction(tx.ID.String(), rawTx)
 145         return nil
 146 }
 147
 148 func (w *Wallet) delExpiredTxs() error {
 149         AnnotatedTx, err := w.GetUnconfirmedTxs("")
 150         if err != nil {
 151                 return err
 152         }
 153         for _, tx := range AnnotatedTx {
 154                 if time.Now().After(time.Unix(int64(tx.Timestamp), 0).Add(MaxUnconfirmedTxDuration)) {
 155                         w.store.DeleteUnconfirmedTransaction(tx.ID.String())
 156                 }
 157         }
 158         return nil
 159 }
 160
 161 //delUnconfirmedTx periodically delete locally stored timeout did not confirm txs
 162 func (w *Wallet) delUnconfirmedTx() {
 163         if err := w.delExpiredTxs(); err != nil {
 164                 log.WithFields(log.Fields{"module": logModule, "err": err}).Error("wallet fail on delUnconfirmedTx")
 165                 return
 166         }
 167         ticker := time.NewTicker(UnconfirmedTxCheckPeriod)
 168         defer ticker.Stop()
 169         for {
 170                 <-ticker.C
 171                 if err := w.delExpiredTxs(); err != nil {
 172                         log.WithFields(log.Fields{"module": logModule, "err": err}).Error("wallet fail on delUnconfirmedTx")
 173                 }
 174         }
 175 }