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				batchInfo, err := p.handleForgeBatch(p.ctx, batchNum)
204				if p.ctx.Err() != nil {
205					continue
206				} else if err != nil {
207					waitDuration = p.cfg.SyncRetryInterval
208					continue
209				}
210				p.batchNum = batchNum
211				select {
212				case batchChSentServerProof <- batchInfo:
213				case <-p.ctx.Done():
214				}
215			}
216		}
217	}()
218
219	p.wg.Add(1)
220	go func() {
221		for {
222			select {
223			case <-p.ctx.Done():
224				log.Info("Pipeline waitServerProofSendEth loop done")
225				p.wg.Done()
226				return
227			case batchInfo := <-batchChSentServerProof:
228				err := p.waitServerProof(p.ctx, batchInfo)
229				// We are done with this serverProof, add it back to the pool
230				p.proversPool.Add(p.ctx, batchInfo.ServerProof)
231				batchInfo.ServerProof = nil
232				if p.ctx.Err() != nil {
233					continue
234				} else if err != nil {
235					log.Errorw("waitServerProof", "err", err)
236					continue
237				}
238				p.txManager.AddBatch(p.ctx, batchInfo)
239			}
240		}
241	}()
242	return nil
243}
244
245// Stop the forging pipeline
246func (p *Pipeline) Stop(ctx context.Context) {
247	if !p.started {
248		log.Fatal("Pipeline already stopped")
249	}
250	p.started = false
251	log.Info("Stopping Pipeline...")
252	p.cancel()
253	p.wg.Wait()
254	for _, prover := range p.provers {
255		if err := prover.Cancel(ctx); ctx.Err() != nil {
256			continue
257		} else if err != nil {
258			log.Errorw("prover.Cancel", "err", err)
259		}
260	}
261}
262
263// sendServerProof sends the circuit inputs to the proof server
264func (p *Pipeline) sendServerProof(ctx context.Context, batchInfo *BatchInfo) error {
265	p.cfg.debugBatchStore(batchInfo)
266
267	// 7. Call the selected idle server proof with BatchBuilder output,
268	// save server proof info for batchNum
269	if err := batchInfo.ServerProof.CalculateProof(ctx, batchInfo.ZKInputs); err != nil {
270		return tracerr.Wrap(err)
271	}
272	return nil
273}
274
275// forgeBatch forges the batchNum batch.
276func (p *Pipeline) forgeBatch(batchNum common.BatchNum) (batchInfo *BatchInfo, err error) {
277	// remove transactions from the pool that have been there for too long
278	_, err = p.purger.InvalidateMaybe(p.l2DB, p.txSelector.LocalAccountsDB(),
279		p.stats.Sync.LastBlock.Num, int64(batchNum))
280	if err != nil {
281		return nil, tracerr.Wrap(err)
282	}
283	_, err = p.purger.PurgeMaybe(p.l2DB, p.stats.Sync.LastBlock.Num, int64(batchNum))
284	if err != nil {
285		return nil, tracerr.Wrap(err)
286	}
287
288	batchInfo = &BatchInfo{BatchNum: batchNum} // to accumulate metadata of the batch
289	batchInfo.Debug.StartTimestamp = time.Now()
290	batchInfo.Debug.StartBlockNum = p.stats.Eth.LastBlock.Num + 1
291
292	selectionCfg := &txselector.SelectionConfig{
293		MaxL1UserTxs:      common.RollupConstMaxL1UserTx,
294		TxProcessorConfig: p.cfg.TxProcessorConfig,
295	}
296
297	var poolL2Txs []common.PoolL2Tx
298	var discardedL2Txs []common.PoolL2Tx
299	var l1UserTxsExtra, l1CoordTxs []common.L1Tx
300	var auths [][]byte
301	var coordIdxs []common.Idx
302
303	// 1. Decide if we forge L2Tx or L1+L2Tx
304	if p.shouldL1L2Batch(batchInfo) {
305		batchInfo.L1Batch = true
306		defer func() {
307			// If there's no error, update the parameters related
308			// to the last L1Batch forged
309			if err == nil {
310				p.lastScheduledL1BatchBlockNum = p.stats.Eth.LastBlock.Num + 1
311				p.lastForgeL1TxsNum++
312			}
313		}()
314		if p.lastForgeL1TxsNum != p.stats.Sync.LastForgeL1TxsNum {
315			return nil, tracerr.Wrap(errLastL1BatchNotSynced)
316		}
317		// 2a: L1+L2 txs
318		l1UserTxs, err := p.historyDB.GetUnforgedL1UserTxs(p.lastForgeL1TxsNum + 1)
319		if err != nil {
320			return nil, tracerr.Wrap(err)
321		}
322		coordIdxs, auths, l1UserTxsExtra, l1CoordTxs, poolL2Txs, discardedL2Txs, err =
323			p.txSelector.GetL1L2TxSelection(selectionCfg, l1UserTxs)
324		if err != nil {
325			return nil, tracerr.Wrap(err)
326		}
327	} else {
328		// 2b: only L2 txs
329		coordIdxs, auths, l1CoordTxs, poolL2Txs, discardedL2Txs, err =
330			p.txSelector.GetL2TxSelection(selectionCfg)
331		if err != nil {
332			return nil, tracerr.Wrap(err)
333		}
334		l1UserTxsExtra = nil
335	}
336
337	// 3.  Save metadata from TxSelector output for BatchNum
338	batchInfo.L1UserTxsExtra = l1UserTxsExtra
339	batchInfo.L1CoordTxs = l1CoordTxs
340	batchInfo.L1CoordinatorTxsAuths = auths
341	batchInfo.CoordIdxs = coordIdxs
342	batchInfo.VerifierIdx = p.cfg.VerifierIdx
343
344	if err := p.l2DB.StartForging(common.TxIDsFromPoolL2Txs(poolL2Txs), batchInfo.BatchNum); err != nil {
345		return nil, tracerr.Wrap(err)
346	}
347	if err := p.l2DB.UpdateTxsInfo(discardedL2Txs); err != nil {
348		return nil, tracerr.Wrap(err)
349	}
350
351	// Invalidate transactions that become invalid beause of
352	// the poolL2Txs selected.  Will mark as invalid the txs that have a
353	// (fromIdx, nonce) which already appears in the selected txs (includes
354	// all the nonces smaller than the current one)
355	err = p.l2DB.InvalidateOldNonces(idxsNonceFromPoolL2Txs(poolL2Txs), batchInfo.BatchNum)
356	if err != nil {
357		return nil, tracerr.Wrap(err)
358	}
359
360	// 4. Call BatchBuilder with TxSelector output
361	configBatch := &batchbuilder.ConfigBatch{
362		TxProcessorConfig: p.cfg.TxProcessorConfig,
363	}
364	zkInputs, err := p.batchBuilder.BuildBatch(coordIdxs, configBatch, l1UserTxsExtra,
365		l1CoordTxs, poolL2Txs)
366	if err != nil {
367		return nil, tracerr.Wrap(err)
368	}
369	l2Txs, err := common.PoolL2TxsToL2Txs(poolL2Txs) // NOTE: This is a big uggly, find a better way
370	if err != nil {
371		return nil, tracerr.Wrap(err)
372	}
373	batchInfo.L2Txs = l2Txs
374
375	// 5. Save metadata from BatchBuilder output for BatchNum
376	batchInfo.ZKInputs = zkInputs
377	batchInfo.Debug.Status = StatusForged
378	p.cfg.debugBatchStore(batchInfo)
379	log.Infow("Pipeline: batch forged internally", "batch", batchInfo.BatchNum)
380
381	return batchInfo, nil
382}
383
384// waitServerProof gets the generated zkProof & sends it to the SmartContract
385func (p *Pipeline) waitServerProof(ctx context.Context, batchInfo *BatchInfo) error {
386	proof, pubInputs, err := batchInfo.ServerProof.GetProof(ctx) // blocking call, until not resolved don't continue. Returns when the proof server has calculated the proof
387	if err != nil {
388		return tracerr.Wrap(err)
389	}
390	batchInfo.Proof = proof
391	batchInfo.PublicInputs = pubInputs
392	batchInfo.ForgeBatchArgs = prepareForgeBatchArgs(batchInfo)
393	batchInfo.Debug.Status = StatusProof
394	p.cfg.debugBatchStore(batchInfo)
395	log.Infow("Pipeline: batch proof calculated", "batch", batchInfo.BatchNum)
396	return nil
397}
398
399func (p *Pipeline) shouldL1L2Batch(batchInfo *BatchInfo) bool {
400	// Take the lastL1BatchBlockNum as the biggest between the last
401	// scheduled one, and the synchronized one.
402	lastL1BatchBlockNum := p.lastScheduledL1BatchBlockNum
403	if p.stats.Sync.LastL1BatchBlock > lastL1BatchBlockNum {
404		lastL1BatchBlockNum = p.stats.Sync.LastL1BatchBlock
405	}
406	// Set Debug information
407	batchInfo.Debug.LastScheduledL1BatchBlockNum = p.lastScheduledL1BatchBlockNum
408	batchInfo.Debug.LastL1BatchBlock = p.stats.Sync.LastL1BatchBlock
409	batchInfo.Debug.LastL1BatchBlockDelta = p.stats.Eth.LastBlock.Num + 1 - lastL1BatchBlockNum
410	batchInfo.Debug.L1BatchBlockScheduleDeadline =
411		int64(float64(p.vars.Rollup.ForgeL1L2BatchTimeout-1) * p.cfg.L1BatchTimeoutPerc)
412	// Return true if we have passed the l1BatchTimeoutPerc portion of the
413	// range before the l1batch timeout.
414	return p.stats.Eth.LastBlock.Num+1-lastL1BatchBlockNum >=
415		int64(float64(p.vars.Rollup.ForgeL1L2BatchTimeout-1)*p.cfg.L1BatchTimeoutPerc)
416}
417
418func prepareForgeBatchArgs(batchInfo *BatchInfo) *eth.RollupForgeBatchArgs {
419	proof := batchInfo.Proof
420	zki := batchInfo.ZKInputs
421	return &eth.RollupForgeBatchArgs{
422		NewLastIdx:            int64(zki.Metadata.NewLastIdxRaw),
423		NewStRoot:             zki.Metadata.NewStateRootRaw.BigInt(),
424		NewExitRoot:           zki.Metadata.NewExitRootRaw.BigInt(),
425		L1UserTxs:             batchInfo.L1UserTxsExtra,
426		L1CoordinatorTxs:      batchInfo.L1CoordTxs,
427		L1CoordinatorTxsAuths: batchInfo.L1CoordinatorTxsAuths,
428		L2TxsData:             batchInfo.L2Txs,
429		FeeIdxCoordinator:     batchInfo.CoordIdxs,
430		// Circuit selector
431		VerifierIdx: batchInfo.VerifierIdx,
432		L1Batch:     batchInfo.L1Batch,
433		ProofA:      [2]*big.Int{proof.PiA[0], proof.PiA[1]},
434		// Implementation of the verifier need a swap on the proofB vector
435		ProofB: [2][2]*big.Int{
436			{proof.PiB[0][1], proof.PiB[0][0]},
437			{proof.PiB[1][1], proof.PiB[1][0]},
438		},
439		ProofC: [2]*big.Int{proof.PiC[0], proof.PiC[1]},
440	}
441}