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) (*Pipeline, error) {
160 return NewPipeline(ctx, c.cfg, c.historyDB, c.l2DB, c.txSelector,
161 c.batchBuilder, c.purger, c.txManager, c.provers, &c.consts)
162}
163
164// MsgSyncBlock indicates an update to the Synchronizer stats
165type MsgSyncBlock struct {
166 Stats synchronizer.Stats
167 Batches []common.BatchData
168 // Vars contains each Smart Contract variables if they are updated, or
169 // nil if they haven't changed.
170 Vars synchronizer.SCVariablesPtr
171}
172
173// MsgSyncReorg indicates a reorg
174type MsgSyncReorg struct {
175 Stats synchronizer.Stats
176 Vars synchronizer.SCVariablesPtr
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+1)
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+1, "batch", stats.Sync.LastBatch)
223 batchNum := common.BatchNum(stats.Sync.LastBatch)
224 var err error
225 if c.pipeline, err = c.newPipeline(ctx); 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 log.Infow("Coordinator: forging state end", "block", stats.Eth.LastBlock.Num+1)
238 c.pipeline.Stop(c.ctx)
239 c.pipeline = nil
240 }
241 }
242 if c.pipeline == nil {
243 // Mark invalid in Pool due to forged L2Txs
244 // for _, batch := range batches {
245 // if err := c.l2DB.InvalidateOldNonces(
246 // idxsNonceFromL2Txs(batch.L2Txs), batch.Batch.BatchNum); err != nil {
247 // return err
248 // }
249 // }
250 if c.purger.CanInvalidate(stats.Sync.LastBlock.Num, stats.Sync.LastBatch) {
251 if err := c.txSelector.Reset(common.BatchNum(stats.Sync.LastBatch)); err != nil {
252 return tracerr.Wrap(err)
253 }
254 }
255 _, err := c.purger.InvalidateMaybe(c.l2DB, c.txSelector.LocalAccountsDB(),
256 stats.Sync.LastBlock.Num, stats.Sync.LastBatch)
257 if err != nil {
258 return tracerr.Wrap(err)
259 }
260 _, err = c.purger.PurgeMaybe(c.l2DB, stats.Sync.LastBlock.Num, stats.Sync.LastBatch)
261 if err != nil {
262 return tracerr.Wrap(err)
263 }
264 }
265 return nil
266}
267
268func (c *Coordinator) handleMsgSyncBlock(ctx context.Context, msg *MsgSyncBlock) error {
269 c.stats = &msg.Stats
270 c.syncSCVars(msg.Vars)
271 if c.pipeline != nil {
272 c.pipeline.SetSyncStatsVars(&msg.Stats, &msg.Vars)
273 }
274 if !c.stats.Synced() {
275 return nil
276 }
277 return c.syncStats(ctx, c.stats)
278}
279
280func (c *Coordinator) handleReorg(ctx context.Context, msg *MsgSyncReorg) error {
281 c.stats = &msg.Stats
282 c.syncSCVars(msg.Vars)
283 if c.pipeline != nil {
284 c.pipeline.SetSyncStatsVars(&msg.Stats, &msg.Vars)
285 }
286 if common.BatchNum(c.stats.Sync.LastBatch) < c.pipelineBatchNum {
287 // There's been a reorg and the batch from which the pipeline
288 // was started was in a block that was discarded. The batch
289 // may not be in the main chain, so we stop the pipeline as a
290 // precaution (it will be started again once the node is in
291 // sync).
292 log.Infow("Coordinator.handleReorg StopPipeline sync.LastBatch < c.pipelineBatchNum",
293 "sync.LastBatch", c.stats.Sync.LastBatch,
294 "c.pipelineBatchNum", c.pipelineBatchNum)
295 if err := c.handleStopPipeline(ctx, "reorg"); err != nil {
296 return tracerr.Wrap(err)
297 }
298 }
299 return nil
300}
301
302func (c *Coordinator) handleStopPipeline(ctx context.Context, reason string) error {
303 if err := c.l2DB.Reorg(common.BatchNum(c.stats.Sync.LastBatch)); err != nil {
304 return tracerr.Wrap(err)
305 }
306 if c.pipeline != nil {
307 c.pipeline.Stop(c.ctx)
308 c.pipeline = nil
309 }
310 if strings.Contains(reason, common.AuctionErrMsgCannotForge) { //nolint:staticcheck
311 // TODO: Check that we are in a slot in which we can't forge
312 }
313 return nil
314}
315
316func (c *Coordinator) handleMsg(ctx context.Context, msg interface{}) error {
317 switch msg := msg.(type) {
318 case MsgSyncBlock:
319 if err := c.handleMsgSyncBlock(ctx, &msg); err != nil {
320 return tracerr.Wrap(fmt.Errorf("Coordinator.handleMsgSyncBlock error: %w", err))
321 }
322 case MsgSyncReorg:
323 if err := c.handleReorg(ctx, &msg); err != nil {
324 return tracerr.Wrap(fmt.Errorf("Coordinator.handleReorg error: %w", err))
325 }
326 case MsgStopPipeline:
327 log.Infow("Coordinator received MsgStopPipeline", "reason", msg.Reason)
328 if err := c.handleStopPipeline(ctx, msg.Reason); err != nil {
329 return tracerr.Wrap(fmt.Errorf("Coordinator.handleStopPipeline: %w", err))
330 }
331 default:
332 log.Fatalw("Coordinator Unexpected Coordinator msg of type %T: %+v", msg, msg)
333 }
334 return nil
335}
336
337// Start the coordinator
338func (c *Coordinator) Start() {
339 if c.started {
340 log.Fatal("Coordinator already started")
341 }
342 c.started = true
343 c.wg.Add(1)
344 go func() {
345 c.txManager.Run(c.ctx)
346 c.wg.Done()
347 }()
348
349 c.wg.Add(1)
350 go func() {
351 waitDuration := time.Duration(longWaitDuration)
352 for {
353 select {
354 case <-c.ctx.Done():
355 log.Info("Coordinator done")
356 c.wg.Done()
357 return
358 case msg := <-c.msgCh:
359 if err := c.handleMsg(c.ctx, msg); c.ctx.Err() != nil {
360 continue
361 } else if err != nil {
362 log.Errorw("Coordinator.handleMsg", "err", err)
363 waitDuration = time.Duration(c.cfg.SyncRetryInterval)
364 continue
365 }
366 waitDuration = time.Duration(longWaitDuration)
367 case <-time.After(waitDuration):
368 if c.stats == nil {
369 waitDuration = time.Duration(longWaitDuration)
370 continue
371 }
372 if err := c.syncStats(c.ctx, c.stats); c.ctx.Err() != nil {
373 continue
374 } else if err != nil {
375 log.Errorw("Coordinator.syncStats", "err", err)
376 waitDuration = time.Duration(c.cfg.SyncRetryInterval)
377 continue
378 }
379 waitDuration = time.Duration(longWaitDuration)
380 }
381 }
382 }()
383}
384
385const stopCtxTimeout = 200 * time.Millisecond
386
387// Stop the coordinator
388func (c *Coordinator) Stop() {
389 if !c.started {
390 log.Fatal("Coordinator already stopped")
391 }
392 c.started = false
393 log.Infow("Stopping Coordinator...")
394 c.cancel()
395 c.wg.Wait()
396 if c.pipeline != nil {
397 ctx, cancel := context.WithTimeout(context.Background(), stopCtxTimeout)
398 defer cancel()
399 c.pipeline.Stop(ctx)
400 c.pipeline = nil
401 }
402}
403
404// TxManager handles everything related to ethereum transactions: It makes the
405// call to forge, waits for transaction confirmation, and keeps checking them
406// until a number of confirmed blocks have passed.
407type TxManager struct {
408 cfg Config
409 ethClient eth.ClientInterface
410 l2DB *l2db.L2DB // Used only to mark forged txs as forged in the L2DB
411 coord *Coordinator // Used only to send messages to stop the pipeline
412 batchCh chan *BatchInfo
413 lastBlockCh chan int64
414 queue []*BatchInfo
415 lastBlock int64
416 // lastConfirmedBatch stores the last BatchNum that who's forge call was confirmed
417 lastConfirmedBatch common.BatchNum
418}
419
420// NewTxManager creates a new TxManager
421func NewTxManager(cfg *Config, ethClient eth.ClientInterface, l2DB *l2db.L2DB,
422 coord *Coordinator) *TxManager {
423 return &TxManager{
424 cfg: *cfg,
425 ethClient: ethClient,
426 l2DB: l2DB,
427 coord: coord,
428 batchCh: make(chan *BatchInfo, queueLen),
429 lastBlockCh: make(chan int64, queueLen),
430 lastBlock: -1,
431 }
432}
433
434// AddBatch is a thread safe method to pass a new batch TxManager to be sent to
435// the smart contract via the forge call
436func (t *TxManager) AddBatch(batchInfo *BatchInfo) {
437 t.batchCh <- batchInfo
438}
439
440// SetLastBlock is a thread safe method to pass the lastBlock to the TxManager
441func (t *TxManager) SetLastBlock(lastBlock int64) {
442 t.lastBlockCh <- lastBlock
443}
444
445func (t *TxManager) rollupForgeBatch(ctx context.Context, batchInfo *BatchInfo) error {
446 batchInfo.Debug.Status = StatusSent
447 batchInfo.Debug.SendBlockNum = t.lastBlock + 1
448 batchInfo.Debug.SendTimestamp = time.Now()
449 batchInfo.Debug.StartToSendDelay = batchInfo.Debug.SendTimestamp.Sub(
450 batchInfo.Debug.StartTimestamp).Seconds()
451 var ethTx *types.Transaction
452 var err error
453 for attempt := 0; attempt < t.cfg.EthClientAttempts; attempt++ {
454 ethTx, err = t.ethClient.RollupForgeBatch(batchInfo.ForgeBatchArgs)
455 if err != nil {
456 if strings.Contains(err.Error(), common.AuctionErrMsgCannotForge) {
457 log.Debugw("TxManager ethClient.RollupForgeBatch", "err", err,
458 "block", t.lastBlock+1)
459 return tracerr.Wrap(err)
460 }
461 log.Errorw("TxManager ethClient.RollupForgeBatch",
462 "attempt", attempt, "err", err, "block", t.lastBlock+1,
463 "batchNum", batchInfo.BatchNum)
464 } else {
465 break
466 }
467 select {
468 case <-ctx.Done():
469 return tracerr.Wrap(common.ErrDone)
470 case <-time.After(t.cfg.EthClientAttemptsDelay):
471 }
472 }
473 if err != nil {
474 return tracerr.Wrap(fmt.Errorf("reached max attempts for ethClient.RollupForgeBatch: %w", err))
475 }
476 batchInfo.EthTx = ethTx
477 log.Infow("TxManager ethClient.RollupForgeBatch", "batch", batchInfo.BatchNum, "tx", ethTx.Hash().Hex())
478 t.cfg.debugBatchStore(batchInfo)
479 if err := t.l2DB.DoneForging(common.TxIDsFromL2Txs(batchInfo.L2Txs), batchInfo.BatchNum); err != nil {
480 return tracerr.Wrap(err)
481 }
482 return nil
483}
484
485func (t *TxManager) ethTransactionReceipt(ctx context.Context, batchInfo *BatchInfo) error {
486 txHash := batchInfo.EthTx.Hash()
487 var receipt *types.Receipt
488 var err error
489 for attempt := 0; attempt < t.cfg.EthClientAttempts; attempt++ {
490 receipt, err = t.ethClient.EthTransactionReceipt(ctx, txHash)
491 if ctx.Err() != nil {
492 continue
493 }
494 if err != nil {
495 log.Errorw("TxManager ethClient.EthTransactionReceipt",
496 "attempt", attempt, "err", err)
497 } else {
498 break
499 }
500 select {
501 case <-ctx.Done():
502 return tracerr.Wrap(common.ErrDone)
503 case <-time.After(t.cfg.EthClientAttemptsDelay):
504 }
505 }
506 if err != nil {
507 return tracerr.Wrap(fmt.Errorf("reached max attempts for ethClient.EthTransactionReceipt: %w", err))
508 }
509 batchInfo.Receipt = receipt
510 t.cfg.debugBatchStore(batchInfo)
511 return nil
512}
513
514func (t *TxManager) handleReceipt(batchInfo *BatchInfo) (*int64, error) {
515 receipt := batchInfo.Receipt
516 if receipt != nil {
517 if receipt.Status == types.ReceiptStatusFailed {
518 batchInfo.Debug.Status = StatusFailed
519 t.cfg.debugBatchStore(batchInfo)
520 log.Errorw("TxManager receipt status is failed", "receipt", receipt)
521 return nil, tracerr.Wrap(fmt.Errorf("ethereum transaction receipt statis is failed"))
522 } else if receipt.Status == types.ReceiptStatusSuccessful {
523 batchInfo.Debug.Status = StatusMined
524 batchInfo.Debug.MineBlockNum = receipt.BlockNumber.Int64()
525 batchInfo.Debug.StartToMineBlocksDelay = batchInfo.Debug.MineBlockNum -
526 batchInfo.Debug.StartBlockNum
527 t.cfg.debugBatchStore(batchInfo)
528 if batchInfo.BatchNum > t.lastConfirmedBatch {
529 t.lastConfirmedBatch = batchInfo.BatchNum
530 }
531 confirm := t.lastBlock - receipt.BlockNumber.Int64()
532 return &confirm, nil
533 }
534 }
535 return nil, nil
536}
537
538const longWaitDuration = 999 * time.Hour
539
540// Run the TxManager
541func (t *TxManager) Run(ctx context.Context) {
542 next := 0
543 waitDuration := time.Duration(longWaitDuration)
544
545 for {
546 select {
547 case <-ctx.Done():
548 log.Info("TxManager done")
549 return
550 case lastBlock := <-t.lastBlockCh:
551 t.lastBlock = lastBlock
552 case batchInfo := <-t.batchCh:
553 if err := t.rollupForgeBatch(ctx, batchInfo); ctx.Err() != nil {
554 continue
555 } else if err != nil {
556 t.coord.SendMsg(MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch call: %v", err)})
557 continue
558 }
559 log.Debugf("ethClient ForgeCall sent, batchNum: %d", batchInfo.BatchNum)
560 t.queue = append(t.queue, batchInfo)
561 waitDuration = t.cfg.TxManagerCheckInterval
562 case <-time.After(waitDuration):
563 if len(t.queue) == 0 {
564 continue
565 }
566 current := next
567 next = (current + 1) % len(t.queue)
568 batchInfo := t.queue[current]
569 if err := t.ethTransactionReceipt(ctx, batchInfo); ctx.Err() != nil {
570 continue
571 } else if err != nil { //nolint:staticcheck
572 // We can't get the receipt for the
573 // transaction, so we can't confirm if it was
574 // mined
575 t.coord.SendMsg(MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch receipt: %v", err)})
576 }
577
578 confirm, err := t.handleReceipt(batchInfo)
579 if err != nil { //nolint:staticcheck
580 // Transaction was rejected
581 t.coord.SendMsg(MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch reject: %v", err)})
582 }
583 if confirm != nil && *confirm >= t.cfg.ConfirmBlocks {
584 log.Debugw("TxManager tx for RollupForgeBatch confirmed",
585 "batch", batchInfo.BatchNum)
586 t.queue = append(t.queue[:current], t.queue[current+1:]...)
587 if len(t.queue) == 0 {
588 waitDuration = longWaitDuration
589 next = 0
590 } else {
591 next = current % len(t.queue)
592 }
593 }
594 }
595 }
596}
597
598type statsVars struct {
599 Stats synchronizer.Stats
600 Vars synchronizer.SCVariablesPtr
601}
602
603// Pipeline manages the forging of batches with parallel server proofs
604type Pipeline struct {
605 cfg Config
606 consts synchronizer.SCConsts
607
608 // state
609 batchNum common.BatchNum
610 lastScheduledL1BatchBlockNum int64
611 lastForgeL1TxsNum int64
612 started bool
613
614 proversPool *ProversPool
615 provers []prover.Client
616 txManager *TxManager
617 historyDB *historydb.HistoryDB
618 l2DB *l2db.L2DB
619 txSelector *txselector.TxSelector
620 batchBuilder *batchbuilder.BatchBuilder
621 purger *Purger
622
623 stats synchronizer.Stats
624 vars synchronizer.SCVariables
625 statsVarsCh chan statsVars
626
627 ctx context.Context
628 wg sync.WaitGroup
629 cancel context.CancelFunc
630}
631
632// NewPipeline creates a new Pipeline
633func NewPipeline(ctx context.Context,
634 cfg Config,
635 historyDB *historydb.HistoryDB,
636 l2DB *l2db.L2DB,
637 txSelector *txselector.TxSelector,
638 batchBuilder *batchbuilder.BatchBuilder,
639 purger *Purger,
640 txManager *TxManager,
641 provers []prover.Client,
642 scConsts *synchronizer.SCConsts,
643) (*Pipeline, error) {
644 proversPool := NewProversPool(len(provers))
645 proversPoolSize := 0
646 for _, prover := range provers {
647 if err := prover.WaitReady(ctx); err != nil {
648 log.Errorw("prover.WaitReady", "err", err)
649 } else {
650 proversPool.Add(prover)
651 proversPoolSize++
652 }
653 }
654 if proversPoolSize == 0 {
655 return nil, tracerr.Wrap(fmt.Errorf("no provers in the pool"))
656 }
657 return &Pipeline{
658 cfg: cfg,
659 historyDB: historyDB,
660 l2DB: l2DB,
661 txSelector: txSelector,
662 batchBuilder: batchBuilder,
663 provers: provers,
664 proversPool: proversPool,
665 purger: purger,
666 txManager: txManager,
667 consts: *scConsts,
668 statsVarsCh: make(chan statsVars, queueLen),
669 }, nil
670}
671
672// SetSyncStatsVars is a thread safe method to sets the synchronizer Stats
673func (p *Pipeline) SetSyncStatsVars(stats *synchronizer.Stats, vars *synchronizer.SCVariablesPtr) {
674 p.statsVarsCh <- statsVars{Stats: *stats, Vars: *vars}
675}
676
677// reset pipeline state
678func (p *Pipeline) reset(batchNum common.BatchNum, lastForgeL1TxsNum int64,
679 stats *synchronizer.Stats, vars *synchronizer.SCVariables) error {
680 p.batchNum = batchNum
681 p.lastForgeL1TxsNum = lastForgeL1TxsNum
682 p.stats = *stats
683 p.vars = *vars
684 p.lastScheduledL1BatchBlockNum = 0
685
686 err := p.txSelector.Reset(p.batchNum)
687 if err != nil {
688 return tracerr.Wrap(err)
689 }
690 err = p.batchBuilder.Reset(p.batchNum, true)
691 if err != nil {
692 return tracerr.Wrap(err)
693 }
694 return nil
695}
696
697func (p *Pipeline) syncSCVars(vars synchronizer.SCVariablesPtr) {
698 if vars.Rollup != nil {
699 p.vars.Rollup = *vars.Rollup
700 }
701 if vars.Auction != nil {
702 p.vars.Auction = *vars.Auction
703 }
704 if vars.WDelayer != nil {
705 p.vars.WDelayer = *vars.WDelayer
706 }
707}
708
709// Start the forging pipeline
710func (p *Pipeline) Start(batchNum common.BatchNum, lastForgeL1TxsNum int64,
711 stats *synchronizer.Stats, vars *synchronizer.SCVariables) error {
712 if p.started {
713 log.Fatal("Pipeline already started")
714 }
715 p.started = true
716
717 if err := p.reset(batchNum, lastForgeL1TxsNum, stats, vars); err != nil {
718 return tracerr.Wrap(err)
719 }
720 p.ctx, p.cancel = context.WithCancel(context.Background())
721
722 queueSize := 1
723 batchChSentServerProof := make(chan *BatchInfo, queueSize)
724
725 p.wg.Add(1)
726 go func() {
727 for {
728 select {
729 case <-p.ctx.Done():
730 log.Info("Pipeline forgeBatch loop done")
731 p.wg.Done()
732 return
733 case statsVars := <-p.statsVarsCh:
734 p.stats = statsVars.Stats
735 p.syncSCVars(statsVars.Vars)
736 default:
737 batchNum = p.batchNum + 1
738 batchInfo, err := p.forgeBatch(batchNum)
739 if p.ctx.Err() != nil {
740 continue
741 } else if err != nil {
742 log.Errorw("forgeBatch", "err", err)
743 continue
744 }
745 // 6. Wait for an available server proof (blocking call)
746 serverProof, err := p.proversPool.Get(p.ctx)
747 if p.ctx.Err() != nil {
748 continue
749 } else if err != nil {
750 log.Errorw("proversPool.Get", "err", err)
751 continue
752 }
753 batchInfo.ServerProof = serverProof
754 if err := p.sendServerProof(p.ctx, batchInfo); p.ctx.Err() != nil {
755 continue
756 } else if err != nil {
757 log.Errorw("sendServerProof", "err", err)
758 batchInfo.ServerProof = nil
759 p.proversPool.Add(serverProof)
760 continue
761 }
762 p.batchNum = batchNum
763 batchChSentServerProof <- batchInfo
764 }
765 }
766 }()
767
768 p.wg.Add(1)
769 go func() {
770 for {
771 select {
772 case <-p.ctx.Done():
773 log.Info("Pipeline waitServerProofSendEth loop done")
774 p.wg.Done()
775 return
776 case batchInfo := <-batchChSentServerProof:
777 err := p.waitServerProof(p.ctx, batchInfo)
778 // We are done with this serverProof, add it back to the pool
779 p.proversPool.Add(batchInfo.ServerProof)
780 batchInfo.ServerProof = nil
781 if p.ctx.Err() != nil {
782 continue
783 }
784 if err != nil {
785 log.Errorw("waitServerProof", "err", err)
786 continue
787 }
788 p.txManager.AddBatch(batchInfo)
789 }
790 }
791 }()
792 return nil
793}
794
795// Stop the forging pipeline
796func (p *Pipeline) Stop(ctx context.Context) {
797 if !p.started {
798 log.Fatal("Pipeline already stopped")
799 }
800 p.started = false
801 log.Info("Stopping Pipeline...")
802 p.cancel()
803 p.wg.Wait()
804 for _, prover := range p.provers {
805 if err := prover.Cancel(ctx); ctx.Err() != nil {
806 continue
807 } else if err != nil {
808 log.Errorw("prover.Cancel", "err", err)
809 }
810 }
811}
812
813// sendServerProof sends the circuit inputs to the proof server
814func (p *Pipeline) sendServerProof(ctx context.Context, batchInfo *BatchInfo) error {
815 p.cfg.debugBatchStore(batchInfo)
816
817 // 7. Call the selected idle server proof with BatchBuilder output,
818 // save server proof info for batchNum
819 if err := batchInfo.ServerProof.CalculateProof(ctx, batchInfo.ZKInputs); err != nil {
820 return tracerr.Wrap(err)
821 }
822 return nil
823}
824
825// forgeBatch the next batch.
826func (p *Pipeline) forgeBatch(batchNum common.BatchNum) (*BatchInfo, error) {
827 // remove transactions from the pool that have been there for too long
828 _, err := p.purger.InvalidateMaybe(p.l2DB, p.txSelector.LocalAccountsDB(),
829 p.stats.Sync.LastBlock.Num, int64(batchNum))
830 if err != nil {
831 return nil, tracerr.Wrap(err)
832 }
833 _, err = p.purger.PurgeMaybe(p.l2DB, p.stats.Sync.LastBlock.Num, int64(batchNum))
834 if err != nil {
835 return nil, tracerr.Wrap(err)
836 }
837
838 batchInfo := BatchInfo{BatchNum: batchNum} // to accumulate metadata of the batch
839 batchInfo.Debug.StartTimestamp = time.Now()
840 batchInfo.Debug.StartBlockNum = p.stats.Eth.LastBlock.Num + 1
841
842 selectionCfg := &txselector.SelectionConfig{
843 MaxL1UserTxs: common.RollupConstMaxL1UserTx,
844 TxProcessorConfig: p.cfg.TxProcessorConfig,
845 }
846
847 var poolL2Txs []common.PoolL2Tx
848 // var feesInfo
849 var l1UserTxsExtra, l1CoordTxs []common.L1Tx
850 var auths [][]byte
851 var coordIdxs []common.Idx
852
853 // 1. Decide if we forge L2Tx or L1+L2Tx
854 if p.shouldL1L2Batch(&batchInfo) {
855 batchInfo.L1Batch = true
856 p.lastScheduledL1BatchBlockNum = p.stats.Eth.LastBlock.Num + 1
857 // 2a: L1+L2 txs
858 p.lastForgeL1TxsNum++
859 l1UserTxs, err := p.historyDB.GetUnforgedL1UserTxs(p.lastForgeL1TxsNum)
860 if err != nil {
861 return nil, tracerr.Wrap(err)
862 }
863 coordIdxs, auths, l1UserTxsExtra, l1CoordTxs, poolL2Txs, err =
864 p.txSelector.GetL1L2TxSelection(selectionCfg, l1UserTxs)
865 if err != nil {
866 return nil, tracerr.Wrap(err)
867 }
868 } else {
869 // 2b: only L2 txs
870 coordIdxs, auths, l1CoordTxs, poolL2Txs, err =
871 p.txSelector.GetL2TxSelection(selectionCfg)
872 if err != nil {
873 return nil, tracerr.Wrap(err)
874 }
875 l1UserTxsExtra = nil
876 }
877
878 // 3. Save metadata from TxSelector output for BatchNum
879 batchInfo.L1UserTxsExtra = l1UserTxsExtra
880 batchInfo.L1CoordTxs = l1CoordTxs
881 batchInfo.L1CoordinatorTxsAuths = auths
882 batchInfo.CoordIdxs = coordIdxs
883 batchInfo.VerifierIdx = p.cfg.VerifierIdx
884
885 if err := p.l2DB.StartForging(common.TxIDsFromPoolL2Txs(poolL2Txs), batchInfo.BatchNum); err != nil {
886 return nil, tracerr.Wrap(err)
887 }
888
889 // Invalidate transactions that become invalid beause of
890 // the poolL2Txs selected. Will mark as invalid the txs that have a
891 // (fromIdx, nonce) which already appears in the selected txs (includes
892 // all the nonces smaller than the current one)
893 err = p.l2DB.InvalidateOldNonces(idxsNonceFromPoolL2Txs(poolL2Txs), batchInfo.BatchNum)
894 if err != nil {
895 return nil, tracerr.Wrap(err)
896 }
897
898 // 4. Call BatchBuilder with TxSelector output
899 configBatch := &batchbuilder.ConfigBatch{
900 ForgerAddress: p.cfg.ForgerAddress,
901 TxProcessorConfig: p.cfg.TxProcessorConfig,
902 }
903 zkInputs, err := p.batchBuilder.BuildBatch(coordIdxs, configBatch, l1UserTxsExtra,
904 l1CoordTxs, poolL2Txs, nil)
905 if err != nil {
906 return nil, tracerr.Wrap(err)
907 }
908 l2Txs, err := common.PoolL2TxsToL2Txs(poolL2Txs) // NOTE: This is a big uggly, find a better way
909 if err != nil {
910 return nil, tracerr.Wrap(err)
911 }
912 batchInfo.L2Txs = l2Txs
913
914 // 5. Save metadata from BatchBuilder output for BatchNum
915 batchInfo.ZKInputs = zkInputs
916 batchInfo.Debug.Status = StatusForged
917 p.cfg.debugBatchStore(&batchInfo)
918
919 return &batchInfo, nil
920}
921
922// waitServerProof gets the generated zkProof & sends it to the SmartContract
923func (p *Pipeline) waitServerProof(ctx context.Context, batchInfo *BatchInfo) error {
924 proof, pubInputs, err := batchInfo.ServerProof.GetProof(ctx) // blocking call, until not resolved don't continue. Returns when the proof server has calculated the proof
925 if err != nil {
926 return tracerr.Wrap(err)
927 }
928 batchInfo.Proof = proof
929 batchInfo.PublicInputs = pubInputs
930 batchInfo.ForgeBatchArgs = prepareForgeBatchArgs(batchInfo)
931 batchInfo.Debug.Status = StatusProof
932 p.cfg.debugBatchStore(batchInfo)
933 return nil
934}
935
936func (p *Pipeline) shouldL1L2Batch(batchInfo *BatchInfo) bool {
937 // Take the lastL1BatchBlockNum as the biggest between the last
938 // scheduled one, and the synchronized one.
939 lastL1BatchBlockNum := p.lastScheduledL1BatchBlockNum
940 if p.stats.Sync.LastL1BatchBlock > lastL1BatchBlockNum {
941 lastL1BatchBlockNum = p.stats.Sync.LastL1BatchBlock
942 }
943 // Set Debug information
944 batchInfo.Debug.LastScheduledL1BatchBlockNum = p.lastScheduledL1BatchBlockNum
945 batchInfo.Debug.LastL1BatchBlock = p.stats.Sync.LastL1BatchBlock
946 batchInfo.Debug.LastL1BatchBlockDelta = p.stats.Eth.LastBlock.Num + 1 - lastL1BatchBlockNum
947 batchInfo.Debug.L1BatchBlockScheduleDeadline =
948 int64(float64(p.vars.Rollup.ForgeL1L2BatchTimeout-1) * p.cfg.L1BatchTimeoutPerc)
949 // Return true if we have passed the l1BatchTimeoutPerc portion of the
950 // range before the l1batch timeout.
951 return p.stats.Eth.LastBlock.Num+1-lastL1BatchBlockNum >=
952 int64(float64(p.vars.Rollup.ForgeL1L2BatchTimeout-1)*p.cfg.L1BatchTimeoutPerc)
953}
954
955func prepareForgeBatchArgs(batchInfo *BatchInfo) *eth.RollupForgeBatchArgs {
956 proof := batchInfo.Proof
957 zki := batchInfo.ZKInputs
958 return ð.RollupForgeBatchArgs{
959 NewLastIdx: int64(zki.Metadata.NewLastIdxRaw),
960 NewStRoot: zki.Metadata.NewStateRootRaw.BigInt(),
961 NewExitRoot: zki.Metadata.NewExitRootRaw.BigInt(),
962 L1UserTxs: batchInfo.L1UserTxsExtra,
963 L1CoordinatorTxs: batchInfo.L1CoordTxs,
964 L1CoordinatorTxsAuths: batchInfo.L1CoordinatorTxsAuths,
965 L2TxsData: batchInfo.L2Txs,
966 FeeIdxCoordinator: batchInfo.CoordIdxs,
967 // Circuit selector
968 VerifierIdx: batchInfo.VerifierIdx,
969 L1Batch: batchInfo.L1Batch,
970 ProofA: [2]*big.Int{proof.PiA[0], proof.PiA[1]},
971 ProofB: [2][2]*big.Int{
972 {proof.PiB[0][0], proof.PiB[0][1]},
973 {proof.PiB[1][0], proof.PiB[1][1]},
974 },
975 ProofC: [2]*big.Int{proof.PiC[0], proof.PiC[1]},
976 }
977}