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/api/stateapiupdater"
19 "github.com/hermeznetwork/hermez-node/batchbuilder"
20 "github.com/hermeznetwork/hermez-node/common"
21 "github.com/hermeznetwork/hermez-node/config"
22 "github.com/hermeznetwork/hermez-node/coordinator"
23 dbUtils "github.com/hermeznetwork/hermez-node/db"
24 "github.com/hermeznetwork/hermez-node/db/historydb"
25 "github.com/hermeznetwork/hermez-node/db/l2db"
26 "github.com/hermeznetwork/hermez-node/db/statedb"
27 "github.com/hermeznetwork/hermez-node/eth"
28 "github.com/hermeznetwork/hermez-node/log"
29 "github.com/hermeznetwork/hermez-node/priceupdater"
30 "github.com/hermeznetwork/hermez-node/prover"
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.MaxFeeUSD,
234 cfg.Coordinator.L2DB.TTL.Duration,
235 apiConnCon,
236 )
237 }
238
239 sync, err := synchronizer.NewSynchronizer(client, historyDB, l2DB, stateDB, synchronizer.Config{
240 StatsRefreshPeriod: cfg.Synchronizer.StatsRefreshPeriod.Duration,
241 ChainID: chainIDU16,
242 })
243 if err != nil {
244 return nil, tracerr.Wrap(err)
245 }
246 initSCVars := sync.SCVars()
247
248 scConsts := common.SCConsts{
249 Rollup: *sync.RollupConstants(),
250 Auction: *sync.AuctionConstants(),
251 WDelayer: *sync.WDelayerConstants(),
252 }
253
254 hdbNodeCfg := historydb.NodeConfig{
255 MaxPoolTxs: cfg.Coordinator.L2DB.MaxTxs,
256 MinFeeUSD: cfg.Coordinator.L2DB.MinFeeUSD,
257 MaxFeeUSD: cfg.Coordinator.L2DB.MaxFeeUSD,
258 ForgeDelay: cfg.Coordinator.ForgeDelay.Duration.Seconds(),
259 }
260 if err := historyDB.SetNodeConfig(&hdbNodeCfg); err != nil {
261 return nil, tracerr.Wrap(err)
262 }
263 hdbConsts := historydb.Constants{
264 SCConsts: common.SCConsts{
265 Rollup: scConsts.Rollup,
266 Auction: scConsts.Auction,
267 WDelayer: scConsts.WDelayer,
268 },
269 ChainID: chainIDU16,
270 HermezAddress: cfg.SmartContracts.Rollup,
271 }
272 if err := historyDB.SetConstants(&hdbConsts); err != nil {
273 return nil, tracerr.Wrap(err)
274 }
275
276 stateAPIUpdater := stateapiupdater.NewUpdater(historyDB, &hdbNodeCfg, initSCVars, &hdbConsts)
277
278 var coord *coordinator.Coordinator
279 if mode == ModeCoordinator {
280 // Unlock FeeAccount EthAddr in the keystore to generate the
281 // account creation authorization
282 if !keyStore.HasAddress(cfg.Coordinator.FeeAccount.Address) {
283 return nil, tracerr.Wrap(fmt.Errorf(
284 "ethereum keystore doesn't have the key for address %v",
285 cfg.Coordinator.FeeAccount.Address))
286 }
287 feeAccount := accounts.Account{
288 Address: cfg.Coordinator.FeeAccount.Address,
289 }
290 if err := keyStore.Unlock(feeAccount,
291 cfg.Coordinator.EthClient.Keystore.Password); err != nil {
292 return nil, tracerr.Wrap(err)
293 }
294 auth := &common.AccountCreationAuth{
295 EthAddr: cfg.Coordinator.FeeAccount.Address,
296 BJJ: cfg.Coordinator.FeeAccount.BJJ,
297 }
298 if err := auth.Sign(func(msg []byte) ([]byte, error) {
299 return keyStore.SignHash(feeAccount, msg)
300 }, chainIDU16, cfg.SmartContracts.Rollup); err != nil {
301 return nil, tracerr.Wrap(err)
302 }
303 coordAccount := &txselector.CoordAccount{
304 Addr: cfg.Coordinator.FeeAccount.Address,
305 BJJ: cfg.Coordinator.FeeAccount.BJJ,
306 AccountCreationAuth: auth.Signature,
307 }
308 txSelector, err := txselector.NewTxSelector(coordAccount,
309 cfg.Coordinator.TxSelector.Path, stateDB, l2DB)
310 if err != nil {
311 return nil, tracerr.Wrap(err)
312 }
313 batchBuilder, err := batchbuilder.NewBatchBuilder(cfg.Coordinator.BatchBuilder.Path,
314 stateDB, 0, uint64(cfg.Coordinator.Circuit.NLevels))
315 if err != nil {
316 return nil, tracerr.Wrap(err)
317 }
318 serverProofs := make([]prover.Client, len(cfg.Coordinator.ServerProofs))
319 for i, serverProofCfg := range cfg.Coordinator.ServerProofs {
320 serverProofs[i] = prover.NewProofServerClient(serverProofCfg.URL,
321 cfg.Coordinator.ProofServerPollInterval.Duration)
322 }
323
324 txProcessorCfg := txprocessor.Config{
325 NLevels: uint32(cfg.Coordinator.Circuit.NLevels),
326 MaxTx: uint32(cfg.Coordinator.Circuit.MaxTx),
327 ChainID: chainIDU16,
328 MaxFeeTx: common.RollupConstMaxFeeIdxCoordinator,
329 MaxL1Tx: common.RollupConstMaxL1Tx,
330 }
331 var verifierIdx int
332 if cfg.Coordinator.Debug.RollupVerifierIndex == nil {
333 verifierIdx, err = scConsts.Rollup.FindVerifierIdx(
334 cfg.Coordinator.Circuit.MaxTx,
335 cfg.Coordinator.Circuit.NLevels,
336 )
337 if err != nil {
338 return nil, tracerr.Wrap(err)
339 }
340 log.Infow("Found verifier that matches circuit config", "verifierIdx", verifierIdx)
341 } else {
342 verifierIdx = *cfg.Coordinator.Debug.RollupVerifierIndex
343 log.Infow("Using debug verifier index from config", "verifierIdx", verifierIdx)
344 if verifierIdx >= len(scConsts.Rollup.Verifiers) {
345 return nil, tracerr.Wrap(
346 fmt.Errorf("verifierIdx (%v) >= "+
347 "len(scConsts.Rollup.Verifiers) (%v)",
348 verifierIdx, len(scConsts.Rollup.Verifiers)))
349 }
350 verifier := scConsts.Rollup.Verifiers[verifierIdx]
351 if verifier.MaxTx != cfg.Coordinator.Circuit.MaxTx ||
352 verifier.NLevels != cfg.Coordinator.Circuit.NLevels {
353 return nil, tracerr.Wrap(
354 fmt.Errorf("Circuit config and verifier params don't match. "+
355 "circuit.MaxTx = %v, circuit.NLevels = %v, "+
356 "verifier.MaxTx = %v, verifier.NLevels = %v",
357 cfg.Coordinator.Circuit.MaxTx, cfg.Coordinator.Circuit.NLevels,
358 verifier.MaxTx, verifier.NLevels,
359 ))
360 }
361 }
362
363 coord, err = coordinator.NewCoordinator(
364 coordinator.Config{
365 ForgerAddress: cfg.Coordinator.ForgerAddress,
366 ConfirmBlocks: cfg.Coordinator.ConfirmBlocks,
367 L1BatchTimeoutPerc: cfg.Coordinator.L1BatchTimeoutPerc,
368 ForgeRetryInterval: cfg.Coordinator.ForgeRetryInterval.Duration,
369 ForgeDelay: cfg.Coordinator.ForgeDelay.Duration,
370 MustForgeAtSlotDeadline: cfg.Coordinator.MustForgeAtSlotDeadline,
371 IgnoreSlotCommitment: cfg.Coordinator.IgnoreSlotCommitment,
372 ForgeOncePerSlotIfTxs: cfg.Coordinator.ForgeOncePerSlotIfTxs,
373 ForgeNoTxsDelay: cfg.Coordinator.ForgeNoTxsDelay.Duration,
374 SyncRetryInterval: cfg.Coordinator.SyncRetryInterval.Duration,
375 PurgeByExtDelInterval: cfg.Coordinator.PurgeByExtDelInterval.Duration,
376 EthClientAttempts: cfg.Coordinator.EthClient.Attempts,
377 EthClientAttemptsDelay: cfg.Coordinator.EthClient.AttemptsDelay.Duration,
378 EthNoReuseNonce: cfg.Coordinator.EthClient.NoReuseNonce,
379 EthTxResendTimeout: cfg.Coordinator.EthClient.TxResendTimeout.Duration,
380 MaxGasPrice: cfg.Coordinator.EthClient.MaxGasPrice,
381 GasPriceIncPerc: cfg.Coordinator.EthClient.GasPriceIncPerc,
382 TxManagerCheckInterval: cfg.Coordinator.EthClient.CheckLoopInterval.Duration,
383 DebugBatchPath: cfg.Coordinator.Debug.BatchPath,
384 Purger: coordinator.PurgerCfg{
385 PurgeBatchDelay: cfg.Coordinator.L2DB.PurgeBatchDelay,
386 InvalidateBatchDelay: cfg.Coordinator.L2DB.InvalidateBatchDelay,
387 PurgeBlockDelay: cfg.Coordinator.L2DB.PurgeBlockDelay,
388 InvalidateBlockDelay: cfg.Coordinator.L2DB.InvalidateBlockDelay,
389 },
390 ForgeBatchGasCost: cfg.Coordinator.EthClient.ForgeBatchGasCost,
391 VerifierIdx: uint8(verifierIdx),
392 TxProcessorConfig: txProcessorCfg,
393 },
394 historyDB,
395 l2DB,
396 txSelector,
397 batchBuilder,
398 serverProofs,
399 client,
400 &scConsts,
401 initSCVars,
402 )
403 if err != nil {
404 return nil, tracerr.Wrap(err)
405 }
406 }
407 var nodeAPI *NodeAPI
408 if cfg.API.Address != "" {
409 if cfg.Debug.GinDebugMode {
410 gin.SetMode(gin.DebugMode)
411 } else {
412 gin.SetMode(gin.ReleaseMode)
413 }
414 if cfg.API.UpdateMetricsInterval.Duration == 0 {
415 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateMetricsInterval: %v",
416 cfg.API.UpdateMetricsInterval.Duration))
417 }
418 if cfg.API.UpdateRecommendedFeeInterval.Duration == 0 {
419 return nil, tracerr.Wrap(fmt.Errorf("invalid cfg.API.UpdateRecommendedFeeInterval: %v",
420 cfg.API.UpdateRecommendedFeeInterval.Duration))
421 }
422 server := gin.Default()
423 coord := false
424 if mode == ModeCoordinator {
425 coord = cfg.Coordinator.API.Coordinator
426 }
427 var err error
428 nodeAPI, err = NewNodeAPI(
429 cfg.API.Address,
430 coord, cfg.API.Explorer,
431 server,
432 historyDB,
433 l2DB,
434 )
435 if err != nil {
436 return nil, tracerr.Wrap(err)
437 }
438 }
439 var debugAPI *debugapi.DebugAPI
440 if cfg.Debug.APIAddress != "" {
441 debugAPI = debugapi.NewDebugAPI(cfg.Debug.APIAddress, stateDB, sync)
442 }
443 priceUpdater, err := priceupdater.NewPriceUpdater(
444 cfg.PriceUpdater.DefaultUpdateMethod,
445 cfg.PriceUpdater.TokensConfig,
446 historyDB,
447 cfg.PriceUpdater.URLBitfinexV2,
448 cfg.PriceUpdater.URLCoinGeckoV3,
449 )
450 if err != nil {
451 return nil, tracerr.Wrap(err)
452 }
453 ctx, cancel := context.WithCancel(context.Background())
454 return &Node{
455 stateAPIUpdater: stateAPIUpdater,
456 nodeAPI: nodeAPI,
457 debugAPI: debugAPI,
458 priceUpdater: priceUpdater,
459 coord: coord,
460 sync: sync,
461 cfg: cfg,
462 mode: mode,
463 sqlConnRead: dbRead,
464 sqlConnWrite: dbWrite,
465 historyDB: historyDB,
466 ctx: ctx,
467 cancel: cancel,
468 }, nil
469}
470
471// APIServer is a server that only runs the API
472type APIServer struct {
473 nodeAPI *NodeAPI
474 mode Mode
475 ctx context.Context
476 wg sync.WaitGroup
477 cancel context.CancelFunc
478}
479
480// NewAPIServer creates a new APIServer
481func NewAPIServer(mode Mode, cfg *config.APIServer) (*APIServer, error) {
482 meddler.Debug = cfg.Debug.MeddlerLogs
483 // Stablish DB connection
484 dbWrite, err := dbUtils.InitSQLDB(
485 cfg.PostgreSQL.PortWrite,
486 cfg.PostgreSQL.HostWrite,
487 cfg.PostgreSQL.UserWrite,
488 cfg.PostgreSQL.PasswordWrite,
489 cfg.PostgreSQL.NameWrite,
490 )
491 if err != nil {
492 return nil, tracerr.Wrap(fmt.Errorf("dbUtils.InitSQLDB: %w", err))
493 }
494 var dbRead *sqlx.DB
495 if cfg.PostgreSQL.HostRead == "" {
496 dbRead = dbWrite
497 } else if cfg.PostgreSQL.HostRead == cfg.PostgreSQL.HostWrite {
498 return nil, tracerr.Wrap(fmt.Errorf(
499 "PostgreSQL.HostRead and PostgreSQL.HostWrite must be different",
500 ))
501 } else {
502 dbRead, err = dbUtils.InitSQLDB(
503 cfg.PostgreSQL.PortRead,
504 cfg.PostgreSQL.HostRead,
505 cfg.PostgreSQL.UserRead,
506 cfg.PostgreSQL.PasswordRead,
507 cfg.PostgreSQL.NameRead,
508 )
509 if err != nil {
510 return nil, tracerr.Wrap(fmt.Errorf("dbUtils.InitSQLDB: %w", err))
511 }
512 }
513 apiConnCon := dbUtils.NewAPIConnectionController(
514 cfg.API.MaxSQLConnections,
515 cfg.API.SQLConnectionTimeout.Duration,
516 )
517
518 historyDB := historydb.NewHistoryDB(dbRead, dbWrite, apiConnCon)
519
520 var l2DB *l2db.L2DB
521 if mode == ModeCoordinator {
522 l2DB = l2db.NewL2DB(
523 dbRead, dbWrite,
524 0,
525 cfg.Coordinator.L2DB.MaxTxs,
526 cfg.Coordinator.L2DB.MinFeeUSD,
527 cfg.Coordinator.L2DB.MaxFeeUSD,
528 0,
529 apiConnCon,
530 )
531 }
532
533 if cfg.Debug.GinDebugMode {
534 gin.SetMode(gin.DebugMode)
535 } else {
536 gin.SetMode(gin.ReleaseMode)
537 }
538 server := gin.Default()
539 coord := false
540 if mode == ModeCoordinator {
541 coord = cfg.Coordinator.API.Coordinator
542 }
543 nodeAPI, err := NewNodeAPI(
544 cfg.API.Address,
545 coord, cfg.API.Explorer,
546 server,
547 historyDB,
548 l2DB,
549 )
550 if err != nil {
551 return nil, tracerr.Wrap(err)
552 }
553 ctx, cancel := context.WithCancel(context.Background())
554 return &APIServer{
555 nodeAPI: nodeAPI,
556 mode: mode,
557 ctx: ctx,
558 cancel: cancel,
559 }, nil
560}
561
562// Start the APIServer
563func (s *APIServer) Start() {
564 log.Infow("Starting api server...", "mode", s.mode)
565 log.Info("Starting NodeAPI...")
566 s.wg.Add(1)
567 go func() {
568 defer func() {
569 log.Info("NodeAPI routine stopped")
570 s.wg.Done()
571 }()
572 if err := s.nodeAPI.Run(s.ctx); err != nil {
573 if s.ctx.Err() != nil {
574 return
575 }
576 log.Fatalw("NodeAPI.Run", "err", err)
577 }
578 }()
579}
580
581// Stop the APIServer
582func (s *APIServer) Stop() {
583 log.Infow("Stopping NodeAPI...")
584 s.cancel()
585 s.wg.Wait()
586}
587
588// NodeAPI holds the node http API
589type NodeAPI struct { //nolint:golint
590 api *api.API
591 engine *gin.Engine
592 addr string
593}
594
595func handleNoRoute(c *gin.Context) {
596 c.JSON(http.StatusNotFound, gin.H{
597 "error": "404 page not found",
598 })
599}
600
601// NewNodeAPI creates a new NodeAPI (which internally calls api.NewAPI)
602func NewNodeAPI(
603 addr string,
604 coordinatorEndpoints, explorerEndpoints bool,
605 server *gin.Engine,
606 hdb *historydb.HistoryDB,
607 l2db *l2db.L2DB,
608) (*NodeAPI, error) {
609 engine := gin.Default()
610 engine.NoRoute(handleNoRoute)
611 engine.Use(cors.Default())
612 _api, err := api.NewAPI(
613 coordinatorEndpoints, explorerEndpoints,
614 engine,
615 hdb,
616 l2db,
617 )
618 if err != nil {
619 return nil, tracerr.Wrap(err)
620 }
621 return &NodeAPI{
622 addr: addr,
623 api: _api,
624 engine: engine,
625 }, nil
626}
627
628// Run starts the http server of the NodeAPI. To stop it, pass a context
629// with cancellation.
630func (a *NodeAPI) Run(ctx context.Context) error {
631 server := &http.Server{
632 Handler: a.engine,
633 // TODO: Figure out best parameters for production
634 ReadTimeout: 30 * time.Second, //nolint:gomnd
635 WriteTimeout: 30 * time.Second, //nolint:gomnd
636 MaxHeaderBytes: 1 << 20, //nolint:gomnd
637 }
638 listener, err := net.Listen("tcp", a.addr)
639 if err != nil {
640 return tracerr.Wrap(err)
641 }
642 log.Infof("NodeAPI is ready at %v", a.addr)
643 go func() {
644 if err := server.Serve(listener); err != nil &&
645 tracerr.Unwrap(err) != http.ErrServerClosed {
646 log.Fatalf("Listen: %s\n", err)
647 }
648 }()
649
650 <-ctx.Done()
651 log.Info("Stopping NodeAPI...")
652 ctxTimeout, cancel := context.WithTimeout(context.Background(), 10*time.Second) //nolint:gomnd
653 defer cancel()
654 if err := server.Shutdown(ctxTimeout); err != nil {
655 return tracerr.Wrap(err)
656 }
657 log.Info("NodeAPI done")
658 return nil
659}
660
661func (n *Node) handleNewBlock(ctx context.Context, stats *synchronizer.Stats,
662 vars *common.SCVariablesPtr, batches []common.BatchData) error {
663 if n.mode == ModeCoordinator {
664 n.coord.SendMsg(ctx, coordinator.MsgSyncBlock{
665 Stats: *stats,
666 Vars: *vars,
667 Batches: batches,
668 })
669 }
670 n.stateAPIUpdater.SetSCVars(vars)
671 if stats.Synced() {
672 if err := n.stateAPIUpdater.UpdateNetworkInfo(
673 stats.Eth.LastBlock, stats.Sync.LastBlock,
674 common.BatchNum(stats.Eth.LastBatchNum),
675 stats.Sync.Auction.CurrentSlot.SlotNum,
676 ); err != nil {
677 log.Errorw("ApiStateUpdater.UpdateNetworkInfo", "err", err)
678 }
679 } else {
680 n.stateAPIUpdater.UpdateNetworkInfoBlock(
681 stats.Eth.LastBlock, stats.Sync.LastBlock,
682 )
683 }
684 if err := n.stateAPIUpdater.Store(); err != nil {
685 return tracerr.Wrap(err)
686 }
687 return nil
688}
689
690func (n *Node) handleReorg(ctx context.Context, stats *synchronizer.Stats,
691 vars *common.SCVariables) error {
692 if n.mode == ModeCoordinator {
693 n.coord.SendMsg(ctx, coordinator.MsgSyncReorg{
694 Stats: *stats,
695 Vars: *vars.AsPtr(),
696 })
697 }
698 n.stateAPIUpdater.SetSCVars(vars.AsPtr())
699 n.stateAPIUpdater.UpdateNetworkInfoBlock(
700 stats.Eth.LastBlock, stats.Sync.LastBlock,
701 )
702 if err := n.stateAPIUpdater.Store(); err != nil {
703 return tracerr.Wrap(err)
704 }
705 return nil
706}
707
708// TODO(Edu): Consider keeping the `lastBlock` inside synchronizer so that we
709// don't have to pass it around.
710func (n *Node) syncLoopFn(ctx context.Context, lastBlock *common.Block) (*common.Block,
711 time.Duration, error) {
712 blockData, discarded, err := n.sync.Sync(ctx, lastBlock)
713 stats := n.sync.Stats()
714 if err != nil {
715 // case: error
716 return nil, n.cfg.Synchronizer.SyncLoopInterval.Duration, tracerr.Wrap(err)
717 } else if discarded != nil {
718 // case: reorg
719 log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
720 vars := n.sync.SCVars()
721 if err := n.handleReorg(ctx, stats, vars); err != nil {
722 return nil, time.Duration(0), tracerr.Wrap(err)
723 }
724 return nil, time.Duration(0), nil
725 } else if blockData != nil {
726 // case: new block
727 vars := common.SCVariablesPtr{
728 Rollup: blockData.Rollup.Vars,
729 Auction: blockData.Auction.Vars,
730 WDelayer: blockData.WDelayer.Vars,
731 }
732 if err := n.handleNewBlock(ctx, stats, &vars, blockData.Rollup.Batches); err != nil {
733 return nil, time.Duration(0), tracerr.Wrap(err)
734 }
735 return &blockData.Block, time.Duration(0), nil
736 } else {
737 // case: no block
738 return lastBlock, n.cfg.Synchronizer.SyncLoopInterval.Duration, nil
739 }
740}
741
742// StartSynchronizer starts the synchronizer
743func (n *Node) StartSynchronizer() {
744 log.Info("Starting Synchronizer...")
745
746 // Trigger a manual call to handleNewBlock with the loaded state of the
747 // synchronizer in order to quickly activate the API and Coordinator
748 // and avoid waiting for the next block. Without this, the API and
749 // Coordinator will not react until the following block (starting from
750 // the last synced one) is synchronized
751 stats := n.sync.Stats()
752 vars := n.sync.SCVars()
753 if err := n.handleNewBlock(n.ctx, stats, vars.AsPtr(), []common.BatchData{}); err != nil {
754 log.Fatalw("Node.handleNewBlock", "err", err)
755 }
756
757 n.wg.Add(1)
758 go func() {
759 var err error
760 var lastBlock *common.Block
761 waitDuration := time.Duration(0)
762 for {
763 select {
764 case <-n.ctx.Done():
765 log.Info("Synchronizer done")
766 n.wg.Done()
767 return
768 case <-time.After(waitDuration):
769 if lastBlock, waitDuration, err = n.syncLoopFn(n.ctx,
770 lastBlock); err != nil {
771 if n.ctx.Err() != nil {
772 continue
773 }
774 if errors.Is(err, eth.ErrBlockHashMismatchEvent) {
775 log.Warnw("Synchronizer.Sync", "err", err)
776 } else if errors.Is(err, synchronizer.ErrUnknownBlock) {
777 log.Warnw("Synchronizer.Sync", "err", err)
778 } else {
779 log.Errorw("Synchronizer.Sync", "err", err)
780 }
781 }
782 }
783 }
784 }()
785
786 n.wg.Add(1)
787 go func() {
788 for {
789 select {
790 case <-n.ctx.Done():
791 log.Info("PriceUpdater done")
792 n.wg.Done()
793 return
794 case <-time.After(n.cfg.PriceUpdater.Interval.Duration):
795 if err := n.priceUpdater.UpdateTokenList(); err != nil {
796 log.Errorw("PriceUpdater.UpdateTokenList()", "err", err)
797 }
798 n.priceUpdater.UpdatePrices(n.ctx)
799 }
800 }
801 }()
802}
803
804// StartDebugAPI starts the DebugAPI
805func (n *Node) StartDebugAPI() {
806 log.Info("Starting DebugAPI...")
807 n.wg.Add(1)
808 go func() {
809 defer func() {
810 log.Info("DebugAPI routine stopped")
811 n.wg.Done()
812 }()
813 if err := n.debugAPI.Run(n.ctx); err != nil {
814 if n.ctx.Err() != nil {
815 return
816 }
817 log.Fatalw("DebugAPI.Run", "err", err)
818 }
819 }()
820}
821
822// StartNodeAPI starts the NodeAPI
823func (n *Node) StartNodeAPI() {
824 log.Info("Starting NodeAPI...")
825 n.wg.Add(1)
826 go func() {
827 defer func() {
828 log.Info("NodeAPI routine stopped")
829 n.wg.Done()
830 }()
831 if err := n.nodeAPI.Run(n.ctx); err != nil {
832 if n.ctx.Err() != nil {
833 return
834 }
835 log.Fatalw("NodeAPI.Run", "err", err)
836 }
837 }()
838
839 n.wg.Add(1)
840 go func() {
841 // Do an initial update on startup
842 if err := n.stateAPIUpdater.UpdateMetrics(); err != nil {
843 log.Errorw("ApiStateUpdater.UpdateMetrics", "err", err)
844 }
845 if err := n.stateAPIUpdater.Store(); err != nil {
846 log.Errorw("ApiStateUpdater.Store", "err", err)
847 }
848 for {
849 select {
850 case <-n.ctx.Done():
851 log.Info("ApiStateUpdater.UpdateMetrics loop done")
852 n.wg.Done()
853 return
854 case <-time.After(n.cfg.API.UpdateMetricsInterval.Duration):
855 if err := n.stateAPIUpdater.UpdateMetrics(); err != nil {
856 log.Errorw("ApiStateUpdater.UpdateMetrics", "err", err)
857 continue
858 }
859 if err := n.stateAPIUpdater.Store(); err != nil {
860 log.Errorw("ApiStateUpdater.Store", "err", err)
861 }
862 }
863 }
864 }()
865
866 n.wg.Add(1)
867 go func() {
868 // Do an initial update on startup
869 if err := n.stateAPIUpdater.UpdateRecommendedFee(); err != nil {
870 log.Errorw("ApiStateUpdater.UpdateRecommendedFee", "err", err)
871 }
872 if err := n.stateAPIUpdater.Store(); err != nil {
873 log.Errorw("ApiStateUpdater.Store", "err", err)
874 }
875 for {
876 select {
877 case <-n.ctx.Done():
878 log.Info("ApiStateUpdaterAPI.UpdateRecommendedFee loop done")
879 n.wg.Done()
880 return
881 case <-time.After(n.cfg.API.UpdateRecommendedFeeInterval.Duration):
882 if err := n.stateAPIUpdater.UpdateRecommendedFee(); err != nil {
883 log.Errorw("ApiStateUpdaterAPI.UpdateRecommendedFee", "err", err)
884 continue
885 }
886 if err := n.stateAPIUpdater.Store(); err != nil {
887 log.Errorw("ApiStateUpdater.Store", "err", err)
888 }
889 }
890 }
891 }()
892}
893
894// Start the node
895func (n *Node) Start() {
896 log.Infow("Starting node...", "mode", n.mode)
897 if n.debugAPI != nil {
898 n.StartDebugAPI()
899 }
900 if n.nodeAPI != nil {
901 n.StartNodeAPI()
902 }
903 if n.mode == ModeCoordinator {
904 log.Info("Starting Coordinator...")
905 n.coord.Start()
906 }
907 n.StartSynchronizer()
908}
909
910// Stop the node
911func (n *Node) Stop() {
912 log.Infow("Stopping node...")
913 n.cancel()
914 n.wg.Wait()
915 if n.mode == ModeCoordinator {
916 log.Info("Stopping Coordinator...")
917 n.coord.Stop()
918 }
919 // Close kv DBs
920 n.sync.StateDB().Close()
921 if n.mode == ModeCoordinator {
922 n.coord.TxSelector().LocalAccountsDB().Close()
923 n.coord.BatchBuilder().LocalStateDB().Close()
924 }
925}