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