From eea2892109afe12acda9d46eb83ee9fad8049ca4 Mon Sep 17 00:00:00 2001
From: Pavel Karpy <p.karpy@yadro.com>
Date: Wed, 7 Dec 2022 20:42:35 +0300
Subject: [PATCH] [#1956] node: Lock shard's mode on its methods switch

Signed-off-by: Pavel Karpy <p.karpy@yadro.com>
---
 CHANGELOG.md                             |  1 +
 pkg/local_object_storage/shard/delete.go | 12 +++++++---
 pkg/local_object_storage/shard/exists.go |  5 +++-
 pkg/local_object_storage/shard/gc.go     | 29 ++++++++++++++++++++----
 pkg/local_object_storage/shard/get.go    |  7 ++++--
 pkg/local_object_storage/shard/inhume.go | 14 +++++++++---
 pkg/local_object_storage/shard/list.go   |  5 +++-
 pkg/local_object_storage/shard/lock.go   |  5 +++-
 pkg/local_object_storage/shard/move.go   |  5 +++-
 pkg/local_object_storage/shard/put.go    |  5 +++-
 pkg/local_object_storage/shard/range.go  |  5 +++-
 pkg/local_object_storage/shard/select.go |  5 +++-
 pkg/local_object_storage/shard/tree.go   | 18 ++++++++++++---
 13 files changed, 94 insertions(+), 22 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index e61567a31..932e3116c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -67,6 +67,7 @@ Changelog for NeoFS Node
 - Synchronizing a tree now longer reports an error for a single-node container (#2154)
 - Prevent leaking goroutines in the tree service (#2162)
 - Do not search for LOCK objects when delete container when session provided (#2152)
+- Race conditions on shard's mode switch (#1956)
 
 ### Removed
 - `-g` option from `neofs-cli control ...` and `neofs-cli container create` commands (#2089)
diff --git a/pkg/local_object_storage/shard/delete.go b/pkg/local_object_storage/shard/delete.go
index 22388a03a..77d1c9772 100644
--- a/pkg/local_object_storage/shard/delete.go
+++ b/pkg/local_object_storage/shard/delete.go
@@ -28,10 +28,16 @@ func (p *DeletePrm) SetAddresses(addr ...oid.Address) {
 // Delete removes data from the shard's writeCache, metaBase and
 // blobStor.
 func (s *Shard) Delete(prm DeletePrm) (DeleteRes, error) {
-	m := s.GetMode()
-	if m.ReadOnly() {
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	return s.delete(prm)
+}
+
+func (s *Shard) delete(prm DeletePrm) (DeleteRes, error) {
+	if s.info.Mode.ReadOnly() {
 		return DeleteRes{}, ErrReadOnlyMode
-	} else if m.NoMetabase() {
+	} else if s.info.Mode.NoMetabase() {
 		return DeleteRes{}, ErrDegradedMode
 	}
 
diff --git a/pkg/local_object_storage/shard/exists.go b/pkg/local_object_storage/shard/exists.go
index 2e711253e..5bb930b62 100644
--- a/pkg/local_object_storage/shard/exists.go
+++ b/pkg/local_object_storage/shard/exists.go
@@ -37,7 +37,10 @@ func (s *Shard) Exists(prm ExistsPrm) (ExistsRes, error) {
 	var exists bool
 	var err error
 
-	if s.GetMode().NoMetabase() {
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.NoMetabase() {
 		var p common.ExistsPrm
 		p.Address = prm.addr
 
diff --git a/pkg/local_object_storage/shard/gc.go b/pkg/local_object_storage/shard/gc.go
index dd76b7bb6..e527d779c 100644
--- a/pkg/local_object_storage/shard/gc.go
+++ b/pkg/local_object_storage/shard/gc.go
@@ -187,7 +187,10 @@ func (gc *gc) stop() {
 // with GC-marked graves.
 // Does nothing if shard is in "read-only" mode.
 func (s *Shard) removeGarbage() {
-	if s.GetMode() != mode.ReadWrite {
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode != mode.ReadWrite {
 		return
 	}
 
@@ -221,7 +224,7 @@ func (s *Shard) removeGarbage() {
 	deletePrm.SetAddresses(buf...)
 
 	// delete accumulated objects
-	_, err = s.Delete(deletePrm)
+	_, err = s.delete(deletePrm)
 	if err != nil {
 		s.log.Warn("could not delete the objects",
 			zap.String("error", err.Error()),
@@ -242,6 +245,13 @@ func (s *Shard) collectExpiredObjects(ctx context.Context, e Event) {
 		return
 	}
 
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.NoMetabase() {
+		return
+	}
+
 	var inhumePrm meta.InhumePrm
 
 	inhumePrm.SetAddresses(expired...)
@@ -284,17 +294,25 @@ func (s *Shard) collectExpiredTombstones(ctx context.Context, e Event) {
 	for {
 		log.Debug("iterating tombstones")
 
-		if s.GetMode().NoMetabase() {
+		s.m.RLock()
+
+		if s.info.Mode.NoMetabase() {
 			s.log.Debug("shard is in a degraded mode, skip collecting expired tombstones")
+			s.m.RUnlock()
+
 			return
 		}
 
 		err := s.metaBase.IterateOverGraveyard(iterPrm)
 		if err != nil {
 			log.Error("iterator over graveyard failed", zap.Error(err))
+			s.m.RUnlock()
+
 			return
 		}
 
+		s.m.RUnlock()
+
 		tssLen := len(tss)
 		if tssLen == 0 {
 			break
@@ -332,7 +350,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() {
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.NoMetabase() {
 		return nil, ErrDegradedMode
 	}
 
diff --git a/pkg/local_object_storage/shard/get.go b/pkg/local_object_storage/shard/get.go
index 9c65db4a7..0ff846e99 100644
--- a/pkg/local_object_storage/shard/get.go
+++ b/pkg/local_object_storage/shard/get.go
@@ -62,6 +62,9 @@ func (r GetRes) HasMeta() bool {
 // Returns an error of type apistatus.ObjectAlreadyRemoved if the requested object has been marked as removed in shard.
 // Returns the object.ErrObjectIsExpired if the object is presented but already expired.
 func (s *Shard) Get(prm GetPrm) (GetRes, error) {
+	s.m.RLock()
+	defer s.m.RUnlock()
+
 	cb := func(stor *blobstor.BlobStor, id []byte) (*objectSDK.Object, error) {
 		var getPrm common.GetPrm
 		getPrm.Address = prm.addr
@@ -79,7 +82,7 @@ func (s *Shard) Get(prm GetPrm) (GetRes, error) {
 		return c.Get(prm.addr)
 	}
 
-	skipMeta := prm.skipMeta || s.GetMode().NoMetabase()
+	skipMeta := prm.skipMeta || s.info.Mode.NoMetabase()
 	obj, hasMeta, err := s.fetchObjectData(prm.addr, skipMeta, cb, wc)
 
 	return GetRes{
@@ -114,7 +117,7 @@ func (s *Shard) fetchObjectData(addr oid.Address, skipMeta bool, cb storFetcher,
 		mPrm.SetAddress(addr)
 
 		mRes, err := s.metaBase.Exists(mPrm)
-		if err != nil && !s.GetMode().NoMetabase() {
+		if err != nil && !s.info.Mode.NoMetabase() {
 			return res, false, err
 		}
 		exists = mRes.Exists()
diff --git a/pkg/local_object_storage/shard/inhume.go b/pkg/local_object_storage/shard/inhume.go
index 41bfc972d..5a84c2c98 100644
--- a/pkg/local_object_storage/shard/inhume.go
+++ b/pkg/local_object_storage/shard/inhume.go
@@ -61,10 +61,13 @@ var ErrLockObjectRemoval = meta.ErrLockObjectRemoval
 //
 // Returns ErrReadOnlyMode error if shard is in "read-only" mode.
 func (s *Shard) Inhume(prm InhumePrm) (InhumeRes, error) {
-	m := s.GetMode()
-	if m.ReadOnly() {
+	s.m.RLock()
+
+	if s.info.Mode.ReadOnly() {
+		s.m.RUnlock()
 		return InhumeRes{}, ErrReadOnlyMode
-	} else if m.NoMetabase() {
+	} else if s.info.Mode.NoMetabase() {
+		s.m.RUnlock()
 		return InhumeRes{}, ErrDegradedMode
 	}
 
@@ -91,6 +94,7 @@ func (s *Shard) Inhume(prm InhumePrm) (InhumeRes, error) {
 	res, err := s.metaBase.Inhume(metaPrm)
 	if err != nil {
 		if errors.Is(err, meta.ErrLockObjectRemoval) {
+			s.m.RUnlock()
 			return InhumeRes{}, ErrLockObjectRemoval
 		}
 
@@ -98,9 +102,13 @@ func (s *Shard) Inhume(prm InhumePrm) (InhumeRes, error) {
 			zap.String("error", err.Error()),
 		)
 
+		s.m.RUnlock()
+
 		return InhumeRes{}, fmt.Errorf("metabase inhume: %w", err)
 	}
 
+	s.m.RUnlock()
+
 	s.decObjectCounterBy(logical, res.AvailableInhumed())
 
 	if deletedLockObjs := res.DeletedLockObjects(); len(deletedLockObjs) != 0 {
diff --git a/pkg/local_object_storage/shard/list.go b/pkg/local_object_storage/shard/list.go
index 6b379ee7e..724a190a7 100644
--- a/pkg/local_object_storage/shard/list.go
+++ b/pkg/local_object_storage/shard/list.go
@@ -64,7 +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() {
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.NoMetabase() {
 		return SelectRes{}, ErrDegradedMode
 	}
 
diff --git a/pkg/local_object_storage/shard/lock.go b/pkg/local_object_storage/shard/lock.go
index b41045279..dd1d05658 100644
--- a/pkg/local_object_storage/shard/lock.go
+++ b/pkg/local_object_storage/shard/lock.go
@@ -15,7 +15,10 @@ import (
 //
 // Locked list should be unique. Panics if it is empty.
 func (s *Shard) Lock(idCnr cid.ID, locker oid.ID, locked []oid.ID) error {
-	m := s.GetMode()
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	m := s.info.Mode
 	if m.ReadOnly() {
 		return ErrReadOnlyMode
 	} else if m.NoMetabase() {
diff --git a/pkg/local_object_storage/shard/move.go b/pkg/local_object_storage/shard/move.go
index a9469cc34..fbe4051d7 100644
--- a/pkg/local_object_storage/shard/move.go
+++ b/pkg/local_object_storage/shard/move.go
@@ -23,7 +23,10 @@ func (p *ToMoveItPrm) SetAddress(addr oid.Address) {
 // ToMoveIt calls metabase.ToMoveIt method to mark object as relocatable to
 // another shard.
 func (s *Shard) ToMoveIt(prm ToMoveItPrm) (ToMoveItRes, error) {
-	m := s.GetMode()
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	m := s.info.Mode
 	if m.ReadOnly() {
 		return ToMoveItRes{}, ErrReadOnlyMode
 	} else if m.NoMetabase() {
diff --git a/pkg/local_object_storage/shard/put.go b/pkg/local_object_storage/shard/put.go
index 67c1edade..33eb3cd25 100644
--- a/pkg/local_object_storage/shard/put.go
+++ b/pkg/local_object_storage/shard/put.go
@@ -30,7 +30,10 @@ func (p *PutPrm) SetObject(obj *object.Object) {
 //
 // Returns ErrReadOnlyMode error if shard is in "read-only" mode.
 func (s *Shard) Put(prm PutPrm) (PutRes, error) {
-	m := s.GetMode()
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	m := s.info.Mode
 	if m.ReadOnly() {
 		return PutRes{}, ErrReadOnlyMode
 	}
diff --git a/pkg/local_object_storage/shard/range.go b/pkg/local_object_storage/shard/range.go
index 8ad39291e..0ff10f6e6 100644
--- a/pkg/local_object_storage/shard/range.go
+++ b/pkg/local_object_storage/shard/range.go
@@ -67,6 +67,9 @@ func (r RngRes) HasMeta() bool {
 // Returns an error of type apistatus.ObjectAlreadyRemoved if the requested object has been marked as removed in shard.
 // Returns the object.ErrObjectIsExpired if the object is presented but already expired.
 func (s *Shard) GetRange(prm RngPrm) (RngRes, error) {
+	s.m.RLock()
+	defer s.m.RUnlock()
+
 	cb := func(stor *blobstor.BlobStor, id []byte) (*object.Object, error) {
 		var getRngPrm common.GetRangePrm
 		getRngPrm.Address = prm.addr
@@ -103,7 +106,7 @@ func (s *Shard) GetRange(prm RngPrm) (RngRes, error) {
 		return obj, nil
 	}
 
-	skipMeta := prm.skipMeta || s.GetMode().NoMetabase()
+	skipMeta := prm.skipMeta || s.info.Mode.NoMetabase()
 	obj, hasMeta, err := s.fetchObjectData(prm.addr, skipMeta, cb, wc)
 
 	return RngRes{
diff --git a/pkg/local_object_storage/shard/select.go b/pkg/local_object_storage/shard/select.go
index 50156d1c3..b03626601 100644
--- a/pkg/local_object_storage/shard/select.go
+++ b/pkg/local_object_storage/shard/select.go
@@ -40,7 +40,10 @@ func (r SelectRes) AddressList() []oid.Address {
 // Returns any error encountered that
 // did not allow to completely select the objects.
 func (s *Shard) Select(prm SelectPrm) (SelectRes, error) {
-	if s.GetMode().NoMetabase() {
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.NoMetabase() {
 		return SelectRes{}, ErrDegradedMode
 	}
 
diff --git a/pkg/local_object_storage/shard/tree.go b/pkg/local_object_storage/shard/tree.go
index 56c3c0e06..f45a06d91 100644
--- a/pkg/local_object_storage/shard/tree.go
+++ b/pkg/local_object_storage/shard/tree.go
@@ -16,7 +16,11 @@ func (s *Shard) TreeMove(d pilorama.CIDDescriptor, treeID string, m *pilorama.Mo
 	if s.pilorama == nil {
 		return nil, ErrPiloramaDisabled
 	}
-	if s.GetMode().ReadOnly() {
+
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.ReadOnly() {
 		return nil, ErrReadOnlyMode
 	}
 	return s.pilorama.TreeMove(d, treeID, m)
@@ -27,7 +31,11 @@ func (s *Shard) TreeAddByPath(d pilorama.CIDDescriptor, treeID string, attr stri
 	if s.pilorama == nil {
 		return nil, ErrPiloramaDisabled
 	}
-	if s.GetMode().ReadOnly() {
+
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.ReadOnly() {
 		return nil, ErrReadOnlyMode
 	}
 	return s.pilorama.TreeAddByPath(d, treeID, attr, path, meta)
@@ -38,7 +46,11 @@ func (s *Shard) TreeApply(d pilorama.CIDDescriptor, treeID string, m *pilorama.M
 	if s.pilorama == nil {
 		return ErrPiloramaDisabled
 	}
-	if s.GetMode().ReadOnly() {
+
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.ReadOnly() {
 		return ErrReadOnlyMode
 	}
 	return s.pilorama.TreeApply(d, treeID, m)