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