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