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
610// NewNodeAPI creates a new NodeAPI (which internally calls api.NewAPI)
611func NewNodeAPI(
612 addr string,
613 coordinatorEndpoints, explorerEndpoints bool,
614 server *gin.Engine,
615 hdb *historydb.HistoryDB,
616 l2db *l2db.L2DB,
617) (*NodeAPI, error) {
618 engine := gin.Default()
619 engine.Use(cors.Default())
620 _api, err := api.NewAPI(
621 coordinatorEndpoints, explorerEndpoints,
622 engine,
623 hdb,
624 l2db,
625 )
626 if err != nil {
627 return nil, tracerr.Wrap(err)
628 }
629 return &NodeAPI{
630 addr: addr,
631 api: _api,
632 engine: engine,
633 }, nil
634}
635
636// Run starts the http server of the NodeAPI. To stop it, pass a context
637// with cancellation.
638func (a *NodeAPI) Run(ctx context.Context) error {
639 server := &http.Server{
640 Handler: a.engine,
641 // TODO: Figure out best parameters for production
642 ReadTimeout: 30 * time.Second, //nolint:gomnd
643 WriteTimeout: 30 * time.Second, //nolint:gomnd
644 MaxHeaderBytes: 1 << 20, //nolint:gomnd
645 }
646 listener, err := net.Listen("tcp", a.addr)
647 if err != nil {
648 return tracerr.Wrap(err)
649 }
650 log.Infof("NodeAPI is ready at %v", a.addr)
651 go func() {
652 if err := server.Serve(listener); err != nil &&
653 tracerr.Unwrap(err) != http.ErrServerClosed {
654 log.Fatalf("Listen: %s\n", err)
655 }
656 }()
657
658 <-ctx.Done()
659 log.Info("Stopping NodeAPI...")
660 ctxTimeout, cancel := context.WithTimeout(context.Background(), 10*time.Second) //nolint:gomnd
661 defer cancel()
662 if err := server.Shutdown(ctxTimeout); err != nil {
663 return tracerr.Wrap(err)
664 }
665 log.Info("NodeAPI done")
666 return nil
667}
668
669func (n *Node) handleNewBlock(ctx context.Context, stats *synchronizer.Stats,
670 vars *common.SCVariablesPtr, batches []common.BatchData) error {
671 if n.mode == ModeCoordinator {
672 n.coord.SendMsg(ctx, coordinator.MsgSyncBlock{
673 Stats: *stats,
674 Vars: *vars,
675 Batches: batches,
676 })
677 }
678 n.stateAPIUpdater.SetSCVars(vars)
679 if stats.Synced() {
680 if err := n.stateAPIUpdater.UpdateNetworkInfo(
681 stats.Eth.LastBlock, stats.Sync.LastBlock,
682 common.BatchNum(stats.Eth.LastBatchNum),
683 stats.Sync.Auction.CurrentSlot.SlotNum,
684 ); err != nil {
685 log.Errorw("ApiStateUpdater.UpdateNetworkInfo", "err", err)
686 }
687 } else {
688 n.stateAPIUpdater.UpdateNetworkInfoBlock(
689 stats.Eth.LastBlock, stats.Sync.LastBlock,
690 )
691 }
692 if err := n.stateAPIUpdater.Store(); err != nil {
693 return tracerr.Wrap(err)
694 }
695 return nil
696}
697
698func (n *Node) handleReorg(ctx context.Context, stats *synchronizer.Stats,
699 vars *common.SCVariables) error {
700 if n.mode == ModeCoordinator {
701 n.coord.SendMsg(ctx, coordinator.MsgSyncReorg{
702 Stats: *stats,
703 Vars: *vars.AsPtr(),
704 })
705 }
706 n.stateAPIUpdater.SetSCVars(vars.AsPtr())
707 n.stateAPIUpdater.UpdateNetworkInfoBlock(
708 stats.Eth.LastBlock, stats.Sync.LastBlock,
709 )
710 if err := n.stateAPIUpdater.Store(); err != nil {
711 return tracerr.Wrap(err)
712 }
713 return nil
714}
715
716// TODO(Edu): Consider keeping the `lastBlock` inside synchronizer so that we
717// don't have to pass it around.
718func (n *Node) syncLoopFn(ctx context.Context, lastBlock *common.Block) (*common.Block,
719 time.Duration, error) {
720 blockData, discarded, err := n.sync.Sync(ctx, lastBlock)
721 stats := n.sync.Stats()
722 if err != nil {
723 // case: error
724 return nil, n.cfg.Synchronizer.SyncLoopInterval.Duration, tracerr.Wrap(err)
725 } else if discarded != nil {
726 // case: reorg
727 log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
728 vars := n.sync.SCVars()
729 if err := n.handleReorg(ctx, stats, vars); err != nil {
730 return nil, time.Duration(0), tracerr.Wrap(err)
731 }
732 return nil, time.Duration(0), nil
733 } else if blockData != nil {
734 // case: new block
735 vars := common.SCVariablesPtr{
736 Rollup: blockData.Rollup.Vars,
737 Auction: blockData.Auction.Vars,
738 WDelayer: blockData.WDelayer.Vars,
739 }
740 if err := n.handleNewBlock(ctx, stats, &vars, blockData.Rollup.Batches); err != nil {
741 return nil, time.Duration(0), tracerr.Wrap(err)
742 }
743 return &blockData.Block, time.Duration(0), nil
744 } else {
745 // case: no block
746 return lastBlock, n.cfg.Synchronizer.SyncLoopInterval.Duration, nil
747 }
748}
749
750// StartSynchronizer starts the synchronizer
751func (n *Node) StartSynchronizer() {
752 log.Info("Starting Synchronizer...")
753
754 // Trigger a manual call to handleNewBlock with the loaded state of the
755 // synchronizer in order to quickly activate the API and Coordinator
756 // and avoid waiting for the next block. Without this, the API and
757 // Coordinator will not react until the following block (starting from
758 // the last synced one) is synchronized
759 stats := n.sync.Stats()
760 vars := n.sync.SCVars()
761 if err := n.handleNewBlock(n.ctx, stats, vars.AsPtr(), []common.BatchData{}); err != nil {
762 log.Fatalw("Node.handleNewBlock", "err", err)
763 }
764
765 n.wg.Add(1)
766 go func() {
767 var err error
768 var lastBlock *common.Block
769 waitDuration := time.Duration(0)
770 for {
771 select {
772 case <-n.ctx.Done():
773 log.Info("Synchronizer done")
774 n.wg.Done()
775 return
776 case <-time.After(waitDuration):
777 if lastBlock, waitDuration, err = n.syncLoopFn(n.ctx,
778 lastBlock); err != nil {
779 if n.ctx.Err() != nil {
780 continue
781 }
782 if errors.Is(err, eth.ErrBlockHashMismatchEvent) {
783 log.Warnw("Synchronizer.Sync", "err", err)
784 } else if errors.Is(err, synchronizer.ErrUnknownBlock) {
785 log.Warnw("Synchronizer.Sync", "err", err)
786 } else {
787 log.Errorw("Synchronizer.Sync", "err", err)
788 }
789 }
790 }
791 }
792 }()
793
794 n.wg.Add(1)
795 go func() {
796 for {
797 select {
798 case <-n.ctx.Done():
799 log.Info("PriceUpdater done")
800 n.wg.Done()
801 return
802 case <-time.After(n.cfg.PriceUpdater.Interval.Duration):
803 if err := n.priceUpdater.UpdateTokenList(); err != nil {
804 log.Errorw("PriceUpdater.UpdateTokenList()", "err", err)
805 }
806 n.priceUpdater.UpdatePrices(n.ctx)
807 }
808 }
809 }()
810}
811
812// StartDebugAPI starts the DebugAPI
813func (n *Node) StartDebugAPI() {
814 log.Info("Starting DebugAPI...")
815 n.wg.Add(1)
816 go func() {
817 defer func() {
818 log.Info("DebugAPI routine stopped")
819 n.wg.Done()
820 }()
821 if err := n.debugAPI.Run(n.ctx); err != nil {
822 if n.ctx.Err() != nil {
823 return
824 }
825 log.Fatalw("DebugAPI.Run", "err", err)
826 }
827 }()
828}
829
830// StartNodeAPI starts the NodeAPI
831func (n *Node) StartNodeAPI() {
832 log.Info("Starting NodeAPI...")
833 n.wg.Add(1)
834 go func() {
835 defer func() {
836 log.Info("NodeAPI routine stopped")
837 n.wg.Done()
838 }()
839 if err := n.nodeAPI.Run(n.ctx); err != nil {
840 if n.ctx.Err() != nil {
841 return
842 }
843 log.Fatalw("NodeAPI.Run", "err", err)
844 }
845 }()
846
847 n.wg.Add(1)
848 go func() {
849 // Do an initial update on startup
850 if err := n.stateAPIUpdater.UpdateMetrics(); err != nil {
851 log.Errorw("ApiStateUpdater.UpdateMetrics", "err", err)
852 }
853 if err := n.stateAPIUpdater.Store(); err != nil {
854 log.Errorw("ApiStateUpdater.Store", "err", err)
855 }
856 for {
857 select {
858 case <-n.ctx.Done():
859 log.Info("ApiStateUpdater.UpdateMetrics loop done")
860 n.wg.Done()
861 return
862 case <-time.After(n.cfg.API.UpdateMetricsInterval.Duration):
863 if err := n.stateAPIUpdater.UpdateMetrics(); err != nil {
864 log.Errorw("ApiStateUpdater.UpdateMetrics", "err", err)
865 continue
866 }
867 if err := n.stateAPIUpdater.Store(); err != nil {
868 log.Errorw("ApiStateUpdater.Store", "err", err)
869 }
870 }
871 }
872 }()
873
874 n.wg.Add(1)
875 go func() {
876 // Do an initial update on startup
877 if err := n.stateAPIUpdater.UpdateRecommendedFee(); err != nil {
878 log.Errorw("ApiStateUpdater.UpdateRecommendedFee", "err", err)
879 }
880 if err := n.stateAPIUpdater.Store(); err != nil {
881 log.Errorw("ApiStateUpdater.Store", "err", err)
882 }
883 for {
884 select {
885 case <-n.ctx.Done():
886 log.Info("ApiStateUpdaterAPI.UpdateRecommendedFee loop done")
887 n.wg.Done()
888 return
889 case <-time.After(n.cfg.API.UpdateRecommendedFeeInterval.Duration):
890 if err := n.stateAPIUpdater.UpdateRecommendedFee(); err != nil {
891 log.Errorw("ApiStateUpdaterAPI.UpdateRecommendedFee", "err", err)
892 continue
893 }
894 if err := n.stateAPIUpdater.Store(); err != nil {
895 log.Errorw("ApiStateUpdater.Store", "err", err)
896 }
897 }
898 }
899 }()
900}
901
902// Start the node
903func (n *Node) Start() {
904 log.Infow("Starting node...", "mode", n.mode)
905 if n.debugAPI != nil {
906 n.StartDebugAPI()
907 }
908 if n.nodeAPI != nil {
909 n.StartNodeAPI()
910 }
911 if n.mode == ModeCoordinator {
912 log.Info("Starting Coordinator...")
913 n.coord.Start()
914 }
915 n.StartSynchronizer()
916}
917
918// Stop the node
919func (n *Node) Stop() {
920 log.Infow("Stopping node...")
921 n.cancel()
922 n.wg.Wait()
923 if n.mode == ModeCoordinator {
924 log.Info("Stopping Coordinator...")
925 n.coord.Stop()
926 }
927 // Close kv DBs
928 n.sync.StateDB().Close()
929 if n.mode == ModeCoordinator {
930 n.coord.TxSelector().LocalAccountsDB().Close()
931 n.coord.BatchBuilder().LocalStateDB().Close()
932 }
933}