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