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 stateDB, err := statedb.NewStateDB(cfg.StateDB.Path, statedb.TypeSynchronizer, 32, chainIDU16)
134 if err != nil {
135 return nil, tracerr.Wrap(err)
136 }
137
138 sync, err := synchronizer.NewSynchronizer(client, historyDB, stateDB, synchronizer.Config{
139 StatsRefreshPeriod: cfg.Synchronizer.StatsRefreshPeriod.Duration,
140 })
141 if err != nil {
142 return nil, tracerr.Wrap(err)
143 }
144 initSCVars := sync.SCVars()
145
146 scConsts := synchronizer.SCConsts{
147 Rollup: *sync.RollupConstants(),
148 Auction: *sync.AuctionConstants(),
149 WDelayer: *sync.WDelayerConstants(),
150 }
151
152 var coord *coordinator.Coordinator
153 var l2DB *l2db.L2DB
154 if mode == ModeCoordinator {
155 l2DB = l2db.NewL2DB(
156 db,
157 cfg.Coordinator.L2DB.SafetyPeriod,
158 cfg.Coordinator.L2DB.MaxTxs,
159 cfg.Coordinator.L2DB.TTL.Duration,
160 )
161 // TODO: Get (maxL1UserTxs, maxL1OperatorTxs, maxTxs) from the smart contract
162 coordAccount := &txselector.CoordAccount{ // TODO TMP
163 Addr: ethCommon.HexToAddress("0xc58d29fA6e86E4FAe04DDcEd660d45BCf3Cb2370"),
164 BJJ: common.EmptyBJJComp,
165 AccountCreationAuth: nil,
166 }
167 txSelector, err := txselector.NewTxSelector(coordAccount, cfg.Coordinator.TxSelector.Path, stateDB, l2DB)
168 if err != nil {
169 return nil, tracerr.Wrap(err)
170 }
171 // TODO: Get (configCircuits []ConfigCircuit, batchNum common.BatchNum, nLevels uint64) from smart contract
172 nLevels := uint64(32) //nolint:gomnd
173 batchBuilder, err := batchbuilder.NewBatchBuilder(cfg.Coordinator.BatchBuilder.Path, stateDB, nil, 0, nLevels)
174 if err != nil {
175 return nil, tracerr.Wrap(err)
176 }
177 if err != nil {
178 return nil, tracerr.Wrap(err)
179 }
180 serverProofs := make([]prover.Client, len(cfg.Coordinator.ServerProofs))
181 for i, serverProofCfg := range cfg.Coordinator.ServerProofs {
182 serverProofs[i] = prover.NewProofServerClient(serverProofCfg.URL,
183 cfg.Coordinator.ProofServerPollInterval.Duration)
184 }
185
186 coord, err = coordinator.NewCoordinator(
187 coordinator.Config{
188 ForgerAddress: cfg.Coordinator.ForgerAddress,
189 ConfirmBlocks: cfg.Coordinator.ConfirmBlocks,
190 L1BatchTimeoutPerc: cfg.Coordinator.L1BatchTimeoutPerc,
191 SyncRetryInterval: cfg.Coordinator.SyncRetryInterval.Duration,
192 EthClientAttempts: cfg.Coordinator.EthClient.Attempts,
193 EthClientAttemptsDelay: cfg.Coordinator.EthClient.AttemptsDelay.Duration,
194 TxManagerCheckInterval: cfg.Coordinator.EthClient.CheckLoopInterval.Duration,
195 DebugBatchPath: cfg.Coordinator.Debug.BatchPath,
196 Purger: coordinator.PurgerCfg{
197 PurgeBatchDelay: cfg.Coordinator.L2DB.PurgeBatchDelay,
198 InvalidateBatchDelay: cfg.Coordinator.L2DB.InvalidateBatchDelay,
199 PurgeBlockDelay: cfg.Coordinator.L2DB.PurgeBlockDelay,
200 InvalidateBlockDelay: cfg.Coordinator.L2DB.InvalidateBlockDelay,
201 },
202 },
203 historyDB,
204 l2DB,
205 txSelector,
206 batchBuilder,
207 serverProofs,
208 client,
209 &scConsts,
210 &synchronizer.SCVariables{
211 Rollup: *initSCVars.Rollup,
212 Auction: *initSCVars.Auction,
213 WDelayer: *initSCVars.WDelayer,
214 },
215 )
216 if err != nil {
217 return nil, tracerr.Wrap(err)
218 }
219 }
220 var nodeAPI *NodeAPI
221 if cfg.API.Address != "" {
222 if cfg.API.UpdateMetricsInterval.Duration == 0 {
223 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateMetricsInterval: %v",
224 cfg.API.UpdateMetricsInterval.Duration))
225 }
226 if cfg.API.UpdateRecommendedFeeInterval.Duration == 0 {
227 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateRecommendedFeeInterval: %v",
228 cfg.API.UpdateRecommendedFeeInterval.Duration))
229 }
230 server := gin.Default()
231 coord := false
232 if mode == ModeCoordinator {
233 coord = cfg.Coordinator.API.Coordinator
234 }
235 var err error
236 nodeAPI, err = NewNodeAPI(
237 cfg.API.Address,
238 coord, cfg.API.Explorer,
239 server,
240 historyDB,
241 stateDB,
242 l2DB,
243 &api.Config{
244 RollupConstants: scConsts.Rollup,
245 AuctionConstants: scConsts.Auction,
246 WDelayerConstants: scConsts.WDelayer,
247 },
248 chainIDU16,
249 )
250 if err != nil {
251 return nil, tracerr.Wrap(err)
252 }
253 nodeAPI.api.SetRollupVariables(*initSCVars.Rollup)
254 nodeAPI.api.SetAuctionVariables(*initSCVars.Auction)
255 nodeAPI.api.SetWDelayerVariables(*initSCVars.WDelayer)
256 }
257 var debugAPI *debugapi.DebugAPI
258 if cfg.Debug.APIAddress != "" {
259 debugAPI = debugapi.NewDebugAPI(cfg.Debug.APIAddress, stateDB, sync)
260 }
261 priceUpdater, err := priceupdater.NewPriceUpdater(cfg.PriceUpdater.URL,
262 priceupdater.APIType(cfg.PriceUpdater.Type), historyDB)
263 if err != nil {
264 return nil, tracerr.Wrap(err)
265 }
266 ctx, cancel := context.WithCancel(context.Background())
267 return &Node{
268 nodeAPI: nodeAPI,
269 debugAPI: debugAPI,
270 priceUpdater: priceUpdater,
271 coord: coord,
272 sync: sync,
273 cfg: cfg,
274 mode: mode,
275 sqlConn: db,
276 ctx: ctx,
277 cancel: cancel,
278 }, nil
279}
280
281// NodeAPI holds the node http API
282type NodeAPI struct { //nolint:golint
283 api *api.API
284 engine *gin.Engine
285 addr string
286}
287
288func handleNoRoute(c *gin.Context) {
289 c.JSON(http.StatusNotFound, gin.H{
290 "error": "404 page not found",
291 })
292}
293
294// NewNodeAPI creates a new NodeAPI (which internally calls api.NewAPI)
295func NewNodeAPI(
296 addr string,
297 coordinatorEndpoints, explorerEndpoints bool,
298 server *gin.Engine,
299 hdb *historydb.HistoryDB,
300 sdb *statedb.StateDB,
301 l2db *l2db.L2DB,
302 config *api.Config,
303 chainID uint16,
304) (*NodeAPI, error) {
305 engine := gin.Default()
306 engine.NoRoute(handleNoRoute)
307 engine.Use(cors.Default())
308 _api, err := api.NewAPI(
309 coordinatorEndpoints, explorerEndpoints,
310 engine,
311 hdb,
312 sdb,
313 l2db,
314 config,
315 chainID,
316 )
317 if err != nil {
318 return nil, tracerr.Wrap(err)
319 }
320 return &NodeAPI{
321 addr: addr,
322 api: _api,
323 engine: engine,
324 }, nil
325}
326
327// Run starts the http server of the NodeAPI. To stop it, pass a context with
328// cancelation.
329func (a *NodeAPI) Run(ctx context.Context) error {
330 server := &http.Server{
331 Addr: a.addr,
332 Handler: a.engine,
333 // TODO: Figure out best parameters for production
334 ReadTimeout: 30 * time.Second, //nolint:gomnd
335 WriteTimeout: 30 * time.Second, //nolint:gomnd
336 MaxHeaderBytes: 1 << 20, //nolint:gomnd
337 }
338 go func() {
339 log.Infof("NodeAPI is ready at %v", a.addr)
340 if err := server.ListenAndServe(); err != nil && tracerr.Unwrap(err) != http.ErrServerClosed {
341 log.Fatalf("Listen: %s\n", err)
342 }
343 }()
344
345 <-ctx.Done()
346 log.Info("Stopping NodeAPI...")
347 ctxTimeout, cancel := context.WithTimeout(context.Background(), 10*time.Second) //nolint:gomnd
348 defer cancel()
349 if err := server.Shutdown(ctxTimeout); err != nil {
350 return tracerr.Wrap(err)
351 }
352 log.Info("NodeAPI done")
353 return nil
354}
355
356func (n *Node) handleNewBlock(stats *synchronizer.Stats, vars synchronizer.SCVariablesPtr,
357 batches []common.BatchData) {
358 if n.mode == ModeCoordinator {
359 n.coord.SendMsg(coordinator.MsgSyncBlock{
360 Stats: *stats,
361 Batches: batches,
362 Vars: synchronizer.SCVariablesPtr{
363 Rollup: vars.Rollup,
364 Auction: vars.Auction,
365 WDelayer: vars.WDelayer,
366 },
367 })
368 }
369 if n.nodeAPI != nil {
370 if vars.Rollup != nil {
371 n.nodeAPI.api.SetRollupVariables(*vars.Rollup)
372 }
373 if vars.Auction != nil {
374 n.nodeAPI.api.SetAuctionVariables(*vars.Auction)
375 }
376 if vars.WDelayer != nil {
377 n.nodeAPI.api.SetWDelayerVariables(*vars.WDelayer)
378 }
379
380 if stats.Synced() {
381 if err := n.nodeAPI.api.UpdateNetworkInfo(
382 stats.Eth.LastBlock, stats.Sync.LastBlock,
383 common.BatchNum(stats.Eth.LastBatch),
384 stats.Sync.Auction.CurrentSlot.SlotNum,
385 ); err != nil {
386 log.Errorw("API.UpdateNetworkInfo", "err", err)
387 }
388 }
389 }
390}
391
392func (n *Node) handleReorg(stats *synchronizer.Stats) {
393 if n.mode == ModeCoordinator {
394 n.coord.SendMsg(coordinator.MsgSyncReorg{
395 Stats: *stats,
396 })
397 }
398 if n.nodeAPI != nil {
399 vars := n.sync.SCVars()
400 n.nodeAPI.api.SetRollupVariables(*vars.Rollup)
401 n.nodeAPI.api.SetAuctionVariables(*vars.Auction)
402 n.nodeAPI.api.SetWDelayerVariables(*vars.WDelayer)
403 n.nodeAPI.api.UpdateNetworkInfoBlock(
404 stats.Eth.LastBlock, stats.Sync.LastBlock,
405 )
406 }
407}
408
409// TODO(Edu): Consider keeping the `lastBlock` inside synchronizer so that we
410// don't have to pass it around.
411func (n *Node) syncLoopFn(ctx context.Context, lastBlock *common.Block) (*common.Block, time.Duration, error) {
412 blockData, discarded, err := n.sync.Sync2(ctx, lastBlock)
413 stats := n.sync.Stats()
414 if err != nil {
415 // case: error
416 return nil, n.cfg.Synchronizer.SyncLoopInterval.Duration, err
417 } else if discarded != nil {
418 // case: reorg
419 log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
420 n.handleReorg(stats)
421 return nil, time.Duration(0), nil
422 } else if blockData != nil {
423 // case: new block
424 n.handleNewBlock(stats, synchronizer.SCVariablesPtr{
425 Rollup: blockData.Rollup.Vars,
426 Auction: blockData.Auction.Vars,
427 WDelayer: blockData.WDelayer.Vars,
428 }, blockData.Rollup.Batches)
429 return &blockData.Block, time.Duration(0), nil
430 } else {
431 // case: no block
432 return lastBlock, n.cfg.Synchronizer.SyncLoopInterval.Duration, nil
433 }
434}
435
436// StartSynchronizer starts the synchronizer
437func (n *Node) StartSynchronizer() {
438 log.Info("Starting Synchronizer...")
439
440 // Trigger a manual call to handleNewBlock with the loaded state of the
441 // synchronizer in order to quickly activate the API and Coordinator
442 // and avoid waiting for the next block. Without this, the API and
443 // Coordinator will not react until the following block (starting from
444 // the last synced one) is synchronized
445 stats := n.sync.Stats()
446 vars := n.sync.SCVars()
447 n.handleNewBlock(stats, vars, []common.BatchData{})
448
449 n.wg.Add(1)
450 go func() {
451 var err error
452 var lastBlock *common.Block
453 waitDuration := time.Duration(0)
454 for {
455 select {
456 case <-n.ctx.Done():
457 log.Info("Synchronizer done")
458 n.wg.Done()
459 return
460 case <-time.After(waitDuration):
461 if lastBlock, waitDuration, err = n.syncLoopFn(n.ctx,
462 lastBlock); err != nil {
463 if n.ctx.Err() != nil {
464 continue
465 }
466 log.Errorw("Synchronizer.Sync", "err", err)
467 }
468 }
469 }
470 }()
471
472 n.wg.Add(1)
473 go func() {
474 for {
475 select {
476 case <-n.ctx.Done():
477 log.Info("PriceUpdater done")
478 n.wg.Done()
479 return
480 case <-time.After(n.cfg.PriceUpdater.Interval.Duration):
481 if err := n.priceUpdater.UpdateTokenList(); err != nil {
482 log.Errorw("PriceUpdater.UpdateTokenList()", "err", err)
483 }
484 n.priceUpdater.UpdatePrices(n.ctx)
485 }
486 }
487 }()
488}
489
490// StartDebugAPI starts the DebugAPI
491func (n *Node) StartDebugAPI() {
492 log.Info("Starting DebugAPI...")
493 n.wg.Add(1)
494 go func() {
495 defer func() {
496 log.Info("DebugAPI routine stopped")
497 n.wg.Done()
498 }()
499 if err := n.debugAPI.Run(n.ctx); err != nil {
500 if n.ctx.Err() != nil {
501 return
502 }
503 log.Fatalw("DebugAPI.Run", "err", err)
504 }
505 }()
506}
507
508// StartNodeAPI starts the NodeAPI
509func (n *Node) StartNodeAPI() {
510 log.Info("Starting NodeAPI...")
511 n.wg.Add(1)
512 go func() {
513 defer func() {
514 log.Info("NodeAPI routine stopped")
515 n.wg.Done()
516 }()
517 if err := n.nodeAPI.Run(n.ctx); err != nil {
518 if n.ctx.Err() != nil {
519 return
520 }
521 log.Fatalw("NodeAPI.Run", "err", err)
522 }
523 }()
524
525 n.wg.Add(1)
526 go func() {
527 for {
528 select {
529 case <-n.ctx.Done():
530 log.Info("API.UpdateMetrics loop done")
531 n.wg.Done()
532 return
533 case <-time.After(n.cfg.API.UpdateMetricsInterval.Duration):
534 if err := n.nodeAPI.api.UpdateMetrics(); err != nil {
535 log.Errorw("API.UpdateMetrics", "err", err)
536 }
537 }
538 }
539 }()
540
541 n.wg.Add(1)
542 go func() {
543 for {
544 select {
545 case <-n.ctx.Done():
546 log.Info("API.UpdateRecommendedFee loop done")
547 n.wg.Done()
548 return
549 case <-time.After(n.cfg.API.UpdateRecommendedFeeInterval.Duration):
550 if err := n.nodeAPI.api.UpdateRecommendedFee(); err != nil {
551 log.Errorw("API.UpdateRecommendedFee", "err", err)
552 }
553 }
554 }
555 }()
556}
557
558// Start the node
559func (n *Node) Start() {
560 log.Infow("Starting node...", "mode", n.mode)
561 if n.debugAPI != nil {
562 n.StartDebugAPI()
563 }
564 if n.nodeAPI != nil {
565 n.StartNodeAPI()
566 }
567 if n.mode == ModeCoordinator {
568 log.Info("Starting Coordinator...")
569 n.coord.Start()
570 }
571 n.StartSynchronizer()
572}
573
574// Stop the node
575func (n *Node) Stop() {
576 log.Infow("Stopping node...")
577 n.cancel()
578 if n.mode == ModeCoordinator {
579 log.Info("Stopping Coordinator...")
580 n.coord.Stop()
581 }
582 n.wg.Wait()
583}