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