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
619// NewNodeAPI creates a new NodeAPI (which internally calls api.NewAPI)
620func NewNodeAPI(
621 addr string,
622 coordinatorEndpoints, explorerEndpoints bool,
623 server *gin.Engine,
624 hdb *historydb.HistoryDB,
625 l2db *l2db.L2DB,
626) (*NodeAPI, error) {
627 engine := gin.Default()
628 engine.Use(cors.Default())
629 _api, err := api.NewAPI(
630 coordinatorEndpoints, explorerEndpoints,
631 engine,
632 hdb,
633 l2db,
634 )
635 if err != nil {
636 return nil, tracerr.Wrap(err)
637 }
638 return &NodeAPI{
639 addr: addr,
640 api: _api,
641 engine: engine,
642 }, nil
643}
644
645// Run starts the http server of the NodeAPI. To stop it, pass a context
646// with cancellation.
647func (a *NodeAPI) Run(ctx context.Context) error {
648 server := &http.Server{
649 Handler: a.engine,
650 // TODO: Figure out best parameters for production
651 ReadTimeout: 30 * time.Second, //nolint:gomnd
652 WriteTimeout: 30 * time.Second, //nolint:gomnd
653 MaxHeaderBytes: 1 << 20, //nolint:gomnd
654 }
655 listener, err := net.Listen("tcp", a.addr)
656 if err != nil {
657 return tracerr.Wrap(err)
658 }
659 log.Infof("NodeAPI is ready at %v", a.addr)
660 go func() {
661 if err := server.Serve(listener); err != nil &&
662 tracerr.Unwrap(err) != http.ErrServerClosed {
663 log.Fatalf("Listen: %s\n", err)
664 }
665 }()
666
667 <-ctx.Done()
668 log.Info("Stopping NodeAPI...")
669 ctxTimeout, cancel := context.WithTimeout(context.Background(), 10*time.Second) //nolint:gomnd
670 defer cancel()
671 if err := server.Shutdown(ctxTimeout); err != nil {
672 return tracerr.Wrap(err)
673 }
674 log.Info("NodeAPI done")
675 return nil
676}
677
678func (n *Node) handleNewBlock(ctx context.Context, stats *synchronizer.Stats,
679 vars *common.SCVariablesPtr, batches []common.BatchData) error {
680 if n.mode == ModeCoordinator {
681 n.coord.SendMsg(ctx, coordinator.MsgSyncBlock{
682 Stats: *stats,
683 Vars: *vars,
684 Batches: batches,
685 })
686 }
687 n.stateAPIUpdater.SetSCVars(vars)
688 if stats.Synced() {
689 if err := n.stateAPIUpdater.UpdateNetworkInfo(
690 stats.Eth.LastBlock, stats.Sync.LastBlock,
691 common.BatchNum(stats.Eth.LastBatchNum),
692 stats.Sync.Auction.CurrentSlot.SlotNum,
693 ); err != nil {
694 log.Errorw("ApiStateUpdater.UpdateNetworkInfo", "err", err)
695 }
696 } else {
697 n.stateAPIUpdater.UpdateNetworkInfoBlock(
698 stats.Eth.LastBlock, stats.Sync.LastBlock,
699 )
700 }
701 if err := n.stateAPIUpdater.Store(); err != nil {
702 return tracerr.Wrap(err)
703 }
704 return nil
705}
706
707func (n *Node) handleReorg(ctx context.Context, stats *synchronizer.Stats,
708 vars *common.SCVariables) error {
709 if n.mode == ModeCoordinator {
710 n.coord.SendMsg(ctx, coordinator.MsgSyncReorg{
711 Stats: *stats,
712 Vars: *vars.AsPtr(),
713 })
714 }
715 n.stateAPIUpdater.SetSCVars(vars.AsPtr())
716 n.stateAPIUpdater.UpdateNetworkInfoBlock(
717 stats.Eth.LastBlock, stats.Sync.LastBlock,
718 )
719 if err := n.stateAPIUpdater.Store(); err != nil {
720 return tracerr.Wrap(err)
721 }
722 return nil
723}
724
725// TODO(Edu): Consider keeping the `lastBlock` inside synchronizer so that we
726// don't have to pass it around.
727func (n *Node) syncLoopFn(ctx context.Context, lastBlock *common.Block) (*common.Block,
728 time.Duration, error) {
729 blockData, discarded, err := n.sync.Sync(ctx, lastBlock)
730 stats := n.sync.Stats()
731 if err != nil {
732 // case: error
733 return nil, n.cfg.Synchronizer.SyncLoopInterval.Duration, tracerr.Wrap(err)
734 } else if discarded != nil {
735 // case: reorg
736 log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
737 vars := n.sync.SCVars()
738 if err := n.handleReorg(ctx, stats, vars); err != nil {
739 return nil, time.Duration(0), tracerr.Wrap(err)
740 }
741 return nil, time.Duration(0), nil
742 } else if blockData != nil {
743 // case: new block
744 vars := common.SCVariablesPtr{
745 Rollup: blockData.Rollup.Vars,
746 Auction: blockData.Auction.Vars,
747 WDelayer: blockData.WDelayer.Vars,
748 }
749 if err := n.handleNewBlock(ctx, stats, &vars, blockData.Rollup.Batches); err != nil {
750 return nil, time.Duration(0), tracerr.Wrap(err)
751 }
752 return &blockData.Block, time.Duration(0), nil
753 } else {
754 // case: no block
755 return lastBlock, n.cfg.Synchronizer.SyncLoopInterval.Duration, nil
756 }
757}
758
759// StartSynchronizer starts the synchronizer
760func (n *Node) StartSynchronizer() {
761 log.Info("Starting Synchronizer...")
762
763 // Trigger a manual call to handleNewBlock with the loaded state of the
764 // synchronizer in order to quickly activate the API and Coordinator
765 // and avoid waiting for the next block. Without this, the API and
766 // Coordinator will not react until the following block (starting from
767 // the last synced one) is synchronized
768 stats := n.sync.Stats()
769 vars := n.sync.SCVars()
770 if err := n.handleNewBlock(n.ctx, stats, vars.AsPtr(), []common.BatchData{}); err != nil {
771 log.Fatalw("Node.handleNewBlock", "err", err)
772 }
773
774 n.wg.Add(1)
775 go func() {
776 var err error
777 var lastBlock *common.Block
778 waitDuration := time.Duration(0)
779 for {
780 select {
781 case <-n.ctx.Done():
782 log.Info("Synchronizer done")
783 n.wg.Done()
784 return
785 case <-time.After(waitDuration):
786 if lastBlock, waitDuration, err = n.syncLoopFn(n.ctx,
787 lastBlock); err != nil {
788 if n.ctx.Err() != nil {
789 continue
790 }
791 if errors.Is(err, eth.ErrBlockHashMismatchEvent) {
792 log.Warnw("Synchronizer.Sync", "err", err)
793 } else if errors.Is(err, synchronizer.ErrUnknownBlock) {
794 log.Warnw("Synchronizer.Sync", "err", err)
795 } else {
796 log.Errorw("Synchronizer.Sync", "err", err)
797 }
798 }
799 }
800 }
801 }()
802
803 n.wg.Add(1)
804 go func() {
805 for {
806 select {
807 case <-n.ctx.Done():
808 log.Info("PriceUpdater done")
809 n.wg.Done()
810 return
811 case <-time.After(n.cfg.PriceUpdater.Interval.Duration):
812 if err := n.priceUpdater.UpdateTokenList(); err != nil {
813 log.Errorw("PriceUpdater.UpdateTokenList()", "err", err)
814 }
815 n.priceUpdater.UpdatePrices(n.ctx)
816 }
817 }
818 }()
819}
820
821// StartDebugAPI starts the DebugAPI
822func (n *Node) StartDebugAPI() {
823 log.Info("Starting DebugAPI...")
824 n.wg.Add(1)
825 go func() {
826 defer func() {
827 log.Info("DebugAPI routine stopped")
828 n.wg.Done()
829 }()
830 if err := n.debugAPI.Run(n.ctx); err != nil {
831 if n.ctx.Err() != nil {
832 return
833 }
834 log.Fatalw("DebugAPI.Run", "err", err)
835 }
836 }()
837}
838
839// StartNodeAPI starts the NodeAPI
840func (n *Node) StartNodeAPI() {
841 log.Info("Starting NodeAPI...")
842 n.wg.Add(1)
843 go func() {
844 defer func() {
845 log.Info("NodeAPI routine stopped")
846 n.wg.Done()
847 }()
848 if err := n.nodeAPI.Run(n.ctx); err != nil {
849 if n.ctx.Err() != nil {
850 return
851 }
852 log.Fatalw("NodeAPI.Run", "err", err)
853 }
854 }()
855
856 n.wg.Add(1)
857 go func() {
858 // Do an initial update on startup
859 if err := n.stateAPIUpdater.UpdateMetrics(); err != nil {
860 log.Errorw("ApiStateUpdater.UpdateMetrics", "err", err)
861 }
862 if err := n.stateAPIUpdater.Store(); err != nil {
863 log.Errorw("ApiStateUpdater.Store", "err", err)
864 }
865 for {
866 select {
867 case <-n.ctx.Done():
868 log.Info("ApiStateUpdater.UpdateMetrics loop done")
869 n.wg.Done()
870 return
871 case <-time.After(n.cfg.API.UpdateMetricsInterval.Duration):
872 if err := n.stateAPIUpdater.UpdateMetrics(); err != nil {
873 log.Errorw("ApiStateUpdater.UpdateMetrics", "err", err)
874 continue
875 }
876 if err := n.stateAPIUpdater.Store(); err != nil {
877 log.Errorw("ApiStateUpdater.Store", "err", err)
878 }
879 }
880 }
881 }()
882
883 n.wg.Add(1)
884 go func() {
885 // Do an initial update on startup
886 if err := n.stateAPIUpdater.UpdateRecommendedFee(); err != nil {
887 log.Errorw("ApiStateUpdater.UpdateRecommendedFee", "err", err)
888 }
889 if err := n.stateAPIUpdater.Store(); err != nil {
890 log.Errorw("ApiStateUpdater.Store", "err", err)
891 }
892 for {
893 select {
894 case <-n.ctx.Done():
895 log.Info("ApiStateUpdaterAPI.UpdateRecommendedFee loop done")
896 n.wg.Done()
897 return
898 case <-time.After(n.cfg.API.UpdateRecommendedFeeInterval.Duration):
899 if err := n.stateAPIUpdater.UpdateRecommendedFee(); err != nil {
900 log.Errorw("ApiStateUpdaterAPI.UpdateRecommendedFee", "err", err)
901 continue
902 }
903 if err := n.stateAPIUpdater.Store(); err != nil {
904 log.Errorw("ApiStateUpdater.Store", "err", err)
905 }
906 }
907 }
908 }()
909}
910
911// Start the node
912func (n *Node) Start() {
913 log.Infow("Starting node...", "mode", n.mode)
914 if n.debugAPI != nil {
915 n.StartDebugAPI()
916 }
917 if n.nodeAPI != nil {
918 n.StartNodeAPI()
919 }
920 if n.mode == ModeCoordinator {
921 log.Info("Starting Coordinator...")
922 n.coord.Start()
923 }
924 n.StartSynchronizer()
925}
926
927// Stop the node
928func (n *Node) Stop() {
929 log.Infow("Stopping node...")
930 n.cancel()
931 n.wg.Wait()
932 if n.mode == ModeCoordinator {
933 log.Info("Stopping Coordinator...")
934 n.coord.Stop()
935 }
936 // Close kv DBs
937 n.sync.StateDB().Close()
938 if n.mode == ModeCoordinator {
939 n.coord.TxSelector().LocalAccountsDB().Close()
940 n.coord.BatchBuilder().LocalStateDB().Close()
941 }
942}