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