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