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
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) (*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: cfg.Coordinator.EthClient.CallGasLimit,
95 DeployGasLimit: cfg.Coordinator.EthClient.DeployGasLimit,
96 GasPriceDiv: cfg.Coordinator.EthClient.GasPriceDiv,
97 ReceiptTimeout: cfg.Coordinator.EthClient.ReceiptTimeout.Duration,
98 IntervalReceiptLoop: cfg.Coordinator.EthClient.ReceiptLoopInterval.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 initSCVars := sync.SCVars()
130
131 scConsts := synchronizer.SCConsts{
132 Rollup: *sync.RollupConstants(),
133 Auction: *sync.AuctionConstants(),
134 WDelayer: *sync.WDelayerConstants(),
135 }
136
137 var coord *coordinator.Coordinator
138 var l2DB *l2db.L2DB
139 if mode == ModeCoordinator {
140 l2DB = l2db.NewL2DB(
141 db,
142 cfg.Coordinator.L2DB.SafetyPeriod,
143 cfg.Coordinator.L2DB.MaxTxs,
144 cfg.Coordinator.L2DB.TTL.Duration,
145 )
146 // TODO: Get (maxL1UserTxs, maxL1OperatorTxs, maxTxs) from the smart contract
147 txSelector, err := txselector.NewTxSelector(cfg.Coordinator.TxSelector.Path, stateDB, l2DB, 10, 10, 10)
148 if err != nil {
149 return nil, tracerr.Wrap(err)
150 }
151 // TODO: Get (configCircuits []ConfigCircuit, batchNum common.BatchNum, nLevels uint64) from smart contract
152 nLevels := uint64(32) //nolint:gomnd
153 batchBuilder, err := batchbuilder.NewBatchBuilder(cfg.Coordinator.BatchBuilder.Path, stateDB, nil, 0, nLevels)
154 if err != nil {
155 return nil, tracerr.Wrap(err)
156 }
157 if err != nil {
158 return nil, tracerr.Wrap(err)
159 }
160 serverProofs := make([]prover.Client, len(cfg.Coordinator.ServerProofs))
161 for i, serverProofCfg := range cfg.Coordinator.ServerProofs {
162 serverProofs[i] = prover.NewProofServerClient(serverProofCfg.URL,
163 cfg.Coordinator.ProofServerPollInterval.Duration)
164 }
165
166 coord, err = coordinator.NewCoordinator(
167 coordinator.Config{
168 ForgerAddress: cfg.Coordinator.ForgerAddress,
169 ConfirmBlocks: cfg.Coordinator.ConfirmBlocks,
170 L1BatchTimeoutPerc: cfg.Coordinator.L1BatchTimeoutPerc,
171 SyncRetryInterval: cfg.Coordinator.SyncRetryInterval.Duration,
172 EthClientAttempts: cfg.Coordinator.EthClient.Attempts,
173 EthClientAttemptsDelay: cfg.Coordinator.EthClient.AttemptsDelay.Duration,
174 TxManagerCheckInterval: cfg.Coordinator.EthClient.CheckLoopInterval.Duration,
175 DebugBatchPath: cfg.Coordinator.Debug.BatchPath,
176 Purger: coordinator.PurgerCfg{
177 PurgeBatchDelay: cfg.Coordinator.L2DB.PurgeBatchDelay,
178 InvalidateBatchDelay: cfg.Coordinator.L2DB.InvalidateBatchDelay,
179 PurgeBlockDelay: cfg.Coordinator.L2DB.PurgeBlockDelay,
180 InvalidateBlockDelay: cfg.Coordinator.L2DB.InvalidateBlockDelay,
181 },
182 },
183 historyDB,
184 l2DB,
185 txSelector,
186 batchBuilder,
187 serverProofs,
188 client,
189 &scConsts,
190 &synchronizer.SCVariables{
191 Rollup: *initSCVars.Rollup,
192 Auction: *initSCVars.Auction,
193 WDelayer: *initSCVars.WDelayer,
194 },
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 = cfg.Coordinator.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 sync: sync,
246 cfg: cfg,
247 mode: mode,
248 sqlConn: db,
249 ctx: ctx,
250 cancel: cancel,
251 }, nil
252}
253
254// NodeAPI holds the node http API
255type NodeAPI struct { //nolint:golint
256 api *api.API
257 engine *gin.Engine
258 addr string
259}
260
261func handleNoRoute(c *gin.Context) {
262 c.JSON(http.StatusNotFound, gin.H{
263 "error": "404 page not found",
264 })
265}
266
267// NewNodeAPI creates a new NodeAPI (which internally calls api.NewAPI)
268func NewNodeAPI(
269 addr string,
270 coordinatorEndpoints, explorerEndpoints bool,
271 server *gin.Engine,
272 hdb *historydb.HistoryDB,
273 sdb *statedb.StateDB,
274 l2db *l2db.L2DB,
275 config *api.Config,
276) (*NodeAPI, error) {
277 engine := gin.Default()
278 engine.NoRoute(handleNoRoute)
279 engine.Use(cors.Default())
280 _api, err := api.NewAPI(
281 coordinatorEndpoints, explorerEndpoints,
282 engine,
283 hdb,
284 sdb,
285 l2db,
286 config,
287 )
288 if err != nil {
289 return nil, tracerr.Wrap(err)
290 }
291 return &NodeAPI{
292 addr: addr,
293 api: _api,
294 engine: engine,
295 }, nil
296}
297
298// Run starts the http server of the NodeAPI. To stop it, pass a context with
299// cancelation.
300func (a *NodeAPI) Run(ctx context.Context) error {
301 server := &http.Server{
302 Addr: a.addr,
303 Handler: a.engine,
304 // TODO: Figure out best parameters for production
305 ReadTimeout: 30 * time.Second, //nolint:gomnd
306 WriteTimeout: 30 * time.Second, //nolint:gomnd
307 MaxHeaderBytes: 1 << 20, //nolint:gomnd
308 }
309 go func() {
310 log.Infof("NodeAPI is ready at %v", a.addr)
311 if err := server.ListenAndServe(); err != nil && tracerr.Unwrap(err) != http.ErrServerClosed {
312 log.Fatalf("Listen: %s\n", err)
313 }
314 }()
315
316 <-ctx.Done()
317 log.Info("Stopping NodeAPI...")
318 ctxTimeout, cancel := context.WithTimeout(context.Background(), 10*time.Second) //nolint:gomnd
319 defer cancel()
320 if err := server.Shutdown(ctxTimeout); err != nil {
321 return tracerr.Wrap(err)
322 }
323 log.Info("NodeAPI done")
324 return nil
325}
326
327func (n *Node) handleNewBlock(stats *synchronizer.Stats, vars synchronizer.SCVariablesPtr,
328 batches []common.BatchData) {
329 if n.mode == ModeCoordinator {
330 n.coord.SendMsg(coordinator.MsgSyncBlock{
331 Stats: *stats,
332 Batches: batches,
333 Vars: synchronizer.SCVariablesPtr{
334 Rollup: vars.Rollup,
335 Auction: vars.Auction,
336 WDelayer: vars.WDelayer,
337 },
338 })
339 }
340 if n.nodeAPI != nil {
341 if vars.Rollup != nil {
342 n.nodeAPI.api.SetRollupVariables(*vars.Rollup)
343 }
344 if vars.Auction != nil {
345 n.nodeAPI.api.SetAuctionVariables(*vars.Auction)
346 }
347 if vars.WDelayer != nil {
348 n.nodeAPI.api.SetWDelayerVariables(*vars.WDelayer)
349 }
350
351 if stats.Synced() {
352 if err := n.nodeAPI.api.UpdateNetworkInfo(
353 stats.Eth.LastBlock, stats.Sync.LastBlock,
354 common.BatchNum(stats.Eth.LastBatch),
355 stats.Sync.Auction.CurrentSlot.SlotNum,
356 ); err != nil {
357 log.Errorw("API.UpdateNetworkInfo", "err", err)
358 }
359 }
360 }
361}
362
363func (n *Node) handleReorg(stats *synchronizer.Stats) {
364 if n.mode == ModeCoordinator {
365 n.coord.SendMsg(coordinator.MsgSyncReorg{
366 Stats: *stats,
367 })
368 }
369 if n.nodeAPI != nil {
370 vars := n.sync.SCVars()
371 n.nodeAPI.api.SetRollupVariables(*vars.Rollup)
372 n.nodeAPI.api.SetAuctionVariables(*vars.Auction)
373 n.nodeAPI.api.SetWDelayerVariables(*vars.WDelayer)
374 n.nodeAPI.api.UpdateNetworkInfoBlock(
375 stats.Eth.LastBlock, stats.Sync.LastBlock,
376 )
377 }
378}
379
380// TODO(Edu): Consider keeping the `lastBlock` inside synchronizer so that we
381// don't have to pass it around.
382func (n *Node) syncLoopFn(lastBlock *common.Block) (*common.Block, time.Duration) {
383 blockData, discarded, err := n.sync.Sync2(n.ctx, lastBlock)
384 stats := n.sync.Stats()
385 if err != nil {
386 // case: error
387 log.Errorw("Synchronizer.Sync", "err", err)
388 return nil, n.cfg.Synchronizer.SyncLoopInterval.Duration
389 } else if discarded != nil {
390 // case: reorg
391 log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
392 n.handleReorg(stats)
393 return nil, time.Duration(0)
394 } else if blockData != nil {
395 // case: new block
396 n.handleNewBlock(stats, synchronizer.SCVariablesPtr{
397 Rollup: blockData.Rollup.Vars,
398 Auction: blockData.Auction.Vars,
399 WDelayer: blockData.WDelayer.Vars,
400 }, blockData.Rollup.Batches)
401 return &blockData.Block, time.Duration(0)
402 } else {
403 // case: no block
404 return lastBlock, n.cfg.Synchronizer.SyncLoopInterval.Duration
405 }
406}
407
408// StartSynchronizer starts the synchronizer
409func (n *Node) StartSynchronizer() {
410 log.Info("Starting Synchronizer...")
411
412 // Trigger a manual call to handleNewBlock with the loaded state of the
413 // synchronizer in order to quickly activate the API and Coordinator
414 // and avoid waiting for the next block. Without this, the API and
415 // Coordinator will not react until the following block (starting from
416 // the last synced one) is synchronized
417 stats := n.sync.Stats()
418 vars := n.sync.SCVars()
419 n.handleNewBlock(stats, vars, []common.BatchData{})
420
421 n.wg.Add(1)
422 go func() {
423 var lastBlock *common.Block
424 waitDuration := time.Duration(0)
425 for {
426 select {
427 case <-n.ctx.Done():
428 log.Info("Synchronizer done")
429 n.wg.Done()
430 return
431 case <-time.After(waitDuration):
432 lastBlock, waitDuration = n.syncLoopFn(lastBlock)
433 }
434 }
435 }()
436 // TODO: Run price updater. This is required by the API and the TxSelector
437}
438
439// StartDebugAPI starts the DebugAPI
440func (n *Node) StartDebugAPI() {
441 log.Info("Starting DebugAPI...")
442 n.wg.Add(1)
443 go func() {
444 defer func() {
445 log.Info("DebugAPI routine stopped")
446 n.wg.Done()
447 }()
448 if err := n.debugAPI.Run(n.ctx); err != nil {
449 log.Fatalw("DebugAPI.Run", "err", err)
450 }
451 }()
452}
453
454// StartNodeAPI starts the NodeAPI
455func (n *Node) StartNodeAPI() {
456 log.Info("Starting NodeAPI...")
457 n.wg.Add(1)
458 go func() {
459 defer func() {
460 log.Info("NodeAPI routine stopped")
461 n.wg.Done()
462 }()
463 if err := n.nodeAPI.Run(n.ctx); err != nil {
464 log.Fatalw("NodeAPI.Run", "err", err)
465 }
466 }()
467
468 n.wg.Add(1)
469 go func() {
470 for {
471 select {
472 case <-n.ctx.Done():
473 log.Info("API.UpdateMetrics loop done")
474 n.wg.Done()
475 return
476 case <-time.After(n.cfg.API.UpdateMetricsInterval.Duration):
477 if err := n.nodeAPI.api.UpdateMetrics(); err != nil {
478 log.Errorw("API.UpdateMetrics", "err", err)
479 }
480 }
481 }
482 }()
483
484 n.wg.Add(1)
485 go func() {
486 for {
487 select {
488 case <-n.ctx.Done():
489 log.Info("API.UpdateRecommendedFee loop done")
490 n.wg.Done()
491 return
492 case <-time.After(n.cfg.API.UpdateRecommendedFeeInterval.Duration):
493 if err := n.nodeAPI.api.UpdateRecommendedFee(); err != nil {
494 log.Errorw("API.UpdateRecommendedFee", "err", err)
495 }
496 }
497 }
498 }()
499}
500
501// Start the node
502func (n *Node) Start() {
503 log.Infow("Starting node...", "mode", n.mode)
504 if n.debugAPI != nil {
505 n.StartDebugAPI()
506 }
507 if n.nodeAPI != nil {
508 n.StartNodeAPI()
509 }
510 if n.mode == ModeCoordinator {
511 log.Info("Starting Coordinator...")
512 n.coord.Start()
513 }
514 n.StartSynchronizer()
515}
516
517// Stop the node
518func (n *Node) Stop() {
519 log.Infow("Stopping node...")
520 n.cancel()
521 if n.mode == ModeCoordinator {
522 log.Info("Stopping Coordinator...")
523 n.coord.Stop()
524 }
525 n.wg.Wait()
526}