feat(localstore): add migration for batch-index (ethersphere#2112)

Author: Peter Mrekaj

cmd/bee/cmd/db.go

@@ -58,7 +58,7 @@ func dbExportCmd(cmd *cobra.Command) {
 
 			path := filepath.Join(dataDir, "localstore")
 
-			storer, err := localstore.New(path, nil, nil, logger)
+			storer, err := localstore.New(path, nil, nil, nil, logger)
 			if err != nil {
 				return fmt.Errorf("localstore: %w", err)
 			}
@@ -118,7 +118,7 @@ func dbImportCmd(cmd *cobra.Command) {
 
 			path := filepath.Join(dataDir, "localstore")
 
-			storer, err := localstore.New(path, nil, nil, logger)
+			storer, err := localstore.New(path, nil, nil, nil, logger)
 			if err != nil {
 				return fmt.Errorf("localstore: %w", err)
 			}

pkg/localstore/gc_test.go

@@ -486,7 +486,7 @@ func TestDB_gcSize(t *testing.T) {
 		t.Fatal(err)
 	}
 	logger := logging.New(ioutil.Discard, 0)
-	db, err := New(dir, baseKey, nil, logger)
+	db, err := New(dir, baseKey, nil, nil, logger)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -514,7 +514,7 @@ func TestDB_gcSize(t *testing.T) {
 	if err := db.Close(); err != nil {
 		t.Fatal(err)
 	}
-	db, err = New(dir, baseKey, nil, logger)
+	db, err = New(dir, baseKey, nil, nil, logger)
 	if err != nil {
 		t.Fatal(err)
 	}

pkg/localstore/localstore.go

@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"github.com/ethersphere/bee/pkg/logging"
+	"github.com/ethersphere/bee/pkg/pinning"
 	"github.com/ethersphere/bee/pkg/postage"
 	"github.com/ethersphere/bee/pkg/postage/batchstore"
 	"github.com/ethersphere/bee/pkg/shed"
@@ -62,6 +63,9 @@ type DB struct {
 	shed *shed.DB
 	tags *tags.Tags
 
+	// stateStore is needed to access the pinning Service.Pins() method.
+	stateStore storage.StateStorer
+
 	// schema name of loaded data
 	schemaName shed.StringField
 
@@ -176,7 +180,7 @@ type Options struct {
 // New returns a new DB.  All fields and indexes are initialized
 // and possible conflicts with schema from existing database is checked.
 // One goroutine for writing batches is created.
-func New(path string, baseKey []byte, o *Options, logger logging.Logger) (db *DB, err error) {
+func New(path string, baseKey []byte, ss storage.StateStorer, o *Options, logger logging.Logger) (db *DB, err error) {
 	if o == nil {
 		// default options
 		o = &Options{
@@ -185,6 +189,7 @@ func New(path string, baseKey []byte, o *Options, logger logging.Logger) (db *DB
 	}
 
 	db = &DB{
+		stateStore:    ss,
 		cacheCapacity: o.Capacity,
 		baseKey:       baseKey,
 		tags:          o.Tags,
@@ -241,7 +246,7 @@ func New(path string, baseKey []byte, o *Options, logger logging.Logger) (db *DB
 	}
 	if schemaName == "" {
 		// initial new localstore run
-		err := db.schemaName.Put(DbSchemaCurrent)
+		err := db.schemaName.Put(DBSchemaCurrent)
 		if err != nil {
 			return nil, err
 		}
@@ -585,6 +590,16 @@ func (db *DB) DebugIndices() (indexInfo map[string]int, err error) {
 	return indexInfo, err
 }
 
+// stateStoreHasPins returns true if the state-store
+// contains any pins, otherwise false is returned.
+func (db *DB) stateStoreHasPins() (bool, error) {
+	pins, err := pinning.NewService(nil, db.stateStore, nil).Pins()
+	if err != nil {
+		return false, err
+	}
+	return len(pins) > 0, nil
+}
+
 // chunkToItem creates new Item with data provided by the Chunk.
 func chunkToItem(ch swarm.Chunk) shed.Item {
 	return shed.Item{

pkg/localstore/localstore_test.go

@@ -158,7 +158,7 @@ func newTestDB(t testing.TB, o *Options) *DB {
 		t.Fatal(err)
 	}
 	logger := logging.New(ioutil.Discard, 0)
-	db, err := New("", baseKey, o, logger)
+	db, err := New("", baseKey, nil, o, logger)
 	if err != nil {
 		t.Fatal(err)
 	}