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