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