9 "github.com/bytom/blockchain/txbuilder"
10 chainjson "github.com/bytom/encoding/json"
11 "github.com/bytom/errors"
12 "github.com/bytom/net/http/httperror"
13 "github.com/bytom/net/http/reqid"
14 "github.com/bytom/protocol/bc/legacy"
15 "github.com/bytom/log"
18 var defaultTxTTL = 5 * time.Minute
20 func (a *BlockchainReactor) actionDecoder(action string) (func([]byte) (txbuilder.Action, error), bool) {
21 var decoder func([]byte) (txbuilder.Action, error)
23 case "control_account":
24 decoder = a.accounts.DecodeControlAction
25 case "control_program":
26 decoder = txbuilder.DecodeControlProgramAction
27 case "control_receiver":
28 decoder = txbuilder.DecodeControlReceiverAction
30 decoder = a.assets.DecodeIssueAction
32 decoder = txbuilder.DecodeRetireAction
34 decoder = a.accounts.DecodeSpendAction
35 case "spend_account_unspent_output":
36 decoder = a.accounts.DecodeSpendUTXOAction
37 case "set_transaction_reference_data":
38 decoder = txbuilder.DecodeSetTxRefDataAction
45 func (a *BlockchainReactor) buildSingle(ctx context.Context, req *BuildRequest) (*txbuilder.Template, error) {
46 err := a.filterAliases(ctx, req)
50 actions := make([]txbuilder.Action, 0, len(req.Actions))
51 for i, act := range req.Actions {
52 typ, ok := act["type"].(string)
54 return nil, errors.WithDetailf(errBadActionType, "no action type provided on action %d", i)
56 decoder, ok := a.actionDecoder(typ)
58 return nil, errors.WithDetailf(errBadActionType, "unknown action type %q on action %d", typ, i)
61 // Remarshal to JSON, the action may have been modified when we
63 b, err := json.Marshal(act)
67 action, err := decoder(b)
69 return nil, errors.WithDetailf(errBadAction, "%s on action %d", err.Error(), i)
71 actions = append(actions, action)
74 ttl := req.TTL.Duration
78 maxTime := time.Now().Add(ttl)
80 tpl, err := txbuilder.Build(ctx, req.Tx, actions, maxTime)
81 if errors.Root(err) == txbuilder.ErrAction {
82 // Format each of the inner errors contained in the data.
83 var formattedErrs []httperror.Response
84 for _, innerErr := range errors.Data(err)["actions"].([]error) {
85 resp := errorFormatter.Format(innerErr)
86 formattedErrs = append(formattedErrs, resp)
88 err = errors.WithData(err, "actions", formattedErrs)
94 // ensure null is never returned for signing instructions
95 if tpl.SigningInstructions == nil {
96 tpl.SigningInstructions = []*txbuilder.SigningInstruction{}
101 // POST /build-transaction
102 func (a *BlockchainReactor) build(ctx context.Context, buildReqs []*BuildRequest) (interface{}, error) {
103 responses := make([]interface{}, len(buildReqs))
104 var wg sync.WaitGroup
105 wg.Add(len(responses))
107 for i := 0; i < len(responses); i++ {
109 subctx := reqid.NewSubContext(ctx, reqid.New())
111 defer batchRecover(subctx, &responses[i])
113 tmpl, err := a.buildSingle(subctx, buildReqs[i])
123 return responses, nil
126 func (a *BlockchainReactor) submitSingle(ctx context.Context, tpl *txbuilder.Template, waitUntil string) (interface{}, error) {
127 if tpl.Transaction == nil {
128 return nil, errors.Wrap(txbuilder.ErrMissingRawTx)
131 err := a.finalizeTxWait(ctx, tpl, waitUntil)
133 return nil, errors.Wrapf(err, "tx %s", tpl.Transaction.ID.String())
136 return map[string]string{"id": tpl.Transaction.ID.String()}, nil
140 // recordSubmittedTx records a lower bound height at which the tx
141 // was first submitted to the tx pool. If this request fails for
142 // some reason, a retry will know to look for the transaction in
143 // blocks starting at this height.
145 // If the tx has already been submitted, it returns the existing
147 func recordSubmittedTx(ctx context.Context, db pg.DB, txHash bc.Hash, currentHeight uint64) (uint64, error) {
149 INSERT INTO submitted_txs (tx_hash, height) VALUES($1, $2)
150 ON CONFLICT DO NOTHING
152 res, err := db.Exec(ctx, insertQ, txHash.Bytes(), currentHeight)
156 inserted, err := res.RowsAffected()
161 return currentHeight, nil
164 // The insert didn't affect any rows, meaning there was already an entry
165 // for this transaction hash.
167 SELECT height FROM submitted_txs WHERE tx_hash = $1
170 err = db.QueryRow(ctx, selectQ, txHash.Bytes()).Scan(&height)
176 // cleanUpSubmittedTxs will periodically delete records of submitted txs
177 // older than a day. This function blocks and only exits when its context
179 func cleanUpSubmittedTxs(ctx context.Context, db pg.DB) {
180 ticker := time.NewTicker(15 * time.Minute)
184 // TODO(jackson): We could avoid expensive bulk deletes by partitioning
185 // the table and DROP-ing tables of expired rows. Partitioning doesn't
186 // play well with ON CONFLICT clauses though, so we would need to rework
187 // how we guarantee uniqueness.
188 const q = `DELETE FROM submitted_txs WHERE submitted_at < now() - interval '1 day'`
189 _, err := db.Exec(ctx, q)
201 // finalizeTxWait calls FinalizeTx and then waits for confirmation of
202 // the transaction. A nil error return means the transaction is
203 // confirmed on the blockchain. ErrRejected means a conflicting tx is
204 // on the blockchain. context.DeadlineExceeded means ctx is an
205 // expiring context that timed out.
206 func (a *BlockchainReactor) finalizeTxWait(ctx context.Context, txTemplate *txbuilder.Template, waitUntil string) error {
207 // Use the current generator height as the lower bound of the block height
208 // that the transaction may appear in.
209 localHeight := a.chain.Height()
210 generatorHeight := localHeight
212 log.Printf(ctx, "localHeight:%v\n", localHeight)
213 // Remember this height in case we retry this submit call.
214 /*height, err := recordSubmittedTx(ctx, a.db, txTemplate.Transaction.ID, generatorHeight)
216 return errors.Wrap(err, "saving tx submitted height")
219 err := txbuilder.FinalizeTx(ctx, a.chain, txTemplate.Transaction)
223 if waitUntil == "none" {
227 _, err = a.waitForTxInBlock(ctx, txTemplate.Transaction, generatorHeight)
231 if waitUntil == "confirmed" {
239 case <-a.pinStore.AllWaiter(height):
246 func (a *BlockchainReactor) waitForTxInBlock(ctx context.Context, tx *legacy.Tx, height uint64) (uint64, error) {
247 log.Printf(ctx, "waitForTxInBlock function.")
254 case <-a.chain.BlockWaiter(height):
255 b, err := a.chain.GetBlock(height)
257 return 0, errors.Wrap(err, "getting block that just landed")
259 for _, confirmed := range b.Transactions {
260 if confirmed.ID == tx.ID {
266 if tx.MaxTime > 0 && tx.MaxTime < b.TimestampMS {
267 return 0, errors.Wrap(txbuilder.ErrRejected, "transaction max time exceeded")
270 // might still be in pool or might be rejected; we can't
271 // tell definitively until its max time elapses.
273 // Re-insert into the pool in case it was dropped.
274 err = txbuilder.FinalizeTx(ctx, a.chain, tx)
279 // TODO(jackson): Do simple rejection checks like checking if
280 // the tx's blockchain prevouts still exist in the state tree.
285 type SubmitArg struct {
286 Transactions []txbuilder.Template
287 Wait chainjson.Duration
288 WaitUntil string `json:"wait_until"` // values none, confirmed, processed. default: processed
291 // POST /submit-transaction
292 func (a *BlockchainReactor) submit(ctx context.Context, x SubmitArg) (interface{}, error) {
293 // Setup a timeout for the provided wait duration.
294 timeout := x.Wait.Duration
296 timeout = 30 * time.Second
298 ctx, cancel := context.WithTimeout(ctx, timeout)
301 responses := make([]interface{}, len(x.Transactions))
302 var wg sync.WaitGroup
303 wg.Add(len(responses))
304 for i := range responses {
306 subctx := reqid.NewSubContext(ctx, reqid.New())
308 defer batchRecover(subctx, &responses[i])
310 tx, err := a.submitSingle(subctx, &x.Transactions[i], x.WaitUntil)
311 log.Printf(ctx, "-----tx:%v\n", tx)
321 return responses, nil