hermez-node/node/node.go
node/node.goRaw
1package node
2
3import (
4 "context"
5 "fmt"
6 "net/http"
7 "sync"
8 "time"
9
10 ethCommon "github.com/ethereum/go-ethereum/common"
11 "github.com/ethereum/go-ethereum/ethclient"
12 "github.com/gin-contrib/cors"
13 "github.com/gin-gonic/gin"
14 "github.com/hermeznetwork/hermez-node/api"
15 "github.com/hermeznetwork/hermez-node/batchbuilder"
16 "github.com/hermeznetwork/hermez-node/common"
17 "github.com/hermeznetwork/hermez-node/config"
18 "github.com/hermeznetwork/hermez-node/coordinator"
19 dbUtils "github.com/hermeznetwork/hermez-node/db"
20 "github.com/hermeznetwork/hermez-node/db/historydb"
21 "github.com/hermeznetwork/hermez-node/db/l2db"
22 "github.com/hermeznetwork/hermez-node/db/statedb"
23 "github.com/hermeznetwork/hermez-node/eth"
24 "github.com/hermeznetwork/hermez-node/log"
25 "github.com/hermeznetwork/hermez-node/priceupdater"
26 "github.com/hermeznetwork/hermez-node/prover"
27 "github.com/hermeznetwork/hermez-node/synchronizer"
28 "github.com/hermeznetwork/hermez-node/test/debugapi"
29 "github.com/hermeznetwork/hermez-node/txselector"
30 "github.com/hermeznetwork/tracerr"
31 "github.com/jmoiron/sqlx"
32)
33
34// Mode sets the working mode of the node (synchronizer or coordinator)
35type Mode string
36
37const (
38 // ModeCoordinator defines the mode of the HermezNode as Coordinator, which
39 // means that the node is set to forge (which also will be synchronizing with
40 // the L1 blockchain state)
41 ModeCoordinator Mode = "coordinator"
42
43 // ModeSynchronizer defines the mode of the HermezNode as Synchronizer, which
44 // means that the node is set to only synchronize with the L1 blockchain state
45 // and will not forge
46 ModeSynchronizer Mode = "synchronizer"
47)
48
49// Node is the Hermez Node
50type Node struct {
51 nodeAPI *NodeAPI
52 debugAPI *debugapi.DebugAPI
53 priceUpdater *priceupdater.PriceUpdater
54 // Coordinator
55 coord *coordinator.Coordinator
56
57 // Synchronizer
58 sync *synchronizer.Synchronizer
59
60 // General
61 cfg *config.Node
62 mode Mode
63 sqlConn *sqlx.DB
64 ctx context.Context
65 wg sync.WaitGroup
66 cancel context.CancelFunc
67}
68
69// NewNode creates a Node
70func NewNode(mode Mode, cfg *config.Node) (*Node, error) {
71 // Stablish DB connection
72 db, err := dbUtils.InitSQLDB(
73 cfg.PostgreSQL.Port,
74 cfg.PostgreSQL.Host,
75 cfg.PostgreSQL.User,
76 cfg.PostgreSQL.Password,
77 cfg.PostgreSQL.Name,
78 )
79 if err != nil {
80 return nil, tracerr.Wrap(err)
81 }
82
83 historyDB := historydb.NewHistoryDB(db)
84
85 ethClient, err := ethclient.Dial(cfg.Web3.URL)
86 if err != nil {
87 return nil, tracerr.Wrap(err)
88 }
89 var ethCfg eth.EthereumConfig
90 if mode == ModeCoordinator {
91 ethCfg = eth.EthereumConfig{
92 CallGasLimit: cfg.Coordinator.EthClient.CallGasLimit,
93 DeployGasLimit: cfg.Coordinator.EthClient.DeployGasLimit,
94 GasPriceDiv: cfg.Coordinator.EthClient.GasPriceDiv,
95 ReceiptTimeout: cfg.Coordinator.EthClient.ReceiptTimeout.Duration,
96 IntervalReceiptLoop: cfg.Coordinator.EthClient.ReceiptLoopInterval.Duration,
97 }
98 }
99 client, err := eth.NewClient(ethClient, nil, nil, ð.ClientConfig{
100 Ethereum: ethCfg,
101 Rollup: eth.RollupConfig{
102 Address: cfg.SmartContracts.Rollup,
103 },
104 Auction: eth.AuctionConfig{
105 Address: cfg.SmartContracts.Auction,
106 TokenHEZ: eth.TokenConfig{
107 Address: cfg.SmartContracts.TokenHEZ,
108 Name: cfg.SmartContracts.TokenHEZName,
109 },
110 },
111 WDelayer: eth.WDelayerConfig{
112 Address: cfg.SmartContracts.WDelayer,
113 },
114 })
115 if err != nil {
116 return nil, tracerr.Wrap(err)
117 }
118
119 chainID, err := client.EthChainID()
120 if err != nil {
121 return nil, tracerr.Wrap(err)
122 }
123 if !chainID.IsUint64() {
124 return nil, tracerr.Wrap(fmt.Errorf("chainID cannot be represented as uint64"))
125 }
126 chainIDU64 := chainID.Uint64()
127 const maxUint16 uint64 = 0xffff
128 if chainIDU64 > maxUint16 {
129 return nil, tracerr.Wrap(fmt.Errorf("chainID overflows uint16"))
130 }
131 chainIDU16 := uint16(chainIDU64)
132
133 const safeStateDBKeep = 128
134 if cfg.StateDB.Keep < safeStateDBKeep {
135 return nil, tracerr.Wrap(fmt.Errorf("cfg.StateDB.Keep = %v < %v, which is unsafe",
136 cfg.StateDB.Keep, safeStateDBKeep))
137 }
138 stateDB, err := statedb.NewStateDB(cfg.StateDB.Path, cfg.StateDB.Keep,
139 statedb.TypeSynchronizer, 32)
140 if err != nil {
141 return nil, tracerr.Wrap(err)
142 }
143
144 sync, err := synchronizer.NewSynchronizer(client, historyDB, stateDB, synchronizer.Config{
145 StatsRefreshPeriod: cfg.Synchronizer.StatsRefreshPeriod.Duration,
146 })
147 if err != nil {
148 return nil, tracerr.Wrap(err)
149 }
150 initSCVars := sync.SCVars()
151
152 scConsts := synchronizer.SCConsts{
153 Rollup: *sync.RollupConstants(),
154 Auction: *sync.AuctionConstants(),
155 WDelayer: *sync.WDelayerConstants(),
156 }
157
158 var coord *coordinator.Coordinator
159 var l2DB *l2db.L2DB
160 if mode == ModeCoordinator {
161 l2DB = l2db.NewL2DB(
162 db,
163 cfg.Coordinator.L2DB.SafetyPeriod,
164 cfg.Coordinator.L2DB.MaxTxs,
165 cfg.Coordinator.L2DB.TTL.Duration,
166 )
167 // TODO: Get (maxL1UserTxs, maxL1OperatorTxs, maxTxs) from the smart contract
168 coordAccount := &txselector.CoordAccount{ // TODO TMP
169 Addr: ethCommon.HexToAddress("0xc58d29fA6e86E4FAe04DDcEd660d45BCf3Cb2370"),
170 BJJ: common.EmptyBJJComp,
171 AccountCreationAuth: nil,
172 }
173 txSelector, err := txselector.NewTxSelector(coordAccount, cfg.Coordinator.TxSelector.Path, stateDB, l2DB)
174 if err != nil {
175 return nil, tracerr.Wrap(err)
176 }
177 // TODO: Get (configCircuits []ConfigCircuit, batchNum common.BatchNum, nLevels uint64) from smart contract
178 nLevels := uint64(32) //nolint:gomnd
179 batchBuilder, err := batchbuilder.NewBatchBuilder(cfg.Coordinator.BatchBuilder.Path, stateDB, nil, 0, nLevels)
180 if err != nil {
181 return nil, tracerr.Wrap(err)
182 }
183 if err != nil {
184 return nil, tracerr.Wrap(err)
185 }
186 serverProofs := make([]prover.Client, len(cfg.Coordinator.ServerProofs))
187 for i, serverProofCfg := range cfg.Coordinator.ServerProofs {
188 serverProofs[i] = prover.NewProofServerClient(serverProofCfg.URL,
189 cfg.Coordinator.ProofServerPollInterval.Duration)
190 }
191
192 coord, err = coordinator.NewCoordinator(
193 coordinator.Config{
194 ForgerAddress: cfg.Coordinator.ForgerAddress,
195 ConfirmBlocks: cfg.Coordinator.ConfirmBlocks,
196 L1BatchTimeoutPerc: cfg.Coordinator.L1BatchTimeoutPerc,
197 SyncRetryInterval: cfg.Coordinator.SyncRetryInterval.Duration,
198 EthClientAttempts: cfg.Coordinator.EthClient.Attempts,
199 EthClientAttemptsDelay: cfg.Coordinator.EthClient.AttemptsDelay.Duration,
200 TxManagerCheckInterval: cfg.Coordinator.EthClient.CheckLoopInterval.Duration,
201 DebugBatchPath: cfg.Coordinator.Debug.BatchPath,
202 Purger: coordinator.PurgerCfg{
203 PurgeBatchDelay: cfg.Coordinator.L2DB.PurgeBatchDelay,
204 InvalidateBatchDelay: cfg.Coordinator.L2DB.InvalidateBatchDelay,
205 PurgeBlockDelay: cfg.Coordinator.L2DB.PurgeBlockDelay,
206 InvalidateBlockDelay: cfg.Coordinator.L2DB.InvalidateBlockDelay,
207 },
208 },
209 historyDB,
210 l2DB,
211 txSelector,
212 batchBuilder,
213 serverProofs,
214 client,
215 &scConsts,
216 &synchronizer.SCVariables{
217 Rollup: *initSCVars.Rollup,
218 Auction: *initSCVars.Auction,
219 WDelayer: *initSCVars.WDelayer,
220 },
221 )
222 if err != nil {
223 return nil, tracerr.Wrap(err)
224 }
225 }
226 var nodeAPI *NodeAPI
227 if cfg.API.Address != "" {
228 if cfg.API.UpdateMetricsInterval.Duration == 0 {
229 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateMetricsInterval: %v",
230 cfg.API.UpdateMetricsInterval.Duration))
231 }
232 if cfg.API.UpdateRecommendedFeeInterval.Duration == 0 {
233 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateRecommendedFeeInterval: %v",
234 cfg.API.UpdateRecommendedFeeInterval.Duration))
235 }
236 server := gin.Default()
237 coord := false
238 if mode == ModeCoordinator {
239 coord = cfg.Coordinator.API.Coordinator
240 }
241 var err error
242 nodeAPI, err = NewNodeAPI(
243 cfg.API.Address,
244 coord, cfg.API.Explorer,
245 server,
246 historyDB,
247 stateDB,
248 l2DB,
249 &api.Config{
250 RollupConstants: scConsts.Rollup,
251 AuctionConstants: scConsts.Auction,
252 WDelayerConstants: scConsts.WDelayer,
253 ChainID: chainIDU16,
254 HermezAddress: cfg.SmartContracts.Rollup,
255 },
256 )
257 if err != nil {
258 return nil, tracerr.Wrap(err)
259 }
260 nodeAPI.api.SetRollupVariables(*initSCVars.Rollup)
261 nodeAPI.api.SetAuctionVariables(*initSCVars.Auction)
262 nodeAPI.api.SetWDelayerVariables(*initSCVars.WDelayer)
263 }
264 var debugAPI *debugapi.DebugAPI
265 if cfg.Debug.APIAddress != "" {
266 debugAPI = debugapi.NewDebugAPI(cfg.Debug.APIAddress, stateDB, sync)
267 }
268 priceUpdater, err := priceupdater.NewPriceUpdater(cfg.PriceUpdater.URL,
269 priceupdater.APIType(cfg.PriceUpdater.Type), historyDB)
270 if err != nil {
271 return nil, tracerr.Wrap(err)
272 }
273 ctx, cancel := context.WithCancel(context.Background())
274 return &Node{
275 nodeAPI: nodeAPI,
276 debugAPI: debugAPI,
277 priceUpdater: priceUpdater,
278 coord: coord,
279 sync: sync,
280 cfg: cfg,
281 mode: mode,
282 sqlConn: db,
283 ctx: ctx,
284 cancel: cancel,
285 }, nil
286}
287
288// NodeAPI holds the node http API
289type NodeAPI struct { //nolint:golint
290 api *api.API
291 engine *gin.Engine
292 addr string
293}
294
295func handleNoRoute(c *gin.Context) {
296 c.JSON(http.StatusNotFound, gin.H{
297 "error": "404 page not found",
298 })
299}
300
301// NewNodeAPI creates a new NodeAPI (which internally calls api.NewAPI)
302func NewNodeAPI(
303 addr string,
304 coordinatorEndpoints, explorerEndpoints bool,
305 server *gin.Engine,
306 hdb *historydb.HistoryDB,
307 sdb *statedb.StateDB,
308 l2db *l2db.L2DB,
309 config *api.Config,
310) (*NodeAPI, error) {
311 engine := gin.Default()
312 engine.NoRoute(handleNoRoute)
313 engine.Use(cors.Default())
314 _api, err := api.NewAPI(
315 coordinatorEndpoints, explorerEndpoints,
316 engine,
317 hdb,
318 sdb,
319 l2db,
320 config,
321 )
322 if err != nil {
323 return nil, tracerr.Wrap(err)
324 }
325 return &NodeAPI{
326 addr: addr,
327 api: _api,
328 engine: engine,
329 }, nil
330}
331
332// Run starts the http server of the NodeAPI. To stop it, pass a context with
333// cancelation.
334func (a *NodeAPI) Run(ctx context.Context) error {
335 server := &http.Server{
336 Addr: a.addr,
337 Handler: a.engine,
338 // TODO: Figure out best parameters for production
339 ReadTimeout: 30 * time.Second, //nolint:gomnd
340 WriteTimeout: 30 * time.Second, //nolint:gomnd
341 MaxHeaderBytes: 1 << 20, //nolint:gomnd
342 }
343 go func() {
344 log.Infof("NodeAPI is ready at %v", a.addr)
345 if err := server.ListenAndServe(); err != nil && tracerr.Unwrap(err) != http.ErrServerClosed {
346 log.Fatalf("Listen: %s\n", err)
347 }
348 }()
349
350 <-ctx.Done()
351 log.Info("Stopping NodeAPI...")
352 ctxTimeout, cancel := context.WithTimeout(context.Background(), 10*time.Second) //nolint:gomnd
353 defer cancel()
354 if err := server.Shutdown(ctxTimeout); err != nil {
355 return tracerr.Wrap(err)
356 }
357 log.Info("NodeAPI done")
358 return nil
359}
360
361func (n *Node) handleNewBlock(stats *synchronizer.Stats, vars synchronizer.SCVariablesPtr,
362 batches []common.BatchData) {
363 if n.mode == ModeCoordinator {
364 n.coord.SendMsg(coordinator.MsgSyncBlock{
365 Stats: *stats,
366 Batches: batches,
367 Vars: synchronizer.SCVariablesPtr{
368 Rollup: vars.Rollup,
369 Auction: vars.Auction,
370 WDelayer: vars.WDelayer,
371 },
372 })
373 }
374 if n.nodeAPI != nil {
375 if vars.Rollup != nil {
376 n.nodeAPI.api.SetRollupVariables(*vars.Rollup)
377 }
378 if vars.Auction != nil {
379 n.nodeAPI.api.SetAuctionVariables(*vars.Auction)
380 }
381 if vars.WDelayer != nil {
382 n.nodeAPI.api.SetWDelayerVariables(*vars.WDelayer)
383 }
384
385 if stats.Synced() {
386 if err := n.nodeAPI.api.UpdateNetworkInfo(
387 stats.Eth.LastBlock, stats.Sync.LastBlock,
388 common.BatchNum(stats.Eth.LastBatch),
389 stats.Sync.Auction.CurrentSlot.SlotNum,
390 ); err != nil {
391 log.Errorw("API.UpdateNetworkInfo", "err", err)
392 }
393 }
394 }
395}
396
397func (n *Node) handleReorg(stats *synchronizer.Stats) {
398 if n.mode == ModeCoordinator {
399 n.coord.SendMsg(coordinator.MsgSyncReorg{
400 Stats: *stats,
401 })
402 }
403 if n.nodeAPI != nil {
404 vars := n.sync.SCVars()
405 n.nodeAPI.api.SetRollupVariables(*vars.Rollup)
406 n.nodeAPI.api.SetAuctionVariables(*vars.Auction)
407 n.nodeAPI.api.SetWDelayerVariables(*vars.WDelayer)
408 n.nodeAPI.api.UpdateNetworkInfoBlock(
409 stats.Eth.LastBlock, stats.Sync.LastBlock,
410 )
411 }
412}
413
414// TODO(Edu): Consider keeping the `lastBlock` inside synchronizer so that we
415// don't have to pass it around.
416func (n *Node) syncLoopFn(ctx context.Context, lastBlock *common.Block) (*common.Block, time.Duration, error) {
417 blockData, discarded, err := n.sync.Sync2(ctx, lastBlock)
418 stats := n.sync.Stats()
419 if err != nil {
420 // case: error
421 return nil, n.cfg.Synchronizer.SyncLoopInterval.Duration, tracerr.Wrap(err)
422 } else if discarded != nil {
423 // case: reorg
424 log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
425 n.handleReorg(stats)
426 return nil, time.Duration(0), nil
427 } else if blockData != nil {
428 // case: new block
429 n.handleNewBlock(stats, synchronizer.SCVariablesPtr{
430 Rollup: blockData.Rollup.Vars,
431 Auction: blockData.Auction.Vars,
432 WDelayer: blockData.WDelayer.Vars,
433 }, blockData.Rollup.Batches)
434 return &blockData.Block, time.Duration(0), nil
435 } else {
436 // case: no block
437 return lastBlock, n.cfg.Synchronizer.SyncLoopInterval.Duration, nil
438 }
439}
440
441// StartSynchronizer starts the synchronizer
442func (n *Node) StartSynchronizer() {
443 log.Info("Starting Synchronizer...")
444
445 // Trigger a manual call to handleNewBlock with the loaded state of the
446 // synchronizer in order to quickly activate the API and Coordinator
447 // and avoid waiting for the next block. Without this, the API and
448 // Coordinator will not react until the following block (starting from
449 // the last synced one) is synchronized
450 stats := n.sync.Stats()
451 vars := n.sync.SCVars()
452 n.handleNewBlock(stats, vars, []common.BatchData{})
453
454 n.wg.Add(1)
455 go func() {
456 var err error
457 var lastBlock *common.Block
458 waitDuration := time.Duration(0)
459 for {
460 select {
461 case <-n.ctx.Done():
462 log.Info("Synchronizer done")
463 n.wg.Done()
464 return
465 case <-time.After(waitDuration):
466 if lastBlock, waitDuration, err = n.syncLoopFn(n.ctx,
467 lastBlock); err != nil {
468 if n.ctx.Err() != nil {
469 continue
470 }
471 log.Errorw("Synchronizer.Sync", "err", err)
472 }
473 }
474 }
475 }()
476
477 n.wg.Add(1)
478 go func() {
479 for {
480 select {
481 case <-n.ctx.Done():
482 log.Info("PriceUpdater done")
483 n.wg.Done()
484 return
485 case <-time.After(n.cfg.PriceUpdater.Interval.Duration):
486 if err := n.priceUpdater.UpdateTokenList(); err != nil {
487 log.Errorw("PriceUpdater.UpdateTokenList()", "err", err)
488 }
489 n.priceUpdater.UpdatePrices(n.ctx)
490 }
491 }
492 }()
493}
494
495// StartDebugAPI starts the DebugAPI
496func (n *Node) StartDebugAPI() {
497 log.Info("Starting DebugAPI...")
498 n.wg.Add(1)
499 go func() {
500 defer func() {
501 log.Info("DebugAPI routine stopped")
502 n.wg.Done()
503 }()
504 if err := n.debugAPI.Run(n.ctx); err != nil {
505 if n.ctx.Err() != nil {
506 return
507 }
508 log.Fatalw("DebugAPI.Run", "err", err)
509 }
510 }()
511}
512
513// StartNodeAPI starts the NodeAPI
514func (n *Node) StartNodeAPI() {
515 log.Info("Starting NodeAPI...")
516 n.wg.Add(1)
517 go func() {
518 defer func() {
519 log.Info("NodeAPI routine stopped")
520 n.wg.Done()
521 }()
522 if err := n.nodeAPI.Run(n.ctx); err != nil {
523 if n.ctx.Err() != nil {
524 return
525 }
526 log.Fatalw("NodeAPI.Run", "err", err)
527 }
528 }()
529
530 n.wg.Add(1)
531 go func() {
532 for {
533 select {
534 case <-n.ctx.Done():
535 log.Info("API.UpdateMetrics loop done")
536 n.wg.Done()
537 return
538 case <-time.After(n.cfg.API.UpdateMetricsInterval.Duration):
539 if err := n.nodeAPI.api.UpdateMetrics(); err != nil {
540 log.Errorw("API.UpdateMetrics", "err", err)
541 }
542 }
543 }
544 }()
545
546 n.wg.Add(1)
547 go func() {
548 for {
549 select {
550 case <-n.ctx.Done():
551 log.Info("API.UpdateRecommendedFee loop done")
552 n.wg.Done()
553 return
554 case <-time.After(n.cfg.API.UpdateRecommendedFeeInterval.Duration):
555 if err := n.nodeAPI.api.UpdateRecommendedFee(); err != nil {
556 log.Errorw("API.UpdateRecommendedFee", "err", err)
557 }
558 }
559 }
560 }()
561}
562
563// Start the node
564func (n *Node) Start() {
565 log.Infow("Starting node...", "mode", n.mode)
566 if n.debugAPI != nil {
567 n.StartDebugAPI()
568 }
569 if n.nodeAPI != nil {
570 n.StartNodeAPI()
571 }
572 if n.mode == ModeCoordinator {
573 log.Info("Starting Coordinator...")
574 n.coord.Start()
575 }
576 n.StartSynchronizer()
577}
578
579// Stop the node
580func (n *Node) Stop() {
581 log.Infow("Stopping node...")
582 n.cancel()
583 if n.mode == ModeCoordinator {
584 log.Info("Stopping Coordinator...")
585 n.coord.Stop()
586 }
587 n.wg.Wait()
588}