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, chainIDU16)
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 },
254 chainIDU16,
255 )
256 if err != nil {
257 return nil, tracerr.Wrap(err)
258 }
259 nodeAPI.api.SetRollupVariables(*initSCVars.Rollup)
260 nodeAPI.api.SetAuctionVariables(*initSCVars.Auction)
261 nodeAPI.api.SetWDelayerVariables(*initSCVars.WDelayer)
262 }
263 var debugAPI *debugapi.DebugAPI
264 if cfg.Debug.APIAddress != "" {
265 debugAPI = debugapi.NewDebugAPI(cfg.Debug.APIAddress, stateDB, sync)
266 }
267 priceUpdater, err := priceupdater.NewPriceUpdater(cfg.PriceUpdater.URL,
268 priceupdater.APIType(cfg.PriceUpdater.Type), historyDB)
269 if err != nil {
270 return nil, tracerr.Wrap(err)
271 }
272 ctx, cancel := context.WithCancel(context.Background())
273 return &Node{
274 nodeAPI: nodeAPI,
275 debugAPI: debugAPI,
276 priceUpdater: priceUpdater,
277 coord: coord,
278 sync: sync,
279 cfg: cfg,
280 mode: mode,
281 sqlConn: db,
282 ctx: ctx,
283 cancel: cancel,
284 }, nil
285}
286
287// NodeAPI holds the node http API
288type NodeAPI struct { //nolint:golint
289 api *api.API
290 engine *gin.Engine
291 addr string
292}
293
294func handleNoRoute(c *gin.Context) {
295 c.JSON(http.StatusNotFound, gin.H{
296 "error": "404 page not found",
297 })
298}
299
300// NewNodeAPI creates a new NodeAPI (which internally calls api.NewAPI)
301func NewNodeAPI(
302 addr string,
303 coordinatorEndpoints, explorerEndpoints bool,
304 server *gin.Engine,
305 hdb *historydb.HistoryDB,
306 sdb *statedb.StateDB,
307 l2db *l2db.L2DB,
308 config *api.Config,
309 chainID uint16,
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 chainID,
322 )
323 if err != nil {
324 return nil, tracerr.Wrap(err)
325 }
326 return &NodeAPI{
327 addr: addr,
328 api: _api,
329 engine: engine,
330 }, nil
331}
332
333// Run starts the http server of the NodeAPI. To stop it, pass a context with
334// cancelation.
335func (a *NodeAPI) Run(ctx context.Context) error {
336 server := &http.Server{
337 Addr: a.addr,
338 Handler: a.engine,
339 // TODO: Figure out best parameters for production
340 ReadTimeout: 30 * time.Second, //nolint:gomnd
341 WriteTimeout: 30 * time.Second, //nolint:gomnd
342 MaxHeaderBytes: 1 << 20, //nolint:gomnd
343 }
344 go func() {
345 log.Infof("NodeAPI is ready at %v", a.addr)
346 if err := server.ListenAndServe(); err != nil && tracerr.Unwrap(err) != http.ErrServerClosed {
347 log.Fatalf("Listen: %s\n", err)
348 }
349 }()
350
351 <-ctx.Done()
352 log.Info("Stopping NodeAPI...")
353 ctxTimeout, cancel := context.WithTimeout(context.Background(), 10*time.Second) //nolint:gomnd
354 defer cancel()
355 if err := server.Shutdown(ctxTimeout); err != nil {
356 return tracerr.Wrap(err)
357 }
358 log.Info("NodeAPI done")
359 return nil
360}
361
362func (n *Node) handleNewBlock(stats *synchronizer.Stats, vars synchronizer.SCVariablesPtr,
363 batches []common.BatchData) {
364 if n.mode == ModeCoordinator {
365 n.coord.SendMsg(coordinator.MsgSyncBlock{
366 Stats: *stats,
367 Batches: batches,
368 Vars: synchronizer.SCVariablesPtr{
369 Rollup: vars.Rollup,
370 Auction: vars.Auction,
371 WDelayer: vars.WDelayer,
372 },
373 })
374 }
375 if n.nodeAPI != nil {
376 if vars.Rollup != nil {
377 n.nodeAPI.api.SetRollupVariables(*vars.Rollup)
378 }
379 if vars.Auction != nil {
380 n.nodeAPI.api.SetAuctionVariables(*vars.Auction)
381 }
382 if vars.WDelayer != nil {
383 n.nodeAPI.api.SetWDelayerVariables(*vars.WDelayer)
384 }
385
386 if stats.Synced() {
387 if err := n.nodeAPI.api.UpdateNetworkInfo(
388 stats.Eth.LastBlock, stats.Sync.LastBlock,
389 common.BatchNum(stats.Eth.LastBatch),
390 stats.Sync.Auction.CurrentSlot.SlotNum,
391 ); err != nil {
392 log.Errorw("API.UpdateNetworkInfo", "err", err)
393 }
394 }
395 }
396}
397
398func (n *Node) handleReorg(stats *synchronizer.Stats) {
399 if n.mode == ModeCoordinator {
400 n.coord.SendMsg(coordinator.MsgSyncReorg{
401 Stats: *stats,
402 })
403 }
404 if n.nodeAPI != nil {
405 vars := n.sync.SCVars()
406 n.nodeAPI.api.SetRollupVariables(*vars.Rollup)
407 n.nodeAPI.api.SetAuctionVariables(*vars.Auction)
408 n.nodeAPI.api.SetWDelayerVariables(*vars.WDelayer)
409 n.nodeAPI.api.UpdateNetworkInfoBlock(
410 stats.Eth.LastBlock, stats.Sync.LastBlock,
411 )
412 }
413}
414
415// TODO(Edu): Consider keeping the `lastBlock` inside synchronizer so that we
416// don't have to pass it around.
417func (n *Node) syncLoopFn(ctx context.Context, lastBlock *common.Block) (*common.Block, time.Duration, error) {
418 blockData, discarded, err := n.sync.Sync2(ctx, lastBlock)
419 stats := n.sync.Stats()
420 if err != nil {
421 // case: error
422 return nil, n.cfg.Synchronizer.SyncLoopInterval.Duration, err
423 } else if discarded != nil {
424 // case: reorg
425 log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
426 n.handleReorg(stats)
427 return nil, time.Duration(0), nil
428 } else if blockData != nil {
429 // case: new block
430 n.handleNewBlock(stats, synchronizer.SCVariablesPtr{
431 Rollup: blockData.Rollup.Vars,
432 Auction: blockData.Auction.Vars,
433 WDelayer: blockData.WDelayer.Vars,
434 }, blockData.Rollup.Batches)
435 return &blockData.Block, time.Duration(0), nil
436 } else {
437 // case: no block
438 return lastBlock, n.cfg.Synchronizer.SyncLoopInterval.Duration, nil
439 }
440}
441
442// StartSynchronizer starts the synchronizer
443func (n *Node) StartSynchronizer() {
444 log.Info("Starting Synchronizer...")
445
446 // Trigger a manual call to handleNewBlock with the loaded state of the
447 // synchronizer in order to quickly activate the API and Coordinator
448 // and avoid waiting for the next block. Without this, the API and
449 // Coordinator will not react until the following block (starting from
450 // the last synced one) is synchronized
451 stats := n.sync.Stats()
452 vars := n.sync.SCVars()
453 n.handleNewBlock(stats, vars, []common.BatchData{})
454
455 n.wg.Add(1)
456 go func() {
457 var err error
458 var lastBlock *common.Block
459 waitDuration := time.Duration(0)
460 for {
461 select {
462 case <-n.ctx.Done():
463 log.Info("Synchronizer done")
464 n.wg.Done()
465 return
466 case <-time.After(waitDuration):
467 if lastBlock, waitDuration, err = n.syncLoopFn(n.ctx,
468 lastBlock); err != nil {
469 if n.ctx.Err() != nil {
470 continue
471 }
472 log.Errorw("Synchronizer.Sync", "err", err)
473 }
474 }
475 }
476 }()
477
478 n.wg.Add(1)
479 go func() {
480 for {
481 select {
482 case <-n.ctx.Done():
483 log.Info("PriceUpdater done")
484 n.wg.Done()
485 return
486 case <-time.After(n.cfg.PriceUpdater.Interval.Duration):
487 if err := n.priceUpdater.UpdateTokenList(); err != nil {
488 log.Errorw("PriceUpdater.UpdateTokenList()", "err", err)
489 }
490 n.priceUpdater.UpdatePrices(n.ctx)
491 }
492 }
493 }()
494}
495
496// StartDebugAPI starts the DebugAPI
497func (n *Node) StartDebugAPI() {
498 log.Info("Starting DebugAPI...")
499 n.wg.Add(1)
500 go func() {
501 defer func() {
502 log.Info("DebugAPI routine stopped")
503 n.wg.Done()
504 }()
505 if err := n.debugAPI.Run(n.ctx); err != nil {
506 if n.ctx.Err() != nil {
507 return
508 }
509 log.Fatalw("DebugAPI.Run", "err", err)
510 }
511 }()
512}
513
514// StartNodeAPI starts the NodeAPI
515func (n *Node) StartNodeAPI() {
516 log.Info("Starting NodeAPI...")
517 n.wg.Add(1)
518 go func() {
519 defer func() {
520 log.Info("NodeAPI routine stopped")
521 n.wg.Done()
522 }()
523 if err := n.nodeAPI.Run(n.ctx); err != nil {
524 if n.ctx.Err() != nil {
525 return
526 }
527 log.Fatalw("NodeAPI.Run", "err", err)
528 }
529 }()
530
531 n.wg.Add(1)
532 go func() {
533 for {
534 select {
535 case <-n.ctx.Done():
536 log.Info("API.UpdateMetrics loop done")
537 n.wg.Done()
538 return
539 case <-time.After(n.cfg.API.UpdateMetricsInterval.Duration):
540 if err := n.nodeAPI.api.UpdateMetrics(); err != nil {
541 log.Errorw("API.UpdateMetrics", "err", err)
542 }
543 }
544 }
545 }()
546
547 n.wg.Add(1)
548 go func() {
549 for {
550 select {
551 case <-n.ctx.Done():
552 log.Info("API.UpdateRecommendedFee loop done")
553 n.wg.Done()
554 return
555 case <-time.After(n.cfg.API.UpdateRecommendedFeeInterval.Duration):
556 if err := n.nodeAPI.api.UpdateRecommendedFee(); err != nil {
557 log.Errorw("API.UpdateRecommendedFee", "err", err)
558 }
559 }
560 }
561 }()
562}
563
564// Start the node
565func (n *Node) Start() {
566 log.Infow("Starting node...", "mode", n.mode)
567 if n.debugAPI != nil {
568 n.StartDebugAPI()
569 }
570 if n.nodeAPI != nil {
571 n.StartNodeAPI()
572 }
573 if n.mode == ModeCoordinator {
574 log.Info("Starting Coordinator...")
575 n.coord.Start()
576 }
577 n.StartSynchronizer()
578}
579
580// Stop the node
581func (n *Node) Stop() {
582 log.Infow("Stopping node...")
583 n.cancel()
584 if n.mode == ModeCoordinator {
585 log.Info("Stopping Coordinator...")
586 n.coord.Stop()
587 }
588 n.wg.Wait()
589}