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