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
175func (c *Coordinator) newPipeline(ctx context.Context) (*Pipeline, error) {
176 return NewPipeline(ctx, c.cfg, c.historyDB, c.l2DB, c.txSelector,
177 c.batchBuilder, c.purger, c.txManager, c.provers, &c.consts)
178}
179
180// MsgSyncBlock indicates an update to the Synchronizer stats
181type MsgSyncBlock struct {
182 Stats synchronizer.Stats
183 Batches []common.BatchData
184 // Vars contains each Smart Contract variables if they are updated, or
185 // nil if they haven't changed.
186 Vars synchronizer.SCVariablesPtr
187}
188
189// MsgSyncReorg indicates a reorg
190type MsgSyncReorg struct {
191 Stats synchronizer.Stats
192 Vars synchronizer.SCVariablesPtr
193}
194
195// MsgStopPipeline indicates a signal to reset the pipeline
196type MsgStopPipeline struct {
197 Reason string
198}
199
200// SendMsg is a thread safe method to pass a message to the Coordinator
201func (c *Coordinator) SendMsg(ctx context.Context, msg interface{}) {
202 select {
203 case c.msgCh <- msg:
204 case <-ctx.Done():
205 }
206}
207
208func (c *Coordinator) syncSCVars(vars synchronizer.SCVariablesPtr) {
209 if vars.Rollup != nil {
210 c.vars.Rollup = *vars.Rollup
211 }
212 if vars.Auction != nil {
213 c.vars.Auction = *vars.Auction
214 }
215 if vars.WDelayer != nil {
216 c.vars.WDelayer = *vars.WDelayer
217 }
218}
219
220func canForge(auctionConstants *common.AuctionConstants, auctionVars *common.AuctionVariables,
221 currentSlot *common.Slot, nextSlot *common.Slot, addr ethCommon.Address, blockNum int64) bool {
222 var slot *common.Slot
223 if currentSlot.StartBlock <= blockNum && blockNum <= currentSlot.EndBlock {
224 slot = currentSlot
225 } else if nextSlot.StartBlock <= blockNum && blockNum <= nextSlot.EndBlock {
226 slot = nextSlot
227 } else {
228 log.Warnw("Coordinator: requested blockNum for canForge is outside slot",
229 "blockNum", blockNum, "currentSlot", currentSlot,
230 "nextSlot", nextSlot,
231 )
232 return false
233 }
234 anyoneForge := false
235 if !slot.ForgerCommitment &&
236 auctionConstants.RelativeBlock(blockNum) >= int64(auctionVars.SlotDeadline) {
237 log.Debugw("Coordinator: anyone can forge in the current slot (slotDeadline passed)",
238 "block", blockNum)
239 anyoneForge = true
240 }
241 if slot.Forger == addr || anyoneForge {
242 return true
243 }
244 return false
245}
246
247func (c *Coordinator) canForge() bool {
248 blockNum := c.stats.Eth.LastBlock.Num + 1
249 return canForge(&c.consts.Auction, &c.vars.Auction,
250 &c.stats.Sync.Auction.CurrentSlot, &c.stats.Sync.Auction.NextSlot,
251 c.cfg.ForgerAddress, blockNum)
252}
253
254func (c *Coordinator) syncStats(ctx context.Context, stats *synchronizer.Stats) error {
255 canForge := c.canForge()
256 if c.pipeline == nil {
257 if canForge {
258 log.Infow("Coordinator: forging state begin", "block",
259 stats.Eth.LastBlock.Num+1, "batch", stats.Sync.LastBatch)
260 batchNum := common.BatchNum(stats.Sync.LastBatch)
261 var err error
262 if c.pipeline, err = c.newPipeline(ctx); err != nil {
263 return tracerr.Wrap(err)
264 }
265 if err := c.pipeline.Start(batchNum, stats, &c.vars); err != nil {
266 c.pipeline = nil
267 return tracerr.Wrap(err)
268 }
269 c.pipelineBatchNum = batchNum
270 }
271 } else {
272 if !canForge {
273 log.Infow("Coordinator: forging state end", "block", stats.Eth.LastBlock.Num+1)
274 c.pipeline.Stop(c.ctx)
275 c.pipeline = nil
276 }
277 }
278 if c.pipeline == nil {
279 // Mark invalid in Pool due to forged L2Txs
280 // for _, batch := range batches {
281 // if err := c.l2DB.InvalidateOldNonces(
282 // idxsNonceFromL2Txs(batch.L2Txs), batch.Batch.BatchNum); err != nil {
283 // return err
284 // }
285 // }
286 if c.purger.CanInvalidate(stats.Sync.LastBlock.Num, stats.Sync.LastBatch) {
287 if err := c.txSelector.Reset(common.BatchNum(stats.Sync.LastBatch)); err != nil {
288 return tracerr.Wrap(err)
289 }
290 }
291 _, err := c.purger.InvalidateMaybe(c.l2DB, c.txSelector.LocalAccountsDB(),
292 stats.Sync.LastBlock.Num, stats.Sync.LastBatch)
293 if err != nil {
294 return tracerr.Wrap(err)
295 }
296 _, err = c.purger.PurgeMaybe(c.l2DB, stats.Sync.LastBlock.Num, stats.Sync.LastBatch)
297 if err != nil {
298 return tracerr.Wrap(err)
299 }
300 }
301 return nil
302}
303
304func (c *Coordinator) handleMsgSyncBlock(ctx context.Context, msg *MsgSyncBlock) error {
305 c.stats = msg.Stats
306 c.syncSCVars(msg.Vars)
307 c.txManager.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
308 if c.pipeline != nil {
309 c.pipeline.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
310 }
311 if !c.stats.Synced() {
312 return nil
313 }
314 return c.syncStats(ctx, &c.stats)
315}
316
317func (c *Coordinator) handleReorg(ctx context.Context, msg *MsgSyncReorg) error {
318 c.stats = msg.Stats
319 c.syncSCVars(msg.Vars)
320 c.txManager.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
321 if c.pipeline != nil {
322 c.pipeline.SetSyncStatsVars(ctx, &msg.Stats, &msg.Vars)
323 }
324 if common.BatchNum(c.stats.Sync.LastBatch) < c.pipelineBatchNum {
325 // There's been a reorg and the batch from which the pipeline
326 // was started was in a block that was discarded. The batch
327 // may not be in the main chain, so we stop the pipeline as a
328 // precaution (it will be started again once the node is in
329 // sync).
330 log.Infow("Coordinator.handleReorg StopPipeline sync.LastBatch < c.pipelineBatchNum",
331 "sync.LastBatch", c.stats.Sync.LastBatch,
332 "c.pipelineBatchNum", c.pipelineBatchNum)
333 if err := c.handleStopPipeline(ctx, "reorg"); err != nil {
334 return tracerr.Wrap(err)
335 }
336 }
337 return nil
338}
339
340func (c *Coordinator) handleStopPipeline(ctx context.Context, reason string) error {
341 if err := c.l2DB.Reorg(common.BatchNum(c.stats.Sync.LastBatch)); err != nil {
342 return tracerr.Wrap(err)
343 }
344 if c.pipeline != nil {
345 c.pipeline.Stop(c.ctx)
346 c.pipeline = nil
347 }
348 if strings.Contains(reason, common.AuctionErrMsgCannotForge) { //nolint:staticcheck
349 // TODO: Check that we are in a slot in which we can't forge
350 }
351 return nil
352}
353
354func (c *Coordinator) handleMsg(ctx context.Context, msg interface{}) error {
355 switch msg := msg.(type) {
356 case MsgSyncBlock:
357 if err := c.handleMsgSyncBlock(ctx, &msg); err != nil {
358 return tracerr.Wrap(fmt.Errorf("Coordinator.handleMsgSyncBlock error: %w", err))
359 }
360 case MsgSyncReorg:
361 if err := c.handleReorg(ctx, &msg); err != nil {
362 return tracerr.Wrap(fmt.Errorf("Coordinator.handleReorg error: %w", err))
363 }
364 case MsgStopPipeline:
365 log.Infow("Coordinator received MsgStopPipeline", "reason", msg.Reason)
366 if err := c.handleStopPipeline(ctx, msg.Reason); err != nil {
367 return tracerr.Wrap(fmt.Errorf("Coordinator.handleStopPipeline: %w", err))
368 }
369 default:
370 log.Fatalw("Coordinator Unexpected Coordinator msg of type %T: %+v", msg, msg)
371 }
372 return nil
373}
374
375// Start the coordinator
376func (c *Coordinator) Start() {
377 if c.started {
378 log.Fatal("Coordinator already started")
379 }
380 c.started = true
381 c.wg.Add(1)
382 go func() {
383 c.txManager.Run(c.ctx)
384 c.wg.Done()
385 }()
386
387 c.wg.Add(1)
388 go func() {
389 waitDuration := longWaitDuration
390 for {
391 select {
392 case <-c.ctx.Done():
393 log.Info("Coordinator done")
394 c.wg.Done()
395 return
396 case msg := <-c.msgCh:
397 if err := c.handleMsg(c.ctx, msg); c.ctx.Err() != nil {
398 continue
399 } else if err != nil {
400 log.Errorw("Coordinator.handleMsg", "err", err)
401 waitDuration = c.cfg.SyncRetryInterval
402 continue
403 }
404 waitDuration = longWaitDuration
405 case <-time.After(waitDuration):
406 if !c.stats.Synced() {
407 waitDuration = longWaitDuration
408 continue
409 }
410 if err := c.syncStats(c.ctx, &c.stats); c.ctx.Err() != nil {
411 continue
412 } else if err != nil {
413 log.Errorw("Coordinator.syncStats", "err", err)
414 waitDuration = c.cfg.SyncRetryInterval
415 continue
416 }
417 waitDuration = longWaitDuration
418 }
419 }
420 }()
421}
422
423const stopCtxTimeout = 200 * time.Millisecond
424
425// Stop the coordinator
426func (c *Coordinator) Stop() {
427 if !c.started {
428 log.Fatal("Coordinator already stopped")
429 }
430 c.started = false
431 log.Infow("Stopping Coordinator...")
432 c.cancel()
433 c.wg.Wait()
434 if c.pipeline != nil {
435 ctx, cancel := context.WithTimeout(context.Background(), stopCtxTimeout)
436 defer cancel()
437 c.pipeline.Stop(ctx)
438 c.pipeline = nil
439 }
440}