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