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