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