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