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