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