hermez-node/coordinator/txmanager.go
coordinator/txmanager.goRaw
1package coordinator
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "math/big"
8 "time"
9
10 "github.com/ethereum/go-ethereum"
11 "github.com/ethereum/go-ethereum/accounts"
12 "github.com/ethereum/go-ethereum/accounts/abi/bind"
13 "github.com/ethereum/go-ethereum/core"
14 "github.com/ethereum/go-ethereum/core/types"
15 "github.com/hermeznetwork/hermez-node/common"
16 "github.com/hermeznetwork/hermez-node/db/l2db"
17 "github.com/hermeznetwork/hermez-node/eth"
18 "github.com/hermeznetwork/hermez-node/log"
19 "github.com/hermeznetwork/hermez-node/synchronizer"
20 "github.com/hermeznetwork/tracerr"
21)
22
23// TxManager handles everything related to ethereum transactions: It makes the
24// call to forge, waits for transaction confirmation, and keeps checking them
25// until a number of confirmed blocks have passed.
26type TxManager struct {
27 cfg Config
28 ethClient eth.ClientInterface
29 l2DB *l2db.L2DB // Used only to mark forged txs as forged in the L2DB
30 coord *Coordinator // Used only to send messages to stop the pipeline
31 batchCh chan *BatchInfo
32 chainID *big.Int
33 account accounts.Account
34 consts synchronizer.SCConsts
35
36 stats synchronizer.Stats
37 vars synchronizer.SCVariables
38 statsVarsCh chan statsVars
39
40 discardPipelineCh chan int // int refers to the pipelineNum
41
42 minPipelineNum int
43 queue Queue
44 // lastSuccessBatch stores the last BatchNum that who's forge call was confirmed
45 lastSuccessBatch common.BatchNum
46 // lastPendingBatch common.BatchNum
47 // accNonce is the account nonce in the last mined block (due to mined txs)
48 accNonce uint64
49 // accNextNonce is the nonce that we should use to send the next tx.
50 // In some cases this will be a reused nonce of an already pending tx.
51 accNextNonce uint64
52
53 lastSentL1BatchBlockNum int64
54}
55
56// NewTxManager creates a new TxManager
57func NewTxManager(ctx context.Context, cfg *Config, ethClient eth.ClientInterface, l2DB *l2db.L2DB,
58 coord *Coordinator, scConsts *synchronizer.SCConsts, initSCVars *synchronizer.SCVariables) (*TxManager, error) {
59 chainID, err := ethClient.EthChainID()
60 if err != nil {
61 return nil, tracerr.Wrap(err)
62 }
63 address, err := ethClient.EthAddress()
64 if err != nil {
65 return nil, tracerr.Wrap(err)
66 }
67 accNonce, err := ethClient.EthNonceAt(ctx, *address, nil)
68 if err != nil {
69 return nil, err
70 }
71 log.Infow("TxManager started", "nonce", accNonce)
72 return &TxManager{
73 cfg: *cfg,
74 ethClient: ethClient,
75 l2DB: l2DB,
76 coord: coord,
77 batchCh: make(chan *BatchInfo, queueLen),
78 statsVarsCh: make(chan statsVars, queueLen),
79 discardPipelineCh: make(chan int, queueLen),
80 account: accounts.Account{
81 Address: *address,
82 },
83 chainID: chainID,
84 consts: *scConsts,
85
86 vars: *initSCVars,
87
88 minPipelineNum: 0,
89 queue: NewQueue(),
90 accNonce: accNonce,
91 accNextNonce: accNonce,
92 }, nil
93}
94
95// AddBatch is a thread safe method to pass a new batch TxManager to be sent to
96// the smart contract via the forge call
97func (t *TxManager) AddBatch(ctx context.Context, batchInfo *BatchInfo) {
98 select {
99 case t.batchCh <- batchInfo:
100 case <-ctx.Done():
101 }
102}
103
104// SetSyncStatsVars is a thread safe method to sets the synchronizer Stats
105func (t *TxManager) SetSyncStatsVars(ctx context.Context, stats *synchronizer.Stats, vars *synchronizer.SCVariablesPtr) {
106 select {
107 case t.statsVarsCh <- statsVars{Stats: *stats, Vars: *vars}:
108 case <-ctx.Done():
109 }
110}
111
112// DiscardPipeline is a thread safe method to notify about a discarded pipeline
113// due to a reorg
114func (t *TxManager) DiscardPipeline(ctx context.Context, pipelineNum int) {
115 select {
116 case t.discardPipelineCh <- pipelineNum:
117 case <-ctx.Done():
118 }
119}
120
121func (t *TxManager) syncSCVars(vars synchronizer.SCVariablesPtr) {
122 updateSCVars(&t.vars, vars)
123}
124
125// NewAuth generates a new auth object for an ethereum transaction
126func (t *TxManager) NewAuth(ctx context.Context) (*bind.TransactOpts, error) {
127 gasPrice, err := t.ethClient.EthSuggestGasPrice(ctx)
128 if err != nil {
129 return nil, tracerr.Wrap(err)
130 }
131 inc := new(big.Int).Set(gasPrice)
132 // TODO: Replace this by a value of percentage
133 const gasPriceDiv = 100
134 inc.Div(inc, new(big.Int).SetUint64(gasPriceDiv))
135 gasPrice.Add(gasPrice, inc)
136 // log.Debugw("TxManager: transaction metadata", "gasPrice", gasPrice)
137
138 auth, err := bind.NewKeyStoreTransactorWithChainID(t.ethClient.EthKeyStore(), t.account, t.chainID)
139 if err != nil {
140 return nil, tracerr.Wrap(err)
141 }
142 auth.Value = big.NewInt(0) // in wei
143 // TODO: Calculate GasLimit based on the contents of the ForgeBatchArgs
144 auth.GasLimit = 1000000
145 auth.GasPrice = gasPrice
146 auth.Nonce = nil
147
148 return auth, nil
149}
150
151func (t *TxManager) shouldSendRollupForgeBatch(batchInfo *BatchInfo) error {
152 nextBlock := t.stats.Eth.LastBlock.Num + 1
153 if !t.canForgeAt(nextBlock) {
154 return tracerr.Wrap(fmt.Errorf("can't forge in the next block: %v", nextBlock))
155 }
156 if t.mustL1L2Batch(nextBlock) && !batchInfo.L1Batch {
157 return tracerr.Wrap(fmt.Errorf("can't forge non-L1Batch in the next block: %v", nextBlock))
158 }
159 margin := t.cfg.SendBatchBlocksMarginCheck
160 if margin != 0 {
161 if !t.canForgeAt(nextBlock + margin) {
162 return tracerr.Wrap(fmt.Errorf("can't forge after %v blocks: %v",
163 margin, nextBlock))
164 }
165 if t.mustL1L2Batch(nextBlock+margin) && !batchInfo.L1Batch {
166 return tracerr.Wrap(fmt.Errorf("can't forge non-L1Batch after %v blocks: %v",
167 margin, nextBlock))
168 }
169 }
170 return nil
171}
172
173func addPerc(v *big.Int, p int64) *big.Int {
174 r := new(big.Int).Set(v)
175 r.Mul(r, big.NewInt(p))
176 // nolint reason: to calculate percentages we divide by 100
177 r.Div(r, big.NewInt(100)) //nolit:gomnd
178 return r.Add(v, r)
179}
180
181func (t *TxManager) sendRollupForgeBatch(ctx context.Context, batchInfo *BatchInfo, resend bool) error {
182 var ethTx *types.Transaction
183 var err error
184 auth, err := t.NewAuth(ctx)
185 if err != nil {
186 return tracerr.Wrap(err)
187 }
188 auth.Nonce = big.NewInt(int64(t.accNextNonce))
189 if resend {
190 auth.Nonce = big.NewInt(int64(batchInfo.EthTx.Nonce()))
191 }
192 for attempt := 0; attempt < t.cfg.EthClientAttempts; attempt++ {
193 if auth.GasPrice.Cmp(t.cfg.MaxGasPrice) > 0 {
194 return tracerr.Wrap(fmt.Errorf("calculated gasPrice (%v) > maxGasPrice (%v)",
195 auth.GasPrice, t.cfg.MaxGasPrice))
196 }
197 // RollupForgeBatch() calls ethclient.SendTransaction()
198 ethTx, err = t.ethClient.RollupForgeBatch(batchInfo.ForgeBatchArgs, auth)
199 if errors.Is(err, core.ErrNonceTooLow) {
200 log.Warnw("TxManager ethClient.RollupForgeBatch incrementing nonce",
201 "err", err, "nonce", auth.Nonce, "batchNum", batchInfo.BatchNum)
202 auth.Nonce.Add(auth.Nonce, big.NewInt(1))
203 attempt--
204 } else if errors.Is(err, core.ErrNonceTooHigh) {
205 log.Warnw("TxManager ethClient.RollupForgeBatch decrementing nonce",
206 "err", err, "nonce", auth.Nonce, "batchNum", batchInfo.BatchNum)
207 auth.Nonce.Sub(auth.Nonce, big.NewInt(1))
208 attempt--
209 } else if errors.Is(err, core.ErrUnderpriced) {
210 log.Warnw("TxManager ethClient.RollupForgeBatch incrementing gasPrice",
211 "err", err, "gasPrice", auth.GasPrice, "batchNum", batchInfo.BatchNum)
212 auth.GasPrice = addPerc(auth.GasPrice, 10)
213 attempt--
214 } else if errors.Is(err, core.ErrReplaceUnderpriced) {
215 log.Warnw("TxManager ethClient.RollupForgeBatch incrementing gasPrice",
216 "err", err, "gasPrice", auth.GasPrice, "batchNum", batchInfo.BatchNum)
217 auth.GasPrice = addPerc(auth.GasPrice, 10)
218 attempt--
219 } else if err != nil {
220 log.Errorw("TxManager ethClient.RollupForgeBatch",
221 "attempt", attempt, "err", err, "block", t.stats.Eth.LastBlock.Num+1,
222 "batchNum", batchInfo.BatchNum)
223 } else {
224 break
225 }
226 select {
227 case <-ctx.Done():
228 return tracerr.Wrap(common.ErrDone)
229 case <-time.After(t.cfg.EthClientAttemptsDelay):
230 }
231 }
232 if err != nil {
233 return tracerr.Wrap(fmt.Errorf("reached max attempts for ethClient.RollupForgeBatch: %w", err))
234 }
235 if !resend {
236 t.accNextNonce = auth.Nonce.Uint64() + 1
237 }
238 batchInfo.EthTx = ethTx
239 log.Infow("TxManager ethClient.RollupForgeBatch", "batch", batchInfo.BatchNum, "tx", ethTx.Hash())
240 now := time.Now()
241 batchInfo.SendTimestamp = now
242
243 if resend {
244 batchInfo.Debug.ResendNum++
245 }
246 batchInfo.Debug.Status = StatusSent
247 batchInfo.Debug.SendBlockNum = t.stats.Eth.LastBlock.Num + 1
248 batchInfo.Debug.SendTimestamp = batchInfo.SendTimestamp
249 batchInfo.Debug.StartToSendDelay = batchInfo.Debug.SendTimestamp.Sub(
250 batchInfo.Debug.StartTimestamp).Seconds()
251 t.cfg.debugBatchStore(batchInfo)
252
253 if !resend {
254 if batchInfo.L1Batch {
255 t.lastSentL1BatchBlockNum = t.stats.Eth.LastBlock.Num + 1
256 }
257 }
258 if err := t.l2DB.DoneForging(common.TxIDsFromL2Txs(batchInfo.L2Txs), batchInfo.BatchNum); err != nil {
259 return tracerr.Wrap(err)
260 }
261 return nil
262}
263
264// checkEthTransactionReceipt takes the txHash from the BatchInfo and stores
265// the corresponding receipt if found
266func (t *TxManager) checkEthTransactionReceipt(ctx context.Context, batchInfo *BatchInfo) error {
267 txHash := batchInfo.EthTx.Hash()
268 var receipt *types.Receipt
269 var err error
270 for attempt := 0; attempt < t.cfg.EthClientAttempts; attempt++ {
271 receipt, err = t.ethClient.EthTransactionReceipt(ctx, txHash)
272 if ctx.Err() != nil {
273 continue
274 } else if tracerr.Unwrap(err) == ethereum.NotFound {
275 err = nil
276 break
277 } else if err != nil {
278 log.Errorw("TxManager ethClient.EthTransactionReceipt",
279 "attempt", attempt, "err", err)
280 } else {
281 break
282 }
283 select {
284 case <-ctx.Done():
285 return tracerr.Wrap(common.ErrDone)
286 case <-time.After(t.cfg.EthClientAttemptsDelay):
287 }
288 }
289 if err != nil {
290 return tracerr.Wrap(fmt.Errorf("reached max attempts for ethClient.EthTransactionReceipt: %w", err))
291 }
292 batchInfo.Receipt = receipt
293 t.cfg.debugBatchStore(batchInfo)
294 return nil
295}
296
297func (t *TxManager) handleReceipt(ctx context.Context, batchInfo *BatchInfo) (*int64, error) {
298 receipt := batchInfo.Receipt
299 if receipt != nil {
300 if batchInfo.EthTx.Nonce()+1 > t.accNonce {
301 t.accNonce = batchInfo.EthTx.Nonce() + 1
302 }
303 if receipt.Status == types.ReceiptStatusFailed {
304 batchInfo.Debug.Status = StatusFailed
305 _, err := t.ethClient.EthCall(ctx, batchInfo.EthTx, receipt.BlockNumber)
306 log.Warnw("TxManager receipt status is failed", "tx", receipt.TxHash,
307 "batch", batchInfo.BatchNum, "block", receipt.BlockNumber.Int64(),
308 "err", err)
309 batchInfo.EthTxErr = err
310 if batchInfo.BatchNum <= t.lastSuccessBatch {
311 t.lastSuccessBatch = batchInfo.BatchNum - 1
312 }
313 t.cfg.debugBatchStore(batchInfo)
314 return nil, tracerr.Wrap(fmt.Errorf(
315 "ethereum transaction receipt status is failed: %w", err))
316 } else if receipt.Status == types.ReceiptStatusSuccessful {
317 batchInfo.Debug.Status = StatusMined
318 batchInfo.Debug.MineBlockNum = receipt.BlockNumber.Int64()
319 batchInfo.Debug.StartToMineBlocksDelay = batchInfo.Debug.MineBlockNum -
320 batchInfo.Debug.StartBlockNum
321 if batchInfo.Debug.StartToMineDelay == 0 {
322 if block, err := t.ethClient.EthBlockByNumber(ctx,
323 receipt.BlockNumber.Int64()); err != nil {
324 log.Warnw("TxManager: ethClient.EthBlockByNumber", "err", err)
325 } else {
326 batchInfo.Debug.SendToMineDelay = block.Timestamp.Sub(
327 batchInfo.Debug.SendTimestamp).Seconds()
328 batchInfo.Debug.StartToMineDelay = block.Timestamp.Sub(
329 batchInfo.Debug.StartTimestamp).Seconds()
330 }
331 }
332 t.cfg.debugBatchStore(batchInfo)
333 if batchInfo.BatchNum > t.lastSuccessBatch {
334 t.lastSuccessBatch = batchInfo.BatchNum
335 }
336 confirm := t.stats.Eth.LastBlock.Num - receipt.BlockNumber.Int64()
337 return &confirm, nil
338 }
339 }
340 return nil, nil
341}
342
343// TODO:
344// - After sending a message: CancelPipeline, stop all consecutive pending Batches (transactions)
345
346// Queue of BatchInfos
347type Queue struct {
348 list []*BatchInfo
349 // nonceByBatchNum map[common.BatchNum]uint64
350 next int
351}
352
353// NewQueue returns a new queue
354func NewQueue() Queue {
355 return Queue{
356 list: make([]*BatchInfo, 0),
357 // nonceByBatchNum: make(map[common.BatchNum]uint64),
358 next: 0,
359 }
360}
361
362// Len is the length of the queue
363func (q *Queue) Len() int {
364 return len(q.list)
365}
366
367// At returns the BatchInfo at position (or nil if position is out of bounds)
368func (q *Queue) At(position int) *BatchInfo {
369 if position >= len(q.list) {
370 return nil
371 }
372 return q.list[position]
373}
374
375// Next returns the next BatchInfo (or nil if queue is empty)
376func (q *Queue) Next() (int, *BatchInfo) {
377 if len(q.list) == 0 {
378 return 0, nil
379 }
380 defer func() { q.next = (q.next + 1) % len(q.list) }()
381 return q.next, q.list[q.next]
382}
383
384// Remove removes the BatchInfo at position
385func (q *Queue) Remove(position int) {
386 // batchInfo := q.list[position]
387 // delete(q.nonceByBatchNum, batchInfo.BatchNum)
388 q.list = append(q.list[:position], q.list[position+1:]...)
389 if len(q.list) == 0 {
390 q.next = 0
391 } else {
392 q.next = position % len(q.list)
393 }
394}
395
396// Push adds a new BatchInfo
397func (q *Queue) Push(batchInfo *BatchInfo) {
398 q.list = append(q.list, batchInfo)
399 // q.nonceByBatchNum[batchInfo.BatchNum] = batchInfo.EthTx.Nonce()
400}
401
402// func (q *Queue) NonceByBatchNum(batchNum common.BatchNum) (uint64, bool) {
403// nonce, ok := q.nonceByBatchNum[batchNum]
404// return nonce, ok
405// }
406
407// Run the TxManager
408func (t *TxManager) Run(ctx context.Context) {
409 waitDuration := longWaitDuration
410
411 var statsVars statsVars
412 select {
413 case statsVars = <-t.statsVarsCh:
414 case <-ctx.Done():
415 }
416 t.stats = statsVars.Stats
417 t.syncSCVars(statsVars.Vars)
418 log.Infow("TxManager: received initial statsVars",
419 "block", t.stats.Eth.LastBlock.Num, "batch", t.stats.Eth.LastBatchNum)
420
421 for {
422 select {
423 case <-ctx.Done():
424 log.Info("TxManager done")
425 return
426 case statsVars := <-t.statsVarsCh:
427 t.stats = statsVars.Stats
428 t.syncSCVars(statsVars.Vars)
429 case pipelineNum := <-t.discardPipelineCh:
430 t.minPipelineNum = pipelineNum + 1
431 if err := t.removeBadBatchInfos(ctx); ctx.Err() != nil {
432 continue
433 } else if err != nil {
434 log.Errorw("TxManager: removeBadBatchInfos", "err", err)
435 continue
436 }
437 case batchInfo := <-t.batchCh:
438 if batchInfo.PipelineNum < t.minPipelineNum {
439 log.Warnw("TxManager: batchInfo received pipelineNum < minPipelineNum",
440 "num", batchInfo.PipelineNum, "minNum", t.minPipelineNum)
441 }
442 if err := t.shouldSendRollupForgeBatch(batchInfo); err != nil {
443 log.Warnw("TxManager: shouldSend", "err", err,
444 "batch", batchInfo.BatchNum)
445 t.coord.SendMsg(ctx, MsgStopPipeline{
446 Reason: fmt.Sprintf("forgeBatch shouldSend: %v", err)})
447 continue
448 }
449 if err := t.sendRollupForgeBatch(ctx, batchInfo, false); ctx.Err() != nil {
450 continue
451 } else if err != nil {
452 // If we reach here it's because our ethNode has
453 // been unable to send the transaction to
454 // ethereum. This could be due to the ethNode
455 // failure, or an invalid transaction (that
456 // can't be mined)
457 log.Warnw("TxManager: forgeBatch send failed", "err", err,
458 "batch", batchInfo.BatchNum)
459 t.coord.SendMsg(ctx, MsgStopPipeline{
460 Reason: fmt.Sprintf("forgeBatch send: %v", err)})
461 continue
462 }
463 t.queue.Push(batchInfo)
464 waitDuration = t.cfg.TxManagerCheckInterval
465 case <-time.After(waitDuration):
466 queuePosition, batchInfo := t.queue.Next()
467 if batchInfo == nil {
468 waitDuration = longWaitDuration
469 continue
470 }
471 if err := t.checkEthTransactionReceipt(ctx, batchInfo); ctx.Err() != nil {
472 continue
473 } else if err != nil { //nolint:staticcheck
474 // Our ethNode is giving an error different
475 // than "not found" when getting the receipt
476 // for the transaction, so we can't figure out
477 // if it was not mined, mined and succesfull or
478 // mined and failed. This could be due to the
479 // ethNode failure.
480 t.coord.SendMsg(ctx, MsgStopPipeline{
481 Reason: fmt.Sprintf("forgeBatch receipt: %v", err)})
482 }
483
484 confirm, err := t.handleReceipt(ctx, batchInfo)
485 if ctx.Err() != nil {
486 continue
487 } else if err != nil { //nolint:staticcheck
488 // Transaction was rejected
489 if err := t.removeBadBatchInfos(ctx); ctx.Err() != nil {
490 continue
491 } else if err != nil {
492 log.Errorw("TxManager: removeBadBatchInfos", "err", err)
493 continue
494 }
495 t.coord.SendMsg(ctx, MsgStopPipeline{
496 Reason: fmt.Sprintf("forgeBatch reject: %v", err)})
497 continue
498 }
499 now := time.Now()
500 if !t.cfg.EthNoReuseNonce && confirm == nil &&
501 now.Sub(batchInfo.SendTimestamp) > t.cfg.EthTxResendTimeout {
502 log.Infow("TxManager: forgeBatch tx not been mined timeout, resending",
503 "tx", batchInfo.EthTx.Hash(), "batch", batchInfo.BatchNum)
504 if err := t.sendRollupForgeBatch(ctx, batchInfo, true); ctx.Err() != nil {
505 continue
506 } else if err != nil {
507 // If we reach here it's because our ethNode has
508 // been unable to send the transaction to
509 // ethereum. This could be due to the ethNode
510 // failure, or an invalid transaction (that
511 // can't be mined)
512 log.Warnw("TxManager: forgeBatch resend failed", "err", err,
513 "batch", batchInfo.BatchNum)
514 t.coord.SendMsg(ctx, MsgStopPipeline{
515 Reason: fmt.Sprintf("forgeBatch resend: %v", err)})
516 continue
517 }
518 }
519
520 if confirm != nil && *confirm >= t.cfg.ConfirmBlocks {
521 log.Debugw("TxManager: forgeBatch tx confirmed",
522 "tx", batchInfo.EthTx.Hash(), "batch", batchInfo.BatchNum)
523 t.queue.Remove(queuePosition)
524 }
525 }
526 }
527}
528
529func (t *TxManager) removeBadBatchInfos(ctx context.Context) error {
530 next := 0
531 for {
532 batchInfo := t.queue.At(next)
533 if batchInfo == nil {
534 break
535 }
536 if err := t.checkEthTransactionReceipt(ctx, batchInfo); ctx.Err() != nil {
537 return nil
538 } else if err != nil {
539 // Our ethNode is giving an error different
540 // than "not found" when getting the receipt
541 // for the transaction, so we can't figure out
542 // if it was not mined, mined and succesfull or
543 // mined and failed. This could be due to the
544 // ethNode failure.
545 next++
546 continue
547 }
548 confirm, err := t.handleReceipt(ctx, batchInfo)
549 if ctx.Err() != nil {
550 return nil
551 } else if err != nil {
552 // Transaction was rejected
553 if t.minPipelineNum <= batchInfo.PipelineNum {
554 t.minPipelineNum = batchInfo.PipelineNum + 1
555 }
556 t.queue.Remove(next)
557 continue
558 }
559 // If tx is pending but is from a cancelled pipeline, remove it
560 // from the queue
561 if confirm == nil {
562 if batchInfo.PipelineNum < t.minPipelineNum {
563 t.queue.Remove(next)
564 continue
565 }
566 }
567 next++
568 }
569 accNonce, err := t.ethClient.EthNonceAt(ctx, t.account.Address, nil)
570 if err != nil {
571 return err
572 }
573 if !t.cfg.EthNoReuseNonce {
574 t.accNextNonce = accNonce
575 }
576 return nil
577}
578
579func (t *TxManager) canForgeAt(blockNum int64) bool {
580 return canForge(&t.consts.Auction, &t.vars.Auction,
581 &t.stats.Sync.Auction.CurrentSlot, &t.stats.Sync.Auction.NextSlot,
582 t.cfg.ForgerAddress, blockNum)
583}
584
585func (t *TxManager) mustL1L2Batch(blockNum int64) bool {
586 lastL1BatchBlockNum := t.lastSentL1BatchBlockNum
587 if t.stats.Sync.LastL1BatchBlock > lastL1BatchBlockNum {
588 lastL1BatchBlockNum = t.stats.Sync.LastL1BatchBlock
589 }
590 return blockNum-lastL1BatchBlockNum >= t.vars.Rollup.ForgeL1L2BatchTimeout-1
591}