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