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