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