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)
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); 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)
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 var ethTx *types.Transaction
447 var err error
448 for attempt := 0; attempt < t.cfg.EthClientAttempts; attempt++ {
449 ethTx, err = t.ethClient.RollupForgeBatch(batchInfo.ForgeBatchArgs)
450 if err != nil {
451 if strings.Contains(err.Error(), common.AuctionErrMsgCannotForge) {
452 log.Debugw("TxManager ethClient.RollupForgeBatch", "err", err,
453 "block", t.lastBlock)
454 return tracerr.Wrap(err)
455 }
456 log.Errorw("TxManager ethClient.RollupForgeBatch",
457 "attempt", attempt, "err", err, "block", t.lastBlock,
458 "batchNum", batchInfo.BatchNum)
459 } else {
460 break
461 }
462 select {
463 case <-ctx.Done():
464 return tracerr.Wrap(common.ErrDone)
465 case <-time.After(t.cfg.EthClientAttemptsDelay):
466 }
467 }
468 if err != nil {
469 return tracerr.Wrap(fmt.Errorf("reached max attempts for ethClient.RollupForgeBatch: %w", err))
470 }
471 batchInfo.EthTx = ethTx
472 batchInfo.Status = StatusSent
473 log.Infow("TxManager ethClient.RollupForgeBatch", "batch", batchInfo.BatchNum, "tx", ethTx.Hash().Hex())
474 t.cfg.debugBatchStore(batchInfo)
475 if err := t.l2DB.DoneForging(common.TxIDsFromL2Txs(batchInfo.L2Txs), batchInfo.BatchNum); err != nil {
476 return tracerr.Wrap(err)
477 }
478 return nil
479}
480
481func (t *TxManager) ethTransactionReceipt(ctx context.Context, batchInfo *BatchInfo) error {
482 txHash := batchInfo.EthTx.Hash()
483 var receipt *types.Receipt
484 var err error
485 for attempt := 0; attempt < t.cfg.EthClientAttempts; attempt++ {
486 receipt, err = t.ethClient.EthTransactionReceipt(ctx, txHash)
487 if ctx.Err() != nil {
488 continue
489 }
490 if err != nil {
491 log.Errorw("TxManager ethClient.EthTransactionReceipt",
492 "attempt", attempt, "err", err)
493 } else {
494 break
495 }
496 select {
497 case <-ctx.Done():
498 return tracerr.Wrap(common.ErrDone)
499 case <-time.After(t.cfg.EthClientAttemptsDelay):
500 }
501 }
502 if err != nil {
503 return tracerr.Wrap(fmt.Errorf("reached max attempts for ethClient.EthTransactionReceipt: %w", err))
504 }
505 batchInfo.Receipt = receipt
506 t.cfg.debugBatchStore(batchInfo)
507 return nil
508}
509
510func (t *TxManager) handleReceipt(batchInfo *BatchInfo) (*int64, error) {
511 receipt := batchInfo.Receipt
512 if receipt != nil {
513 if receipt.Status == types.ReceiptStatusFailed {
514 batchInfo.Status = StatusFailed
515 t.cfg.debugBatchStore(batchInfo)
516 log.Errorw("TxManager receipt status is failed", "receipt", receipt)
517 return nil, tracerr.Wrap(fmt.Errorf("ethereum transaction receipt statis is failed"))
518 } else if receipt.Status == types.ReceiptStatusSuccessful {
519 batchInfo.Status = StatusMined
520 t.cfg.debugBatchStore(batchInfo)
521 if batchInfo.BatchNum > t.lastConfirmedBatch {
522 t.lastConfirmedBatch = batchInfo.BatchNum
523 }
524 confirm := t.lastBlock - receipt.BlockNumber.Int64()
525 return &confirm, nil
526 }
527 }
528 return nil, nil
529}
530
531const longWaitDuration = 999 * time.Hour
532
533// Run the TxManager
534func (t *TxManager) Run(ctx context.Context) {
535 next := 0
536 waitDuration := time.Duration(longWaitDuration)
537
538 for {
539 select {
540 case <-ctx.Done():
541 log.Info("TxManager done")
542 return
543 case lastBlock := <-t.lastBlockCh:
544 t.lastBlock = lastBlock
545 case batchInfo := <-t.batchCh:
546 if err := t.rollupForgeBatch(ctx, batchInfo); ctx.Err() != nil {
547 continue
548 } else if err != nil {
549 t.coord.SendMsg(MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch call: %v", err)})
550 continue
551 }
552 log.Debugf("ethClient ForgeCall sent, batchNum: %d", batchInfo.BatchNum)
553 t.queue = append(t.queue, batchInfo)
554 waitDuration = t.cfg.TxManagerCheckInterval
555 case <-time.After(waitDuration):
556 if len(t.queue) == 0 {
557 continue
558 }
559 current := next
560 next = (current + 1) % len(t.queue)
561 batchInfo := t.queue[current]
562 if err := t.ethTransactionReceipt(ctx, batchInfo); ctx.Err() != nil {
563 continue
564 } else if err != nil { //nolint:staticcheck
565 // We can't get the receipt for the
566 // transaction, so we can't confirm if it was
567 // mined
568 t.coord.SendMsg(MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch receipt: %v", err)})
569 }
570
571 confirm, err := t.handleReceipt(batchInfo)
572 if err != nil { //nolint:staticcheck
573 // Transaction was rejected
574 t.coord.SendMsg(MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch reject: %v", err)})
575 }
576 if confirm != nil && *confirm >= t.cfg.ConfirmBlocks {
577 log.Debugw("TxManager tx for RollupForgeBatch confirmed",
578 "batch", batchInfo.BatchNum)
579 t.queue = append(t.queue[:current], t.queue[current+1:]...)
580 if len(t.queue) == 0 {
581 waitDuration = longWaitDuration
582 next = 0
583 } else {
584 next = current % len(t.queue)
585 }
586 }
587 }
588 }
589}
590
591type statsVars struct {
592 Stats synchronizer.Stats
593 Vars synchronizer.SCVariablesPtr
594}
595
596// Pipeline manages the forging of batches with parallel server proofs
597type Pipeline struct {
598 cfg Config
599 consts synchronizer.SCConsts
600
601 // state
602 batchNum common.BatchNum
603 lastScheduledL1BatchBlockNum int64
604 lastForgeL1TxsNum int64
605 started bool
606
607 proversPool *ProversPool
608 provers []prover.Client
609 txManager *TxManager
610 historyDB *historydb.HistoryDB
611 l2DB *l2db.L2DB
612 txSelector *txselector.TxSelector
613 batchBuilder *batchbuilder.BatchBuilder
614 purger *Purger
615
616 stats synchronizer.Stats
617 vars synchronizer.SCVariables
618 statsVarsCh chan statsVars
619
620 ctx context.Context
621 wg sync.WaitGroup
622 cancel context.CancelFunc
623}
624
625// NewPipeline creates a new Pipeline
626func NewPipeline(ctx context.Context,
627 cfg Config,
628 historyDB *historydb.HistoryDB,
629 l2DB *l2db.L2DB,
630 txSelector *txselector.TxSelector,
631 batchBuilder *batchbuilder.BatchBuilder,
632 purger *Purger,
633 txManager *TxManager,
634 provers []prover.Client,
635 scConsts *synchronizer.SCConsts,
636) (*Pipeline, error) {
637 proversPool := NewProversPool(len(provers))
638 proversPoolSize := 0
639 for _, prover := range provers {
640 if err := prover.WaitReady(ctx); err != nil {
641 log.Errorw("prover.WaitReady", "err", err)
642 } else {
643 proversPool.Add(prover)
644 proversPoolSize++
645 }
646 }
647 if proversPoolSize == 0 {
648 return nil, tracerr.Wrap(fmt.Errorf("no provers in the pool"))
649 }
650 return &Pipeline{
651 cfg: cfg,
652 historyDB: historyDB,
653 l2DB: l2DB,
654 txSelector: txSelector,
655 batchBuilder: batchBuilder,
656 provers: provers,
657 proversPool: proversPool,
658 purger: purger,
659 txManager: txManager,
660 consts: *scConsts,
661 statsVarsCh: make(chan statsVars, queueLen),
662 }, nil
663}
664
665// SetSyncStatsVars is a thread safe method to sets the synchronizer Stats
666func (p *Pipeline) SetSyncStatsVars(stats *synchronizer.Stats, vars *synchronizer.SCVariablesPtr) {
667 p.statsVarsCh <- statsVars{Stats: *stats, Vars: *vars}
668}
669
670// reset pipeline state
671func (p *Pipeline) reset(batchNum common.BatchNum, lastForgeL1TxsNum int64,
672 stats *synchronizer.Stats, vars *synchronizer.SCVariables) error {
673 p.batchNum = batchNum
674 p.lastForgeL1TxsNum = lastForgeL1TxsNum
675 p.stats = *stats
676 p.vars = *vars
677 p.lastScheduledL1BatchBlockNum = 0
678
679 err := p.txSelector.Reset(p.batchNum)
680 if err != nil {
681 return tracerr.Wrap(err)
682 }
683 err = p.batchBuilder.Reset(p.batchNum, true)
684 if err != nil {
685 return tracerr.Wrap(err)
686 }
687 return nil
688}
689
690func (p *Pipeline) syncSCVars(vars synchronizer.SCVariablesPtr) {
691 if vars.Rollup != nil {
692 p.vars.Rollup = *vars.Rollup
693 }
694 if vars.Auction != nil {
695 p.vars.Auction = *vars.Auction
696 }
697 if vars.WDelayer != nil {
698 p.vars.WDelayer = *vars.WDelayer
699 }
700}
701
702// Start the forging pipeline
703func (p *Pipeline) Start(batchNum common.BatchNum, lastForgeL1TxsNum int64,
704 stats *synchronizer.Stats, vars *synchronizer.SCVariables) error {
705 if p.started {
706 log.Fatal("Pipeline already started")
707 }
708 p.started = true
709
710 if err := p.reset(batchNum, lastForgeL1TxsNum, stats, vars); err != nil {
711 return tracerr.Wrap(err)
712 }
713 p.ctx, p.cancel = context.WithCancel(context.Background())
714
715 queueSize := 1
716 batchChSentServerProof := make(chan *BatchInfo, queueSize)
717
718 p.wg.Add(1)
719 go func() {
720 for {
721 select {
722 case <-p.ctx.Done():
723 log.Info("Pipeline forgeBatch loop done")
724 p.wg.Done()
725 return
726 case statsVars := <-p.statsVarsCh:
727 p.stats = statsVars.Stats
728 p.syncSCVars(statsVars.Vars)
729 default:
730 batchNum = p.batchNum + 1
731 batchInfo, err := p.forgeBatch(batchNum)
732 if p.ctx.Err() != nil {
733 continue
734 } else if err != nil {
735 log.Errorw("forgeBatch", "err", err)
736 continue
737 }
738 // 6. Wait for an available server proof (blocking call)
739 serverProof, err := p.proversPool.Get(p.ctx)
740 if p.ctx.Err() != nil {
741 continue
742 } else if err != nil {
743 log.Errorw("proversPool.Get", "err", err)
744 continue
745 }
746 batchInfo.ServerProof = serverProof
747 if err := p.sendServerProof(p.ctx, batchInfo); p.ctx.Err() != nil {
748 continue
749 } else if err != nil {
750 log.Errorw("sendServerProof", "err", err)
751 batchInfo.ServerProof = nil
752 p.proversPool.Add(serverProof)
753 continue
754 }
755 p.batchNum = batchNum
756 batchChSentServerProof <- batchInfo
757 }
758 }
759 }()
760
761 p.wg.Add(1)
762 go func() {
763 for {
764 select {
765 case <-p.ctx.Done():
766 log.Info("Pipeline waitServerProofSendEth loop done")
767 p.wg.Done()
768 return
769 case batchInfo := <-batchChSentServerProof:
770 err := p.waitServerProof(p.ctx, batchInfo)
771 // We are done with this serverProof, add it back to the pool
772 p.proversPool.Add(batchInfo.ServerProof)
773 batchInfo.ServerProof = nil
774 if p.ctx.Err() != nil {
775 continue
776 }
777 if err != nil {
778 log.Errorw("waitServerProof", "err", err)
779 continue
780 }
781 p.txManager.AddBatch(batchInfo)
782 }
783 }
784 }()
785 return nil
786}
787
788// Stop the forging pipeline
789func (p *Pipeline) Stop(ctx context.Context) {
790 if !p.started {
791 log.Fatal("Pipeline already stopped")
792 }
793 p.started = false
794 log.Info("Stopping Pipeline...")
795 p.cancel()
796 p.wg.Wait()
797 for _, prover := range p.provers {
798 if err := prover.Cancel(ctx); err != nil {
799 log.Errorw("prover.Cancel", "err", err)
800 }
801 }
802}
803
804// sendServerProof sends the circuit inputs to the proof server
805func (p *Pipeline) sendServerProof(ctx context.Context, batchInfo *BatchInfo) error {
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 if err := batchInfo.ServerProof.CalculateProof(ctx, batchInfo.ZKInputs); err != nil {
811 return tracerr.Wrap(err)
812 }
813 return nil
814}
815
816// forgeBatch the next batch.
817func (p *Pipeline) forgeBatch(batchNum common.BatchNum) (*BatchInfo, error) {
818 // remove transactions from the pool that have been there for too long
819 _, err := p.purger.InvalidateMaybe(p.l2DB, p.txSelector.LocalAccountsDB(),
820 p.stats.Sync.LastBlock.Num, int64(batchNum))
821 if err != nil {
822 return nil, tracerr.Wrap(err)
823 }
824 _, err = p.purger.PurgeMaybe(p.l2DB, p.stats.Sync.LastBlock.Num, int64(batchNum))
825 if err != nil {
826 return nil, tracerr.Wrap(err)
827 }
828
829 batchInfo := BatchInfo{BatchNum: batchNum} // to accumulate metadata of the batch
830
831 selectionCfg := &txselector.SelectionConfig{
832 MaxL1UserTxs: common.RollupConstMaxL1UserTx,
833 TxProcessorConfig: p.cfg.TxProcessorConfig,
834 }
835
836 var poolL2Txs []common.PoolL2Tx
837 // var feesInfo
838 var l1UserTxsExtra, l1CoordTxs []common.L1Tx
839 var auths [][]byte
840 var coordIdxs []common.Idx
841
842 // 1. Decide if we forge L2Tx or L1+L2Tx
843 if p.shouldL1L2Batch() {
844 batchInfo.L1Batch = true
845 p.lastScheduledL1BatchBlockNum = p.stats.Eth.LastBlock.Num
846 // 2a: L1+L2 txs
847 p.lastForgeL1TxsNum++
848 l1UserTxs, err := p.historyDB.GetUnforgedL1UserTxs(p.lastForgeL1TxsNum)
849 if err != nil {
850 return nil, tracerr.Wrap(err)
851 }
852 coordIdxs, auths, l1UserTxsExtra, l1CoordTxs, poolL2Txs, err =
853 p.txSelector.GetL1L2TxSelection(selectionCfg, l1UserTxs)
854 if err != nil {
855 return nil, tracerr.Wrap(err)
856 }
857 } else {
858 // 2b: only L2 txs
859 coordIdxs, auths, l1CoordTxs, poolL2Txs, err =
860 p.txSelector.GetL2TxSelection(selectionCfg)
861 if err != nil {
862 return nil, tracerr.Wrap(err)
863 }
864 l1UserTxsExtra = nil
865 }
866
867 // 3. Save metadata from TxSelector output for BatchNum
868 batchInfo.L1UserTxsExtra = l1UserTxsExtra
869 batchInfo.L1CoordTxs = l1CoordTxs
870 batchInfo.L1CoordinatorTxsAuths = auths
871 batchInfo.CoordIdxs = coordIdxs
872 batchInfo.VerifierIdx = p.cfg.VerifierIdx
873
874 if err := p.l2DB.StartForging(common.TxIDsFromPoolL2Txs(poolL2Txs), batchInfo.BatchNum); err != nil {
875 return nil, tracerr.Wrap(err)
876 }
877
878 // Invalidate transactions that become invalid beause of
879 // the poolL2Txs selected. Will mark as invalid the txs that have a
880 // (fromIdx, nonce) which already appears in the selected txs (includes
881 // all the nonces smaller than the current one)
882 err = p.l2DB.InvalidateOldNonces(idxsNonceFromPoolL2Txs(poolL2Txs), batchInfo.BatchNum)
883 if err != nil {
884 return nil, tracerr.Wrap(err)
885 }
886
887 // 4. Call BatchBuilder with TxSelector output
888 configBatch := &batchbuilder.ConfigBatch{
889 ForgerAddress: p.cfg.ForgerAddress,
890 TxProcessorConfig: p.cfg.TxProcessorConfig,
891 }
892 zkInputs, err := p.batchBuilder.BuildBatch(coordIdxs, configBatch, l1UserTxsExtra,
893 l1CoordTxs, poolL2Txs, nil)
894 if err != nil {
895 return nil, tracerr.Wrap(err)
896 }
897 l2Txs, err := common.PoolL2TxsToL2Txs(poolL2Txs) // NOTE: This is a big uggly, find a better way
898 if err != nil {
899 return nil, tracerr.Wrap(err)
900 }
901 batchInfo.L2Txs = l2Txs
902
903 // 5. Save metadata from BatchBuilder output for BatchNum
904 batchInfo.ZKInputs = zkInputs
905 batchInfo.Status = StatusForged
906 p.cfg.debugBatchStore(&batchInfo)
907
908 return &batchInfo, nil
909}
910
911// waitServerProof gets the generated zkProof & sends it to the SmartContract
912func (p *Pipeline) waitServerProof(ctx context.Context, batchInfo *BatchInfo) error {
913 proof, pubInputs, err := batchInfo.ServerProof.GetProof(ctx) // blocking call, until not resolved don't continue. Returns when the proof server has calculated the proof
914 if err != nil {
915 return tracerr.Wrap(err)
916 }
917 batchInfo.Proof = proof
918 batchInfo.PublicInputs = pubInputs
919 batchInfo.ForgeBatchArgs = prepareForgeBatchArgs(batchInfo)
920 batchInfo.Status = StatusProof
921 p.cfg.debugBatchStore(batchInfo)
922 return nil
923}
924
925func (p *Pipeline) shouldL1L2Batch() bool {
926 // Take the lastL1BatchBlockNum as the biggest between the last
927 // scheduled one, and the synchronized one.
928 lastL1BatchBlockNum := p.lastScheduledL1BatchBlockNum
929 if p.stats.Sync.LastL1BatchBlock > lastL1BatchBlockNum {
930 lastL1BatchBlockNum = p.stats.Sync.LastL1BatchBlock
931 }
932 // Return true if we have passed the l1BatchTimeoutPerc portion of the
933 // range before the l1batch timeout.
934 if p.stats.Eth.LastBlock.Num-lastL1BatchBlockNum >=
935 int64(float64(p.vars.Rollup.ForgeL1L2BatchTimeout-1)*p.cfg.L1BatchTimeoutPerc) {
936 return true
937 }
938 return false
939}
940
941func prepareForgeBatchArgs(batchInfo *BatchInfo) *eth.RollupForgeBatchArgs {
942 proof := batchInfo.Proof
943 zki := batchInfo.ZKInputs
944 return ð.RollupForgeBatchArgs{
945 NewLastIdx: int64(zki.Metadata.NewLastIdxRaw),
946 NewStRoot: zki.Metadata.NewStateRootRaw.BigInt(),
947 NewExitRoot: zki.Metadata.NewExitRootRaw.BigInt(),
948 L1UserTxs: batchInfo.L1UserTxsExtra,
949 L1CoordinatorTxs: batchInfo.L1CoordTxs,
950 L1CoordinatorTxsAuths: batchInfo.L1CoordinatorTxsAuths,
951 L2TxsData: batchInfo.L2Txs,
952 FeeIdxCoordinator: batchInfo.CoordIdxs,
953 // Circuit selector
954 VerifierIdx: batchInfo.VerifierIdx,
955 L1Batch: batchInfo.L1Batch,
956 ProofA: [2]*big.Int{proof.PiA[0], proof.PiA[1]},
957 ProofB: [2][2]*big.Int{
958 {proof.PiB[0][0], proof.PiB[0][1]},
959 {proof.PiB[1][0], proof.PiB[1][1]},
960 },
961 ProofC: [2]*big.Int{proof.PiC[0], proof.PiC[1]},
962 }
963}