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