hermez-node/synchronizer/synchronizer.go
synchronizer/synchronizer.goRaw
1package synchronizer
2
3import (
4 "context"
5 "database/sql"
6 "errors"
7 "sync"
8
9 "github.com/hermeznetwork/hermez-node/common"
10 "github.com/hermeznetwork/hermez-node/db/historydb"
11 "github.com/hermeznetwork/hermez-node/db/statedb"
12 "github.com/hermeznetwork/hermez-node/eth"
13 "github.com/hermeznetwork/hermez-node/log"
14)
15
16var (
17 // ErrNotAbleToSync is used when there is not possible to find a valid block to sync
18 ErrNotAbleToSync = errors.New("it has not been possible to synchronize any block")
19)
20
21// rollupData contains information returned by the Rollup SC
22type rollupData struct {
23 l1Txs []*common.L1Tx
24 batches []*BatchData
25 // withdrawals []*common.ExitInfo
26 registeredTokens []*common.Token
27 rollupVars *common.RollupVars
28}
29
30// NewRollupData creates an empty rollupData with the slices initialized.
31func newRollupData() rollupData {
32 return rollupData{
33 l1Txs: make([]*common.L1Tx, 0),
34 batches: make([]*BatchData, 0),
35 // withdrawals: make([]*common.ExitInfo, 0),
36 registeredTokens: make([]*common.Token, 0),
37 }
38}
39
40// auctionData contains information returned by the Action SC
41type auctionData struct {
42 bids []*common.Bid
43 coordinators []*common.Coordinator
44 auctionVars *common.AuctionVars
45}
46
47// newAuctionData creates an empty auctionData with the slices initialized.
48func newAuctionData() *auctionData {
49 return &auctionData{
50 bids: make([]*common.Bid, 0),
51 coordinators: make([]*common.Coordinator, 0),
52 }
53}
54
55// BatchData contains information about Batches from the contracts
56type BatchData struct {
57 l1UserTxs []*common.L1Tx
58 l1CoordinatorTxs []*common.L1Tx
59 l2Txs []*common.L2Tx
60 createdAccounts []*common.Account
61 exitTree []common.ExitInfo
62 batch *common.Batch
63}
64
65// NewBatchData creates an empty BatchData with the slices initialized.
66func NewBatchData() *BatchData {
67 return &BatchData{
68 l1UserTxs: make([]*common.L1Tx, 0),
69 l1CoordinatorTxs: make([]*common.L1Tx, 0),
70 l2Txs: make([]*common.L2Tx, 0),
71 createdAccounts: make([]*common.Account, 0),
72 exitTree: make([]common.ExitInfo, 0),
73 }
74}
75
76// BlockData contains information about Blocks from the contracts
77type BlockData struct {
78 block *common.Block
79 // Rollup
80 l1Txs []*common.L1Tx // TODO: Answer: User? Coordinator? Both?
81 batches []*BatchData // TODO: Also contains L1Txs!
82 // withdrawals []*common.ExitInfo // TODO
83 registeredTokens []*common.Token
84 rollupVars *common.RollupVars
85 // Auction
86 bids []*common.Bid
87 coordinators []*common.Coordinator
88 auctionVars *common.AuctionVars
89 // WithdrawalDelayer
90 withdrawalDelayerVars *common.WithdrawalDelayerVars
91}
92
93// Synchronizer implements the Synchronizer type
94type Synchronizer struct {
95 ethClient *eth.Client
96 historyDB *historydb.HistoryDB
97 stateDB *statedb.StateDB
98 firstSavedBlock *common.Block
99 mux sync.Mutex
100}
101
102// NewSynchronizer creates a new Synchronizer
103func NewSynchronizer(ethClient *eth.Client, historyDB *historydb.HistoryDB, stateDB *statedb.StateDB) *Synchronizer {
104 s := &Synchronizer{
105 ethClient: ethClient,
106 historyDB: historyDB,
107 stateDB: stateDB,
108 }
109 return s
110}
111
112// TODO: Be smart about locking: only lock during the read/write operations
113
114// Sync updates History and State DB with information from the blockchain
115// TODO: Return true if a new block was processed
116// TODO: Add argument: maximum number of blocks to process
117// TODO: Check reorgs in the middle of syncing a block. Probably make
118// rollupSync, auctionSync and withdrawalSync return the block hash.
119func (s *Synchronizer) Sync() error {
120 // Avoid new sync while performing one
121 s.mux.Lock()
122 defer s.mux.Unlock()
123
124 var nextBlockNum int64 // next block number to sync
125
126 // Get lastSavedBlock from History DB
127 lastSavedBlock, err := s.historyDB.GetLastBlock()
128 if err != nil && err != sql.ErrNoRows {
129 return err
130 }
131 // If we don't have any stored block, we must do a full sync starting from the rollup genesis block
132 if err == sql.ErrNoRows {
133 // TODO: Query rollup constants and genesis information, store them
134 nextBlockNum = 1234 // TODO: Replace this with genesisBlockNum
135 } else {
136 // Get the latest block we have in History DB from blockchain to detect a reorg
137 ethBlock, err := s.ethClient.EthBlockByNumber(context.Background(), lastSavedBlock.EthBlockNum)
138 if err != nil {
139 return err
140 }
141
142 if ethBlock.Hash != lastSavedBlock.Hash {
143 // Reorg detected
144 log.Debugf("Reorg Detected...")
145 _, err := s.reorg(lastSavedBlock)
146 if err != nil {
147 return err
148 }
149
150 lastSavedBlock, err = s.historyDB.GetLastBlock()
151 if err != nil {
152 return err
153 }
154 }
155 nextBlockNum = lastSavedBlock.EthBlockNum + 1
156 }
157
158 log.Debugf("Syncing...")
159
160 // Get latest blockNum in blockchain
161 latestBlockNum, err := s.ethClient.EthCurrentBlock()
162 if err != nil {
163 return err
164 }
165
166 log.Debugf("Blocks to sync: %v (firstBlockToSync: %v, latestBlock: %v)", latestBlockNum-nextBlockNum+1, nextBlockNum, latestBlockNum)
167
168 for nextBlockNum < latestBlockNum {
169 ethBlock, err := s.ethClient.EthBlockByNumber(context.Background(), nextBlockNum)
170 if err != nil {
171 return err
172 }
173 // TODO: Check that the obtianed ethBlock.ParentHash == prevEthBlock.Hash; if not, reorg!
174
175 // TODO: Send the ethHash in rollupSync(), auctionSync() and
176 // wdelayerSync() and make sure they all use the same block
177 // hash.
178
179 // Get data from the rollup contract
180 rollupData, err := s.rollupSync(nextBlockNum)
181 if err != nil {
182 return err
183 }
184
185 // Get data from the auction contract
186 auctionData, err := s.auctionSync(nextBlockNum)
187 if err != nil {
188 return err
189 }
190
191 // Get data from the WithdrawalDelayer contract
192 wdelayerData, err := s.wdelayerSync(nextBlockNum)
193 if err != nil {
194 return err
195 }
196
197 // Group all the block data into the structs to save into HistoryDB
198 var blockData BlockData
199
200 blockData.block = ethBlock
201
202 if rollupData != nil {
203 blockData.l1Txs = rollupData.l1Txs
204 blockData.batches = rollupData.batches
205 // blockData.withdrawals = rollupData.withdrawals // TODO
206 blockData.registeredTokens = rollupData.registeredTokens
207 blockData.rollupVars = rollupData.rollupVars
208 }
209
210 if auctionData != nil {
211 blockData.bids = auctionData.bids
212 blockData.coordinators = auctionData.coordinators
213 blockData.auctionVars = auctionData.auctionVars
214 }
215
216 if wdelayerData != nil {
217 blockData.withdrawalDelayerVars = wdelayerData
218 }
219
220 // Add rollupData and auctionData once the method is updated
221 // TODO: Save Whole Struct -> AddBlockSCData(blockData)
222 err = s.historyDB.AddBlock(blockData.block)
223 if err != nil {
224 return err
225 }
226 }
227
228 return nil
229}
230
231// reorg manages a reorg, updating History and State DB as needed. Keeps
232// checking previous blocks from the HistoryDB against the blockchain until a
233// block hash match is found. All future blocks in the HistoryDB and
234// corresponding batches in StateBD are discarded. Returns the last valid
235// blockNum from the HistoryDB.
236func (s *Synchronizer) reorg(uncleBlock *common.Block) (int64, error) {
237 var block *common.Block
238 blockNum := uncleBlock.EthBlockNum
239 found := false
240
241 log.Debugf("Reorg first uncle block: %v", blockNum)
242
243 // Iterate History DB and the blokchain looking for the latest valid block
244 for !found && blockNum > s.firstSavedBlock.EthBlockNum {
245 ethBlock, err := s.ethClient.EthBlockByNumber(context.Background(), blockNum)
246 if err != nil {
247 return 0, err
248 }
249
250 block, err = s.historyDB.GetBlock(blockNum)
251 if err != nil {
252 return 0, err
253 }
254 if block.Hash == ethBlock.Hash {
255 found = true
256 log.Debugf("Found valid block: %v", blockNum)
257 } else {
258 log.Debugf("Discarding block: %v", blockNum)
259 }
260
261 blockNum--
262 }
263
264 if found {
265 // Set History DB and State DB to the correct state
266 err := s.historyDB.Reorg(block.EthBlockNum)
267 if err != nil {
268 return 0, err
269 }
270
271 batchNum, err := s.historyDB.GetLastBatchNum()
272 if err != nil && err != sql.ErrNoRows {
273 return 0, err
274 }
275 if batchNum != 0 {
276 err = s.stateDB.Reset(batchNum)
277 if err != nil {
278 return 0, err
279 }
280 }
281
282 return block.EthBlockNum, nil
283 }
284
285 return 0, ErrNotAbleToSync
286}
287
288// Status returns current status values from the Synchronizer
289func (s *Synchronizer) Status() (*common.SyncStatus, error) {
290 // Avoid possible inconsistencies
291 s.mux.Lock()
292 defer s.mux.Unlock()
293
294 var status *common.SyncStatus
295
296 // TODO: Join all queries to the DB into a single transaction so that
297 // we can remove the mutex locking here:
298 // - HistoryDB.GetLastBlock
299 // - HistoryDB.GetLastBatchNum
300 // - HistoryDB.GetCurrentForgerAddr
301 // - HistoryDB.GetNextForgerAddr
302
303 // Get latest block in History DB
304 lastSavedBlock, err := s.historyDB.GetLastBlock()
305 if err != nil {
306 return nil, err
307 }
308 status.CurrentBlock = lastSavedBlock.EthBlockNum
309
310 // Get latest batch in History DB
311 lastSavedBatch, err := s.historyDB.GetLastBatchNum()
312 if err != nil && err != sql.ErrNoRows {
313 return nil, err
314 }
315 status.CurrentBatch = lastSavedBatch
316
317 // Get latest blockNum in blockchain
318 latestBlockNum, err := s.ethClient.EthCurrentBlock()
319 if err != nil {
320 return nil, err
321 }
322
323 // TODO: Get CurrentForgerAddr & NextForgerAddr from the Auction SC / Or from the HistoryDB
324
325 // Check if Synchronizer is synchronized
326 status.Synchronized = status.CurrentBlock == latestBlockNum
327 return status, nil
328}
329
330// rollupSync gets information from the Rollup Contract
331func (s *Synchronizer) rollupSync(blockNum int64) (*rollupData, error) {
332 var rollupData = newRollupData()
333 // var forgeL1TxsNum int64
334 var numAccounts int
335
336 // Get rollup events in the block
337 rollupEvents, _, err := s.ethClient.RollupEventsByBlock(blockNum)
338 if err != nil {
339 return nil, err
340 }
341
342 // TODO: Replace GetLastL1TxsNum by GetNextL1TxsNum
343 var nextForgeL1TxsNum int64
344 nextForgeL1TxsNumPtr, err := s.historyDB.GetLastL1TxsNum()
345 if err != nil {
346 return nil, err
347 }
348 if nextForgeL1TxsNumPtr != nil {
349 nextForgeL1TxsNum = *nextForgeL1TxsNumPtr + 1
350 } else {
351 nextForgeL1TxsNum = 0
352 }
353
354 // Get newLastIdx that will be used to complete the accounts
355 // idx, err := s.getIdx(rollupEvents)
356 // if err != nil {
357 // return nil, err
358 // }
359
360 // Get L1UserTX
361 rollupData.l1Txs, err = getL1UserTx(rollupEvents.L1UserTx, blockNum)
362 if err != nil {
363 return nil, err
364 }
365
366 // Get ForgeBatch events to get the L1CoordinatorTxs
367 for _, fbEvent := range rollupEvents.ForgeBatch {
368 batchData := NewBatchData()
369 position := 0
370
371 // Get the input for each Tx
372 forgeBatchArgs, err := s.ethClient.RollupForgeBatchArgs(fbEvent.EthTxHash)
373 if err != nil {
374 return nil, err
375 }
376 forgeL1TxsNum := nextForgeL1TxsNum
377 // Check if this is a L1Batch to get L1 Tx from it
378 if forgeBatchArgs.L1Batch {
379 // Get L1 User Txs from History DB
380 // TODO: Get L1TX from HistoryDB filtered by toforgeL1txNum & fromidx = 0 and
381 // update batch number and add accounts to createdAccounts updating idx
382
383 // l1UserTxs, err := s.historyDB.GetL1UserTxs(nextForgeL1TxsNum)
384 // If HistoryDB doesn't have L1UserTxs at
385 // nextForgeL1TxsNum, check if they exist in
386 // rollupData.l1Txs. This could happen because in a
387 // block there could be multiple batches with L1Batch =
388 // true (although it's a very rare case). If the
389 // L1UserTxs are not in rollupData.l1Txs, use an empty
390 // array (this happens when the L1UserTxs queue is
391 // frozen but didn't store any tx).
392 l1UserTxs := []common.L1Tx{}
393 position = len(l1UserTxs)
394
395 // Get L1 Coordinator Txs
396 for _, l1CoordinatorTx := range forgeBatchArgs.L1CoordinatorTxs {
397 l1CoordinatorTx.Position = position
398 l1CoordinatorTx.ToForgeL1TxsNum = &forgeL1TxsNum
399 l1CoordinatorTx.UserOrigin = false
400 l1CoordinatorTx.EthBlockNum = blockNum
401 bn := new(common.BatchNum)
402 *bn = common.BatchNum(fbEvent.BatchNum)
403 l1CoordinatorTx.BatchNum = bn
404 l1CoordinatorTx, err = common.NewL1Tx(l1CoordinatorTx)
405 if err != nil {
406 return nil, err
407 }
408
409 batchData.l1CoordinatorTxs = append(batchData.l1CoordinatorTxs, l1CoordinatorTx)
410
411 // Check if we have to register an account
412 // if l1CoordinatorTx.FromIdx == 0 {
413 // account := common.Account{
414 // // TODO: Uncommnent when common.account has IDx
415 // // IDx: common.Idx(idx),
416 // TokenID: l1CoordinatorTx.TokenID,
417 // Nonce: 0,
418 // Balance: l1CoordinatorTx.LoadAmount,
419 // PublicKey: l1CoordinatorTx.FromBJJ,
420 // EthAddr: l1CoordinatorTx.FromEthAddr,
421 // }
422 // idx++
423 // batchData.createdAccounts = append(batchData.createdAccounts, &account)
424 // numAccounts++
425 // }
426 position++
427 }
428 nextForgeL1TxsNum++
429 }
430
431 // Get L2Txs
432 poolL2Txs := common.L2TxsToPoolL2Txs(forgeBatchArgs.L2Txs) // TODO: This is a big uggly, find a better way
433
434 // Get exitTree
435 // TODO: Get createdAccounts from ProcessTxs()
436 // TODO: Get CollectedFees from ProcessTxs()
437 // TODO: Pass forgeBatchArgs.FeeIdxCoordinator to ProcessTxs()
438 _, exitInfo, err := s.stateDB.ProcessTxs(batchData.l1UserTxs, batchData.l1CoordinatorTxs, poolL2Txs)
439 if err != nil {
440 return nil, err
441 }
442
443 l2Txs, err := common.PoolL2TxsToL2Txs(poolL2Txs) // TODO: This is a big uggly, find a better way
444 if err != nil {
445 return nil, err
446 }
447 batchData.l2Txs = append(batchData.l2Txs, l2Txs...)
448
449 batchData.exitTree = exitInfo
450
451 // Get Batch information
452 batch := &common.Batch{
453 BatchNum: common.BatchNum(fbEvent.BatchNum),
454 EthBlockNum: blockNum,
455 // ForgerAddr: , TODO: Get it from ethClient -> Add ForgerAddr to RollupEventForgeBatch
456 // CollectedFees: , TODO: Clarify where to get them if they are still needed
457 StateRoot: common.Hash(forgeBatchArgs.NewStRoot.Bytes()),
458 NumAccounts: numAccounts,
459 ExitRoot: common.Hash(forgeBatchArgs.NewExitRoot.Bytes()),
460 ForgeL1TxsNum: &forgeL1TxsNum,
461 // SlotNum: TODO: Calculate once ethClient provides the info // calculate from blockNum + ethClient Constants
462 }
463 batchData.batch = batch
464 rollupData.batches = append(rollupData.batches, batchData)
465 }
466
467 // Get Registered Tokens
468 for _, eAddToken := range rollupEvents.AddToken {
469 var token *common.Token
470
471 token.TokenID = common.TokenID(eAddToken.TokenID)
472 token.EthAddr = eAddToken.Address
473 token.EthBlockNum = blockNum
474
475 // TODO: Add external information consulting SC about it using Address
476 rollupData.registeredTokens = append(rollupData.registeredTokens, token)
477 }
478
479 // TODO: rollupEvents.UpdateForgeL1L2BatchTimeout
480 // TODO: rollupEvents.UpdateFeeAddToken
481 // TODO: rollupEvents.WithdrawEvent
482
483 // TODO: Emergency Mechanism
484 // TODO: Variables
485 // TODO: Constants
486
487 return &rollupData, nil
488}
489
490// auctionSync gets information from the Auction Contract
491func (s *Synchronizer) auctionSync(blockNum int64) (*auctionData, error) {
492 var auctionData = newAuctionData()
493
494 // Get auction events in the block
495 auctionEvents, _, err := s.ethClient.AuctionEventsByBlock(blockNum)
496 if err != nil {
497 return nil, err
498 }
499
500 // Get bids
501 for _, eNewBid := range auctionEvents.NewBid {
502 bid := &common.Bid{
503 SlotNum: common.SlotNum(eNewBid.Slot),
504 BidValue: eNewBid.BidAmount,
505 ForgerAddr: eNewBid.CoordinatorForger,
506 EthBlockNum: blockNum,
507 }
508 auctionData.bids = append(auctionData.bids, bid)
509 }
510
511 // Get Coordinators
512 for _, eNewCoordinator := range auctionEvents.NewCoordinator {
513 coordinator := &common.Coordinator{
514 Forger: eNewCoordinator.ForgerAddress,
515 WithdrawAddr: eNewCoordinator.WithdrawalAddress,
516 URL: eNewCoordinator.CoordinatorURL,
517 }
518 auctionData.coordinators = append(auctionData.coordinators, coordinator)
519 }
520
521 // TODO: NewSlotDeadline
522 // TODO: NewClosedAuctionSlots
523 // TODO: NewOutbidding
524 // TODO: NewDonationAddress
525 // TODO: NewBootCoordinator
526 // TODO: NewOpenAuctionSlots
527 // TODO: NewAllocationRatio
528 // TODO: NewForgeAllocated
529 // TODO: NewDefaultSlotSetBid
530 // TODO: NewForge
531 // TODO: HEZClaimed
532
533 // TODO: Think about separating new coordinaors from coordinator updated
534
535 // Get Coordinators from updates
536 for _, eCoordinatorUpdated := range auctionEvents.CoordinatorUpdated {
537 coordinator := &common.Coordinator{
538 Forger: eCoordinatorUpdated.ForgerAddress,
539 WithdrawAddr: eCoordinatorUpdated.WithdrawalAddress,
540 URL: eCoordinatorUpdated.CoordinatorURL,
541 }
542 auctionData.coordinators = append(auctionData.coordinators, coordinator)
543 }
544
545 // TODO: VARS
546 // TODO: CONSTANTS
547
548 return auctionData, nil
549}
550
551// wdelayerSync gets information from the Withdrawal Delayer Contract
552func (s *Synchronizer) wdelayerSync(blockNum int64) (*common.WithdrawalDelayerVars, error) {
553 // TODO: VARS
554 // TODO: CONSTANTS
555
556 return nil, nil
557}
558
559// func (s *Synchronizer) getIdx(rollupEvents *eth.RollupEvents) (int64, error) {
560// // TODO: FIXME: There will be an error here when `len(rollupEvents.ForgeBatch) == 0`
561// lastForgeBatch := rollupEvents.ForgeBatch[len(rollupEvents.ForgeBatch)-1]
562//
563// // TODO: RollupForgeBatchArgs is already called in `rollupSync`.
564// // Ideally it should not need to be called twice for the same batch.
565// // Get the input for forgeBatch
566// forgeBatchArgs, err := s.ethClient.RollupForgeBatchArgs(lastForgeBatch.EthTxHash)
567// if err != nil {
568// return 0, err
569// }
570//
571// return forgeBatchArgs.NewLastIdx + 1, nil
572// }
573
574func getL1UserTx(l1UserTxEvents []eth.RollupEventL1UserTx, blockNum int64) ([]*common.L1Tx, error) {
575 l1Txs := make([]*common.L1Tx, 0)
576
577 for _, eL1UserTx := range l1UserTxEvents {
578 // Fill aditional Tx fields
579 toForge := new(int64)
580 *toForge = eL1UserTx.ToForgeL1TxsNum
581 eL1UserTx.L1Tx.ToForgeL1TxsNum = toForge
582 eL1UserTx.L1Tx.Position = eL1UserTx.Position
583 eL1UserTx.L1Tx.UserOrigin = true
584 eL1UserTx.L1Tx.EthBlockNum = blockNum
585 nL1Tx, err := common.NewL1Tx(&eL1UserTx.L1Tx)
586 if err != nil {
587 return nil, err
588 }
589 eL1UserTx.L1Tx = *nL1Tx
590
591 l1Txs = append(l1Txs, &eL1UserTx.L1Tx)
592 }
593 return l1Txs, nil
594}