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/hermeznetwork/tracerr"
28 "github.com/jmoiron/sqlx"
29)
30
31// Mode sets the working mode of the node (synchronizer or coordinator)
32type Mode string
33
34const (
35 // ModeCoordinator defines the mode of the HermezNode as Coordinator, which
36 // means that the node is set to forge (which also will be synchronizing with
37 // the L1 blockchain state)
38 ModeCoordinator Mode = "coordinator"
39
40 // ModeSynchronizer defines the mode of the HermezNode as Synchronizer, which
41 // means that the node is set to only synchronize with the L1 blockchain state
42 // and will not forge
43 ModeSynchronizer Mode = "synchronizer"
44)
45
46// Node is the Hermez Node
47type Node struct {
48 nodeAPI *NodeAPI
49 debugAPI *debugapi.DebugAPI
50 // Coordinator
51 coord *coordinator.Coordinator
52 coordCfg *config.Coordinator
53
54 // Synchronizer
55 sync *synchronizer.Synchronizer
56
57 // General
58 cfg *config.Node
59 mode Mode
60 sqlConn *sqlx.DB
61 ctx context.Context
62 wg sync.WaitGroup
63 cancel context.CancelFunc
64}
65
66// NewNode creates a Node
67func NewNode(mode Mode, cfg *config.Node, coordCfg *config.Coordinator) (*Node, error) {
68 // Stablish DB connection
69 db, err := dbUtils.InitSQLDB(
70 cfg.PostgreSQL.Port,
71 cfg.PostgreSQL.Host,
72 cfg.PostgreSQL.User,
73 cfg.PostgreSQL.Password,
74 cfg.PostgreSQL.Name,
75 )
76 if err != nil {
77 return nil, tracerr.Wrap(err)
78 }
79
80 historyDB := historydb.NewHistoryDB(db)
81
82 stateDB, err := statedb.NewStateDB(cfg.StateDB.Path, statedb.TypeSynchronizer, 32)
83 if err != nil {
84 return nil, tracerr.Wrap(err)
85 }
86
87 ethClient, err := ethclient.Dial(cfg.Web3.URL)
88 if err != nil {
89 return nil, tracerr.Wrap(err)
90 }
91 var ethCfg eth.EthereumConfig
92 if mode == ModeCoordinator {
93 ethCfg = eth.EthereumConfig{
94 CallGasLimit: coordCfg.EthClient.CallGasLimit,
95 DeployGasLimit: coordCfg.EthClient.DeployGasLimit,
96 GasPriceDiv: coordCfg.EthClient.GasPriceDiv,
97 ReceiptTimeout: coordCfg.EthClient.ReceiptTimeout.Duration,
98 IntervalReceiptLoop: coordCfg.EthClient.IntervalReceiptLoop.Duration,
99 }
100 }
101 client, err := eth.NewClient(ethClient, nil, nil, ð.ClientConfig{
102 Ethereum: ethCfg,
103 Rollup: eth.RollupConfig{
104 Address: cfg.SmartContracts.Rollup,
105 },
106 Auction: eth.AuctionConfig{
107 Address: cfg.SmartContracts.Auction,
108 TokenHEZ: eth.TokenConfig{
109 Address: cfg.SmartContracts.TokenHEZ,
110 Name: cfg.SmartContracts.TokenHEZName,
111 },
112 },
113 WDelayer: eth.WDelayerConfig{
114 Address: cfg.SmartContracts.WDelayer,
115 },
116 })
117 if err != nil {
118 return nil, tracerr.Wrap(err)
119 }
120
121 sync, err := synchronizer.NewSynchronizer(client, historyDB, stateDB, synchronizer.Config{
122 StartBlockNum: cfg.Synchronizer.StartBlockNum,
123 InitialVariables: cfg.Synchronizer.InitialVariables,
124 StatsRefreshPeriod: cfg.Synchronizer.StatsRefreshPeriod.Duration,
125 })
126 if err != nil {
127 return nil, tracerr.Wrap(err)
128 }
129 varsRollup, varsAuction, varsWDelayer := sync.SCVars()
130 initSCVars := synchronizer.SCVariables{
131 Rollup: *varsRollup,
132 Auction: *varsAuction,
133 WDelayer: *varsWDelayer,
134 }
135
136 scConsts := synchronizer.SCConsts{
137 Rollup: *sync.RollupConstants(),
138 Auction: *sync.AuctionConstants(),
139 WDelayer: *sync.WDelayerConstants(),
140 }
141
142 var coord *coordinator.Coordinator
143 var l2DB *l2db.L2DB
144 if mode == ModeCoordinator {
145 l2DB = l2db.NewL2DB(
146 db,
147 coordCfg.L2DB.SafetyPeriod,
148 coordCfg.L2DB.MaxTxs,
149 coordCfg.L2DB.TTL.Duration,
150 )
151 // TODO: Get (maxL1UserTxs, maxL1OperatorTxs, maxTxs) from the smart contract
152 txSelector, err := txselector.NewTxSelector(coordCfg.TxSelector.Path, stateDB, l2DB, 10, 10, 10)
153 if err != nil {
154 return nil, tracerr.Wrap(err)
155 }
156 // TODO: Get (configCircuits []ConfigCircuit, batchNum common.BatchNum, nLevels uint64) from smart contract
157 nLevels := uint64(32) //nolint:gomnd
158 batchBuilder, err := batchbuilder.NewBatchBuilder(coordCfg.BatchBuilder.Path, stateDB, nil, 0, nLevels)
159 if err != nil {
160 return nil, tracerr.Wrap(err)
161 }
162 if err != nil {
163 return nil, tracerr.Wrap(err)
164 }
165 serverProofs := make([]coordinator.ServerProofInterface, len(coordCfg.ServerProofs))
166 for i, serverProofCfg := range coordCfg.ServerProofs {
167 serverProofs[i] = coordinator.NewServerProof(serverProofCfg.URL)
168 }
169
170 coord, err = coordinator.NewCoordinator(
171 coordinator.Config{
172 ForgerAddress: coordCfg.ForgerAddress,
173 ConfirmBlocks: coordCfg.ConfirmBlocks,
174 },
175 historyDB,
176 txSelector,
177 batchBuilder,
178 serverProofs,
179 client,
180 &scConsts,
181 &initSCVars,
182 )
183 if err != nil {
184 return nil, tracerr.Wrap(err)
185 }
186 }
187 var nodeAPI *NodeAPI
188 if cfg.API.Address != "" {
189 if cfg.API.UpdateMetricsInterval.Duration == 0 {
190 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateMetricsInterval: %v",
191 cfg.API.UpdateMetricsInterval.Duration))
192 }
193 if cfg.API.UpdateRecommendedFeeInterval.Duration == 0 {
194 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateRecommendedFeeInterval: %v",
195 cfg.API.UpdateRecommendedFeeInterval.Duration))
196 }
197 server := gin.Default()
198 coord := false
199 if mode == ModeCoordinator {
200 coord = coordCfg.API.Coordinator
201 }
202 var err error
203 nodeAPI, err = NewNodeAPI(
204 cfg.API.Address,
205 coord, cfg.API.Explorer,
206 server,
207 historyDB,
208 stateDB,
209 l2DB,
210 &api.Config{
211 RollupConstants: scConsts.Rollup,
212 AuctionConstants: scConsts.Auction,
213 WDelayerConstants: scConsts.WDelayer,
214 },
215 )
216 if err != nil {
217 return nil, tracerr.Wrap(err)
218 }
219 nodeAPI.api.SetRollupVariables(initSCVars.Rollup)
220 nodeAPI.api.SetAuctionVariables(initSCVars.Auction)
221 nodeAPI.api.SetWDelayerVariables(initSCVars.WDelayer)
222 }
223 var debugAPI *debugapi.DebugAPI
224 if cfg.Debug.APIAddress != "" {
225 debugAPI = debugapi.NewDebugAPI(cfg.Debug.APIAddress, stateDB, sync)
226 }
227 ctx, cancel := context.WithCancel(context.Background())
228 return &Node{
229 nodeAPI: nodeAPI,
230 debugAPI: debugAPI,
231 coord: coord,
232 coordCfg: coordCfg,
233 sync: sync,
234 cfg: cfg,
235 mode: mode,
236 sqlConn: db,
237 ctx: ctx,
238 cancel: cancel,
239 }, nil
240}
241
242// NodeAPI holds the node http API
243type NodeAPI struct { //nolint:golint
244 api *api.API
245 engine *gin.Engine
246 addr string
247}
248
249func handleNoRoute(c *gin.Context) {
250 c.JSON(http.StatusNotFound, gin.H{
251 "error": "404 page not found",
252 })
253}
254
255// NewNodeAPI creates a new NodeAPI (which internally calls api.NewAPI)
256func NewNodeAPI(
257 addr string,
258 coordinatorEndpoints, explorerEndpoints bool,
259 server *gin.Engine,
260 hdb *historydb.HistoryDB,
261 sdb *statedb.StateDB,
262 l2db *l2db.L2DB,
263 config *api.Config,
264) (*NodeAPI, error) {
265 engine := gin.Default()
266 engine.NoRoute(handleNoRoute)
267 engine.Use(cors.Default())
268 _api, err := api.NewAPI(
269 coordinatorEndpoints, explorerEndpoints,
270 engine,
271 hdb,
272 sdb,
273 l2db,
274 config,
275 )
276 if err != nil {
277 return nil, tracerr.Wrap(err)
278 }
279 return &NodeAPI{
280 addr: addr,
281 api: _api,
282 engine: engine,
283 }, nil
284}
285
286// Run starts the http server of the NodeAPI. To stop it, pass a context with
287// cancelation.
288func (a *NodeAPI) Run(ctx context.Context) error {
289 server := &http.Server{
290 Addr: a.addr,
291 Handler: a.engine,
292 // TODO: Figure out best parameters for production
293 ReadTimeout: 30 * time.Second, //nolint:gomnd
294 WriteTimeout: 30 * time.Second, //nolint:gomnd
295 MaxHeaderBytes: 1 << 20, //nolint:gomnd
296 }
297 go func() {
298 log.Infof("NodeAPI is ready at %v", a.addr)
299 if err := server.ListenAndServe(); err != nil && tracerr.Unwrap(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 tracerr.Wrap(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}