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