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