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 "github.com/ethereum/go-ethereum/ethclient"
11 "github.com/gin-contrib/cors"
12 "github.com/gin-gonic/gin"
13 "github.com/hermeznetwork/hermez-node/api"
14 "github.com/hermeznetwork/hermez-node/batchbuilder"
15 "github.com/hermeznetwork/hermez-node/common"
16 "github.com/hermeznetwork/hermez-node/config"
17 "github.com/hermeznetwork/hermez-node/coordinator"
18 dbUtils "github.com/hermeznetwork/hermez-node/db"
19 "github.com/hermeznetwork/hermez-node/db/historydb"
20 "github.com/hermeznetwork/hermez-node/db/l2db"
21 "github.com/hermeznetwork/hermez-node/db/statedb"
22 "github.com/hermeznetwork/hermez-node/eth"
23 "github.com/hermeznetwork/hermez-node/log"
24 "github.com/hermeznetwork/hermez-node/prover"
25 "github.com/hermeznetwork/hermez-node/synchronizer"
26 "github.com/hermeznetwork/hermez-node/test/debugapi"
27 "github.com/hermeznetwork/hermez-node/txselector"
28 "github.com/hermeznetwork/tracerr"
29 "github.com/jmoiron/sqlx"
30)
31
32// Mode sets the working mode of the node (synchronizer or coordinator)
33type Mode string
34
35const (
36 // ModeCoordinator defines the mode of the HermezNode as Coordinator, which
37 // means that the node is set to forge (which also will be synchronizing with
38 // the L1 blockchain state)
39 ModeCoordinator Mode = "coordinator"
40
41 // ModeSynchronizer defines the mode of the HermezNode as Synchronizer, which
42 // means that the node is set to only synchronize with the L1 blockchain state
43 // and will not forge
44 ModeSynchronizer Mode = "synchronizer"
45)
46
47// Node is the Hermez Node
48type Node struct {
49 nodeAPI *NodeAPI
50 debugAPI *debugapi.DebugAPI
51 // Coordinator
52 coord *coordinator.Coordinator
53 coordCfg *config.Coordinator
54
55 // Synchronizer
56 sync *synchronizer.Synchronizer
57
58 // General
59 cfg *config.Node
60 mode Mode
61 sqlConn *sqlx.DB
62 ctx context.Context
63 wg sync.WaitGroup
64 cancel context.CancelFunc
65}
66
67// NewNode creates a Node
68func NewNode(mode Mode, cfg *config.Node, coordCfg *config.Coordinator) (*Node, error) {
69 // Stablish DB connection
70 db, err := dbUtils.InitSQLDB(
71 cfg.PostgreSQL.Port,
72 cfg.PostgreSQL.Host,
73 cfg.PostgreSQL.User,
74 cfg.PostgreSQL.Password,
75 cfg.PostgreSQL.Name,
76 )
77 if err != nil {
78 return nil, tracerr.Wrap(err)
79 }
80
81 historyDB := historydb.NewHistoryDB(db)
82
83 stateDB, err := statedb.NewStateDB(cfg.StateDB.Path, statedb.TypeSynchronizer, 32)
84 if err != nil {
85 return nil, tracerr.Wrap(err)
86 }
87
88 ethClient, err := ethclient.Dial(cfg.Web3.URL)
89 if err != nil {
90 return nil, tracerr.Wrap(err)
91 }
92 var ethCfg eth.EthereumConfig
93 if mode == ModeCoordinator {
94 ethCfg = eth.EthereumConfig{
95 CallGasLimit: coordCfg.EthClient.CallGasLimit,
96 DeployGasLimit: coordCfg.EthClient.DeployGasLimit,
97 GasPriceDiv: coordCfg.EthClient.GasPriceDiv,
98 ReceiptTimeout: coordCfg.EthClient.ReceiptTimeout.Duration,
99 IntervalReceiptLoop: coordCfg.EthClient.ReceiptLoopInterval.Duration,
100 }
101 }
102 client, err := eth.NewClient(ethClient, nil, nil, ð.ClientConfig{
103 Ethereum: ethCfg,
104 Rollup: eth.RollupConfig{
105 Address: cfg.SmartContracts.Rollup,
106 },
107 Auction: eth.AuctionConfig{
108 Address: cfg.SmartContracts.Auction,
109 TokenHEZ: eth.TokenConfig{
110 Address: cfg.SmartContracts.TokenHEZ,
111 Name: cfg.SmartContracts.TokenHEZName,
112 },
113 },
114 WDelayer: eth.WDelayerConfig{
115 Address: cfg.SmartContracts.WDelayer,
116 },
117 })
118 if err != nil {
119 return nil, tracerr.Wrap(err)
120 }
121
122 sync, err := synchronizer.NewSynchronizer(client, historyDB, stateDB, synchronizer.Config{
123 StartBlockNum: cfg.Synchronizer.StartBlockNum,
124 InitialVariables: cfg.Synchronizer.InitialVariables,
125 StatsRefreshPeriod: cfg.Synchronizer.StatsRefreshPeriod.Duration,
126 })
127 if err != nil {
128 return nil, tracerr.Wrap(err)
129 }
130 varsRollup, varsAuction, varsWDelayer := sync.SCVars()
131 initSCVars := synchronizer.SCVariables{
132 Rollup: *varsRollup,
133 Auction: *varsAuction,
134 WDelayer: *varsWDelayer,
135 }
136
137 scConsts := synchronizer.SCConsts{
138 Rollup: *sync.RollupConstants(),
139 Auction: *sync.AuctionConstants(),
140 WDelayer: *sync.WDelayerConstants(),
141 }
142
143 var coord *coordinator.Coordinator
144 var l2DB *l2db.L2DB
145 if mode == ModeCoordinator {
146 l2DB = l2db.NewL2DB(
147 db,
148 coordCfg.L2DB.SafetyPeriod,
149 coordCfg.L2DB.MaxTxs,
150 coordCfg.L2DB.TTL.Duration,
151 )
152 // TODO: Get (maxL1UserTxs, maxL1OperatorTxs, maxTxs) from the smart contract
153 txSelector, err := txselector.NewTxSelector(coordCfg.TxSelector.Path, stateDB, l2DB, 10, 10, 10)
154 if err != nil {
155 return nil, tracerr.Wrap(err)
156 }
157 // TODO: Get (configCircuits []ConfigCircuit, batchNum common.BatchNum, nLevels uint64) from smart contract
158 nLevels := uint64(32) //nolint:gomnd
159 batchBuilder, err := batchbuilder.NewBatchBuilder(coordCfg.BatchBuilder.Path, stateDB, nil, 0, nLevels)
160 if err != nil {
161 return nil, tracerr.Wrap(err)
162 }
163 if err != nil {
164 return nil, tracerr.Wrap(err)
165 }
166 serverProofs := make([]prover.Client, len(coordCfg.ServerProofs))
167 for i, serverProofCfg := range coordCfg.ServerProofs {
168 serverProofs[i] = prover.NewProofServerClient(serverProofCfg.URL,
169 coordCfg.ProofServerPollInterval.Duration)
170 }
171
172 coord, err = coordinator.NewCoordinator(
173 coordinator.Config{
174 ForgerAddress: coordCfg.ForgerAddress,
175 ConfirmBlocks: coordCfg.ConfirmBlocks,
176 L1BatchTimeoutPerc: coordCfg.L1BatchTimeoutPerc,
177 EthClientAttempts: coordCfg.EthClient.Attempts,
178 EthClientAttemptsDelay: coordCfg.EthClient.AttemptsDelay.Duration,
179 TxManagerCheckInterval: coordCfg.EthClient.CheckLoopInterval.Duration,
180 DebugBatchPath: coordCfg.Debug.BatchPath,
181 Purger: coordinator.PurgerCfg{
182 PurgeBatchDelay: coordCfg.L2DB.PurgeBatchDelay,
183 InvalidateBatchDelay: coordCfg.L2DB.InvalidateBatchDelay,
184 PurgeBlockDelay: coordCfg.L2DB.PurgeBlockDelay,
185 InvalidateBlockDelay: coordCfg.L2DB.InvalidateBlockDelay,
186 },
187 },
188 historyDB,
189 l2DB,
190 txSelector,
191 batchBuilder,
192 serverProofs,
193 client,
194 &scConsts,
195 &initSCVars,
196 )
197 if err != nil {
198 return nil, tracerr.Wrap(err)
199 }
200 }
201 var nodeAPI *NodeAPI
202 if cfg.API.Address != "" {
203 if cfg.API.UpdateMetricsInterval.Duration == 0 {
204 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateMetricsInterval: %v",
205 cfg.API.UpdateMetricsInterval.Duration))
206 }
207 if cfg.API.UpdateRecommendedFeeInterval.Duration == 0 {
208 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateRecommendedFeeInterval: %v",
209 cfg.API.UpdateRecommendedFeeInterval.Duration))
210 }
211 server := gin.Default()
212 coord := false
213 if mode == ModeCoordinator {
214 coord = coordCfg.API.Coordinator
215 }
216 var err error
217 nodeAPI, err = NewNodeAPI(
218 cfg.API.Address,
219 coord, cfg.API.Explorer,
220 server,
221 historyDB,
222 stateDB,
223 l2DB,
224 &api.Config{
225 RollupConstants: scConsts.Rollup,
226 AuctionConstants: scConsts.Auction,
227 WDelayerConstants: scConsts.WDelayer,
228 },
229 )
230 if err != nil {
231 return nil, tracerr.Wrap(err)
232 }
233 nodeAPI.api.SetRollupVariables(initSCVars.Rollup)
234 nodeAPI.api.SetAuctionVariables(initSCVars.Auction)
235 nodeAPI.api.SetWDelayerVariables(initSCVars.WDelayer)
236 }
237 var debugAPI *debugapi.DebugAPI
238 if cfg.Debug.APIAddress != "" {
239 debugAPI = debugapi.NewDebugAPI(cfg.Debug.APIAddress, stateDB, sync)
240 }
241 ctx, cancel := context.WithCancel(context.Background())
242 return &Node{
243 nodeAPI: nodeAPI,
244 debugAPI: debugAPI,
245 coord: coord,
246 coordCfg: coordCfg,
247 sync: sync,
248 cfg: cfg,
249 mode: mode,
250 sqlConn: db,
251 ctx: ctx,
252 cancel: cancel,
253 }, nil
254}
255
256// NodeAPI holds the node http API
257type NodeAPI struct { //nolint:golint
258 api *api.API
259 engine *gin.Engine
260 addr string
261}
262
263func handleNoRoute(c *gin.Context) {
264 c.JSON(http.StatusNotFound, gin.H{
265 "error": "404 page not found",
266 })
267}
268
269// NewNodeAPI creates a new NodeAPI (which internally calls api.NewAPI)
270func NewNodeAPI(
271 addr string,
272 coordinatorEndpoints, explorerEndpoints bool,
273 server *gin.Engine,
274 hdb *historydb.HistoryDB,
275 sdb *statedb.StateDB,
276 l2db *l2db.L2DB,
277 config *api.Config,
278) (*NodeAPI, error) {
279 engine := gin.Default()
280 engine.NoRoute(handleNoRoute)
281 engine.Use(cors.Default())
282 _api, err := api.NewAPI(
283 coordinatorEndpoints, explorerEndpoints,
284 engine,
285 hdb,
286 sdb,
287 l2db,
288 config,
289 )
290 if err != nil {
291 return nil, tracerr.Wrap(err)
292 }
293 return &NodeAPI{
294 addr: addr,
295 api: _api,
296 engine: engine,
297 }, nil
298}
299
300// Run starts the http server of the NodeAPI. To stop it, pass a context with
301// cancelation.
302func (a *NodeAPI) Run(ctx context.Context) error {
303 server := &http.Server{
304 Addr: a.addr,
305 Handler: a.engine,
306 // TODO: Figure out best parameters for production
307 ReadTimeout: 30 * time.Second, //nolint:gomnd
308 WriteTimeout: 30 * time.Second, //nolint:gomnd
309 MaxHeaderBytes: 1 << 20, //nolint:gomnd
310 }
311 go func() {
312 log.Infof("NodeAPI is ready at %v", a.addr)
313 if err := server.ListenAndServe(); err != nil && tracerr.Unwrap(err) != http.ErrServerClosed {
314 log.Fatalf("Listen: %s\n", err)
315 }
316 }()
317
318 <-ctx.Done()
319 log.Info("Stopping NodeAPI...")
320 ctxTimeout, cancel := context.WithTimeout(context.Background(), 10*time.Second) //nolint:gomnd
321 defer cancel()
322 if err := server.Shutdown(ctxTimeout); err != nil {
323 return tracerr.Wrap(err)
324 }
325 log.Info("NodeAPI done")
326 return nil
327}
328
329// TODO(Edu): Consider keeping the `lastBlock` inside synchronizer so that we
330// don't have to pass it around.
331func (n *Node) syncLoopFn(lastBlock *common.Block) (*common.Block, time.Duration) {
332 blockData, discarded, err := n.sync.Sync2(n.ctx, lastBlock)
333 stats := n.sync.Stats()
334 if err != nil {
335 // case: error
336 log.Errorw("Synchronizer.Sync", "err", err)
337 return nil, n.cfg.Synchronizer.SyncLoopInterval.Duration
338 } else if discarded != nil {
339 // case: reorg
340 log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
341 if n.mode == ModeCoordinator {
342 n.coord.SendMsg(coordinator.MsgSyncReorg{
343 Stats: *stats,
344 })
345 }
346 if n.nodeAPI != nil {
347 rollup, auction, wDelayer := n.sync.SCVars()
348 n.nodeAPI.api.SetRollupVariables(*rollup)
349 n.nodeAPI.api.SetAuctionVariables(*auction)
350 n.nodeAPI.api.SetWDelayerVariables(*wDelayer)
351 n.nodeAPI.api.UpdateNetworkInfoBlock(
352 stats.Eth.LastBlock, stats.Sync.LastBlock,
353 )
354 }
355 return nil, time.Duration(0)
356 } else if blockData != nil {
357 // case: new block
358 if n.mode == ModeCoordinator {
359 if stats.Synced() && (blockData.Rollup.Vars != nil ||
360 blockData.Auction.Vars != nil ||
361 blockData.WDelayer.Vars != nil) {
362 n.coord.SendMsg(coordinator.MsgSyncSCVars{
363 Rollup: blockData.Rollup.Vars,
364 Auction: blockData.Auction.Vars,
365 WDelayer: blockData.WDelayer.Vars,
366 })
367 }
368 n.coord.SendMsg(coordinator.MsgSyncBlock{
369 Stats: *stats,
370 Batches: blockData.Rollup.Batches,
371 })
372 }
373 if n.nodeAPI != nil {
374 if blockData.Rollup.Vars != nil {
375 n.nodeAPI.api.SetRollupVariables(*blockData.Rollup.Vars)
376 }
377 if blockData.Auction.Vars != nil {
378 n.nodeAPI.api.SetAuctionVariables(*blockData.Auction.Vars)
379 }
380 if blockData.WDelayer.Vars != nil {
381 n.nodeAPI.api.SetWDelayerVariables(*blockData.WDelayer.Vars)
382 }
383
384 if stats.Synced() {
385 if err := n.nodeAPI.api.UpdateNetworkInfo(
386 stats.Eth.LastBlock, stats.Sync.LastBlock,
387 common.BatchNum(stats.Eth.LastBatch),
388 stats.Sync.Auction.CurrentSlot.SlotNum,
389 ); err != nil {
390 log.Errorw("API.UpdateNetworkInfo", "err", err)
391 }
392 }
393 }
394 return &blockData.Block, time.Duration(0)
395 } else {
396 // case: no block
397 return lastBlock, n.cfg.Synchronizer.SyncLoopInterval.Duration
398 }
399}
400
401// StartSynchronizer starts the synchronizer
402func (n *Node) StartSynchronizer() {
403 log.Info("Starting Synchronizer...")
404 n.wg.Add(1)
405 go func() {
406 var lastBlock *common.Block
407 waitDuration := time.Duration(0)
408 for {
409 select {
410 case <-n.ctx.Done():
411 log.Info("Synchronizer done")
412 n.wg.Done()
413 return
414 case <-time.After(waitDuration):
415 lastBlock, waitDuration = n.syncLoopFn(lastBlock)
416 }
417 }
418 }()
419 // TODO: Run price updater. This is required by the API and the TxSelector
420}
421
422// StartDebugAPI starts the DebugAPI
423func (n *Node) StartDebugAPI() {
424 log.Info("Starting DebugAPI...")
425 n.wg.Add(1)
426 go func() {
427 defer func() {
428 log.Info("DebugAPI routine stopped")
429 n.wg.Done()
430 }()
431 if err := n.debugAPI.Run(n.ctx); err != nil {
432 log.Fatalw("DebugAPI.Run", "err", err)
433 }
434 }()
435}
436
437// StartNodeAPI starts the NodeAPI
438func (n *Node) StartNodeAPI() {
439 log.Info("Starting NodeAPI...")
440 n.wg.Add(1)
441 go func() {
442 defer func() {
443 log.Info("NodeAPI routine stopped")
444 n.wg.Done()
445 }()
446 if err := n.nodeAPI.Run(n.ctx); err != nil {
447 log.Fatalw("NodeAPI.Run", "err", err)
448 }
449 }()
450
451 n.wg.Add(1)
452 go func() {
453 for {
454 select {
455 case <-n.ctx.Done():
456 log.Info("API.UpdateMetrics loop done")
457 n.wg.Done()
458 return
459 case <-time.After(n.cfg.API.UpdateMetricsInterval.Duration):
460 if err := n.nodeAPI.api.UpdateMetrics(); err != nil {
461 log.Errorw("API.UpdateMetrics", "err", err)
462 }
463 }
464 }
465 }()
466
467 n.wg.Add(1)
468 go func() {
469 for {
470 select {
471 case <-n.ctx.Done():
472 log.Info("API.UpdateRecommendedFee loop done")
473 n.wg.Done()
474 return
475 case <-time.After(n.cfg.API.UpdateRecommendedFeeInterval.Duration):
476 if err := n.nodeAPI.api.UpdateRecommendedFee(); err != nil {
477 log.Errorw("API.UpdateRecommendedFee", "err", err)
478 }
479 }
480 }
481 }()
482}
483
484// Start the node
485func (n *Node) Start() {
486 log.Infow("Starting node...", "mode", n.mode)
487 if n.debugAPI != nil {
488 n.StartDebugAPI()
489 }
490 if n.nodeAPI != nil {
491 n.StartNodeAPI()
492 }
493 if n.mode == ModeCoordinator {
494 log.Info("Starting Coordinator...")
495 n.coord.Start()
496 }
497 n.StartSynchronizer()
498}
499
500// Stop the node
501func (n *Node) Stop() {
502 log.Infow("Stopping node...")
503 n.cancel()
504 if n.mode == ModeCoordinator {
505 log.Info("Stopping Coordinator...")
506 n.coord.Stop()
507 }
508 n.wg.Wait()
509}