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