/
kv.go
116 lines (104 loc) · 2.86 KB
/
kv.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
// Package slasherkv defines a bolt-db, key-value store implementation
// of the slasher database interface for Prysm.
package slasherkv
import (
"context"
"os"
"path"
"time"
"github.com/pkg/errors"
"github.com/prysmaticlabs/prysm/v4/beacon-chain/db/iface"
"github.com/prysmaticlabs/prysm/v4/config/params"
"github.com/prysmaticlabs/prysm/v4/io/file"
bolt "go.etcd.io/bbolt"
)
var _ iface.SlasherDatabase = (*Store)(nil)
const (
// DatabaseFileName is the name of the beacon node database.
DatabaseFileName = "slasher.db"
boltAllocSize = 8 * 1024 * 1024
// Specifies the initial mmap size of bolt.
mmapSize = 536870912
)
// Store defines an implementation of the Prysm Database interface
// using BoltDB as the underlying persistent kv-store for Ethereum consensus.
type Store struct {
db *bolt.DB
databasePath string
ctx context.Context
}
// NewKVStore initializes a new boltDB key-value store at the directory
// path specified, creates the kv-buckets based on the schema, and stores
// an open connection db object as a property of the Store struct.
func NewKVStore(ctx context.Context, dirPath string) (*Store, error) {
hasDir, err := file.HasDir(dirPath)
if err != nil {
return nil, err
}
if !hasDir {
if err := file.MkdirAll(dirPath); err != nil {
return nil, err
}
}
datafile := path.Join(dirPath, DatabaseFileName)
boltDB, err := bolt.Open(
datafile,
params.BeaconIoConfig().ReadWritePermissions,
&bolt.Options{
Timeout: 1 * time.Second,
InitialMmapSize: mmapSize,
},
)
if err != nil {
if errors.Is(err, bolt.ErrTimeout) {
return nil, errors.New("cannot obtain database lock, database may be in use by another process")
}
return nil, err
}
boltDB.AllocSize = boltAllocSize
kv := &Store{
db: boltDB,
databasePath: dirPath,
ctx: ctx,
}
if err := kv.db.Update(func(tx *bolt.Tx) error {
return createBuckets(
tx,
// Slasher buckets.
attestedEpochsByValidator,
attestationRecordsBucket,
attestationDataRootsBucket,
proposalRecordsBucket,
slasherChunksBucket,
)
}); err != nil {
return nil, err
}
return kv, err
}
// ClearDB removes the previously stored database in the data directory.
func (s *Store) ClearDB() error {
if _, err := os.Stat(s.databasePath); os.IsNotExist(err) {
return nil
}
if err := os.Remove(path.Join(s.databasePath, DatabaseFileName)); err != nil {
return errors.Wrap(err, "could not remove database file")
}
return nil
}
// Close closes the underlying BoltDB database.
func (s *Store) Close() error {
return s.db.Close()
}
// DatabasePath at which this database writes files.
func (s *Store) DatabasePath() string {
return s.databasePath
}
func createBuckets(tx *bolt.Tx, buckets ...[]byte) error {
for _, bucket := range buckets {
if _, err := tx.CreateBucketIfNotExists(bucket); err != nil {
return err
}
}
return nil
}