hermez-node/node/node.go
node/node.goRaw
1package node
2
3import (
4 "context"
5 "net/http"
6 "sync"
7 "time"
8
9 "github.com/ethereum/go-ethereum/ethclient"
10 "github.com/gin-contrib/cors"
11 "github.com/gin-gonic/gin"
12 "github.com/hermeznetwork/hermez-node/api"
13 "github.com/hermeznetwork/hermez-node/batchbuilder"
14 "github.com/hermeznetwork/hermez-node/common"
15 "github.com/hermeznetwork/hermez-node/config"
16 "github.com/hermeznetwork/hermez-node/coordinator"
17 dbUtils "github.com/hermeznetwork/hermez-node/db"
18 "github.com/hermeznetwork/hermez-node/db/historydb"
19 "github.com/hermeznetwork/hermez-node/db/l2db"
20 "github.com/hermeznetwork/hermez-node/db/statedb"
21 "github.com/hermeznetwork/hermez-node/eth"
22 "github.com/hermeznetwork/hermez-node/log"
23 "github.com/hermeznetwork/hermez-node/synchronizer"
24 "github.com/hermeznetwork/hermez-node/test/debugapi"
25 "github.com/hermeznetwork/hermez-node/txselector"
26 "github.com/jmoiron/sqlx"
27)
28
29// Mode sets the working mode of the node (synchronizer or coordinator)
30type Mode string
31
32const (
33 // ModeCoordinator defines the mode of the HermezNode as Coordinator, which
34 // means that the node is set to forge (which also will be synchronizing with
35 // the L1 blockchain state)
36 ModeCoordinator Mode = "coordinator"
37
38 // ModeSynchronizer defines the mode of the HermezNode as Synchronizer, which
39 // means that the node is set to only synchronize with the L1 blockchain state
40 // and will not forge
41 ModeSynchronizer Mode = "synchronizer"
42)
43
44// Node is the Hermez Node
45type Node struct {
46 nodeAPI *NodeAPI
47 debugAPI *debugapi.DebugAPI
48 // Coordinator
49 coord *coordinator.Coordinator
50 coordCfg *config.Coordinator
51
52 // Synchronizer
53 sync *synchronizer.Synchronizer
54
55 // General
56 cfg *config.Node
57 mode Mode
58 sqlConn *sqlx.DB
59 ctx context.Context
60 wg sync.WaitGroup
61 cancel context.CancelFunc
62}
63
64// NewNode creates a Node
65func NewNode(mode Mode, cfg *config.Node, coordCfg *config.Coordinator) (*Node, error) {
66 // Stablish DB connection
67 db, err := dbUtils.InitSQLDB(
68 cfg.PostgreSQL.Port,
69 cfg.PostgreSQL.Host,
70 cfg.PostgreSQL.User,
71 cfg.PostgreSQL.Password,
72 cfg.PostgreSQL.Name,
73 )
74 if err != nil {
75 return nil, err
76 }
77
78 historyDB := historydb.NewHistoryDB(db)
79
80 stateDB, err := statedb.NewStateDB(cfg.StateDB.Path, statedb.TypeSynchronizer, 32)
81 if err != nil {
82 return nil, err
83 }
84
85 ethClient, err := ethclient.Dial(cfg.Web3.URL)
86 if err != nil {
87 return nil, err
88 }
89 var ethCfg eth.EthereumConfig
90 if mode == ModeCoordinator {
91 ethCfg = eth.EthereumConfig{
92 CallGasLimit: coordCfg.EthClient.CallGasLimit,
93 DeployGasLimit: coordCfg.EthClient.DeployGasLimit,
94 GasPriceDiv: coordCfg.EthClient.GasPriceDiv,
95 ReceiptTimeout: coordCfg.EthClient.ReceiptTimeout.Duration,
96 IntervalReceiptLoop: coordCfg.EthClient.IntervalReceiptLoop.Duration,
97 }
98 }
99 client, err := eth.NewClient(ethClient, nil, nil, ð.ClientConfig{
100 Ethereum: ethCfg,
101 Rollup: eth.RollupConfig{
102 Address: cfg.SmartContracts.Rollup,
103 },
104 Auction: eth.AuctionConfig{
105 Address: cfg.SmartContracts.Auction,
106 TokenHEZ: eth.TokenConfig{
107 Address: cfg.SmartContracts.TokenHEZ,
108 Name: cfg.SmartContracts.TokenHEZName,
109 },
110 },
111 WDelayer: eth.WDelayerConfig{
112 Address: cfg.SmartContracts.WDelayer,
113 },
114 })
115 if err != nil {
116 return nil, err
117 }
118
119 sync, err := synchronizer.NewSynchronizer(client, historyDB, stateDB, synchronizer.Config{
120 StartBlockNum: cfg.Synchronizer.StartBlockNum,
121 InitialVariables: cfg.Synchronizer.InitialVariables,
122 StatsRefreshPeriod: cfg.Synchronizer.StatsRefreshPeriod.Duration,
123 })
124 if err != nil {
125 return nil, err
126 }
127 varsRollup, varsAuction, varsWDelayer := sync.SCVars()
128 initSCVars := synchronizer.SCVariables{
129 Rollup: *varsRollup,
130 Auction: *varsAuction,
131 WDelayer: *varsWDelayer,
132 }
133
134 scConsts := synchronizer.SCConsts{
135 Rollup: *sync.RollupConstants(),
136 Auction: *sync.AuctionConstants(),
137 WDelayer: *sync.WDelayerConstants(),
138 }
139
140 var coord *coordinator.Coordinator
141 var l2DB *l2db.L2DB
142 if mode == ModeCoordinator {
143 l2DB = l2db.NewL2DB(
144 db,
145 coordCfg.L2DB.SafetyPeriod,
146 coordCfg.L2DB.MaxTxs,
147 coordCfg.L2DB.TTL.Duration,
148 )
149 // TODO: Get (maxL1UserTxs, maxL1OperatorTxs, maxTxs) from the smart contract
150 txSelector, err := txselector.NewTxSelector(coordCfg.TxSelector.Path, stateDB, l2DB, 10, 10, 10)
151 if err != nil {
152 return nil, err
153 }
154 // TODO: Get (configCircuits []ConfigCircuit, batchNum common.BatchNum, nLevels uint64) from smart contract
155 nLevels := uint64(32) //nolint:gomnd
156 batchBuilder, err := batchbuilder.NewBatchBuilder(coordCfg.BatchBuilder.Path, stateDB, nil, 0, nLevels)
157 if err != nil {
158 return nil, err
159 }
160 if err != nil {
161 return nil, err
162 }
163 serverProofs := make([]coordinator.ServerProofInterface, len(coordCfg.ServerProofs))
164 for i, serverProofCfg := range coordCfg.ServerProofs {
165 serverProofs[i] = coordinator.NewServerProof(serverProofCfg.URL)
166 }
167
168 coord = coordinator.NewCoordinator(
169 coordinator.Config{
170 ForgerAddress: coordCfg.ForgerAddress,
171 ConfirmBlocks: coordCfg.ConfirmBlocks,
172 },
173 historyDB,
174 txSelector,
175 batchBuilder,
176 serverProofs,
177 client,
178 &scConsts,
179 &initSCVars,
180 )
181 }
182 var nodeAPI *NodeAPI
183 if cfg.API.Address != "" {
184 server := gin.Default()
185 coord := false
186 if mode == ModeCoordinator {
187 coord = coordCfg.API.Coordinator
188 }
189 var err error
190 nodeAPI, err = NewNodeAPI(
191 cfg.API.Address,
192 coord, cfg.API.Explorer,
193 server,
194 historyDB,
195 stateDB,
196 l2DB,
197 &api.Config{
198 RollupConstants: scConsts.Rollup,
199 AuctionConstants: scConsts.Auction,
200 WDelayerConstants: scConsts.WDelayer,
201 },
202 )
203 if err != nil {
204 return nil, err
205 }
206 nodeAPI.api.SetRollupVariables(initSCVars.Rollup)
207 nodeAPI.api.SetAuctionVariables(initSCVars.Auction)
208 nodeAPI.api.SetWDelayerVariables(initSCVars.WDelayer)
209 }
210 var debugAPI *debugapi.DebugAPI
211 if cfg.Debug.APIAddress != "" {
212 debugAPI = debugapi.NewDebugAPI(cfg.Debug.APIAddress, stateDB, sync)
213 }
214 ctx, cancel := context.WithCancel(context.Background())
215 return &Node{
216 nodeAPI: nodeAPI,
217 debugAPI: debugAPI,
218 coord: coord,
219 coordCfg: coordCfg,
220 sync: sync,
221 cfg: cfg,
222 mode: mode,
223 sqlConn: db,
224 ctx: ctx,
225 cancel: cancel,
226 }, nil
227}
228
229// NodeAPI holds the node http API
230type NodeAPI struct { //nolint:golint
231 api *api.API
232 engine *gin.Engine
233 addr string
234}
235
236func handleNoRoute(c *gin.Context) {
237 c.JSON(http.StatusNotFound, gin.H{
238 "error": "404 page not found",
239 })
240}
241
242// NewNodeAPI creates a new NodeAPI (which internally calls api.NewAPI)
243func NewNodeAPI(
244 addr string,
245 coordinatorEndpoints, explorerEndpoints bool,
246 server *gin.Engine,
247 hdb *historydb.HistoryDB,
248 sdb *statedb.StateDB,
249 l2db *l2db.L2DB,
250 config *api.Config,
251) (*NodeAPI, error) {
252 engine := gin.Default()
253 engine.NoRoute(handleNoRoute)
254 engine.Use(cors.Default())
255 _api, err := api.NewAPI(
256 coordinatorEndpoints, explorerEndpoints,
257 engine,
258 hdb,
259 sdb,
260 l2db,
261 config,
262 )
263 if err != nil {
264 return nil, err
265 }
266 return &NodeAPI{
267 addr: addr,
268 api: _api,
269 engine: engine,
270 }, nil
271}
272
273// Run starts the http server of the NodeAPI. To stop it, pass a context with
274// cancelation.
275func (a *NodeAPI) Run(ctx context.Context) error {
276 server := &http.Server{
277 Addr: a.addr,
278 Handler: a.engine,
279 // TODO: Figure out best parameters for production
280 ReadTimeout: 30 * time.Second, //nolint:gomnd
281 WriteTimeout: 30 * time.Second, //nolint:gomnd
282 MaxHeaderBytes: 1 << 20, //nolint:gomnd
283 }
284 go func() {
285 log.Infof("NodeAPI is ready at %v", a.addr)
286 if err := server.ListenAndServe(); err != nil &&
287 err != http.ErrServerClosed {
288 log.Fatalf("Listen: %s\n", err)
289 }
290 }()
291
292 <-ctx.Done()
293 log.Info("Stopping NodeAPI...")
294 ctxTimeout, cancel := context.WithTimeout(context.Background(), 10*time.Second) //nolint:gomnd
295 defer cancel()
296 if err := server.Shutdown(ctxTimeout); err != nil {
297 return err
298 }
299 log.Info("NodeAPI done")
300 return nil
301}
302
303// TODO(Edu): Consider keeping the `lastBlock` inside synchronizer so that we
304// don't have to pass it around.
305func (n *Node) syncLoopFn(lastBlock *common.Block) (*common.Block, time.Duration) {
306 if blockData, discarded, err := n.sync.Sync2(n.ctx, lastBlock); err != nil {
307 // case: error
308 log.Errorw("Synchronizer.Sync", "error", err)
309 return nil, n.cfg.Synchronizer.SyncLoopInterval.Duration
310 } else if discarded != nil {
311 // case: reorg
312 log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
313 if n.mode == ModeCoordinator {
314 n.coord.SendMsg(coordinator.MsgSyncReorg{})
315 }
316 if n.nodeAPI != nil {
317 rollup, auction, wDelayer := n.sync.SCVars()
318 n.nodeAPI.api.SetRollupVariables(*rollup)
319 n.nodeAPI.api.SetAuctionVariables(*auction)
320 n.nodeAPI.api.SetWDelayerVariables(*wDelayer)
321
322 // TODO: n.nodeAPI.api.UpdateNetworkInfo()
323 }
324 return nil, time.Duration(0)
325 } else if blockData != nil {
326 // case: new block
327 stats := n.sync.Stats()
328 if n.mode == ModeCoordinator {
329 if stats.Synced() && (blockData.Rollup.Vars != nil ||
330 blockData.Auction.Vars != nil ||
331 blockData.WDelayer.Vars != nil) {
332 n.coord.SendMsg(coordinator.MsgSyncSCVars{
333 Rollup: blockData.Rollup.Vars,
334 Auction: blockData.Auction.Vars,
335 WDelayer: blockData.WDelayer.Vars,
336 })
337 }
338 n.coord.SendMsg(coordinator.MsgSyncStats{
339 Stats: *stats,
340 })
341 }
342 if n.nodeAPI != nil {
343 if blockData.Rollup.Vars != nil {
344 n.nodeAPI.api.SetRollupVariables(*blockData.Rollup.Vars)
345 }
346 if blockData.Auction.Vars != nil {
347 n.nodeAPI.api.SetAuctionVariables(*blockData.Auction.Vars)
348 }
349 if blockData.WDelayer.Vars != nil {
350 n.nodeAPI.api.SetWDelayerVariables(*blockData.WDelayer.Vars)
351 }
352
353 // TODO: n.nodeAPI.api.UpdateNetworkInfo()
354 }
355 return &blockData.Block, time.Duration(0)
356 } else {
357 // case: no block
358 return lastBlock, n.cfg.Synchronizer.SyncLoopInterval.Duration
359 }
360}
361
362// StartSynchronizer starts the synchronizer
363func (n *Node) StartSynchronizer() {
364 log.Info("Starting Synchronizer...")
365 n.wg.Add(1)
366 go func() {
367 var lastBlock *common.Block
368 waitDuration := time.Duration(0)
369 for {
370 select {
371 case <-n.ctx.Done():
372 log.Info("Synchronizer done")
373 n.wg.Done()
374 return
375 case <-time.After(waitDuration):
376 lastBlock, waitDuration = n.syncLoopFn(lastBlock)
377 }
378 }
379 }()
380 // TODO: Run price updater. This is required by the API and the TxSelector
381}
382
383// StartDebugAPI starts the DebugAPI
384func (n *Node) StartDebugAPI() {
385 log.Info("Starting DebugAPI...")
386 n.wg.Add(1)
387 go func() {
388 defer func() {
389 log.Info("DebugAPI routine stopped")
390 n.wg.Done()
391 }()
392 if err := n.debugAPI.Run(n.ctx); err != nil {
393 log.Fatalw("DebugAPI.Run", "err", err)
394 }
395 }()
396}
397
398// StartNodeAPI starts the NodeAPI
399func (n *Node) StartNodeAPI() {
400 log.Info("Starting NodeAPI...")
401 n.wg.Add(1)
402 go func() {
403 defer func() {
404 log.Info("NodeAPI routine stopped")
405 n.wg.Done()
406 }()
407 if err := n.nodeAPI.Run(n.ctx); err != nil {
408 log.Fatalw("NodeAPI.Run", "err", err)
409 }
410 }()
411}
412
413// Start the node
414func (n *Node) Start() {
415 log.Infow("Starting node...", "mode", n.mode)
416 if n.debugAPI != nil {
417 n.StartDebugAPI()
418 }
419 if n.nodeAPI != nil {
420 n.StartNodeAPI()
421 }
422 if n.mode == ModeCoordinator {
423 log.Info("Starting Coordinator...")
424 n.coord.Start()
425 }
426 n.StartSynchronizer()
427}
428
429// Stop the node
430func (n *Node) Stop() {
431 log.Infow("Stopping node...")
432 n.cancel()
433 if n.mode == ModeCoordinator {
434 log.Info("Stopping Coordinator...")
435 n.coord.Stop()
436 }
437 n.wg.Wait()
438}