1package coordinator
  2
  3import (
  4	"context"
  5	"fmt"
  6	"math/big"
  7	"sync"
  8	"time"
  9
 10	"github.com/hermeznetwork/hermez-node/batchbuilder"
 11	"github.com/hermeznetwork/hermez-node/common"
 12	"github.com/hermeznetwork/hermez-node/db/historydb"
 13	"github.com/hermeznetwork/hermez-node/db/l2db"
 14	"github.com/hermeznetwork/hermez-node/eth"
 15	"github.com/hermeznetwork/hermez-node/log"
 16	"github.com/hermeznetwork/hermez-node/prover"
 17	"github.com/hermeznetwork/hermez-node/synchronizer"
 18	"github.com/hermeznetwork/hermez-node/txselector"
 19	"github.com/hermeznetwork/tracerr"
 20)
 21
 22type statsVars struct {
 23	Stats synchronizer.Stats
 24	Vars  synchronizer.SCVariablesPtr
 25}
 26
 27// Pipeline manages the forging of batches with parallel server proofs
 28type Pipeline struct {
 29	cfg    Config
 30	consts synchronizer.SCConsts
 31
 32	// state
 33	batchNum                     common.BatchNum
 34	lastScheduledL1BatchBlockNum int64
 35	lastForgeL1TxsNum            int64
 36	started                      bool
 37
 38	proversPool  *ProversPool
 39	provers      []prover.Client
 40	txManager    *TxManager
 41	historyDB    *historydb.HistoryDB
 42	l2DB         *l2db.L2DB
 43	txSelector   *txselector.TxSelector
 44	batchBuilder *batchbuilder.BatchBuilder
 45	purger       *Purger
 46
 47	stats       synchronizer.Stats
 48	vars        synchronizer.SCVariables
 49	statsVarsCh chan statsVars
 50
 51	ctx    context.Context
 52	wg     sync.WaitGroup
 53	cancel context.CancelFunc
 54}
 55
 56// NewPipeline creates a new Pipeline
 57func NewPipeline(ctx context.Context,
 58	cfg Config,
 59	historyDB *historydb.HistoryDB,
 60	l2DB *l2db.L2DB,
 61	txSelector *txselector.TxSelector,
 62	batchBuilder *batchbuilder.BatchBuilder,
 63	purger *Purger,
 64	txManager *TxManager,
 65	provers []prover.Client,
 66	scConsts *synchronizer.SCConsts,
 67) (*Pipeline, error) {
 68	proversPool := NewProversPool(len(provers))
 69	proversPoolSize := 0
 70	for _, prover := range provers {
 71		if err := prover.WaitReady(ctx); err != nil {
 72			log.Errorw("prover.WaitReady", "err", err)
 73		} else {
 74			proversPool.Add(ctx, prover)
 75			proversPoolSize++
 76		}
 77	}
 78	if proversPoolSize == 0 {
 79		return nil, tracerr.Wrap(fmt.Errorf("no provers in the pool"))
 80	}
 81	return &Pipeline{
 82		cfg:          cfg,
 83		historyDB:    historyDB,
 84		l2DB:         l2DB,
 85		txSelector:   txSelector,
 86		batchBuilder: batchBuilder,
 87		provers:      provers,
 88		proversPool:  proversPool,
 89		purger:       purger,
 90		txManager:    txManager,
 91		consts:       *scConsts,
 92		statsVarsCh:  make(chan statsVars, queueLen),
 93	}, nil
 94}
 95
 96// SetSyncStatsVars is a thread safe method to sets the synchronizer Stats
 97func (p *Pipeline) SetSyncStatsVars(ctx context.Context, stats *synchronizer.Stats, vars *synchronizer.SCVariablesPtr) {
 98	select {
 99	case p.statsVarsCh <- statsVars{Stats: *stats, Vars: *vars}:
100	case <-ctx.Done():
101	}
102}
103
104// reset pipeline state
105func (p *Pipeline) reset(batchNum common.BatchNum,
106	stats *synchronizer.Stats, vars *synchronizer.SCVariables) error {
107	p.batchNum = batchNum
108	p.lastForgeL1TxsNum = stats.Sync.LastForgeL1TxsNum
109	p.stats = *stats
110	p.vars = *vars
111	p.lastScheduledL1BatchBlockNum = 0
112
113	err := p.txSelector.Reset(p.batchNum)
114	if err != nil {
115		return tracerr.Wrap(err)
116	}
117	err = p.batchBuilder.Reset(p.batchNum, true)
118	if err != nil {
119		return tracerr.Wrap(err)
120	}
121	return nil
122}
123
124func (p *Pipeline) syncSCVars(vars synchronizer.SCVariablesPtr) {
125	if vars.Rollup != nil {
126		p.vars.Rollup = *vars.Rollup
127	}
128	if vars.Auction != nil {
129		p.vars.Auction = *vars.Auction
130	}
131	if vars.WDelayer != nil {
132		p.vars.WDelayer = *vars.WDelayer
133	}
134}
135
136// handleForgeBatch calls p.forgeBatch to forge the batch and get the zkInputs,
137// and then waits for an available proof server and sends the zkInputs to it so
138// that the proof computation begins.
139func (p *Pipeline) handleForgeBatch(ctx context.Context, batchNum common.BatchNum) (*BatchInfo, error) {
140	batchInfo, err := p.forgeBatch(batchNum)
141	if ctx.Err() != nil {
142		return nil, ctx.Err()
143	} else if err != nil {
144		if tracerr.Unwrap(err) == errLastL1BatchNotSynced {
145			log.Warnw("forgeBatch: scheduled L1Batch too early", "err", err,
146				"lastForgeL1TxsNum", p.lastForgeL1TxsNum,
147				"syncLastForgeL1TxsNum", p.stats.Sync.LastForgeL1TxsNum)
148		} else {
149			log.Errorw("forgeBatch", "err", err)
150		}
151		return nil, err
152	}
153	// 6. Wait for an available server proof (blocking call)
154	serverProof, err := p.proversPool.Get(ctx)
155	if ctx.Err() != nil {
156		return nil, ctx.Err()
157	} else if err != nil {
158		log.Errorw("proversPool.Get", "err", err)
159		return nil, err
160	}
161	batchInfo.ServerProof = serverProof
162	if err := p.sendServerProof(ctx, batchInfo); ctx.Err() != nil {
163		return nil, ctx.Err()
164	} else if err != nil {
165		log.Errorw("sendServerProof", "err", err)
166		batchInfo.ServerProof = nil
167		p.proversPool.Add(ctx, serverProof)
168		return nil, err
169	}
170	return batchInfo, nil
171}
172
173// Start the forging pipeline
174func (p *Pipeline) Start(batchNum common.BatchNum,
175	stats *synchronizer.Stats, vars *synchronizer.SCVariables) error {
176	if p.started {
177		log.Fatal("Pipeline already started")
178	}
179	p.started = true
180
181	if err := p.reset(batchNum, stats, vars); err != nil {
182		return tracerr.Wrap(err)
183	}
184	p.ctx, p.cancel = context.WithCancel(context.Background())
185
186	queueSize := 1
187	batchChSentServerProof := make(chan *BatchInfo, queueSize)
188
189	p.wg.Add(1)
190	go func() {
191		waitDuration := zeroDuration
192		for {
193			select {
194			case <-p.ctx.Done():
195				log.Info("Pipeline forgeBatch loop done")
196				p.wg.Done()
197				return
198			case statsVars := <-p.statsVarsCh:
199				p.stats = statsVars.Stats
200				p.syncSCVars(statsVars.Vars)
201			case <-time.After(waitDuration):
202				batchNum = p.batchNum + 1
203				if batchInfo, err := p.handleForgeBatch(p.ctx, batchNum); err != nil {
204					waitDuration = p.cfg.SyncRetryInterval
205					continue
206				} else {
207					p.batchNum = batchNum
208					select {
209					case batchChSentServerProof <- batchInfo:
210					case <-p.ctx.Done():
211					}
212				}
213			}
214		}
215	}()
216
217	p.wg.Add(1)
218	go func() {
219		for {
220			select {
221			case <-p.ctx.Done():
222				log.Info("Pipeline waitServerProofSendEth loop done")
223				p.wg.Done()
224				return
225			case batchInfo := <-batchChSentServerProof:
226				err := p.waitServerProof(p.ctx, batchInfo)
227				// We are done with this serverProof, add it back to the pool
228				p.proversPool.Add(p.ctx, batchInfo.ServerProof)
229				batchInfo.ServerProof = nil
230				if p.ctx.Err() != nil {
231					continue
232				} else if err != nil {
233					log.Errorw("waitServerProof", "err", err)
234					continue
235				}
236				p.txManager.AddBatch(p.ctx, batchInfo)
237			}
238		}
239	}()
240	return nil
241}
242
243// Stop the forging pipeline
244func (p *Pipeline) Stop(ctx context.Context) {
245	if !p.started {
246		log.Fatal("Pipeline already stopped")
247	}
248	p.started = false
249	log.Info("Stopping Pipeline...")
250	p.cancel()
251	p.wg.Wait()
252	for _, prover := range p.provers {
253		if err := prover.Cancel(ctx); ctx.Err() != nil {
254			continue
255		} else if err != nil {
256			log.Errorw("prover.Cancel", "err", err)
257		}
258	}
259}
260
261// sendServerProof sends the circuit inputs to the proof server
262func (p *Pipeline) sendServerProof(ctx context.Context, batchInfo *BatchInfo) error {
263	p.cfg.debugBatchStore(batchInfo)
264
265	// 7. Call the selected idle server proof with BatchBuilder output,
266	// save server proof info for batchNum
267	if err := batchInfo.ServerProof.CalculateProof(ctx, batchInfo.ZKInputs); err != nil {
268		return tracerr.Wrap(err)
269	}
270	return nil
271}
272
273// forgeBatch forges the batchNum batch.
274func (p *Pipeline) forgeBatch(batchNum common.BatchNum) (batchInfo *BatchInfo, err error) {
275	// remove transactions from the pool that have been there for too long
276	_, err = p.purger.InvalidateMaybe(p.l2DB, p.txSelector.LocalAccountsDB(),
277		p.stats.Sync.LastBlock.Num, int64(batchNum))
278	if err != nil {
279		return nil, tracerr.Wrap(err)
280	}
281	_, err = p.purger.PurgeMaybe(p.l2DB, p.stats.Sync.LastBlock.Num, int64(batchNum))
282	if err != nil {
283		return nil, tracerr.Wrap(err)
284	}
285
286	batchInfo = &BatchInfo{BatchNum: batchNum} // to accumulate metadata of the batch
287	batchInfo.Debug.StartTimestamp = time.Now()
288	batchInfo.Debug.StartBlockNum = p.stats.Eth.LastBlock.Num + 1
289
290	selectionCfg := &txselector.SelectionConfig{
291		MaxL1UserTxs:      common.RollupConstMaxL1UserTx,
292		TxProcessorConfig: p.cfg.TxProcessorConfig,
293	}
294
295	var poolL2Txs []common.PoolL2Tx
296	var discardedL2Txs []common.PoolL2Tx
297	var l1UserTxsExtra, l1CoordTxs []common.L1Tx
298	var auths [][]byte
299	var coordIdxs []common.Idx
300
301	// 1. Decide if we forge L2Tx or L1+L2Tx
302	if p.shouldL1L2Batch(batchInfo) {
303		batchInfo.L1Batch = true
304		defer func() {
305			// If there's no error, update the parameters related
306			// to the last L1Batch forged
307			if err == nil {
308				p.lastScheduledL1BatchBlockNum = p.stats.Eth.LastBlock.Num + 1
309				p.lastForgeL1TxsNum++
310			}
311		}()
312		if p.lastForgeL1TxsNum != p.stats.Sync.LastForgeL1TxsNum {
313			return nil, tracerr.Wrap(errLastL1BatchNotSynced)
314		}
315		// 2a: L1+L2 txs
316		l1UserTxs, err := p.historyDB.GetUnforgedL1UserTxs(p.lastForgeL1TxsNum + 1)
317		if err != nil {
318			return nil, tracerr.Wrap(err)
319		}
320		coordIdxs, auths, l1UserTxsExtra, l1CoordTxs, poolL2Txs, discardedL2Txs, err =
321			p.txSelector.GetL1L2TxSelection(selectionCfg, l1UserTxs)
322		if err != nil {
323			return nil, tracerr.Wrap(err)
324		}
325	} else {
326		// 2b: only L2 txs
327		coordIdxs, auths, l1CoordTxs, poolL2Txs, discardedL2Txs, err =
328			p.txSelector.GetL2TxSelection(selectionCfg)
329		if err != nil {
330			return nil, tracerr.Wrap(err)
331		}
332		l1UserTxsExtra = nil
333	}
334
335	// 3.  Save metadata from TxSelector output for BatchNum
336	batchInfo.L1UserTxsExtra = l1UserTxsExtra
337	batchInfo.L1CoordTxs = l1CoordTxs
338	batchInfo.L1CoordinatorTxsAuths = auths
339	batchInfo.CoordIdxs = coordIdxs
340	batchInfo.VerifierIdx = p.cfg.VerifierIdx
341
342	if err := p.l2DB.StartForging(common.TxIDsFromPoolL2Txs(poolL2Txs), batchInfo.BatchNum); err != nil {
343		return nil, tracerr.Wrap(err)
344	}
345	if err := p.l2DB.UpdateTxsInfo(discardedL2Txs); err != nil {
346		return nil, tracerr.Wrap(err)
347	}
348
349	// Invalidate transactions that become invalid beause of
350	// the poolL2Txs selected.  Will mark as invalid the txs that have a
351	// (fromIdx, nonce) which already appears in the selected txs (includes
352	// all the nonces smaller than the current one)
353	err = p.l2DB.InvalidateOldNonces(idxsNonceFromPoolL2Txs(poolL2Txs), batchInfo.BatchNum)
354	if err != nil {
355		return nil, tracerr.Wrap(err)
356	}
357
358	// 4. Call BatchBuilder with TxSelector output
359	configBatch := &batchbuilder.ConfigBatch{
360		TxProcessorConfig: p.cfg.TxProcessorConfig,
361	}
362	zkInputs, err := p.batchBuilder.BuildBatch(coordIdxs, configBatch, l1UserTxsExtra,
363		l1CoordTxs, poolL2Txs)
364	if err != nil {
365		return nil, tracerr.Wrap(err)
366	}
367	l2Txs, err := common.PoolL2TxsToL2Txs(poolL2Txs) // NOTE: This is a big uggly, find a better way
368	if err != nil {
369		return nil, tracerr.Wrap(err)
370	}
371	batchInfo.L2Txs = l2Txs
372
373	// 5. Save metadata from BatchBuilder output for BatchNum
374	batchInfo.ZKInputs = zkInputs
375	batchInfo.Debug.Status = StatusForged
376	p.cfg.debugBatchStore(batchInfo)
377	log.Infow("Pipeline: batch forged internally", "batch", batchInfo.BatchNum)
378
379	return batchInfo, nil
380}
381
382// waitServerProof gets the generated zkProof & sends it to the SmartContract
383func (p *Pipeline) waitServerProof(ctx context.Context, batchInfo *BatchInfo) error {
384	proof, pubInputs, err := batchInfo.ServerProof.GetProof(ctx) // blocking call, until not resolved don't continue. Returns when the proof server has calculated the proof
385	if err != nil {
386		return tracerr.Wrap(err)
387	}
388	batchInfo.Proof = proof
389	batchInfo.PublicInputs = pubInputs
390	batchInfo.ForgeBatchArgs = prepareForgeBatchArgs(batchInfo)
391	batchInfo.Debug.Status = StatusProof
392	p.cfg.debugBatchStore(batchInfo)
393	log.Infow("Pipeline: batch proof calculated", "batch", batchInfo.BatchNum)
394	return nil
395}
396
397func (p *Pipeline) shouldL1L2Batch(batchInfo *BatchInfo) bool {
398	// Take the lastL1BatchBlockNum as the biggest between the last
399	// scheduled one, and the synchronized one.
400	lastL1BatchBlockNum := p.lastScheduledL1BatchBlockNum
401	if p.stats.Sync.LastL1BatchBlock > lastL1BatchBlockNum {
402		lastL1BatchBlockNum = p.stats.Sync.LastL1BatchBlock
403	}
404	// Set Debug information
405	batchInfo.Debug.LastScheduledL1BatchBlockNum = p.lastScheduledL1BatchBlockNum
406	batchInfo.Debug.LastL1BatchBlock = p.stats.Sync.LastL1BatchBlock
407	batchInfo.Debug.LastL1BatchBlockDelta = p.stats.Eth.LastBlock.Num + 1 - lastL1BatchBlockNum
408	batchInfo.Debug.L1BatchBlockScheduleDeadline =
409		int64(float64(p.vars.Rollup.ForgeL1L2BatchTimeout-1) * p.cfg.L1BatchTimeoutPerc)
410	// Return true if we have passed the l1BatchTimeoutPerc portion of the
411	// range before the l1batch timeout.
412	return p.stats.Eth.LastBlock.Num+1-lastL1BatchBlockNum >=
413		int64(float64(p.vars.Rollup.ForgeL1L2BatchTimeout-1)*p.cfg.L1BatchTimeoutPerc)
414}
415
416func prepareForgeBatchArgs(batchInfo *BatchInfo) *eth.RollupForgeBatchArgs {
417	proof := batchInfo.Proof
418	zki := batchInfo.ZKInputs
419	return &eth.RollupForgeBatchArgs{
420		NewLastIdx:            int64(zki.Metadata.NewLastIdxRaw),
421		NewStRoot:             zki.Metadata.NewStateRootRaw.BigInt(),
422		NewExitRoot:           zki.Metadata.NewExitRootRaw.BigInt(),
423		L1UserTxs:             batchInfo.L1UserTxsExtra,
424		L1CoordinatorTxs:      batchInfo.L1CoordTxs,
425		L1CoordinatorTxsAuths: batchInfo.L1CoordinatorTxsAuths,
426		L2TxsData:             batchInfo.L2Txs,
427		FeeIdxCoordinator:     batchInfo.CoordIdxs,
428		// Circuit selector
429		VerifierIdx: batchInfo.VerifierIdx,
430		L1Batch:     batchInfo.L1Batch,
431		ProofA:      [2]*big.Int{proof.PiA[0], proof.PiA[1]},
432		// Implementation of the verifier need a swap on the proofB vector
433		ProofB: [2][2]*big.Int{
434			{proof.PiB[0][1], proof.PiB[0][0]},
435			{proof.PiB[1][1], proof.PiB[1][0]},
436		},
437		ProofC: [2]*big.Int{proof.PiC[0], proof.PiC[1]},
438	}
439}