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