arnaucube
arnaucube/hermez-nodePublic Self-hosted repository
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, &eth.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		WDelayer: eth.WDelayerConfig{
106			Address: cfg.SmartContracts.WDelayer,
107		},
108	})
109	if err != nil {
110		return nil, err
111	}
112
113	sync, err := synchronizer.NewSynchronizer(client, historyDB, stateDB, synchronizer.Config{
114		StartBlockNum:    cfg.Synchronizer.StartBlockNum,
115		InitialVariables: cfg.Synchronizer.InitialVariables,
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, 1)
180	n.stoppedGetProofCallForge = make(chan bool, 1)
181	n.stoppedForgeCallConfirm = make(chan bool, 1)
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	// stopped channel is size 1 so that the defer doesn't block
253	n.stoppedSync = make(chan bool, 1)
254	go func() {
255		defer func() {
256			n.stoppedSync <- true
257		}()
258		var lastBlock *common.Block
259		d := time.Duration(0)
260		for {
261			select {
262			case <-n.ctx.Done():
263				log.Info("Synchronizer stopped")
264				return
265			case <-time.After(d):
266				if blockData, discarded, err := n.sync.Sync2(n.ctx, lastBlock); err != nil {
267					log.Errorw("Synchronizer.Sync", "error", err)
268					lastBlock = nil
269					d = n.cfg.Synchronizer.SyncLoopInterval.Duration
270				} else if discarded != nil {
271					log.Infow("Synchronizer.Sync reorg", "discarded", *discarded)
272					lastBlock = nil
273					d = time.Duration(0)
274				} else if blockData != nil {
275					lastBlock = &blockData.Block
276					d = time.Duration(0)
277				} else {
278					d = n.cfg.Synchronizer.SyncLoopInterval.Duration
279				}
280			}
281		}
282	}()
283	// TODO: Run price updater.  This is required by the API and the TxSelector
284}
285
286// WaitStopSynchronizer waits for the synchronizer to stop
287func (n *Node) WaitStopSynchronizer() {
288	log.Info("Waiting for Synchronizer to stop...")
289	<-n.stoppedSync
290}
291
292// Start the node
293func (n *Node) Start() {
294	log.Infow("Starting node...", "mode", n.mode)
295	if n.mode == ModeCoordinator {
296		n.StartCoordinator()
297	}
298	n.StartSynchronizer()
299}
300
301// Stop the node
302func (n *Node) Stop() {
303	log.Infow("Stopping node...")
304	n.cancel()
305	if n.mode == ModeCoordinator {
306		n.StopCoordinator()
307	}
308	n.WaitStopSynchronizer()
309}