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