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