hermez-node/node/node.go
node/node.goRaw
1/*
2Package node does the initialization of all the required objects to either run
3as a synchronizer or as a coordinator.
4
5The Node contains several goroutines that run in the background or that
6periodically perform tasks. One of this goroutines periodically calls the
7`Synchronizer.Sync` function, allowing the synchronization of one block at a
8time. After every call to `Synchronizer.Sync`, the Node sends a message to the
9Coordinator to notify it about the new synced block (and associated state) or
10reorg (and resetted state) in case one happens.
11
12Other goroutines perform tasks such as: updating the token prices, update
13metrics stored in the historyDB, update recommended fee stored in the
14historyDB, run the http API server, run the debug http API server, etc.
15*/
16package node
17
18import (
19 "context"
20 "errors"
21 "fmt"
22 "net"
23 "net/http"
24 "regexp"
25 "sync"
26 "time"
27
28 "github.com/ethereum/go-ethereum/accounts"
29 ethKeystore "github.com/ethereum/go-ethereum/accounts/keystore"
30 "github.com/ethereum/go-ethereum/ethclient"
31 "github.com/gin-contrib/cors"
32 "github.com/gin-gonic/gin"
33 "github.com/hermeznetwork/hermez-node/api"
34 "github.com/hermeznetwork/hermez-node/api/stateapiupdater"
35 "github.com/hermeznetwork/hermez-node/batchbuilder"
36 "github.com/hermeznetwork/hermez-node/common"
37 "github.com/hermeznetwork/hermez-node/config"
38 "github.com/hermeznetwork/hermez-node/coordinator"
39 dbUtils "github.com/hermeznetwork/hermez-node/db"
40 "github.com/hermeznetwork/hermez-node/db/historydb"
41 "github.com/hermeznetwork/hermez-node/db/l2db"
42 "github.com/hermeznetwork/hermez-node/db/statedb"
43 "github.com/hermeznetwork/hermez-node/eth"
44 "github.com/hermeznetwork/hermez-node/log"
45 "github.com/hermeznetwork/hermez-node/priceupdater"
46 "github.com/hermeznetwork/hermez-node/prover"
47 "github.com/hermeznetwork/hermez-node/synchronizer"
48 "github.com/hermeznetwork/hermez-node/test/debugapi"
49 "github.com/hermeznetwork/hermez-node/txprocessor"
50 "github.com/hermeznetwork/hermez-node/txselector"
51 "github.com/hermeznetwork/tracerr"
52 "github.com/jmoiron/sqlx"
53 "github.com/russross/meddler"
54)
55
56// Mode sets the working mode of the node (synchronizer or coordinator)
57type Mode string
58
59const (
60 // ModeCoordinator defines the mode of the HermezNode as Coordinator, which
61 // means that the node is set to forge (which also will be synchronizing with
62 // the L1 blockchain state)
63 ModeCoordinator Mode = "coordinator"
64
65 // ModeSynchronizer defines the mode of the HermezNode as Synchronizer, which
66 // means that the node is set to only synchronize with the L1 blockchain state
67 // and will not forge
68 ModeSynchronizer Mode = "synchronizer"
69)
70
71// Node is the Hermez Node
72type Node struct {
73 nodeAPI *NodeAPI
74 stateAPIUpdater *stateapiupdater.Updater
75 debugAPI *debugapi.DebugAPI
76 priceUpdater *priceupdater.PriceUpdater
77 // Coordinator
78 coord *coordinator.Coordinator
79
80 // Synchronizer
81 sync *synchronizer.Synchronizer
82
83 // General
84 cfg *config.Node
85 mode Mode
86 sqlConnRead *sqlx.DB
87 sqlConnWrite *sqlx.DB
88 historyDB *historydb.HistoryDB
89 ctx context.Context
90 wg sync.WaitGroup
91 cancel context.CancelFunc
92}
93
94// NewNode creates a Node
95func NewNode(mode Mode, cfg *config.Node) (*Node, error) {
96 meddler.Debug = cfg.Debug.MeddlerLogs
97 // Stablish DB connection
98 dbWrite, err := dbUtils.InitSQLDB(
99 cfg.PostgreSQL.PortWrite,
100 cfg.PostgreSQL.HostWrite,
101 cfg.PostgreSQL.UserWrite,
102 cfg.PostgreSQL.PasswordWrite,
103 cfg.PostgreSQL.NameWrite,
104 )
105 if err != nil {
106 return nil, tracerr.Wrap(fmt.Errorf("dbUtils.InitSQLDB: %w", err))
107 }
108 var dbRead *sqlx.DB
109 if cfg.PostgreSQL.HostRead == "" {
110 dbRead = dbWrite
111 } else if cfg.PostgreSQL.HostRead == cfg.PostgreSQL.HostWrite {
112 return nil, tracerr.Wrap(fmt.Errorf(
113 "PostgreSQL.HostRead and PostgreSQL.HostWrite must be different",
114 ))
115 } else {
116 dbRead, err = dbUtils.InitSQLDB(
117 cfg.PostgreSQL.PortRead,
118 cfg.PostgreSQL.HostRead,
119 cfg.PostgreSQL.UserRead,
120 cfg.PostgreSQL.PasswordRead,
121 cfg.PostgreSQL.NameRead,
122 )
123 if err != nil {
124 return nil, tracerr.Wrap(fmt.Errorf("dbUtils.InitSQLDB: %w", err))
125 }
126 }
127 var apiConnCon *dbUtils.APIConnectionController
128 if cfg.API.Explorer || mode == ModeCoordinator {
129 apiConnCon = dbUtils.NewAPIConnectionController(
130 cfg.API.MaxSQLConnections,
131 cfg.API.SQLConnectionTimeout.Duration,
132 )
133 }
134
135 historyDB := historydb.NewHistoryDB(dbRead, dbWrite, apiConnCon)
136
137 ethClient, err := ethclient.Dial(cfg.Web3.URL)
138 if err != nil {
139 return nil, tracerr.Wrap(err)
140 }
141 var ethCfg eth.EthereumConfig
142 var account *accounts.Account
143 var keyStore *ethKeystore.KeyStore
144 if mode == ModeCoordinator {
145 ethCfg = eth.EthereumConfig{
146 CallGasLimit: 0, // cfg.Coordinator.EthClient.CallGasLimit,
147 GasPriceDiv: 0, // cfg.Coordinator.EthClient.GasPriceDiv,
148 }
149
150 scryptN := ethKeystore.StandardScryptN
151 scryptP := ethKeystore.StandardScryptP
152 if cfg.Coordinator.Debug.LightScrypt {
153 scryptN = ethKeystore.LightScryptN
154 scryptP = ethKeystore.LightScryptP
155 }
156 keyStore = ethKeystore.NewKeyStore(cfg.Coordinator.EthClient.Keystore.Path,
157 scryptN, scryptP)
158
159 balance, err := ethClient.BalanceAt(context.TODO(), cfg.Coordinator.ForgerAddress, nil)
160 if err != nil {
161 return nil, tracerr.Wrap(err)
162 }
163
164 minForgeBalance := cfg.Coordinator.MinimumForgeAddressBalance
165 if minForgeBalance != nil && balance.Cmp(minForgeBalance) == -1 {
166 return nil, tracerr.Wrap(fmt.Errorf(
167 "forger account balance is less than cfg.Coordinator.MinimumForgeAddressBalance: %v < %v",
168 balance, minForgeBalance))
169 }
170 log.Infow("forger ethereum account balance",
171 "addr", cfg.Coordinator.ForgerAddress,
172 "balance", balance,
173 "minForgeBalance", minForgeBalance,
174 )
175
176 // Unlock Coordinator ForgerAddr in the keystore to make calls
177 // to ForgeBatch in the smart contract
178 if !keyStore.HasAddress(cfg.Coordinator.ForgerAddress) {
179 return nil, tracerr.Wrap(fmt.Errorf(
180 "ethereum keystore doesn't have the key for address %v",
181 cfg.Coordinator.ForgerAddress))
182 }
183 account = &accounts.Account{
184 Address: cfg.Coordinator.ForgerAddress,
185 }
186 if err := keyStore.Unlock(*account,
187 cfg.Coordinator.EthClient.Keystore.Password); err != nil {
188 return nil, tracerr.Wrap(err)
189 }
190 log.Infow("Forger ethereum account unlocked in the keystore",
191 "addr", cfg.Coordinator.ForgerAddress)
192 }
193 client, err := eth.NewClient(ethClient, account, keyStore, ð.ClientConfig{
194 Ethereum: ethCfg,
195 Rollup: eth.RollupConfig{
196 Address: cfg.SmartContracts.Rollup,
197 },
198 Auction: eth.AuctionConfig{
199 Address: cfg.SmartContracts.Auction,
200 TokenHEZ: eth.TokenConfig{
201 Address: cfg.SmartContracts.TokenHEZ,
202 Name: cfg.SmartContracts.TokenHEZName,
203 },
204 },
205 WDelayer: eth.WDelayerConfig{
206 Address: cfg.SmartContracts.WDelayer,
207 },
208 })
209 if err != nil {
210 return nil, tracerr.Wrap(err)
211 }
212
213 chainID, err := client.EthChainID()
214 if err != nil {
215 return nil, tracerr.Wrap(err)
216 }
217 if !chainID.IsUint64() {
218 return nil, tracerr.Wrap(fmt.Errorf("chainID cannot be represented as uint64"))
219 }
220 chainIDU64 := chainID.Uint64()
221 const maxUint16 uint64 = 0xffff
222 if chainIDU64 > maxUint16 {
223 return nil, tracerr.Wrap(fmt.Errorf("chainID overflows uint16"))
224 }
225 chainIDU16 := uint16(chainIDU64)
226
227 const safeStateDBKeep = 128
228 if cfg.StateDB.Keep < safeStateDBKeep {
229 return nil, tracerr.Wrap(fmt.Errorf("cfg.StateDB.Keep = %v < %v, which is unsafe",
230 cfg.StateDB.Keep, safeStateDBKeep))
231 }
232 stateDB, err := statedb.NewStateDB(statedb.Config{
233 Path: cfg.StateDB.Path,
234 Keep: cfg.StateDB.Keep,
235 Type: statedb.TypeSynchronizer,
236 NLevels: statedb.MaxNLevels,
237 })
238 if err != nil {
239 return nil, tracerr.Wrap(err)
240 }
241
242 var l2DB *l2db.L2DB
243 if mode == ModeCoordinator {
244 l2DB = l2db.NewL2DB(
245 dbRead, dbWrite,
246 cfg.Coordinator.L2DB.SafetyPeriod,
247 cfg.Coordinator.L2DB.MaxTxs,
248 cfg.Coordinator.L2DB.MinFeeUSD,
249 cfg.Coordinator.L2DB.MaxFeeUSD,
250 cfg.Coordinator.L2DB.TTL.Duration,
251 apiConnCon,
252 )
253 }
254
255 sync, err := synchronizer.NewSynchronizer(client, historyDB, l2DB, stateDB, synchronizer.Config{
256 StatsRefreshPeriod: cfg.Synchronizer.StatsRefreshPeriod.Duration,
257 ChainID: chainIDU16,
258 })
259 if err != nil {
260 return nil, tracerr.Wrap(err)
261 }
262 initSCVars := sync.SCVars()
263
264 scConsts := common.SCConsts{
265 Rollup: *sync.RollupConstants(),
266 Auction: *sync.AuctionConstants(),
267 WDelayer: *sync.WDelayerConstants(),
268 }
269
270 hdbNodeCfg := historydb.NodeConfig{
271 MaxPoolTxs: cfg.Coordinator.L2DB.MaxTxs,
272 MinFeeUSD: cfg.Coordinator.L2DB.MinFeeUSD,
273 MaxFeeUSD: cfg.Coordinator.L2DB.MaxFeeUSD,
274 ForgeDelay: cfg.Coordinator.ForgeDelay.Duration.Seconds(),
275 }
276 if err := historyDB.SetNodeConfig(&hdbNodeCfg); err != nil {
277 return nil, tracerr.Wrap(err)
278 }
279 hdbConsts := historydb.Constants{
280 SCConsts: common.SCConsts{
281 Rollup: scConsts.Rollup,
282 Auction: scConsts.Auction,
283 WDelayer: scConsts.WDelayer,
284 },
285 ChainID: chainIDU16,
286 HermezAddress: cfg.SmartContracts.Rollup,
287 }
288 if err := historyDB.SetConstants(&hdbConsts); err != nil {
289 return nil, tracerr.Wrap(err)
290 }
291
292 stateAPIUpdater := stateapiupdater.NewUpdater(historyDB, &hdbNodeCfg, initSCVars, &hdbConsts)
293
294 var coord *coordinator.Coordinator
295 if mode == ModeCoordinator {
296 // Unlock FeeAccount EthAddr in the keystore to generate the
297 // account creation authorization
298 if !keyStore.HasAddress(cfg.Coordinator.FeeAccount.Address) {
299 return nil, tracerr.Wrap(fmt.Errorf(
300 "ethereum keystore doesn't have the key for address %v",
301 cfg.Coordinator.FeeAccount.Address))
302 }
303 feeAccount := accounts.Account{
304 Address: cfg.Coordinator.FeeAccount.Address,
305 }
306 if err := keyStore.Unlock(feeAccount,
307 cfg.Coordinator.EthClient.Keystore.Password); err != nil {
308 return nil, tracerr.Wrap(err)
309 }
310 auth := &common.AccountCreationAuth{
311 EthAddr: cfg.Coordinator.FeeAccount.Address,
312 BJJ: cfg.Coordinator.FeeAccount.BJJ,
313 }
314 if err := auth.Sign(func(msg []byte) ([]byte, error) {
315 return keyStore.SignHash(feeAccount, msg)
316 }, chainIDU16, cfg.SmartContracts.Rollup); err != nil {
317 return nil, tracerr.Wrap(err)
318 }
319 coordAccount := &txselector.CoordAccount{
320 Addr: cfg.Coordinator.FeeAccount.Address,
321 BJJ: cfg.Coordinator.FeeAccount.BJJ,
322 AccountCreationAuth: auth.Signature,
323 }
324 txSelector, err := txselector.NewTxSelector(coordAccount,
325 cfg.Coordinator.TxSelector.Path, stateDB, l2DB)
326 if err != nil {
327 return nil, tracerr.Wrap(err)
328 }
329 batchBuilder, err := batchbuilder.NewBatchBuilder(cfg.Coordinator.BatchBuilder.Path,
330 stateDB, 0, uint64(cfg.Coordinator.Circuit.NLevels))
331 if err != nil {
332 return nil, tracerr.Wrap(err)
333 }
334 serverProofs := make([]prover.Client, len(cfg.Coordinator.ServerProofs))
335 for i, serverProofCfg := range cfg.Coordinator.ServerProofs {
336 serverProofs[i] = prover.NewProofServerClient(serverProofCfg.URL,
337 cfg.Coordinator.ProofServerPollInterval.Duration)
338 }
339
340 txProcessorCfg := txprocessor.Config{
341 NLevels: uint32(cfg.Coordinator.Circuit.NLevels),
342 MaxTx: uint32(cfg.Coordinator.Circuit.MaxTx),
343 ChainID: chainIDU16,
344 MaxFeeTx: common.RollupConstMaxFeeIdxCoordinator,
345 MaxL1Tx: common.RollupConstMaxL1Tx,
346 }
347 var verifierIdx int
348 if cfg.Coordinator.Debug.RollupVerifierIndex == nil {
349 verifierIdx, err = scConsts.Rollup.FindVerifierIdx(
350 cfg.Coordinator.Circuit.MaxTx,
351 cfg.Coordinator.Circuit.NLevels,
352 )
353 if err != nil {
354 return nil, tracerr.Wrap(err)
355 }
356 log.Infow("Found verifier that matches circuit config", "verifierIdx", verifierIdx)
357 } else {
358 verifierIdx = *cfg.Coordinator.Debug.RollupVerifierIndex
359 log.Infow("Using debug verifier index from config", "verifierIdx", verifierIdx)
360 if verifierIdx >= len(scConsts.Rollup.Verifiers) {
361 return nil, tracerr.Wrap(
362 fmt.Errorf("verifierIdx (%v) >= "+
363 "len(scConsts.Rollup.Verifiers) (%v)",
364 verifierIdx, len(scConsts.Rollup.Verifiers)))
365 }
366 verifier := scConsts.Rollup.Verifiers[verifierIdx]
367 if verifier.MaxTx != cfg.Coordinator.Circuit.MaxTx ||
368 verifier.NLevels != cfg.Coordinator.Circuit.NLevels {
369 return nil, tracerr.Wrap(
370 fmt.Errorf("Circuit config and verifier params don't match. "+
371 "circuit.MaxTx = %v, circuit.NLevels = %v, "+
372 "verifier.MaxTx = %v, verifier.NLevels = %v",
373 cfg.Coordinator.Circuit.MaxTx, cfg.Coordinator.Circuit.NLevels,
374 verifier.MaxTx, verifier.NLevels,
375 ))
376 }
377 }
378
379 coord, err = coordinator.NewCoordinator(
380 coordinator.Config{
381 ForgerAddress: cfg.Coordinator.ForgerAddress,
382 ConfirmBlocks: cfg.Coordinator.ConfirmBlocks,
383 L1BatchTimeoutPerc: cfg.Coordinator.L1BatchTimeoutPerc,
384 ForgeRetryInterval: cfg.Coordinator.ForgeRetryInterval.Duration,
385 ForgeDelay: cfg.Coordinator.ForgeDelay.Duration,
386 MustForgeAtSlotDeadline: cfg.Coordinator.MustForgeAtSlotDeadline,
387 IgnoreSlotCommitment: cfg.Coordinator.IgnoreSlotCommitment,
388 ForgeOncePerSlotIfTxs: cfg.Coordinator.ForgeOncePerSlotIfTxs,
389 ForgeNoTxsDelay: cfg.Coordinator.ForgeNoTxsDelay.Duration,
390 SyncRetryInterval: cfg.Coordinator.SyncRetryInterval.Duration,
391 PurgeByExtDelInterval: cfg.Coordinator.PurgeByExtDelInterval.Duration,
392 EthClientAttempts: cfg.Coordinator.EthClient.Attempts,
393 EthClientAttemptsDelay: cfg.Coordinator.EthClient.AttemptsDelay.Duration,
394 EthNoReuseNonce: cfg.Coordinator.EthClient.NoReuseNonce,
395 EthTxResendTimeout: cfg.Coordinator.EthClient.TxResendTimeout.Duration,
396 MaxGasPrice: cfg.Coordinator.EthClient.MaxGasPrice,
397 GasPriceIncPerc: cfg.Coordinator.EthClient.GasPriceIncPerc,
398 TxManagerCheckInterval: cfg.Coordinator.EthClient.CheckLoopInterval.Duration,
399 DebugBatchPath: cfg.Coordinator.Debug.BatchPath,
400 Purger: coordinator.PurgerCfg{
401 PurgeBatchDelay: cfg.Coordinator.L2DB.PurgeBatchDelay,
402 InvalidateBatchDelay: cfg.Coordinator.L2DB.InvalidateBatchDelay,
403 PurgeBlockDelay: cfg.Coordinator.L2DB.PurgeBlockDelay,
404 InvalidateBlockDelay: cfg.Coordinator.L2DB.InvalidateBlockDelay,
405 },
406 ForgeBatchGasCost: cfg.Coordinator.EthClient.ForgeBatchGasCost,
407 VerifierIdx: uint8(verifierIdx),
408 TxProcessorConfig: txProcessorCfg,
409 },
410 historyDB,
411 l2DB,
412 txSelector,
413 batchBuilder,
414 serverProofs,
415 client,
416 &scConsts,
417 initSCVars,
418 )
419 if err != nil {
420 return nil, tracerr.Wrap(err)
421 }
422 }
423 var nodeAPI *NodeAPI
424 if cfg.API.Address != "" {
425 if cfg.Debug.GinDebugMode {
426 gin.SetMode(gin.DebugMode)
427 } else {
428 gin.SetMode(gin.ReleaseMode)
429 }
430 if cfg.API.UpdateMetricsInterval.Duration == 0 {
431 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateMetricsInterval: %v",
432 cfg.API.UpdateMetricsInterval.Duration))
433 }
434 if cfg.API.UpdateRecommendedFeeInterval.Duration == 0 {
435 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateRecommendedFeeInterval: %v",
436 cfg.API.UpdateRecommendedFeeInterval.Duration))
437 }
438 server := gin.Default()
439 coord := false
440 if mode == ModeCoordinator {
441 coord = cfg.Coordinator.API.Coordinator
442 }
443 var err error
444 nodeAPI, err = NewNodeAPI(
445 cfg.API.Address,
446 coord, cfg.API.Explorer,
447 server,
448 historyDB,
449 l2DB,
450 )
451 if err != nil {
452 return nil, tracerr.Wrap(err)
453 }
454 }
455 var debugAPI *debugapi.DebugAPI
456 if cfg.Debug.APIAddress != "" {
457 debugAPI = debugapi.NewDebugAPI(cfg.Debug.APIAddress, stateDB, sync)
458 }
459 priceUpdater, err := priceupdater.NewPriceUpdater(
460 cfg.PriceUpdater.DefaultUpdateMethod,
461 cfg.PriceUpdater.TokensConfig,
462 historyDB,
463 cfg.PriceUpdater.URLBitfinexV2,
464 cfg.PriceUpdater.URLCoinGeckoV3,
465 )
466 if err != nil {
467 return nil, tracerr.Wrap(err)
468 }
469 ctx, cancel := context.WithCancel(context.Background())
470 return &Node{
471 stateAPIUpdater: stateAPIUpdater,
472 nodeAPI: nodeAPI,
473 debugAPI: debugAPI,
474 priceUpdater: priceUpdater,
475 coord: coord,
476 sync: sync,
477 cfg: cfg,
478 mode: mode,
479 sqlConnRead: dbRead,
480 sqlConnWrite: dbWrite,
481 historyDB: historyDB,
482 ctx: ctx,
483 cancel: cancel,
484 }, nil
485}
486
487// APIServer is a server that only runs the API
488type APIServer struct {
489 nodeAPI *NodeAPI
490 mode Mode
491 ctx context.Context
492 wg sync.WaitGroup
493 cancel context.CancelFunc
494}
495
496// NewAPIServer creates a new APIServer
497func NewAPIServer(mode Mode, cfg *config.APIServer) (*APIServer, error) {
498 meddler.Debug = cfg.Debug.MeddlerLogs
499 // Stablish DB connection
500 dbWrite, err := dbUtils.InitSQLDB(
501 cfg.PostgreSQL.PortWrite,
502 cfg.PostgreSQL.HostWrite,
503 cfg.PostgreSQL.UserWrite,
504 cfg.PostgreSQL.PasswordWrite,
505 cfg.PostgreSQL.NameWrite,
506 )
507 if err != nil {
508 return nil, tracerr.Wrap(fmt.Errorf("dbUtils.InitSQLDB: %w", err))
509 }
510 var dbRead *sqlx.DB
511 if cfg.PostgreSQL.HostRead == "" {
512 dbRead = dbWrite
513 } else if cfg.PostgreSQL.HostRead == cfg.PostgreSQL.HostWrite {
514 return nil, tracerr.Wrap(fmt.Errorf(
515 "PostgreSQL.HostRead and PostgreSQL.HostWrite must be different",
516 ))
517 } else {
518 dbRead, err = dbUtils.InitSQLDB(
519 cfg.PostgreSQL.PortRead,
520 cfg.PostgreSQL.HostRead,
521 cfg.PostgreSQL.UserRead,
522 cfg.PostgreSQL.PasswordRead,
523 cfg.PostgreSQL.NameRead,
524 )
525 if err != nil {
526 return nil, tracerr.Wrap(fmt.Errorf("dbUtils.InitSQLDB: %w", err))
527 }
528 }
529 apiConnCon := dbUtils.NewAPIConnectionController(
530 cfg.API.MaxSQLConnections,
531 cfg.API.SQLConnectionTimeout.Duration,
532 )
533
534 historyDB := historydb.NewHistoryDB(dbRead, dbWrite, apiConnCon)
535
536 var l2DB *l2db.L2DB
537 if mode == ModeCoordinator {
538 l2DB = l2db.NewL2DB(
539 dbRead, dbWrite,
540 0,
541 cfg.Coordinator.L2DB.MaxTxs,
542 cfg.Coordinator.L2DB.MinFeeUSD,
543 cfg.Coordinator.L2DB.MaxFeeUSD,
544 0,
545 apiConnCon,
546 )
547 }
548
549 if cfg.Debug.GinDebugMode {
550 gin.SetMode(gin.DebugMode)
551 } else {
552 gin.SetMode(gin.ReleaseMode)
553 }
554 server := gin.Default()
555 coord := false
556 if mode == ModeCoordinator {
557 coord = cfg.Coordinator.API.Coordinator
558 }
559 nodeAPI, err := NewNodeAPI(
560 cfg.API.Address,
561 coord, cfg.API.Explorer,
562 server,
563 historyDB,
564 l2DB,
565 )
566 if err != nil {
567 return nil, tracerr.Wrap(err)
568 }
569 ctx, cancel := context.WithCancel(context.Background())
570 return &APIServer{
571 nodeAPI: nodeAPI,
572 mode: mode,
573 ctx: ctx,
574 cancel: cancel,
575 }, nil
576}
577
578// Start the APIServer
579func (s *APIServer) Start() {
580 log.Infow("Starting api server...", "mode", s.mode)
581 log.Info("Starting NodeAPI...")
582 s.wg.Add(1)
583 go func() {
584 defer func() {
585 log.Info("NodeAPI routine stopped")
586 s.wg.Done()
587 }()
588 if err := s.nodeAPI.Run(s.ctx); err != nil {
589 if s.ctx.Err() != nil {
590 return
591 }
592 log.Fatalw("NodeAPI.Run", "err", err)
593 }
594 }()
595}
596
597// Stop the APIServer
598func (s *APIServer) Stop() {
599 log.Infow("Stopping NodeAPI...")
600 s.cancel()
601 s.wg.Wait()
602}
603
604// NodeAPI holds the node http API
605type NodeAPI struct { //nolint:golint
606 api *api.API
607 engine *gin.Engine
608 addr string
609}
610
611func handleNoRoute(c *gin.Context) {
612 matched, _ := regexp.MatchString(`^/v[0-9]+/`, c.Request.URL.Path)
613 if !matched {
614 c.JSON(http.StatusNotFound, gin.H{
615 "error": "Version not provided, please provide a valid version in the path such as v1",
616 })
617 return
618 }
619 c.JSON(http.StatusNotFound, gin.H{
620 "error": "404 page not found",
621 })
622}
623
624// NewNodeAPI creates a new NodeAPI (which internally calls api.NewAPI)
625func NewNodeAPI(
626 addr string,
627 coordinatorEndpoints, explorerEndpoints bool,
628 server *gin.Engine,
629 hdb *historydb.HistoryDB,
630 l2db *l2db.L2DB,
631) (*NodeAPI, error) {
632 engine := gin.Default()
633 engine.NoRoute(handleNoRoute)
634 engine.Use(cors.Default())
635 _api, err := api.NewAPI(
636 coordinatorEndpoints, explorerEndpoints,
637 engine,
638 hdb,
639 l2db,
640 )
641 if err != nil {
642 return nil, tracerr.Wrap(err)
643 }
644 return &NodeAPI{
645 addr: addr,
646 api: _api,
647 engine: engine,
648 }, nil
649}
650
651// Run starts the http server of the NodeAPI. To stop it, pass a context
652// with cancellation.
653func (a *NodeAPI) Run(ctx context.Context) error {
654 server := &http.Server{
655 Handler: a.engine,
656 // TODO: Figure out best parameters for production
657 ReadTimeout: 30 * time.Second, //nolint:gomnd
658 WriteTimeout: 30 * time.Second, //nolint:gomnd
659 MaxHeaderBytes: 1 << 20, //nolint:gomnd
660 }
661 listener, err := net.Listen("tcp", a.addr)
662 if err != nil {
663 return tracerr.Wrap(err)
664 }
665 log.Infof("NodeAPI is ready at %v", a.addr)
666 go func() {
667 if err := server.Serve(listener); err != nil &&
668 tracerr.Unwrap(err) != http.ErrServerClosed {
669 log.Fatalf("Listen: %s\n", err)
670 }
671 }()
672
673 <-ctx.Done()
674 log.Info("Stopping NodeAPI...")
675 ctxTimeout, cancel := context.WithTimeout(context.Background(), 10*time.Second) //nolint:gomnd
676 defer cancel()
677 if err := server.Shutdown(ctxTimeout); err != nil {
678 return tracerr.Wrap(err)
679 }
680 log.Info("NodeAPI done")
681 return nil
682}
683
684func (n *Node) handleNewBlock(ctx context.Context, stats *synchronizer.Stats,
685 vars *common.SCVariablesPtr, batches []common.BatchData) error {
686 if n.mode == ModeCoordinator {
687 n.coord.SendMsg(ctx, coordinator.MsgSyncBlock{
688 Stats: *stats,
689 Vars: *vars,
690 Batches: batches,
691 })
692 }
693 n.stateAPIUpdater.SetSCVars(vars)
694 if stats.Synced() {
695 if err := n.stateAPIUpdater.UpdateNetworkInfo(
696 stats.Eth.LastBlock, stats.Sync.LastBlock,
697 common.BatchNum(stats.Eth.LastBatchNum),
698 stats.Sync.Auction.CurrentSlot.SlotNum,
699 ); err != nil {
700 log.Errorw("ApiStateUpdater.UpdateNetworkInfo", "err", err)
701 }
702 } else {
703 n.stateAPIUpdater.UpdateNetworkInfoBlock(
704 stats.Eth.LastBlock, stats.Sync.LastBlock,
705 )
706 }
707 if err := n.stateAPIUpdater.Store(); err != nil {
708 return tracerr.Wrap(err)
709 }
710 return nil
711}
712
713func (n *Node) handleReorg(ctx context.Context, stats *synchronizer.Stats,
714 vars *common.SCVariables) error {
715 if n.mode == ModeCoordinator {
716 n.coord.SendMsg(ctx, coordinator.MsgSyncReorg{
717 Stats: *stats,
718 Vars: *vars.AsPtr(),
719 })
720 }
721 n.stateAPIUpdater.SetSCVars(vars.AsPtr())
722 n.stateAPIUpdater.UpdateNetworkInfoBlock(
723 stats.Eth.LastBlock, stats.Sync.LastBlock,
724 )
725 if err := n.stateAPIUpdater.Store(); err != nil {
726 return tracerr.Wrap(err)
727 }
728 return nil
729}
730
731// TODO(Edu): Consider keeping the `lastBlock` inside synchronizer so that we
732// don't have to pass it around.
733func (n *Node) syncLoopFn(ctx context.Context, lastBlock *common.Block) (*common.Block,
734 time.Duration, error) {
735 blockData, discarded, err := n.sync.Sync(ctx, lastBlock)
736 stats := n.sync.Stats()
737 if err != nil {
738 // case: error
739 return nil, n.cfg.Synchronizer.SyncLoopInterval.Duration, tracerr.Wrap(err)
740 } else if discarded != nil {
741 // case: reorg
742 log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
743 vars := n.sync.SCVars()
744 if err := n.handleReorg(ctx, stats, vars); err != nil {
745 return nil, time.Duration(0), tracerr.Wrap(err)
746 }
747 return nil, time.Duration(0), nil
748 } else if blockData != nil {
749 // case: new block
750 vars := common.SCVariablesPtr{
751 Rollup: blockData.Rollup.Vars,
752 Auction: blockData.Auction.Vars,
753 WDelayer: blockData.WDelayer.Vars,
754 }
755 if err := n.handleNewBlock(ctx, stats, &vars, blockData.Rollup.Batches); err != nil {
756 return nil, time.Duration(0), tracerr.Wrap(err)
757 }
758 return &blockData.Block, time.Duration(0), nil
759 } else {
760 // case: no block
761 return lastBlock, n.cfg.Synchronizer.SyncLoopInterval.Duration, nil
762 }
763}
764
765// StartSynchronizer starts the synchronizer
766func (n *Node) StartSynchronizer() {
767 log.Info("Starting Synchronizer...")
768
769 // Trigger a manual call to handleNewBlock with the loaded state of the
770 // synchronizer in order to quickly activate the API and Coordinator
771 // and avoid waiting for the next block. Without this, the API and
772 // Coordinator will not react until the following block (starting from
773 // the last synced one) is synchronized
774 stats := n.sync.Stats()
775 vars := n.sync.SCVars()
776 if err := n.handleNewBlock(n.ctx, stats, vars.AsPtr(), []common.BatchData{}); err != nil {
777 log.Fatalw("Node.handleNewBlock", "err", err)
778 }
779
780 n.wg.Add(1)
781 go func() {
782 var err error
783 var lastBlock *common.Block
784 waitDuration := time.Duration(0)
785 for {
786 select {
787 case <-n.ctx.Done():
788 log.Info("Synchronizer done")
789 n.wg.Done()
790 return
791 case <-time.After(waitDuration):
792 if lastBlock, waitDuration, err = n.syncLoopFn(n.ctx,
793 lastBlock); err != nil {
794 if n.ctx.Err() != nil {
795 continue
796 }
797 if errors.Is(err, eth.ErrBlockHashMismatchEvent) {
798 log.Warnw("Synchronizer.Sync", "err", err)
799 } else if errors.Is(err, synchronizer.ErrUnknownBlock) {
800 log.Warnw("Synchronizer.Sync", "err", err)
801 } else {
802 log.Errorw("Synchronizer.Sync", "err", err)
803 }
804 }
805 }
806 }
807 }()
808
809 n.wg.Add(1)
810 go func() {
811 for {
812 select {
813 case <-n.ctx.Done():
814 log.Info("PriceUpdater done")
815 n.wg.Done()
816 return
817 case <-time.After(n.cfg.PriceUpdater.Interval.Duration):
818 if err := n.priceUpdater.UpdateTokenList(); err != nil {
819 log.Errorw("PriceUpdater.UpdateTokenList()", "err", err)
820 }
821 n.priceUpdater.UpdatePrices(n.ctx)
822 }
823 }
824 }()
825}
826
827// StartDebugAPI starts the DebugAPI
828func (n *Node) StartDebugAPI() {
829 log.Info("Starting DebugAPI...")
830 n.wg.Add(1)
831 go func() {
832 defer func() {
833 log.Info("DebugAPI routine stopped")
834 n.wg.Done()
835 }()
836 if err := n.debugAPI.Run(n.ctx); err != nil {
837 if n.ctx.Err() != nil {
838 return
839 }
840 log.Fatalw("DebugAPI.Run", "err", err)
841 }
842 }()
843}
844
845// StartNodeAPI starts the NodeAPI
846func (n *Node) StartNodeAPI() {
847 log.Info("Starting NodeAPI...")
848 n.wg.Add(1)
849 go func() {
850 defer func() {
851 log.Info("NodeAPI routine stopped")
852 n.wg.Done()
853 }()
854 if err := n.nodeAPI.Run(n.ctx); err != nil {
855 if n.ctx.Err() != nil {
856 return
857 }
858 log.Fatalw("NodeAPI.Run", "err", err)
859 }
860 }()
861
862 n.wg.Add(1)
863 go func() {
864 // Do an initial update on startup
865 if err := n.stateAPIUpdater.UpdateMetrics(); err != nil {
866 log.Errorw("ApiStateUpdater.UpdateMetrics", "err", err)
867 }
868 if err := n.stateAPIUpdater.Store(); err != nil {
869 log.Errorw("ApiStateUpdater.Store", "err", err)
870 }
871 for {
872 select {
873 case <-n.ctx.Done():
874 log.Info("ApiStateUpdater.UpdateMetrics loop done")
875 n.wg.Done()
876 return
877 case <-time.After(n.cfg.API.UpdateMetricsInterval.Duration):
878 if err := n.stateAPIUpdater.UpdateMetrics(); err != nil {
879 log.Errorw("ApiStateUpdater.UpdateMetrics", "err", err)
880 continue
881 }
882 if err := n.stateAPIUpdater.Store(); err != nil {
883 log.Errorw("ApiStateUpdater.Store", "err", err)
884 }
885 }
886 }
887 }()
888
889 n.wg.Add(1)
890 go func() {
891 // Do an initial update on startup
892 if err := n.stateAPIUpdater.UpdateRecommendedFee(); err != nil {
893 log.Errorw("ApiStateUpdater.UpdateRecommendedFee", "err", err)
894 }
895 if err := n.stateAPIUpdater.Store(); err != nil {
896 log.Errorw("ApiStateUpdater.Store", "err", err)
897 }
898 for {
899 select {
900 case <-n.ctx.Done():
901 log.Info("ApiStateUpdaterAPI.UpdateRecommendedFee loop done")
902 n.wg.Done()
903 return
904 case <-time.After(n.cfg.API.UpdateRecommendedFeeInterval.Duration):
905 if err := n.stateAPIUpdater.UpdateRecommendedFee(); err != nil {
906 log.Errorw("ApiStateUpdaterAPI.UpdateRecommendedFee", "err", err)
907 continue
908 }
909 if err := n.stateAPIUpdater.Store(); err != nil {
910 log.Errorw("ApiStateUpdater.Store", "err", err)
911 }
912 }
913 }
914 }()
915}
916
917// Start the node
918func (n *Node) Start() {
919 log.Infow("Starting node...", "mode", n.mode)
920 if n.debugAPI != nil {
921 n.StartDebugAPI()
922 }
923 if n.nodeAPI != nil {
924 n.StartNodeAPI()
925 }
926 if n.mode == ModeCoordinator {
927 log.Info("Starting Coordinator...")
928 n.coord.Start()
929 }
930 n.StartSynchronizer()
931}
932
933// Stop the node
934func (n *Node) Stop() {
935 log.Infow("Stopping node...")
936 n.cancel()
937 n.wg.Wait()
938 if n.mode == ModeCoordinator {
939 log.Info("Stopping Coordinator...")
940 n.coord.Stop()
941 }
942 // Close kv DBs
943 n.sync.StateDB().Close()
944 if n.mode == ModeCoordinator {
945 n.coord.TxSelector().LocalAccountsDB().Close()
946 n.coord.BatchBuilder().LocalStateDB().Close()
947 }
948}