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 // EthClientAttemptsDelay is delay between attempts do do an eth client
89 // RPC call
90 EthClientAttemptsDelay time.Duration
91 // EthTxResendTimeout is the timeout after which a non-mined ethereum
92 // transaction will be resent (reusing the nonce) with a newly
93 // calculated gas price
94 EthTxResendTimeout time.Duration
95 // EthNoReuseNonce disables reusing nonces of pending transactions for
96 // new replacement transactions
97 EthNoReuseNonce bool
98 // MaxGasPrice is the maximum gas price allowed for ethereum
99 // transactions
100 MaxGasPrice *big.Int
101 // GasPriceIncPerc is the percentage increase of gas price set in an
102 // ethereum transaction from the suggested gas price by the ehtereum
103 // node
104 GasPriceIncPerc int64
105 // TxManagerCheckInterval is the waiting interval between receipt
106 // checks of ethereum transactions in the TxManager
107 TxManagerCheckInterval time.Duration
108 // DebugBatchPath if set, specifies the path where batchInfo is stored
109 // in JSON in every step/update of the pipeline
110 DebugBatchPath string
111 Purger PurgerCfg
112 // VerifierIdx is the index of the verifier contract registered in the
113 // smart contract
114 VerifierIdx uint8
115 TxProcessorConfig txprocessor.Config
116}
117
118func (c *Config) debugBatchStore(batchInfo *BatchInfo) {
119 if c.DebugBatchPath != "" {
120 if err := batchInfo.DebugStore(c.DebugBatchPath); err != nil {
121 log.Warnw("Error storing debug BatchInfo",
122 "path", c.DebugBatchPath, "err", err)
123 }
124 }
125}
126
127type fromBatch struct {
128 BatchNum common.BatchNum
129 ForgerAddr ethCommon.Address
130 StateRoot *big.Int
131}
132
133// Coordinator implements the Coordinator type
134type Coordinator struct {
135 // State
136 pipelineNum int // Pipeline sequential number. The first pipeline is 1
137 pipelineFromBatch fromBatch // batch from which we started the pipeline
138 provers []prover.Client
139 consts synchronizer.SCConsts
140 vars synchronizer.SCVariables
141 stats synchronizer.Stats
142 started bool
143
144 cfg Config
145
146 historyDB *historydb.HistoryDB
147 l2DB *l2db.L2DB
148 txSelector *txselector.TxSelector
149 batchBuilder *batchbuilder.BatchBuilder
150
151 msgCh chan interface{}
152 ctx context.Context
153 wg sync.WaitGroup
154 cancel context.CancelFunc
155
156 pipeline *Pipeline
157 lastNonFailedBatchNum common.BatchNum
158
159 purger *Purger
160 txManager *TxManager
161}
162
163// NewCoordinator creates a new Coordinator
164func NewCoordinator(cfg Config,
165 historyDB *historydb.HistoryDB,
166 l2DB *l2db.L2DB,
167 txSelector *txselector.TxSelector,
168 batchBuilder *batchbuilder.BatchBuilder,
169 serverProofs []prover.Client,
170 ethClient eth.ClientInterface,
171 scConsts *synchronizer.SCConsts,
172 initSCVars *synchronizer.SCVariables,
173) (*Coordinator, error) {
174 // nolint reason: hardcoded `1.0`, by design the percentage can't be over 100%
175 if cfg.L1BatchTimeoutPerc >= 1.0 { //nolint:gomnd
176 return nil, tracerr.Wrap(fmt.Errorf("invalid value for Config.L1BatchTimeoutPerc (%v >= 1.0)",
177 cfg.L1BatchTimeoutPerc))
178 }
179 if cfg.EthClientAttempts < 1 {
180 return nil, tracerr.Wrap(fmt.Errorf("invalid value for Config.EthClientAttempts (%v < 1)",
181 cfg.EthClientAttempts))
182 }
183
184 if cfg.DebugBatchPath != "" {
185 if err := os.MkdirAll(cfg.DebugBatchPath, 0744); err != nil {
186 return nil, tracerr.Wrap(err)
187 }
188 }
189
190 purger := Purger{
191 cfg: cfg.Purger,
192 lastPurgeBlock: 0,
193 lastPurgeBatch: 0,
194 lastInvalidateBlock: 0,
195 lastInvalidateBatch: 0,
196 }
197
198 ctx, cancel := context.WithCancel(context.Background())
199 c := Coordinator{
200 pipelineNum: 0,
201 pipelineFromBatch: fromBatch{
202 BatchNum: 0,
203 ForgerAddr: ethCommon.Address{},
204 StateRoot: big.NewInt(0),
205 },
206 provers: serverProofs,
207 consts: *scConsts,
208 vars: *initSCVars,
209
210 cfg: cfg,
211
212 historyDB: historyDB,
213 l2DB: l2DB,
214 txSelector: txSelector,
215 batchBuilder: batchBuilder,
216
217 purger: &purger,
218
219 msgCh: make(chan interface{}),
220 ctx: ctx,
221 // wg
222 cancel: cancel,
223 }
224 ctxTimeout, ctxTimeoutCancel := context.WithTimeout(ctx, 1*time.Second)
225 defer ctxTimeoutCancel()
226 txManager, err := NewTxManager(ctxTimeout, &cfg, ethClient, l2DB, &c,
227 scConsts, initSCVars)
228 if err != nil {
229 return nil, tracerr.Wrap(err)
230 }
231 c.txManager = txManager
232 // Set Eth LastBlockNum to -1 in stats so that stats.Synced() is
233 // guaranteed to return false before it's updated with a real stats
234 c.stats.Eth.LastBlock.Num = -1
235 return &c, nil
236}
237
238// TxSelector returns the inner TxSelector
239func (c *Coordinator) TxSelector() *txselector.TxSelector {
240 return c.txSelector
241}
242
243// BatchBuilder returns the inner BatchBuilder
244func (c *Coordinator) BatchBuilder() *batchbuilder.BatchBuilder {
245 return c.batchBuilder
246}
247
248func (c *Coordinator) newPipeline(ctx context.Context) (*Pipeline, error) {
249 c.pipelineNum++
250 return NewPipeline(ctx, c.cfg, c.pipelineNum, c.historyDB, c.l2DB, c.txSelector,
251 c.batchBuilder, c.purger, c, c.txManager, c.provers, &c.consts)
252}
253
254// MsgSyncBlock indicates an update to the Synchronizer stats
255type MsgSyncBlock struct {
256 Stats synchronizer.Stats
257 Batches []common.BatchData
258 // Vars contains each Smart Contract variables if they are updated, or
259 // nil if they haven't changed.
260 Vars synchronizer.SCVariablesPtr
261}
262
263// MsgSyncReorg indicates a reorg
264type MsgSyncReorg struct {
265 Stats synchronizer.Stats
266 Vars synchronizer.SCVariablesPtr
267}
268
269// MsgStopPipeline indicates a signal to reset the pipeline
270type MsgStopPipeline struct {
271 Reason string
272 // FailedBatchNum indicates the first batchNum that failed in the
273 // pipeline. If FailedBatchNum is 0, it should be ignored.
274 FailedBatchNum common.BatchNum
275}
276
277// SendMsg is a thread safe method to pass a message to the Coordinator
278func (c *Coordinator) SendMsg(ctx context.Context, msg interface{}) {
279 select {
280 case c.msgCh <- msg:
281 case <-ctx.Done():
282 }
283}
284
285func updateSCVars(vars *synchronizer.SCVariables, update synchronizer.SCVariablesPtr) {
286 if update.Rollup != nil {
287 vars.Rollup = *update.Rollup
288 }
289 if update.Auction != nil {
290 vars.Auction = *update.Auction
291 }
292 if update.WDelayer != nil {
293 vars.WDelayer = *update.WDelayer
294 }
295}
296
297func (c *Coordinator) syncSCVars(vars synchronizer.SCVariablesPtr) {
298 updateSCVars(&c.vars, vars)
299}
300
301func canForge(auctionConstants *common.AuctionConstants, auctionVars *common.AuctionVariables,
302 currentSlot *common.Slot, nextSlot *common.Slot, addr ethCommon.Address, blockNum int64) bool {
303 if blockNum < auctionConstants.GenesisBlockNum {
304 log.Infow("canForge: requested blockNum is < genesis", "blockNum", blockNum,
305 "genesis", auctionConstants.GenesisBlockNum)
306 return false
307 }
308 var slot *common.Slot
309 if currentSlot.StartBlock <= blockNum && blockNum <= currentSlot.EndBlock {
310 slot = currentSlot
311 } else if nextSlot.StartBlock <= blockNum && blockNum <= nextSlot.EndBlock {
312 slot = nextSlot
313 } else {
314 log.Warnw("canForge: requested blockNum is outside current and next slot",
315 "blockNum", blockNum, "currentSlot", currentSlot,
316 "nextSlot", nextSlot,
317 )
318 return false
319 }
320 anyoneForge := false
321 if !slot.ForgerCommitment &&
322 auctionConstants.RelativeBlock(blockNum) >= int64(auctionVars.SlotDeadline) {
323 log.Debugw("canForge: anyone can forge in the current slot (slotDeadline passed)",
324 "block", blockNum)
325 anyoneForge = true
326 }
327 if slot.Forger == addr || anyoneForge {
328 return true
329 }
330 log.Debugw("canForge: can't forge", "slot.Forger", slot.Forger)
331 return false
332}
333
334func (c *Coordinator) canForgeAt(blockNum int64) bool {
335 return canForge(&c.consts.Auction, &c.vars.Auction,
336 &c.stats.Sync.Auction.CurrentSlot, &c.stats.Sync.Auction.NextSlot,
337 c.cfg.ForgerAddress, blockNum)
338}
339
340func (c *Coordinator) canForge() bool {
341 blockNum := c.stats.Eth.LastBlock.Num + 1
342 return canForge(&c.consts.Auction, &c.vars.Auction,
343 &c.stats.Sync.Auction.CurrentSlot, &c.stats.Sync.Auction.NextSlot,
344 c.cfg.ForgerAddress, blockNum)
345}
346
347func (c *Coordinator) syncStats(ctx context.Context, stats *synchronizer.Stats) error {
348 nextBlock := c.stats.Eth.LastBlock.Num + 1
349 canForge := c.canForgeAt(nextBlock)
350 if c.cfg.ScheduleBatchBlocksAheadCheck != 0 && canForge {
351 canForge = c.canForgeAt(nextBlock + c.cfg.ScheduleBatchBlocksAheadCheck)
352 }
353 if c.pipeline == nil {
354 relativeBlock := c.consts.Auction.RelativeBlock(nextBlock)
355 if canForge && relativeBlock < c.cfg.StartSlotBlocksDelay {
356 log.Debugf("Coordinator: delaying pipeline start due to "+
357 "relativeBlock (%v) < cfg.StartSlotBlocksDelay (%v)",
358 relativeBlock, c.cfg.StartSlotBlocksDelay)
359 } else if canForge {
360 log.Infow("Coordinator: forging state begin", "block",
361 stats.Eth.LastBlock.Num+1, "batch", stats.Sync.LastBatch.BatchNum)
362 fromBatch := fromBatch{
363 BatchNum: stats.Sync.LastBatch.BatchNum,
364 ForgerAddr: stats.Sync.LastBatch.ForgerAddr,
365 StateRoot: stats.Sync.LastBatch.StateRoot,
366 }
367 if c.lastNonFailedBatchNum > fromBatch.BatchNum {
368 fromBatch.BatchNum = c.lastNonFailedBatchNum
369 fromBatch.ForgerAddr = c.cfg.ForgerAddress
370 fromBatch.StateRoot = big.NewInt(0)
371 }
372 var err error
373 if c.pipeline, err = c.newPipeline(ctx); err != nil {
374 return tracerr.Wrap(err)
375 }
376 c.pipelineFromBatch = fromBatch
377 if err := c.pipeline.Start(fromBatch.BatchNum, stats, &c.vars); err != nil {
378 c.pipeline = nil
379 return tracerr.Wrap(err)
380 }
381 }
382 } else {
383 if !canForge {
384 log.Infow("Coordinator: forging state end", "block", stats.Eth.LastBlock.Num+1)
385 c.pipeline.Stop(c.ctx)
386 c.pipeline = nil
387 }
388 }
389 if c.pipeline == nil {
390 if _, err := c.purger.InvalidateMaybe(c.l2DB, c.txSelector.LocalAccountsDB(),
391 stats.Sync.LastBlock.Num, int64(stats.Sync.LastBatch.BatchNum)); err != nil {
392 return tracerr.Wrap(err)
393 }
394 if _, err := c.purger.PurgeMaybe(c.l2DB, stats.Sync.LastBlock.Num,
395 int64(stats.Sync.LastBatch.BatchNum)); err != nil {
396 return tracerr.Wrap(err)
397 }
398 }
399 return nil
400}
401
402func (c *Coordinator) handleMsgSyncBlock(ctx context.Context, msg *MsgSyncBlock) error {
403 c.stats = msg.Stats
404 c.syncSCVars(msg.Vars)
405 c.txManager.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
406 if c.pipeline != nil {
407 c.pipeline.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
408 }
409 if !c.stats.Synced() {
410 return nil
411 }
412 return c.syncStats(ctx, &c.stats)
413}
414
415func (c *Coordinator) handleReorg(ctx context.Context, msg *MsgSyncReorg) error {
416 c.stats = msg.Stats
417 c.syncSCVars(msg.Vars)
418 c.txManager.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
419 if c.pipeline != nil {
420 c.pipeline.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
421 }
422 if c.stats.Sync.LastBatch.ForgerAddr != c.cfg.ForgerAddress &&
423 (c.stats.Sync.LastBatch.StateRoot == nil || c.pipelineFromBatch.StateRoot == nil ||
424 c.stats.Sync.LastBatch.StateRoot.Cmp(c.pipelineFromBatch.StateRoot) != 0) {
425 // There's been a reorg and the batch state root from which the
426 // pipeline was started has changed (probably because it was in
427 // a block that was discarded), and it was sent by a different
428 // coordinator than us. That batch may never be in the main
429 // chain, so we stop the pipeline (it will be started again
430 // once the node is in sync).
431 log.Infow("Coordinator.handleReorg StopPipeline sync.LastBatch.ForgerAddr != cfg.ForgerAddr "+
432 "& sync.LastBatch.StateRoot != pipelineFromBatch.StateRoot",
433 "sync.LastBatch.StateRoot", c.stats.Sync.LastBatch.StateRoot,
434 "pipelineFromBatch.StateRoot", c.pipelineFromBatch.StateRoot)
435 c.txManager.DiscardPipeline(ctx, c.pipelineNum)
436 if err := c.handleStopPipeline(ctx, "reorg", 0); err != nil {
437 return tracerr.Wrap(err)
438 }
439 }
440 return nil
441}
442
443// handleStopPipeline handles stopping the pipeline. If failedBatchNum is 0,
444// the next pipeline will start from the last state of the synchronizer,
445// otherwise, it will state from failedBatchNum-1.
446func (c *Coordinator) handleStopPipeline(ctx context.Context, reason string, failedBatchNum common.BatchNum) error {
447 batchNum := c.stats.Sync.LastBatch.BatchNum
448 if failedBatchNum != 0 {
449 batchNum = failedBatchNum - 1
450 }
451 if c.pipeline != nil {
452 c.pipeline.Stop(c.ctx)
453 c.pipeline = nil
454 }
455 if err := c.l2DB.Reorg(batchNum); err != nil {
456 return tracerr.Wrap(err)
457 }
458 c.lastNonFailedBatchNum = batchNum
459 return nil
460}
461
462func (c *Coordinator) handleMsg(ctx context.Context, msg interface{}) error {
463 switch msg := msg.(type) {
464 case MsgSyncBlock:
465 if err := c.handleMsgSyncBlock(ctx, &msg); err != nil {
466 return tracerr.Wrap(fmt.Errorf("Coordinator.handleMsgSyncBlock error: %w", err))
467 }
468 case MsgSyncReorg:
469 if err := c.handleReorg(ctx, &msg); err != nil {
470 return tracerr.Wrap(fmt.Errorf("Coordinator.handleReorg error: %w", err))
471 }
472 case MsgStopPipeline:
473 log.Infow("Coordinator received MsgStopPipeline", "reason", msg.Reason)
474 if err := c.handleStopPipeline(ctx, msg.Reason, msg.FailedBatchNum); err != nil {
475 return tracerr.Wrap(fmt.Errorf("Coordinator.handleStopPipeline: %w", err))
476 }
477 default:
478 log.Fatalw("Coordinator Unexpected Coordinator msg of type %T: %+v", msg, msg)
479 }
480 return nil
481}
482
483// Start the coordinator
484func (c *Coordinator) Start() {
485 if c.started {
486 log.Fatal("Coordinator already started")
487 }
488 c.started = true
489 c.wg.Add(1)
490 go func() {
491 c.txManager.Run(c.ctx)
492 c.wg.Done()
493 }()
494
495 c.wg.Add(1)
496 go func() {
497 waitDuration := longWaitDuration
498 for {
499 select {
500 case <-c.ctx.Done():
501 log.Info("Coordinator done")
502 c.wg.Done()
503 return
504 case msg := <-c.msgCh:
505 if err := c.handleMsg(c.ctx, msg); c.ctx.Err() != nil {
506 continue
507 } else if err != nil {
508 log.Errorw("Coordinator.handleMsg", "err", err)
509 waitDuration = c.cfg.SyncRetryInterval
510 continue
511 }
512 waitDuration = longWaitDuration
513 case <-time.After(waitDuration):
514 if !c.stats.Synced() {
515 waitDuration = longWaitDuration
516 continue
517 }
518 if err := c.syncStats(c.ctx, &c.stats); c.ctx.Err() != nil {
519 continue
520 } else if err != nil {
521 log.Errorw("Coordinator.syncStats", "err", err)
522 waitDuration = c.cfg.SyncRetryInterval
523 continue
524 }
525 waitDuration = longWaitDuration
526 }
527 }
528 }()
529}
530
531const stopCtxTimeout = 200 * time.Millisecond
532
533// Stop the coordinator
534func (c *Coordinator) Stop() {
535 if !c.started {
536 log.Fatal("Coordinator already stopped")
537 }
538 c.started = false
539 log.Infow("Stopping Coordinator...")
540 c.cancel()
541 c.wg.Wait()
542 if c.pipeline != nil {
543 ctx, cancel := context.WithTimeout(context.Background(), stopCtxTimeout)
544 defer cancel()
545 c.pipeline.Stop(ctx)
546 c.pipeline = nil
547 }
548}