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