hermez-node/db/kvdb/kvdb.go
db/kvdb/kvdb.goRaw
1// Package kvdb provides a key-value database with Checkpoints & Resets system
2package kvdb
3
4import (
5 "fmt"
6 "io/ioutil"
7 "os"
8 "path"
9 "sort"
10 "strings"
11 "sync"
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/db"
17 "github.com/iden3/go-merkletree/db/pebble"
18)
19
20const (
21 // PathBatchNum defines the subpath of the Batch Checkpoint in the
22 // subpath of the KVDB
23 PathBatchNum = "BatchNum"
24 // PathCurrent defines the subpath of the current Batch in the subpath
25 // of the KVDB
26 PathCurrent = "current"
27 // PathLast defines the subpath of the last Batch in the subpath
28 // of the StateDB
29 PathLast = "last"
30 // DefaultKeep is the default value for the Keep parameter
31 DefaultKeep = 128
32)
33
34var (
35 // KeyCurrentBatch is used as key in the db to store the current BatchNum
36 KeyCurrentBatch = []byte("k:currentbatch")
37 // keyCurrentIdx is used as key in the db to store the CurrentIdx
38 keyCurrentIdx = []byte("k:idx")
39 // ErrNoLast is returned when the KVDB has been configured to not have
40 // a Last checkpoint but a Last method is used
41 ErrNoLast = fmt.Errorf("no last checkpoint")
42)
43
44// KVDB represents the Key-Value DB object
45type KVDB struct {
46 cfg Config
47 db *pebble.Storage
48 // CurrentIdx holds the current Idx that the BatchBuilder is using
49 CurrentIdx common.Idx
50 CurrentBatch common.BatchNum
51 m sync.Mutex
52 last *Last
53}
54
55// Last is a consistent view to the last batch of the stateDB that can
56// be queried concurrently.
57type Last struct {
58 db *pebble.Storage
59 path string
60 rw sync.RWMutex
61}
62
63func (k *Last) setNew() error {
64 k.rw.Lock()
65 defer k.rw.Unlock()
66 if k.db != nil {
67 k.db.Close()
68 k.db = nil
69 }
70 lastPath := path.Join(k.path, PathLast)
71 if err := os.RemoveAll(lastPath); err != nil {
72 return tracerr.Wrap(err)
73 }
74 db, err := pebble.NewPebbleStorage(lastPath, false)
75 if err != nil {
76 return tracerr.Wrap(err)
77 }
78 k.db = db
79 return nil
80}
81
82func (k *Last) set(kvdb *KVDB, batchNum common.BatchNum) error {
83 k.rw.Lock()
84 defer k.rw.Unlock()
85 if k.db != nil {
86 k.db.Close()
87 k.db = nil
88 }
89 lastPath := path.Join(k.path, PathLast)
90 if err := kvdb.MakeCheckpointFromTo(batchNum, lastPath); err != nil {
91 return tracerr.Wrap(err)
92 }
93 db, err := pebble.NewPebbleStorage(lastPath, false)
94 if err != nil {
95 return tracerr.Wrap(err)
96 }
97 k.db = db
98 return nil
99}
100
101func (k *Last) close() {
102 k.rw.Lock()
103 defer k.rw.Unlock()
104 if k.db != nil {
105 k.db.Close()
106 k.db = nil
107 }
108}
109
110// Config of the KVDB
111type Config struct {
112 // Path where the checkpoints will be stored
113 Path string
114 // Keep is the number of old checkpoints to keep. If 0, all
115 // checkpoints are kept.
116 Keep int
117 // At every checkpoint, check that there are no gaps between the
118 // checkpoints
119 NoGapsCheck bool
120 // NoLast skips having an opened DB with a checkpoint to the last
121 // batchNum for thread-safe reads.
122 NoLast bool
123}
124
125// NewKVDB creates a new KVDB, allowing to use an in-memory or in-disk storage.
126// Checkpoints older than the value defined by `keep` will be deleted.
127// func NewKVDB(pathDB string, keep int) (*KVDB, error) {
128func NewKVDB(cfg Config) (*KVDB, error) {
129 var sto *pebble.Storage
130 var err error
131 sto, err = pebble.NewPebbleStorage(path.Join(cfg.Path, PathCurrent), false)
132 if err != nil {
133 return nil, tracerr.Wrap(err)
134 }
135
136 var last *Last
137 if !cfg.NoLast {
138 last = &Last{
139 path: cfg.Path,
140 }
141 }
142 kvdb := &KVDB{
143 cfg: cfg,
144 db: sto,
145 last: last,
146 }
147 // load currentBatch
148 kvdb.CurrentBatch, err = kvdb.GetCurrentBatch()
149 if err != nil {
150 return nil, tracerr.Wrap(err)
151 }
152
153 // make reset (get checkpoint) at currentBatch
154 err = kvdb.reset(kvdb.CurrentBatch, true)
155 if err != nil {
156 return nil, tracerr.Wrap(err)
157 }
158
159 return kvdb, nil
160}
161
162// LastRead is a thread-safe method to query the last KVDB
163func (k *KVDB) LastRead(fn func(db *pebble.Storage) error) error {
164 if k.last == nil {
165 return tracerr.Wrap(ErrNoLast)
166 }
167 k.last.rw.RLock()
168 defer k.last.rw.RUnlock()
169 return fn(k.last.db)
170}
171
172// DB returns the *pebble.Storage from the KVDB
173func (k *KVDB) DB() *pebble.Storage {
174 return k.db
175}
176
177// StorageWithPrefix returns the db.Storage with the given prefix from the
178// current KVDB
179func (k *KVDB) StorageWithPrefix(prefix []byte) db.Storage {
180 return k.db.WithPrefix(prefix)
181}
182
183// Reset resets the KVDB to the checkpoint at the given batchNum. Reset does
184// not delete the checkpoints between old current and the new current, those
185// checkpoints will remain in the storage, and eventually will be deleted when
186// MakeCheckpoint overwrites them.
187func (k *KVDB) Reset(batchNum common.BatchNum) error {
188 return k.reset(batchNum, true)
189}
190
191// reset resets the KVDB to the checkpoint at the given batchNum. Reset does
192// not delete the checkpoints between old current and the new current, those
193// checkpoints will remain in the storage, and eventually will be deleted when
194// MakeCheckpoint overwrites them. `closeCurrent` will close the currently
195// opened db before doing the reset.
196func (k *KVDB) reset(batchNum common.BatchNum, closeCurrent bool) error {
197 currentPath := path.Join(k.cfg.Path, PathCurrent)
198
199 if closeCurrent && k.db != nil {
200 k.db.Close()
201 k.db = nil
202 }
203 // remove 'current'
204 if err := os.RemoveAll(currentPath); err != nil {
205 return tracerr.Wrap(err)
206 }
207 // remove all checkpoints > batchNum
208 list, err := k.ListCheckpoints()
209 if err != nil {
210 return tracerr.Wrap(err)
211 }
212 // Find first batch that is greater than batchNum, and delete
213 // everything after that
214 start := 0
215 for ; start < len(list); start++ {
216 if common.BatchNum(list[start]) > batchNum {
217 break
218 }
219 }
220 for _, bn := range list[start:] {
221 if err := k.DeleteCheckpoint(common.BatchNum(bn)); err != nil {
222 return tracerr.Wrap(err)
223 }
224 }
225
226 if batchNum == 0 {
227 // if batchNum == 0, open the new fresh 'current'
228 sto, err := pebble.NewPebbleStorage(currentPath, false)
229 if err != nil {
230 return tracerr.Wrap(err)
231 }
232 k.db = sto
233 k.CurrentIdx = common.RollupConstReservedIDx // 255
234 k.CurrentBatch = 0
235 if k.last != nil {
236 if err := k.last.setNew(); err != nil {
237 return tracerr.Wrap(err)
238 }
239 }
240
241 return nil
242 }
243
244 // copy 'batchNum' to 'current'
245 if err := k.MakeCheckpointFromTo(batchNum, currentPath); err != nil {
246 return tracerr.Wrap(err)
247 }
248 // copy 'batchNum' to 'last'
249 if k.last != nil {
250 if err := k.last.set(k, batchNum); err != nil {
251 return tracerr.Wrap(err)
252 }
253 }
254
255 // open the new 'current'
256 sto, err := pebble.NewPebbleStorage(currentPath, false)
257 if err != nil {
258 return tracerr.Wrap(err)
259 }
260 k.db = sto
261
262 // get currentBatch num
263 k.CurrentBatch, err = k.GetCurrentBatch()
264 if err != nil {
265 return tracerr.Wrap(err)
266 }
267 // idx is obtained from the statedb reset
268 k.CurrentIdx, err = k.GetCurrentIdx()
269 if err != nil {
270 return tracerr.Wrap(err)
271 }
272
273 return nil
274}
275
276// ResetFromSynchronizer performs a reset in the KVDB getting the state from
277// synchronizerKVDB for the given batchNum.
278func (k *KVDB) ResetFromSynchronizer(batchNum common.BatchNum, synchronizerKVDB *KVDB) error {
279 if synchronizerKVDB == nil {
280 return tracerr.Wrap(fmt.Errorf("synchronizerKVDB can not be nil"))
281 }
282
283 currentPath := path.Join(k.cfg.Path, PathCurrent)
284 if k.db != nil {
285 k.db.Close()
286 k.db = nil
287 }
288
289 // remove 'current'
290 if err := os.RemoveAll(currentPath); err != nil {
291 return tracerr.Wrap(err)
292 }
293 // remove all checkpoints
294 list, err := k.ListCheckpoints()
295 if err != nil {
296 return tracerr.Wrap(err)
297 }
298 for _, bn := range list {
299 if err := k.DeleteCheckpoint(common.BatchNum(bn)); err != nil {
300 return tracerr.Wrap(err)
301 }
302 }
303
304 if batchNum == 0 {
305 // if batchNum == 0, open the new fresh 'current'
306 sto, err := pebble.NewPebbleStorage(currentPath, false)
307 if err != nil {
308 return tracerr.Wrap(err)
309 }
310 k.db = sto
311 k.CurrentIdx = common.RollupConstReservedIDx // 255
312 k.CurrentBatch = 0
313
314 return nil
315 }
316
317 checkpointPath := path.Join(k.cfg.Path, fmt.Sprintf("%s%d", PathBatchNum, batchNum))
318
319 // copy synchronizer'BatchNumX' to 'BatchNumX'
320 if err := synchronizerKVDB.MakeCheckpointFromTo(batchNum, checkpointPath); err != nil {
321 return tracerr.Wrap(err)
322 }
323
324 // copy 'BatchNumX' to 'current'
325 err = k.MakeCheckpointFromTo(batchNum, currentPath)
326 if err != nil {
327 return tracerr.Wrap(err)
328 }
329
330 // open the new 'current'
331 sto, err := pebble.NewPebbleStorage(currentPath, false)
332 if err != nil {
333 return tracerr.Wrap(err)
334 }
335 k.db = sto
336
337 // get currentBatch num
338 k.CurrentBatch, err = k.GetCurrentBatch()
339 if err != nil {
340 return tracerr.Wrap(err)
341 }
342 // get currentIdx
343 k.CurrentIdx, err = k.GetCurrentIdx()
344 if err != nil {
345 return tracerr.Wrap(err)
346 }
347
348 return nil
349}
350
351// GetCurrentBatch returns the current BatchNum stored in the KVDB
352func (k *KVDB) GetCurrentBatch() (common.BatchNum, error) {
353 cbBytes, err := k.db.Get(KeyCurrentBatch)
354 if tracerr.Unwrap(err) == db.ErrNotFound {
355 return 0, nil
356 }
357 if err != nil {
358 return 0, tracerr.Wrap(err)
359 }
360 return common.BatchNumFromBytes(cbBytes)
361}
362
363// setCurrentBatch stores the current BatchNum in the KVDB
364func (k *KVDB) setCurrentBatch() error {
365 tx, err := k.db.NewTx()
366 if err != nil {
367 return tracerr.Wrap(err)
368 }
369 err = tx.Put(KeyCurrentBatch, k.CurrentBatch.Bytes())
370 if err != nil {
371 return tracerr.Wrap(err)
372 }
373 if err := tx.Commit(); err != nil {
374 return tracerr.Wrap(err)
375 }
376 return nil
377}
378
379// GetCurrentIdx returns the stored Idx from the KVDB, which is the last Idx
380// used for an Account in the k.
381func (k *KVDB) GetCurrentIdx() (common.Idx, error) {
382 idxBytes, err := k.db.Get(keyCurrentIdx)
383 if tracerr.Unwrap(err) == db.ErrNotFound {
384 return common.RollupConstReservedIDx, nil // 255, nil
385 }
386 if err != nil {
387 return 0, tracerr.Wrap(err)
388 }
389 return common.IdxFromBytes(idxBytes[:])
390}
391
392// SetCurrentIdx stores Idx in the KVDB
393func (k *KVDB) SetCurrentIdx(idx common.Idx) error {
394 k.CurrentIdx = idx
395
396 tx, err := k.db.NewTx()
397 if err != nil {
398 return tracerr.Wrap(err)
399 }
400 idxBytes, err := idx.Bytes()
401 if err != nil {
402 return tracerr.Wrap(err)
403 }
404 err = tx.Put(keyCurrentIdx, idxBytes[:])
405 if err != nil {
406 return tracerr.Wrap(err)
407 }
408 if err := tx.Commit(); err != nil {
409 return tracerr.Wrap(err)
410 }
411 return nil
412}
413
414// MakeCheckpoint does a checkpoint at the given batchNum in the defined path.
415// Internally this advances & stores the current BatchNum, and then stores a
416// Checkpoint of the current state of the k.
417func (k *KVDB) MakeCheckpoint() error {
418 // advance currentBatch
419 k.CurrentBatch++
420
421 checkpointPath := path.Join(k.cfg.Path, fmt.Sprintf("%s%d", PathBatchNum, k.CurrentBatch))
422
423 if err := k.setCurrentBatch(); err != nil {
424 return tracerr.Wrap(err)
425 }
426
427 // if checkpoint BatchNum already exist in disk, delete it
428 if _, err := os.Stat(checkpointPath); os.IsNotExist(err) {
429 } else if err != nil {
430 return tracerr.Wrap(err)
431 } else {
432 if err := os.RemoveAll(checkpointPath); err != nil {
433 return tracerr.Wrap(err)
434 }
435 }
436
437 // execute Checkpoint
438 if err := k.db.Pebble().Checkpoint(checkpointPath); err != nil {
439 return tracerr.Wrap(err)
440 }
441 // copy 'CurrentBatch' to 'last'
442 if k.last != nil {
443 if err := k.last.set(k, k.CurrentBatch); err != nil {
444 return tracerr.Wrap(err)
445 }
446 }
447 // delete old checkpoints
448 if err := k.deleteOldCheckpoints(); err != nil {
449 return tracerr.Wrap(err)
450 }
451
452 return nil
453}
454
455// CheckpointExists returns true if the checkpoint exists
456func (k *KVDB) CheckpointExists(batchNum common.BatchNum) (bool, error) {
457 source := path.Join(k.cfg.Path, fmt.Sprintf("%s%d", PathBatchNum, batchNum))
458 if _, err := os.Stat(source); os.IsNotExist(err) {
459 return false, nil
460 } else if err != nil {
461 return false, tracerr.Wrap(err)
462 }
463 return true, nil
464}
465
466// DeleteCheckpoint removes if exist the checkpoint of the given batchNum
467func (k *KVDB) DeleteCheckpoint(batchNum common.BatchNum) error {
468 checkpointPath := path.Join(k.cfg.Path, fmt.Sprintf("%s%d", PathBatchNum, batchNum))
469
470 if _, err := os.Stat(checkpointPath); os.IsNotExist(err) {
471 return tracerr.Wrap(fmt.Errorf("Checkpoint with batchNum %d does not exist in DB", batchNum))
472 } else if err != nil {
473 return tracerr.Wrap(err)
474 }
475
476 return os.RemoveAll(checkpointPath)
477}
478
479// ListCheckpoints returns the list of batchNums of the checkpoints, sorted.
480// If there's a gap between the list of checkpoints, an error is returned.
481func (k *KVDB) ListCheckpoints() ([]int, error) {
482 files, err := ioutil.ReadDir(k.cfg.Path)
483 if err != nil {
484 return nil, tracerr.Wrap(err)
485 }
486 checkpoints := []int{}
487 var checkpoint int
488 pattern := fmt.Sprintf("%s%%d", PathBatchNum)
489 for _, file := range files {
490 fileName := file.Name()
491 if file.IsDir() && strings.HasPrefix(fileName, PathBatchNum) {
492 if _, err := fmt.Sscanf(fileName, pattern, &checkpoint); err != nil {
493 return nil, tracerr.Wrap(err)
494 }
495 checkpoints = append(checkpoints, checkpoint)
496 }
497 }
498 sort.Ints(checkpoints)
499 if !k.cfg.NoGapsCheck && len(checkpoints) > 0 {
500 first := checkpoints[0]
501 for _, checkpoint := range checkpoints[1:] {
502 first++
503 if checkpoint != first {
504 log.Errorw("gap between checkpoints", "checkpoints", checkpoints)
505 return nil, tracerr.Wrap(fmt.Errorf("checkpoint gap at %v", checkpoint))
506 }
507 }
508 }
509 return checkpoints, nil
510}
511
512// deleteOldCheckpoints deletes old checkpoints when there are more than
513// `s.keep` checkpoints
514func (k *KVDB) deleteOldCheckpoints() error {
515 list, err := k.ListCheckpoints()
516 if err != nil {
517 return tracerr.Wrap(err)
518 }
519 if k.cfg.Keep > 0 && len(list) > k.cfg.Keep {
520 for _, checkpoint := range list[:len(list)-k.cfg.Keep] {
521 if err := k.DeleteCheckpoint(common.BatchNum(checkpoint)); err != nil {
522 return tracerr.Wrap(err)
523 }
524 }
525 }
526 return nil
527}
528
529// MakeCheckpointFromTo makes a checkpoint from the current db at fromBatchNum
530// to the dest folder. This method is locking, so it can be called from
531// multiple places at the same time.
532func (k *KVDB) MakeCheckpointFromTo(fromBatchNum common.BatchNum, dest string) error {
533 source := path.Join(k.cfg.Path, fmt.Sprintf("%s%d", PathBatchNum, fromBatchNum))
534 if _, err := os.Stat(source); os.IsNotExist(err) {
535 // if kvdb does not have checkpoint at batchNum, return err
536 return tracerr.Wrap(fmt.Errorf("Checkpoint \"%v\" does not exist", source))
537 } else if err != nil {
538 return tracerr.Wrap(err)
539 }
540 // By locking we allow calling MakeCheckpointFromTo from multiple
541 // places at the same time for the same stateDB. This allows the
542 // synchronizer to do a reset to a batchNum at the same time as the
543 // pipeline is doing a txSelector.Reset and batchBuilder.Reset from
544 // synchronizer to the same batchNum
545 k.m.Lock()
546 defer k.m.Unlock()
547 return PebbleMakeCheckpoint(source, dest)
548}
549
550// PebbleMakeCheckpoint is a hepler function to make a pebble checkpoint from
551// source to dest.
552func PebbleMakeCheckpoint(source, dest string) error {
553 // Remove dest folder (if it exists) before doing the checkpoint
554 if _, err := os.Stat(dest); os.IsNotExist(err) {
555 } else if err != nil {
556 return tracerr.Wrap(err)
557 } else {
558 if err := os.RemoveAll(dest); err != nil {
559 return tracerr.Wrap(err)
560 }
561 }
562
563 sto, err := pebble.NewPebbleStorage(source, false)
564 if err != nil {
565 return tracerr.Wrap(err)
566 }
567 defer sto.Close()
568
569 // execute Checkpoint
570 err = sto.Pebble().Checkpoint(dest)
571 if err != nil {
572 return tracerr.Wrap(err)
573 }
574
575 return nil
576}
577
578// Close the DB
579func (k *KVDB) Close() {
580 if k.db != nil {
581 k.db.Close()
582 k.db = nil
583 }
584 if k.last != nil {
585 k.last.close()
586 }
587}