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 ForgeOncePerSlotIfTxs: cfg.Coordinator.ForgeOncePerSlotIfTxs,
371 ForgeNoTxsDelay: cfg.Coordinator.ForgeNoTxsDelay.Duration,
372 SyncRetryInterval: cfg.Coordinator.SyncRetryInterval.Duration,
373 PurgeByExtDelInterval: cfg.Coordinator.PurgeByExtDelInterval.Duration,
374 EthClientAttempts: cfg.Coordinator.EthClient.Attempts,
375 EthClientAttemptsDelay: cfg.Coordinator.EthClient.AttemptsDelay.Duration,
376 EthNoReuseNonce: cfg.Coordinator.EthClient.NoReuseNonce,
377 EthTxResendTimeout: cfg.Coordinator.EthClient.TxResendTimeout.Duration,
378 MaxGasPrice: cfg.Coordinator.EthClient.MaxGasPrice,
379 GasPriceIncPerc: cfg.Coordinator.EthClient.GasPriceIncPerc,
380 TxManagerCheckInterval: cfg.Coordinator.EthClient.CheckLoopInterval.Duration,
381 DebugBatchPath: cfg.Coordinator.Debug.BatchPath,
382 Purger: coordinator.PurgerCfg{
383 PurgeBatchDelay: cfg.Coordinator.L2DB.PurgeBatchDelay,
384 InvalidateBatchDelay: cfg.Coordinator.L2DB.InvalidateBatchDelay,
385 PurgeBlockDelay: cfg.Coordinator.L2DB.PurgeBlockDelay,
386 InvalidateBlockDelay: cfg.Coordinator.L2DB.InvalidateBlockDelay,
387 },
388 ForgeBatchGasCost: cfg.Coordinator.EthClient.ForgeBatchGasCost,
389 VerifierIdx: uint8(verifierIdx),
390 TxProcessorConfig: txProcessorCfg,
391 },
392 historyDB,
393 l2DB,
394 txSelector,
395 batchBuilder,
396 serverProofs,
397 client,
398 &scConsts,
399 initSCVars,
400 )
401 if err != nil {
402 return nil, tracerr.Wrap(err)
403 }
404 }
405 var nodeAPI *NodeAPI
406 if cfg.API.Address != "" {
407 if cfg.Debug.GinDebugMode {
408 gin.SetMode(gin.DebugMode)
409 } else {
410 gin.SetMode(gin.ReleaseMode)
411 }
412 if cfg.API.UpdateMetricsInterval.Duration == 0 {
413 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateMetricsInterval: %v",
414 cfg.API.UpdateMetricsInterval.Duration))
415 }
416 if cfg.API.UpdateRecommendedFeeInterval.Duration == 0 {
417 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateRecommendedFeeInterval: %v",
418 cfg.API.UpdateRecommendedFeeInterval.Duration))
419 }
420 server := gin.Default()
421 coord := false
422 if mode == ModeCoordinator {
423 coord = cfg.Coordinator.API.Coordinator
424 }
425 var err error
426 nodeAPI, err = NewNodeAPI(
427 cfg.API.Address,
428 coord, cfg.API.Explorer,
429 server,
430 historyDB,
431 l2DB,
432 )
433 if err != nil {
434 return nil, tracerr.Wrap(err)
435 }
436 }
437 var debugAPI *debugapi.DebugAPI
438 if cfg.Debug.APIAddress != "" {
439 debugAPI = debugapi.NewDebugAPI(cfg.Debug.APIAddress, stateDB, sync)
440 }
441 priceUpdater, err := priceupdater.NewPriceUpdater(
442 cfg.PriceUpdater.DefaultUpdateMethod,
443 cfg.PriceUpdater.TokensConfig,
444 historyDB,
445 cfg.PriceUpdater.URLBitfinexV2,
446 cfg.PriceUpdater.URLCoinGeckoV3,
447 )
448 if err != nil {
449 return nil, tracerr.Wrap(err)
450 }
451 ctx, cancel := context.WithCancel(context.Background())
452 return &Node{
453 stateAPIUpdater: stateAPIUpdater,
454 nodeAPI: nodeAPI,
455 debugAPI: debugAPI,
456 priceUpdater: priceUpdater,
457 coord: coord,
458 sync: sync,
459 cfg: cfg,
460 mode: mode,
461 sqlConnRead: dbRead,
462 sqlConnWrite: dbWrite,
463 historyDB: historyDB,
464 ctx: ctx,
465 cancel: cancel,
466 }, nil
467}
468
469// APIServer is a server that only runs the API
470type APIServer struct {
471 nodeAPI *NodeAPI
472 mode Mode
473 ctx context.Context
474 wg sync.WaitGroup
475 cancel context.CancelFunc
476}
477
478// NewAPIServer creates a new APIServer
479func NewAPIServer(mode Mode, cfg *config.APIServer) (*APIServer, error) {
480 meddler.Debug = cfg.Debug.MeddlerLogs
481 // Stablish DB connection
482 dbWrite, err := dbUtils.InitSQLDB(
483 cfg.PostgreSQL.PortWrite,
484 cfg.PostgreSQL.HostWrite,
485 cfg.PostgreSQL.UserWrite,
486 cfg.PostgreSQL.PasswordWrite,
487 cfg.PostgreSQL.NameWrite,
488 )
489 if err != nil {
490 return nil, tracerr.Wrap(fmt.Errorf("dbUtils.InitSQLDB: %w", err))
491 }
492 var dbRead *sqlx.DB
493 if cfg.PostgreSQL.HostRead == "" {
494 dbRead = dbWrite
495 } else if cfg.PostgreSQL.HostRead == cfg.PostgreSQL.HostWrite {
496 return nil, tracerr.Wrap(fmt.Errorf(
497 "PostgreSQL.HostRead and PostgreSQL.HostWrite must be different",
498 ))
499 } else {
500 dbRead, err = dbUtils.InitSQLDB(
501 cfg.PostgreSQL.PortRead,
502 cfg.PostgreSQL.HostRead,
503 cfg.PostgreSQL.UserRead,
504 cfg.PostgreSQL.PasswordRead,
505 cfg.PostgreSQL.NameRead,
506 )
507 if err != nil {
508 return nil, tracerr.Wrap(fmt.Errorf("dbUtils.InitSQLDB: %w", err))
509 }
510 }
511 apiConnCon := dbUtils.NewAPIConnectionController(
512 cfg.API.MaxSQLConnections,
513 cfg.API.SQLConnectionTimeout.Duration,
514 )
515
516 historyDB := historydb.NewHistoryDB(dbRead, dbWrite, apiConnCon)
517
518 var l2DB *l2db.L2DB
519 if mode == ModeCoordinator {
520 l2DB = l2db.NewL2DB(
521 dbRead, dbWrite,
522 0,
523 cfg.Coordinator.L2DB.MaxTxs,
524 cfg.Coordinator.L2DB.MinFeeUSD,
525 0,
526 apiConnCon,
527 )
528 }
529
530 if cfg.Debug.GinDebugMode {
531 gin.SetMode(gin.DebugMode)
532 } else {
533 gin.SetMode(gin.ReleaseMode)
534 }
535 server := gin.Default()
536 coord := false
537 if mode == ModeCoordinator {
538 coord = cfg.Coordinator.API.Coordinator
539 }
540 nodeAPI, err := NewNodeAPI(
541 cfg.API.Address,
542 coord, cfg.API.Explorer,
543 server,
544 historyDB,
545 l2DB,
546 )
547 if err != nil {
548 return nil, tracerr.Wrap(err)
549 }
550 ctx, cancel := context.WithCancel(context.Background())
551 return &APIServer{
552 nodeAPI: nodeAPI,
553 mode: mode,
554 ctx: ctx,
555 cancel: cancel,
556 }, nil
557}
558
559// Start the APIServer
560func (s *APIServer) Start() {
561 log.Infow("Starting api server...", "mode", s.mode)
562 log.Info("Starting NodeAPI...")
563 s.wg.Add(1)
564 go func() {
565 defer func() {
566 log.Info("NodeAPI routine stopped")
567 s.wg.Done()
568 }()
569 if err := s.nodeAPI.Run(s.ctx); err != nil {
570 if s.ctx.Err() != nil {
571 return
572 }
573 log.Fatalw("NodeAPI.Run", "err", err)
574 }
575 }()
576}
577
578// Stop the APIServer
579func (s *APIServer) Stop() {
580 log.Infow("Stopping NodeAPI...")
581 s.cancel()
582 s.wg.Wait()
583}
584
585// NodeAPI holds the node http API
586type NodeAPI struct { //nolint:golint
587 api *api.API
588 engine *gin.Engine
589 addr string
590}
591
592func handleNoRoute(c *gin.Context) {
593 c.JSON(http.StatusNotFound, gin.H{
594 "error": "404 page not found",
595 })
596}
597
598// NewNodeAPI creates a new NodeAPI (which internally calls api.NewAPI)
599func NewNodeAPI(
600 addr string,
601 coordinatorEndpoints, explorerEndpoints bool,
602 server *gin.Engine,
603 hdb *historydb.HistoryDB,
604 l2db *l2db.L2DB,
605) (*NodeAPI, error) {
606 engine := gin.Default()
607 engine.NoRoute(handleNoRoute)
608 engine.Use(cors.Default())
609 _api, err := api.NewAPI(
610 coordinatorEndpoints, explorerEndpoints,
611 engine,
612 hdb,
613 l2db,
614 )
615 if err != nil {
616 return nil, tracerr.Wrap(err)
617 }
618 return &NodeAPI{
619 addr: addr,
620 api: _api,
621 engine: engine,
622 }, nil
623}
624
625// Run starts the http server of the NodeAPI. To stop it, pass a context
626// with cancellation.
627func (a *NodeAPI) Run(ctx context.Context) error {
628 server := &http.Server{
629 Handler: a.engine,
630 // TODO: Figure out best parameters for production
631 ReadTimeout: 30 * time.Second, //nolint:gomnd
632 WriteTimeout: 30 * time.Second, //nolint:gomnd
633 MaxHeaderBytes: 1 << 20, //nolint:gomnd
634 }
635 listener, err := net.Listen("tcp", a.addr)
636 if err != nil {
637 return tracerr.Wrap(err)
638 }
639 log.Infof("NodeAPI is ready at %v", a.addr)
640 go func() {
641 if err := server.Serve(listener); err != nil &&
642 tracerr.Unwrap(err) != http.ErrServerClosed {
643 log.Fatalf("Listen: %s\n", err)
644 }
645 }()
646
647 <-ctx.Done()
648 log.Info("Stopping NodeAPI...")
649 ctxTimeout, cancel := context.WithTimeout(context.Background(), 10*time.Second) //nolint:gomnd
650 defer cancel()
651 if err := server.Shutdown(ctxTimeout); err != nil {
652 return tracerr.Wrap(err)
653 }
654 log.Info("NodeAPI done")
655 return nil
656}
657
658func (n *Node) handleNewBlock(ctx context.Context, stats *synchronizer.Stats,
659 vars *common.SCVariablesPtr, batches []common.BatchData) error {
660 if n.mode == ModeCoordinator {
661 n.coord.SendMsg(ctx, coordinator.MsgSyncBlock{
662 Stats: *stats,
663 Vars: *vars,
664 Batches: batches,
665 })
666 }
667 n.stateAPIUpdater.SetSCVars(vars)
668 if stats.Synced() {
669 if err := n.stateAPIUpdater.UpdateNetworkInfo(
670 stats.Eth.LastBlock, stats.Sync.LastBlock,
671 common.BatchNum(stats.Eth.LastBatchNum),
672 stats.Sync.Auction.CurrentSlot.SlotNum,
673 ); err != nil {
674 log.Errorw("ApiStateUpdater.UpdateNetworkInfo", "err", err)
675 }
676 } else {
677 n.stateAPIUpdater.UpdateNetworkInfoBlock(
678 stats.Eth.LastBlock, stats.Sync.LastBlock,
679 )
680 }
681 if err := n.stateAPIUpdater.Store(); err != nil {
682 return tracerr.Wrap(err)
683 }
684 return nil
685}
686
687func (n *Node) handleReorg(ctx context.Context, stats *synchronizer.Stats,
688 vars *common.SCVariables) error {
689 if n.mode == ModeCoordinator {
690 n.coord.SendMsg(ctx, coordinator.MsgSyncReorg{
691 Stats: *stats,
692 Vars: *vars.AsPtr(),
693 })
694 }
695 n.stateAPIUpdater.SetSCVars(vars.AsPtr())
696 n.stateAPIUpdater.UpdateNetworkInfoBlock(
697 stats.Eth.LastBlock, stats.Sync.LastBlock,
698 )
699 if err := n.stateAPIUpdater.Store(); err != nil {
700 return tracerr.Wrap(err)
701 }
702 return nil
703}
704
705// TODO(Edu): Consider keeping the `lastBlock` inside synchronizer so that we
706// don't have to pass it around.
707func (n *Node) syncLoopFn(ctx context.Context, lastBlock *common.Block) (*common.Block,
708 time.Duration, error) {
709 blockData, discarded, err := n.sync.Sync(ctx, lastBlock)
710 stats := n.sync.Stats()
711 if err != nil {
712 // case: error
713 return nil, n.cfg.Synchronizer.SyncLoopInterval.Duration, tracerr.Wrap(err)
714 } else if discarded != nil {
715 // case: reorg
716 log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
717 vars := n.sync.SCVars()
718 if err := n.handleReorg(ctx, stats, vars); err != nil {
719 return nil, time.Duration(0), tracerr.Wrap(err)
720 }
721 return nil, time.Duration(0), nil
722 } else if blockData != nil {
723 // case: new block
724 vars := common.SCVariablesPtr{
725 Rollup: blockData.Rollup.Vars,
726 Auction: blockData.Auction.Vars,
727 WDelayer: blockData.WDelayer.Vars,
728 }
729 if err := n.handleNewBlock(ctx, stats, &vars, blockData.Rollup.Batches); err != nil {
730 return nil, time.Duration(0), tracerr.Wrap(err)
731 }
732 return &blockData.Block, time.Duration(0), nil
733 } else {
734 // case: no block
735 return lastBlock, n.cfg.Synchronizer.SyncLoopInterval.Duration, nil
736 }
737}
738
739// StartSynchronizer starts the synchronizer
740func (n *Node) StartSynchronizer() {
741 log.Info("Starting Synchronizer...")
742
743 // Trigger a manual call to handleNewBlock with the loaded state of the
744 // synchronizer in order to quickly activate the API and Coordinator
745 // and avoid waiting for the next block. Without this, the API and
746 // Coordinator will not react until the following block (starting from
747 // the last synced one) is synchronized
748 stats := n.sync.Stats()
749 vars := n.sync.SCVars()
750 if err := n.handleNewBlock(n.ctx, stats, vars.AsPtr(), []common.BatchData{}); err != nil {
751 log.Fatalw("Node.handleNewBlock", "err", err)
752 }
753
754 n.wg.Add(1)
755 go func() {
756 var err error
757 var lastBlock *common.Block
758 waitDuration := time.Duration(0)
759 for {
760 select {
761 case <-n.ctx.Done():
762 log.Info("Synchronizer done")
763 n.wg.Done()
764 return
765 case <-time.After(waitDuration):
766 if lastBlock, waitDuration, err = n.syncLoopFn(n.ctx,
767 lastBlock); err != nil {
768 if n.ctx.Err() != nil {
769 continue
770 }
771 if errors.Is(err, eth.ErrBlockHashMismatchEvent) {
772 log.Warnw("Synchronizer.Sync", "err", err)
773 } else if errors.Is(err, synchronizer.ErrUnknownBlock) {
774 log.Warnw("Synchronizer.Sync", "err", err)
775 } else {
776 log.Errorw("Synchronizer.Sync", "err", err)
777 }
778 }
779 }
780 }
781 }()
782
783 n.wg.Add(1)
784 go func() {
785 for {
786 select {
787 case <-n.ctx.Done():
788 log.Info("PriceUpdater done")
789 n.wg.Done()
790 return
791 case <-time.After(n.cfg.PriceUpdater.Interval.Duration):
792 if err := n.priceUpdater.UpdateTokenList(); err != nil {
793 log.Errorw("PriceUpdater.UpdateTokenList()", "err", err)
794 }
795 n.priceUpdater.UpdatePrices(n.ctx)
796 }
797 }
798 }()
799}
800
801// StartDebugAPI starts the DebugAPI
802func (n *Node) StartDebugAPI() {
803 log.Info("Starting DebugAPI...")
804 n.wg.Add(1)
805 go func() {
806 defer func() {
807 log.Info("DebugAPI routine stopped")
808 n.wg.Done()
809 }()
810 if err := n.debugAPI.Run(n.ctx); err != nil {
811 if n.ctx.Err() != nil {
812 return
813 }
814 log.Fatalw("DebugAPI.Run", "err", err)
815 }
816 }()
817}
818
819// StartNodeAPI starts the NodeAPI
820func (n *Node) StartNodeAPI() {
821 log.Info("Starting NodeAPI...")
822 n.wg.Add(1)
823 go func() {
824 defer func() {
825 log.Info("NodeAPI routine stopped")
826 n.wg.Done()
827 }()
828 if err := n.nodeAPI.Run(n.ctx); err != nil {
829 if n.ctx.Err() != nil {
830 return
831 }
832 log.Fatalw("NodeAPI.Run", "err", err)
833 }
834 }()
835
836 n.wg.Add(1)
837 go func() {
838 // Do an initial update on startup
839 if err := n.stateAPIUpdater.UpdateMetrics(); err != nil {
840 log.Errorw("ApiStateUpdater.UpdateMetrics", "err", err)
841 }
842 if err := n.stateAPIUpdater.Store(); err != nil {
843 log.Errorw("ApiStateUpdater.Store", "err", err)
844 }
845 for {
846 select {
847 case <-n.ctx.Done():
848 log.Info("ApiStateUpdater.UpdateMetrics loop done")
849 n.wg.Done()
850 return
851 case <-time.After(n.cfg.API.UpdateMetricsInterval.Duration):
852 if err := n.stateAPIUpdater.UpdateMetrics(); err != nil {
853 log.Errorw("ApiStateUpdater.UpdateMetrics", "err", err)
854 continue
855 }
856 if err := n.stateAPIUpdater.Store(); err != nil {
857 log.Errorw("ApiStateUpdater.Store", "err", err)
858 }
859 }
860 }
861 }()
862
863 n.wg.Add(1)
864 go func() {
865 // Do an initial update on startup
866 if err := n.stateAPIUpdater.UpdateRecommendedFee(); err != nil {
867 log.Errorw("ApiStateUpdater.UpdateRecommendedFee", "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("ApiStateUpdaterAPI.UpdateRecommendedFee loop done")
876 n.wg.Done()
877 return
878 case <-time.After(n.cfg.API.UpdateRecommendedFeeInterval.Duration):
879 if err := n.stateAPIUpdater.UpdateRecommendedFee(); err != nil {
880 log.Errorw("ApiStateUpdaterAPI.UpdateRecommendedFee", "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
891// Start the node
892func (n *Node) Start() {
893 log.Infow("Starting node...", "mode", n.mode)
894 if n.debugAPI != nil {
895 n.StartDebugAPI()
896 }
897 if n.nodeAPI != nil {
898 n.StartNodeAPI()
899 }
900 if n.mode == ModeCoordinator {
901 log.Info("Starting Coordinator...")
902 n.coord.Start()
903 }
904 n.StartSynchronizer()
905}
906
907// Stop the node
908func (n *Node) Stop() {
909 log.Infow("Stopping node...")
910 n.cancel()
911 n.wg.Wait()
912 if n.mode == ModeCoordinator {
913 log.Info("Stopping Coordinator...")
914 n.coord.Stop()
915 }
916 // Close kv DBs
917 n.sync.StateDB().Close()
918 if n.mode == ModeCoordinator {
919 n.coord.TxSelector().LocalAccountsDB().Close()
920 n.coord.BatchBuilder().LocalStateDB().Close()
921 }
922}