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