hermez-node/node/node.go
node/node.goRaw
1package node
2
3import (
4 "context"
5 "sync"
6 "time"
7
8 "github.com/ethereum/go-ethereum/ethclient"
9 "github.com/hermeznetwork/hermez-node/batchbuilder"
10 "github.com/hermeznetwork/hermez-node/common"
11 "github.com/hermeznetwork/hermez-node/config"
12 "github.com/hermeznetwork/hermez-node/coordinator"
13 dbUtils "github.com/hermeznetwork/hermez-node/db"
14 "github.com/hermeznetwork/hermez-node/db/historydb"
15 "github.com/hermeznetwork/hermez-node/db/l2db"
16 "github.com/hermeznetwork/hermez-node/db/statedb"
17 "github.com/hermeznetwork/hermez-node/eth"
18 "github.com/hermeznetwork/hermez-node/log"
19 "github.com/hermeznetwork/hermez-node/synchronizer"
20 "github.com/hermeznetwork/hermez-node/test/debugapi"
21 "github.com/hermeznetwork/hermez-node/txselector"
22 "github.com/jmoiron/sqlx"
23)
24
25// Mode sets the working mode of the node (synchronizer or coordinator)
26type Mode string
27
28const (
29 // ModeCoordinator defines the mode of the HermezNode as Coordinator, which
30 // means that the node is set to forge (which also will be synchronizing with
31 // the L1 blockchain state)
32 ModeCoordinator Mode = "coordinator"
33
34 // ModeSynchronizer defines the mode of the HermezNode as Synchronizer, which
35 // means that the node is set to only synchronize with the L1 blockchain state
36 // and will not forge
37 ModeSynchronizer Mode = "synchronizer"
38)
39
40// Node is the Hermez Node
41type Node struct {
42 debugAPI *debugapi.DebugAPI
43 // Coordinator
44 coord *coordinator.Coordinator
45 coordCfg *config.Coordinator
46 stopForge chan bool
47 stopGetProofCallForge chan bool
48 stopForgeCallConfirm chan bool
49 stoppedForge chan bool
50 stoppedGetProofCallForge chan bool
51 stoppedForgeCallConfirm chan bool
52
53 // Synchronizer
54 sync *synchronizer.Synchronizer
55
56 // General
57 cfg *config.Node
58 mode Mode
59 sqlConn *sqlx.DB
60 ctx context.Context
61 wg sync.WaitGroup
62 cancel context.CancelFunc
63}
64
65// NewNode creates a Node
66func NewNode(mode Mode, cfg *config.Node, coordCfg *config.Coordinator) (*Node, error) {
67 // Stablish DB connection
68 db, err := dbUtils.InitSQLDB(
69 cfg.PostgreSQL.Port,
70 cfg.PostgreSQL.Host,
71 cfg.PostgreSQL.User,
72 cfg.PostgreSQL.Password,
73 cfg.PostgreSQL.Name,
74 )
75 if err != nil {
76 return nil, err
77 }
78
79 historyDB := historydb.NewHistoryDB(db)
80
81 stateDB, err := statedb.NewStateDB(cfg.StateDB.Path, statedb.TypeSynchronizer, 32)
82 if err != nil {
83 return nil, err
84 }
85
86 ethClient, err := ethclient.Dial(cfg.Web3.URL)
87 if err != nil {
88 return nil, err
89 }
90 client, err := eth.NewClient(ethClient, nil, nil, ð.ClientConfig{
91 Ethereum: eth.EthereumConfig{
92 CallGasLimit: cfg.EthClient.CallGasLimit,
93 DeployGasLimit: cfg.EthClient.DeployGasLimit,
94 GasPriceDiv: cfg.EthClient.GasPriceDiv,
95 ReceiptTimeout: cfg.EthClient.ReceiptTimeout.Duration,
96 IntervalReceiptLoop: cfg.EthClient.IntervalReceiptLoop.Duration,
97 },
98 Rollup: eth.RollupConfig{
99 Address: cfg.SmartContracts.Rollup,
100 },
101 Auction: eth.AuctionConfig{
102 Address: cfg.SmartContracts.Auction,
103 TokenHEZ: eth.TokenConfig{
104 Address: cfg.SmartContracts.TokenHEZ,
105 Name: cfg.SmartContracts.TokenHEZName,
106 },
107 },
108 WDelayer: eth.WDelayerConfig{
109 Address: cfg.SmartContracts.WDelayer,
110 },
111 })
112 if err != nil {
113 return nil, err
114 }
115
116 sync, err := synchronizer.NewSynchronizer(client, historyDB, stateDB, synchronizer.Config{
117 StartBlockNum: cfg.Synchronizer.StartBlockNum,
118 InitialVariables: cfg.Synchronizer.InitialVariables,
119 StatsRefreshPeriod: cfg.Synchronizer.StatsRefreshPeriod.Duration,
120 })
121 if err != nil {
122 return nil, err
123 }
124
125 var coord *coordinator.Coordinator
126 if mode == ModeCoordinator {
127 l2DB := l2db.NewL2DB(
128 db,
129 coordCfg.L2DB.SafetyPeriod,
130 coordCfg.L2DB.MaxTxs,
131 coordCfg.L2DB.TTL.Duration,
132 )
133 // TODO: Get (maxL1UserTxs, maxL1OperatorTxs, maxTxs) from the smart contract
134 txSelector, err := txselector.NewTxSelector(coordCfg.TxSelector.Path, stateDB, l2DB, 10, 10, 10)
135 if err != nil {
136 return nil, err
137 }
138 // TODO: Get (configCircuits []ConfigCircuit, batchNum common.BatchNum, nLevels uint64) from smart contract
139 nLevels := uint64(32) //nolint:gomnd
140 batchBuilder, err := batchbuilder.NewBatchBuilder(coordCfg.BatchBuilder.Path, stateDB, nil, 0, nLevels)
141 if err != nil {
142 return nil, err
143 }
144 if err != nil {
145 return nil, err
146 }
147 serverProofs := make([]coordinator.ServerProofInterface, len(coordCfg.ServerProofs))
148 for i, serverProofCfg := range coordCfg.ServerProofs {
149 serverProofs[i] = coordinator.NewServerProof(serverProofCfg.URL)
150 }
151 coord = coordinator.NewCoordinator(
152 coordinator.Config{
153 ForgerAddress: coordCfg.ForgerAddress,
154 },
155 historyDB,
156 txSelector,
157 batchBuilder,
158 serverProofs,
159 client,
160 )
161 }
162 var debugAPI *debugapi.DebugAPI
163 if cfg.Debug.APIAddress != "" {
164 debugAPI = debugapi.NewDebugAPI(cfg.Debug.APIAddress, stateDB, sync)
165 }
166 ctx, cancel := context.WithCancel(context.Background())
167 return &Node{
168 debugAPI: debugAPI,
169 coord: coord,
170 coordCfg: coordCfg,
171 sync: sync,
172 cfg: cfg,
173 mode: mode,
174 sqlConn: db,
175 ctx: ctx,
176 cancel: cancel,
177 }, nil
178}
179
180// StartCoordinator starts the coordinator
181func (n *Node) StartCoordinator() {
182 log.Info("Starting Coordinator...")
183
184 // TODO: Replace stopXXX by context
185 // TODO: Replace stoppedXXX by waitgroup
186
187 n.stopForge = make(chan bool)
188 n.stopGetProofCallForge = make(chan bool)
189 n.stopForgeCallConfirm = make(chan bool)
190
191 n.stoppedForge = make(chan bool, 1)
192 n.stoppedGetProofCallForge = make(chan bool, 1)
193 n.stoppedForgeCallConfirm = make(chan bool, 1)
194
195 queueSize := 1
196 batchCh0 := make(chan *coordinator.BatchInfo, queueSize)
197 batchCh1 := make(chan *coordinator.BatchInfo, queueSize)
198
199 go func() {
200 defer func() { n.stoppedForge <- true }()
201 for {
202 select {
203 case <-n.stopForge:
204 return
205 default:
206 if forge, err := n.coord.ForgeLoopFn(batchCh0, n.stopForge); err == coordinator.ErrStop {
207 return
208 } else if err != nil {
209 log.Errorw("Coordinator.ForgeLoopFn", "error", err)
210 } else if !forge {
211 time.Sleep(n.coordCfg.ForgeLoopInterval.Duration)
212 }
213 }
214 }
215 }()
216 go func() {
217 defer func() { n.stoppedGetProofCallForge <- true }()
218 for {
219 select {
220 case <-n.stopGetProofCallForge:
221 return
222 default:
223 if err := n.coord.GetProofCallForgeLoopFn(
224 batchCh0, batchCh1, n.stopGetProofCallForge); err == coordinator.ErrStop {
225 return
226 } else if err != nil {
227 log.Errorw("Coordinator.GetProofCallForgeLoopFn", "error", err)
228 }
229 }
230 }
231 }()
232 go func() {
233 defer func() { n.stoppedForgeCallConfirm <- true }()
234 for {
235 select {
236 case <-n.stopForgeCallConfirm:
237 return
238 default:
239 if err := n.coord.ForgeCallConfirmLoopFn(
240 batchCh1, n.stopForgeCallConfirm); err == coordinator.ErrStop {
241 return
242 } else if err != nil {
243 log.Errorw("Coordinator.ForgeCallConfirmLoopFn", "error", err)
244 }
245 }
246 }
247 }()
248}
249
250// StopCoordinator stops the coordinator
251func (n *Node) StopCoordinator() {
252 log.Info("Stopping Coordinator...")
253 n.stopForge <- true
254 n.stopGetProofCallForge <- true
255 n.stopForgeCallConfirm <- true
256 <-n.stoppedForge
257 <-n.stoppedGetProofCallForge
258 <-n.stoppedForgeCallConfirm
259}
260
261// StartSynchronizer starts the synchronizer
262func (n *Node) StartSynchronizer() {
263 log.Info("Starting Synchronizer...")
264 n.wg.Add(1)
265 go func() {
266 defer func() {
267 log.Info("Synchronizer routine stopped")
268 n.wg.Done()
269 }()
270 var lastBlock *common.Block
271 d := time.Duration(0)
272 for {
273 select {
274 case <-n.ctx.Done():
275 log.Info("Synchronizer done")
276 return
277 case <-time.After(d):
278 if blockData, discarded, err := n.sync.Sync2(n.ctx, lastBlock); err != nil {
279 log.Errorw("Synchronizer.Sync", "error", err)
280 lastBlock = nil
281 d = n.cfg.Synchronizer.SyncLoopInterval.Duration
282 } else if discarded != nil {
283 log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
284 lastBlock = nil
285 d = time.Duration(0)
286 } else if blockData != nil {
287 lastBlock = &blockData.Block
288 d = time.Duration(0)
289 } else {
290 d = n.cfg.Synchronizer.SyncLoopInterval.Duration
291 }
292 }
293 }
294 }()
295 // TODO: Run price updater. This is required by the API and the TxSelector
296}
297
298// StartDebugAPI starts the DebugAPI
299func (n *Node) StartDebugAPI() {
300 log.Info("Starting DebugAPI...")
301 n.wg.Add(1)
302 go func() {
303 defer func() {
304 log.Info("DebugAPI routine stopped")
305 n.wg.Done()
306 }()
307 if err := n.debugAPI.Run(n.ctx); err != nil {
308 log.Fatalw("DebugAPI.Run", "err", err)
309 }
310 }()
311}
312
313// Start the node
314func (n *Node) Start() {
315 log.Infow("Starting node...", "mode", n.mode)
316 if n.debugAPI != nil {
317 n.StartDebugAPI()
318 }
319 if n.mode == ModeCoordinator {
320 n.StartCoordinator()
321 }
322 n.StartSynchronizer()
323}
324
325// Stop the node
326func (n *Node) Stop() {
327 log.Infow("Stopping node...")
328 n.cancel()
329 if n.mode == ModeCoordinator {
330 n.StopCoordinator()
331 }
332 n.wg.Wait()
333}