// Package kvdb provides a key-value database with Checkpoints & Resets system package kvdb import ( "fmt" "io/ioutil" "os" "path" "sort" "strings" "sync" "github.com/hermeznetwork/hermez-node/common" "github.com/hermeznetwork/hermez-node/log" "github.com/hermeznetwork/tracerr" "github.com/iden3/go-merkletree/db" "github.com/iden3/go-merkletree/db/pebble" ) const ( // PathBatchNum defines the subpath of the Batch Checkpoint in the // subpath of the KVDB PathBatchNum = "BatchNum" // PathCurrent defines the subpath of the current Batch in the subpath // of the KVDB PathCurrent = "current" // PathLast defines the subpath of the last Batch in the subpath // of the StateDB PathLast = "last" ) var ( // KeyCurrentBatch is used as key in the db to store the current BatchNum KeyCurrentBatch = []byte("k:currentbatch") // keyCurrentIdx is used as key in the db to store the CurrentIdx keyCurrentIdx = []byte("k:idx") ) // KVDB represents the Key-Value DB object type KVDB struct { path string db *pebble.Storage // CurrentIdx holds the current Idx that the BatchBuilder is using CurrentIdx common.Idx CurrentBatch common.BatchNum keep int m sync.Mutex last *Last } // Last is a consistent view to the last batch of the stateDB that can // be queried concurrently. type Last struct { db *pebble.Storage path string rw sync.RWMutex } func (k *Last) setNew() error { k.rw.Lock() defer k.rw.Unlock() if k.db != nil { k.db.Close() } lastPath := path.Join(k.path, PathLast) err := os.RemoveAll(lastPath) if err != nil { return tracerr.Wrap(err) } db, err := pebble.NewPebbleStorage(path.Join(k.path, lastPath), false) if err != nil { return tracerr.Wrap(err) } k.db = db return nil } func (k *Last) set(kvdb *KVDB, batchNum common.BatchNum) error { k.rw.Lock() defer k.rw.Unlock() if k.db != nil { k.db.Close() } lastPath := path.Join(k.path, PathLast) if err := kvdb.MakeCheckpointFromTo(batchNum, lastPath); err != nil { return tracerr.Wrap(err) } db, err := pebble.NewPebbleStorage(lastPath, false) if err != nil { return tracerr.Wrap(err) } k.db = db return nil } func (k *Last) close() { k.rw.Lock() defer k.rw.Unlock() k.db.Close() } // NewKVDB creates a new KVDB, allowing to use an in-memory or in-disk storage. // Checkpoints older than the value defined by `keep` will be deleted. func NewKVDB(pathDB string, keep int) (*KVDB, error) { var sto *pebble.Storage var err error sto, err = pebble.NewPebbleStorage(path.Join(pathDB, PathCurrent), false) if err != nil { return nil, tracerr.Wrap(err) } kvdb := &KVDB{ path: pathDB, db: sto, keep: keep, last: &Last{ path: pathDB, }, } // load currentBatch kvdb.CurrentBatch, err = kvdb.GetCurrentBatch() if err != nil { return nil, tracerr.Wrap(err) } // make reset (get checkpoint) at currentBatch err = kvdb.reset(kvdb.CurrentBatch, true) if err != nil { return nil, tracerr.Wrap(err) } return kvdb, nil } // LastRead is a thread-safe method to query the last KVDB func (kvdb *KVDB) LastRead(fn func(db *pebble.Storage) error) error { kvdb.last.rw.RLock() defer kvdb.last.rw.RUnlock() return fn(kvdb.last.db) } // DB returns the *pebble.Storage from the KVDB func (kvdb *KVDB) DB() *pebble.Storage { return kvdb.db } // StorageWithPrefix returns the db.Storage with the given prefix from the // current KVDB func (kvdb *KVDB) StorageWithPrefix(prefix []byte) db.Storage { return kvdb.db.WithPrefix(prefix) } // Reset resets the KVDB to the checkpoint at the given batchNum. Reset does // not delete the checkpoints between old current and the new current, those // checkpoints will remain in the storage, and eventually will be deleted when // MakeCheckpoint overwrites them. func (kvdb *KVDB) Reset(batchNum common.BatchNum) error { return kvdb.reset(batchNum, true) } // reset resets the KVDB to the checkpoint at the given batchNum. Reset does // not delete the checkpoints between old current and the new current, those // checkpoints will remain in the storage, and eventually will be deleted when // MakeCheckpoint overwrites them. `closeCurrent` will close the currently // opened db before doing the reset. func (kvdb *KVDB) reset(batchNum common.BatchNum, closeCurrent bool) error { currentPath := path.Join(kvdb.path, PathCurrent) if closeCurrent { if err := kvdb.db.Pebble().Close(); err != nil { return tracerr.Wrap(err) } } // remove 'current' err := os.RemoveAll(currentPath) if err != nil { return tracerr.Wrap(err) } // remove all checkpoints > batchNum list, err := kvdb.ListCheckpoints() if err != nil { return tracerr.Wrap(err) } // Find first batch that is greater than batchNum, and delete // everything after that start := 0 for ; start < len(list); start++ { if common.BatchNum(list[start]) > batchNum { break } } for _, bn := range list[start:] { if err := kvdb.DeleteCheckpoint(common.BatchNum(bn)); err != nil { return tracerr.Wrap(err) } } if batchNum == 0 { // if batchNum == 0, open the new fresh 'current' sto, err := pebble.NewPebbleStorage(currentPath, false) if err != nil { return tracerr.Wrap(err) } kvdb.db = sto kvdb.CurrentIdx = common.RollupConstReservedIDx // 255 kvdb.CurrentBatch = 0 if err := kvdb.last.setNew(); err != nil { return tracerr.Wrap(err) } return nil } // copy 'batchNum' to 'current' if err := kvdb.MakeCheckpointFromTo(batchNum, currentPath); err != nil { return tracerr.Wrap(err) } // copy 'batchNum' to 'last' if err := kvdb.last.set(kvdb, batchNum); err != nil { return tracerr.Wrap(err) } // open the new 'current' sto, err := pebble.NewPebbleStorage(currentPath, false) if err != nil { return tracerr.Wrap(err) } kvdb.db = sto // get currentBatch num kvdb.CurrentBatch, err = kvdb.GetCurrentBatch() if err != nil { return tracerr.Wrap(err) } // idx is obtained from the statedb reset kvdb.CurrentIdx, err = kvdb.GetCurrentIdx() if err != nil { return tracerr.Wrap(err) } return nil } // ResetFromSynchronizer performs a reset in the KVDB getting the state from // synchronizerKVDB for the given batchNum. func (kvdb *KVDB) ResetFromSynchronizer(batchNum common.BatchNum, synchronizerKVDB *KVDB) error { if synchronizerKVDB == nil { return tracerr.Wrap(fmt.Errorf("synchronizerKVDB can not be nil")) } currentPath := path.Join(kvdb.path, PathCurrent) if err := kvdb.db.Pebble().Close(); err != nil { return tracerr.Wrap(err) } // remove 'current' err := os.RemoveAll(currentPath) if err != nil { return tracerr.Wrap(err) } // remove all checkpoints list, err := kvdb.ListCheckpoints() if err != nil { return tracerr.Wrap(err) } for _, bn := range list { if err := kvdb.DeleteCheckpoint(common.BatchNum(bn)); err != nil { return tracerr.Wrap(err) } } if batchNum == 0 { // if batchNum == 0, open the new fresh 'current' sto, err := pebble.NewPebbleStorage(currentPath, false) if err != nil { return tracerr.Wrap(err) } kvdb.db = sto kvdb.CurrentIdx = common.RollupConstReservedIDx // 255 kvdb.CurrentBatch = 0 return nil } checkpointPath := path.Join(kvdb.path, fmt.Sprintf("%s%d", PathBatchNum, batchNum)) // copy synchronizer'BatchNumX' to 'BatchNumX' if err := synchronizerKVDB.MakeCheckpointFromTo(batchNum, checkpointPath); err != nil { return tracerr.Wrap(err) } // copy 'BatchNumX' to 'current' err = kvdb.MakeCheckpointFromTo(batchNum, currentPath) if err != nil { return tracerr.Wrap(err) } // open the new 'current' sto, err := pebble.NewPebbleStorage(currentPath, false) if err != nil { return tracerr.Wrap(err) } kvdb.db = sto // get currentBatch num kvdb.CurrentBatch, err = kvdb.GetCurrentBatch() if err != nil { return tracerr.Wrap(err) } // get currentIdx kvdb.CurrentIdx, err = kvdb.GetCurrentIdx() if err != nil { return tracerr.Wrap(err) } return nil } // GetCurrentBatch returns the current BatchNum stored in the KVDB func (kvdb *KVDB) GetCurrentBatch() (common.BatchNum, error) { cbBytes, err := kvdb.db.Get(KeyCurrentBatch) if tracerr.Unwrap(err) == db.ErrNotFound { return 0, nil } if err != nil { return 0, tracerr.Wrap(err) } return common.BatchNumFromBytes(cbBytes) } // setCurrentBatch stores the current BatchNum in the KVDB func (kvdb *KVDB) setCurrentBatch() error { tx, err := kvdb.db.NewTx() if err != nil { return tracerr.Wrap(err) } err = tx.Put(KeyCurrentBatch, kvdb.CurrentBatch.Bytes()) if err != nil { return tracerr.Wrap(err) } if err := tx.Commit(); err != nil { return tracerr.Wrap(err) } return nil } // GetCurrentIdx returns the stored Idx from the KVDB, which is the last Idx // used for an Account in the KVDB. func (kvdb *KVDB) GetCurrentIdx() (common.Idx, error) { idxBytes, err := kvdb.db.Get(keyCurrentIdx) if tracerr.Unwrap(err) == db.ErrNotFound { return common.RollupConstReservedIDx, nil // 255, nil } if err != nil { return 0, tracerr.Wrap(err) } return common.IdxFromBytes(idxBytes[:]) } // SetCurrentIdx stores Idx in the KVDB func (kvdb *KVDB) SetCurrentIdx(idx common.Idx) error { kvdb.CurrentIdx = idx tx, err := kvdb.db.NewTx() if err != nil { return tracerr.Wrap(err) } idxBytes, err := idx.Bytes() if err != nil { return tracerr.Wrap(err) } err = tx.Put(keyCurrentIdx, idxBytes[:]) if err != nil { return tracerr.Wrap(err) } if err := tx.Commit(); err != nil { return tracerr.Wrap(err) } return nil } // 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 KVDB. func (kvdb *KVDB) MakeCheckpoint() error { // advance currentBatch kvdb.CurrentBatch++ checkpointPath := path.Join(kvdb.path, fmt.Sprintf("%s%d", PathBatchNum, kvdb.CurrentBatch)) if err := kvdb.setCurrentBatch(); err != nil { return tracerr.Wrap(err) } // if checkpoint BatchNum already exist in disk, delete it if _, err := os.Stat(checkpointPath); !os.IsNotExist(err) { err := os.RemoveAll(checkpointPath) if err != nil { return tracerr.Wrap(err) } } else if err != nil && !os.IsNotExist(err) { return tracerr.Wrap(err) } // execute Checkpoint if err := kvdb.db.Pebble().Checkpoint(checkpointPath); err != nil { return tracerr.Wrap(err) } // copy 'CurrentBatch' to 'last' if err := kvdb.last.set(kvdb, kvdb.CurrentBatch); err != nil { return tracerr.Wrap(err) } // delete old checkpoints if err := kvdb.deleteOldCheckpoints(); err != nil { return tracerr.Wrap(err) } return nil } // DeleteCheckpoint removes if exist the checkpoint of the given batchNum func (kvdb *KVDB) DeleteCheckpoint(batchNum common.BatchNum) error { checkpointPath := path.Join(kvdb.path, fmt.Sprintf("%s%d", PathBatchNum, batchNum)) if _, err := os.Stat(checkpointPath); os.IsNotExist(err) { return tracerr.Wrap(fmt.Errorf("Checkpoint with batchNum %d does not exist in DB", batchNum)) } return os.RemoveAll(checkpointPath) } // ListCheckpoints returns the list of batchNums of the checkpoints, sorted. // If there's a gap between the list of checkpoints, an error is returned. func (kvdb *KVDB) ListCheckpoints() ([]int, error) { files, err := ioutil.ReadDir(kvdb.path) if err != nil { return nil, tracerr.Wrap(err) } checkpoints := []int{} var checkpoint int pattern := fmt.Sprintf("%s%%d", PathBatchNum) for _, file := range files { fileName := file.Name() if file.IsDir() && strings.HasPrefix(fileName, PathBatchNum) { if _, err := fmt.Sscanf(fileName, pattern, &checkpoint); err != nil { return nil, tracerr.Wrap(err) } checkpoints = append(checkpoints, checkpoint) } } sort.Ints(checkpoints) if len(checkpoints) > 0 { first := checkpoints[0] for _, checkpoint := range checkpoints[1:] { first++ if checkpoint != first { log.Errorw("GAP", "checkpoints", checkpoints) return nil, tracerr.Wrap(fmt.Errorf("checkpoint gap at %v", checkpoint)) } } } return checkpoints, nil } // deleteOldCheckpoints deletes old checkpoints when there are more than // `s.keep` checkpoints func (kvdb *KVDB) deleteOldCheckpoints() error { list, err := kvdb.ListCheckpoints() if err != nil { return tracerr.Wrap(err) } if len(list) > kvdb.keep { for _, checkpoint := range list[:len(list)-kvdb.keep] { if err := kvdb.DeleteCheckpoint(common.BatchNum(checkpoint)); err != nil { return tracerr.Wrap(err) } } } return nil } // MakeCheckpointFromTo makes a checkpoint from the current db at fromBatchNum // to the dest folder. This method is locking, so it can be called from // multiple places at the same time. func (kvdb *KVDB) MakeCheckpointFromTo(fromBatchNum common.BatchNum, dest string) error { source := path.Join(kvdb.path, fmt.Sprintf("%s%d", PathBatchNum, fromBatchNum)) if _, err := os.Stat(source); os.IsNotExist(err) { // if kvdb does not have checkpoint at batchNum, return err return tracerr.Wrap(fmt.Errorf("Checkpoint \"%v\" does not exist", source)) } // By locking we allow calling MakeCheckpointFromTo from multiple // places at the same time for the same stateDB. This allows the // synchronizer to do a reset to a batchNum at the same time as the // pipeline is doing a txSelector.Reset and batchBuilder.Reset from // synchronizer to the same batchNum kvdb.m.Lock() defer kvdb.m.Unlock() return pebbleMakeCheckpoint(source, dest) } func pebbleMakeCheckpoint(source, dest string) error { // Remove dest folder (if it exists) before doing the checkpoint if _, err := os.Stat(dest); !os.IsNotExist(err) { err := os.RemoveAll(dest) if err != nil { return tracerr.Wrap(err) } } else if err != nil && !os.IsNotExist(err) { return tracerr.Wrap(err) } sto, err := pebble.NewPebbleStorage(source, false) if err != nil { return tracerr.Wrap(err) } defer func() { errClose := sto.Pebble().Close() if errClose != nil { log.Errorw("Pebble.Close", "err", errClose) } }() // execute Checkpoint err = sto.Pebble().Checkpoint(dest) if err != nil { return tracerr.Wrap(err) } return nil } // Close the DB func (kvdb *KVDB) Close() { kvdb.db.Close() kvdb.last.close() }