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