hermez-node/db/statedb/statedb.go
db/statedb/statedb.goRaw
1package statedb
2
3import (
4 "errors"
5 "fmt"
6 "os"
7 "strconv"
8
9 "github.com/hermeznetwork/hermez-node/common"
10 "github.com/hermeznetwork/hermez-node/log"
11 "github.com/iden3/go-merkletree"
12 "github.com/iden3/go-merkletree/db"
13 "github.com/iden3/go-merkletree/db/pebble"
14)
15
16// TODO(Edu): Document here how StateDB is kept consistent
17
18var (
19 // ErrStateDBWithoutMT is used when a method that requires a MerkleTree
20 // is called in a StateDB that does not have a MerkleTree defined
21 ErrStateDBWithoutMT = errors.New("Can not call method to use MerkleTree in a StateDB without MerkleTree")
22
23 // ErrAccountAlreadyExists is used when CreateAccount is called and the
24 // Account already exists
25 ErrAccountAlreadyExists = errors.New("Can not CreateAccount because Account already exists")
26
27 // ErrToIdxNotFound is used when trying to get the ToIdx from ToEthAddr
28 // or ToEthAddr&ToBJJ
29 ErrToIdxNotFound = errors.New("ToIdx can not be found")
30
31 // KeyCurrentBatch is used as key in the db to store the current BatchNum
32 KeyCurrentBatch = []byte("currentbatch")
33)
34
35const (
36 // PathStateDB defines the subpath of the StateDB
37 PathStateDB = "/statedb"
38 // PathBatchNum defines the subpath of the Batch Checkpoint in the
39 // subpath of the StateDB
40 PathBatchNum = "/BatchNum"
41 // PathCurrent defines the subpath of the current Batch in the subpath
42 // of the StateDB
43 PathCurrent = "/current"
44 // TypeSynchronizer defines a StateDB used by the Synchronizer, that
45 // generates the ExitTree when processing the txs
46 TypeSynchronizer = "synchronizer"
47 // TypeTxSelector defines a StateDB used by the TxSelector, without
48 // computing ExitTree neither the ZKInputs
49 TypeTxSelector = "txselector"
50 // TypeBatchBuilder defines a StateDB used by the BatchBuilder, that
51 // generates the ExitTree and the ZKInput when processing the txs
52 TypeBatchBuilder = "batchbuilder"
53)
54
55// TypeStateDB determines the type of StateDB
56type TypeStateDB string
57
58// StateDB represents the StateDB object
59type StateDB struct {
60 path string
61 currentBatch common.BatchNum
62 db *pebble.PebbleStorage
63 mt *merkletree.MerkleTree
64 typ TypeStateDB
65 // idx holds the current Idx that the BatchBuilder is using
66 idx common.Idx
67 zki *common.ZKInputs
68 i int // i is the current transaction index in the ZKInputs generation (zki)
69}
70
71// NewStateDB creates a new StateDB, allowing to use an in-memory or in-disk
72// storage
73func NewStateDB(path string, typ TypeStateDB, nLevels int) (*StateDB, error) {
74 var sto *pebble.PebbleStorage
75 var err error
76 sto, err = pebble.NewPebbleStorage(path+PathStateDB+PathCurrent, false)
77 if err != nil {
78 return nil, err
79 }
80
81 var mt *merkletree.MerkleTree = nil
82 if typ == TypeSynchronizer || typ == TypeBatchBuilder {
83 mt, err = merkletree.NewMerkleTree(sto, nLevels)
84 if err != nil {
85 return nil, err
86 }
87 }
88 if typ == TypeTxSelector && nLevels != 0 {
89 return nil, fmt.Errorf("invalid StateDB parameters: StateDB type==TypeStateDB can not have nLevels!=0")
90 }
91
92 sdb := &StateDB{
93 path: path + PathStateDB,
94 db: sto,
95 mt: mt,
96 typ: typ,
97 }
98
99 // load currentBatch
100 sdb.currentBatch, err = sdb.GetCurrentBatch()
101 if err != nil {
102 return nil, err
103 }
104
105 // make reset (get checkpoint) at currentBatch
106 err = sdb.reset(sdb.currentBatch, false)
107 if err != nil {
108 return nil, err
109 }
110
111 return sdb, nil
112}
113
114// DB returns the *pebble.PebbleStorage from the StateDB
115func (s *StateDB) DB() *pebble.PebbleStorage {
116 return s.db
117}
118
119// GetCurrentBatch returns the current BatchNum stored in the StateDB
120func (s *StateDB) GetCurrentBatch() (common.BatchNum, error) {
121 cbBytes, err := s.db.Get(KeyCurrentBatch)
122 if err == db.ErrNotFound {
123 return 0, nil
124 }
125 if err != nil {
126 return 0, err
127 }
128 return common.BatchNumFromBytes(cbBytes)
129}
130
131// setCurrentBatch stores the current BatchNum in the StateDB
132func (s *StateDB) setCurrentBatch() error {
133 tx, err := s.db.NewTx()
134 if err != nil {
135 return err
136 }
137 err = tx.Put(KeyCurrentBatch, s.currentBatch.Bytes())
138 if err != nil {
139 return err
140 }
141 if err := tx.Commit(); err != nil {
142 return err
143 }
144 return nil
145}
146
147// MakeCheckpoint does a checkpoint at the given batchNum in the defined path. Internally this advances & stores the current BatchNum, and then stores a Checkpoint of the current state of the StateDB.
148func (s *StateDB) MakeCheckpoint() error {
149 // advance currentBatch
150 s.currentBatch++
151
152 checkpointPath := s.path + PathBatchNum + strconv.Itoa(int(s.currentBatch))
153
154 err := s.setCurrentBatch()
155 if err != nil {
156 return err
157 }
158
159 // if checkpoint BatchNum already exist in disk, delete it
160 if _, err := os.Stat(checkpointPath); !os.IsNotExist(err) {
161 err := os.RemoveAll(checkpointPath)
162 if err != nil {
163 return err
164 }
165 } else if err != nil && !os.IsNotExist(err) {
166 return err
167 }
168
169 // execute Checkpoint
170 err = s.db.Pebble().Checkpoint(checkpointPath)
171 if err != nil {
172 return err
173 }
174
175 return nil
176}
177
178// DeleteCheckpoint removes if exist the checkpoint of the given batchNum
179func (s *StateDB) DeleteCheckpoint(batchNum common.BatchNum) error {
180 checkpointPath := s.path + PathBatchNum + strconv.Itoa(int(batchNum))
181
182 if _, err := os.Stat(checkpointPath); os.IsNotExist(err) {
183 return fmt.Errorf("Checkpoint with batchNum %d does not exist in DB", batchNum)
184 }
185
186 return os.RemoveAll(checkpointPath)
187}
188
189func pebbleMakeCheckpoint(source, dest string) error {
190 // Remove dest folder (if it exists) before doing the checkpoint
191 if _, err := os.Stat(dest); !os.IsNotExist(err) {
192 err := os.RemoveAll(dest)
193 if err != nil {
194 return err
195 }
196 } else if err != nil && !os.IsNotExist(err) {
197 return err
198 }
199
200 sto, err := pebble.NewPebbleStorage(source, false)
201 if err != nil {
202 return err
203 }
204 defer func() {
205 errClose := sto.Pebble().Close()
206 if errClose != nil {
207 log.Errorw("Pebble.Close", "err", errClose)
208 }
209 }()
210
211 // execute Checkpoint
212 err = sto.Pebble().Checkpoint(dest)
213 if err != nil {
214 return err
215 }
216
217 return nil
218}
219
220// Reset resets the StateDB to the checkpoint at the given batchNum. Reset
221// does not delete the checkpoints between old current and the new current,
222// those checkpoints will remain in the storage, and eventually will be
223// deleted when MakeCheckpoint overwrites them.
224func (s *StateDB) Reset(batchNum common.BatchNum) error {
225 return s.reset(batchNum, true)
226}
227
228// reset resets the StateDB to the checkpoint at the given batchNum. Reset
229// does not delete the checkpoints between old current and the new current,
230// those checkpoints will remain in the storage, and eventually will be
231// deleted when MakeCheckpoint overwrites them. `closeCurrent` will close the
232// currently opened db before doing the reset.
233func (s *StateDB) reset(batchNum common.BatchNum, closeCurrent bool) error {
234 checkpointPath := s.path + PathBatchNum + strconv.Itoa(int(batchNum))
235 currentPath := s.path + PathCurrent
236
237 if closeCurrent {
238 if err := s.db.Pebble().Close(); err != nil {
239 return err
240 }
241 }
242 // remove 'current'
243 err := os.RemoveAll(currentPath)
244 if err != nil {
245 return err
246 }
247 if batchNum == 0 {
248 // if batchNum == 0, open the new fresh 'current'
249 sto, err := pebble.NewPebbleStorage(currentPath, false)
250 if err != nil {
251 return err
252 }
253 s.db = sto
254 s.idx = 255
255 s.currentBatch = batchNum
256 return nil
257 }
258
259 // copy 'BatchNumX' to 'current'
260 err = pebbleMakeCheckpoint(checkpointPath, currentPath)
261 if err != nil {
262 return err
263 }
264
265 // open the new 'current'
266 sto, err := pebble.NewPebbleStorage(currentPath, false)
267 if err != nil {
268 return err
269 }
270 s.db = sto
271
272 // get currentBatch num
273 s.currentBatch, err = s.GetCurrentBatch()
274 if err != nil {
275 return err
276 }
277 // idx is obtained from the statedb reset
278 s.idx, err = s.getIdx()
279 if err != nil {
280 return err
281 }
282
283 if s.mt != nil {
284 // open the MT for the current s.db
285 mt, err := merkletree.NewMerkleTree(s.db, s.mt.MaxLevels())
286 if err != nil {
287 return err
288 }
289 s.mt = mt
290 }
291
292 return nil
293}
294
295// GetAccount returns the account for the given Idx
296func (s *StateDB) GetAccount(idx common.Idx) (*common.Account, error) {
297 return getAccountInTreeDB(s.db, idx)
298}
299
300// getAccountInTreeDB is abstracted from StateDB to be used from StateDB and
301// from ExitTree. GetAccount returns the account for the given Idx
302func getAccountInTreeDB(sto db.Storage, idx common.Idx) (*common.Account, error) {
303 idxBytes, err := idx.Bytes()
304 if err != nil {
305 return nil, err
306 }
307 vBytes, err := sto.Get(idxBytes[:])
308 if err != nil {
309 return nil, err
310 }
311 accBytes, err := sto.Get(vBytes)
312 if err != nil {
313 return nil, err
314 }
315 var b [32 * common.NLeafElems]byte
316 copy(b[:], accBytes)
317 return common.AccountFromBytes(b)
318}
319
320// CreateAccount creates a new Account in the StateDB for the given Idx. If
321// StateDB.mt==nil, MerkleTree is not affected, otherwise updates the
322// MerkleTree, returning a CircomProcessorProof.
323func (s *StateDB) CreateAccount(idx common.Idx, account *common.Account) (*merkletree.CircomProcessorProof, error) {
324 cpp, err := createAccountInTreeDB(s.db, s.mt, idx, account)
325 if err != nil {
326 return cpp, err
327 }
328 // store idx by EthAddr & BJJ
329 err = s.setIdxByEthAddrBJJ(idx, account.EthAddr, account.PublicKey)
330 return cpp, err
331}
332
333// createAccountInTreeDB is abstracted from StateDB to be used from StateDB and
334// from ExitTree. Creates a new Account in the StateDB for the given Idx. If
335// StateDB.mt==nil, MerkleTree is not affected, otherwise updates the
336// MerkleTree, returning a CircomProcessorProof.
337func createAccountInTreeDB(sto db.Storage, mt *merkletree.MerkleTree, idx common.Idx, account *common.Account) (*merkletree.CircomProcessorProof, error) {
338 // store at the DB the key: v, and value: leaf.Bytes()
339 v, err := account.HashValue()
340 if err != nil {
341 return nil, err
342 }
343 accountBytes, err := account.Bytes()
344 if err != nil {
345 return nil, err
346 }
347
348 // store the Leaf value
349 tx, err := sto.NewTx()
350 if err != nil {
351 return nil, err
352 }
353
354 idxBytes, err := idx.Bytes()
355 if err != nil {
356 return nil, err
357 }
358 _, err = tx.Get(idxBytes[:])
359 if err != db.ErrNotFound {
360 return nil, ErrAccountAlreadyExists
361 }
362
363 err = tx.Put(v.Bytes(), accountBytes[:])
364 if err != nil {
365 return nil, err
366 }
367 err = tx.Put(idxBytes[:], v.Bytes())
368 if err != nil {
369 return nil, err
370 }
371
372 if err := tx.Commit(); err != nil {
373 return nil, err
374 }
375
376 if mt != nil {
377 return mt.AddAndGetCircomProof(idx.BigInt(), v)
378 }
379
380 return nil, nil
381}
382
383// UpdateAccount updates the Account in the StateDB for the given Idx. If
384// StateDB.mt==nil, MerkleTree is not affected, otherwise updates the
385// MerkleTree, returning a CircomProcessorProof.
386func (s *StateDB) UpdateAccount(idx common.Idx, account *common.Account) (*merkletree.CircomProcessorProof, error) {
387 return updateAccountInTreeDB(s.db, s.mt, idx, account)
388}
389
390// updateAccountInTreeDB is abstracted from StateDB to be used from StateDB and
391// from ExitTree. Updates the Account in the StateDB for the given Idx. If
392// StateDB.mt==nil, MerkleTree is not affected, otherwise updates the
393// MerkleTree, returning a CircomProcessorProof.
394func updateAccountInTreeDB(sto db.Storage, mt *merkletree.MerkleTree, idx common.Idx, account *common.Account) (*merkletree.CircomProcessorProof, error) {
395 // store at the DB the key: v, and value: account.Bytes()
396 v, err := account.HashValue()
397 if err != nil {
398 return nil, err
399 }
400 accountBytes, err := account.Bytes()
401 if err != nil {
402 return nil, err
403 }
404
405 tx, err := sto.NewTx()
406 if err != nil {
407 return nil, err
408 }
409 err = tx.Put(v.Bytes(), accountBytes[:])
410 if err != nil {
411 return nil, err
412 }
413 idxBytes, err := idx.Bytes()
414 if err != nil {
415 return nil, err
416 }
417 err = tx.Put(idxBytes[:], v.Bytes())
418 if err != nil {
419 return nil, err
420 }
421
422 if err := tx.Commit(); err != nil {
423 return nil, err
424 }
425
426 if mt != nil {
427 return mt.Update(idx.BigInt(), v)
428 }
429 return nil, nil
430}
431
432// MTGetProof returns the CircomVerifierProof for a given Idx
433func (s *StateDB) MTGetProof(idx common.Idx) (*merkletree.CircomVerifierProof, error) {
434 if s.mt == nil {
435 return nil, ErrStateDBWithoutMT
436 }
437 return s.mt.GenerateCircomVerifierProof(idx.BigInt(), s.mt.Root())
438}
439
440// LocalStateDB represents the local StateDB which allows to make copies from
441// the synchronizer StateDB, and is used by the tx-selector and the
442// batch-builder. LocalStateDB is an in-memory storage.
443type LocalStateDB struct {
444 *StateDB
445 synchronizerStateDB *StateDB
446}
447
448// NewLocalStateDB returns a new LocalStateDB connected to the given
449// synchronizerDB
450func NewLocalStateDB(path string, synchronizerDB *StateDB, typ TypeStateDB, nLevels int) (*LocalStateDB, error) {
451 s, err := NewStateDB(path, typ, nLevels)
452 if err != nil {
453 return nil, err
454 }
455 return &LocalStateDB{
456 s,
457 synchronizerDB,
458 }, nil
459}
460
461// Reset performs a reset in the LocaStateDB. If fromSynchronizer is true, it
462// gets the state from LocalStateDB.synchronizerStateDB for the given batchNum. If fromSynchronizer is false, get the state from LocalStateDB checkpoints.
463func (l *LocalStateDB) Reset(batchNum common.BatchNum, fromSynchronizer bool) error {
464 if batchNum == 0 {
465 l.idx = 0
466 return nil
467 }
468
469 synchronizerCheckpointPath := l.synchronizerStateDB.path + PathBatchNum + strconv.Itoa(int(batchNum))
470 checkpointPath := l.path + PathBatchNum + strconv.Itoa(int(batchNum))
471 currentPath := l.path + PathCurrent
472
473 if fromSynchronizer {
474 // use checkpoint from SynchronizerStateDB
475 if _, err := os.Stat(synchronizerCheckpointPath); os.IsNotExist(err) {
476 // if synchronizerStateDB does not have checkpoint at batchNum, return err
477 return fmt.Errorf("Checkpoint not exist in Synchronizer")
478 }
479
480 if err := l.db.Pebble().Close(); err != nil {
481 return err
482 }
483 // remove 'current'
484 err := os.RemoveAll(currentPath)
485 if err != nil {
486 return err
487 }
488 // copy synchronizer'BatchNumX' to 'current'
489 err = pebbleMakeCheckpoint(synchronizerCheckpointPath, currentPath)
490 if err != nil {
491 return err
492 }
493 // copy synchronizer'BatchNumX' to 'BatchNumX'
494 err = pebbleMakeCheckpoint(synchronizerCheckpointPath, checkpointPath)
495 if err != nil {
496 return err
497 }
498
499 // open the new 'current'
500 sto, err := pebble.NewPebbleStorage(currentPath, false)
501 if err != nil {
502 return err
503 }
504 l.db = sto
505
506 // get currentBatch num
507 l.currentBatch, err = l.GetCurrentBatch()
508 if err != nil {
509 return err
510 }
511 // open the MT for the current s.db
512 mt, err := merkletree.NewMerkleTree(l.db, l.mt.MaxLevels())
513 if err != nil {
514 return err
515 }
516 l.mt = mt
517
518 return nil
519 }
520 // use checkpoint from LocalStateDB
521 return l.StateDB.reset(batchNum, true)
522}