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/accounts"
11 ethKeystore "github.com/ethereum/go-ethereum/accounts/keystore"
12 ethCommon "github.com/ethereum/go-ethereum/common"
13 "github.com/ethereum/go-ethereum/ethclient"
14 "github.com/gin-contrib/cors"
15 "github.com/gin-gonic/gin"
16 "github.com/hermeznetwork/hermez-node/api"
17 "github.com/hermeznetwork/hermez-node/batchbuilder"
18 "github.com/hermeznetwork/hermez-node/common"
19 "github.com/hermeznetwork/hermez-node/config"
20 "github.com/hermeznetwork/hermez-node/coordinator"
21 dbUtils "github.com/hermeznetwork/hermez-node/db"
22 "github.com/hermeznetwork/hermez-node/db/historydb"
23 "github.com/hermeznetwork/hermez-node/db/l2db"
24 "github.com/hermeznetwork/hermez-node/db/statedb"
25 "github.com/hermeznetwork/hermez-node/eth"
26 "github.com/hermeznetwork/hermez-node/log"
27 "github.com/hermeznetwork/hermez-node/priceupdater"
28 "github.com/hermeznetwork/hermez-node/prover"
29 "github.com/hermeznetwork/hermez-node/synchronizer"
30 "github.com/hermeznetwork/hermez-node/test/debugapi"
31 "github.com/hermeznetwork/hermez-node/txselector"
32 "github.com/hermeznetwork/tracerr"
33 "github.com/jmoiron/sqlx"
34)
35
36// Mode sets the working mode of the node (synchronizer or coordinator)
37type Mode string
38
39const (
40 // ModeCoordinator defines the mode of the HermezNode as Coordinator, which
41 // means that the node is set to forge (which also will be synchronizing with
42 // the L1 blockchain state)
43 ModeCoordinator Mode = "coordinator"
44
45 // ModeSynchronizer defines the mode of the HermezNode as Synchronizer, which
46 // means that the node is set to only synchronize with the L1 blockchain state
47 // and will not forge
48 ModeSynchronizer Mode = "synchronizer"
49)
50
51// Node is the Hermez Node
52type Node struct {
53 nodeAPI *NodeAPI
54 debugAPI *debugapi.DebugAPI
55 priceUpdater *priceupdater.PriceUpdater
56 // Coordinator
57 coord *coordinator.Coordinator
58
59 // Synchronizer
60 sync *synchronizer.Synchronizer
61
62 // General
63 cfg *config.Node
64 mode Mode
65 sqlConn *sqlx.DB
66 ctx context.Context
67 wg sync.WaitGroup
68 cancel context.CancelFunc
69}
70
71// NewNode creates a Node
72func NewNode(mode Mode, cfg *config.Node) (*Node, error) {
73 // Stablish DB connection
74 db, err := dbUtils.InitSQLDB(
75 cfg.PostgreSQL.Port,
76 cfg.PostgreSQL.Host,
77 cfg.PostgreSQL.User,
78 cfg.PostgreSQL.Password,
79 cfg.PostgreSQL.Name,
80 )
81 if err != nil {
82 return nil, tracerr.Wrap(err)
83 }
84
85 historyDB := historydb.NewHistoryDB(db)
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 var account *accounts.Account
93 var keyStore *ethKeystore.KeyStore
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 scryptN := ethKeystore.StandardScryptN
104 scryptP := ethKeystore.StandardScryptP
105 if cfg.Coordinator.Debug.LightScrypt {
106 scryptN = ethKeystore.LightScryptN
107 scryptP = ethKeystore.LightScryptP
108 }
109 keyStore = ethKeystore.NewKeyStore(cfg.Coordinator.EthClient.Keystore.Path,
110 scryptN, scryptP)
111 if !keyStore.HasAddress(cfg.Coordinator.ForgerAddress) {
112 return nil, tracerr.Wrap(fmt.Errorf(
113 "ethereum keystore doesn't have the key for address %v",
114 cfg.Coordinator.ForgerAddress))
115 }
116 account = &accounts.Account{
117 Address: cfg.Coordinator.ForgerAddress,
118 }
119 if err := keyStore.Unlock(*account,
120 cfg.Coordinator.EthClient.Keystore.Password); err != nil {
121 return nil, tracerr.Wrap(err)
122 }
123 log.Infow("Forger ethereum account unlocked in the keystore",
124 "addr", cfg.Coordinator.ForgerAddress)
125 }
126 client, err := eth.NewClient(ethClient, account, keyStore, ð.ClientConfig{
127 Ethereum: ethCfg,
128 Rollup: eth.RollupConfig{
129 Address: cfg.SmartContracts.Rollup,
130 },
131 Auction: eth.AuctionConfig{
132 Address: cfg.SmartContracts.Auction,
133 TokenHEZ: eth.TokenConfig{
134 Address: cfg.SmartContracts.TokenHEZ,
135 Name: cfg.SmartContracts.TokenHEZName,
136 },
137 },
138 WDelayer: eth.WDelayerConfig{
139 Address: cfg.SmartContracts.WDelayer,
140 },
141 })
142 if err != nil {
143 return nil, tracerr.Wrap(err)
144 }
145
146 chainID, err := client.EthChainID()
147 if err != nil {
148 return nil, tracerr.Wrap(err)
149 }
150 if !chainID.IsUint64() {
151 return nil, tracerr.Wrap(fmt.Errorf("chainID cannot be represented as uint64"))
152 }
153 chainIDU64 := chainID.Uint64()
154 const maxUint16 uint64 = 0xffff
155 if chainIDU64 > maxUint16 {
156 return nil, tracerr.Wrap(fmt.Errorf("chainID overflows uint16"))
157 }
158 chainIDU16 := uint16(chainIDU64)
159
160 const safeStateDBKeep = 128
161 if cfg.StateDB.Keep < safeStateDBKeep {
162 return nil, tracerr.Wrap(fmt.Errorf("cfg.StateDB.Keep = %v < %v, which is unsafe",
163 cfg.StateDB.Keep, safeStateDBKeep))
164 }
165 stateDB, err := statedb.NewStateDB(cfg.StateDB.Path, cfg.StateDB.Keep,
166 statedb.TypeSynchronizer, 32)
167 if err != nil {
168 return nil, tracerr.Wrap(err)
169 }
170
171 sync, err := synchronizer.NewSynchronizer(client, historyDB, stateDB, synchronizer.Config{
172 StatsRefreshPeriod: cfg.Synchronizer.StatsRefreshPeriod.Duration,
173 })
174 if err != nil {
175 return nil, tracerr.Wrap(err)
176 }
177 initSCVars := sync.SCVars()
178
179 scConsts := synchronizer.SCConsts{
180 Rollup: *sync.RollupConstants(),
181 Auction: *sync.AuctionConstants(),
182 WDelayer: *sync.WDelayerConstants(),
183 }
184
185 var coord *coordinator.Coordinator
186 var l2DB *l2db.L2DB
187 if mode == ModeCoordinator {
188 l2DB = l2db.NewL2DB(
189 db,
190 cfg.Coordinator.L2DB.SafetyPeriod,
191 cfg.Coordinator.L2DB.MaxTxs,
192 cfg.Coordinator.L2DB.TTL.Duration,
193 )
194 // TODO: Get (maxL1UserTxs, maxL1OperatorTxs, maxTxs) from the smart contract
195 coordAccount := &txselector.CoordAccount{ // TODO TMP
196 Addr: ethCommon.HexToAddress("0xc58d29fA6e86E4FAe04DDcEd660d45BCf3Cb2370"),
197 BJJ: common.EmptyBJJComp,
198 AccountCreationAuth: nil,
199 }
200 txSelector, err := txselector.NewTxSelector(coordAccount, cfg.Coordinator.TxSelector.Path, stateDB, l2DB)
201 if err != nil {
202 return nil, tracerr.Wrap(err)
203 }
204 // TODO: Get (configCircuits []ConfigCircuit, batchNum common.BatchNum, nLevels uint64) from smart contract
205 nLevels := uint64(32) //nolint:gomnd
206 batchBuilder, err := batchbuilder.NewBatchBuilder(cfg.Coordinator.BatchBuilder.Path, stateDB, nil, 0, nLevels)
207 if err != nil {
208 return nil, tracerr.Wrap(err)
209 }
210 if err != nil {
211 return nil, tracerr.Wrap(err)
212 }
213 serverProofs := make([]prover.Client, len(cfg.Coordinator.ServerProofs))
214 for i, serverProofCfg := range cfg.Coordinator.ServerProofs {
215 serverProofs[i] = prover.NewProofServerClient(serverProofCfg.URL,
216 cfg.Coordinator.ProofServerPollInterval.Duration)
217 }
218
219 coord, err = coordinator.NewCoordinator(
220 coordinator.Config{
221 ForgerAddress: cfg.Coordinator.ForgerAddress,
222 ConfirmBlocks: cfg.Coordinator.ConfirmBlocks,
223 L1BatchTimeoutPerc: cfg.Coordinator.L1BatchTimeoutPerc,
224 SyncRetryInterval: cfg.Coordinator.SyncRetryInterval.Duration,
225 EthClientAttempts: cfg.Coordinator.EthClient.Attempts,
226 EthClientAttemptsDelay: cfg.Coordinator.EthClient.AttemptsDelay.Duration,
227 TxManagerCheckInterval: cfg.Coordinator.EthClient.CheckLoopInterval.Duration,
228 DebugBatchPath: cfg.Coordinator.Debug.BatchPath,
229 Purger: coordinator.PurgerCfg{
230 PurgeBatchDelay: cfg.Coordinator.L2DB.PurgeBatchDelay,
231 InvalidateBatchDelay: cfg.Coordinator.L2DB.InvalidateBatchDelay,
232 PurgeBlockDelay: cfg.Coordinator.L2DB.PurgeBlockDelay,
233 InvalidateBlockDelay: cfg.Coordinator.L2DB.InvalidateBlockDelay,
234 },
235 },
236 historyDB,
237 l2DB,
238 txSelector,
239 batchBuilder,
240 serverProofs,
241 client,
242 &scConsts,
243 &synchronizer.SCVariables{
244 Rollup: *initSCVars.Rollup,
245 Auction: *initSCVars.Auction,
246 WDelayer: *initSCVars.WDelayer,
247 },
248 )
249 if err != nil {
250 return nil, tracerr.Wrap(err)
251 }
252 }
253 var nodeAPI *NodeAPI
254 if cfg.API.Address != "" {
255 if cfg.API.UpdateMetricsInterval.Duration == 0 {
256 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateMetricsInterval: %v",
257 cfg.API.UpdateMetricsInterval.Duration))
258 }
259 if cfg.API.UpdateRecommendedFeeInterval.Duration == 0 {
260 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateRecommendedFeeInterval: %v",
261 cfg.API.UpdateRecommendedFeeInterval.Duration))
262 }
263 server := gin.Default()
264 coord := false
265 if mode == ModeCoordinator {
266 coord = cfg.Coordinator.API.Coordinator
267 }
268 var err error
269 nodeAPI, err = NewNodeAPI(
270 cfg.API.Address,
271 coord, cfg.API.Explorer,
272 server,
273 historyDB,
274 stateDB,
275 l2DB,
276 &api.Config{
277 RollupConstants: scConsts.Rollup,
278 AuctionConstants: scConsts.Auction,
279 WDelayerConstants: scConsts.WDelayer,
280 ChainID: chainIDU16,
281 HermezAddress: cfg.SmartContracts.Rollup,
282 },
283 )
284 if err != nil {
285 return nil, tracerr.Wrap(err)
286 }
287 nodeAPI.api.SetRollupVariables(*initSCVars.Rollup)
288 nodeAPI.api.SetAuctionVariables(*initSCVars.Auction)
289 nodeAPI.api.SetWDelayerVariables(*initSCVars.WDelayer)
290 }
291 var debugAPI *debugapi.DebugAPI
292 if cfg.Debug.APIAddress != "" {
293 debugAPI = debugapi.NewDebugAPI(cfg.Debug.APIAddress, stateDB, sync)
294 }
295 priceUpdater, err := priceupdater.NewPriceUpdater(cfg.PriceUpdater.URL,
296 priceupdater.APIType(cfg.PriceUpdater.Type), historyDB)
297 if err != nil {
298 return nil, tracerr.Wrap(err)
299 }
300 ctx, cancel := context.WithCancel(context.Background())
301 return &Node{
302 nodeAPI: nodeAPI,
303 debugAPI: debugAPI,
304 priceUpdater: priceUpdater,
305 coord: coord,
306 sync: sync,
307 cfg: cfg,
308 mode: mode,
309 sqlConn: db,
310 ctx: ctx,
311 cancel: cancel,
312 }, nil
313}
314
315// NodeAPI holds the node http API
316type NodeAPI struct { //nolint:golint
317 api *api.API
318 engine *gin.Engine
319 addr string
320}
321
322func handleNoRoute(c *gin.Context) {
323 c.JSON(http.StatusNotFound, gin.H{
324 "error": "404 page not found",
325 })
326}
327
328// NewNodeAPI creates a new NodeAPI (which internally calls api.NewAPI)
329func NewNodeAPI(
330 addr string,
331 coordinatorEndpoints, explorerEndpoints bool,
332 server *gin.Engine,
333 hdb *historydb.HistoryDB,
334 sdb *statedb.StateDB,
335 l2db *l2db.L2DB,
336 config *api.Config,
337) (*NodeAPI, error) {
338 engine := gin.Default()
339 engine.NoRoute(handleNoRoute)
340 engine.Use(cors.Default())
341 _api, err := api.NewAPI(
342 coordinatorEndpoints, explorerEndpoints,
343 engine,
344 hdb,
345 sdb,
346 l2db,
347 config,
348 )
349 if err != nil {
350 return nil, tracerr.Wrap(err)
351 }
352 return &NodeAPI{
353 addr: addr,
354 api: _api,
355 engine: engine,
356 }, nil
357}
358
359// Run starts the http server of the NodeAPI. To stop it, pass a context with
360// cancelation.
361func (a *NodeAPI) Run(ctx context.Context) error {
362 server := &http.Server{
363 Addr: a.addr,
364 Handler: a.engine,
365 // TODO: Figure out best parameters for production
366 ReadTimeout: 30 * time.Second, //nolint:gomnd
367 WriteTimeout: 30 * time.Second, //nolint:gomnd
368 MaxHeaderBytes: 1 << 20, //nolint:gomnd
369 }
370 go func() {
371 log.Infof("NodeAPI is ready at %v", a.addr)
372 if err := server.ListenAndServe(); err != nil && tracerr.Unwrap(err) != http.ErrServerClosed {
373 log.Fatalf("Listen: %s\n", err)
374 }
375 }()
376
377 <-ctx.Done()
378 log.Info("Stopping NodeAPI...")
379 ctxTimeout, cancel := context.WithTimeout(context.Background(), 10*time.Second) //nolint:gomnd
380 defer cancel()
381 if err := server.Shutdown(ctxTimeout); err != nil {
382 return tracerr.Wrap(err)
383 }
384 log.Info("NodeAPI done")
385 return nil
386}
387
388func (n *Node) handleNewBlock(stats *synchronizer.Stats, vars synchronizer.SCVariablesPtr,
389 batches []common.BatchData) {
390 if n.mode == ModeCoordinator {
391 n.coord.SendMsg(coordinator.MsgSyncBlock{
392 Stats: *stats,
393 Batches: batches,
394 Vars: synchronizer.SCVariablesPtr{
395 Rollup: vars.Rollup,
396 Auction: vars.Auction,
397 WDelayer: vars.WDelayer,
398 },
399 })
400 }
401 if n.nodeAPI != nil {
402 if vars.Rollup != nil {
403 n.nodeAPI.api.SetRollupVariables(*vars.Rollup)
404 }
405 if vars.Auction != nil {
406 n.nodeAPI.api.SetAuctionVariables(*vars.Auction)
407 }
408 if vars.WDelayer != nil {
409 n.nodeAPI.api.SetWDelayerVariables(*vars.WDelayer)
410 }
411
412 if stats.Synced() {
413 if err := n.nodeAPI.api.UpdateNetworkInfo(
414 stats.Eth.LastBlock, stats.Sync.LastBlock,
415 common.BatchNum(stats.Eth.LastBatch),
416 stats.Sync.Auction.CurrentSlot.SlotNum,
417 ); err != nil {
418 log.Errorw("API.UpdateNetworkInfo", "err", err)
419 }
420 }
421 }
422}
423
424func (n *Node) handleReorg(stats *synchronizer.Stats) {
425 if n.mode == ModeCoordinator {
426 n.coord.SendMsg(coordinator.MsgSyncReorg{
427 Stats: *stats,
428 })
429 }
430 if n.nodeAPI != nil {
431 vars := n.sync.SCVars()
432 n.nodeAPI.api.SetRollupVariables(*vars.Rollup)
433 n.nodeAPI.api.SetAuctionVariables(*vars.Auction)
434 n.nodeAPI.api.SetWDelayerVariables(*vars.WDelayer)
435 n.nodeAPI.api.UpdateNetworkInfoBlock(
436 stats.Eth.LastBlock, stats.Sync.LastBlock,
437 )
438 }
439}
440
441// TODO(Edu): Consider keeping the `lastBlock` inside synchronizer so that we
442// don't have to pass it around.
443func (n *Node) syncLoopFn(ctx context.Context, lastBlock *common.Block) (*common.Block, time.Duration, error) {
444 blockData, discarded, err := n.sync.Sync2(ctx, lastBlock)
445 stats := n.sync.Stats()
446 if err != nil {
447 // case: error
448 return nil, n.cfg.Synchronizer.SyncLoopInterval.Duration, tracerr.Wrap(err)
449 } else if discarded != nil {
450 // case: reorg
451 log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
452 n.handleReorg(stats)
453 return nil, time.Duration(0), nil
454 } else if blockData != nil {
455 // case: new block
456 n.handleNewBlock(stats, synchronizer.SCVariablesPtr{
457 Rollup: blockData.Rollup.Vars,
458 Auction: blockData.Auction.Vars,
459 WDelayer: blockData.WDelayer.Vars,
460 }, blockData.Rollup.Batches)
461 return &blockData.Block, time.Duration(0), nil
462 } else {
463 // case: no block
464 return lastBlock, n.cfg.Synchronizer.SyncLoopInterval.Duration, nil
465 }
466}
467
468// StartSynchronizer starts the synchronizer
469func (n *Node) StartSynchronizer() {
470 log.Info("Starting Synchronizer...")
471
472 // Trigger a manual call to handleNewBlock with the loaded state of the
473 // synchronizer in order to quickly activate the API and Coordinator
474 // and avoid waiting for the next block. Without this, the API and
475 // Coordinator will not react until the following block (starting from
476 // the last synced one) is synchronized
477 stats := n.sync.Stats()
478 vars := n.sync.SCVars()
479 n.handleNewBlock(stats, vars, []common.BatchData{})
480
481 n.wg.Add(1)
482 go func() {
483 var err error
484 var lastBlock *common.Block
485 waitDuration := time.Duration(0)
486 for {
487 select {
488 case <-n.ctx.Done():
489 log.Info("Synchronizer done")
490 n.wg.Done()
491 return
492 case <-time.After(waitDuration):
493 if lastBlock, waitDuration, err = n.syncLoopFn(n.ctx,
494 lastBlock); err != nil {
495 if n.ctx.Err() != nil {
496 continue
497 }
498 log.Errorw("Synchronizer.Sync", "err", err)
499 }
500 }
501 }
502 }()
503
504 n.wg.Add(1)
505 go func() {
506 for {
507 select {
508 case <-n.ctx.Done():
509 log.Info("PriceUpdater done")
510 n.wg.Done()
511 return
512 case <-time.After(n.cfg.PriceUpdater.Interval.Duration):
513 if err := n.priceUpdater.UpdateTokenList(); err != nil {
514 log.Errorw("PriceUpdater.UpdateTokenList()", "err", err)
515 }
516 n.priceUpdater.UpdatePrices(n.ctx)
517 }
518 }
519 }()
520}
521
522// StartDebugAPI starts the DebugAPI
523func (n *Node) StartDebugAPI() {
524 log.Info("Starting DebugAPI...")
525 n.wg.Add(1)
526 go func() {
527 defer func() {
528 log.Info("DebugAPI routine stopped")
529 n.wg.Done()
530 }()
531 if err := n.debugAPI.Run(n.ctx); err != nil {
532 if n.ctx.Err() != nil {
533 return
534 }
535 log.Fatalw("DebugAPI.Run", "err", err)
536 }
537 }()
538}
539
540// StartNodeAPI starts the NodeAPI
541func (n *Node) StartNodeAPI() {
542 log.Info("Starting NodeAPI...")
543 n.wg.Add(1)
544 go func() {
545 defer func() {
546 log.Info("NodeAPI routine stopped")
547 n.wg.Done()
548 }()
549 if err := n.nodeAPI.Run(n.ctx); err != nil {
550 if n.ctx.Err() != nil {
551 return
552 }
553 log.Fatalw("NodeAPI.Run", "err", err)
554 }
555 }()
556
557 n.wg.Add(1)
558 go func() {
559 for {
560 select {
561 case <-n.ctx.Done():
562 log.Info("API.UpdateMetrics loop done")
563 n.wg.Done()
564 return
565 case <-time.After(n.cfg.API.UpdateMetricsInterval.Duration):
566 if err := n.nodeAPI.api.UpdateMetrics(); err != nil {
567 log.Errorw("API.UpdateMetrics", "err", err)
568 }
569 }
570 }
571 }()
572
573 n.wg.Add(1)
574 go func() {
575 for {
576 select {
577 case <-n.ctx.Done():
578 log.Info("API.UpdateRecommendedFee loop done")
579 n.wg.Done()
580 return
581 case <-time.After(n.cfg.API.UpdateRecommendedFeeInterval.Duration):
582 if err := n.nodeAPI.api.UpdateRecommendedFee(); err != nil {
583 log.Errorw("API.UpdateRecommendedFee", "err", err)
584 }
585 }
586 }
587 }()
588}
589
590// Start the node
591func (n *Node) Start() {
592 log.Infow("Starting node...", "mode", n.mode)
593 if n.debugAPI != nil {
594 n.StartDebugAPI()
595 }
596 if n.nodeAPI != nil {
597 n.StartNodeAPI()
598 }
599 if n.mode == ModeCoordinator {
600 log.Info("Starting Coordinator...")
601 n.coord.Start()
602 }
603 n.StartSynchronizer()
604}
605
606// Stop the node
607func (n *Node) Stop() {
608 log.Infow("Stopping node...")
609 n.cancel()
610 if n.mode == ModeCoordinator {
611 log.Info("Stopping Coordinator...")
612 n.coord.Stop()
613 }
614 n.wg.Wait()
615}