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