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