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 for i := batchNum + 1; i <= kvdb.CurrentBatch; i++ {
114 if err := kvdb.DeleteCheckpoint(i); err != nil {
115 return tracerr.Wrap(err)
116 }
117 }
118 if batchNum == 0 {
119 // if batchNum == 0, open the new fresh 'current'
120 sto, err := pebble.NewPebbleStorage(currentPath, false)
121 if err != nil {
122 return tracerr.Wrap(err)
123 }
124 kvdb.db = sto
125 kvdb.CurrentIdx = 255
126 kvdb.CurrentBatch = 0
127
128 return nil
129 }
130
131 checkpointPath := path.Join(kvdb.path, fmt.Sprintf("%s%d", PathBatchNum, batchNum))
132 // copy 'BatchNumX' to 'current'
133 err = pebbleMakeCheckpoint(checkpointPath, currentPath)
134 if err != nil {
135 return tracerr.Wrap(err)
136 }
137
138 // open the new 'current'
139 sto, err := pebble.NewPebbleStorage(currentPath, false)
140 if err != nil {
141 return tracerr.Wrap(err)
142 }
143 kvdb.db = sto
144
145 // get currentBatch num
146 kvdb.CurrentBatch, err = kvdb.GetCurrentBatch()
147 if err != nil {
148 return tracerr.Wrap(err)
149 }
150 // idx is obtained from the statedb reset
151 kvdb.CurrentIdx, err = kvdb.GetCurrentIdx()
152 if err != nil {
153 return tracerr.Wrap(err)
154 }
155
156 return nil
157}
158
159// ResetFromSynchronizer performs a reset in the KVDB getting the state from
160// synchronizerKVDB for the given batchNum.
161func (kvdb *KVDB) ResetFromSynchronizer(batchNum common.BatchNum, synchronizerKVDB *KVDB) error {
162 if synchronizerKVDB == nil {
163 return tracerr.Wrap(fmt.Errorf("synchronizerKVDB can not be nil"))
164 }
165
166 if batchNum == 0 {
167 kvdb.CurrentIdx = 0
168 return nil
169 }
170 synchronizerCheckpointPath := path.Join(synchronizerKVDB.path,
171 fmt.Sprintf("%s%d", PathBatchNum, batchNum))
172 checkpointPath := path.Join(kvdb.path, fmt.Sprintf("%s%d", PathBatchNum, batchNum))
173 currentPath := path.Join(kvdb.path, PathCurrent)
174
175 // use checkpoint from synchronizerKVDB
176 if _, err := os.Stat(synchronizerCheckpointPath); os.IsNotExist(err) {
177 // if synchronizerKVDB does not have checkpoint at batchNum, return err
178 return tracerr.Wrap(fmt.Errorf("Checkpoint \"%v\" not exist in Synchronizer",
179 synchronizerCheckpointPath))
180 }
181
182 if err := kvdb.db.Pebble().Close(); err != nil {
183 return tracerr.Wrap(err)
184 }
185 // remove 'current'
186 err := os.RemoveAll(currentPath)
187 if err != nil {
188 return tracerr.Wrap(err)
189 }
190 // copy synchronizer'BatchNumX' to 'current'
191 err = pebbleMakeCheckpoint(synchronizerCheckpointPath, currentPath)
192 if err != nil {
193 return tracerr.Wrap(err)
194 }
195 // copy synchronizer'BatchNumX' to 'BatchNumX'
196 err = pebbleMakeCheckpoint(synchronizerCheckpointPath, checkpointPath)
197 if err != nil {
198 return tracerr.Wrap(err)
199 }
200
201 // open the new 'current'
202 sto, err := pebble.NewPebbleStorage(currentPath, false)
203 if err != nil {
204 return tracerr.Wrap(err)
205 }
206 kvdb.db = sto
207
208 // get currentBatch num
209 kvdb.CurrentBatch, err = kvdb.GetCurrentBatch()
210 if err != nil {
211 return tracerr.Wrap(err)
212 }
213 // get currentIdx
214 kvdb.CurrentIdx, err = kvdb.GetCurrentIdx()
215 if err != nil {
216 return tracerr.Wrap(err)
217 }
218
219 return nil
220}
221
222// GetCurrentBatch returns the current BatchNum stored in the KVDB
223func (kvdb *KVDB) GetCurrentBatch() (common.BatchNum, error) {
224 cbBytes, err := kvdb.db.Get(KeyCurrentBatch)
225 if tracerr.Unwrap(err) == db.ErrNotFound {
226 return 0, nil
227 }
228 if err != nil {
229 return 0, tracerr.Wrap(err)
230 }
231 return common.BatchNumFromBytes(cbBytes)
232}
233
234// setCurrentBatch stores the current BatchNum in the KVDB
235func (kvdb *KVDB) setCurrentBatch() error {
236 tx, err := kvdb.db.NewTx()
237 if err != nil {
238 return tracerr.Wrap(err)
239 }
240 err = tx.Put(KeyCurrentBatch, kvdb.CurrentBatch.Bytes())
241 if err != nil {
242 return tracerr.Wrap(err)
243 }
244 if err := tx.Commit(); err != nil {
245 return tracerr.Wrap(err)
246 }
247 return nil
248}
249
250// GetCurrentIdx returns the stored Idx from the KVDB, which is the last Idx
251// used for an Account in the KVDB.
252func (kvdb *KVDB) GetCurrentIdx() (common.Idx, error) {
253 idxBytes, err := kvdb.db.Get(keyCurrentIdx)
254 if tracerr.Unwrap(err) == db.ErrNotFound {
255 return 0, nil
256 }
257 if err != nil {
258 return 0, tracerr.Wrap(err)
259 }
260 return common.IdxFromBytes(idxBytes[:])
261}
262
263// SetCurrentIdx stores Idx in the KVDB
264func (kvdb *KVDB) SetCurrentIdx(idx common.Idx) error {
265 kvdb.CurrentIdx = idx
266
267 tx, err := kvdb.db.NewTx()
268 if err != nil {
269 return tracerr.Wrap(err)
270 }
271 idxBytes, err := idx.Bytes()
272 if err != nil {
273 return tracerr.Wrap(err)
274 }
275 err = tx.Put(keyCurrentIdx, idxBytes[:])
276 if err != nil {
277 return tracerr.Wrap(err)
278 }
279 if err := tx.Commit(); err != nil {
280 return tracerr.Wrap(err)
281 }
282 return nil
283}
284
285// MakeCheckpoint does a checkpoint at the given batchNum in the defined path.
286// Internally this advances & stores the current BatchNum, and then stores a
287// Checkpoint of the current state of the KVDB.
288func (kvdb *KVDB) MakeCheckpoint() error {
289 // advance currentBatch
290 kvdb.CurrentBatch++
291 log.Debugw("Making KVDB checkpoint", "batch", kvdb.CurrentBatch)
292
293 checkpointPath := path.Join(kvdb.path, fmt.Sprintf("%s%d", PathBatchNum, kvdb.CurrentBatch))
294
295 if err := kvdb.setCurrentBatch(); err != nil {
296 return tracerr.Wrap(err)
297 }
298
299 // if checkpoint BatchNum already exist in disk, delete it
300 if _, err := os.Stat(checkpointPath); !os.IsNotExist(err) {
301 err := os.RemoveAll(checkpointPath)
302 if err != nil {
303 return tracerr.Wrap(err)
304 }
305 } else if err != nil && !os.IsNotExist(err) {
306 return tracerr.Wrap(err)
307 }
308
309 // execute Checkpoint
310 if err := kvdb.db.Pebble().Checkpoint(checkpointPath); err != nil {
311 return tracerr.Wrap(err)
312 }
313 // delete old checkpoints
314 if err := kvdb.deleteOldCheckpoints(); err != nil {
315 return tracerr.Wrap(err)
316 }
317
318 return nil
319}
320
321// DeleteCheckpoint removes if exist the checkpoint of the given batchNum
322func (kvdb *KVDB) DeleteCheckpoint(batchNum common.BatchNum) error {
323 checkpointPath := path.Join(kvdb.path, fmt.Sprintf("%s%d", PathBatchNum, batchNum))
324
325 if _, err := os.Stat(checkpointPath); os.IsNotExist(err) {
326 return tracerr.Wrap(fmt.Errorf("Checkpoint with batchNum %d does not exist in DB", batchNum))
327 }
328
329 return os.RemoveAll(checkpointPath)
330}
331
332// ListCheckpoints returns the list of batchNums of the checkpoints, sorted.
333// If there's a gap between the list of checkpoints, an error is returned.
334func (kvdb *KVDB) ListCheckpoints() ([]int, error) {
335 files, err := ioutil.ReadDir(kvdb.path)
336 if err != nil {
337 return nil, tracerr.Wrap(err)
338 }
339 checkpoints := []int{}
340 var checkpoint int
341 pattern := fmt.Sprintf("%s%%d", PathBatchNum)
342 for _, file := range files {
343 fileName := file.Name()
344 if file.IsDir() && strings.HasPrefix(fileName, PathBatchNum) {
345 if _, err := fmt.Sscanf(fileName, pattern, &checkpoint); err != nil {
346 return nil, tracerr.Wrap(err)
347 }
348 checkpoints = append(checkpoints, checkpoint)
349 }
350 }
351 sort.Ints(checkpoints)
352 if len(checkpoints) > 0 {
353 first := checkpoints[0]
354 for _, checkpoint := range checkpoints[1:] {
355 first++
356 if checkpoint != first {
357 return nil, tracerr.Wrap(fmt.Errorf("checkpoint gap at %v", checkpoint))
358 }
359 }
360 }
361 return checkpoints, nil
362}
363
364// deleteOldCheckpoints deletes old checkpoints when there are more than
365// `s.keep` checkpoints
366func (kvdb *KVDB) deleteOldCheckpoints() error {
367 list, err := kvdb.ListCheckpoints()
368 if err != nil {
369 return tracerr.Wrap(err)
370 }
371 if len(list) > kvdb.keep {
372 for _, checkpoint := range list[:len(list)-kvdb.keep] {
373 if err := kvdb.DeleteCheckpoint(common.BatchNum(checkpoint)); err != nil {
374 return tracerr.Wrap(err)
375 }
376 }
377 }
378 return nil
379}
380
381func pebbleMakeCheckpoint(source, dest string) error {
382 // Remove dest folder (if it exists) before doing the checkpoint
383 if _, err := os.Stat(dest); !os.IsNotExist(err) {
384 err := os.RemoveAll(dest)
385 if err != nil {
386 return tracerr.Wrap(err)
387 }
388 } else if err != nil && !os.IsNotExist(err) {
389 return tracerr.Wrap(err)
390 }
391
392 sto, err := pebble.NewPebbleStorage(source, false)
393 if err != nil {
394 return tracerr.Wrap(err)
395 }
396 defer func() {
397 errClose := sto.Pebble().Close()
398 if errClose != nil {
399 log.Errorw("Pebble.Close", "err", errClose)
400 }
401 }()
402
403 // execute Checkpoint
404 err = sto.Pebble().Checkpoint(dest)
405 if err != nil {
406 return tracerr.Wrap(err)
407 }
408
409 return nil
410}