1package coordinator
  2
  3import (
  4	"context"
  5	"database/sql"
  6	"fmt"
  7	"math/big"
  8	"strconv"
  9	"sync"
 10	"time"
 11
 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/metric"
 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
 25type statsVars struct {
 26	Stats synchronizer.Stats
 27	Vars  common.SCVariablesPtr
 28}
 29
 30type state struct {
 31	batchNum                     common.BatchNum
 32	lastScheduledL1BatchBlockNum int64
 33	lastForgeL1TxsNum            int64
 34	lastSlotForged               int64
 35}
 36
 37// Pipeline manages the forging of batches with parallel server proofs
 38type Pipeline struct {
 39	num    int
 40	cfg    Config
 41	consts common.SCConsts
 42
 43	// state
 44	state         state
 45	started       bool
 46	rw            sync.RWMutex
 47	errAtBatchNum common.BatchNum
 48	lastForgeTime time.Time
 49
 50	proversPool           *ProversPool
 51	provers               []prover.Client
 52	coord                 *Coordinator
 53	txManager             *TxManager
 54	historyDB             *historydb.HistoryDB
 55	l2DB                  *l2db.L2DB
 56	txSelector            *txselector.TxSelector
 57	batchBuilder          *batchbuilder.BatchBuilder
 58	mutexL2DBUpdateDelete *sync.Mutex
 59	purger                *Purger
 60
 61	stats       synchronizer.Stats
 62	vars        common.SCVariables
 63	statsVarsCh chan statsVars
 64
 65	ctx    context.Context
 66	wg     sync.WaitGroup
 67	cancel context.CancelFunc
 68}
 69
 70func (p *Pipeline) setErrAtBatchNum(batchNum common.BatchNum) {
 71	p.rw.Lock()
 72	defer p.rw.Unlock()
 73	p.errAtBatchNum = batchNum
 74}
 75
 76func (p *Pipeline) getErrAtBatchNum() common.BatchNum {
 77	p.rw.RLock()
 78	defer p.rw.RUnlock()
 79	return p.errAtBatchNum
 80}
 81
 82// NewPipeline creates a new Pipeline
 83func NewPipeline(ctx context.Context,
 84	cfg Config,
 85	num int, // Pipeline sequential number
 86	historyDB *historydb.HistoryDB,
 87	l2DB *l2db.L2DB,
 88	txSelector *txselector.TxSelector,
 89	batchBuilder *batchbuilder.BatchBuilder,
 90	mutexL2DBUpdateDelete *sync.Mutex,
 91	purger *Purger,
 92	coord *Coordinator,
 93	txManager *TxManager,
 94	provers []prover.Client,
 95	scConsts *common.SCConsts,
 96) (*Pipeline, error) {
 97	proversPool := NewProversPool(len(provers))
 98	proversPoolSize := 0
 99	for _, prover := range provers {
100		if err := prover.WaitReady(ctx); err != nil {
101			log.Errorw("prover.WaitReady", "err", err)
102		} else {
103			proversPool.Add(ctx, prover)
104			proversPoolSize++
105		}
106	}
107	if proversPoolSize == 0 {
108		return nil, tracerr.Wrap(fmt.Errorf("no provers in the pool"))
109	}
110	return &Pipeline{
111		num:                   num,
112		cfg:                   cfg,
113		historyDB:             historyDB,
114		l2DB:                  l2DB,
115		txSelector:            txSelector,
116		batchBuilder:          batchBuilder,
117		provers:               provers,
118		proversPool:           proversPool,
119		mutexL2DBUpdateDelete: mutexL2DBUpdateDelete,
120		purger:                purger,
121		coord:                 coord,
122		txManager:             txManager,
123		consts:                *scConsts,
124		statsVarsCh:           make(chan statsVars, queueLen),
125	}, nil
126}
127
128// SetSyncStatsVars is a thread safe method to sets the synchronizer Stats
129func (p *Pipeline) SetSyncStatsVars(ctx context.Context, stats *synchronizer.Stats,
130	vars *common.SCVariablesPtr) {
131	select {
132	case p.statsVarsCh <- statsVars{Stats: *stats, Vars: *vars}:
133	case <-ctx.Done():
134	}
135}
136
137// reset pipeline state
138func (p *Pipeline) reset(batchNum common.BatchNum,
139	stats *synchronizer.Stats, vars *common.SCVariables) error {
140	p.state = state{
141		batchNum:                     batchNum,
142		lastForgeL1TxsNum:            stats.Sync.LastForgeL1TxsNum,
143		lastScheduledL1BatchBlockNum: 0,
144		lastSlotForged:               -1,
145	}
146	p.stats = *stats
147	p.vars = *vars
148
149	// Reset the StateDB in TxSelector and BatchBuilder from the
150	// synchronizer only if the checkpoint we reset from either:
151	// a. Doesn't exist in the TxSelector/BatchBuilder
152	// b. The batch has already been synced by the synchronizer and has a
153	//    different MTRoot than the BatchBuilder
154	// Otherwise, reset from the local checkpoint.
155
156	// First attempt to reset from local checkpoint if such checkpoint exists
157	existsTxSelector, err := p.txSelector.LocalAccountsDB().CheckpointExists(p.state.batchNum)
158	if err != nil {
159		return tracerr.Wrap(err)
160	}
161	fromSynchronizerTxSelector := !existsTxSelector
162	if err := p.txSelector.Reset(p.state.batchNum, fromSynchronizerTxSelector); err != nil {
163		return tracerr.Wrap(err)
164	}
165	existsBatchBuilder, err := p.batchBuilder.LocalStateDB().CheckpointExists(p.state.batchNum)
166	if err != nil {
167		return tracerr.Wrap(err)
168	}
169	fromSynchronizerBatchBuilder := !existsBatchBuilder
170	if err := p.batchBuilder.Reset(p.state.batchNum, fromSynchronizerBatchBuilder); err != nil {
171		return tracerr.Wrap(err)
172	}
173
174	// After reset, check that if the batch exists in the historyDB, the
175	// stateRoot matches with the local one, if not, force a reset from
176	// synchronizer
177	batch, err := p.historyDB.GetBatch(p.state.batchNum)
178	if tracerr.Unwrap(err) == sql.ErrNoRows {
179		// nothing to do
180	} else if err != nil {
181		return tracerr.Wrap(err)
182	} else {
183		localStateRoot := p.batchBuilder.LocalStateDB().MT.Root().BigInt()
184		if batch.StateRoot.Cmp(localStateRoot) != 0 {
185			log.Debugw("localStateRoot (%v) != historyDB stateRoot (%v).  "+
186				"Forcing reset from Synchronizer", localStateRoot, batch.StateRoot)
187			// StateRoot from synchronizer doesn't match StateRoot
188			// from batchBuilder, force a reset from synchronizer
189			if err := p.txSelector.Reset(p.state.batchNum, true); err != nil {
190				return tracerr.Wrap(err)
191			}
192			if err := p.batchBuilder.Reset(p.state.batchNum, true); err != nil {
193				return tracerr.Wrap(err)
194			}
195		}
196	}
197	return nil
198}
199
200func (p *Pipeline) syncSCVars(vars common.SCVariablesPtr) {
201	updateSCVars(&p.vars, vars)
202}
203
204// handleForgeBatch waits for an available proof server, calls p.forgeBatch to
205// forge the batch and get the zkInputs, and then  sends the zkInputs to the
206// selected proof server so that the proof computation begins.
207func (p *Pipeline) handleForgeBatch(ctx context.Context,
208	batchNum common.BatchNum) (batchInfo *BatchInfo, err error) {
209	// 1. Wait for an available serverProof (blocking call)
210	serverProof, err := p.proversPool.Get(ctx)
211	if ctx.Err() != nil {
212		return nil, ctx.Err()
213	} else if err != nil {
214		log.Errorw("proversPool.Get", "err", err)
215		return nil, tracerr.Wrap(err)
216	}
217	defer func() {
218		// If we encounter any error (notice that this function returns
219		// errors to notify that a batch is not forged not only because
220		// of unexpected errors but also due to benign causes), add the
221		// serverProof back to the pool
222		if err != nil {
223			p.proversPool.Add(ctx, serverProof)
224		}
225	}()
226
227	// 2. Forge the batch internally (make a selection of txs and prepare
228	// all the smart contract arguments)
229	var skipReason *string
230	p.mutexL2DBUpdateDelete.Lock()
231	batchInfo, skipReason, err = p.forgeBatch(batchNum)
232	p.mutexL2DBUpdateDelete.Unlock()
233	if ctx.Err() != nil {
234		return nil, ctx.Err()
235	} else if err != nil {
236		if tracerr.Unwrap(err) == errLastL1BatchNotSynced {
237			log.Warnw("forgeBatch: scheduled L1Batch too early", "err", err,
238				"lastForgeL1TxsNum", p.state.lastForgeL1TxsNum,
239				"syncLastForgeL1TxsNum", p.stats.Sync.LastForgeL1TxsNum)
240		} else {
241			log.Errorw("forgeBatch", "err", err)
242		}
243		return nil, tracerr.Wrap(err)
244	} else if skipReason != nil {
245		log.Debugw("skipping batch", "batch", batchNum, "reason", *skipReason)
246		return nil, tracerr.Wrap(errSkipBatchByPolicy)
247	}
248
249	// 3. Send the ZKInputs to the proof server
250	batchInfo.ServerProof = serverProof
251	batchInfo.ProofStart = time.Now()
252	if err := p.sendServerProof(ctx, batchInfo); ctx.Err() != nil {
253		return nil, ctx.Err()
254	} else if err != nil {
255		log.Errorw("sendServerProof", "err", err)
256		return nil, tracerr.Wrap(err)
257	}
258	return batchInfo, nil
259}
260
261// Start the forging pipeline
262func (p *Pipeline) Start(batchNum common.BatchNum,
263	stats *synchronizer.Stats, vars *common.SCVariables) error {
264	if p.started {
265		log.Fatal("Pipeline already started")
266	}
267	p.started = true
268
269	if err := p.reset(batchNum, stats, vars); err != nil {
270		return tracerr.Wrap(err)
271	}
272	p.ctx, p.cancel = context.WithCancel(context.Background())
273
274	queueSize := 1
275	batchChSentServerProof := make(chan *BatchInfo, queueSize)
276
277	p.wg.Add(1)
278	go func() {
279		timer := time.NewTimer(zeroDuration)
280		for {
281			select {
282			case <-p.ctx.Done():
283				log.Info("Pipeline forgeBatch loop done")
284				p.wg.Done()
285				return
286			case statsVars := <-p.statsVarsCh:
287				p.stats = statsVars.Stats
288				p.syncSCVars(statsVars.Vars)
289			case <-timer.C:
290				timer.Reset(p.cfg.ForgeRetryInterval)
291				// Once errAtBatchNum != 0, we stop forging
292				// batches because there's been an error and we
293				// wait for the pipeline to be stopped.
294				if p.getErrAtBatchNum() != 0 {
295					continue
296				}
297				batchNum = p.state.batchNum + 1
298				batchInfo, err := p.handleForgeBatch(p.ctx, batchNum)
299				if p.ctx.Err() != nil {
300					continue
301				} else if tracerr.Unwrap(err) == errLastL1BatchNotSynced ||
302					tracerr.Unwrap(err) == errSkipBatchByPolicy {
303					continue
304				} else if err != nil {
305					p.setErrAtBatchNum(batchNum)
306					p.coord.SendMsg(p.ctx, MsgStopPipeline{
307						Reason: fmt.Sprintf(
308							"Pipeline.handleForgBatch: %v", err),
309						FailedBatchNum: batchNum,
310					})
311					continue
312				}
313				p.lastForgeTime = time.Now()
314
315				p.state.batchNum = batchNum
316				select {
317				case batchChSentServerProof <- batchInfo:
318				case <-p.ctx.Done():
319				}
320				if !timer.Stop() {
321					<-timer.C
322				}
323				timer.Reset(zeroDuration)
324			}
325		}
326	}()
327
328	p.wg.Add(1)
329	go func() {
330		for {
331			select {
332			case <-p.ctx.Done():
333				log.Info("Pipeline waitServerProofSendEth loop done")
334				p.wg.Done()
335				return
336			case batchInfo := <-batchChSentServerProof:
337				// Once errAtBatchNum != 0, we stop forging
338				// batches because there's been an error and we
339				// wait for the pipeline to be stopped.
340				if p.getErrAtBatchNum() != 0 {
341					continue
342				}
343				err := p.waitServerProof(p.ctx, batchInfo)
344				if p.ctx.Err() != nil {
345					continue
346				} else if err != nil {
347					log.Errorw("waitServerProof", "err", err)
348					p.setErrAtBatchNum(batchInfo.BatchNum)
349					p.coord.SendMsg(p.ctx, MsgStopPipeline{
350						Reason: fmt.Sprintf(
351							"Pipeline.waitServerProof: %v", err),
352						FailedBatchNum: batchInfo.BatchNum,
353					})
354					continue
355				}
356				// We are done with this serverProof, add it back to the pool
357				p.proversPool.Add(p.ctx, batchInfo.ServerProof)
358				p.txManager.AddBatch(p.ctx, batchInfo)
359			}
360		}
361	}()
362	return nil
363}
364
365// Stop the forging pipeline
366func (p *Pipeline) Stop(ctx context.Context) {
367	if !p.started {
368		log.Fatal("Pipeline already stopped")
369	}
370	p.started = false
371	log.Info("Stopping Pipeline...")
372	p.cancel()
373	p.wg.Wait()
374	for _, prover := range p.provers {
375		if err := prover.Cancel(ctx); ctx.Err() != nil {
376			continue
377		} else if err != nil {
378			log.Errorw("prover.Cancel", "err", err)
379		}
380	}
381}
382
383// sendServerProof sends the circuit inputs to the proof server
384func (p *Pipeline) sendServerProof(ctx context.Context, batchInfo *BatchInfo) error {
385	p.cfg.debugBatchStore(batchInfo)
386
387	// 7. Call the selected idle server proof with BatchBuilder output,
388	// save server proof info for batchNum
389	if err := batchInfo.ServerProof.CalculateProof(ctx, batchInfo.ZKInputs); err != nil {
390		return tracerr.Wrap(err)
391	}
392	return nil
393}
394
395// slotCommitted returns true if the current slot has already been committed
396func (p *Pipeline) slotCommitted() bool {
397	// Synchronizer has synchronized a batch in the current slot (setting
398	// CurrentSlot.ForgerCommitment) or the pipeline has already
399	// internally-forged a batch in the current slot
400	return p.stats.Sync.Auction.CurrentSlot.ForgerCommitment ||
401		p.stats.Sync.Auction.CurrentSlot.SlotNum == p.state.lastSlotForged
402}
403
404// forgePolicySkipPreSelection is called before doing a tx selection in a batch to
405// determine by policy if we should forge the batch or not.  Returns true and
406// the reason when the forging of the batch must be skipped.
407func (p *Pipeline) forgePolicySkipPreSelection(now time.Time) (bool, string) {
408	// Check if the slot is not yet fulfilled
409	slotCommitted := p.slotCommitted()
410	if p.cfg.ForgeOncePerSlotIfTxs {
411		if slotCommitted {
412			return true, "cfg.ForgeOncePerSlotIfTxs = true and slot already committed"
413		}
414		return false, ""
415	}
416	// Determine if we must commit the slot
417	if !p.cfg.IgnoreSlotCommitment && !slotCommitted {
418		return false, ""
419	}
420
421	// If we haven't reached the ForgeDelay, skip forging the batch
422	if now.Sub(p.lastForgeTime) < p.cfg.ForgeDelay {
423		return true, "we haven't reached the forge delay"
424	}
425	return false, ""
426}
427
428// forgePolicySkipPostSelection is called after doing a tx selection in a batch to
429// determine by policy if we should forge the batch or not.  Returns true and
430// the reason when the forging of the batch must be skipped.
431func (p *Pipeline) forgePolicySkipPostSelection(now time.Time, l1UserTxsExtra, l1CoordTxs []common.L1Tx,
432	poolL2Txs []common.PoolL2Tx, batchInfo *BatchInfo) (bool, string, error) {
433	// Check if the slot is not yet fulfilled
434	slotCommitted := p.slotCommitted()
435
436	pendingTxs := true
437	if len(l1UserTxsExtra) == 0 && len(l1CoordTxs) == 0 && len(poolL2Txs) == 0 {
438		if batchInfo.L1Batch {
439			// Query the number of unforged L1UserTxs
440			// (either in a open queue or in a frozen
441			// not-yet-forged queue).
442			count, err := p.historyDB.GetUnforgedL1UserTxsCount()
443			if err != nil {
444				return false, "", err
445			}
446			// If there are future L1UserTxs, we forge a
447			// batch to advance the queues to be able to
448			// forge the L1UserTxs in the future.
449			// Otherwise, skip.
450			if count == 0 {
451				pendingTxs = false
452			}
453		} else {
454			pendingTxs = false
455		}
456	}
457
458	if p.cfg.ForgeOncePerSlotIfTxs {
459		if slotCommitted {
460			return true, "cfg.ForgeOncePerSlotIfTxs = true and slot already committed",
461				nil
462		}
463		if pendingTxs {
464			return false, "", nil
465		}
466		return true, "cfg.ForgeOncePerSlotIfTxs = true and no pending txs",
467			nil
468	}
469
470	// Determine if we must commit the slot
471	if !p.cfg.IgnoreSlotCommitment && !slotCommitted {
472		return false, "", nil
473	}
474
475	// check if there is no txs to forge, no l1UserTxs in the open queue to
476	// freeze and we haven't reached the ForgeNoTxsDelay
477	if now.Sub(p.lastForgeTime) < p.cfg.ForgeNoTxsDelay {
478		if !pendingTxs {
479			return true, "no txs to forge and we haven't reached the forge no txs delay",
480				nil
481		}
482	}
483	return false, "", nil
484}
485
486// forgeBatch forges the batchNum batch.
487func (p *Pipeline) forgeBatch(batchNum common.BatchNum) (batchInfo *BatchInfo,
488	skipReason *string, err error) {
489	// remove transactions from the pool that have been there for too long
490	_, err = p.purger.InvalidateMaybe(p.l2DB, p.txSelector.LocalAccountsDB(),
491		p.stats.Sync.LastBlock.Num, int64(batchNum))
492	if err != nil {
493		return nil, nil, tracerr.Wrap(err)
494	}
495	_, err = p.purger.PurgeMaybe(p.l2DB, p.stats.Sync.LastBlock.Num, int64(batchNum))
496	if err != nil {
497		return nil, nil, tracerr.Wrap(err)
498	}
499	// Structure to accumulate data and metadata of the batch
500	now := time.Now()
501	batchInfo = &BatchInfo{PipelineNum: p.num, BatchNum: batchNum}
502	batchInfo.Debug.StartTimestamp = now
503	batchInfo.Debug.StartBlockNum = p.stats.Eth.LastBlock.Num + 1
504
505	var poolL2Txs []common.PoolL2Tx
506	var discardedL2Txs []common.PoolL2Tx
507	var l1UserTxs, l1CoordTxs []common.L1Tx
508	var auths [][]byte
509	var coordIdxs []common.Idx
510
511	if skip, reason := p.forgePolicySkipPreSelection(now); skip {
512		return nil, &reason, nil
513	}
514
515	// 1. Decide if we forge L2Tx or L1+L2Tx
516	if p.shouldL1L2Batch(batchInfo) {
517		batchInfo.L1Batch = true
518		if p.state.lastForgeL1TxsNum != p.stats.Sync.LastForgeL1TxsNum {
519			return nil, nil, tracerr.Wrap(errLastL1BatchNotSynced)
520		}
521		// 2a: L1+L2 txs
522		_l1UserTxs, err := p.historyDB.GetUnforgedL1UserTxs(p.state.lastForgeL1TxsNum + 1)
523		if err != nil {
524			return nil, nil, tracerr.Wrap(err)
525		}
526		// l1UserFutureTxs are the l1UserTxs that are not being forged
527		// in the next batch, but that are also in the queue for the
528		// future batches
529		l1UserFutureTxs, err := p.historyDB.GetUnforgedL1UserFutureTxs(p.state.lastForgeL1TxsNum + 1)
530		if err != nil {
531			return nil, nil, tracerr.Wrap(err)
532		}
533
534		coordIdxs, auths, l1UserTxs, l1CoordTxs, poolL2Txs, discardedL2Txs, err =
535			p.txSelector.GetL1L2TxSelection(p.cfg.TxProcessorConfig, _l1UserTxs, l1UserFutureTxs)
536		if err != nil {
537			return nil, nil, tracerr.Wrap(err)
538		}
539	} else {
540		// get l1UserFutureTxs which are all the l1 pending in all the
541		// queues
542		l1UserFutureTxs, err := p.historyDB.GetUnforgedL1UserFutureTxs(p.state.lastForgeL1TxsNum) //nolint:gomnd
543		if err != nil {
544			return nil, nil, tracerr.Wrap(err)
545		}
546
547		// 2b: only L2 txs
548		coordIdxs, auths, l1CoordTxs, poolL2Txs, discardedL2Txs, err =
549			p.txSelector.GetL2TxSelection(p.cfg.TxProcessorConfig, l1UserFutureTxs)
550		if err != nil {
551			return nil, nil, tracerr.Wrap(err)
552		}
553		l1UserTxs = nil
554	}
555
556	if skip, reason, err := p.forgePolicySkipPostSelection(now,
557		l1UserTxs, l1CoordTxs, poolL2Txs, batchInfo); err != nil {
558		return nil, nil, tracerr.Wrap(err)
559	} else if skip {
560		if err := p.txSelector.Reset(batchInfo.BatchNum-1, false); err != nil {
561			return nil, nil, tracerr.Wrap(err)
562		}
563		return nil, &reason, tracerr.Wrap(err)
564	}
565
566	if batchInfo.L1Batch {
567		p.state.lastScheduledL1BatchBlockNum = p.stats.Eth.LastBlock.Num + 1
568		p.state.lastForgeL1TxsNum++
569	}
570
571	// 3.  Save metadata from TxSelector output for BatchNum
572	batchInfo.L1UserTxs = l1UserTxs
573	batchInfo.L1CoordTxs = l1CoordTxs
574	batchInfo.L1CoordinatorTxsAuths = auths
575	batchInfo.CoordIdxs = coordIdxs
576	batchInfo.VerifierIdx = p.cfg.VerifierIdx
577
578	if err := p.l2DB.StartForging(common.TxIDsFromPoolL2Txs(poolL2Txs),
579		batchInfo.BatchNum); err != nil {
580		return nil, nil, tracerr.Wrap(err)
581	}
582	if err := p.l2DB.UpdateTxsInfo(discardedL2Txs); err != nil {
583		return nil, nil, tracerr.Wrap(err)
584	}
585
586	// Invalidate transactions that become invalid because of
587	// the poolL2Txs selected.  Will mark as invalid the txs that have a
588	// (fromIdx, nonce) which already appears in the selected txs (includes
589	// all the nonces smaller than the current one)
590	err = p.l2DB.InvalidateOldNonces(idxsNonceFromPoolL2Txs(poolL2Txs), batchInfo.BatchNum)
591	if err != nil {
592		return nil, nil, tracerr.Wrap(err)
593	}
594
595	// 4. Call BatchBuilder with TxSelector output
596	configBatch := &batchbuilder.ConfigBatch{
597		TxProcessorConfig: p.cfg.TxProcessorConfig,
598	}
599	zkInputs, err := p.batchBuilder.BuildBatch(coordIdxs, configBatch, l1UserTxs,
600		l1CoordTxs, poolL2Txs)
601	if err != nil {
602		return nil, nil, tracerr.Wrap(err)
603	}
604	l2Txs, err := common.PoolL2TxsToL2Txs(poolL2Txs) // NOTE: This is a big uggly, find a better way
605	if err != nil {
606		return nil, nil, tracerr.Wrap(err)
607	}
608	batchInfo.L2Txs = l2Txs
609
610	// 5. Save metadata from BatchBuilder output for BatchNum
611	batchInfo.ZKInputs = zkInputs
612	batchInfo.Debug.Status = StatusForged
613	p.cfg.debugBatchStore(batchInfo)
614	log.Infow("Pipeline: batch forged internally", "batch", batchInfo.BatchNum)
615
616	p.state.lastSlotForged = p.stats.Sync.Auction.CurrentSlot.SlotNum
617
618	return batchInfo, nil, nil
619}
620
621// waitServerProof gets the generated zkProof & sends it to the SmartContract
622func (p *Pipeline) waitServerProof(ctx context.Context, batchInfo *BatchInfo) error {
623	defer metric.MeasureDuration(metric.WaitServerProof, batchInfo.ProofStart,
624		batchInfo.BatchNum.BigInt().String(), strconv.Itoa(batchInfo.PipelineNum))
625
626	proof, pubInputs, err := batchInfo.ServerProof.GetProof(ctx) // blocking call,
627	// until not resolved don't continue. Returns when the proof server has calculated the proof
628	if err != nil {
629		return tracerr.Wrap(err)
630	}
631	batchInfo.Proof = proof
632	batchInfo.PublicInputs = pubInputs
633	batchInfo.ForgeBatchArgs = prepareForgeBatchArgs(batchInfo)
634	batchInfo.Debug.Status = StatusProof
635	p.cfg.debugBatchStore(batchInfo)
636	log.Infow("Pipeline: batch proof calculated", "batch", batchInfo.BatchNum)
637	return nil
638}
639
640func (p *Pipeline) shouldL1L2Batch(batchInfo *BatchInfo) bool {
641	// Take the lastL1BatchBlockNum as the biggest between the last
642	// scheduled one, and the synchronized one.
643	lastL1BatchBlockNum := p.state.lastScheduledL1BatchBlockNum
644	if p.stats.Sync.LastL1BatchBlock > lastL1BatchBlockNum {
645		lastL1BatchBlockNum = p.stats.Sync.LastL1BatchBlock
646	}
647	// Set Debug information
648	batchInfo.Debug.LastScheduledL1BatchBlockNum = p.state.lastScheduledL1BatchBlockNum
649	batchInfo.Debug.LastL1BatchBlock = p.stats.Sync.LastL1BatchBlock
650	batchInfo.Debug.LastL1BatchBlockDelta = p.stats.Eth.LastBlock.Num + 1 - lastL1BatchBlockNum
651	batchInfo.Debug.L1BatchBlockScheduleDeadline =
652		int64(float64(p.vars.Rollup.ForgeL1L2BatchTimeout-1) * p.cfg.L1BatchTimeoutPerc)
653	// Return true if we have passed the l1BatchTimeoutPerc portion of the
654	// range before the l1batch timeout.
655	return p.stats.Eth.LastBlock.Num+1-lastL1BatchBlockNum >=
656		int64(float64(p.vars.Rollup.ForgeL1L2BatchTimeout-1)*p.cfg.L1BatchTimeoutPerc)
657}
658
659func prepareForgeBatchArgs(batchInfo *BatchInfo) *eth.RollupForgeBatchArgs {
660	proof := batchInfo.Proof
661	zki := batchInfo.ZKInputs
662	return &eth.RollupForgeBatchArgs{
663		NewLastIdx:            int64(zki.Metadata.NewLastIdxRaw),
664		NewStRoot:             zki.Metadata.NewStateRootRaw.BigInt(),
665		NewExitRoot:           zki.Metadata.NewExitRootRaw.BigInt(),
666		L1UserTxs:             batchInfo.L1UserTxs,
667		L1CoordinatorTxs:      batchInfo.L1CoordTxs,
668		L1CoordinatorTxsAuths: batchInfo.L1CoordinatorTxsAuths,
669		L2TxsData:             batchInfo.L2Txs,
670		FeeIdxCoordinator:     batchInfo.CoordIdxs,
671		// Circuit selector
672		VerifierIdx: batchInfo.VerifierIdx,
673		L1Batch:     batchInfo.L1Batch,
674		ProofA:      [2]*big.Int{proof.PiA[0], proof.PiA[1]},
675		// Implementation of the verifier need a swap on the proofB vector
676		ProofB: [2][2]*big.Int{
677			{proof.PiB[0][1], proof.PiB[0][0]},
678			{proof.PiB[1][1], proof.PiB[1][0]},
679		},
680		ProofC: [2]*big.Int{proof.PiC[0], proof.PiC[1]},
681	}
682}