hermez-node/coordinator/coordinator.go
coordinator/coordinator.goRaw
1package coordinator
2
3import (
4 "context"
5 "fmt"
6 "os"
7 "strings"
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 // EthClientAttempts is the number of attempts to do an eth client RPC
46 // call before giving up
47 EthClientAttempts int
48 // ForgeRetryInterval is the waiting interval between calls forge a
49 // batch after an error
50 ForgeRetryInterval time.Duration
51 // SyncRetryInterval is the waiting interval between calls to the main
52 // handler of a synced block after an error
53 SyncRetryInterval time.Duration
54 // EthClientAttemptsDelay is delay between attempts do do an eth client
55 // RPC call
56 EthClientAttemptsDelay time.Duration
57 // TxManagerCheckInterval is the waiting interval between receipt
58 // checks of ethereum transactions in the TxManager
59 TxManagerCheckInterval time.Duration
60 // DebugBatchPath if set, specifies the path where batchInfo is stored
61 // in JSON in every step/update of the pipeline
62 DebugBatchPath string
63 Purger PurgerCfg
64 VerifierIdx uint8
65 TxProcessorConfig txprocessor.Config
66}
67
68func (c *Config) debugBatchStore(batchInfo *BatchInfo) {
69 if c.DebugBatchPath != "" {
70 if err := batchInfo.DebugStore(c.DebugBatchPath); err != nil {
71 log.Warnw("Error storing debug BatchInfo",
72 "path", c.DebugBatchPath, "err", err)
73 }
74 }
75}
76
77// Coordinator implements the Coordinator type
78type Coordinator struct {
79 // State
80 pipelineBatchNum common.BatchNum // batchNum from which we started the pipeline
81 provers []prover.Client
82 consts synchronizer.SCConsts
83 vars synchronizer.SCVariables
84 stats synchronizer.Stats
85 started bool
86
87 cfg Config
88
89 historyDB *historydb.HistoryDB
90 l2DB *l2db.L2DB
91 txSelector *txselector.TxSelector
92 batchBuilder *batchbuilder.BatchBuilder
93
94 msgCh chan interface{}
95 ctx context.Context
96 wg sync.WaitGroup
97 cancel context.CancelFunc
98
99 pipeline *Pipeline
100
101 purger *Purger
102 txManager *TxManager
103}
104
105// NewCoordinator creates a new Coordinator
106func NewCoordinator(cfg Config,
107 historyDB *historydb.HistoryDB,
108 l2DB *l2db.L2DB,
109 txSelector *txselector.TxSelector,
110 batchBuilder *batchbuilder.BatchBuilder,
111 serverProofs []prover.Client,
112 ethClient eth.ClientInterface,
113 scConsts *synchronizer.SCConsts,
114 initSCVars *synchronizer.SCVariables,
115) (*Coordinator, error) {
116 // nolint reason: hardcoded `1.0`, by design the percentage can't be over 100%
117 if cfg.L1BatchTimeoutPerc >= 1.0 { //nolint:gomnd
118 return nil, tracerr.Wrap(fmt.Errorf("invalid value for Config.L1BatchTimeoutPerc (%v >= 1.0)",
119 cfg.L1BatchTimeoutPerc))
120 }
121 if cfg.EthClientAttempts < 1 {
122 return nil, tracerr.Wrap(fmt.Errorf("invalid value for Config.EthClientAttempts (%v < 1)",
123 cfg.EthClientAttempts))
124 }
125
126 if cfg.DebugBatchPath != "" {
127 if err := os.MkdirAll(cfg.DebugBatchPath, 0744); err != nil {
128 return nil, tracerr.Wrap(err)
129 }
130 }
131
132 purger := Purger{
133 cfg: cfg.Purger,
134 lastPurgeBlock: 0,
135 lastPurgeBatch: 0,
136 lastInvalidateBlock: 0,
137 lastInvalidateBatch: 0,
138 }
139
140 ctx, cancel := context.WithCancel(context.Background())
141 c := Coordinator{
142 pipelineBatchNum: -1,
143 provers: serverProofs,
144 consts: *scConsts,
145 vars: *initSCVars,
146
147 cfg: cfg,
148
149 historyDB: historyDB,
150 l2DB: l2DB,
151 txSelector: txSelector,
152 batchBuilder: batchBuilder,
153
154 purger: &purger,
155
156 msgCh: make(chan interface{}),
157 ctx: ctx,
158 // wg
159 cancel: cancel,
160 }
161 ctxTimeout, ctxTimeoutCancel := context.WithTimeout(ctx, 1*time.Second)
162 defer ctxTimeoutCancel()
163 txManager, err := NewTxManager(ctxTimeout, &cfg, ethClient, l2DB, &c,
164 scConsts, initSCVars)
165 if err != nil {
166 return nil, tracerr.Wrap(err)
167 }
168 c.txManager = txManager
169 // Set Eth LastBlockNum to -1 in stats so that stats.Synced() is
170 // guaranteed to return false before it's updated with a real stats
171 c.stats.Eth.LastBlock.Num = -1
172 return &c, nil
173}
174
175// TxSelector returns the inner TxSelector
176func (c *Coordinator) TxSelector() *txselector.TxSelector {
177 return c.txSelector
178}
179
180// BatchBuilder returns the inner BatchBuilder
181func (c *Coordinator) BatchBuilder() *batchbuilder.BatchBuilder {
182 return c.batchBuilder
183}
184
185func (c *Coordinator) newPipeline(ctx context.Context) (*Pipeline, error) {
186 return NewPipeline(ctx, c.cfg, c.historyDB, c.l2DB, c.txSelector,
187 c.batchBuilder, c.purger, c.txManager, c.provers, &c.consts)
188}
189
190// MsgSyncBlock indicates an update to the Synchronizer stats
191type MsgSyncBlock struct {
192 Stats synchronizer.Stats
193 Batches []common.BatchData
194 // Vars contains each Smart Contract variables if they are updated, or
195 // nil if they haven't changed.
196 Vars synchronizer.SCVariablesPtr
197}
198
199// MsgSyncReorg indicates a reorg
200type MsgSyncReorg struct {
201 Stats synchronizer.Stats
202 Vars synchronizer.SCVariablesPtr
203}
204
205// MsgStopPipeline indicates a signal to reset the pipeline
206type MsgStopPipeline struct {
207 Reason string
208}
209
210// SendMsg is a thread safe method to pass a message to the Coordinator
211func (c *Coordinator) SendMsg(ctx context.Context, msg interface{}) {
212 select {
213 case c.msgCh <- msg:
214 case <-ctx.Done():
215 }
216}
217
218func (c *Coordinator) syncSCVars(vars synchronizer.SCVariablesPtr) {
219 if vars.Rollup != nil {
220 c.vars.Rollup = *vars.Rollup
221 }
222 if vars.Auction != nil {
223 c.vars.Auction = *vars.Auction
224 }
225 if vars.WDelayer != nil {
226 c.vars.WDelayer = *vars.WDelayer
227 }
228}
229
230func canForge(auctionConstants *common.AuctionConstants, auctionVars *common.AuctionVariables,
231 currentSlot *common.Slot, nextSlot *common.Slot, addr ethCommon.Address, blockNum int64) bool {
232 var slot *common.Slot
233 if currentSlot.StartBlock <= blockNum && blockNum <= currentSlot.EndBlock {
234 slot = currentSlot
235 } else if nextSlot.StartBlock <= blockNum && blockNum <= nextSlot.EndBlock {
236 slot = nextSlot
237 } else {
238 log.Warnw("Coordinator: requested blockNum for canForge is outside slot",
239 "blockNum", blockNum, "currentSlot", currentSlot,
240 "nextSlot", nextSlot,
241 )
242 return false
243 }
244 anyoneForge := false
245 if !slot.ForgerCommitment &&
246 auctionConstants.RelativeBlock(blockNum) >= int64(auctionVars.SlotDeadline) {
247 log.Debugw("Coordinator: anyone can forge in the current slot (slotDeadline passed)",
248 "block", blockNum)
249 anyoneForge = true
250 }
251 if slot.Forger == addr || anyoneForge {
252 return true
253 }
254 return false
255}
256
257func (c *Coordinator) canForge() bool {
258 blockNum := c.stats.Eth.LastBlock.Num + 1
259 return canForge(&c.consts.Auction, &c.vars.Auction,
260 &c.stats.Sync.Auction.CurrentSlot, &c.stats.Sync.Auction.NextSlot,
261 c.cfg.ForgerAddress, blockNum)
262}
263
264func (c *Coordinator) syncStats(ctx context.Context, stats *synchronizer.Stats) error {
265 canForge := c.canForge()
266 if c.pipeline == nil {
267 if canForge {
268 log.Infow("Coordinator: forging state begin", "block",
269 stats.Eth.LastBlock.Num+1, "batch", stats.Sync.LastBatch)
270 batchNum := common.BatchNum(stats.Sync.LastBatch)
271 var err error
272 if c.pipeline, err = c.newPipeline(ctx); err != nil {
273 return tracerr.Wrap(err)
274 }
275 if err := c.pipeline.Start(batchNum, stats, &c.vars); err != nil {
276 c.pipeline = nil
277 return tracerr.Wrap(err)
278 }
279 c.pipelineBatchNum = batchNum
280 }
281 } else {
282 if !canForge {
283 log.Infow("Coordinator: forging state end", "block", stats.Eth.LastBlock.Num+1)
284 c.pipeline.Stop(c.ctx)
285 c.pipeline = nil
286 }
287 }
288 if c.pipeline == nil {
289 // Mark invalid in Pool due to forged L2Txs
290 // for _, batch := range batches {
291 // if err := c.l2DB.InvalidateOldNonces(
292 // idxsNonceFromL2Txs(batch.L2Txs), batch.Batch.BatchNum); err != nil {
293 // return err
294 // }
295 // }
296 if c.purger.CanInvalidate(stats.Sync.LastBlock.Num, stats.Sync.LastBatch) {
297 if err := c.txSelector.Reset(common.BatchNum(stats.Sync.LastBatch)); err != nil {
298 return tracerr.Wrap(err)
299 }
300 }
301 _, err := c.purger.InvalidateMaybe(c.l2DB, c.txSelector.LocalAccountsDB(),
302 stats.Sync.LastBlock.Num, stats.Sync.LastBatch)
303 if err != nil {
304 return tracerr.Wrap(err)
305 }
306 _, err = c.purger.PurgeMaybe(c.l2DB, stats.Sync.LastBlock.Num, stats.Sync.LastBatch)
307 if err != nil {
308 return tracerr.Wrap(err)
309 }
310 }
311 return nil
312}
313
314func (c *Coordinator) handleMsgSyncBlock(ctx context.Context, msg *MsgSyncBlock) error {
315 c.stats = msg.Stats
316 c.syncSCVars(msg.Vars)
317 c.txManager.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
318 if c.pipeline != nil {
319 c.pipeline.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
320 }
321 if !c.stats.Synced() {
322 return nil
323 }
324 return c.syncStats(ctx, &c.stats)
325}
326
327func (c *Coordinator) handleReorg(ctx context.Context, msg *MsgSyncReorg) error {
328 c.stats = msg.Stats
329 c.syncSCVars(msg.Vars)
330 c.txManager.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
331 if c.pipeline != nil {
332 c.pipeline.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
333 }
334 if common.BatchNum(c.stats.Sync.LastBatch) < c.pipelineBatchNum {
335 // There's been a reorg and the batch from which the pipeline
336 // was started was in a block that was discarded. The batch
337 // may not be in the main chain, so we stop the pipeline as a
338 // precaution (it will be started again once the node is in
339 // sync).
340 log.Infow("Coordinator.handleReorg StopPipeline sync.LastBatch < c.pipelineBatchNum",
341 "sync.LastBatch", c.stats.Sync.LastBatch,
342 "c.pipelineBatchNum", c.pipelineBatchNum)
343 if err := c.handleStopPipeline(ctx, "reorg"); err != nil {
344 return tracerr.Wrap(err)
345 }
346 }
347 return nil
348}
349
350func (c *Coordinator) handleStopPipeline(ctx context.Context, reason string) error {
351 if c.pipeline != nil {
352 c.pipeline.Stop(c.ctx)
353 c.pipeline = nil
354 }
355 if err := c.l2DB.Reorg(common.BatchNum(c.stats.Sync.LastBatch)); err != nil {
356 return tracerr.Wrap(err)
357 }
358 if strings.Contains(reason, common.AuctionErrMsgCannotForge) { //nolint:staticcheck
359 // TODO: Check that we are in a slot in which we can't forge
360 }
361 return nil
362}
363
364func (c *Coordinator) handleMsg(ctx context.Context, msg interface{}) error {
365 switch msg := msg.(type) {
366 case MsgSyncBlock:
367 if err := c.handleMsgSyncBlock(ctx, &msg); err != nil {
368 return tracerr.Wrap(fmt.Errorf("Coordinator.handleMsgSyncBlock error: %w", err))
369 }
370 case MsgSyncReorg:
371 if err := c.handleReorg(ctx, &msg); err != nil {
372 return tracerr.Wrap(fmt.Errorf("Coordinator.handleReorg error: %w", err))
373 }
374 case MsgStopPipeline:
375 log.Infow("Coordinator received MsgStopPipeline", "reason", msg.Reason)
376 if err := c.handleStopPipeline(ctx, msg.Reason); err != nil {
377 return tracerr.Wrap(fmt.Errorf("Coordinator.handleStopPipeline: %w", err))
378 }
379 default:
380 log.Fatalw("Coordinator Unexpected Coordinator msg of type %T: %+v", msg, msg)
381 }
382 return nil
383}
384
385// Start the coordinator
386func (c *Coordinator) Start() {
387 if c.started {
388 log.Fatal("Coordinator already started")
389 }
390 c.started = true
391 c.wg.Add(1)
392 go func() {
393 c.txManager.Run(c.ctx)
394 c.wg.Done()
395 }()
396
397 c.wg.Add(1)
398 go func() {
399 waitDuration := longWaitDuration
400 for {
401 select {
402 case <-c.ctx.Done():
403 log.Info("Coordinator done")
404 c.wg.Done()
405 return
406 case msg := <-c.msgCh:
407 if err := c.handleMsg(c.ctx, msg); c.ctx.Err() != nil {
408 continue
409 } else if err != nil {
410 log.Errorw("Coordinator.handleMsg", "err", err)
411 waitDuration = c.cfg.SyncRetryInterval
412 continue
413 }
414 waitDuration = longWaitDuration
415 case <-time.After(waitDuration):
416 if !c.stats.Synced() {
417 waitDuration = longWaitDuration
418 continue
419 }
420 if err := c.syncStats(c.ctx, &c.stats); c.ctx.Err() != nil {
421 continue
422 } else if err != nil {
423 log.Errorw("Coordinator.syncStats", "err", err)
424 waitDuration = c.cfg.SyncRetryInterval
425 continue
426 }
427 waitDuration = longWaitDuration
428 }
429 }
430 }()
431}
432
433const stopCtxTimeout = 200 * time.Millisecond
434
435// Stop the coordinator
436func (c *Coordinator) Stop() {
437 if !c.started {
438 log.Fatal("Coordinator already stopped")
439 }
440 c.started = false
441 log.Infow("Stopping Coordinator...")
442 c.cancel()
443 c.wg.Wait()
444 if c.pipeline != nil {
445 ctx, cancel := context.WithTimeout(context.Background(), stopCtxTimeout)
446 defer cancel()
447 c.pipeline.Stop(ctx)
448 c.pipeline = nil
449 }
450}