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