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.IntervalReceiptLoop.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 }
170
171 coord, err = coordinator.NewCoordinator(
172 coordinator.Config{
173 ForgerAddress: coordCfg.ForgerAddress,
174 ConfirmBlocks: coordCfg.ConfirmBlocks,
175 L1BatchTimeoutPerc: coordCfg.L1BatchTimeoutPerc,
176 EthClientAttempts: coordCfg.EthClient.Attempts,
177 EthClientAttemptsDelay: coordCfg.EthClient.AttemptsDelay.Duration,
178 TxManagerCheckInterval: coordCfg.EthClient.IntervalCheckLoop.Duration,
179 DebugBatchPath: coordCfg.Debug.BatchPath,
180 Purger: coordinator.PurgerCfg{
181 PurgeBatchDelay: coordCfg.L2DB.PurgeBatchDelay,
182 InvalidateBatchDelay: coordCfg.L2DB.InvalidateBatchDelay,
183 PurgeBlockDelay: coordCfg.L2DB.PurgeBlockDelay,
184 InvalidateBlockDelay: coordCfg.L2DB.InvalidateBlockDelay,
185 },
186 },
187 historyDB,
188 l2DB,
189 txSelector,
190 batchBuilder,
191 serverProofs,
192 client,
193 &scConsts,
194 &initSCVars,
195 )
196 if err != nil {
197 return nil, tracerr.Wrap(err)
198 }
199 }
200 var nodeAPI *NodeAPI
201 if cfg.API.Address != "" {
202 if cfg.API.UpdateMetricsInterval.Duration == 0 {
203 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateMetricsInterval: %v",
204 cfg.API.UpdateMetricsInterval.Duration))
205 }
206 if cfg.API.UpdateRecommendedFeeInterval.Duration == 0 {
207 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateRecommendedFeeInterval: %v",
208 cfg.API.UpdateRecommendedFeeInterval.Duration))
209 }
210 server := gin.Default()
211 coord := false
212 if mode == ModeCoordinator {
213 coord = coordCfg.API.Coordinator
214 }
215 var err error
216 nodeAPI, err = NewNodeAPI(
217 cfg.API.Address,
218 coord, cfg.API.Explorer,
219 server,
220 historyDB,
221 stateDB,
222 l2DB,
223 &api.Config{
224 RollupConstants: scConsts.Rollup,
225 AuctionConstants: scConsts.Auction,
226 WDelayerConstants: scConsts.WDelayer,
227 },
228 )
229 if err != nil {
230 return nil, tracerr.Wrap(err)
231 }
232 nodeAPI.api.SetRollupVariables(initSCVars.Rollup)
233 nodeAPI.api.SetAuctionVariables(initSCVars.Auction)
234 nodeAPI.api.SetWDelayerVariables(initSCVars.WDelayer)
235 }
236 var debugAPI *debugapi.DebugAPI
237 if cfg.Debug.APIAddress != "" {
238 debugAPI = debugapi.NewDebugAPI(cfg.Debug.APIAddress, stateDB, sync)
239 }
240 ctx, cancel := context.WithCancel(context.Background())
241 return &Node{
242 nodeAPI: nodeAPI,
243 debugAPI: debugAPI,
244 coord: coord,
245 coordCfg: coordCfg,
246 sync: sync,
247 cfg: cfg,
248 mode: mode,
249 sqlConn: db,
250 ctx: ctx,
251 cancel: cancel,
252 }, nil
253}
254
255// NodeAPI holds the node http API
256type NodeAPI struct { //nolint:golint
257 api *api.API
258 engine *gin.Engine
259 addr string
260}
261
262func handleNoRoute(c *gin.Context) {
263 c.JSON(http.StatusNotFound, gin.H{
264 "error": "404 page not found",
265 })
266}
267
268// NewNodeAPI creates a new NodeAPI (which internally calls api.NewAPI)
269func NewNodeAPI(
270 addr string,
271 coordinatorEndpoints, explorerEndpoints bool,
272 server *gin.Engine,
273 hdb *historydb.HistoryDB,
274 sdb *statedb.StateDB,
275 l2db *l2db.L2DB,
276 config *api.Config,
277) (*NodeAPI, error) {
278 engine := gin.Default()
279 engine.NoRoute(handleNoRoute)
280 engine.Use(cors.Default())
281 _api, err := api.NewAPI(
282 coordinatorEndpoints, explorerEndpoints,
283 engine,
284 hdb,
285 sdb,
286 l2db,
287 config,
288 )
289 if err != nil {
290 return nil, tracerr.Wrap(err)
291 }
292 return &NodeAPI{
293 addr: addr,
294 api: _api,
295 engine: engine,
296 }, nil
297}
298
299// Run starts the http server of the NodeAPI. To stop it, pass a context with
300// cancelation.
301func (a *NodeAPI) Run(ctx context.Context) error {
302 server := &http.Server{
303 Addr: a.addr,
304 Handler: a.engine,
305 // TODO: Figure out best parameters for production
306 ReadTimeout: 30 * time.Second, //nolint:gomnd
307 WriteTimeout: 30 * time.Second, //nolint:gomnd
308 MaxHeaderBytes: 1 << 20, //nolint:gomnd
309 }
310 go func() {
311 log.Infof("NodeAPI is ready at %v", a.addr)
312 if err := server.ListenAndServe(); err != nil && tracerr.Unwrap(err) != http.ErrServerClosed {
313 log.Fatalf("Listen: %s\n", err)
314 }
315 }()
316
317 <-ctx.Done()
318 log.Info("Stopping NodeAPI...")
319 ctxTimeout, cancel := context.WithTimeout(context.Background(), 10*time.Second) //nolint:gomnd
320 defer cancel()
321 if err := server.Shutdown(ctxTimeout); err != nil {
322 return tracerr.Wrap(err)
323 }
324 log.Info("NodeAPI done")
325 return nil
326}
327
328// TODO(Edu): Consider keeping the `lastBlock` inside synchronizer so that we
329// don't have to pass it around.
330func (n *Node) syncLoopFn(lastBlock *common.Block) (*common.Block, time.Duration) {
331 blockData, discarded, err := n.sync.Sync2(n.ctx, lastBlock)
332 stats := n.sync.Stats()
333 if err != nil {
334 // case: error
335 log.Errorw("Synchronizer.Sync", "err", err)
336 return nil, n.cfg.Synchronizer.SyncLoopInterval.Duration
337 } else if discarded != nil {
338 // case: reorg
339 log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
340 if n.mode == ModeCoordinator {
341 n.coord.SendMsg(coordinator.MsgSyncReorg{
342 Stats: *stats,
343 })
344 }
345 if n.nodeAPI != nil {
346 rollup, auction, wDelayer := n.sync.SCVars()
347 n.nodeAPI.api.SetRollupVariables(*rollup)
348 n.nodeAPI.api.SetAuctionVariables(*auction)
349 n.nodeAPI.api.SetWDelayerVariables(*wDelayer)
350 n.nodeAPI.api.UpdateNetworkInfoBlock(
351 stats.Eth.LastBlock, stats.Sync.LastBlock,
352 )
353 }
354 return nil, time.Duration(0)
355 } else if blockData != nil {
356 // case: new block
357 if n.mode == ModeCoordinator {
358 if stats.Synced() && (blockData.Rollup.Vars != nil ||
359 blockData.Auction.Vars != nil ||
360 blockData.WDelayer.Vars != nil) {
361 n.coord.SendMsg(coordinator.MsgSyncSCVars{
362 Rollup: blockData.Rollup.Vars,
363 Auction: blockData.Auction.Vars,
364 WDelayer: blockData.WDelayer.Vars,
365 })
366 }
367 n.coord.SendMsg(coordinator.MsgSyncBlock{
368 Stats: *stats,
369 Batches: blockData.Rollup.Batches,
370 })
371 }
372 if n.nodeAPI != nil {
373 if blockData.Rollup.Vars != nil {
374 n.nodeAPI.api.SetRollupVariables(*blockData.Rollup.Vars)
375 }
376 if blockData.Auction.Vars != nil {
377 n.nodeAPI.api.SetAuctionVariables(*blockData.Auction.Vars)
378 }
379 if blockData.WDelayer.Vars != nil {
380 n.nodeAPI.api.SetWDelayerVariables(*blockData.WDelayer.Vars)
381 }
382
383 if stats.Synced() {
384 if err := n.nodeAPI.api.UpdateNetworkInfo(
385 stats.Eth.LastBlock, stats.Sync.LastBlock,
386 common.BatchNum(stats.Eth.LastBatch),
387 stats.Sync.Auction.CurrentSlot.SlotNum,
388 ); err != nil {
389 log.Errorw("API.UpdateNetworkInfo", "err", err)
390 }
391 }
392 }
393 return &blockData.Block, time.Duration(0)
394 } else {
395 // case: no block
396 return lastBlock, n.cfg.Synchronizer.SyncLoopInterval.Duration
397 }
398}
399
400// StartSynchronizer starts the synchronizer
401func (n *Node) StartSynchronizer() {
402 log.Info("Starting Synchronizer...")
403 n.wg.Add(1)
404 go func() {
405 var lastBlock *common.Block
406 waitDuration := time.Duration(0)
407 for {
408 select {
409 case <-n.ctx.Done():
410 log.Info("Synchronizer done")
411 n.wg.Done()
412 return
413 case <-time.After(waitDuration):
414 lastBlock, waitDuration = n.syncLoopFn(lastBlock)
415 }
416 }
417 }()
418 // TODO: Run price updater. This is required by the API and the TxSelector
419}
420
421// StartDebugAPI starts the DebugAPI
422func (n *Node) StartDebugAPI() {
423 log.Info("Starting DebugAPI...")
424 n.wg.Add(1)
425 go func() {
426 defer func() {
427 log.Info("DebugAPI routine stopped")
428 n.wg.Done()
429 }()
430 if err := n.debugAPI.Run(n.ctx); err != nil {
431 log.Fatalw("DebugAPI.Run", "err", err)
432 }
433 }()
434}
435
436// StartNodeAPI starts the NodeAPI
437func (n *Node) StartNodeAPI() {
438 log.Info("Starting NodeAPI...")
439 n.wg.Add(1)
440 go func() {
441 defer func() {
442 log.Info("NodeAPI routine stopped")
443 n.wg.Done()
444 }()
445 if err := n.nodeAPI.Run(n.ctx); err != nil {
446 log.Fatalw("NodeAPI.Run", "err", err)
447 }
448 }()
449
450 n.wg.Add(1)
451 go func() {
452 for {
453 select {
454 case <-n.ctx.Done():
455 log.Info("API.UpdateMetrics loop done")
456 n.wg.Done()
457 return
458 case <-time.After(n.cfg.API.UpdateMetricsInterval.Duration):
459 if err := n.nodeAPI.api.UpdateMetrics(); err != nil {
460 log.Errorw("API.UpdateMetrics", "err", err)
461 }
462 }
463 }
464 }()
465
466 n.wg.Add(1)
467 go func() {
468 for {
469 select {
470 case <-n.ctx.Done():
471 log.Info("API.UpdateRecommendedFee loop done")
472 n.wg.Done()
473 return
474 case <-time.After(n.cfg.API.UpdateRecommendedFeeInterval.Duration):
475 if err := n.nodeAPI.api.UpdateRecommendedFee(); err != nil {
476 log.Errorw("API.UpdateRecommendedFee", "err", err)
477 }
478 }
479 }
480 }()
481}
482
483// Start the node
484func (n *Node) Start() {
485 log.Infow("Starting node...", "mode", n.mode)
486 if n.debugAPI != nil {
487 n.StartDebugAPI()
488 }
489 if n.nodeAPI != nil {
490 n.StartNodeAPI()
491 }
492 if n.mode == ModeCoordinator {
493 log.Info("Starting Coordinator...")
494 n.coord.Start()
495 }
496 n.StartSynchronizer()
497}
498
499// Stop the node
500func (n *Node) Stop() {
501 log.Infow("Stopping node...")
502 n.cancel()
503 if n.mode == ModeCoordinator {
504 log.Info("Stopping Coordinator...")
505 n.coord.Stop()
506 }
507 n.wg.Wait()
508}