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