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 ethCommon "github.com/ethereum/go-ethereum/common"
11 "github.com/ethereum/go-ethereum/ethclient"
12 "github.com/gin-contrib/cors"
13 "github.com/gin-gonic/gin"
14 "github.com/hermeznetwork/hermez-node/api"
15 "github.com/hermeznetwork/hermez-node/batchbuilder"
16 "github.com/hermeznetwork/hermez-node/common"
17 "github.com/hermeznetwork/hermez-node/config"
18 "github.com/hermeznetwork/hermez-node/coordinator"
19 dbUtils "github.com/hermeznetwork/hermez-node/db"
20 "github.com/hermeznetwork/hermez-node/db/historydb"
21 "github.com/hermeznetwork/hermez-node/db/l2db"
22 "github.com/hermeznetwork/hermez-node/db/statedb"
23 "github.com/hermeznetwork/hermez-node/eth"
24 "github.com/hermeznetwork/hermez-node/log"
25 "github.com/hermeznetwork/hermez-node/prover"
26 "github.com/hermeznetwork/hermez-node/synchronizer"
27 "github.com/hermeznetwork/hermez-node/test/debugapi"
28 "github.com/hermeznetwork/hermez-node/txselector"
29 "github.com/hermeznetwork/tracerr"
30 "github.com/jmoiron/sqlx"
31)
32
33// Mode sets the working mode of the node (synchronizer or coordinator)
34type Mode string
35
36const (
37 // ModeCoordinator defines the mode of the HermezNode as Coordinator, which
38 // means that the node is set to forge (which also will be synchronizing with
39 // the L1 blockchain state)
40 ModeCoordinator Mode = "coordinator"
41
42 // ModeSynchronizer defines the mode of the HermezNode as Synchronizer, which
43 // means that the node is set to only synchronize with the L1 blockchain state
44 // and will not forge
45 ModeSynchronizer Mode = "synchronizer"
46)
47
48// Node is the Hermez Node
49type Node struct {
50 nodeAPI *NodeAPI
51 debugAPI *debugapi.DebugAPI
52 // Coordinator
53 coord *coordinator.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) (*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: cfg.Coordinator.EthClient.CallGasLimit,
96 DeployGasLimit: cfg.Coordinator.EthClient.DeployGasLimit,
97 GasPriceDiv: cfg.Coordinator.EthClient.GasPriceDiv,
98 ReceiptTimeout: cfg.Coordinator.EthClient.ReceiptTimeout.Duration,
99 IntervalReceiptLoop: cfg.Coordinator.EthClient.ReceiptLoopInterval.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 initSCVars := sync.SCVars()
131
132 scConsts := synchronizer.SCConsts{
133 Rollup: *sync.RollupConstants(),
134 Auction: *sync.AuctionConstants(),
135 WDelayer: *sync.WDelayerConstants(),
136 }
137
138 var coord *coordinator.Coordinator
139 var l2DB *l2db.L2DB
140 if mode == ModeCoordinator {
141 l2DB = l2db.NewL2DB(
142 db,
143 cfg.Coordinator.L2DB.SafetyPeriod,
144 cfg.Coordinator.L2DB.MaxTxs,
145 cfg.Coordinator.L2DB.TTL.Duration,
146 )
147 // TODO: Get (maxL1UserTxs, maxL1OperatorTxs, maxTxs) from the smart contract
148 coordAccount := &txselector.CoordAccount{ // TODO TMP
149 Addr: ethCommon.HexToAddress("0xc58d29fA6e86E4FAe04DDcEd660d45BCf3Cb2370"),
150 BJJ: nil,
151 AccountCreationAuth: nil,
152 }
153 txSelector, err := txselector.NewTxSelector(coordAccount, cfg.Coordinator.TxSelector.Path, stateDB, l2DB)
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(cfg.Coordinator.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(cfg.Coordinator.ServerProofs))
167 for i, serverProofCfg := range cfg.Coordinator.ServerProofs {
168 serverProofs[i] = prover.NewProofServerClient(serverProofCfg.URL,
169 cfg.Coordinator.ProofServerPollInterval.Duration)
170 }
171
172 coord, err = coordinator.NewCoordinator(
173 coordinator.Config{
174 ForgerAddress: cfg.Coordinator.ForgerAddress,
175 ConfirmBlocks: cfg.Coordinator.ConfirmBlocks,
176 L1BatchTimeoutPerc: cfg.Coordinator.L1BatchTimeoutPerc,
177 SyncRetryInterval: cfg.Coordinator.SyncRetryInterval.Duration,
178 EthClientAttempts: cfg.Coordinator.EthClient.Attempts,
179 EthClientAttemptsDelay: cfg.Coordinator.EthClient.AttemptsDelay.Duration,
180 TxManagerCheckInterval: cfg.Coordinator.EthClient.CheckLoopInterval.Duration,
181 DebugBatchPath: cfg.Coordinator.Debug.BatchPath,
182 Purger: coordinator.PurgerCfg{
183 PurgeBatchDelay: cfg.Coordinator.L2DB.PurgeBatchDelay,
184 InvalidateBatchDelay: cfg.Coordinator.L2DB.InvalidateBatchDelay,
185 PurgeBlockDelay: cfg.Coordinator.L2DB.PurgeBlockDelay,
186 InvalidateBlockDelay: cfg.Coordinator.L2DB.InvalidateBlockDelay,
187 },
188 },
189 historyDB,
190 l2DB,
191 txSelector,
192 batchBuilder,
193 serverProofs,
194 client,
195 &scConsts,
196 &synchronizer.SCVariables{
197 Rollup: *initSCVars.Rollup,
198 Auction: *initSCVars.Auction,
199 WDelayer: *initSCVars.WDelayer,
200 },
201 )
202 if err != nil {
203 return nil, tracerr.Wrap(err)
204 }
205 }
206 var nodeAPI *NodeAPI
207 if cfg.API.Address != "" {
208 if cfg.API.UpdateMetricsInterval.Duration == 0 {
209 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateMetricsInterval: %v",
210 cfg.API.UpdateMetricsInterval.Duration))
211 }
212 if cfg.API.UpdateRecommendedFeeInterval.Duration == 0 {
213 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateRecommendedFeeInterval: %v",
214 cfg.API.UpdateRecommendedFeeInterval.Duration))
215 }
216 server := gin.Default()
217 coord := false
218 if mode == ModeCoordinator {
219 coord = cfg.Coordinator.API.Coordinator
220 }
221 var err error
222 nodeAPI, err = NewNodeAPI(
223 cfg.API.Address,
224 coord, cfg.API.Explorer,
225 server,
226 historyDB,
227 stateDB,
228 l2DB,
229 &api.Config{
230 RollupConstants: scConsts.Rollup,
231 AuctionConstants: scConsts.Auction,
232 WDelayerConstants: scConsts.WDelayer,
233 },
234 )
235 if err != nil {
236 return nil, tracerr.Wrap(err)
237 }
238 nodeAPI.api.SetRollupVariables(*initSCVars.Rollup)
239 nodeAPI.api.SetAuctionVariables(*initSCVars.Auction)
240 nodeAPI.api.SetWDelayerVariables(*initSCVars.WDelayer)
241 }
242 var debugAPI *debugapi.DebugAPI
243 if cfg.Debug.APIAddress != "" {
244 debugAPI = debugapi.NewDebugAPI(cfg.Debug.APIAddress, stateDB, sync)
245 }
246 ctx, cancel := context.WithCancel(context.Background())
247 return &Node{
248 nodeAPI: nodeAPI,
249 debugAPI: debugAPI,
250 coord: coord,
251 sync: sync,
252 cfg: cfg,
253 mode: mode,
254 sqlConn: db,
255 ctx: ctx,
256 cancel: cancel,
257 }, nil
258}
259
260// NodeAPI holds the node http API
261type NodeAPI struct { //nolint:golint
262 api *api.API
263 engine *gin.Engine
264 addr string
265}
266
267func handleNoRoute(c *gin.Context) {
268 c.JSON(http.StatusNotFound, gin.H{
269 "error": "404 page not found",
270 })
271}
272
273// NewNodeAPI creates a new NodeAPI (which internally calls api.NewAPI)
274func NewNodeAPI(
275 addr string,
276 coordinatorEndpoints, explorerEndpoints bool,
277 server *gin.Engine,
278 hdb *historydb.HistoryDB,
279 sdb *statedb.StateDB,
280 l2db *l2db.L2DB,
281 config *api.Config,
282) (*NodeAPI, error) {
283 engine := gin.Default()
284 engine.NoRoute(handleNoRoute)
285 engine.Use(cors.Default())
286 _api, err := api.NewAPI(
287 coordinatorEndpoints, explorerEndpoints,
288 engine,
289 hdb,
290 sdb,
291 l2db,
292 config,
293 )
294 if err != nil {
295 return nil, tracerr.Wrap(err)
296 }
297 return &NodeAPI{
298 addr: addr,
299 api: _api,
300 engine: engine,
301 }, nil
302}
303
304// Run starts the http server of the NodeAPI. To stop it, pass a context with
305// cancelation.
306func (a *NodeAPI) Run(ctx context.Context) error {
307 server := &http.Server{
308 Addr: a.addr,
309 Handler: a.engine,
310 // TODO: Figure out best parameters for production
311 ReadTimeout: 30 * time.Second, //nolint:gomnd
312 WriteTimeout: 30 * time.Second, //nolint:gomnd
313 MaxHeaderBytes: 1 << 20, //nolint:gomnd
314 }
315 go func() {
316 log.Infof("NodeAPI is ready at %v", a.addr)
317 if err := server.ListenAndServe(); err != nil && tracerr.Unwrap(err) != http.ErrServerClosed {
318 log.Fatalf("Listen: %s\n", err)
319 }
320 }()
321
322 <-ctx.Done()
323 log.Info("Stopping NodeAPI...")
324 ctxTimeout, cancel := context.WithTimeout(context.Background(), 10*time.Second) //nolint:gomnd
325 defer cancel()
326 if err := server.Shutdown(ctxTimeout); err != nil {
327 return tracerr.Wrap(err)
328 }
329 log.Info("NodeAPI done")
330 return nil
331}
332
333func (n *Node) handleNewBlock(stats *synchronizer.Stats, vars synchronizer.SCVariablesPtr,
334 batches []common.BatchData) {
335 if n.mode == ModeCoordinator {
336 n.coord.SendMsg(coordinator.MsgSyncBlock{
337 Stats: *stats,
338 Batches: batches,
339 Vars: synchronizer.SCVariablesPtr{
340 Rollup: vars.Rollup,
341 Auction: vars.Auction,
342 WDelayer: vars.WDelayer,
343 },
344 })
345 }
346 if n.nodeAPI != nil {
347 if vars.Rollup != nil {
348 n.nodeAPI.api.SetRollupVariables(*vars.Rollup)
349 }
350 if vars.Auction != nil {
351 n.nodeAPI.api.SetAuctionVariables(*vars.Auction)
352 }
353 if vars.WDelayer != nil {
354 n.nodeAPI.api.SetWDelayerVariables(*vars.WDelayer)
355 }
356
357 if stats.Synced() {
358 if err := n.nodeAPI.api.UpdateNetworkInfo(
359 stats.Eth.LastBlock, stats.Sync.LastBlock,
360 common.BatchNum(stats.Eth.LastBatch),
361 stats.Sync.Auction.CurrentSlot.SlotNum,
362 ); err != nil {
363 log.Errorw("API.UpdateNetworkInfo", "err", err)
364 }
365 }
366 }
367}
368
369func (n *Node) handleReorg(stats *synchronizer.Stats) {
370 if n.mode == ModeCoordinator {
371 n.coord.SendMsg(coordinator.MsgSyncReorg{
372 Stats: *stats,
373 })
374 }
375 if n.nodeAPI != nil {
376 vars := n.sync.SCVars()
377 n.nodeAPI.api.SetRollupVariables(*vars.Rollup)
378 n.nodeAPI.api.SetAuctionVariables(*vars.Auction)
379 n.nodeAPI.api.SetWDelayerVariables(*vars.WDelayer)
380 n.nodeAPI.api.UpdateNetworkInfoBlock(
381 stats.Eth.LastBlock, stats.Sync.LastBlock,
382 )
383 }
384}
385
386// TODO(Edu): Consider keeping the `lastBlock` inside synchronizer so that we
387// don't have to pass it around.
388func (n *Node) syncLoopFn(lastBlock *common.Block) (*common.Block, time.Duration) {
389 blockData, discarded, err := n.sync.Sync2(n.ctx, lastBlock)
390 stats := n.sync.Stats()
391 if err != nil {
392 // case: error
393 log.Errorw("Synchronizer.Sync", "err", err)
394 return nil, n.cfg.Synchronizer.SyncLoopInterval.Duration
395 } else if discarded != nil {
396 // case: reorg
397 log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
398 n.handleReorg(stats)
399 return nil, time.Duration(0)
400 } else if blockData != nil {
401 // case: new block
402 n.handleNewBlock(stats, synchronizer.SCVariablesPtr{
403 Rollup: blockData.Rollup.Vars,
404 Auction: blockData.Auction.Vars,
405 WDelayer: blockData.WDelayer.Vars,
406 }, blockData.Rollup.Batches)
407 return &blockData.Block, time.Duration(0)
408 } else {
409 // case: no block
410 return lastBlock, n.cfg.Synchronizer.SyncLoopInterval.Duration
411 }
412}
413
414// StartSynchronizer starts the synchronizer
415func (n *Node) StartSynchronizer() {
416 log.Info("Starting Synchronizer...")
417
418 // Trigger a manual call to handleNewBlock with the loaded state of the
419 // synchronizer in order to quickly activate the API and Coordinator
420 // and avoid waiting for the next block. Without this, the API and
421 // Coordinator will not react until the following block (starting from
422 // the last synced one) is synchronized
423 stats := n.sync.Stats()
424 vars := n.sync.SCVars()
425 n.handleNewBlock(stats, vars, []common.BatchData{})
426
427 n.wg.Add(1)
428 go func() {
429 var lastBlock *common.Block
430 waitDuration := time.Duration(0)
431 for {
432 select {
433 case <-n.ctx.Done():
434 log.Info("Synchronizer done")
435 n.wg.Done()
436 return
437 case <-time.After(waitDuration):
438 lastBlock, waitDuration = n.syncLoopFn(lastBlock)
439 }
440 }
441 }()
442 // TODO: Run price updater. This is required by the API and the TxSelector
443}
444
445// StartDebugAPI starts the DebugAPI
446func (n *Node) StartDebugAPI() {
447 log.Info("Starting DebugAPI...")
448 n.wg.Add(1)
449 go func() {
450 defer func() {
451 log.Info("DebugAPI routine stopped")
452 n.wg.Done()
453 }()
454 if err := n.debugAPI.Run(n.ctx); err != nil {
455 log.Fatalw("DebugAPI.Run", "err", err)
456 }
457 }()
458}
459
460// StartNodeAPI starts the NodeAPI
461func (n *Node) StartNodeAPI() {
462 log.Info("Starting NodeAPI...")
463 n.wg.Add(1)
464 go func() {
465 defer func() {
466 log.Info("NodeAPI routine stopped")
467 n.wg.Done()
468 }()
469 if err := n.nodeAPI.Run(n.ctx); err != nil {
470 log.Fatalw("NodeAPI.Run", "err", err)
471 }
472 }()
473
474 n.wg.Add(1)
475 go func() {
476 for {
477 select {
478 case <-n.ctx.Done():
479 log.Info("API.UpdateMetrics loop done")
480 n.wg.Done()
481 return
482 case <-time.After(n.cfg.API.UpdateMetricsInterval.Duration):
483 if err := n.nodeAPI.api.UpdateMetrics(); err != nil {
484 log.Errorw("API.UpdateMetrics", "err", err)
485 }
486 }
487 }
488 }()
489
490 n.wg.Add(1)
491 go func() {
492 for {
493 select {
494 case <-n.ctx.Done():
495 log.Info("API.UpdateRecommendedFee loop done")
496 n.wg.Done()
497 return
498 case <-time.After(n.cfg.API.UpdateRecommendedFeeInterval.Duration):
499 if err := n.nodeAPI.api.UpdateRecommendedFee(); err != nil {
500 log.Errorw("API.UpdateRecommendedFee", "err", err)
501 }
502 }
503 }
504 }()
505}
506
507// Start the node
508func (n *Node) Start() {
509 log.Infow("Starting node...", "mode", n.mode)
510 if n.debugAPI != nil {
511 n.StartDebugAPI()
512 }
513 if n.nodeAPI != nil {
514 n.StartNodeAPI()
515 }
516 if n.mode == ModeCoordinator {
517 log.Info("Starting Coordinator...")
518 n.coord.Start()
519 }
520 n.StartSynchronizer()
521}
522
523// Stop the node
524func (n *Node) Stop() {
525 log.Infow("Stopping node...")
526 n.cancel()
527 if n.mode == ModeCoordinator {
528 log.Info("Stopping Coordinator...")
529 n.coord.Stop()
530 }
531 n.wg.Wait()
532}