Make metabase.ContainerSize actually read-only

CHANGELOG.md

@@ -25,6 +25,7 @@ Changelog for NeoFS Node
 - Inability to provide session to NeoFS CLI in a NeoFS-binary format (#1933)
 - `neofs-adm` now works correctly with a committee of more than 4 nodes (#1949, #1959)
 - Closing a shard now waits until GC background workers stop (#1964)
+- Make it possible to use `shard.ContainerSize` in read-only mode (#1975)
 
 ### Removed
 ### Updated

pkg/local_object_storage/engine/container.go

@@ -1,6 +1,8 @@
 package engine
 
 import (
+	"errors"
+
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard"
 	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
 	"go.uber.org/zap"
@@ -75,7 +77,7 @@ func (e *StorageEngine) containerSize(prm ContainerSizePrm) (res ContainerSizeRe
 		csPrm.SetContainerID(prm.cnr)
 
 		csRes, err := sh.Shard.ContainerSize(csPrm)
-		if err != nil {
+		if err != nil && !errors.Is(err, shard.ErrDegradedMode) {
 			e.reportShardError(sh, "can't get container size", err,
 				zap.Stringer("container_id", prm.cnr),
 			)
@@ -124,7 +126,9 @@ func (e *StorageEngine) listContainers() (ListContainersRes, error) {
 	e.iterateOverUnsortedShards(func(sh hashedShard) (stop bool) {
 		res, err := sh.Shard.ListContainers(shard.ListContainersPrm{})
 		if err != nil {
-			e.reportShardError(sh, "can't get list of containers", err)
+			if !errors.Is(err, shard.ErrDegradedMode) {
+				e.reportShardError(sh, "can't get list of containers", err)
+			}
 			return false
 		}
 

pkg/local_object_storage/engine/evacuate.go

@@ -125,7 +125,7 @@ mainLoop:
 			//  because ListWithCursor works only with the metabase.
 			listRes, err := sh.ListWithCursor(listPrm)
 			if err != nil {
-				if errors.Is(err, meta.ErrEndOfListing) {
+				if errors.Is(err, meta.ErrEndOfListing) || errors.Is(err, shard.ErrDegradedMode) {
 					continue mainLoop
 				}
 				return res, err

pkg/local_object_storage/engine/select.go

@@ -1,6 +1,8 @@
 package engine
 
 import (
+	"errors"
+
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard"
 	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
 	"github.com/nspcc-dev/neofs-sdk-go/object"
@@ -109,7 +111,9 @@ func (e *StorageEngine) list(limit uint64) (SelectRes, error) {
 	e.iterateOverUnsortedShards(func(sh hashedShard) (stop bool) {
 		res, err := sh.List() // consider limit result of shard iterator
 		if err != nil {
-			e.reportShardError(sh, "could not select objects from shard", err)
+			if !errors.Is(err, shard.ErrDegradedMode) {
+				e.reportShardError(sh, "could not select objects from shard", err)
+			}
 		} else {
 			for _, addr := range res.AddressList() { // save only unique values
 				if _, ok := uniqueMap[addr.EncodeToString()]; !ok {

pkg/local_object_storage/metabase/containers.go

@@ -11,6 +11,10 @@ func (db *DB) Containers() (list []cid.ID, err error) {
 	db.modeMtx.RLock()
 	defer db.modeMtx.RUnlock()
 
+	if db.mode.NoMetabase() {
+		return nil, ErrDegradedMode
+	}
+
 	err = db.boltDB.View(func(tx *bbolt.Tx) error {
 		list, err = db.containers(tx)
 
@@ -38,7 +42,7 @@ func (db *DB) containers(tx *bbolt.Tx) ([]cid.ID, error) {
 }
 
 func (db *DB) ContainerSize(id cid.ID) (size uint64, err error) {
-	err = db.boltDB.Update(func(tx *bbolt.Tx) error {
+	err = db.boltDB.View(func(tx *bbolt.Tx) error {
 		size, err = db.containerSize(tx, id)
 
 		return err
@@ -48,11 +52,7 @@ func (db *DB) ContainerSize(id cid.ID) (size uint64, err error) {
 }
 
 func (db *DB) containerSize(tx *bbolt.Tx, id cid.ID) (uint64, error) {
-	containerVolume, err := tx.CreateBucketIfNotExists(containerVolumeBucketName)
-	if err != nil {
-		return 0, err
-	}
-
+	containerVolume := tx.Bucket(containerVolumeBucketName)
 	key := make([]byte, cidSize)
 	id.Encode(key)
 
@@ -78,11 +78,7 @@ func parseContainerSize(v []byte) uint64 {
 }
 
 func changeContainerSize(tx *bbolt.Tx, id cid.ID, delta uint64, increase bool) error {
-	containerVolume, err := tx.CreateBucketIfNotExists(containerVolumeBucketName)
-	if err != nil {
-		return err
-	}
-
+	containerVolume := tx.Bucket(containerVolumeBucketName)
 	key := make([]byte, cidSize)
 	id.Encode(key)
 

pkg/local_object_storage/metabase/control.go

@@ -94,6 +94,7 @@ func (db *DB) init(reset bool) error {
 		string(toMoveItBucketName):        {},
 		string(garbageBucketName):         {},
 		string(shardInfoBucket):           {},
+		string(bucketNameLocked):          {},
 	}
 
 	return db.boltDB.Update(func(tx *bbolt.Tx) error {
@@ -106,16 +107,18 @@ func (db *DB) init(reset bool) error {
 			}
 		}
 		for k := range mStaticBuckets {
-			b, err := tx.CreateBucketIfNotExists([]byte(k))
-			if err != nil {
-				return fmt.Errorf("could not create static bucket %s: %w", k, err)
-			}
-
+			name := []byte(k)
 			if reset {
-				if err = resetBucket(b); err != nil {
-					return fmt.Errorf("could not reset static bucket %s: %w", k, err)
+				err := tx.DeleteBucket(name)
+				if err != nil && !errors.Is(err, bbolt.ErrBucketNotFound) {
+					return fmt.Errorf("could not delete static bucket %s: %w", k, err)
 				}
 			}
+
+			_, err := tx.CreateBucketIfNotExists(name)
+			if err != nil {
+				return fmt.Errorf("could not create static bucket %s: %w", k, err)
+			}
 		}
 
 		if !reset {

pkg/local_object_storage/metabase/graveyard.go

@@ -164,10 +164,6 @@ func (db *DB) iterateDeletedObj(tx *bbolt.Tx, h kvHandler, offset *oid.Address)
 		panic(fmt.Sprintf("metabase: unknown iteration object hadler: %T", t))
 	}
 
-	if bkt == nil {
-		return nil
-	}
-
 	c := bkt.Cursor()
 	var k, v []byte
 

pkg/local_object_storage/metabase/inhume.go

@@ -92,6 +92,10 @@ func (db *DB) Inhume(prm InhumePrm) (res InhumeRes, err error) {
 	db.modeMtx.RLock()
 	defer db.modeMtx.RUnlock()
 
+	if db.mode.NoMetabase() {
+		return InhumeRes{}, ErrDegradedMode
+	}
+
 	currEpoch := db.epochState.CurrentEpoch()
 	var inhumed uint64
 

pkg/local_object_storage/metabase/iterators.go

@@ -125,9 +125,6 @@ func (db *DB) IterateCoveredByTombstones(tss map[string]oid.Address, h func(oid.
 
 func (db *DB) iterateCoveredByTombstones(tx *bbolt.Tx, tss map[string]oid.Address, h func(oid.Address) error) error {
 	bktGraveyard := tx.Bucket(graveyardBucketName)
-	if bktGraveyard == nil {
-		return nil
-	}
 
 	err := bktGraveyard.ForEach(func(k, v []byte) error {
 		var addr oid.Address

pkg/local_object_storage/metabase/lock.go

@@ -44,10 +44,7 @@ func (db *DB) Lock(cnr cid.ID, locker oid.ID, locked []oid.ID) error {
 			return apistatus.LockNonRegularObject{}
 		}
 
-		bucketLocked, err := tx.CreateBucketIfNotExists(bucketNameLocked)
-		if err != nil {
-			return fmt.Errorf("create global bucket for locked objects: %w", err)
-		}
+		bucketLocked := tx.Bucket(bucketNameLocked)
 
 		cnr.Encode(key)
 		bucketLockedContainer, err := bucketLocked.CreateBucketIfNotExists(key)

pkg/local_object_storage/metabase/movable.go

@@ -56,11 +56,7 @@ func (db *DB) ToMoveIt(prm ToMoveItPrm) (res ToMoveItRes, err error) {
 	key = addressKey(prm.addr, key)
 
 	err = db.boltDB.Update(func(tx *bbolt.Tx) error {
-		toMoveIt, err := tx.CreateBucketIfNotExists(toMoveItBucketName)
-		if err != nil {
-			return err
-		}
-
+		toMoveIt := tx.Bucket(toMoveItBucketName)
 		return toMoveIt.Put(key, zeroValue)
 	})
 
@@ -77,10 +73,6 @@ func (db *DB) DoNotMove(prm DoNotMovePrm) (res DoNotMoveRes, err error) {
 
 	err = db.boltDB.Update(func(tx *bbolt.Tx) error {
 		toMoveIt := tx.Bucket(toMoveItBucketName)
-		if toMoveIt == nil {
-			return nil
-		}
-
 		return toMoveIt.Delete(key)
 	})
 
@@ -96,10 +88,6 @@ func (db *DB) Movable(_ MovablePrm) (MovableRes, error) {
 
 	err := db.boltDB.View(func(tx *bbolt.Tx) error {
 		toMoveIt := tx.Bucket(toMoveItBucketName)
-		if toMoveIt == nil {
-			return nil
-		}
-
 		return toMoveIt.ForEach(func(k, v []byte) error {
 			strAddrs = append(strAddrs, string(k))
 

pkg/local_object_storage/metabase/util.go

@@ -223,17 +223,6 @@ func objectKey(obj oid.ID, key []byte) []byte {
 	return key[:objectKeySize]
 }
 
-// removes all bucket elements.
-func resetBucket(b *bbolt.Bucket) error {
-	return b.ForEach(func(k, v []byte) error {
-		if v != nil {
-			return b.Delete(k)
-		}
-
-		return b.DeleteBucket(k)
-	})
-}
-
 // if meets irregular object container in objs - returns its type, otherwise returns object.TypeRegular.
 //
 // firstIrregularObjectType(tx, cnr, obj) usage allows getting object type.

pkg/local_object_storage/shard/container.go

@@ -23,6 +23,13 @@ func (r ContainerSizeRes) Size() uint64 {
 }
 
 func (s *Shard) ContainerSize(prm ContainerSizePrm) (ContainerSizeRes, error) {
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.NoMetabase() {
+		return ContainerSizeRes{}, ErrDegradedMode
+	}
+
 	size, err := s.metaBase.ContainerSize(prm.cnr)
 	if err != nil {
 		return ContainerSizeRes{}, fmt.Errorf("could not get container size: %w", err)

pkg/local_object_storage/shard/gc.go

@@ -284,6 +284,11 @@ func (s *Shard) collectExpiredTombstones(ctx context.Context, e Event) {
 	for {
 		log.Debug("iterating tombstones")
 
+		if s.GetMode().NoMetabase() {
+			s.log.Debug("shard is in a degraded mode, skip collecting expired tombstones")
+			return
+		}
+
 		err := s.metaBase.IterateOverGraveyard(iterPrm)
 		if err != nil {
 			log.Error("iterator over graveyard failed", zap.Error(err))
@@ -327,6 +332,10 @@ func (s *Shard) collectExpiredLocks(ctx context.Context, e Event) {
 }
 
 func (s *Shard) getExpiredObjects(ctx context.Context, epoch uint64, typeCond func(object.Type) bool) ([]oid.Address, error) {
+	if s.GetMode().NoMetabase() {
+		return nil, ErrDegradedMode
+	}
+
 	var expired []oid.Address
 
 	err := s.metaBase.IterateExpired(epoch, func(expiredObject *meta.ExpiredObject) error {
@@ -351,6 +360,10 @@ func (s *Shard) getExpiredObjects(ctx context.Context, epoch uint64, typeCond fu
 //
 // Does not modify tss.
 func (s *Shard) HandleExpiredTombstones(tss []meta.TombstonedObject) {
+	if s.GetMode().NoMetabase() {
+		return
+	}
+
 	// Mark tombstones as garbage.
 	var pInhume meta.InhumePrm
 
@@ -385,6 +398,10 @@ func (s *Shard) HandleExpiredTombstones(tss []meta.TombstonedObject) {
 // HandleExpiredLocks unlocks all objects which were locked by lockers.
 // If successful, marks lockers themselves as garbage.
 func (s *Shard) HandleExpiredLocks(lockers []oid.Address) {
+	if s.GetMode().NoMetabase() {
+		return
+	}
+
 	err := s.metaBase.FreeLockedBy(lockers)
 	if err != nil {
 		s.log.Warn("failure to unlock objects",
@@ -412,6 +429,10 @@ func (s *Shard) HandleExpiredLocks(lockers []oid.Address) {
 
 // HandleDeletedLocks unlocks all objects which were locked by lockers.
 func (s *Shard) HandleDeletedLocks(lockers []oid.Address) {
+	if s.GetMode().NoMetabase() {
+		return
+	}
+
 	err := s.metaBase.FreeLockedBy(lockers)
 	if err != nil {
 		s.log.Warn("failure to unlock objects",

pkg/local_object_storage/shard/list.go

@@ -64,6 +64,10 @@ func (r ListWithCursorRes) Cursor() *Cursor {
 
 // List returns all objects physically stored in the Shard.
 func (s *Shard) List() (res SelectRes, err error) {
+	if s.GetMode().NoMetabase() {
+		return SelectRes{}, ErrDegradedMode
+	}
+
 	lst, err := s.metaBase.Containers()
 	if err != nil {
 		return res, fmt.Errorf("can't list stored containers: %w", err)
@@ -93,6 +97,10 @@ func (s *Shard) List() (res SelectRes, err error) {
 }
 
 func (s *Shard) ListContainers(_ ListContainersPrm) (ListContainersRes, error) {
+	if s.GetMode().NoMetabase() {
+		return ListContainersRes{}, ErrDegradedMode
+	}
+
 	containers, err := s.metaBase.Containers()
 	if err != nil {
 		return ListContainersRes{}, fmt.Errorf("could not get list of containers: %w", err)