hermez-node/node/node.go
node/node.goRaw
1package node
2
3import (
4 "context"
5 "fmt"
6 "net/http"
7 "strings"
8 "sync"
9 "time"
10
11 ethCommon "github.com/ethereum/go-ethereum/common"
12 "github.com/ethereum/go-ethereum/ethclient"
13 "github.com/gin-contrib/cors"
14 "github.com/gin-gonic/gin"
15 "github.com/hermeznetwork/hermez-node/api"
16 "github.com/hermeznetwork/hermez-node/batchbuilder"
17 "github.com/hermeznetwork/hermez-node/common"
18 "github.com/hermeznetwork/hermez-node/config"
19 "github.com/hermeznetwork/hermez-node/coordinator"
20 dbUtils "github.com/hermeznetwork/hermez-node/db"
21 "github.com/hermeznetwork/hermez-node/db/historydb"
22 "github.com/hermeznetwork/hermez-node/db/l2db"
23 "github.com/hermeznetwork/hermez-node/db/statedb"
24 "github.com/hermeznetwork/hermez-node/eth"
25 "github.com/hermeznetwork/hermez-node/log"
26 "github.com/hermeznetwork/hermez-node/priceupdater"
27 "github.com/hermeznetwork/hermez-node/prover"
28 "github.com/hermeznetwork/hermez-node/synchronizer"
29 "github.com/hermeznetwork/hermez-node/test/debugapi"
30 "github.com/hermeznetwork/hermez-node/txselector"
31 "github.com/hermeznetwork/tracerr"
32 "github.com/jmoiron/sqlx"
33)
34
35// Mode sets the working mode of the node (synchronizer or coordinator)
36type Mode string
37
38const (
39 // ModeCoordinator defines the mode of the HermezNode as Coordinator, which
40 // means that the node is set to forge (which also will be synchronizing with
41 // the L1 blockchain state)
42 ModeCoordinator Mode = "coordinator"
43
44 // ModeSynchronizer defines the mode of the HermezNode as Synchronizer, which
45 // means that the node is set to only synchronize with the L1 blockchain state
46 // and will not forge
47 ModeSynchronizer Mode = "synchronizer"
48)
49
50// Node is the Hermez Node
51type Node struct {
52 nodeAPI *NodeAPI
53 debugAPI *debugapi.DebugAPI
54 priceUpdater *priceupdater.PriceUpdater
55 // Coordinator
56 coord *coordinator.Coordinator
57
58 // Synchronizer
59 sync *synchronizer.Synchronizer
60
61 // General
62 cfg *config.Node
63 mode Mode
64 sqlConn *sqlx.DB
65 ctx context.Context
66 wg sync.WaitGroup
67 cancel context.CancelFunc
68}
69
70// NewNode creates a Node
71func NewNode(mode Mode, cfg *config.Node) (*Node, error) {
72 // Stablish DB connection
73 db, err := dbUtils.InitSQLDB(
74 cfg.PostgreSQL.Port,
75 cfg.PostgreSQL.Host,
76 cfg.PostgreSQL.User,
77 cfg.PostgreSQL.Password,
78 cfg.PostgreSQL.Name,
79 )
80 if err != nil {
81 return nil, tracerr.Wrap(err)
82 }
83
84 historyDB := historydb.NewHistoryDB(db)
85
86 ethClient, err := ethclient.Dial(cfg.Web3.URL)
87 if err != nil {
88 return nil, tracerr.Wrap(err)
89 }
90 var ethCfg eth.EthereumConfig
91 if mode == ModeCoordinator {
92 ethCfg = eth.EthereumConfig{
93 CallGasLimit: cfg.Coordinator.EthClient.CallGasLimit,
94 DeployGasLimit: cfg.Coordinator.EthClient.DeployGasLimit,
95 GasPriceDiv: cfg.Coordinator.EthClient.GasPriceDiv,
96 ReceiptTimeout: cfg.Coordinator.EthClient.ReceiptTimeout.Duration,
97 IntervalReceiptLoop: cfg.Coordinator.EthClient.ReceiptLoopInterval.Duration,
98 }
99 }
100 client, err := eth.NewClient(ethClient, nil, nil, ð.ClientConfig{
101 Ethereum: ethCfg,
102 Rollup: eth.RollupConfig{
103 Address: cfg.SmartContracts.Rollup,
104 },
105 Auction: eth.AuctionConfig{
106 Address: cfg.SmartContracts.Auction,
107 TokenHEZ: eth.TokenConfig{
108 Address: cfg.SmartContracts.TokenHEZ,
109 Name: cfg.SmartContracts.TokenHEZName,
110 },
111 },
112 WDelayer: eth.WDelayerConfig{
113 Address: cfg.SmartContracts.WDelayer,
114 },
115 })
116 if err != nil {
117 return nil, tracerr.Wrap(err)
118 }
119
120 chainID, err := client.EthChainID()
121 if err != nil {
122 return nil, tracerr.Wrap(err)
123 }
124 if !chainID.IsUint64() {
125 return nil, tracerr.Wrap(fmt.Errorf("chainID cannot be represented as uint64"))
126 }
127 chainIDU64 := chainID.Uint64()
128 const maxUint16 uint64 = 0xffff
129 if chainIDU64 > maxUint16 {
130 return nil, tracerr.Wrap(fmt.Errorf("chainID overflows uint16"))
131 }
132 chainIDU16 := uint16(chainIDU64)
133
134 stateDB, err := statedb.NewStateDB(cfg.StateDB.Path, statedb.TypeSynchronizer, 32, chainIDU16)
135 if err != nil {
136 return nil, tracerr.Wrap(err)
137 }
138
139 sync, err := synchronizer.NewSynchronizer(client, historyDB, stateDB, synchronizer.Config{
140 StatsRefreshPeriod: cfg.Synchronizer.StatsRefreshPeriod.Duration,
141 })
142 if err != nil {
143 return nil, tracerr.Wrap(err)
144 }
145 initSCVars := sync.SCVars()
146
147 scConsts := synchronizer.SCConsts{
148 Rollup: *sync.RollupConstants(),
149 Auction: *sync.AuctionConstants(),
150 WDelayer: *sync.WDelayerConstants(),
151 }
152
153 var coord *coordinator.Coordinator
154 var l2DB *l2db.L2DB
155 if mode == ModeCoordinator {
156 l2DB = l2db.NewL2DB(
157 db,
158 cfg.Coordinator.L2DB.SafetyPeriod,
159 cfg.Coordinator.L2DB.MaxTxs,
160 cfg.Coordinator.L2DB.TTL.Duration,
161 )
162 // TODO: Get (maxL1UserTxs, maxL1OperatorTxs, maxTxs) from the smart contract
163 coordAccount := &txselector.CoordAccount{ // TODO TMP
164 Addr: ethCommon.HexToAddress("0xc58d29fA6e86E4FAe04DDcEd660d45BCf3Cb2370"),
165 BJJ: common.EmptyBJJComp,
166 AccountCreationAuth: nil,
167 }
168 txSelector, err := txselector.NewTxSelector(coordAccount, cfg.Coordinator.TxSelector.Path, stateDB, l2DB)
169 if err != nil {
170 return nil, tracerr.Wrap(err)
171 }
172 // TODO: Get (configCircuits []ConfigCircuit, batchNum common.BatchNum, nLevels uint64) from smart contract
173 nLevels := uint64(32) //nolint:gomnd
174 batchBuilder, err := batchbuilder.NewBatchBuilder(cfg.Coordinator.BatchBuilder.Path, stateDB, nil, 0, nLevels)
175 if err != nil {
176 return nil, tracerr.Wrap(err)
177 }
178 if err != nil {
179 return nil, tracerr.Wrap(err)
180 }
181 serverProofs := make([]prover.Client, len(cfg.Coordinator.ServerProofs))
182 for i, serverProofCfg := range cfg.Coordinator.ServerProofs {
183 serverProofs[i] = prover.NewProofServerClient(serverProofCfg.URL,
184 cfg.Coordinator.ProofServerPollInterval.Duration)
185 }
186
187 coord, err = coordinator.NewCoordinator(
188 coordinator.Config{
189 ForgerAddress: cfg.Coordinator.ForgerAddress,
190 ConfirmBlocks: cfg.Coordinator.ConfirmBlocks,
191 L1BatchTimeoutPerc: cfg.Coordinator.L1BatchTimeoutPerc,
192 SyncRetryInterval: cfg.Coordinator.SyncRetryInterval.Duration,
193 EthClientAttempts: cfg.Coordinator.EthClient.Attempts,
194 EthClientAttemptsDelay: cfg.Coordinator.EthClient.AttemptsDelay.Duration,
195 TxManagerCheckInterval: cfg.Coordinator.EthClient.CheckLoopInterval.Duration,
196 DebugBatchPath: cfg.Coordinator.Debug.BatchPath,
197 Purger: coordinator.PurgerCfg{
198 PurgeBatchDelay: cfg.Coordinator.L2DB.PurgeBatchDelay,
199 InvalidateBatchDelay: cfg.Coordinator.L2DB.InvalidateBatchDelay,
200 PurgeBlockDelay: cfg.Coordinator.L2DB.PurgeBlockDelay,
201 InvalidateBlockDelay: cfg.Coordinator.L2DB.InvalidateBlockDelay,
202 },
203 },
204 historyDB,
205 l2DB,
206 txSelector,
207 batchBuilder,
208 serverProofs,
209 client,
210 &scConsts,
211 &synchronizer.SCVariables{
212 Rollup: *initSCVars.Rollup,
213 Auction: *initSCVars.Auction,
214 WDelayer: *initSCVars.WDelayer,
215 },
216 )
217 if err != nil {
218 return nil, tracerr.Wrap(err)
219 }
220 }
221 var nodeAPI *NodeAPI
222 if cfg.API.Address != "" {
223 if cfg.API.UpdateMetricsInterval.Duration == 0 {
224 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateMetricsInterval: %v",
225 cfg.API.UpdateMetricsInterval.Duration))
226 }
227 if cfg.API.UpdateRecommendedFeeInterval.Duration == 0 {
228 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateRecommendedFeeInterval: %v",
229 cfg.API.UpdateRecommendedFeeInterval.Duration))
230 }
231 server := gin.Default()
232 coord := false
233 if mode == ModeCoordinator {
234 coord = cfg.Coordinator.API.Coordinator
235 }
236 var err error
237 nodeAPI, err = NewNodeAPI(
238 cfg.API.Address,
239 coord, cfg.API.Explorer,
240 server,
241 historyDB,
242 stateDB,
243 l2DB,
244 &api.Config{
245 RollupConstants: scConsts.Rollup,
246 AuctionConstants: scConsts.Auction,
247 WDelayerConstants: scConsts.WDelayer,
248 ChainID: chainIDU16,
249 HermezAddress: cfg.SmartContracts.Rollup,
250 },
251 )
252 if err != nil {
253 return nil, tracerr.Wrap(err)
254 }
255 nodeAPI.api.SetRollupVariables(*initSCVars.Rollup)
256 nodeAPI.api.SetAuctionVariables(*initSCVars.Auction)
257 nodeAPI.api.SetWDelayerVariables(*initSCVars.WDelayer)
258 }
259 var debugAPI *debugapi.DebugAPI
260 if cfg.Debug.APIAddress != "" {
261 debugAPI = debugapi.NewDebugAPI(cfg.Debug.APIAddress, stateDB, sync)
262 }
263 priceUpdater, err := priceupdater.NewPriceUpdater(cfg.PriceUpdater.URL,
264 priceupdater.APIType(cfg.PriceUpdater.Type), historyDB)
265 if err != nil {
266 return nil, tracerr.Wrap(err)
267 }
268 ctx, cancel := context.WithCancel(context.Background())
269 return &Node{
270 nodeAPI: nodeAPI,
271 debugAPI: debugAPI,
272 priceUpdater: priceUpdater,
273 coord: coord,
274 sync: sync,
275 cfg: cfg,
276 mode: mode,
277 sqlConn: db,
278 ctx: ctx,
279 cancel: cancel,
280 }, nil
281}
282
283// NodeAPI holds the node http API
284type NodeAPI struct { //nolint:golint
285 api *api.API
286 engine *gin.Engine
287 addr string
288}
289
290func handleNoRoute(c *gin.Context) {
291 c.JSON(http.StatusNotFound, gin.H{
292 "error": "404 page not found",
293 })
294}
295
296// NewNodeAPI creates a new NodeAPI (which internally calls api.NewAPI)
297func NewNodeAPI(
298 addr string,
299 coordinatorEndpoints, explorerEndpoints bool,
300 server *gin.Engine,
301 hdb *historydb.HistoryDB,
302 sdb *statedb.StateDB,
303 l2db *l2db.L2DB,
304 config *api.Config,
305) (*NodeAPI, error) {
306 engine := gin.Default()
307 engine.NoRoute(handleNoRoute)
308 engine.Use(cors.Default())
309 _api, err := api.NewAPI(
310 coordinatorEndpoints, explorerEndpoints,
311 engine,
312 hdb,
313 sdb,
314 l2db,
315 config,
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(lastBlock *common.Block) (*common.Block, time.Duration) {
412 blockData, discarded, err := n.sync.Sync2(n.ctx, lastBlock)
413 stats := n.sync.Stats()
414 if err != nil {
415 // case: error
416 if strings.Contains(err.Error(), "context canceled") {
417 log.Warnw("Synchronizer.Sync", "err", err)
418 } else {
419 log.Errorw("Synchronizer.Sync", "err", err)
420 }
421 return nil, n.cfg.Synchronizer.SyncLoopInterval.Duration
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)
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)
435 } else {
436 // case: no block
437 return lastBlock, n.cfg.Synchronizer.SyncLoopInterval.Duration
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 lastBlock *common.Block
457 waitDuration := time.Duration(0)
458 for {
459 select {
460 case <-n.ctx.Done():
461 log.Info("Synchronizer done")
462 n.wg.Done()
463 return
464 case <-time.After(waitDuration):
465 lastBlock, waitDuration = n.syncLoopFn(lastBlock)
466 }
467 }
468 }()
469
470 n.wg.Add(1)
471 go func() {
472 for {
473 select {
474 case <-n.ctx.Done():
475 log.Info("PriceUpdater done")
476 n.wg.Done()
477 return
478 case <-time.After(n.cfg.PriceUpdater.Interval.Duration):
479 if err := n.priceUpdater.UpdateTokenList(); err != nil {
480 log.Errorw("PriceUpdater.UpdateTokenList()", "err", err)
481 }
482 n.priceUpdater.UpdatePrices(n.ctx)
483 }
484 }
485 }()
486}
487
488// StartDebugAPI starts the DebugAPI
489func (n *Node) StartDebugAPI() {
490 log.Info("Starting DebugAPI...")
491 n.wg.Add(1)
492 go func() {
493 defer func() {
494 log.Info("DebugAPI routine stopped")
495 n.wg.Done()
496 }()
497 if err := n.debugAPI.Run(n.ctx); err != nil {
498 log.Fatalw("DebugAPI.Run", "err", err)
499 }
500 }()
501}
502
503// StartNodeAPI starts the NodeAPI
504func (n *Node) StartNodeAPI() {
505 log.Info("Starting NodeAPI...")
506 n.wg.Add(1)
507 go func() {
508 defer func() {
509 log.Info("NodeAPI routine stopped")
510 n.wg.Done()
511 }()
512 if err := n.nodeAPI.Run(n.ctx); err != nil {
513 log.Fatalw("NodeAPI.Run", "err", err)
514 }
515 }()
516
517 n.wg.Add(1)
518 go func() {
519 for {
520 select {
521 case <-n.ctx.Done():
522 log.Info("API.UpdateMetrics loop done")
523 n.wg.Done()
524 return
525 case <-time.After(n.cfg.API.UpdateMetricsInterval.Duration):
526 if err := n.nodeAPI.api.UpdateMetrics(); err != nil {
527 log.Errorw("API.UpdateMetrics", "err", err)
528 }
529 }
530 }
531 }()
532
533 n.wg.Add(1)
534 go func() {
535 for {
536 select {
537 case <-n.ctx.Done():
538 log.Info("API.UpdateRecommendedFee loop done")
539 n.wg.Done()
540 return
541 case <-time.After(n.cfg.API.UpdateRecommendedFeeInterval.Duration):
542 if err := n.nodeAPI.api.UpdateRecommendedFee(); err != nil {
543 log.Errorw("API.UpdateRecommendedFee", "err", err)
544 }
545 }
546 }
547 }()
548}
549
550// Start the node
551func (n *Node) Start() {
552 log.Infow("Starting node...", "mode", n.mode)
553 if n.debugAPI != nil {
554 n.StartDebugAPI()
555 }
556 if n.nodeAPI != nil {
557 n.StartNodeAPI()
558 }
559 if n.mode == ModeCoordinator {
560 log.Info("Starting Coordinator...")
561 n.coord.Start()
562 }
563 n.StartSynchronizer()
564}
565
566// Stop the node
567func (n *Node) Stop() {
568 log.Infow("Stopping node...")
569 n.cancel()
570 if n.mode == ModeCoordinator {
571 log.Info("Stopping Coordinator...")
572 n.coord.Stop()
573 }
574 n.wg.Wait()
575}