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 "github.com/russross/meddler"
36)
37
38// Mode sets the working mode of the node (synchronizer or coordinator)
39type Mode string
40
41const (
42 // ModeCoordinator defines the mode of the HermezNode as Coordinator, which
43 // means that the node is set to forge (which also will be synchronizing with
44 // the L1 blockchain state)
45 ModeCoordinator Mode = "coordinator"
46
47 // ModeSynchronizer defines the mode of the HermezNode as Synchronizer, which
48 // means that the node is set to only synchronize with the L1 blockchain state
49 // and will not forge
50 ModeSynchronizer Mode = "synchronizer"
51)
52
53// Node is the Hermez Node
54type Node struct {
55 nodeAPI *NodeAPI
56 debugAPI *debugapi.DebugAPI
57 priceUpdater *priceupdater.PriceUpdater
58 // Coordinator
59 coord *coordinator.Coordinator
60
61 // Synchronizer
62 sync *synchronizer.Synchronizer
63
64 // General
65 cfg *config.Node
66 mode Mode
67 sqlConn *sqlx.DB
68 ctx context.Context
69 wg sync.WaitGroup
70 cancel context.CancelFunc
71}
72
73// NewNode creates a Node
74func NewNode(mode Mode, cfg *config.Node) (*Node, error) {
75 meddler.Debug = cfg.Debug.MeddlerLogs
76 // Stablish DB connection
77 db, err := dbUtils.InitSQLDB(
78 cfg.PostgreSQL.Port,
79 cfg.PostgreSQL.Host,
80 cfg.PostgreSQL.User,
81 cfg.PostgreSQL.Password,
82 cfg.PostgreSQL.Name,
83 )
84 if err != nil {
85 return nil, tracerr.Wrap(err)
86 }
87
88 historyDB := historydb.NewHistoryDB(db)
89
90 ethClient, err := ethclient.Dial(cfg.Web3.URL)
91 if err != nil {
92 return nil, tracerr.Wrap(err)
93 }
94 var ethCfg eth.EthereumConfig
95 var account *accounts.Account
96 var keyStore *ethKeystore.KeyStore
97 if mode == ModeCoordinator {
98 ethCfg = eth.EthereumConfig{
99 CallGasLimit: cfg.Coordinator.EthClient.CallGasLimit,
100 GasPriceDiv: cfg.Coordinator.EthClient.GasPriceDiv,
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 batchBuilder, err := batchbuilder.NewBatchBuilder(cfg.Coordinator.BatchBuilder.Path,
206 stateDB, nil, 0, uint64(cfg.Coordinator.Circuit.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 txProcessorCfg := txprocessor.Config{
220 NLevels: uint32(cfg.Coordinator.Circuit.NLevels),
221 MaxTx: uint32(cfg.Coordinator.Circuit.MaxTx),
222 ChainID: chainIDU16,
223 MaxFeeTx: common.RollupConstMaxFeeIdxCoordinator,
224 MaxL1Tx: common.RollupConstMaxL1Tx,
225 }
226 verifierIdx, err := scConsts.Rollup.FindVerifierIdx(
227 cfg.Coordinator.Circuit.MaxTx,
228 cfg.Coordinator.Circuit.NLevels,
229 )
230 if err != nil {
231 return nil, tracerr.Wrap(err)
232 }
233 log.Infow("Found verifier that matches circuit config", "verifierIdx", verifierIdx)
234
235 coord, err = coordinator.NewCoordinator(
236 coordinator.Config{
237 ForgerAddress: cfg.Coordinator.ForgerAddress,
238 ConfirmBlocks: cfg.Coordinator.ConfirmBlocks,
239 L1BatchTimeoutPerc: cfg.Coordinator.L1BatchTimeoutPerc,
240 SyncRetryInterval: cfg.Coordinator.SyncRetryInterval.Duration,
241 EthClientAttempts: cfg.Coordinator.EthClient.Attempts,
242 EthClientAttemptsDelay: cfg.Coordinator.EthClient.AttemptsDelay.Duration,
243 TxManagerCheckInterval: cfg.Coordinator.EthClient.CheckLoopInterval.Duration,
244 DebugBatchPath: cfg.Coordinator.Debug.BatchPath,
245 Purger: coordinator.PurgerCfg{
246 PurgeBatchDelay: cfg.Coordinator.L2DB.PurgeBatchDelay,
247 InvalidateBatchDelay: cfg.Coordinator.L2DB.InvalidateBatchDelay,
248 PurgeBlockDelay: cfg.Coordinator.L2DB.PurgeBlockDelay,
249 InvalidateBlockDelay: cfg.Coordinator.L2DB.InvalidateBlockDelay,
250 },
251 VerifierIdx: uint8(verifierIdx),
252 TxProcessorConfig: txProcessorCfg,
253 },
254 historyDB,
255 l2DB,
256 txSelector,
257 batchBuilder,
258 serverProofs,
259 client,
260 &scConsts,
261 &synchronizer.SCVariables{
262 Rollup: *initSCVars.Rollup,
263 Auction: *initSCVars.Auction,
264 WDelayer: *initSCVars.WDelayer,
265 },
266 )
267 if err != nil {
268 return nil, tracerr.Wrap(err)
269 }
270 }
271 var nodeAPI *NodeAPI
272 if cfg.API.Address != "" {
273 if cfg.API.UpdateMetricsInterval.Duration == 0 {
274 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateMetricsInterval: %v",
275 cfg.API.UpdateMetricsInterval.Duration))
276 }
277 if cfg.API.UpdateRecommendedFeeInterval.Duration == 0 {
278 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateRecommendedFeeInterval: %v",
279 cfg.API.UpdateRecommendedFeeInterval.Duration))
280 }
281 server := gin.Default()
282 coord := false
283 if mode == ModeCoordinator {
284 coord = cfg.Coordinator.API.Coordinator
285 }
286 var err error
287 nodeAPI, err = NewNodeAPI(
288 cfg.API.Address,
289 coord, cfg.API.Explorer,
290 server,
291 historyDB,
292 stateDB,
293 l2DB,
294 &api.Config{
295 RollupConstants: scConsts.Rollup,
296 AuctionConstants: scConsts.Auction,
297 WDelayerConstants: scConsts.WDelayer,
298 ChainID: chainIDU16,
299 HermezAddress: cfg.SmartContracts.Rollup,
300 },
301 )
302 if err != nil {
303 return nil, tracerr.Wrap(err)
304 }
305 nodeAPI.api.SetRollupVariables(*initSCVars.Rollup)
306 nodeAPI.api.SetAuctionVariables(*initSCVars.Auction)
307 nodeAPI.api.SetWDelayerVariables(*initSCVars.WDelayer)
308 }
309 var debugAPI *debugapi.DebugAPI
310 if cfg.Debug.APIAddress != "" {
311 debugAPI = debugapi.NewDebugAPI(cfg.Debug.APIAddress, stateDB, sync)
312 }
313 priceUpdater, err := priceupdater.NewPriceUpdater(cfg.PriceUpdater.URL,
314 priceupdater.APIType(cfg.PriceUpdater.Type), historyDB)
315 if err != nil {
316 return nil, tracerr.Wrap(err)
317 }
318 ctx, cancel := context.WithCancel(context.Background())
319 return &Node{
320 nodeAPI: nodeAPI,
321 debugAPI: debugAPI,
322 priceUpdater: priceUpdater,
323 coord: coord,
324 sync: sync,
325 cfg: cfg,
326 mode: mode,
327 sqlConn: db,
328 ctx: ctx,
329 cancel: cancel,
330 }, nil
331}
332
333// NodeAPI holds the node http API
334type NodeAPI struct { //nolint:golint
335 api *api.API
336 engine *gin.Engine
337 addr string
338}
339
340func handleNoRoute(c *gin.Context) {
341 c.JSON(http.StatusNotFound, gin.H{
342 "error": "404 page not found",
343 })
344}
345
346// NewNodeAPI creates a new NodeAPI (which internally calls api.NewAPI)
347func NewNodeAPI(
348 addr string,
349 coordinatorEndpoints, explorerEndpoints bool,
350 server *gin.Engine,
351 hdb *historydb.HistoryDB,
352 sdb *statedb.StateDB,
353 l2db *l2db.L2DB,
354 config *api.Config,
355) (*NodeAPI, error) {
356 engine := gin.Default()
357 engine.NoRoute(handleNoRoute)
358 engine.Use(cors.Default())
359 _api, err := api.NewAPI(
360 coordinatorEndpoints, explorerEndpoints,
361 engine,
362 hdb,
363 sdb,
364 l2db,
365 config,
366 )
367 if err != nil {
368 return nil, tracerr.Wrap(err)
369 }
370 return &NodeAPI{
371 addr: addr,
372 api: _api,
373 engine: engine,
374 }, nil
375}
376
377// Run starts the http server of the NodeAPI. To stop it, pass a context with
378// cancelation.
379func (a *NodeAPI) Run(ctx context.Context) error {
380 server := &http.Server{
381 Addr: a.addr,
382 Handler: a.engine,
383 // TODO: Figure out best parameters for production
384 ReadTimeout: 30 * time.Second, //nolint:gomnd
385 WriteTimeout: 30 * time.Second, //nolint:gomnd
386 MaxHeaderBytes: 1 << 20, //nolint:gomnd
387 }
388 go func() {
389 log.Infof("NodeAPI is ready at %v", a.addr)
390 if err := server.ListenAndServe(); err != nil && tracerr.Unwrap(err) != http.ErrServerClosed {
391 log.Fatalf("Listen: %s\n", err)
392 }
393 }()
394
395 <-ctx.Done()
396 log.Info("Stopping NodeAPI...")
397 ctxTimeout, cancel := context.WithTimeout(context.Background(), 10*time.Second) //nolint:gomnd
398 defer cancel()
399 if err := server.Shutdown(ctxTimeout); err != nil {
400 return tracerr.Wrap(err)
401 }
402 log.Info("NodeAPI done")
403 return nil
404}
405
406func (n *Node) handleNewBlock(stats *synchronizer.Stats, vars synchronizer.SCVariablesPtr,
407 batches []common.BatchData) {
408 if n.mode == ModeCoordinator {
409 n.coord.SendMsg(coordinator.MsgSyncBlock{
410 Stats: *stats,
411 Batches: batches,
412 Vars: synchronizer.SCVariablesPtr{
413 Rollup: vars.Rollup,
414 Auction: vars.Auction,
415 WDelayer: vars.WDelayer,
416 },
417 })
418 }
419 if n.nodeAPI != nil {
420 if vars.Rollup != nil {
421 n.nodeAPI.api.SetRollupVariables(*vars.Rollup)
422 }
423 if vars.Auction != nil {
424 n.nodeAPI.api.SetAuctionVariables(*vars.Auction)
425 }
426 if vars.WDelayer != nil {
427 n.nodeAPI.api.SetWDelayerVariables(*vars.WDelayer)
428 }
429
430 if stats.Synced() {
431 if err := n.nodeAPI.api.UpdateNetworkInfo(
432 stats.Eth.LastBlock, stats.Sync.LastBlock,
433 common.BatchNum(stats.Eth.LastBatch),
434 stats.Sync.Auction.CurrentSlot.SlotNum,
435 ); err != nil {
436 log.Errorw("API.UpdateNetworkInfo", "err", err)
437 }
438 }
439 }
440}
441
442func (n *Node) handleReorg(stats *synchronizer.Stats) {
443 if n.mode == ModeCoordinator {
444 n.coord.SendMsg(coordinator.MsgSyncReorg{
445 Stats: *stats,
446 })
447 }
448 if n.nodeAPI != nil {
449 vars := n.sync.SCVars()
450 n.nodeAPI.api.SetRollupVariables(*vars.Rollup)
451 n.nodeAPI.api.SetAuctionVariables(*vars.Auction)
452 n.nodeAPI.api.SetWDelayerVariables(*vars.WDelayer)
453 n.nodeAPI.api.UpdateNetworkInfoBlock(
454 stats.Eth.LastBlock, stats.Sync.LastBlock,
455 )
456 }
457}
458
459// TODO(Edu): Consider keeping the `lastBlock` inside synchronizer so that we
460// don't have to pass it around.
461func (n *Node) syncLoopFn(ctx context.Context, lastBlock *common.Block) (*common.Block, time.Duration, error) {
462 blockData, discarded, err := n.sync.Sync2(ctx, lastBlock)
463 stats := n.sync.Stats()
464 if err != nil {
465 // case: error
466 return nil, n.cfg.Synchronizer.SyncLoopInterval.Duration, tracerr.Wrap(err)
467 } else if discarded != nil {
468 // case: reorg
469 log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
470 n.handleReorg(stats)
471 return nil, time.Duration(0), nil
472 } else if blockData != nil {
473 // case: new block
474 n.handleNewBlock(stats, synchronizer.SCVariablesPtr{
475 Rollup: blockData.Rollup.Vars,
476 Auction: blockData.Auction.Vars,
477 WDelayer: blockData.WDelayer.Vars,
478 }, blockData.Rollup.Batches)
479 return &blockData.Block, time.Duration(0), nil
480 } else {
481 // case: no block
482 return lastBlock, n.cfg.Synchronizer.SyncLoopInterval.Duration, nil
483 }
484}
485
486// StartSynchronizer starts the synchronizer
487func (n *Node) StartSynchronizer() {
488 log.Info("Starting Synchronizer...")
489
490 // Trigger a manual call to handleNewBlock with the loaded state of the
491 // synchronizer in order to quickly activate the API and Coordinator
492 // and avoid waiting for the next block. Without this, the API and
493 // Coordinator will not react until the following block (starting from
494 // the last synced one) is synchronized
495 stats := n.sync.Stats()
496 vars := n.sync.SCVars()
497 n.handleNewBlock(stats, vars, []common.BatchData{})
498
499 n.wg.Add(1)
500 go func() {
501 var err error
502 var lastBlock *common.Block
503 waitDuration := time.Duration(0)
504 for {
505 select {
506 case <-n.ctx.Done():
507 log.Info("Synchronizer done")
508 n.wg.Done()
509 return
510 case <-time.After(waitDuration):
511 if lastBlock, waitDuration, err = n.syncLoopFn(n.ctx,
512 lastBlock); err != nil {
513 if n.ctx.Err() != nil {
514 continue
515 }
516 log.Errorw("Synchronizer.Sync", "err", err)
517 }
518 }
519 }
520 }()
521
522 n.wg.Add(1)
523 go func() {
524 for {
525 select {
526 case <-n.ctx.Done():
527 log.Info("PriceUpdater done")
528 n.wg.Done()
529 return
530 case <-time.After(n.cfg.PriceUpdater.Interval.Duration):
531 if err := n.priceUpdater.UpdateTokenList(); err != nil {
532 log.Errorw("PriceUpdater.UpdateTokenList()", "err", err)
533 }
534 n.priceUpdater.UpdatePrices(n.ctx)
535 }
536 }
537 }()
538}
539
540// StartDebugAPI starts the DebugAPI
541func (n *Node) StartDebugAPI() {
542 log.Info("Starting DebugAPI...")
543 n.wg.Add(1)
544 go func() {
545 defer func() {
546 log.Info("DebugAPI routine stopped")
547 n.wg.Done()
548 }()
549 if err := n.debugAPI.Run(n.ctx); err != nil {
550 if n.ctx.Err() != nil {
551 return
552 }
553 log.Fatalw("DebugAPI.Run", "err", err)
554 }
555 }()
556}
557
558// StartNodeAPI starts the NodeAPI
559func (n *Node) StartNodeAPI() {
560 log.Info("Starting NodeAPI...")
561 n.wg.Add(1)
562 go func() {
563 defer func() {
564 log.Info("NodeAPI routine stopped")
565 n.wg.Done()
566 }()
567 if err := n.nodeAPI.Run(n.ctx); err != nil {
568 if n.ctx.Err() != nil {
569 return
570 }
571 log.Fatalw("NodeAPI.Run", "err", err)
572 }
573 }()
574
575 n.wg.Add(1)
576 go func() {
577 for {
578 select {
579 case <-n.ctx.Done():
580 log.Info("API.UpdateMetrics loop done")
581 n.wg.Done()
582 return
583 case <-time.After(n.cfg.API.UpdateMetricsInterval.Duration):
584 if err := n.nodeAPI.api.UpdateMetrics(); err != nil {
585 log.Errorw("API.UpdateMetrics", "err", err)
586 }
587 }
588 }
589 }()
590
591 n.wg.Add(1)
592 go func() {
593 for {
594 select {
595 case <-n.ctx.Done():
596 log.Info("API.UpdateRecommendedFee loop done")
597 n.wg.Done()
598 return
599 case <-time.After(n.cfg.API.UpdateRecommendedFeeInterval.Duration):
600 if err := n.nodeAPI.api.UpdateRecommendedFee(); err != nil {
601 log.Errorw("API.UpdateRecommendedFee", "err", err)
602 }
603 }
604 }
605 }()
606}
607
608// Start the node
609func (n *Node) Start() {
610 log.Infow("Starting node...", "mode", n.mode)
611 if n.debugAPI != nil {
612 n.StartDebugAPI()
613 }
614 if n.nodeAPI != nil {
615 n.StartNodeAPI()
616 }
617 if n.mode == ModeCoordinator {
618 log.Info("Starting Coordinator...")
619 n.coord.Start()
620 }
621 n.StartSynchronizer()
622}
623
624// Stop the node
625func (n *Node) Stop() {
626 log.Infow("Stopping node...")
627 n.cancel()
628 if n.mode == ModeCoordinator {
629 log.Info("Stopping Coordinator...")
630 n.coord.Stop()
631 }
632 n.wg.Wait()
633}