hermez-node/coordinator/coordinator.go
coordinator/coordinator.goRaw
1package coordinator
2
3import (
4 "context"
5 "fmt"
6 "strings"
7 "sync"
8 "time"
9
10 ethCommon "github.com/ethereum/go-ethereum/common"
11 "github.com/ethereum/go-ethereum/core/types"
12 "github.com/hermeznetwork/hermez-node/batchbuilder"
13 "github.com/hermeznetwork/hermez-node/common"
14 "github.com/hermeznetwork/hermez-node/db/historydb"
15 "github.com/hermeznetwork/hermez-node/db/l2db"
16 "github.com/hermeznetwork/hermez-node/eth"
17 "github.com/hermeznetwork/hermez-node/log"
18 "github.com/hermeznetwork/hermez-node/prover"
19 "github.com/hermeznetwork/hermez-node/synchronizer"
20 "github.com/hermeznetwork/hermez-node/txselector"
21 "github.com/hermeznetwork/tracerr"
22)
23
24const queueLen = 16
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 pipelineBatchNum common.BatchNum // batchNum from which we started the pipeline
63 provers []prover.Client
64 consts synchronizer.SCConsts
65 vars synchronizer.SCVariables
66 started bool
67
68 cfg Config
69
70 historyDB *historydb.HistoryDB
71 l2DB *l2db.L2DB
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 l2DB *l2db.L2DB,
89 txSelector *txselector.TxSelector,
90 batchBuilder *batchbuilder.BatchBuilder,
91 serverProofs []prover.Client,
92 ethClient eth.ClientInterface,
93 scConsts *synchronizer.SCConsts,
94 initSCVars *synchronizer.SCVariables,
95) (*Coordinator, error) {
96 // nolint reason: hardcoded `1.0`, by design the percentage can't be over 100%
97 if cfg.L1BatchTimeoutPerc >= 1.0 { //nolint:gomnd
98 return nil, tracerr.Wrap(fmt.Errorf("invalid value for Config.L1BatchTimeoutPerc (%v >= 1.0)",
99 cfg.L1BatchTimeoutPerc))
100 }
101 if cfg.EthClientAttempts < 1 {
102 return nil, tracerr.Wrap(fmt.Errorf("invalid value for Config.EthClientAttempts (%v < 1)",
103 cfg.EthClientAttempts))
104 }
105
106 ctx, cancel := context.WithCancel(context.Background())
107 c := Coordinator{
108 pipelineBatchNum: -1,
109 provers: serverProofs,
110 consts: *scConsts,
111 vars: *initSCVars,
112
113 cfg: cfg,
114
115 historyDB: historyDB,
116 l2DB: l2DB,
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 := NewTxManager(&cfg, ethClient, l2DB, &c)
128 c.txManager = txManager
129 return &c, nil
130}
131
132func (c *Coordinator) newPipeline(ctx context.Context) (*Pipeline, error) {
133 return NewPipeline(ctx, c.cfg, c.historyDB, c.l2DB,
134 c.txSelector, c.batchBuilder, c.txManager, c.provers, &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 Stats synchronizer.Stats
152}
153
154// MsgStopPipeline indicates a signal to reset the pipeline
155type MsgStopPipeline struct {
156 Reason string
157}
158
159// SendMsg is a thread safe method to pass a message to the Coordinator
160func (c *Coordinator) SendMsg(msg interface{}) {
161 c.msgCh <- msg
162}
163
164func (c *Coordinator) handleMsgSyncSCVars(msg *MsgSyncSCVars) {
165 if msg.Rollup != nil {
166 c.vars.Rollup = *msg.Rollup
167 }
168 if msg.Auction != nil {
169 c.vars.Auction = *msg.Auction
170 }
171 if msg.WDelayer != nil {
172 c.vars.WDelayer = *msg.WDelayer
173 }
174}
175
176func (c *Coordinator) canForge(stats *synchronizer.Stats) bool {
177 anyoneForge := false
178 if stats.Sync.Auction.CurrentSlot.BatchesLen == 0 &&
179 c.consts.Auction.RelativeBlock(stats.Eth.LastBlock.Num+1) > int64(c.vars.Auction.SlotDeadline) {
180 log.Debug("Coordinator: anyone can forge in the current slot (slotDeadline passed)")
181 anyoneForge = true
182 }
183 if stats.Sync.Auction.CurrentSlot.Forger == c.cfg.ForgerAddress || anyoneForge {
184 return true
185 }
186 return false
187}
188
189func (c *Coordinator) handleMsgSyncStats(ctx context.Context, stats *synchronizer.Stats) error {
190 if !stats.Synced() {
191 return nil
192 }
193 c.txManager.SetLastBlock(stats.Eth.LastBlock.Num)
194
195 canForge := c.canForge(stats)
196 if c.pipeline == nil {
197 if canForge {
198 log.Infow("Coordinator: forging state begin", "block", stats.Eth.LastBlock.Num,
199 "batch", stats.Sync.LastBatch)
200 batchNum := common.BatchNum(stats.Sync.LastBatch)
201 var err error
202 if c.pipeline, err = c.newPipeline(ctx); err != nil {
203 return tracerr.Wrap(err)
204 }
205 if err := c.pipeline.Start(batchNum, stats.Sync.LastForgeL1TxsNum,
206 stats, &c.vars); err != nil {
207 c.pipeline = nil
208 return tracerr.Wrap(err)
209 }
210 c.pipelineBatchNum = batchNum
211 }
212 } else {
213 if canForge {
214 c.pipeline.SetSyncStats(stats)
215 } else {
216 log.Infow("Coordinator: forging state end", "block", stats.Eth.LastBlock.Num)
217 c.pipeline.Stop(c.ctx)
218 c.pipeline = nil
219 }
220 }
221 return nil
222}
223
224func (c *Coordinator) handleStopPipeline(ctx context.Context, reason string) error {
225 if c.pipeline != nil {
226 c.pipeline.Stop(c.ctx)
227 c.pipeline = nil
228 }
229 if strings.Contains(reason, common.AuctionErrMsgCannotForge) { //nolint:staticcheck
230 // TODO: Check that we are in a slot in which we can't forge
231 }
232 return nil
233}
234
235// Start the coordinator
236func (c *Coordinator) Start() {
237 if c.started {
238 log.Fatal("Coordinator already started")
239 }
240 c.started = true
241 c.wg.Add(1)
242 go func() {
243 c.txManager.Run(c.ctx)
244 c.wg.Done()
245 }()
246
247 c.wg.Add(1)
248 go func() {
249 for {
250 select {
251 case <-c.ctx.Done():
252 log.Info("Coordinator done")
253 c.wg.Done()
254 return
255 case msg := <-c.msgCh:
256 switch msg := msg.(type) {
257 case MsgSyncStats:
258 stats := msg.Stats
259 if err := c.handleMsgSyncStats(c.ctx, &stats); common.IsErrDone(err) {
260 continue
261 } else if err != nil {
262 log.Errorw("Coordinator.handleMsgSyncStats error", "err", err)
263 continue
264 }
265 case MsgSyncReorg:
266 if err := c.handleReorg(c.ctx, &msg.Stats); common.IsErrDone(err) {
267 continue
268 } else if err != nil {
269 log.Errorw("Coordinator.handleReorg error", "err", err)
270 continue
271 }
272 case MsgStopPipeline:
273 log.Infow("Coordinator received MsgStopPipeline", "reason", msg.Reason)
274 if err := c.handleStopPipeline(c.ctx, msg.Reason); common.IsErrDone(err) {
275 continue
276 } else if err != nil {
277 log.Errorw("Coordinator.handleStopPipeline", "err", err)
278 }
279 case MsgSyncSCVars:
280 c.handleMsgSyncSCVars(&msg)
281 default:
282 log.Fatalw("Coordinator Unexpected Coordinator msg of type %T: %+v", msg, msg)
283 }
284 }
285 }
286 }()
287}
288
289const stopCtxTimeout = 200 * time.Millisecond
290
291// Stop the coordinator
292func (c *Coordinator) Stop() {
293 if !c.started {
294 log.Fatal("Coordinator already stopped")
295 }
296 c.started = false
297 log.Infow("Stopping Coordinator...")
298 c.cancel()
299 c.wg.Wait()
300 if c.pipeline != nil {
301 ctx, cancel := context.WithTimeout(context.Background(), stopCtxTimeout)
302 defer cancel()
303 c.pipeline.Stop(ctx)
304 c.pipeline = nil
305 }
306}
307
308func (c *Coordinator) handleReorg(ctx context.Context, stats *synchronizer.Stats) error {
309 if common.BatchNum(stats.Sync.LastBatch) < c.pipelineBatchNum {
310 // There's been a reorg and the batch from which the pipeline
311 // was started was in a block that was discarded. The batch
312 // may not be in the main chain, so we stop the pipeline as a
313 // precaution (it will be started again once the node is in
314 // sync).
315 log.Infow("Coordinator.handleReorg StopPipeline sync.LastBatch < c.pipelineBatchNum",
316 "sync.LastBatch", stats.Sync.LastBatch,
317 "c.pipelineBatchNum", c.pipelineBatchNum)
318 if err := c.handleStopPipeline(ctx, "reorg"); err != nil {
319 return tracerr.Wrap(err)
320 }
321 if err := c.l2DB.Reorg(common.BatchNum(stats.Sync.LastBatch)); err != nil {
322 return tracerr.Wrap(err)
323 }
324 }
325 return nil
326}
327
328// TxManager handles everything related to ethereum transactions: It makes the
329// call to forge, waits for transaction confirmation, and keeps checking them
330// until a number of confirmed blocks have passed.
331type TxManager struct {
332 cfg Config
333 ethClient eth.ClientInterface
334 l2DB *l2db.L2DB // Used only to mark forged txs as forged in the L2DB
335 coord *Coordinator // Used only to send messages to stop the pipeline
336 batchCh chan *BatchInfo
337 lastBlockCh chan int64
338 queue []*BatchInfo
339 lastBlock int64
340 // lastConfirmedBatch stores the last BatchNum that who's forge call was confirmed
341 lastConfirmedBatch common.BatchNum
342}
343
344// NewTxManager creates a new TxManager
345func NewTxManager(cfg *Config, ethClient eth.ClientInterface, l2DB *l2db.L2DB,
346 coord *Coordinator) *TxManager {
347 return &TxManager{
348 cfg: *cfg,
349 ethClient: ethClient,
350 l2DB: l2DB,
351 coord: coord,
352 batchCh: make(chan *BatchInfo, queueLen),
353 lastBlockCh: make(chan int64, queueLen),
354 lastBlock: -1,
355 }
356}
357
358// AddBatch is a thread safe method to pass a new batch TxManager to be sent to
359// the smart contract via the forge call
360func (t *TxManager) AddBatch(batchInfo *BatchInfo) {
361 t.batchCh <- batchInfo
362}
363
364// SetLastBlock is a thread safe method to pass the lastBlock to the TxManager
365func (t *TxManager) SetLastBlock(lastBlock int64) {
366 t.lastBlockCh <- lastBlock
367}
368
369func (t *TxManager) rollupForgeBatch(ctx context.Context, batchInfo *BatchInfo) error {
370 var ethTx *types.Transaction
371 var err error
372 for attempt := 0; attempt < t.cfg.EthClientAttempts; attempt++ {
373 ethTx, err = t.ethClient.RollupForgeBatch(batchInfo.ForgeBatchArgs)
374 if err != nil {
375 if strings.Contains(err.Error(), common.AuctionErrMsgCannotForge) {
376 log.Debugw("TxManager ethClient.RollupForgeBatch", "err", err,
377 "block", t.lastBlock)
378 return tracerr.Wrap(err)
379 }
380 log.Errorw("TxManager ethClient.RollupForgeBatch",
381 "attempt", attempt, "err", err, "block", t.lastBlock)
382 } else {
383 break
384 }
385 select {
386 case <-ctx.Done():
387 return tracerr.Wrap(common.ErrDone)
388 case <-time.After(t.cfg.EthClientAttemptsDelay):
389 }
390 }
391 if err != nil {
392 return tracerr.Wrap(fmt.Errorf("reached max attempts for ethClient.RollupForgeBatch: %w", err))
393 }
394 batchInfo.EthTx = ethTx
395 log.Infow("TxManager ethClient.RollupForgeBatch", "batch", batchInfo.BatchNum, "tx", ethTx.Hash().Hex())
396 t.cfg.debugBatchStore(batchInfo)
397 if err := t.l2DB.DoneForging(l2TxsIDs(batchInfo.L2Txs), batchInfo.BatchNum); err != nil {
398 return tracerr.Wrap(err)
399 }
400 return nil
401}
402
403func (t *TxManager) ethTransactionReceipt(ctx context.Context, batchInfo *BatchInfo) error {
404 txHash := batchInfo.EthTx.Hash()
405 var receipt *types.Receipt
406 var err error
407 for attempt := 0; attempt < t.cfg.EthClientAttempts; attempt++ {
408 receipt, err = t.ethClient.EthTransactionReceipt(ctx, txHash)
409 if err != nil {
410 log.Errorw("TxManager ethClient.EthTransactionReceipt",
411 "attempt", attempt, "err", err)
412 } else {
413 break
414 }
415 select {
416 case <-ctx.Done():
417 return tracerr.Wrap(common.ErrDone)
418 case <-time.After(t.cfg.EthClientAttemptsDelay):
419 }
420 }
421 if err != nil {
422 return tracerr.Wrap(fmt.Errorf("reached max attempts for ethClient.EthTransactionReceipt: %w", err))
423 }
424 batchInfo.Receipt = receipt
425 t.cfg.debugBatchStore(batchInfo)
426 return nil
427}
428
429func (t *TxManager) handleReceipt(batchInfo *BatchInfo) (*int64, error) {
430 receipt := batchInfo.Receipt
431 if receipt != nil {
432 if receipt.Status == types.ReceiptStatusFailed {
433 log.Errorw("TxManager receipt status is failed", "receipt", receipt)
434 return nil, tracerr.Wrap(fmt.Errorf("ethereum transaction receipt statis is failed"))
435 } else if receipt.Status == types.ReceiptStatusSuccessful {
436 if batchInfo.BatchNum > t.lastConfirmedBatch {
437 t.lastConfirmedBatch = batchInfo.BatchNum
438 }
439 confirm := t.lastBlock - receipt.BlockNumber.Int64()
440 return &confirm, nil
441 }
442 }
443 return nil, nil
444}
445
446const longWaitTime = 999 * time.Hour
447
448// Run the TxManager
449func (t *TxManager) Run(ctx context.Context) {
450 next := 0
451 waitTime := time.Duration(longWaitTime)
452 for {
453 select {
454 case <-ctx.Done():
455 log.Info("TxManager done")
456 return
457 case lastBlock := <-t.lastBlockCh:
458 t.lastBlock = lastBlock
459 case batchInfo := <-t.batchCh:
460 if err := t.rollupForgeBatch(ctx, batchInfo); common.IsErrDone(err) {
461 continue
462 } else if err != nil {
463 t.coord.SendMsg(MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch call: %v", err)})
464 continue
465 }
466 log.Debugf("ethClient ForgeCall sent, batchNum: %d", batchInfo.BatchNum)
467 t.queue = append(t.queue, batchInfo)
468 waitTime = t.cfg.TxManagerCheckInterval
469 case <-time.After(waitTime):
470 if len(t.queue) == 0 {
471 continue
472 }
473 current := next
474 next = (current + 1) % len(t.queue)
475 batchInfo := t.queue[current]
476 err := t.ethTransactionReceipt(ctx, batchInfo)
477 if common.IsErrDone(err) {
478 continue
479 } else if err != nil { //nolint:staticcheck
480 // We can't get the receipt for the
481 // transaction, so we can't confirm if it was
482 // mined
483 t.coord.SendMsg(MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch receipt: %v", err)})
484 }
485
486 confirm, err := t.handleReceipt(batchInfo)
487 if err != nil { //nolint:staticcheck
488 // Transaction was rejected
489 t.coord.SendMsg(MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch reject: %v", err)})
490 }
491 if confirm != nil && *confirm >= t.cfg.ConfirmBlocks {
492 log.Debugw("TxManager tx for RollupForgeBatch confirmed",
493 "batch", batchInfo.BatchNum)
494 t.queue = append(t.queue[:current], t.queue[current+1:]...)
495 if len(t.queue) == 0 {
496 waitTime = longWaitTime
497 next = 0
498 } else {
499 next = current % len(t.queue)
500 }
501 }
502 }
503 }
504}
505
506// Pipeline manages the forging of batches with parallel server proofs
507type Pipeline struct {
508 cfg Config
509 consts synchronizer.SCConsts
510
511 // state
512 batchNum common.BatchNum
513 vars synchronizer.SCVariables
514 lastScheduledL1BatchBlockNum int64
515 lastForgeL1TxsNum int64
516 started bool
517
518 proversPool *ProversPool
519 provers []prover.Client
520 txManager *TxManager
521 historyDB *historydb.HistoryDB
522 l2DB *l2db.L2DB
523 txSelector *txselector.TxSelector
524 batchBuilder *batchbuilder.BatchBuilder
525
526 stats synchronizer.Stats
527 statsCh chan synchronizer.Stats
528
529 ctx context.Context
530 wg sync.WaitGroup
531 cancel context.CancelFunc
532}
533
534// NewPipeline creates a new Pipeline
535func NewPipeline(ctx context.Context,
536 cfg Config,
537 historyDB *historydb.HistoryDB,
538 l2DB *l2db.L2DB,
539 txSelector *txselector.TxSelector,
540 batchBuilder *batchbuilder.BatchBuilder,
541 txManager *TxManager,
542 provers []prover.Client,
543 scConsts *synchronizer.SCConsts,
544) (*Pipeline, error) {
545 proversPool := NewProversPool(len(provers))
546 proversPoolSize := 0
547 for _, prover := range provers {
548 if err := prover.WaitReady(ctx); err != nil {
549 log.Errorw("prover.WaitReady", "err", err)
550 } else {
551 proversPool.Add(prover)
552 proversPoolSize++
553 }
554 }
555 if proversPoolSize == 0 {
556 return nil, tracerr.Wrap(fmt.Errorf("no provers in the pool"))
557 }
558 return &Pipeline{
559 cfg: cfg,
560 historyDB: historyDB,
561 l2DB: l2DB,
562 txSelector: txSelector,
563 batchBuilder: batchBuilder,
564 provers: provers,
565 proversPool: proversPool,
566 txManager: txManager,
567 consts: *scConsts,
568 statsCh: make(chan synchronizer.Stats, queueLen),
569 }, nil
570}
571
572// SetSyncStats is a thread safe method to sets the synchronizer Stats
573func (p *Pipeline) SetSyncStats(stats *synchronizer.Stats) {
574 p.statsCh <- *stats
575}
576
577// Start the forging pipeline
578func (p *Pipeline) Start(batchNum common.BatchNum, lastForgeL1TxsNum int64,
579 syncStats *synchronizer.Stats, initSCVars *synchronizer.SCVariables) error {
580 if p.started {
581 log.Fatal("Pipeline already started")
582 }
583 p.started = true
584
585 // Reset pipeline state
586 p.batchNum = batchNum
587 p.lastForgeL1TxsNum = lastForgeL1TxsNum
588 p.vars = *initSCVars
589 p.lastScheduledL1BatchBlockNum = 0
590
591 p.ctx, p.cancel = context.WithCancel(context.Background())
592
593 err := p.txSelector.Reset(p.batchNum)
594 if err != nil {
595 return tracerr.Wrap(err)
596 }
597 err = p.batchBuilder.Reset(p.batchNum, true)
598 if err != nil {
599 return tracerr.Wrap(err)
600 }
601
602 queueSize := 1
603 batchChSentServerProof := make(chan *BatchInfo, queueSize)
604
605 p.wg.Add(1)
606 go func() {
607 for {
608 select {
609 case <-p.ctx.Done():
610 log.Debug("Pipeline forgeSendServerProof loop done")
611 p.wg.Done()
612 return
613 case syncStats := <-p.statsCh:
614 p.stats = syncStats
615 default:
616 p.batchNum = p.batchNum + 1
617 batchInfo, err := p.forgeSendServerProof(p.ctx, p.batchNum)
618 if common.IsErrDone(err) {
619 continue
620 }
621 if err != nil {
622 log.Errorw("forgeSendServerProof", "err", err)
623 continue
624 }
625 batchChSentServerProof <- batchInfo
626 }
627 }
628 }()
629
630 p.wg.Add(1)
631 go func() {
632 for {
633 select {
634 case <-p.ctx.Done():
635 log.Debug("Pipeline waitServerProofSendEth loop done")
636 p.wg.Done()
637 return
638 case batchInfo := <-batchChSentServerProof:
639 err := p.waitServerProof(p.ctx, batchInfo)
640 if common.IsErrDone(err) {
641 continue
642 }
643 if err != nil {
644 log.Errorw("waitServerProof", "err", err)
645 continue
646 }
647 p.txManager.AddBatch(batchInfo)
648 }
649 }
650 }()
651 return nil
652}
653
654// Stop the forging pipeline
655func (p *Pipeline) Stop(ctx context.Context) {
656 if !p.started {
657 log.Fatal("Pipeline already stopped")
658 }
659 p.started = false
660 log.Debug("Stopping Pipeline...")
661 p.cancel()
662 p.wg.Wait()
663 for _, prover := range p.provers {
664 if err := prover.Cancel(ctx); err != nil {
665 log.Errorw("prover.Cancel", "err", err)
666 }
667 }
668}
669
670func l2TxsIDs(txs []common.PoolL2Tx) []common.TxID {
671 txIDs := make([]common.TxID, len(txs))
672 for i, tx := range txs {
673 txIDs[i] = tx.TxID
674 }
675 return txIDs
676}
677
678// forgeSendServerProof the next batch, wait for a proof server to be available and send the
679// circuit inputs to the proof server.
680func (p *Pipeline) forgeSendServerProof(ctx context.Context, batchNum common.BatchNum) (*BatchInfo, error) {
681 // remove transactions from the pool that have been there for too long
682 err := p.l2DB.Purge(common.BatchNum(p.stats.Sync.LastBatch))
683 if err != nil {
684 return nil, tracerr.Wrap(err)
685 }
686
687 batchInfo := BatchInfo{BatchNum: batchNum} // to accumulate metadata of the batch
688
689 var poolL2Txs []common.PoolL2Tx
690 // var feesInfo
691 var l1UserTxsExtra, l1CoordTxs []common.L1Tx
692 // 1. Decide if we forge L2Tx or L1+L2Tx
693 if p.shouldL1L2Batch() {
694 p.lastScheduledL1BatchBlockNum = p.stats.Eth.LastBlock.Num
695 // 2a: L1+L2 txs
696 p.lastForgeL1TxsNum++
697 l1UserTxs, err := p.historyDB.GetL1UserTxs(p.lastForgeL1TxsNum)
698 if err != nil {
699 return nil, tracerr.Wrap(err)
700 }
701 l1UserTxsExtra, l1CoordTxs, poolL2Txs, err = p.txSelector.GetL1L2TxSelection([]common.Idx{}, batchNum, l1UserTxs) // TODO once feesInfo is added to method return, add the var
702 if err != nil {
703 return nil, tracerr.Wrap(err)
704 }
705 } else {
706 // 2b: only L2 txs
707 l1CoordTxs, poolL2Txs, err = p.txSelector.GetL2TxSelection([]common.Idx{}, batchNum)
708 if err != nil {
709 return nil, tracerr.Wrap(err)
710 }
711 l1UserTxsExtra = nil
712 }
713
714 // 3. Save metadata from TxSelector output for BatchNum
715 // TODO feesInfo
716 batchInfo.L1UserTxsExtra = l1UserTxsExtra
717 batchInfo.L1CoordTxs = l1CoordTxs
718 batchInfo.L2Txs = poolL2Txs
719
720 if err := p.l2DB.StartForging(l2TxsIDs(batchInfo.L2Txs), batchInfo.BatchNum); err != nil {
721 return nil, tracerr.Wrap(err)
722 }
723
724 // Run purger to invalidate transactions that become invalid beause of
725 // the poolL2Txs selected. Will mark as invalid the txs that have a
726 // (fromIdx, nonce) which already appears in the selected txs (includes
727 // all the nonces smaller than the current one)
728 err = p.purgeInvalidDueToL2TxsSelection(poolL2Txs)
729 if err != nil {
730 return nil, tracerr.Wrap(err)
731 }
732
733 // 4. Call BatchBuilder with TxSelector output
734 configBatch := &batchbuilder.ConfigBatch{
735 ForgerAddress: p.cfg.ForgerAddress,
736 }
737 zkInputs, err := p.batchBuilder.BuildBatch([]common.Idx{}, configBatch,
738 l1UserTxsExtra, l1CoordTxs, poolL2Txs, nil) // TODO []common.TokenID --> feesInfo
739 if err != nil {
740 return nil, tracerr.Wrap(err)
741 }
742
743 // 5. Save metadata from BatchBuilder output for BatchNum
744 batchInfo.ZKInputs = zkInputs
745 p.cfg.debugBatchStore(&batchInfo)
746
747 // 6. Wait for an available server proof blocking call
748 serverProof, err := p.proversPool.Get(ctx)
749 if err != nil {
750 return nil, tracerr.Wrap(err)
751 }
752 batchInfo.ServerProof = serverProof
753 defer func() {
754 // If there's an error further on, add the serverProof back to
755 // the pool
756 if err != nil {
757 p.proversPool.Add(serverProof)
758 }
759 }()
760 p.cfg.debugBatchStore(&batchInfo)
761
762 // 7. Call the selected idle server proof with BatchBuilder output,
763 // save server proof info for batchNum
764 err = batchInfo.ServerProof.CalculateProof(zkInputs)
765 if err != nil {
766 return nil, tracerr.Wrap(err)
767 }
768
769 return &batchInfo, nil
770}
771
772// waitServerProof gets the generated zkProof & sends it to the SmartContract
773func (p *Pipeline) waitServerProof(ctx context.Context, batchInfo *BatchInfo) error {
774 proof, err := batchInfo.ServerProof.GetProof(ctx) // blocking call, until not resolved don't continue. Returns when the proof server has calculated the proof
775 if err != nil {
776 return tracerr.Wrap(err)
777 }
778 p.proversPool.Add(batchInfo.ServerProof)
779 batchInfo.ServerProof = nil
780 batchInfo.Proof = proof
781 batchInfo.ForgeBatchArgs = p.prepareForgeBatchArgs(batchInfo)
782 batchInfo.TxStatus = TxStatusPending
783 p.cfg.debugBatchStore(batchInfo)
784 return nil
785}
786
787func (p *Pipeline) purgeInvalidDueToL2TxsSelection(l2Txs []common.PoolL2Tx) error {
788 return nil // TODO
789}
790
791func (p *Pipeline) shouldL1L2Batch() bool {
792 // Take the lastL1BatchBlockNum as the biggest between the last
793 // scheduled one, and the synchronized one.
794 lastL1BatchBlockNum := p.lastScheduledL1BatchBlockNum
795 if p.stats.Sync.LastL1BatchBlock > lastL1BatchBlockNum {
796 lastL1BatchBlockNum = p.stats.Sync.LastL1BatchBlock
797 }
798 // Return true if we have passed the l1BatchTimeoutPerc portion of the
799 // range before the l1batch timeout.
800 if p.stats.Eth.LastBlock.Num-lastL1BatchBlockNum >=
801 int64(float64(p.vars.Rollup.ForgeL1L2BatchTimeout)*p.cfg.L1BatchTimeoutPerc) {
802 return true
803 }
804 return false
805}
806
807func (p *Pipeline) prepareForgeBatchArgs(batchInfo *BatchInfo) *eth.RollupForgeBatchArgs {
808 // TODO
809 return ð.RollupForgeBatchArgs{}
810}