aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorAnton Evangelatov <anton.evangelatov@gmail.com>2018-10-03 20:31:59 +0800
committerGitHub <noreply@github.com>2018-10-03 20:31:59 +0800
commit303b99663e963a520aaa44eca68e042d9fe230af (patch)
tree8913a0c5cc2e5578b43695e00f07f0bb6789d1a0
parent14bef9a2dba1f6370c694779962e742e9853fdc6 (diff)
downloadgo-tangerine-303b99663e963a520aaa44eca68e042d9fe230af.tar.gz
go-tangerine-303b99663e963a520aaa44eca68e042d9fe230af.tar.zst
go-tangerine-303b99663e963a520aaa44eca68e042d9fe230af.zip
swarm: schemas and migrations (#17813)
-rw-r--r--cmd/swarm/db.go15
-rw-r--r--cmd/swarm/main.go8
-rw-r--r--swarm/storage/database.go9
-rw-r--r--swarm/storage/ldbstore.go33
-rw-r--r--swarm/storage/localstore.go39
-rw-r--r--swarm/storage/schema.go6
-rw-r--r--swarm/swarm.go5
7 files changed, 81 insertions, 34 deletions
diff --git a/cmd/swarm/db.go b/cmd/swarm/db.go
index fe03f2d16..107fbf100 100644
--- a/cmd/swarm/db.go
+++ b/cmd/swarm/db.go
@@ -93,21 +93,6 @@ func dbImport(ctx *cli.Context) {
log.Info(fmt.Sprintf("successfully imported %d chunks", count))
}
-func dbClean(ctx *cli.Context) {
- args := ctx.Args()
- if len(args) != 2 {
- utils.Fatalf("invalid arguments, please specify <chunkdb> (path to a local chunk database) and the base key")
- }
-
- store, err := openLDBStore(args[0], common.Hex2Bytes(args[1]))
- if err != nil {
- utils.Fatalf("error opening local chunk database: %s", err)
- }
- defer store.Close()
-
- store.Cleanup()
-}
-
func openLDBStore(path string, basekey []byte) (*storage.LDBStore, error) {
if _, err := os.Stat(filepath.Join(path, "CURRENT")); err != nil {
return nil, fmt.Errorf("invalid chunkdb path: %s", err)
diff --git a/cmd/swarm/main.go b/cmd/swarm/main.go
index 5acf87c71..14d8b4c6b 100644
--- a/cmd/swarm/main.go
+++ b/cmd/swarm/main.go
@@ -537,14 +537,6 @@ pv(1) tool to get a progress bar:
pv chunks.tar | swarm db import ~/.ethereum/swarm/bzz-KEY/chunks -`,
},
- {
- Action: dbClean,
- CustomHelpTemplate: helpTemplate,
- Name: "clean",
- Usage: "remove corrupt entries from a local chunk database",
- ArgsUsage: "<chunkdb>",
- Description: "Remove corrupt entries from a local chunk database",
- },
},
},
diff --git a/swarm/storage/database.go b/swarm/storage/database.go
index 3b5d003de..e25fce31f 100644
--- a/swarm/storage/database.go
+++ b/swarm/storage/database.go
@@ -20,8 +20,6 @@ package storage
// no need for queueing/caching
import (
- "fmt"
-
"github.com/ethereum/go-ethereum/metrics"
"github.com/syndtr/goleveldb/leveldb"
"github.com/syndtr/goleveldb/leveldb/iterator"
@@ -46,13 +44,10 @@ func NewLDBDatabase(file string) (*LDBDatabase, error) {
return database, nil
}
-func (db *LDBDatabase) Put(key []byte, value []byte) {
+func (db *LDBDatabase) Put(key []byte, value []byte) error {
metrics.GetOrRegisterCounter("ldbdatabase.put", nil).Inc(1)
- err := db.db.Put(key, value, nil)
- if err != nil {
- fmt.Println("Error put", err)
- }
+ return db.db.Put(key, value, nil)
}
func (db *LDBDatabase) Get(key []byte) ([]byte, error) {
diff --git a/swarm/storage/ldbstore.go b/swarm/storage/ldbstore.go
index bde627394..2a7f51cb3 100644
--- a/swarm/storage/ldbstore.go
+++ b/swarm/storage/ldbstore.go
@@ -37,7 +37,6 @@ import (
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/rlp"
- ch "github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/log"
"github.com/ethereum/go-ethereum/swarm/storage/mock"
"github.com/syndtr/goleveldb/leveldb"
@@ -61,6 +60,7 @@ var (
keyDataIdx = []byte{4}
keyData = byte(6)
keyDistanceCnt = byte(7)
+ keySchema = []byte{8}
)
var (
@@ -418,8 +418,8 @@ func (s *LDBStore) Import(in io.Reader) (int64, error) {
}
}
-func (s *LDBStore) Cleanup() {
- //Iterates over the database and checks that there are no chunks bigger than 4kb
+//Cleanup iterates over the database and deletes chunks if they pass the `f` condition
+func (s *LDBStore) Cleanup(f func(*chunk) bool) {
var errorsFound, removed, total int
it := s.db.NewIterator()
@@ -471,7 +471,8 @@ func (s *LDBStore) Cleanup() {
cs := int64(binary.LittleEndian.Uint64(c.sdata[:8]))
log.Trace("chunk", "key", fmt.Sprintf("%x", key), "ck", fmt.Sprintf("%x", ck), "dkey", fmt.Sprintf("%x", datakey), "dataidx", index.Idx, "po", po, "len data", len(data), "len sdata", len(c.sdata), "size", cs)
- if len(c.sdata) > ch.DefaultSize+8 {
+ // if chunk is to be removed
+ if f(c) {
log.Warn("chunk for cleanup", "key", fmt.Sprintf("%x", key), "ck", fmt.Sprintf("%x", ck), "dkey", fmt.Sprintf("%x", datakey), "dataidx", index.Idx, "po", po, "len data", len(data), "len sdata", len(c.sdata), "size", cs)
s.delete(index.Idx, getIndexKey(key[1:]), po)
removed++
@@ -730,6 +731,30 @@ func (s *LDBStore) tryAccessIdx(ikey []byte, index *dpaDBIndex) bool {
return true
}
+// GetSchema is returning the current named schema of the datastore as read from LevelDB
+func (s *LDBStore) GetSchema() (string, error) {
+ s.lock.Lock()
+ defer s.lock.Unlock()
+
+ data, err := s.db.Get(keySchema)
+ if err != nil {
+ if err == leveldb.ErrNotFound {
+ return "", nil
+ }
+ return "", err
+ }
+
+ return string(data), nil
+}
+
+// PutSchema is saving a named schema to the LevelDB datastore
+func (s *LDBStore) PutSchema(schema string) error {
+ s.lock.Lock()
+ defer s.lock.Unlock()
+
+ return s.db.Put(keySchema, []byte(schema))
+}
+
func (s *LDBStore) Get(_ context.Context, addr Address) (chunk Chunk, err error) {
metrics.GetOrRegisterCounter("ldbstore.get", nil).Inc(1)
log.Trace("ldbstore.get", "key", addr)
diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go
index 04701ee69..b28f62524 100644
--- a/swarm/storage/localstore.go
+++ b/swarm/storage/localstore.go
@@ -184,3 +184,42 @@ func (ls *LocalStore) Iterator(from uint64, to uint64, po uint8, f func(Address,
func (ls *LocalStore) Close() {
ls.DbStore.Close()
}
+
+// Migrate checks the datastore schema vs the runtime schema, and runs migrations if they don't match
+func (ls *LocalStore) Migrate() error {
+ schema, err := ls.DbStore.GetSchema()
+ if err != nil {
+ log.Error(err.Error())
+ return err
+ }
+
+ log.Debug("found schema", "schema", schema, "runtime-schema", CurrentDbSchema)
+ if schema != CurrentDbSchema {
+ // run migrations
+
+ if schema == "" {
+ log.Debug("running migrations for", "schema", schema, "runtime-schema", CurrentDbSchema)
+
+ cleanupFunc := func(c *chunk) bool {
+ // if one of the ls.Validators passes, it means a chunk is of particular type and it is valid
+ valid := false
+ for _, v := range ls.Validators {
+ if valid = v.Validate(c.Address(), c.Data()); valid {
+ break
+ }
+ }
+ return valid
+ }
+
+ ls.DbStore.Cleanup(cleanupFunc)
+
+ err := ls.DbStore.PutSchema(DbSchemaPurity)
+ if err != nil {
+ log.Error(err.Error())
+ return err
+ }
+ }
+ }
+
+ return nil
+}
diff --git a/swarm/storage/schema.go b/swarm/storage/schema.go
new file mode 100644
index 000000000..fb8498a29
--- /dev/null
+++ b/swarm/storage/schema.go
@@ -0,0 +1,6 @@
+package storage
+
+// "purity" is the first formal schema of LevelDB we release together with Swarm 0.3.5
+const DbSchemaPurity = "purity"
+
+const CurrentDbSchema = DbSchemaPurity
diff --git a/swarm/swarm.go b/swarm/swarm.go
index 0cd56d4eb..5b6938754 100644
--- a/swarm/swarm.go
+++ b/swarm/swarm.go
@@ -197,6 +197,11 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e
resourceHandler,
}
+ err = lstore.Migrate()
+ if err != nil {
+ return nil, err
+ }
+
log.Debug("Setup local storage")
self.bzz = network.NewBzz(bzzconfig, to, stateStore, stream.Spec, self.streamer.Run)