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