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