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