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