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)
111 if err != nil {
112 return nil, err
113 }
114
115 var coord *coordinator.Coordinator
116 if mode == ModeCoordinator {
117 l2DB := l2db.NewL2DB(
118 db,
119 coordCfg.L2DB.SafetyPeriod,
120 coordCfg.L2DB.MaxTxs,
121 coordCfg.L2DB.TTL.Duration,
122 )
123 // TODO: Get (maxL1UserTxs, maxL1OperatorTxs, maxTxs) from the smart contract
124 txSelector, err := txselector.NewTxSelector(coordCfg.TxSelector.Path, stateDB, l2DB, 10, 10, 10)
125 if err != nil {
126 return nil, err
127 }
128 // TODO: Get (configCircuits []ConfigCircuit, batchNum common.BatchNum, nLevels uint64) from smart contract
129 nLevels := uint64(32) //nolint:gomnd
130 batchBuilder, err := batchbuilder.NewBatchBuilder(coordCfg.BatchBuilder.Path, stateDB, nil, 0, nLevels)
131 if err != nil {
132 return nil, err
133 }
134 if err != nil {
135 return nil, err
136 }
137 serverProofs := make([]coordinator.ServerProofInterface, len(coordCfg.ServerProofs))
138 for i, serverProofCfg := range coordCfg.ServerProofs {
139 serverProofs[i] = coordinator.NewServerProof(serverProofCfg.URL)
140 }
141 coord = coordinator.NewCoordinator(
142 coordinator.Config{
143 ForgerAddress: coordCfg.ForgerAddress,
144 },
145 historyDB,
146 txSelector,
147 batchBuilder,
148 serverProofs,
149 client,
150 )
151 }
152 ctx, cancel := context.WithCancel(context.Background())
153 return &Node{
154 coord: coord,
155 coordCfg: coordCfg,
156 sync: sync,
157 cfg: cfg,
158 mode: mode,
159 sqlConn: db,
160 ctx: ctx,
161 cancel: cancel,
162 }, nil
163}
164
165// StartCoordinator starts the coordinator
166func (n *Node) StartCoordinator() {
167 log.Info("Starting Coordinator...")
168
169 n.stopForge = make(chan bool)
170 n.stopGetProofCallForge = make(chan bool)
171 n.stopForgeCallConfirm = make(chan bool)
172
173 n.stoppedForge = make(chan bool)
174 n.stoppedGetProofCallForge = make(chan bool)
175 n.stoppedForgeCallConfirm = make(chan bool)
176
177 queueSize := 1
178 batchCh0 := make(chan *coordinator.BatchInfo, queueSize)
179 batchCh1 := make(chan *coordinator.BatchInfo, queueSize)
180
181 go func() {
182 defer func() { n.stoppedForge <- true }()
183 for {
184 select {
185 case <-n.stopForge:
186 return
187 default:
188 if forge, err := n.coord.ForgeLoopFn(batchCh0, n.stopForge); err == coordinator.ErrStop {
189 return
190 } else if err != nil {
191 log.Errorw("Coordinator.ForgeLoopFn", "error", err)
192 } else if !forge {
193 time.Sleep(n.coordCfg.ForgeLoopInterval.Duration)
194 }
195 }
196 }
197 }()
198 go func() {
199 defer func() { n.stoppedGetProofCallForge <- true }()
200 for {
201 select {
202 case <-n.stopGetProofCallForge:
203 return
204 default:
205 if err := n.coord.GetProofCallForgeLoopFn(
206 batchCh0, batchCh1, n.stopGetProofCallForge); err == coordinator.ErrStop {
207 return
208 } else if err != nil {
209 log.Errorw("Coordinator.GetProofCallForgeLoopFn", "error", err)
210 }
211 }
212 }
213 }()
214 go func() {
215 defer func() { n.stoppedForgeCallConfirm <- true }()
216 for {
217 select {
218 case <-n.stopForgeCallConfirm:
219 return
220 default:
221 if err := n.coord.ForgeCallConfirmLoopFn(
222 batchCh1, n.stopForgeCallConfirm); err == coordinator.ErrStop {
223 return
224 } else if err != nil {
225 log.Errorw("Coordinator.ForgeCallConfirmLoopFn", "error", err)
226 }
227 }
228 }
229 }()
230}
231
232// StopCoordinator stops the coordinator
233func (n *Node) StopCoordinator() {
234 log.Info("Stopping Coordinator...")
235 n.stopForge <- true
236 n.stopGetProofCallForge <- true
237 n.stopForgeCallConfirm <- true
238 <-n.stoppedForge
239 <-n.stoppedGetProofCallForge
240 <-n.stoppedForgeCallConfirm
241}
242
243// StartSynchronizer starts the synchronizer
244func (n *Node) StartSynchronizer() {
245 log.Info("Starting Synchronizer...")
246 n.stoppedSync = make(chan bool)
247 go func() {
248 defer func() { n.stoppedSync <- true }()
249 var lastBlock *common.Block
250 d := time.Duration(0)
251 for {
252 select {
253 case <-n.ctx.Done():
254 log.Info("Coordinator stopped")
255 return
256 case <-time.After(d):
257 if blockData, discarded, err := n.sync.Sync2(n.ctx, lastBlock); err != nil {
258 log.Errorw("Synchronizer.Sync", "error", err)
259 lastBlock = nil
260 d = n.cfg.Synchronizer.SyncLoopInterval.Duration
261 } else if discarded != nil {
262 log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
263 lastBlock = nil
264 d = time.Duration(0)
265 } else if blockData != nil {
266 lastBlock = &blockData.Block
267 d = time.Duration(0)
268 } else {
269 d = n.cfg.Synchronizer.SyncLoopInterval.Duration
270 }
271 }
272 }
273 }()
274 // TODO: Run price updater. This is required by the API and the TxSelector
275}
276
277// WaitStopSynchronizer waits for the synchronizer to stop
278func (n *Node) WaitStopSynchronizer() {
279 log.Info("Waiting for Synchronizer to stop...")
280 <-n.stoppedSync
281}
282
283// Start the node
284func (n *Node) Start() {
285 log.Infow("Starting node...", "mode", n.mode)
286 if n.mode == ModeCoordinator {
287 n.StartCoordinator()
288 }
289 n.StartSynchronizer()
290}
291
292// Stop the node
293func (n *Node) Stop() {
294 log.Infow("Stopping node...")
295 n.cancel()
296 if n.mode == ModeCoordinator {
297 n.StopCoordinator()
298 }
299 n.WaitStopSynchronizer()
300}