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