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