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