hermez-node/coordinator/coordinator.go
coordinator/coordinator.goRaw
1package coordinator
2
3import (
4 "context"
5 "fmt"
6 "math/big"
7 "os"
8 "sync"
9 "time"
10
11 ethCommon "github.com/ethereum/go-ethereum/common"
12 "github.com/hermeznetwork/hermez-node/batchbuilder"
13 "github.com/hermeznetwork/hermez-node/common"
14 "github.com/hermeznetwork/hermez-node/db/historydb"
15 "github.com/hermeznetwork/hermez-node/db/l2db"
16 "github.com/hermeznetwork/hermez-node/eth"
17 "github.com/hermeznetwork/hermez-node/log"
18 "github.com/hermeznetwork/hermez-node/prover"
19 "github.com/hermeznetwork/hermez-node/synchronizer"
20 "github.com/hermeznetwork/hermez-node/txprocessor"
21 "github.com/hermeznetwork/hermez-node/txselector"
22 "github.com/hermeznetwork/tracerr"
23)
24
25var (
26 errLastL1BatchNotSynced = fmt.Errorf("last L1Batch not synced yet")
27 errForgeNoTxsBeforeDelay = fmt.Errorf("no txs to forge and we haven't reached the forge no txs delay")
28 errForgeBeforeDelay = fmt.Errorf("we haven't reached the forge delay")
29)
30
31const (
32 queueLen = 16
33 longWaitDuration = 999 * time.Hour
34 zeroDuration = 0 * time.Second
35)
36
37// Config contains the Coordinator configuration
38type Config struct {
39 // ForgerAddress is the address under which this coordinator is forging
40 ForgerAddress ethCommon.Address
41 // ConfirmBlocks is the number of confirmation blocks to wait for sent
42 // ethereum transactions before forgetting about them
43 ConfirmBlocks int64
44 // L1BatchTimeoutPerc is the portion of the range before the L1Batch
45 // timeout that will trigger a schedule to forge an L1Batch
46 L1BatchTimeoutPerc float64
47 // StartSlotBlocksDelay is the number of blocks of delay to wait before
48 // starting the pipeline when we reach a slot in which we can forge.
49 StartSlotBlocksDelay int64
50 // ScheduleBatchBlocksAheadCheck is the number of blocks ahead in which
51 // the forger address is checked to be allowed to forge (apart from
52 // checking the next block), used to decide when to stop scheduling new
53 // batches (by stopping the pipeline).
54 // For example, if we are at block 10 and ScheduleBatchBlocksAheadCheck
55 // is 5, eventhough at block 11 we canForge, the pipeline will be
56 // stopped if we can't forge at block 15.
57 // This value should be the expected number of blocks it takes between
58 // scheduling a batch and having it mined.
59 ScheduleBatchBlocksAheadCheck int64
60 // SendBatchBlocksMarginCheck is the number of margin blocks ahead in
61 // which the coordinator is also checked to be allowed to forge, apart
62 // from the next block; used to decide when to stop sending batches to
63 // the smart contract.
64 // For example, if we are at block 10 and SendBatchBlocksMarginCheck is
65 // 5, eventhough at block 11 we canForge, the batch will be discarded
66 // if we can't forge at block 15.
67 // This value should be the expected number of blocks it takes between
68 // sending a batch and having it mined.
69 SendBatchBlocksMarginCheck int64
70 // EthClientAttempts is the number of attempts to do an eth client RPC
71 // call before giving up
72 EthClientAttempts int
73 // ForgeRetryInterval is the waiting interval between calls forge a
74 // batch after an error
75 ForgeRetryInterval time.Duration
76 // ForgeDelay is the delay after which a batch is forged if the slot is
77 // already committed. If set to 0s, the coordinator will continuously
78 // forge at the maximum rate.
79 ForgeDelay time.Duration
80 // ForgeNoTxsDelay is the delay after which a batch is forged even if
81 // there are no txs to forge if the slot is already committed. If set
82 // to 0s, the coordinator will continuously forge even if the batches
83 // are empty.
84 ForgeNoTxsDelay time.Duration
85 // SyncRetryInterval is the waiting interval between calls to the main
86 // handler of a synced block after an error
87 SyncRetryInterval time.Duration
88 // PurgeByExtDelInterval is the waiting interval between calls
89 // to the PurgeByExternalDelete function of the l2db which deletes
90 // pending txs externally marked by the column `external_delete`
91 PurgeByExtDelInterval time.Duration
92 // EthClientAttemptsDelay is delay between attempts do do an eth client
93 // RPC call
94 EthClientAttemptsDelay time.Duration
95 // EthTxResendTimeout is the timeout after which a non-mined ethereum
96 // transaction will be resent (reusing the nonce) with a newly
97 // calculated gas price
98 EthTxResendTimeout time.Duration
99 // EthNoReuseNonce disables reusing nonces of pending transactions for
100 // new replacement transactions
101 EthNoReuseNonce bool
102 // MaxGasPrice is the maximum gas price allowed for ethereum
103 // transactions
104 MaxGasPrice *big.Int
105 // GasPriceIncPerc is the percentage increase of gas price set in an
106 // ethereum transaction from the suggested gas price by the ehtereum
107 // node
108 GasPriceIncPerc int64
109 // TxManagerCheckInterval is the waiting interval between receipt
110 // checks of ethereum transactions in the TxManager
111 TxManagerCheckInterval time.Duration
112 // DebugBatchPath if set, specifies the path where batchInfo is stored
113 // in JSON in every step/update of the pipeline
114 DebugBatchPath string
115 Purger PurgerCfg
116 // VerifierIdx is the index of the verifier contract registered in the
117 // smart contract
118 VerifierIdx uint8
119 TxProcessorConfig txprocessor.Config
120}
121
122func (c *Config) debugBatchStore(batchInfo *BatchInfo) {
123 if c.DebugBatchPath != "" {
124 if err := batchInfo.DebugStore(c.DebugBatchPath); err != nil {
125 log.Warnw("Error storing debug BatchInfo",
126 "path", c.DebugBatchPath, "err", err)
127 }
128 }
129}
130
131type fromBatch struct {
132 BatchNum common.BatchNum
133 ForgerAddr ethCommon.Address
134 StateRoot *big.Int
135}
136
137// Coordinator implements the Coordinator type
138type Coordinator struct {
139 // State
140 pipelineNum int // Pipeline sequential number. The first pipeline is 1
141 pipelineFromBatch fromBatch // batch from which we started the pipeline
142 provers []prover.Client
143 consts synchronizer.SCConsts
144 vars synchronizer.SCVariables
145 stats synchronizer.Stats
146 started bool
147
148 cfg Config
149
150 historyDB *historydb.HistoryDB
151 l2DB *l2db.L2DB
152 txSelector *txselector.TxSelector
153 batchBuilder *batchbuilder.BatchBuilder
154
155 msgCh chan interface{}
156 ctx context.Context
157 wg sync.WaitGroup
158 cancel context.CancelFunc
159
160 // mutexL2DBUpdateDelete protects updates to the L2DB so that
161 // these two processes always happen exclusively:
162 // - Pipeline taking pending txs, running through the TxProcessor and
163 // marking selected txs as forging
164 // - Coordinator deleting pending txs that have been marked with
165 // `external_delete`.
166 // Without this mutex, the coordinator could delete a pending txs that
167 // has just been selected by the TxProcessor in the pipeline.
168 mutexL2DBUpdateDelete sync.Mutex
169 pipeline *Pipeline
170 lastNonFailedBatchNum common.BatchNum
171
172 purger *Purger
173 txManager *TxManager
174}
175
176// NewCoordinator creates a new Coordinator
177func NewCoordinator(cfg Config,
178 historyDB *historydb.HistoryDB,
179 l2DB *l2db.L2DB,
180 txSelector *txselector.TxSelector,
181 batchBuilder *batchbuilder.BatchBuilder,
182 serverProofs []prover.Client,
183 ethClient eth.ClientInterface,
184 scConsts *synchronizer.SCConsts,
185 initSCVars *synchronizer.SCVariables,
186) (*Coordinator, error) {
187 // nolint reason: hardcoded `1.0`, by design the percentage can't be over 100%
188 if cfg.L1BatchTimeoutPerc >= 1.0 { //nolint:gomnd
189 return nil, tracerr.Wrap(fmt.Errorf("invalid value for Config.L1BatchTimeoutPerc (%v >= 1.0)",
190 cfg.L1BatchTimeoutPerc))
191 }
192 if cfg.EthClientAttempts < 1 {
193 return nil, tracerr.Wrap(fmt.Errorf("invalid value for Config.EthClientAttempts (%v < 1)",
194 cfg.EthClientAttempts))
195 }
196
197 if cfg.DebugBatchPath != "" {
198 if err := os.MkdirAll(cfg.DebugBatchPath, 0744); err != nil {
199 return nil, tracerr.Wrap(err)
200 }
201 }
202
203 purger := Purger{
204 cfg: cfg.Purger,
205 lastPurgeBlock: 0,
206 lastPurgeBatch: 0,
207 lastInvalidateBlock: 0,
208 lastInvalidateBatch: 0,
209 }
210
211 ctx, cancel := context.WithCancel(context.Background())
212 c := Coordinator{
213 pipelineNum: 0,
214 pipelineFromBatch: fromBatch{
215 BatchNum: 0,
216 ForgerAddr: ethCommon.Address{},
217 StateRoot: big.NewInt(0),
218 },
219 provers: serverProofs,
220 consts: *scConsts,
221 vars: *initSCVars,
222
223 cfg: cfg,
224
225 historyDB: historyDB,
226 l2DB: l2DB,
227 txSelector: txSelector,
228 batchBuilder: batchBuilder,
229
230 purger: &purger,
231
232 msgCh: make(chan interface{}),
233 ctx: ctx,
234 // wg
235 cancel: cancel,
236 }
237 ctxTimeout, ctxTimeoutCancel := context.WithTimeout(ctx, 1*time.Second)
238 defer ctxTimeoutCancel()
239 txManager, err := NewTxManager(ctxTimeout, &cfg, ethClient, l2DB, &c,
240 scConsts, initSCVars)
241 if err != nil {
242 return nil, tracerr.Wrap(err)
243 }
244 c.txManager = txManager
245 // Set Eth LastBlockNum to -1 in stats so that stats.Synced() is
246 // guaranteed to return false before it's updated with a real stats
247 c.stats.Eth.LastBlock.Num = -1
248 return &c, nil
249}
250
251// TxSelector returns the inner TxSelector
252func (c *Coordinator) TxSelector() *txselector.TxSelector {
253 return c.txSelector
254}
255
256// BatchBuilder returns the inner BatchBuilder
257func (c *Coordinator) BatchBuilder() *batchbuilder.BatchBuilder {
258 return c.batchBuilder
259}
260
261func (c *Coordinator) newPipeline(ctx context.Context) (*Pipeline, error) {
262 c.pipelineNum++
263 return NewPipeline(ctx, c.cfg, c.pipelineNum, c.historyDB, c.l2DB, c.txSelector,
264 c.batchBuilder, &c.mutexL2DBUpdateDelete, c.purger, c, c.txManager,
265 c.provers, &c.consts)
266}
267
268// MsgSyncBlock indicates an update to the Synchronizer stats
269type MsgSyncBlock struct {
270 Stats synchronizer.Stats
271 Batches []common.BatchData
272 // Vars contains each Smart Contract variables if they are updated, or
273 // nil if they haven't changed.
274 Vars synchronizer.SCVariablesPtr
275}
276
277// MsgSyncReorg indicates a reorg
278type MsgSyncReorg struct {
279 Stats synchronizer.Stats
280 Vars synchronizer.SCVariablesPtr
281}
282
283// MsgStopPipeline indicates a signal to reset the pipeline
284type MsgStopPipeline struct {
285 Reason string
286 // FailedBatchNum indicates the first batchNum that failed in the
287 // pipeline. If FailedBatchNum is 0, it should be ignored.
288 FailedBatchNum common.BatchNum
289}
290
291// SendMsg is a thread safe method to pass a message to the Coordinator
292func (c *Coordinator) SendMsg(ctx context.Context, msg interface{}) {
293 select {
294 case c.msgCh <- msg:
295 case <-ctx.Done():
296 }
297}
298
299func updateSCVars(vars *synchronizer.SCVariables, update synchronizer.SCVariablesPtr) {
300 if update.Rollup != nil {
301 vars.Rollup = *update.Rollup
302 }
303 if update.Auction != nil {
304 vars.Auction = *update.Auction
305 }
306 if update.WDelayer != nil {
307 vars.WDelayer = *update.WDelayer
308 }
309}
310
311func (c *Coordinator) syncSCVars(vars synchronizer.SCVariablesPtr) {
312 updateSCVars(&c.vars, vars)
313}
314
315func canForge(auctionConstants *common.AuctionConstants, auctionVars *common.AuctionVariables,
316 currentSlot *common.Slot, nextSlot *common.Slot, addr ethCommon.Address, blockNum int64) bool {
317 if blockNum < auctionConstants.GenesisBlockNum {
318 log.Infow("canForge: requested blockNum is < genesis", "blockNum", blockNum,
319 "genesis", auctionConstants.GenesisBlockNum)
320 return false
321 }
322 var slot *common.Slot
323 if currentSlot.StartBlock <= blockNum && blockNum <= currentSlot.EndBlock {
324 slot = currentSlot
325 } else if nextSlot.StartBlock <= blockNum && blockNum <= nextSlot.EndBlock {
326 slot = nextSlot
327 } else {
328 log.Warnw("canForge: requested blockNum is outside current and next slot",
329 "blockNum", blockNum, "currentSlot", currentSlot,
330 "nextSlot", nextSlot,
331 )
332 return false
333 }
334 anyoneForge := false
335 if !slot.ForgerCommitment &&
336 auctionConstants.RelativeBlock(blockNum) >= int64(auctionVars.SlotDeadline) {
337 log.Debugw("canForge: anyone can forge in the current slot (slotDeadline passed)",
338 "block", blockNum)
339 anyoneForge = true
340 }
341 if slot.Forger == addr || anyoneForge {
342 return true
343 }
344 log.Debugw("canForge: can't forge", "slot.Forger", slot.Forger)
345 return false
346}
347
348func (c *Coordinator) canForgeAt(blockNum int64) bool {
349 return canForge(&c.consts.Auction, &c.vars.Auction,
350 &c.stats.Sync.Auction.CurrentSlot, &c.stats.Sync.Auction.NextSlot,
351 c.cfg.ForgerAddress, blockNum)
352}
353
354func (c *Coordinator) canForge() bool {
355 blockNum := c.stats.Eth.LastBlock.Num + 1
356 return canForge(&c.consts.Auction, &c.vars.Auction,
357 &c.stats.Sync.Auction.CurrentSlot, &c.stats.Sync.Auction.NextSlot,
358 c.cfg.ForgerAddress, blockNum)
359}
360
361func (c *Coordinator) syncStats(ctx context.Context, stats *synchronizer.Stats) error {
362 nextBlock := c.stats.Eth.LastBlock.Num + 1
363 canForge := c.canForgeAt(nextBlock)
364 if c.cfg.ScheduleBatchBlocksAheadCheck != 0 && canForge {
365 canForge = c.canForgeAt(nextBlock + c.cfg.ScheduleBatchBlocksAheadCheck)
366 }
367 if c.pipeline == nil {
368 relativeBlock := c.consts.Auction.RelativeBlock(nextBlock)
369 if canForge && relativeBlock < c.cfg.StartSlotBlocksDelay {
370 log.Debugf("Coordinator: delaying pipeline start due to "+
371 "relativeBlock (%v) < cfg.StartSlotBlocksDelay (%v)",
372 relativeBlock, c.cfg.StartSlotBlocksDelay)
373 } else if canForge {
374 log.Infow("Coordinator: forging state begin", "block",
375 stats.Eth.LastBlock.Num+1, "batch", stats.Sync.LastBatch.BatchNum)
376 fromBatch := fromBatch{
377 BatchNum: stats.Sync.LastBatch.BatchNum,
378 ForgerAddr: stats.Sync.LastBatch.ForgerAddr,
379 StateRoot: stats.Sync.LastBatch.StateRoot,
380 }
381 if c.lastNonFailedBatchNum > fromBatch.BatchNum {
382 fromBatch.BatchNum = c.lastNonFailedBatchNum
383 fromBatch.ForgerAddr = c.cfg.ForgerAddress
384 fromBatch.StateRoot = big.NewInt(0)
385 }
386 // Before starting the pipeline make sure we reset any
387 // l2tx from the pool that was forged in a batch that
388 // didn't end up being mined. We are already doing
389 // this in handleStopPipeline, but we do it again as a
390 // failsafe in case the last synced batchnum is
391 // different than in the previous call to l2DB.Reorg,
392 // or in case the node was restarted when there was a
393 // started batch that included l2txs but was not mined.
394 if err := c.l2DB.Reorg(fromBatch.BatchNum); err != nil {
395 return tracerr.Wrap(err)
396 }
397 var err error
398 if c.pipeline, err = c.newPipeline(ctx); err != nil {
399 return tracerr.Wrap(err)
400 }
401 c.pipelineFromBatch = fromBatch
402 // Start the pipeline
403 if err := c.pipeline.Start(fromBatch.BatchNum, stats, &c.vars); err != nil {
404 c.pipeline = nil
405 return tracerr.Wrap(err)
406 }
407 }
408 } else {
409 if !canForge {
410 log.Infow("Coordinator: forging state end", "block", stats.Eth.LastBlock.Num+1)
411 c.pipeline.Stop(c.ctx)
412 c.pipeline = nil
413 }
414 }
415 if c.pipeline == nil {
416 if _, err := c.purger.InvalidateMaybe(c.l2DB, c.txSelector.LocalAccountsDB(),
417 stats.Sync.LastBlock.Num, int64(stats.Sync.LastBatch.BatchNum)); err != nil {
418 return tracerr.Wrap(err)
419 }
420 if _, err := c.purger.PurgeMaybe(c.l2DB, stats.Sync.LastBlock.Num,
421 int64(stats.Sync.LastBatch.BatchNum)); err != nil {
422 return tracerr.Wrap(err)
423 }
424 }
425 return nil
426}
427
428func (c *Coordinator) handleMsgSyncBlock(ctx context.Context, msg *MsgSyncBlock) error {
429 c.stats = msg.Stats
430 c.syncSCVars(msg.Vars)
431 c.txManager.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
432 if c.pipeline != nil {
433 c.pipeline.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
434 }
435 if !c.stats.Synced() {
436 return nil
437 }
438 return c.syncStats(ctx, &c.stats)
439}
440
441func (c *Coordinator) handleReorg(ctx context.Context, msg *MsgSyncReorg) error {
442 c.stats = msg.Stats
443 c.syncSCVars(msg.Vars)
444 c.txManager.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
445 if c.pipeline != nil {
446 c.pipeline.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
447 }
448 if c.stats.Sync.LastBatch.ForgerAddr != c.cfg.ForgerAddress &&
449 (c.stats.Sync.LastBatch.StateRoot == nil || c.pipelineFromBatch.StateRoot == nil ||
450 c.stats.Sync.LastBatch.StateRoot.Cmp(c.pipelineFromBatch.StateRoot) != 0) {
451 // There's been a reorg and the batch state root from which the
452 // pipeline was started has changed (probably because it was in
453 // a block that was discarded), and it was sent by a different
454 // coordinator than us. That batch may never be in the main
455 // chain, so we stop the pipeline (it will be started again
456 // once the node is in sync).
457 log.Infow("Coordinator.handleReorg StopPipeline sync.LastBatch.ForgerAddr != cfg.ForgerAddr "+
458 "& sync.LastBatch.StateRoot != pipelineFromBatch.StateRoot",
459 "sync.LastBatch.StateRoot", c.stats.Sync.LastBatch.StateRoot,
460 "pipelineFromBatch.StateRoot", c.pipelineFromBatch.StateRoot)
461 c.txManager.DiscardPipeline(ctx, c.pipelineNum)
462 if err := c.handleStopPipeline(ctx, "reorg", 0); err != nil {
463 return tracerr.Wrap(err)
464 }
465 }
466 return nil
467}
468
469// handleStopPipeline handles stopping the pipeline. If failedBatchNum is 0,
470// the next pipeline will start from the last state of the synchronizer,
471// otherwise, it will state from failedBatchNum-1.
472func (c *Coordinator) handleStopPipeline(ctx context.Context, reason string, failedBatchNum common.BatchNum) error {
473 batchNum := c.stats.Sync.LastBatch.BatchNum
474 if failedBatchNum != 0 {
475 batchNum = failedBatchNum - 1
476 }
477 if c.pipeline != nil {
478 c.pipeline.Stop(c.ctx)
479 c.pipeline = nil
480 }
481 if err := c.l2DB.Reorg(batchNum); err != nil {
482 return tracerr.Wrap(err)
483 }
484 c.lastNonFailedBatchNum = batchNum
485 return nil
486}
487
488func (c *Coordinator) handleMsg(ctx context.Context, msg interface{}) error {
489 switch msg := msg.(type) {
490 case MsgSyncBlock:
491 if err := c.handleMsgSyncBlock(ctx, &msg); err != nil {
492 return tracerr.Wrap(fmt.Errorf("Coordinator.handleMsgSyncBlock error: %w", err))
493 }
494 case MsgSyncReorg:
495 if err := c.handleReorg(ctx, &msg); err != nil {
496 return tracerr.Wrap(fmt.Errorf("Coordinator.handleReorg error: %w", err))
497 }
498 case MsgStopPipeline:
499 log.Infow("Coordinator received MsgStopPipeline", "reason", msg.Reason)
500 if err := c.handleStopPipeline(ctx, msg.Reason, msg.FailedBatchNum); err != nil {
501 return tracerr.Wrap(fmt.Errorf("Coordinator.handleStopPipeline: %w", err))
502 }
503 default:
504 log.Fatalw("Coordinator Unexpected Coordinator msg of type %T: %+v", msg, msg)
505 }
506 return nil
507}
508
509// Start the coordinator
510func (c *Coordinator) Start() {
511 if c.started {
512 log.Fatal("Coordinator already started")
513 }
514 c.started = true
515 c.wg.Add(1)
516 go func() {
517 c.txManager.Run(c.ctx)
518 c.wg.Done()
519 }()
520
521 c.wg.Add(1)
522 go func() {
523 waitCh := time.After(longWaitDuration)
524 for {
525 select {
526 case <-c.ctx.Done():
527 log.Info("Coordinator done")
528 c.wg.Done()
529 return
530 case msg := <-c.msgCh:
531 if err := c.handleMsg(c.ctx, msg); c.ctx.Err() != nil {
532 continue
533 } else if err != nil {
534 log.Errorw("Coordinator.handleMsg", "err", err)
535 waitCh = time.After(c.cfg.SyncRetryInterval)
536 continue
537 }
538 waitCh = time.After(longWaitDuration)
539 case <-waitCh:
540 if !c.stats.Synced() {
541 waitCh = time.After(longWaitDuration)
542 continue
543 }
544 if err := c.syncStats(c.ctx, &c.stats); c.ctx.Err() != nil {
545 waitCh = time.After(longWaitDuration)
546 continue
547 } else if err != nil {
548 log.Errorw("Coordinator.syncStats", "err", err)
549 waitCh = time.After(c.cfg.SyncRetryInterval)
550 continue
551 }
552 waitCh = time.After(longWaitDuration)
553 }
554 }
555 }()
556
557 c.wg.Add(1)
558 go func() {
559 for {
560 select {
561 case <-c.ctx.Done():
562 log.Info("Coordinator L2DB.PurgeByExternalDelete loop done")
563 c.wg.Done()
564 return
565 case <-time.After(c.cfg.PurgeByExtDelInterval):
566 c.mutexL2DBUpdateDelete.Lock()
567 if err := c.l2DB.PurgeByExternalDelete(); err != nil {
568 log.Errorw("L2DB.PurgeByExternalDelete", "err", err)
569 }
570 c.mutexL2DBUpdateDelete.Unlock()
571 }
572 }
573 }()
574}
575
576const stopCtxTimeout = 200 * time.Millisecond
577
578// Stop the coordinator
579func (c *Coordinator) Stop() {
580 if !c.started {
581 log.Fatal("Coordinator already stopped")
582 }
583 c.started = false
584 log.Infow("Stopping Coordinator...")
585 c.cancel()
586 c.wg.Wait()
587 if c.pipeline != nil {
588 ctx, cancel := context.WithTimeout(context.Background(), stopCtxTimeout)
589 defer cancel()
590 c.pipeline.Stop(ctx)
591 c.pipeline = nil
592 }
593}