hermez-node/coordinator/coordinator.go
coordinator/coordinator.goRaw
1package coordinator
2
3import (
4 "context"
5 "fmt"
6 "sync"
7 "time"
8
9 ethCommon "github.com/ethereum/go-ethereum/common"
10 "github.com/ethereum/go-ethereum/core/types"
11 "github.com/hermeznetwork/hermez-node/batchbuilder"
12 "github.com/hermeznetwork/hermez-node/common"
13 "github.com/hermeznetwork/hermez-node/db/historydb"
14 "github.com/hermeznetwork/hermez-node/eth"
15 "github.com/hermeznetwork/hermez-node/log"
16 "github.com/hermeznetwork/hermez-node/synchronizer"
17 "github.com/hermeznetwork/hermez-node/txselector"
18 "github.com/hermeznetwork/tracerr"
19)
20
21var errTODO = fmt.Errorf("TODO")
22
23// ErrDone is returned when the function is stopped asynchronously via a done
24// (terminated) context. It doesn't indicate an error.
25var ErrDone = fmt.Errorf("done")
26
27// Config contains the Coordinator configuration
28type Config struct {
29 // ForgerAddress is the address under which this coordinator is forging
30 ForgerAddress ethCommon.Address
31 // ConfirmBlocks is the number of confirmation blocks to wait for sent
32 // ethereum transactions before forgetting about them
33 ConfirmBlocks int64
34 // L1BatchTimeoutPerc is the portion of the range before the L1Batch
35 // timeout that will trigger a schedule to forge an L1Batch
36 L1BatchTimeoutPerc float64
37 // EthClientAttempts is the number of attempts to do an eth client RPC
38 // call before giving up
39 EthClientAttempts int
40 // EthClientAttemptsDelay is delay between attempts do do an eth client
41 // RPC call
42 EthClientAttemptsDelay time.Duration
43 // TxManagerCheckInterval is the waiting interval between receipt
44 // checks of ethereum transactions in the TxManager
45 TxManagerCheckInterval time.Duration
46 // DebugBatchPath if set, specifies the path where batchInfo is stored
47 // in JSON in every step/update of the pipeline
48 DebugBatchPath string
49}
50
51func (c *Config) debugBatchStore(batchInfo *BatchInfo) {
52 if c.DebugBatchPath != "" {
53 if err := batchInfo.DebugStore(c.DebugBatchPath); err != nil {
54 log.Warnw("Error storing debug BatchInfo",
55 "path", c.DebugBatchPath, "err", err)
56 }
57 }
58}
59
60// Coordinator implements the Coordinator type
61type Coordinator struct {
62 // State
63 batchNum common.BatchNum
64 serverProofs []ServerProofInterface
65 consts synchronizer.SCConsts
66 vars synchronizer.SCVariables
67 started bool
68
69 cfg Config
70
71 historyDB *historydb.HistoryDB
72 txSelector *txselector.TxSelector
73 batchBuilder *batchbuilder.BatchBuilder
74
75 msgCh chan interface{}
76 ctx context.Context
77 wg sync.WaitGroup
78 cancel context.CancelFunc
79
80 pipeline *Pipeline
81
82 txManager *TxManager
83}
84
85// NewCoordinator creates a new Coordinator
86func NewCoordinator(cfg Config,
87 historyDB *historydb.HistoryDB,
88 txSelector *txselector.TxSelector,
89 batchBuilder *batchbuilder.BatchBuilder,
90 serverProofs []ServerProofInterface,
91 ethClient eth.ClientInterface,
92 scConsts *synchronizer.SCConsts,
93 initSCVars *synchronizer.SCVariables,
94) (*Coordinator, error) {
95 // nolint reason: hardcoded `1.0`, by design the percentage can't be over 100%
96 if cfg.L1BatchTimeoutPerc >= 1.0 { //nolint:gomnd
97 return nil, tracerr.Wrap(fmt.Errorf("invalid value for Config.L1BatchTimeoutPerc (%v >= 1.0)",
98 cfg.L1BatchTimeoutPerc))
99 }
100 if cfg.EthClientAttempts < 1 {
101 return nil, tracerr.Wrap(fmt.Errorf("invalid value for Config.EthClientAttempts (%v < 1)",
102 cfg.EthClientAttempts))
103 }
104
105 txManager := NewTxManager(&cfg, ethClient)
106
107 ctx, cancel := context.WithCancel(context.Background())
108 c := Coordinator{
109 batchNum: -1,
110 serverProofs: serverProofs,
111 consts: *scConsts,
112 vars: *initSCVars,
113
114 cfg: cfg,
115
116 historyDB: historyDB,
117 txSelector: txSelector,
118 batchBuilder: batchBuilder,
119
120 // ethClient: ethClient,
121
122 msgCh: make(chan interface{}),
123 ctx: ctx,
124 // wg
125 cancel: cancel,
126
127 txManager: txManager,
128 }
129 return &c, nil
130}
131
132func (c *Coordinator) newPipeline() *Pipeline {
133 return NewPipeline(c.cfg, c.historyDB, c.txSelector, c.batchBuilder,
134 c.txManager, c.serverProofs, &c.consts)
135}
136
137// MsgSyncStats indicates an update to the Synchronizer stats
138type MsgSyncStats struct {
139 Stats synchronizer.Stats
140}
141
142// MsgSyncSCVars indicates an update to Smart Contract Vars
143type MsgSyncSCVars struct {
144 Rollup *common.RollupVariables
145 Auction *common.AuctionVariables
146 WDelayer *common.WDelayerVariables
147}
148
149// MsgSyncReorg indicates a reorg
150type MsgSyncReorg struct {
151}
152
153// SendMsg is a thread safe method to pass a message to the Coordinator
154func (c *Coordinator) SendMsg(msg interface{}) {
155 c.msgCh <- msg
156}
157
158func (c *Coordinator) handleMsgSyncSCVars(msg *MsgSyncSCVars) {
159 if msg.Rollup != nil {
160 c.vars.Rollup = *msg.Rollup
161 }
162 if msg.Auction != nil {
163 c.vars.Auction = *msg.Auction
164 }
165 if msg.WDelayer != nil {
166 c.vars.WDelayer = *msg.WDelayer
167 }
168}
169
170func (c *Coordinator) canForge(stats *synchronizer.Stats) bool {
171 anyoneForge := false
172 if stats.Sync.Auction.CurrentSlot.BatchesLen == 0 &&
173 c.consts.Auction.RelativeBlock(stats.Eth.LastBlock.Num+1) > int64(c.vars.Auction.SlotDeadline) {
174 log.Debug("Coordinator: anyone can forge in the current slot (slotDeadline passed)")
175 anyoneForge = true
176 }
177 if stats.Sync.Auction.CurrentSlot.Forger == c.cfg.ForgerAddress || anyoneForge {
178 return true
179 }
180 return false
181}
182
183func (c *Coordinator) handleMsgSyncStats(stats *synchronizer.Stats) error {
184 if !stats.Synced() {
185 return nil
186 }
187 c.txManager.SetLastBlock(stats.Eth.LastBlock.Num)
188
189 canForge := c.canForge(stats)
190 if c.pipeline == nil {
191 if canForge {
192 log.Info("Coordinator: forging state begin")
193 batchNum := common.BatchNum(stats.Sync.LastBatch)
194 c.pipeline = c.newPipeline()
195 if err := c.pipeline.Start(batchNum, stats, &c.vars); err != nil {
196 return tracerr.Wrap(err)
197 }
198 }
199 } else {
200 if canForge {
201 c.pipeline.SetSyncStats(stats)
202 } else {
203 log.Info("Coordinator: forging state end")
204 c.pipeline.Stop()
205 c.pipeline = nil
206 }
207 }
208 return nil
209}
210
211// Start the coordinator
212func (c *Coordinator) Start() {
213 if c.started {
214 log.Fatal("Coordinator already started")
215 }
216 c.started = true
217 c.wg.Add(1)
218 go func() {
219 c.txManager.Run(c.ctx)
220 c.wg.Done()
221 }()
222
223 c.wg.Add(1)
224 go func() {
225 for {
226 select {
227 case <-c.ctx.Done():
228 log.Info("Coordinator done")
229 c.wg.Done()
230 return
231 case msg := <-c.msgCh:
232 switch msg := msg.(type) {
233 case MsgSyncStats:
234 stats := msg.Stats
235 if err := c.handleMsgSyncStats(&stats); err != nil {
236 log.Errorw("Coordinator.handleMsgSyncStats error", "err", err)
237 continue
238 }
239 case MsgSyncReorg:
240 if err := c.handleReorg(); err != nil {
241 log.Errorw("Coordinator.handleReorg error", "err", err)
242 continue
243 }
244 case MsgSyncSCVars:
245 c.handleMsgSyncSCVars(&msg)
246 default:
247 log.Fatalw("Coordinator Unexpected Coordinator msg of type %T: %+v", msg, msg)
248 }
249 }
250 }
251 }()
252}
253
254// Stop the coordinator
255func (c *Coordinator) Stop() {
256 if !c.started {
257 log.Fatal("Coordinator already stopped")
258 }
259 c.started = false
260 log.Infow("Stopping Coordinator...")
261 c.cancel()
262 c.wg.Wait()
263 if c.pipeline != nil {
264 c.pipeline.Stop()
265 c.pipeline = nil
266 }
267}
268
269func (c *Coordinator) handleReorg() error {
270 return nil // TODO
271}
272
273// TxManager handles everything related to ethereum transactions: It makes the
274// call to forge, waits for transaction confirmation, and keeps checking them
275// until a number of confirmed blocks have passed.
276type TxManager struct {
277 cfg Config
278 ethClient eth.ClientInterface
279 batchCh chan *BatchInfo
280 lastBlockCh chan int64
281 queue []*BatchInfo
282 lastBlock int64
283}
284
285// NewTxManager creates a new TxManager
286func NewTxManager(cfg *Config, ethClient eth.ClientInterface) *TxManager {
287 return &TxManager{
288 cfg: *cfg,
289 ethClient: ethClient,
290 // TODO: Find best queue size
291 batchCh: make(chan *BatchInfo, 16), //nolint:gomnd
292 // TODO: Find best queue size
293 lastBlockCh: make(chan int64, 16), //nolint:gomnd
294 lastBlock: -1,
295 }
296}
297
298// AddBatch is a thread safe method to pass a new batch TxManager to be sent to
299// the smart contract via the forge call
300func (t *TxManager) AddBatch(batchInfo *BatchInfo) {
301 t.batchCh <- batchInfo
302}
303
304// SetLastBlock is a thread safe method to pass the lastBlock to the TxManager
305func (t *TxManager) SetLastBlock(lastBlock int64) {
306 t.lastBlockCh <- lastBlock
307}
308
309func (t *TxManager) rollupForgeBatch(ctx context.Context, batchInfo *BatchInfo) error {
310 var ethTx *types.Transaction
311 var err error
312 for attempt := 0; attempt < t.cfg.EthClientAttempts; attempt++ {
313 ethTx, err = t.ethClient.RollupForgeBatch(batchInfo.ForgeBatchArgs)
314 if err != nil {
315 log.Errorw("TxManager ethClient.RollupForgeBatch",
316 "attempt", attempt, "err", err)
317 } else {
318 break
319 }
320 select {
321 case <-ctx.Done():
322 return tracerr.Wrap(ErrDone)
323 case <-time.After(t.cfg.EthClientAttemptsDelay):
324 }
325 }
326 if err != nil {
327 return tracerr.Wrap(fmt.Errorf("reached max attempts for ethClient.RollupForgeBatch: %w", err))
328 }
329 batchInfo.EthTx = ethTx
330 t.cfg.debugBatchStore(batchInfo)
331 return nil
332}
333
334func (t *TxManager) ethTransactionReceipt(ctx context.Context, batchInfo *BatchInfo) error {
335 txHash := batchInfo.EthTx.Hash()
336 var receipt *types.Receipt
337 var err error
338 for attempt := 0; attempt < t.cfg.EthClientAttempts; attempt++ {
339 receipt, err = t.ethClient.EthTransactionReceipt(ctx, txHash)
340 if err != nil {
341 log.Errorw("TxManager ethClient.EthTransactionReceipt",
342 "attempt", attempt, "err", err)
343 } else {
344 break
345 }
346 select {
347 case <-ctx.Done():
348 return tracerr.Wrap(ErrDone)
349 case <-time.After(t.cfg.EthClientAttemptsDelay):
350 }
351 }
352 if err != nil {
353 return tracerr.Wrap(fmt.Errorf("reached max attempts for ethClient.EthTransactionReceipt: %w", err))
354 }
355 batchInfo.Receipt = receipt
356 t.cfg.debugBatchStore(batchInfo)
357 return nil
358}
359
360func (t *TxManager) handleReceipt(batchInfo *BatchInfo) (*int64, error) {
361 receipt := batchInfo.Receipt
362 if receipt != nil {
363 if receipt.Status == types.ReceiptStatusFailed {
364 log.Errorw("TxManager receipt status is failed", "receipt", receipt)
365 return nil, tracerr.Wrap(fmt.Errorf("ethereum transaction receipt statis is failed"))
366 } else if receipt.Status == types.ReceiptStatusSuccessful {
367 confirm := t.lastBlock - receipt.BlockNumber.Int64()
368 return &confirm, nil
369 }
370 }
371 return nil, nil
372}
373
374const longWaitTime = 999 * time.Hour
375
376// Run the TxManager
377func (t *TxManager) Run(ctx context.Context) {
378 next := 0
379 waitTime := time.Duration(longWaitTime)
380 for {
381 select {
382 case <-ctx.Done():
383 log.Info("TxManager done")
384 return
385 case lastBlock := <-t.lastBlockCh:
386 t.lastBlock = lastBlock
387 case batchInfo := <-t.batchCh:
388 if err := t.rollupForgeBatch(ctx, batchInfo); tracerr.Unwrap(err) == ErrDone {
389 continue
390 } else if err != nil {
391 // TODO: Reset pipeline
392 continue
393 }
394 log.Debugf("ethClient ForgeCall sent, batchNum: %d", batchInfo.BatchNum)
395 t.queue = append(t.queue, batchInfo)
396 waitTime = t.cfg.TxManagerCheckInterval
397 case <-time.After(waitTime):
398 if len(t.queue) == 0 {
399 continue
400 }
401 batchInfo := t.queue[next]
402 err := t.ethTransactionReceipt(ctx, batchInfo)
403 if tracerr.Unwrap(err) == ErrDone {
404 continue
405 } else if err != nil { //nolint:staticcheck
406 // We can't get the receipt for the
407 // transaction, so we can't confirm if it was
408 // mined
409 // TODO: Reset pipeline
410 }
411
412 confirm, err := t.handleReceipt(batchInfo)
413 if err != nil { //nolint:staticcheck
414 // Transaction was rejected
415 // TODO: Reset pipeline
416 }
417 if confirm != nil && *confirm >= t.cfg.ConfirmBlocks {
418 log.Debugw("TxManager tx for RollupForgeBatch confirmed",
419 "batchNum", batchInfo.BatchNum)
420 t.queue = t.queue[1:]
421 if len(t.queue) == 0 {
422 waitTime = longWaitTime
423 }
424 }
425 if len(t.queue) == 0 {
426 next = 0
427 } else {
428 next = (next + 1) % len(t.queue)
429 }
430 }
431 }
432}
433
434// Pipeline manages the forging of batches with parallel server proofs
435type Pipeline struct {
436 cfg Config
437 consts synchronizer.SCConsts
438
439 // state
440 batchNum common.BatchNum
441 vars synchronizer.SCVariables
442 lastScheduledL1BatchBlockNum int64
443 started bool
444
445 serverProofPool *ServerProofPool
446 txManager *TxManager
447 historyDB *historydb.HistoryDB
448 txSelector *txselector.TxSelector
449 batchBuilder *batchbuilder.BatchBuilder
450
451 stats synchronizer.Stats
452 statsCh chan synchronizer.Stats
453
454 ctx context.Context
455 wg sync.WaitGroup
456 cancel context.CancelFunc
457}
458
459// NewPipeline creates a new Pipeline
460func NewPipeline(cfg Config,
461 historyDB *historydb.HistoryDB,
462 txSelector *txselector.TxSelector,
463 batchBuilder *batchbuilder.BatchBuilder,
464 txManager *TxManager,
465 serverProofs []ServerProofInterface,
466 scConsts *synchronizer.SCConsts,
467) *Pipeline {
468 serverProofPool := NewServerProofPool(len(serverProofs))
469 for _, serverProof := range serverProofs {
470 serverProofPool.Add(serverProof)
471 }
472 return &Pipeline{
473 cfg: cfg,
474 historyDB: historyDB,
475 txSelector: txSelector,
476 batchBuilder: batchBuilder,
477 serverProofPool: serverProofPool,
478 txManager: txManager,
479 consts: *scConsts,
480 // TODO: Find best queue size
481 statsCh: make(chan synchronizer.Stats, 16), //nolint:gomnd
482 }
483}
484
485// SetSyncStats is a thread safe method to sets the synchronizer Stats
486func (p *Pipeline) SetSyncStats(stats *synchronizer.Stats) {
487 p.statsCh <- *stats
488}
489
490// Start the forging pipeline
491func (p *Pipeline) Start(batchNum common.BatchNum,
492 syncStats *synchronizer.Stats, initSCVars *synchronizer.SCVariables) error {
493 if p.started {
494 log.Fatal("Pipeline already started")
495 }
496 p.started = true
497
498 // Reset pipeline state
499 p.batchNum = batchNum
500 p.vars = *initSCVars
501 p.lastScheduledL1BatchBlockNum = 0
502
503 p.ctx, p.cancel = context.WithCancel(context.Background())
504
505 err := p.txSelector.Reset(p.batchNum)
506 if err != nil {
507 log.Errorw("Pipeline: TxSelector.Reset", "error", err)
508 return tracerr.Wrap(err)
509 }
510 err = p.batchBuilder.Reset(p.batchNum, true)
511 if err != nil {
512 log.Errorw("Pipeline: BatchBuilder.Reset", "error", err)
513 return tracerr.Wrap(err)
514 }
515
516 queueSize := 1
517 batchChSentServerProof := make(chan *BatchInfo, queueSize)
518
519 p.wg.Add(1)
520 go func() {
521 for {
522 select {
523 case <-p.ctx.Done():
524 log.Debug("Pipeline forgeSendServerProof loop done")
525 p.wg.Done()
526 return
527 case syncStats := <-p.statsCh:
528 p.stats = syncStats
529 default:
530 p.batchNum = p.batchNum + 1
531 batchInfo, err := p.forgeSendServerProof(p.ctx, p.batchNum)
532 if tracerr.Unwrap(err) == ErrDone {
533 continue
534 }
535 if err != nil {
536 log.Errorw("forgeSendServerProof", "err", err)
537 continue
538 }
539 batchChSentServerProof <- batchInfo
540 }
541 }
542 }()
543
544 p.wg.Add(1)
545 go func() {
546 for {
547 select {
548 case <-p.ctx.Done():
549 log.Debug("Pipeline waitServerProofSendEth loop done")
550 p.wg.Done()
551 return
552 case batchInfo := <-batchChSentServerProof:
553 err := p.waitServerProof(p.ctx, batchInfo)
554 if tracerr.Unwrap(err) == ErrDone {
555 continue
556 }
557 if err != nil {
558 log.Errorw("waitServerProof", "err", err)
559 continue
560 }
561 p.txManager.AddBatch(batchInfo)
562 }
563 }
564 }()
565 return nil
566}
567
568// Stop the forging pipeline
569func (p *Pipeline) Stop() {
570 if !p.started {
571 log.Fatal("Pipeline already stopped")
572 }
573 p.started = false
574 log.Debug("Stopping Pipeline...")
575 p.cancel()
576 p.wg.Wait()
577 // TODO: Cancel all proofServers with pending proofs
578}
579
580// forgeSendServerProof the next batch, wait for a proof server to be available and send the
581// circuit inputs to the proof server.
582func (p *Pipeline) forgeSendServerProof(ctx context.Context, batchNum common.BatchNum) (*BatchInfo, error) {
583 // remove transactions from the pool that have been there for too long
584 err := p.purgeRemoveByTimeout()
585 if err != nil {
586 return nil, tracerr.Wrap(err)
587 }
588
589 batchInfo := BatchInfo{BatchNum: batchNum} // to accumulate metadata of the batch
590
591 var poolL2Txs []common.PoolL2Tx
592 // var feesInfo
593 var l1UserTxsExtra, l1OperatorTxs []common.L1Tx
594 // 1. Decide if we forge L2Tx or L1+L2Tx
595 if p.shouldL1L2Batch() {
596 p.lastScheduledL1BatchBlockNum = p.stats.Eth.LastBatch
597 // 2a: L1+L2 txs
598 // l1UserTxs, toForgeL1TxsNumber := c.historyDB.GetNextL1UserTxs() // TODO once HistoryDB is ready, uncomment
599 var l1UserTxs []common.L1Tx = nil // tmp, depends on HistoryDB
600 l1UserTxsExtra, l1OperatorTxs, poolL2Txs, err = p.txSelector.GetL1L2TxSelection([]common.Idx{}, batchNum, l1UserTxs) // TODO once feesInfo is added to method return, add the var
601 if err != nil {
602 return nil, tracerr.Wrap(err)
603 }
604 } else {
605 // 2b: only L2 txs
606 _, poolL2Txs, err = p.txSelector.GetL2TxSelection([]common.Idx{}, batchNum) // TODO once feesInfo is added to method return, add the var
607 if err != nil {
608 return nil, tracerr.Wrap(err)
609 }
610 l1UserTxsExtra = nil
611 l1OperatorTxs = nil
612 }
613
614 // Run purger to invalidate transactions that become invalid beause of
615 // the poolL2Txs selected. Will mark as invalid the txs that have a
616 // (fromIdx, nonce) which already appears in the selected txs (includes
617 // all the nonces smaller than the current one)
618 err = p.purgeInvalidDueToL2TxsSelection(poolL2Txs)
619 if err != nil {
620 return nil, tracerr.Wrap(err)
621 }
622
623 // 3. Save metadata from TxSelector output for BatchNum
624 // batchInfo.SetTxsInfo(l1UserTxsExtra, l1OperatorTxs, poolL2Txs) // TODO feesInfo
625 batchInfo.L1UserTxsExtra = l1UserTxsExtra
626 batchInfo.L1OperatorTxs = l1OperatorTxs
627 batchInfo.L2Txs = poolL2Txs
628
629 // 4. Call BatchBuilder with TxSelector output
630 configBatch := &batchbuilder.ConfigBatch{
631 ForgerAddress: p.cfg.ForgerAddress,
632 }
633 zkInputs, err := p.batchBuilder.BuildBatch([]common.Idx{}, configBatch, l1UserTxsExtra, l1OperatorTxs, poolL2Txs, nil) // TODO []common.TokenID --> feesInfo
634 if err != nil {
635 return nil, tracerr.Wrap(err)
636 }
637
638 // 5. Save metadata from BatchBuilder output for BatchNum
639 batchInfo.ZKInputs = zkInputs
640 p.cfg.debugBatchStore(&batchInfo)
641
642 // 6. Wait for an available server proof blocking call
643 serverProof, err := p.serverProofPool.Get(ctx)
644 if err != nil {
645 return nil, tracerr.Wrap(err)
646 }
647 batchInfo.ServerProof = serverProof
648 defer func() {
649 // If there's an error further on, add the serverProof back to
650 // the pool
651 if err != nil {
652 p.serverProofPool.Add(serverProof)
653 }
654 }()
655 p.cfg.debugBatchStore(&batchInfo)
656
657 // 7. Call the selected idle server proof with BatchBuilder output,
658 // save server proof info for batchNum
659 err = batchInfo.ServerProof.CalculateProof(zkInputs)
660 if err != nil {
661 return nil, tracerr.Wrap(err)
662 }
663
664 return &batchInfo, nil
665}
666
667// waitServerProof gets the generated zkProof & sends it to the SmartContract
668func (p *Pipeline) waitServerProof(ctx context.Context, batchInfo *BatchInfo) error {
669 proof, err := batchInfo.ServerProof.GetProof(ctx) // blocking call, until not resolved don't continue. Returns when the proof server has calculated the proof
670 if err != nil {
671 return tracerr.Wrap(err)
672 }
673 p.serverProofPool.Add(batchInfo.ServerProof)
674 batchInfo.ServerProof = nil
675 batchInfo.Proof = proof
676 batchInfo.ForgeBatchArgs = p.prepareForgeBatchArgs(batchInfo)
677 batchInfo.TxStatus = TxStatusPending
678 p.cfg.debugBatchStore(batchInfo)
679 return nil
680}
681
682// isForgeSequence returns true if the node is the Forger in the current ethereum block
683// func (c *Coordinator) isForgeSequence() (bool, error) {
684// // TODO: Consider checking if we can forge by quering the Synchronizer instead of using ethClient
685// blockNum, err := c.ethClient.EthLastBlock()
686// if err != nil {
687// return false, err
688// }
689// addr, err := c.ethClient.EthAddress()
690// if err != nil {
691// return false, err
692// }
693// return c.ethClient.AuctionCanForge(*addr, blockNum+1)
694// }
695
696func (p *Pipeline) purgeRemoveByTimeout() error {
697 return nil // TODO
698}
699
700func (p *Pipeline) purgeInvalidDueToL2TxsSelection(l2Txs []common.PoolL2Tx) error {
701 return nil // TODO
702}
703
704func (p *Pipeline) shouldL1L2Batch() bool {
705 // Take the lastL1BatchBlockNum as the biggest between the last
706 // scheduled one, and the synchronized one.
707 lastL1BatchBlockNum := p.lastScheduledL1BatchBlockNum
708 if p.stats.Sync.LastL1BatchBlock > lastL1BatchBlockNum {
709 lastL1BatchBlockNum = p.stats.Sync.LastL1BatchBlock
710 }
711 // Return true if we have passed the l1BatchTimeoutPerc portion of the
712 // range before the l1batch timeout.
713 if p.stats.Eth.LastBlock.Num-lastL1BatchBlockNum >=
714 int64(float64(p.vars.Rollup.ForgeL1L2BatchTimeout)*p.cfg.L1BatchTimeoutPerc) {
715 return true
716 }
717 return false
718}
719
720func (p *Pipeline) prepareForgeBatchArgs(batchInfo *BatchInfo) *eth.RollupForgeBatchArgs {
721 // TODO
722 return ð.RollupForgeBatchArgs{}
723}