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