hermez-node/db/statedb/statedb.go
db/statedb/statedb.goRaw
1package statedb
2
3import (
4 "errors"
5 "fmt"
6 "io/ioutil"
7 "math/big"
8 "os"
9 "path"
10 "sort"
11 "strings"
12
13 "github.com/hermeznetwork/hermez-node/common"
14 "github.com/hermeznetwork/hermez-node/log"
15 "github.com/hermeznetwork/tracerr"
16 "github.com/iden3/go-merkletree"
17 "github.com/iden3/go-merkletree/db"
18 "github.com/iden3/go-merkletree/db/pebble"
19)
20
21// TODO(Edu): Document here how StateDB is kept consistent
22
23var (
24 // ErrStateDBWithoutMT is used when a method that requires a MerkleTree
25 // is called in a StateDB that does not have a MerkleTree defined
26 ErrStateDBWithoutMT = errors.New("Can not call method to use MerkleTree in a StateDB without MerkleTree")
27
28 // ErrAccountAlreadyExists is used when CreateAccount is called and the
29 // Account already exists
30 ErrAccountAlreadyExists = errors.New("Can not CreateAccount because Account already exists")
31
32 // ErrToIdxNotFound is used when trying to get the ToIdx from ToEthAddr
33 // or ToEthAddr&ToBJJ
34 ErrToIdxNotFound = errors.New("ToIdx can not be found")
35 // ErrGetIdxNoCase is used when trying to get the Idx from EthAddr &
36 // BJJ with not compatible combination
37 ErrGetIdxNoCase = errors.New("Can not get Idx due unexpected combination of ethereum Address & BabyJubJub PublicKey")
38
39 // KeyCurrentBatch is used as key in the db to store the current BatchNum
40 KeyCurrentBatch = []byte("k:currentbatch")
41
42 // PrefixKeyIdx is the key prefix for idx in the db
43 PrefixKeyIdx = []byte("i:")
44 // PrefixKeyAccHash is the key prefix for account hash in the db
45 PrefixKeyAccHash = []byte("h:")
46 // PrefixKeyMT is the key prefix for merkle tree in the db
47 PrefixKeyMT = []byte("m:")
48 // PrefixKeyAddr is the key prefix for address in the db
49 PrefixKeyAddr = []byte("a:")
50 // PrefixKeyAddrBJJ is the key prefix for address-babyjubjub in the db
51 PrefixKeyAddrBJJ = []byte("ab:")
52)
53
54const (
55 // PathBatchNum defines the subpath of the Batch Checkpoint in the
56 // subpath of the StateDB
57 PathBatchNum = "BatchNum"
58 // PathCurrent defines the subpath of the current Batch in the subpath
59 // of the StateDB
60 PathCurrent = "current"
61 // TypeSynchronizer defines a StateDB used by the Synchronizer, that
62 // generates the ExitTree when processing the txs
63 TypeSynchronizer = "synchronizer"
64 // TypeTxSelector defines a StateDB used by the TxSelector, without
65 // computing ExitTree neither the ZKInputs
66 TypeTxSelector = "txselector"
67 // TypeBatchBuilder defines a StateDB used by the BatchBuilder, that
68 // generates the ExitTree and the ZKInput when processing the txs
69 TypeBatchBuilder = "batchbuilder"
70)
71
72// TypeStateDB determines the type of StateDB
73type TypeStateDB string
74
75// StateDB represents the StateDB object
76type StateDB struct {
77 path string
78 currentBatch common.BatchNum
79 db *pebble.PebbleStorage
80 mt *merkletree.MerkleTree
81 typ TypeStateDB
82 chainID uint16
83 // idx holds the current Idx that the BatchBuilder is using
84 idx common.Idx
85 zki *common.ZKInputs
86 // i is the current transaction index in the ZKInputs generation (zki)
87 i int
88 // AccumulatedFees contains the accumulated fees for each token (Coord
89 // Idx) in the processed batch
90 AccumulatedFees map[common.Idx]*big.Int
91 keep int
92}
93
94// NewStateDB creates a new StateDB, allowing to use an in-memory or in-disk
95// storage. Checkpoints older than the value defined by `keep` will be
96// deleted.
97func NewStateDB(pathDB string, keep int, typ TypeStateDB, nLevels int, chainID uint16) (*StateDB, error) {
98 var sto *pebble.PebbleStorage
99 var err error
100 sto, err = pebble.NewPebbleStorage(path.Join(pathDB, PathCurrent), false)
101 if err != nil {
102 return nil, tracerr.Wrap(err)
103 }
104
105 var mt *merkletree.MerkleTree = nil
106 if typ == TypeSynchronizer || typ == TypeBatchBuilder {
107 mt, err = merkletree.NewMerkleTree(sto.WithPrefix(PrefixKeyMT), nLevels)
108 if err != nil {
109 return nil, tracerr.Wrap(err)
110 }
111 }
112 if typ == TypeTxSelector && nLevels != 0 {
113 return nil, tracerr.Wrap(fmt.Errorf("invalid StateDB parameters: StateDB type==TypeStateDB can not have nLevels!=0"))
114 }
115
116 sdb := &StateDB{
117 path: pathDB,
118 db: sto,
119 mt: mt,
120 typ: typ,
121 chainID: chainID,
122 keep: keep,
123 }
124
125 // load currentBatch
126 sdb.currentBatch, err = sdb.GetCurrentBatch()
127 if err != nil {
128 return nil, tracerr.Wrap(err)
129 }
130
131 // make reset (get checkpoint) at currentBatch
132 err = sdb.reset(sdb.currentBatch, false)
133 if err != nil {
134 return nil, tracerr.Wrap(err)
135 }
136
137 return sdb, nil
138}
139
140// DB returns the *pebble.PebbleStorage from the StateDB
141func (s *StateDB) DB() *pebble.PebbleStorage {
142 return s.db
143}
144
145// GetCurrentBatch returns the current BatchNum stored in the StateDB
146func (s *StateDB) GetCurrentBatch() (common.BatchNum, error) {
147 cbBytes, err := s.db.Get(KeyCurrentBatch)
148 if tracerr.Unwrap(err) == db.ErrNotFound {
149 return 0, nil
150 }
151 if err != nil {
152 return 0, tracerr.Wrap(err)
153 }
154 return common.BatchNumFromBytes(cbBytes)
155}
156
157// setCurrentBatch stores the current BatchNum in the StateDB
158func (s *StateDB) setCurrentBatch() error {
159 tx, err := s.db.NewTx()
160 if err != nil {
161 return tracerr.Wrap(err)
162 }
163 err = tx.Put(KeyCurrentBatch, s.currentBatch.Bytes())
164 if err != nil {
165 return tracerr.Wrap(err)
166 }
167 if err := tx.Commit(); err != nil {
168 return tracerr.Wrap(err)
169 }
170 return nil
171}
172
173// 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.
174func (s *StateDB) MakeCheckpoint() error {
175 // advance currentBatch
176 s.currentBatch++
177 log.Debugw("Making StateDB checkpoint", "batch", s.currentBatch, "type", s.typ)
178
179 checkpointPath := path.Join(s.path, fmt.Sprintf("%s%d", PathBatchNum, s.currentBatch))
180
181 if err := s.setCurrentBatch(); err != nil {
182 return tracerr.Wrap(err)
183 }
184
185 // if checkpoint BatchNum already exist in disk, delete it
186 if _, err := os.Stat(checkpointPath); !os.IsNotExist(err) {
187 err := os.RemoveAll(checkpointPath)
188 if err != nil {
189 return tracerr.Wrap(err)
190 }
191 } else if err != nil && !os.IsNotExist(err) {
192 return tracerr.Wrap(err)
193 }
194
195 // execute Checkpoint
196 if err := s.db.Pebble().Checkpoint(checkpointPath); err != nil {
197 return tracerr.Wrap(err)
198 }
199 // delete old checkpoints
200 if err := s.deleteOldCheckpoints(); err != nil {
201 return tracerr.Wrap(err)
202 }
203
204 return nil
205}
206
207// DeleteCheckpoint removes if exist the checkpoint of the given batchNum
208func (s *StateDB) DeleteCheckpoint(batchNum common.BatchNum) error {
209 checkpointPath := path.Join(s.path, fmt.Sprintf("%s%d", PathBatchNum, batchNum))
210
211 if _, err := os.Stat(checkpointPath); os.IsNotExist(err) {
212 return tracerr.Wrap(fmt.Errorf("Checkpoint with batchNum %d does not exist in DB", batchNum))
213 }
214
215 return os.RemoveAll(checkpointPath)
216}
217
218// listCheckpoints returns the list of batchNums of the checkpoints, sorted.
219// If there's a gap between the list of checkpoints, an error is returned.
220func (s *StateDB) listCheckpoints() ([]int, error) {
221 files, err := ioutil.ReadDir(s.path)
222 if err != nil {
223 return nil, tracerr.Wrap(err)
224 }
225 checkpoints := []int{}
226 var checkpoint int
227 pattern := fmt.Sprintf("%s%%d", PathBatchNum)
228 for _, file := range files {
229 fileName := file.Name()
230 if file.IsDir() && strings.HasPrefix(fileName, PathBatchNum) {
231 if _, err := fmt.Sscanf(fileName, pattern, &checkpoint); err != nil {
232 return nil, tracerr.Wrap(err)
233 }
234 checkpoints = append(checkpoints, checkpoint)
235 }
236 }
237 sort.Ints(checkpoints)
238 if len(checkpoints) > 0 {
239 first := checkpoints[0]
240 for _, checkpoint := range checkpoints[1:] {
241 first++
242 if checkpoint != first {
243 return nil, tracerr.Wrap(fmt.Errorf("checkpoint gap at %v", checkpoint))
244 }
245 }
246 }
247 return checkpoints, nil
248}
249
250// deleteOldCheckpoints deletes old checkpoints when there are more than
251// `s.keep` checkpoints
252func (s *StateDB) deleteOldCheckpoints() error {
253 list, err := s.listCheckpoints()
254 if err != nil {
255 return tracerr.Wrap(err)
256 }
257 if len(list) > s.keep {
258 for _, checkpoint := range list[:len(list)-s.keep] {
259 if err := s.DeleteCheckpoint(common.BatchNum(checkpoint)); err != nil {
260 return tracerr.Wrap(err)
261 }
262 }
263 }
264 return nil
265}
266
267func pebbleMakeCheckpoint(source, dest string) error {
268 // Remove dest folder (if it exists) before doing the checkpoint
269 if _, err := os.Stat(dest); !os.IsNotExist(err) {
270 err := os.RemoveAll(dest)
271 if err != nil {
272 return tracerr.Wrap(err)
273 }
274 } else if err != nil && !os.IsNotExist(err) {
275 return tracerr.Wrap(err)
276 }
277
278 sto, err := pebble.NewPebbleStorage(source, false)
279 if err != nil {
280 return tracerr.Wrap(err)
281 }
282 defer func() {
283 errClose := sto.Pebble().Close()
284 if errClose != nil {
285 log.Errorw("Pebble.Close", "err", errClose)
286 }
287 }()
288
289 // execute Checkpoint
290 err = sto.Pebble().Checkpoint(dest)
291 if err != nil {
292 return tracerr.Wrap(err)
293 }
294
295 return nil
296}
297
298// Reset resets the StateDB to the checkpoint at the given batchNum. Reset
299// does not delete the checkpoints between old current and the new current,
300// those checkpoints will remain in the storage, and eventually will be
301// deleted when MakeCheckpoint overwrites them.
302func (s *StateDB) Reset(batchNum common.BatchNum) error {
303 return s.reset(batchNum, true)
304}
305
306// reset resets the StateDB to the checkpoint at the given batchNum. Reset
307// does not delete the checkpoints between old current and the new current,
308// those checkpoints will remain in the storage, and eventually will be
309// deleted when MakeCheckpoint overwrites them. `closeCurrent` will close the
310// currently opened db before doing the reset.
311func (s *StateDB) reset(batchNum common.BatchNum, closeCurrent bool) error {
312 currentPath := path.Join(s.path, PathCurrent)
313
314 if closeCurrent {
315 if err := s.db.Pebble().Close(); err != nil {
316 return tracerr.Wrap(err)
317 }
318 }
319 // remove 'current'
320 err := os.RemoveAll(currentPath)
321 if err != nil {
322 return tracerr.Wrap(err)
323 }
324 // remove all checkpoints > batchNum
325 for i := batchNum + 1; i <= s.currentBatch; i++ {
326 if err := s.DeleteCheckpoint(i); err != nil {
327 return tracerr.Wrap(err)
328 }
329 }
330 if batchNum == 0 {
331 // if batchNum == 0, open the new fresh 'current'
332 sto, err := pebble.NewPebbleStorage(currentPath, false)
333 if err != nil {
334 return tracerr.Wrap(err)
335 }
336 s.db = sto
337 s.idx = 255
338 s.currentBatch = batchNum
339
340 if s.mt != nil {
341 // open the MT for the current s.db
342 mt, err := merkletree.NewMerkleTree(s.db.WithPrefix(PrefixKeyMT), s.mt.MaxLevels())
343 if err != nil {
344 return tracerr.Wrap(err)
345 }
346 s.mt = mt
347 }
348 return nil
349 }
350
351 checkpointPath := path.Join(s.path, fmt.Sprintf("%s%d", PathBatchNum, batchNum))
352 // copy 'BatchNumX' to 'current'
353 err = pebbleMakeCheckpoint(checkpointPath, currentPath)
354 if err != nil {
355 return tracerr.Wrap(err)
356 }
357
358 // open the new 'current'
359 sto, err := pebble.NewPebbleStorage(currentPath, false)
360 if err != nil {
361 return tracerr.Wrap(err)
362 }
363 s.db = sto
364
365 // get currentBatch num
366 s.currentBatch, err = s.GetCurrentBatch()
367 if err != nil {
368 return tracerr.Wrap(err)
369 }
370 // idx is obtained from the statedb reset
371 s.idx, err = s.GetIdx()
372 if err != nil {
373 return tracerr.Wrap(err)
374 }
375
376 if s.mt != nil {
377 // open the MT for the current s.db
378 mt, err := merkletree.NewMerkleTree(s.db.WithPrefix(PrefixKeyMT), s.mt.MaxLevels())
379 if err != nil {
380 return tracerr.Wrap(err)
381 }
382 s.mt = mt
383 }
384
385 return nil
386}
387
388// GetAccount returns the account for the given Idx
389func (s *StateDB) GetAccount(idx common.Idx) (*common.Account, error) {
390 return getAccountInTreeDB(s.db, idx)
391}
392
393// GetAccounts returns all the accounts in the db. Use for debugging pruposes
394// only.
395func (s *StateDB) GetAccounts() ([]common.Account, error) {
396 idxDB := s.db.WithPrefix(PrefixKeyIdx)
397 idxs := []common.Idx{}
398 // NOTE: Current implementation of Iterate in the pebble interface is
399 // not efficient, as it iterates over all keys. Improve it following
400 // this example: https://github.com/cockroachdb/pebble/pull/923/files
401 if err := idxDB.Iterate(func(k []byte, v []byte) (bool, error) {
402 idx, err := common.IdxFromBytes(k)
403 if err != nil {
404 return false, tracerr.Wrap(err)
405 }
406 idxs = append(idxs, idx)
407 return true, nil
408 }); err != nil {
409 return nil, tracerr.Wrap(err)
410 }
411 accs := []common.Account{}
412 for i := range idxs {
413 acc, err := s.GetAccount(idxs[i])
414 if err != nil {
415 return nil, tracerr.Wrap(err)
416 }
417 accs = append(accs, *acc)
418 }
419 return accs, nil
420}
421
422// getAccountInTreeDB is abstracted from StateDB to be used from StateDB and
423// from ExitTree. GetAccount returns the account for the given Idx
424func getAccountInTreeDB(sto db.Storage, idx common.Idx) (*common.Account, error) {
425 idxBytes, err := idx.Bytes()
426 if err != nil {
427 return nil, tracerr.Wrap(err)
428 }
429 vBytes, err := sto.Get(append(PrefixKeyIdx, idxBytes[:]...))
430 if err != nil {
431 return nil, tracerr.Wrap(err)
432 }
433 accBytes, err := sto.Get(append(PrefixKeyAccHash, vBytes...))
434 if err != nil {
435 return nil, tracerr.Wrap(err)
436 }
437 var b [32 * common.NLeafElems]byte
438 copy(b[:], accBytes)
439 account, err := common.AccountFromBytes(b)
440 if err != nil {
441 return nil, tracerr.Wrap(err)
442 }
443 account.Idx = idx
444 return account, nil
445}
446
447// CreateAccount creates a new Account in the StateDB for the given Idx. If
448// StateDB.mt==nil, MerkleTree is not affected, otherwise updates the
449// MerkleTree, returning a CircomProcessorProof.
450func (s *StateDB) CreateAccount(idx common.Idx, account *common.Account) (*merkletree.CircomProcessorProof, error) {
451 cpp, err := createAccountInTreeDB(s.db, s.mt, idx, account)
452 if err != nil {
453 return cpp, tracerr.Wrap(err)
454 }
455 // store idx by EthAddr & BJJ
456 err = s.setIdxByEthAddrBJJ(idx, account.EthAddr, account.PublicKey, account.TokenID)
457 return cpp, tracerr.Wrap(err)
458}
459
460// createAccountInTreeDB is abstracted from StateDB to be used from StateDB and
461// from ExitTree. Creates a new Account in the StateDB for the given Idx. If
462// StateDB.mt==nil, MerkleTree is not affected, otherwise updates the
463// MerkleTree, returning a CircomProcessorProof.
464func createAccountInTreeDB(sto db.Storage, mt *merkletree.MerkleTree, idx common.Idx, account *common.Account) (*merkletree.CircomProcessorProof, error) {
465 // store at the DB the key: v, and value: leaf.Bytes()
466 v, err := account.HashValue()
467 if err != nil {
468 return nil, tracerr.Wrap(err)
469 }
470 accountBytes, err := account.Bytes()
471 if err != nil {
472 return nil, tracerr.Wrap(err)
473 }
474
475 // store the Leaf value
476 tx, err := sto.NewTx()
477 if err != nil {
478 return nil, tracerr.Wrap(err)
479 }
480
481 idxBytes, err := idx.Bytes()
482 if err != nil {
483 return nil, tracerr.Wrap(err)
484 }
485 _, err = tx.Get(append(PrefixKeyIdx, idxBytes[:]...))
486 if tracerr.Unwrap(err) != db.ErrNotFound {
487 return nil, tracerr.Wrap(ErrAccountAlreadyExists)
488 }
489
490 err = tx.Put(append(PrefixKeyAccHash, v.Bytes()...), accountBytes[:])
491 if err != nil {
492 return nil, tracerr.Wrap(err)
493 }
494 err = tx.Put(append(PrefixKeyIdx, idxBytes[:]...), v.Bytes())
495 if err != nil {
496 return nil, tracerr.Wrap(err)
497 }
498
499 if err := tx.Commit(); err != nil {
500 return nil, tracerr.Wrap(err)
501 }
502
503 if mt != nil {
504 return mt.AddAndGetCircomProof(idx.BigInt(), v)
505 }
506
507 return nil, nil
508}
509
510// UpdateAccount updates the Account in the StateDB for the given Idx. If
511// StateDB.mt==nil, MerkleTree is not affected, otherwise updates the
512// MerkleTree, returning a CircomProcessorProof.
513func (s *StateDB) UpdateAccount(idx common.Idx, account *common.Account) (*merkletree.CircomProcessorProof, error) {
514 return updateAccountInTreeDB(s.db, s.mt, idx, account)
515}
516
517// updateAccountInTreeDB is abstracted from StateDB to be used from StateDB and
518// from ExitTree. Updates the Account in the StateDB for the given Idx. If
519// StateDB.mt==nil, MerkleTree is not affected, otherwise updates the
520// MerkleTree, returning a CircomProcessorProof.
521func updateAccountInTreeDB(sto db.Storage, mt *merkletree.MerkleTree, idx common.Idx, account *common.Account) (*merkletree.CircomProcessorProof, error) {
522 // store at the DB the key: v, and value: account.Bytes()
523 v, err := account.HashValue()
524 if err != nil {
525 return nil, tracerr.Wrap(err)
526 }
527 accountBytes, err := account.Bytes()
528 if err != nil {
529 return nil, tracerr.Wrap(err)
530 }
531
532 tx, err := sto.NewTx()
533 if err != nil {
534 return nil, tracerr.Wrap(err)
535 }
536 err = tx.Put(append(PrefixKeyAccHash, v.Bytes()...), accountBytes[:])
537 if err != nil {
538 return nil, tracerr.Wrap(err)
539 }
540 idxBytes, err := idx.Bytes()
541 if err != nil {
542 return nil, tracerr.Wrap(err)
543 }
544 err = tx.Put(append(PrefixKeyIdx, idxBytes[:]...), v.Bytes())
545 if err != nil {
546 return nil, tracerr.Wrap(err)
547 }
548
549 if err := tx.Commit(); err != nil {
550 return nil, tracerr.Wrap(err)
551 }
552
553 if mt != nil {
554 proof, err := mt.Update(idx.BigInt(), v)
555 return proof, tracerr.Wrap(err)
556 }
557 return nil, nil
558}
559
560// MTGetProof returns the CircomVerifierProof for a given Idx
561func (s *StateDB) MTGetProof(idx common.Idx) (*merkletree.CircomVerifierProof, error) {
562 if s.mt == nil {
563 return nil, tracerr.Wrap(ErrStateDBWithoutMT)
564 }
565 return s.mt.GenerateCircomVerifierProof(idx.BigInt(), s.mt.Root())
566}
567
568// MTGetRoot returns the current root of the underlying Merkle Tree
569func (s *StateDB) MTGetRoot() *big.Int {
570 return s.mt.Root().BigInt()
571}
572
573// MerkleTree returns the underlying StateDB merkle tree. It can be nil.
574func (s *StateDB) MerkleTree() *merkletree.MerkleTree {
575 return s.mt
576}
577
578// LocalStateDB represents the local StateDB which allows to make copies from
579// the synchronizer StateDB, and is used by the tx-selector and the
580// batch-builder. LocalStateDB is an in-memory storage.
581type LocalStateDB struct {
582 *StateDB
583 synchronizerStateDB *StateDB
584}
585
586// NewLocalStateDB returns a new LocalStateDB connected to the given
587// synchronizerDB. Checkpoints older than the value defined by `keep` will be
588// deleted.
589func NewLocalStateDB(path string, keep int, synchronizerDB *StateDB, typ TypeStateDB,
590 nLevels int) (*LocalStateDB, error) {
591 s, err := NewStateDB(path, keep, typ, nLevels, synchronizerDB.chainID)
592 if err != nil {
593 return nil, tracerr.Wrap(err)
594 }
595 return &LocalStateDB{
596 s,
597 synchronizerDB,
598 }, nil
599}
600
601// Reset performs a reset in the LocaStateDB. If fromSynchronizer is true, it
602// gets the state from LocalStateDB.synchronizerStateDB for the given batchNum. If fromSynchronizer is false, get the state from LocalStateDB checkpoints.
603func (l *LocalStateDB) Reset(batchNum common.BatchNum, fromSynchronizer bool) error {
604 if batchNum == 0 {
605 l.idx = 0
606 return nil
607 }
608
609 synchronizerCheckpointPath := path.Join(l.synchronizerStateDB.path,
610 fmt.Sprintf("%s%d", PathBatchNum, batchNum))
611 checkpointPath := path.Join(l.path, fmt.Sprintf("%s%d", PathBatchNum, batchNum))
612 currentPath := path.Join(l.path, PathCurrent)
613
614 if fromSynchronizer {
615 // use checkpoint from SynchronizerStateDB
616 if _, err := os.Stat(synchronizerCheckpointPath); os.IsNotExist(err) {
617 // if synchronizerStateDB does not have checkpoint at batchNum, return err
618 return tracerr.Wrap(fmt.Errorf("Checkpoint \"%v\" not exist in Synchronizer",
619 synchronizerCheckpointPath))
620 }
621
622 if err := l.db.Pebble().Close(); err != nil {
623 return tracerr.Wrap(err)
624 }
625 // remove 'current'
626 err := os.RemoveAll(currentPath)
627 if err != nil {
628 return tracerr.Wrap(err)
629 }
630 // copy synchronizer'BatchNumX' to 'current'
631 err = pebbleMakeCheckpoint(synchronizerCheckpointPath, currentPath)
632 if err != nil {
633 return tracerr.Wrap(err)
634 }
635 // copy synchronizer'BatchNumX' to 'BatchNumX'
636 err = pebbleMakeCheckpoint(synchronizerCheckpointPath, checkpointPath)
637 if err != nil {
638 return tracerr.Wrap(err)
639 }
640
641 // open the new 'current'
642 sto, err := pebble.NewPebbleStorage(currentPath, false)
643 if err != nil {
644 return tracerr.Wrap(err)
645 }
646 l.db = sto
647
648 // get currentBatch num
649 l.currentBatch, err = l.GetCurrentBatch()
650 if err != nil {
651 return tracerr.Wrap(err)
652 }
653 // open the MT for the current s.db
654 if l.mt != nil {
655 mt, err := merkletree.NewMerkleTree(l.db.WithPrefix(PrefixKeyMT), l.mt.MaxLevels())
656 if err != nil {
657 return tracerr.Wrap(err)
658 }
659 l.mt = mt
660 }
661
662 return nil
663 }
664 // use checkpoint from LocalStateDB
665 return l.StateDB.reset(batchNum, true)
666}