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