hermez-node/coordinator/txmanager.go
coordinator/txmanager.goRaw
1package coordinator
2
3import (
4 "context"
5 "fmt"
6 "math/big"
7 "strings"
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, batchInfo *BatchInfo) (*bind.TransactOpts, error) {
127 gasPrice, err := t.ethClient.EthSuggestGasPrice(ctx)
128 if err != nil {
129 return nil, tracerr.Wrap(err)
130 }
131 if t.cfg.GasPriceIncPerc != 0 {
132 inc := new(big.Int).Set(gasPrice)
133 inc.Mul(inc, new(big.Int).SetInt64(t.cfg.GasPriceIncPerc))
134 // nolint reason: to calculate percentages we use 100
135 inc.Div(inc, new(big.Int).SetUint64(100)) //nolint:gomnd
136 gasPrice.Add(gasPrice, inc)
137 }
138
139 // log.Debugw("TxManager: transaction metadata", "gasPrice", gasPrice)
140
141 auth, err := bind.NewKeyStoreTransactorWithChainID(t.ethClient.EthKeyStore(), t.account, t.chainID)
142 if err != nil {
143 return nil, tracerr.Wrap(err)
144 }
145 auth.Value = big.NewInt(0) // in wei
146
147 gasLimit := t.cfg.ForgeBatchGasCost.Fixed +
148 uint64(len(batchInfo.L1UserTxsExtra))*t.cfg.ForgeBatchGasCost.L1UserTx +
149 uint64(len(batchInfo.L1CoordTxs))*t.cfg.ForgeBatchGasCost.L1CoordTx +
150 uint64(len(batchInfo.L2Txs))*t.cfg.ForgeBatchGasCost.L2Tx
151 auth.GasLimit = gasLimit
152 auth.GasPrice = gasPrice
153 auth.Nonce = nil
154
155 return auth, nil
156}
157
158func (t *TxManager) shouldSendRollupForgeBatch(batchInfo *BatchInfo) error {
159 nextBlock := t.stats.Eth.LastBlock.Num + 1
160 if !t.canForgeAt(nextBlock) {
161 return tracerr.Wrap(fmt.Errorf("can't forge in the next block: %v", nextBlock))
162 }
163 if t.mustL1L2Batch(nextBlock) && !batchInfo.L1Batch {
164 return tracerr.Wrap(fmt.Errorf("can't forge non-L1Batch in the next block: %v", nextBlock))
165 }
166 margin := t.cfg.SendBatchBlocksMarginCheck
167 if margin != 0 {
168 if !t.canForgeAt(nextBlock + margin) {
169 return tracerr.Wrap(fmt.Errorf("can't forge after %v blocks: %v",
170 margin, nextBlock))
171 }
172 if t.mustL1L2Batch(nextBlock+margin) && !batchInfo.L1Batch {
173 return tracerr.Wrap(fmt.Errorf("can't forge non-L1Batch after %v blocks: %v",
174 margin, nextBlock))
175 }
176 }
177 return nil
178}
179
180func addPerc(v *big.Int, p int64) *big.Int {
181 r := new(big.Int).Set(v)
182 r.Mul(r, big.NewInt(p))
183 // nolint reason: to calculate percentages we divide by 100
184 r.Div(r, big.NewInt(100)) //nolit:gomnd
185 // If the increase is 0, force it to be 1 so that a gas increase
186 // doesn't result in the same value, making the transaction to be equal
187 // than before.
188 if r.Cmp(big.NewInt(0)) == 0 {
189 r = big.NewInt(1)
190 }
191 return r.Add(v, r)
192}
193
194func (t *TxManager) sendRollupForgeBatch(ctx context.Context, batchInfo *BatchInfo, resend bool) error {
195 var ethTx *types.Transaction
196 var err error
197 var auth *bind.TransactOpts
198 if resend {
199 auth = batchInfo.Auth
200 auth.GasPrice = addPerc(auth.GasPrice, 10)
201 } else {
202 auth, err = t.NewAuth(ctx, batchInfo)
203 if err != nil {
204 return tracerr.Wrap(err)
205 }
206 batchInfo.Auth = auth
207 auth.Nonce = big.NewInt(int64(t.accNextNonce))
208 }
209 for attempt := 0; attempt < t.cfg.EthClientAttempts; attempt++ {
210 if auth.GasPrice.Cmp(t.cfg.MaxGasPrice) > 0 {
211 return tracerr.Wrap(fmt.Errorf("calculated gasPrice (%v) > maxGasPrice (%v)",
212 auth.GasPrice, t.cfg.MaxGasPrice))
213 }
214 // RollupForgeBatch() calls ethclient.SendTransaction()
215 ethTx, err = t.ethClient.RollupForgeBatch(batchInfo.ForgeBatchArgs, auth)
216 // We check the errors via strings because we match the
217 // definition of the error from geth, with the string returned
218 // via RPC obtained by the client.
219 if err == nil {
220 break
221 } else if strings.Contains(err.Error(), core.ErrNonceTooLow.Error()) {
222 log.Warnw("TxManager ethClient.RollupForgeBatch incrementing nonce",
223 "err", err, "nonce", auth.Nonce, "batchNum", batchInfo.BatchNum)
224 auth.Nonce.Add(auth.Nonce, big.NewInt(1))
225 attempt--
226 } else if strings.Contains(err.Error(), core.ErrNonceTooHigh.Error()) {
227 log.Warnw("TxManager ethClient.RollupForgeBatch decrementing nonce",
228 "err", err, "nonce", auth.Nonce, "batchNum", batchInfo.BatchNum)
229 auth.Nonce.Sub(auth.Nonce, big.NewInt(1))
230 attempt--
231 } else if strings.Contains(err.Error(), core.ErrReplaceUnderpriced.Error()) {
232 log.Warnw("TxManager ethClient.RollupForgeBatch incrementing gasPrice",
233 "err", err, "gasPrice", auth.GasPrice, "batchNum", batchInfo.BatchNum)
234 auth.GasPrice = addPerc(auth.GasPrice, 10)
235 attempt--
236 } else if strings.Contains(err.Error(), core.ErrUnderpriced.Error()) {
237 log.Warnw("TxManager ethClient.RollupForgeBatch incrementing gasPrice",
238 "err", err, "gasPrice", auth.GasPrice, "batchNum", batchInfo.BatchNum)
239 auth.GasPrice = addPerc(auth.GasPrice, 10)
240 attempt--
241 } else {
242 log.Errorw("TxManager ethClient.RollupForgeBatch",
243 "attempt", attempt, "err", err, "block", t.stats.Eth.LastBlock.Num+1,
244 "batchNum", batchInfo.BatchNum)
245 }
246 select {
247 case <-ctx.Done():
248 return tracerr.Wrap(common.ErrDone)
249 case <-time.After(t.cfg.EthClientAttemptsDelay):
250 }
251 }
252 if err != nil {
253 return tracerr.Wrap(fmt.Errorf("reached max attempts for ethClient.RollupForgeBatch: %w", err))
254 }
255 if !resend {
256 t.accNextNonce = auth.Nonce.Uint64() + 1
257 }
258 batchInfo.EthTx = ethTx
259 log.Infow("TxManager ethClient.RollupForgeBatch", "batch", batchInfo.BatchNum, "tx", ethTx.Hash())
260 now := time.Now()
261 batchInfo.SendTimestamp = now
262
263 if resend {
264 batchInfo.Debug.ResendNum++
265 }
266 batchInfo.Debug.Status = StatusSent
267 batchInfo.Debug.SendBlockNum = t.stats.Eth.LastBlock.Num + 1
268 batchInfo.Debug.SendTimestamp = batchInfo.SendTimestamp
269 batchInfo.Debug.StartToSendDelay = batchInfo.Debug.SendTimestamp.Sub(
270 batchInfo.Debug.StartTimestamp).Seconds()
271 t.cfg.debugBatchStore(batchInfo)
272
273 if !resend {
274 if batchInfo.L1Batch {
275 t.lastSentL1BatchBlockNum = t.stats.Eth.LastBlock.Num + 1
276 }
277 }
278 if err := t.l2DB.DoneForging(common.TxIDsFromL2Txs(batchInfo.L2Txs), batchInfo.BatchNum); err != nil {
279 return tracerr.Wrap(err)
280 }
281 return nil
282}
283
284// checkEthTransactionReceipt takes the txHash from the BatchInfo and stores
285// the corresponding receipt if found
286func (t *TxManager) checkEthTransactionReceipt(ctx context.Context, batchInfo *BatchInfo) error {
287 txHash := batchInfo.EthTx.Hash()
288 var receipt *types.Receipt
289 var err error
290 for attempt := 0; attempt < t.cfg.EthClientAttempts; attempt++ {
291 receipt, err = t.ethClient.EthTransactionReceipt(ctx, txHash)
292 if ctx.Err() != nil {
293 continue
294 } else if tracerr.Unwrap(err) == ethereum.NotFound {
295 err = nil
296 break
297 } else if err != nil {
298 log.Errorw("TxManager ethClient.EthTransactionReceipt",
299 "attempt", attempt, "err", err)
300 } else {
301 break
302 }
303 select {
304 case <-ctx.Done():
305 return tracerr.Wrap(common.ErrDone)
306 case <-time.After(t.cfg.EthClientAttemptsDelay):
307 }
308 }
309 if err != nil {
310 return tracerr.Wrap(fmt.Errorf("reached max attempts for ethClient.EthTransactionReceipt: %w", err))
311 }
312 batchInfo.Receipt = receipt
313 t.cfg.debugBatchStore(batchInfo)
314 return nil
315}
316
317func (t *TxManager) handleReceipt(ctx context.Context, batchInfo *BatchInfo) (*int64, error) {
318 receipt := batchInfo.Receipt
319 if receipt != nil {
320 if batchInfo.EthTx.Nonce()+1 > t.accNonce {
321 t.accNonce = batchInfo.EthTx.Nonce() + 1
322 }
323 if receipt.Status == types.ReceiptStatusFailed {
324 batchInfo.Debug.Status = StatusFailed
325 _, err := t.ethClient.EthCall(ctx, batchInfo.EthTx, receipt.BlockNumber)
326 log.Warnw("TxManager receipt status is failed", "tx", receipt.TxHash,
327 "batch", batchInfo.BatchNum, "block", receipt.BlockNumber.Int64(),
328 "err", err)
329 batchInfo.EthTxErr = err
330 if batchInfo.BatchNum <= t.lastSuccessBatch {
331 t.lastSuccessBatch = batchInfo.BatchNum - 1
332 }
333 t.cfg.debugBatchStore(batchInfo)
334 return nil, tracerr.Wrap(fmt.Errorf(
335 "ethereum transaction receipt status is failed: %w", err))
336 } else if receipt.Status == types.ReceiptStatusSuccessful {
337 batchInfo.Debug.Status = StatusMined
338 batchInfo.Debug.MineBlockNum = receipt.BlockNumber.Int64()
339 batchInfo.Debug.StartToMineBlocksDelay = batchInfo.Debug.MineBlockNum -
340 batchInfo.Debug.StartBlockNum
341 if batchInfo.Debug.StartToMineDelay == 0 {
342 if block, err := t.ethClient.EthBlockByNumber(ctx,
343 receipt.BlockNumber.Int64()); err != nil {
344 log.Warnw("TxManager: ethClient.EthBlockByNumber", "err", err)
345 } else {
346 batchInfo.Debug.SendToMineDelay = block.Timestamp.Sub(
347 batchInfo.Debug.SendTimestamp).Seconds()
348 batchInfo.Debug.StartToMineDelay = block.Timestamp.Sub(
349 batchInfo.Debug.StartTimestamp).Seconds()
350 }
351 }
352 t.cfg.debugBatchStore(batchInfo)
353 if batchInfo.BatchNum > t.lastSuccessBatch {
354 t.lastSuccessBatch = batchInfo.BatchNum
355 }
356 confirm := t.stats.Eth.LastBlock.Num - receipt.BlockNumber.Int64()
357 return &confirm, nil
358 }
359 }
360 return nil, nil
361}
362
363// TODO:
364// - After sending a message: CancelPipeline, stop all consecutive pending Batches (transactions)
365
366// Queue of BatchInfos
367type Queue struct {
368 list []*BatchInfo
369 // nonceByBatchNum map[common.BatchNum]uint64
370 next int
371}
372
373// NewQueue returns a new queue
374func NewQueue() Queue {
375 return Queue{
376 list: make([]*BatchInfo, 0),
377 // nonceByBatchNum: make(map[common.BatchNum]uint64),
378 next: 0,
379 }
380}
381
382// Len is the length of the queue
383func (q *Queue) Len() int {
384 return len(q.list)
385}
386
387// At returns the BatchInfo at position (or nil if position is out of bounds)
388func (q *Queue) At(position int) *BatchInfo {
389 if position >= len(q.list) {
390 return nil
391 }
392 return q.list[position]
393}
394
395// Next returns the next BatchInfo (or nil if queue is empty)
396func (q *Queue) Next() (int, *BatchInfo) {
397 if len(q.list) == 0 {
398 return 0, nil
399 }
400 defer func() { q.next = (q.next + 1) % len(q.list) }()
401 return q.next, q.list[q.next]
402}
403
404// Remove removes the BatchInfo at position
405func (q *Queue) Remove(position int) {
406 // batchInfo := q.list[position]
407 // delete(q.nonceByBatchNum, batchInfo.BatchNum)
408 q.list = append(q.list[:position], q.list[position+1:]...)
409 if len(q.list) == 0 {
410 q.next = 0
411 } else {
412 q.next = position % len(q.list)
413 }
414}
415
416// Push adds a new BatchInfo
417func (q *Queue) Push(batchInfo *BatchInfo) {
418 q.list = append(q.list, batchInfo)
419 // q.nonceByBatchNum[batchInfo.BatchNum] = batchInfo.EthTx.Nonce()
420}
421
422// func (q *Queue) NonceByBatchNum(batchNum common.BatchNum) (uint64, bool) {
423// nonce, ok := q.nonceByBatchNum[batchNum]
424// return nonce, ok
425// }
426
427// Run the TxManager
428func (t *TxManager) Run(ctx context.Context) {
429 var statsVars statsVars
430 select {
431 case statsVars = <-t.statsVarsCh:
432 case <-ctx.Done():
433 }
434 t.stats = statsVars.Stats
435 t.syncSCVars(statsVars.Vars)
436 log.Infow("TxManager: received initial statsVars",
437 "block", t.stats.Eth.LastBlock.Num, "batch", t.stats.Eth.LastBatchNum)
438
439 timer := time.NewTimer(longWaitDuration)
440 for {
441 select {
442 case <-ctx.Done():
443 log.Info("TxManager done")
444 return
445 case statsVars := <-t.statsVarsCh:
446 t.stats = statsVars.Stats
447 t.syncSCVars(statsVars.Vars)
448 case pipelineNum := <-t.discardPipelineCh:
449 t.minPipelineNum = pipelineNum + 1
450 if err := t.removeBadBatchInfos(ctx); ctx.Err() != nil {
451 continue
452 } else if err != nil {
453 log.Errorw("TxManager: removeBadBatchInfos", "err", err)
454 continue
455 }
456 case batchInfo := <-t.batchCh:
457 if batchInfo.PipelineNum < t.minPipelineNum {
458 log.Warnw("TxManager: batchInfo received pipelineNum < minPipelineNum",
459 "num", batchInfo.PipelineNum, "minNum", t.minPipelineNum)
460 }
461 if err := t.shouldSendRollupForgeBatch(batchInfo); err != nil {
462 log.Warnw("TxManager: shouldSend", "err", err,
463 "batch", batchInfo.BatchNum)
464 t.coord.SendMsg(ctx, MsgStopPipeline{
465 Reason: fmt.Sprintf("forgeBatch shouldSend: %v", err)})
466 continue
467 }
468 if err := t.sendRollupForgeBatch(ctx, batchInfo, false); ctx.Err() != nil {
469 continue
470 } else if err != nil {
471 // If we reach here it's because our ethNode has
472 // been unable to send the transaction to
473 // ethereum. This could be due to the ethNode
474 // failure, or an invalid transaction (that
475 // can't be mined)
476 log.Warnw("TxManager: forgeBatch send failed", "err", err,
477 "batch", batchInfo.BatchNum)
478 t.coord.SendMsg(ctx, MsgStopPipeline{
479 Reason: fmt.Sprintf("forgeBatch send: %v", err)})
480 continue
481 }
482 t.queue.Push(batchInfo)
483 if !timer.Stop() {
484 <-timer.C
485 }
486 timer.Reset(t.cfg.TxManagerCheckInterval)
487 case <-timer.C:
488 queuePosition, batchInfo := t.queue.Next()
489 if batchInfo == nil {
490 timer.Reset(longWaitDuration)
491 continue
492 }
493 timer.Reset(t.cfg.TxManagerCheckInterval)
494 if err := t.checkEthTransactionReceipt(ctx, batchInfo); ctx.Err() != nil {
495 continue
496 } else if err != nil { //nolint:staticcheck
497 // Our ethNode is giving an error different
498 // than "not found" when getting the receipt
499 // for the transaction, so we can't figure out
500 // if it was not mined, mined and succesfull or
501 // mined and failed. This could be due to the
502 // ethNode failure.
503 t.coord.SendMsg(ctx, MsgStopPipeline{
504 Reason: fmt.Sprintf("forgeBatch receipt: %v", err)})
505 }
506
507 confirm, err := t.handleReceipt(ctx, batchInfo)
508 if ctx.Err() != nil {
509 continue
510 } else if err != nil { //nolint:staticcheck
511 // Transaction was rejected
512 if err := t.removeBadBatchInfos(ctx); ctx.Err() != nil {
513 continue
514 } else if err != nil {
515 log.Errorw("TxManager: removeBadBatchInfos", "err", err)
516 continue
517 }
518 t.coord.SendMsg(ctx, MsgStopPipeline{
519 Reason: fmt.Sprintf("forgeBatch reject: %v", err)})
520 continue
521 }
522 now := time.Now()
523 if !t.cfg.EthNoReuseNonce && confirm == nil &&
524 now.Sub(batchInfo.SendTimestamp) > t.cfg.EthTxResendTimeout {
525 log.Infow("TxManager: forgeBatch tx not been mined timeout, resending",
526 "tx", batchInfo.EthTx.Hash(), "batch", batchInfo.BatchNum)
527 if err := t.sendRollupForgeBatch(ctx, batchInfo, true); ctx.Err() != nil {
528 continue
529 } else if err != nil {
530 // If we reach here it's because our ethNode has
531 // been unable to send the transaction to
532 // ethereum. This could be due to the ethNode
533 // failure, or an invalid transaction (that
534 // can't be mined)
535 log.Warnw("TxManager: forgeBatch resend failed", "err", err,
536 "batch", batchInfo.BatchNum)
537 t.coord.SendMsg(ctx, MsgStopPipeline{
538 Reason: fmt.Sprintf("forgeBatch resend: %v", err)})
539 continue
540 }
541 }
542
543 if confirm != nil && *confirm >= t.cfg.ConfirmBlocks {
544 log.Debugw("TxManager: forgeBatch tx confirmed",
545 "tx", batchInfo.EthTx.Hash(), "batch", batchInfo.BatchNum)
546 t.queue.Remove(queuePosition)
547 }
548 }
549 }
550}
551
552func (t *TxManager) removeBadBatchInfos(ctx context.Context) error {
553 next := 0
554 for {
555 batchInfo := t.queue.At(next)
556 if batchInfo == nil {
557 break
558 }
559 if err := t.checkEthTransactionReceipt(ctx, batchInfo); ctx.Err() != nil {
560 return nil
561 } else if err != nil {
562 // Our ethNode is giving an error different
563 // than "not found" when getting the receipt
564 // for the transaction, so we can't figure out
565 // if it was not mined, mined and succesfull or
566 // mined and failed. This could be due to the
567 // ethNode failure.
568 next++
569 continue
570 }
571 confirm, err := t.handleReceipt(ctx, batchInfo)
572 if ctx.Err() != nil {
573 return nil
574 } else if err != nil {
575 // Transaction was rejected
576 if t.minPipelineNum <= batchInfo.PipelineNum {
577 t.minPipelineNum = batchInfo.PipelineNum + 1
578 }
579 t.queue.Remove(next)
580 continue
581 }
582 // If tx is pending but is from a cancelled pipeline, remove it
583 // from the queue
584 if confirm == nil {
585 if batchInfo.PipelineNum < t.minPipelineNum {
586 t.queue.Remove(next)
587 continue
588 }
589 }
590 next++
591 }
592 accNonce, err := t.ethClient.EthNonceAt(ctx, t.account.Address, nil)
593 if err != nil {
594 return err
595 }
596 if !t.cfg.EthNoReuseNonce {
597 t.accNextNonce = accNonce
598 }
599 return nil
600}
601
602func (t *TxManager) canForgeAt(blockNum int64) bool {
603 return canForge(&t.consts.Auction, &t.vars.Auction,
604 &t.stats.Sync.Auction.CurrentSlot, &t.stats.Sync.Auction.NextSlot,
605 t.cfg.ForgerAddress, blockNum)
606}
607
608func (t *TxManager) mustL1L2Batch(blockNum int64) bool {
609 lastL1BatchBlockNum := t.lastSentL1BatchBlockNum
610 if t.stats.Sync.LastL1BatchBlock > lastL1BatchBlockNum {
611 lastL1BatchBlockNum = t.stats.Sync.LastL1BatchBlock
612 }
613 return blockNum-lastL1BatchBlockNum >= t.vars.Rollup.ForgeL1L2BatchTimeout-1
614}