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