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