From fdeea1dface8a2865fbe14ca54701185a13f1d94 Mon Sep 17 00:00:00 2001
From: Pavel Karpy <carpawell@nspcc.ru>
Date: Tue, 15 Nov 2022 15:46:32 +0300
Subject: [PATCH] [#2057] meta: Fix concurrent mode changes

Includes:
1. mode change read lock operation in every exported method that r/w the
underlying database;
2. returning `ErrDegradedMode` logical error if any exported method is
called in degraded (without a metabase) mode.

Signed-off-by: Pavel Karpy <carpawell@nspcc.ru>
---
 CHANGELOG.md                                  |  1 +
 .../metabase/containers.go                    |  7 +++++++
 pkg/local_object_storage/metabase/control.go  | 14 +++++++++++++
 pkg/local_object_storage/metabase/counter.go  |  7 +++++++
 pkg/local_object_storage/metabase/delete.go   |  4 ++++
 pkg/local_object_storage/metabase/exists.go   |  4 ++++
 .../metabase/graveyard.go                     | 21 +++++++++++++++++++
 pkg/local_object_storage/metabase/info.go     |  3 +++
 .../metabase/iterators.go                     | 14 +++++++++++++
 pkg/local_object_storage/metabase/list.go     |  4 ++++
 pkg/local_object_storage/metabase/lock.go     | 18 ++++++++++++++++
 pkg/local_object_storage/metabase/movable.go  | 12 +++++++++++
 pkg/local_object_storage/metabase/put.go      |  4 ++++
 pkg/local_object_storage/metabase/select.go   |  4 ++++
 pkg/local_object_storage/metabase/shard_id.go | 14 +++++++++++++
 .../metabase/storage_id.go                    | 11 ++++++++++
 16 files changed, 142 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3e256944b..ba1897aca 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -33,6 +33,7 @@ Changelog for NeoFS Node
 - Assembly process triggered by a request with a bearer token (#2040)
 - Losing locking context after metabase resync (#1502)
 - Removing all trees by container ID if tree ID is empty in `pilorama.Forest.TreeDrop` (#1940)
+- Concurrent mode changes in the metabase and blobstor (#2057)
 
 ### Removed
 ### Updated
diff --git a/pkg/local_object_storage/metabase/containers.go b/pkg/local_object_storage/metabase/containers.go
index 278f4adf3..c506ee641 100644
--- a/pkg/local_object_storage/metabase/containers.go
+++ b/pkg/local_object_storage/metabase/containers.go
@@ -42,6 +42,13 @@ func (db *DB) containers(tx *bbolt.Tx) ([]cid.ID, error) {
 }
 
 func (db *DB) ContainerSize(id cid.ID) (size uint64, err error) {
+	db.modeMtx.RLock()
+	defer db.modeMtx.RUnlock()
+
+	if db.mode.NoMetabase() {
+		return 0, ErrDegradedMode
+	}
+
 	err = db.boltDB.View(func(tx *bbolt.Tx) error {
 		size, err = db.containerSize(tx, id)
 
diff --git a/pkg/local_object_storage/metabase/control.go b/pkg/local_object_storage/metabase/control.go
index 018fe075b..23e99a4fb 100644
--- a/pkg/local_object_storage/metabase/control.go
+++ b/pkg/local_object_storage/metabase/control.go
@@ -81,6 +81,13 @@ func (db *DB) Init() error {
 // Reset resets metabase. Works similar to Init but cleans up all static buckets and
 // removes all dynamic (CID-dependent) ones in non-blank BoltDB instances.
 func (db *DB) Reset() error {
+	db.modeMtx.RLock()
+	defer db.modeMtx.RUnlock()
+
+	if db.mode.NoMetabase() {
+		return ErrDegradedMode
+	}
+
 	return db.init(true)
 }
 
@@ -147,6 +154,13 @@ func (db *DB) init(reset bool) error {
 
 // SyncCounters forces to synchronize the object counters.
 func (db *DB) SyncCounters() error {
+	db.modeMtx.RLock()
+	defer db.modeMtx.RUnlock()
+
+	if db.mode.NoMetabase() {
+		return ErrDegradedMode
+	}
+
 	return db.boltDB.Update(func(tx *bbolt.Tx) error {
 		return syncCounter(tx, true)
 	})
diff --git a/pkg/local_object_storage/metabase/counter.go b/pkg/local_object_storage/metabase/counter.go
index 5085c3ee0..8ce2152a2 100644
--- a/pkg/local_object_storage/metabase/counter.go
+++ b/pkg/local_object_storage/metabase/counter.go
@@ -43,6 +43,13 @@ func (o ObjectCounters) Phy() uint64 {
 // Returns only the errors that do not allow reading counter
 // in Bolt database.
 func (db *DB) ObjectCounters() (cc ObjectCounters, err error) {
+	db.modeMtx.RLock()
+	defer db.modeMtx.RUnlock()
+
+	if db.mode.NoMetabase() {
+		return ObjectCounters{}, ErrDegradedMode
+	}
+
 	err = db.boltDB.View(func(tx *bbolt.Tx) error {
 		b := tx.Bucket(shardInfoBucket)
 		if b != nil {
diff --git a/pkg/local_object_storage/metabase/delete.go b/pkg/local_object_storage/metabase/delete.go
index e69372b70..c990844c7 100644
--- a/pkg/local_object_storage/metabase/delete.go
+++ b/pkg/local_object_storage/metabase/delete.go
@@ -57,6 +57,10 @@ func (db *DB) Delete(prm DeletePrm) (DeleteRes, error) {
 	db.modeMtx.RLock()
 	defer db.modeMtx.RUnlock()
 
+	if db.mode.NoMetabase() {
+		return DeleteRes{}, ErrDegradedMode
+	}
+
 	var rawRemoved uint64
 	var availableRemoved uint64
 	var err error
diff --git a/pkg/local_object_storage/metabase/exists.go b/pkg/local_object_storage/metabase/exists.go
index efd77066d..a2c10fab5 100644
--- a/pkg/local_object_storage/metabase/exists.go
+++ b/pkg/local_object_storage/metabase/exists.go
@@ -44,6 +44,10 @@ func (db *DB) Exists(prm ExistsPrm) (res ExistsRes, err error) {
 	db.modeMtx.RLock()
 	defer db.modeMtx.RUnlock()
 
+	if db.mode.NoMetabase() {
+		return res, ErrDegradedMode
+	}
+
 	currEpoch := db.epochState.CurrentEpoch()
 
 	err = db.boltDB.View(func(tx *bbolt.Tx) error {
diff --git a/pkg/local_object_storage/metabase/graveyard.go b/pkg/local_object_storage/metabase/graveyard.go
index cba91d7dc..f079119b8 100644
--- a/pkg/local_object_storage/metabase/graveyard.go
+++ b/pkg/local_object_storage/metabase/graveyard.go
@@ -58,6 +58,13 @@ func (g *GarbageIterationPrm) SetOffset(offset oid.Address) {
 // If h returns ErrInterruptIterator, nil returns immediately.
 // Returns other errors of h directly.
 func (db *DB) IterateOverGarbage(p GarbageIterationPrm) error {
+	db.modeMtx.RLock()
+	defer db.modeMtx.RUnlock()
+
+	if db.mode.NoMetabase() {
+		return ErrDegradedMode
+	}
+
 	return db.boltDB.View(func(tx *bbolt.Tx) error {
 		return db.iterateDeletedObj(tx, gcHandler{p.h}, p.offset)
 	})
@@ -118,6 +125,13 @@ func (g *GraveyardIterationPrm) SetOffset(offset oid.Address) {
 // If h returns ErrInterruptIterator, nil returns immediately.
 // Returns other errors of h directly.
 func (db *DB) IterateOverGraveyard(p GraveyardIterationPrm) error {
+	db.modeMtx.RLock()
+	defer db.modeMtx.RUnlock()
+
+	if db.mode.NoMetabase() {
+		return ErrDegradedMode
+	}
+
 	return db.boltDB.View(func(tx *bbolt.Tx) error {
 		return db.iterateDeletedObj(tx, graveyardHandler{p.h}, p.offset)
 	})
@@ -218,6 +232,13 @@ func graveFromKV(k, v []byte) (res TombstonedObject, err error) {
 //
 // Returns any error appeared during deletion process.
 func (db *DB) DropGraves(tss []TombstonedObject) error {
+	db.modeMtx.RLock()
+	defer db.modeMtx.RUnlock()
+
+	if db.mode.NoMetabase() {
+		return ErrDegradedMode
+	}
+
 	buf := make([]byte, addressKeySize)
 
 	return db.boltDB.Update(func(tx *bbolt.Tx) error {
diff --git a/pkg/local_object_storage/metabase/info.go b/pkg/local_object_storage/metabase/info.go
index 7d1bfefb7..4d6a06239 100644
--- a/pkg/local_object_storage/metabase/info.go
+++ b/pkg/local_object_storage/metabase/info.go
@@ -15,5 +15,8 @@ type Info struct {
 
 // DumpInfo returns information about the DB.
 func (db *DB) DumpInfo() Info {
+	db.modeMtx.RLock()
+	defer db.modeMtx.RUnlock()
+
 	return db.info
 }
diff --git a/pkg/local_object_storage/metabase/iterators.go b/pkg/local_object_storage/metabase/iterators.go
index d5c8173ee..6deadde27 100644
--- a/pkg/local_object_storage/metabase/iterators.go
+++ b/pkg/local_object_storage/metabase/iterators.go
@@ -44,6 +44,13 @@ var ErrInterruptIterator = logicerr.New("iterator is interrupted")
 // If h returns ErrInterruptIterator, nil returns immediately.
 // Returns other errors of h directly.
 func (db *DB) IterateExpired(epoch uint64, h ExpiredObjectHandler) error {
+	db.modeMtx.RLock()
+	defer db.modeMtx.RUnlock()
+
+	if db.mode.NoMetabase() {
+		return ErrDegradedMode
+	}
+
 	return db.boltDB.View(func(tx *bbolt.Tx) error {
 		return db.iterateExpired(tx, epoch, h)
 	})
@@ -119,6 +126,13 @@ func (db *DB) iterateExpired(tx *bbolt.Tx, epoch uint64, h ExpiredObjectHandler)
 //
 // Does not modify tss.
 func (db *DB) IterateCoveredByTombstones(tss map[string]oid.Address, h func(oid.Address) error) error {
+	db.modeMtx.RLock()
+	defer db.modeMtx.RUnlock()
+
+	if db.mode.NoMetabase() {
+		return ErrDegradedMode
+	}
+
 	return db.boltDB.View(func(tx *bbolt.Tx) error {
 		return db.iterateCoveredByTombstones(tx, tss, h)
 	})
diff --git a/pkg/local_object_storage/metabase/list.go b/pkg/local_object_storage/metabase/list.go
index 4bd065be9..4dbcc0026 100644
--- a/pkg/local_object_storage/metabase/list.go
+++ b/pkg/local_object_storage/metabase/list.go
@@ -64,6 +64,10 @@ func (db *DB) ListWithCursor(prm ListPrm) (res ListRes, err error) {
 	db.modeMtx.RLock()
 	defer db.modeMtx.RUnlock()
 
+	if db.mode.NoMetabase() {
+		return res, ErrDegradedMode
+	}
+
 	result := make([]objectcore.AddressWithType, 0, prm.count)
 
 	err = db.boltDB.View(func(tx *bbolt.Tx) error {
diff --git a/pkg/local_object_storage/metabase/lock.go b/pkg/local_object_storage/metabase/lock.go
index a15a9b6f7..dbd37a8cf 100644
--- a/pkg/local_object_storage/metabase/lock.go
+++ b/pkg/local_object_storage/metabase/lock.go
@@ -29,6 +29,10 @@ func (db *DB) Lock(cnr cid.ID, locker oid.ID, locked []oid.ID) error {
 	db.modeMtx.RLock()
 	defer db.modeMtx.RUnlock()
 
+	if db.mode.NoMetabase() {
+		return ErrDegradedMode
+	}
+
 	if len(locked) == 0 {
 		panic("empty locked list")
 	}
@@ -91,6 +95,13 @@ func (db *DB) Lock(cnr cid.ID, locker oid.ID, locked []oid.ID) error {
 
 // FreeLockedBy unlocks all objects in DB which are locked by lockers.
 func (db *DB) FreeLockedBy(lockers []oid.Address) error {
+	db.modeMtx.RLock()
+	defer db.modeMtx.RUnlock()
+
+	if db.mode.NoMetabase() {
+		return ErrDegradedMode
+	}
+
 	return db.boltDB.Update(func(tx *bbolt.Tx) error {
 		var err error
 
@@ -202,6 +213,13 @@ func (i IsLockedRes) Locked() bool {
 //
 // Returns only non-logical errors related to underlying database.
 func (db *DB) IsLocked(prm IsLockedPrm) (res IsLockedRes, err error) {
+	db.modeMtx.RLock()
+	defer db.modeMtx.RUnlock()
+
+	if db.mode.NoMetabase() {
+		return res, ErrDegradedMode
+	}
+
 	return res, db.boltDB.View(func(tx *bbolt.Tx) error {
 		res.locked = objectLocked(tx, prm.addr.Container(), prm.addr.Object())
 		return nil
diff --git a/pkg/local_object_storage/metabase/movable.go b/pkg/local_object_storage/metabase/movable.go
index 4ae0b052b..22538d0b2 100644
--- a/pkg/local_object_storage/metabase/movable.go
+++ b/pkg/local_object_storage/metabase/movable.go
@@ -52,6 +52,10 @@ func (db *DB) ToMoveIt(prm ToMoveItPrm) (res ToMoveItRes, err error) {
 	db.modeMtx.RLock()
 	defer db.modeMtx.RUnlock()
 
+	if db.mode.NoMetabase() {
+		return res, ErrDegradedMode
+	}
+
 	key := make([]byte, addressKeySize)
 	key = addressKey(prm.addr, key)
 
@@ -68,6 +72,10 @@ func (db *DB) DoNotMove(prm DoNotMovePrm) (res DoNotMoveRes, err error) {
 	db.modeMtx.RLock()
 	defer db.modeMtx.RUnlock()
 
+	if db.mode.NoMetabase() {
+		return res, ErrDegradedMode
+	}
+
 	key := make([]byte, addressKeySize)
 	key = addressKey(prm.addr, key)
 
@@ -84,6 +92,10 @@ func (db *DB) Movable(_ MovablePrm) (MovableRes, error) {
 	db.modeMtx.RLock()
 	defer db.modeMtx.RUnlock()
 
+	if db.mode.NoMetabase() {
+		return MovableRes{}, ErrDegradedMode
+	}
+
 	var strAddrs []string
 
 	err := db.boltDB.View(func(tx *bbolt.Tx) error {
diff --git a/pkg/local_object_storage/metabase/put.go b/pkg/local_object_storage/metabase/put.go
index b39005633..5d4787805 100644
--- a/pkg/local_object_storage/metabase/put.go
+++ b/pkg/local_object_storage/metabase/put.go
@@ -56,6 +56,10 @@ func (db *DB) Put(prm PutPrm) (res PutRes, err error) {
 	db.modeMtx.RLock()
 	defer db.modeMtx.RUnlock()
 
+	if db.mode.NoMetabase() {
+		return res, ErrDegradedMode
+	}
+
 	currEpoch := db.epochState.CurrentEpoch()
 
 	err = db.boltDB.Batch(func(tx *bbolt.Tx) error {
diff --git a/pkg/local_object_storage/metabase/select.go b/pkg/local_object_storage/metabase/select.go
index 1da56b12f..0b5492f90 100644
--- a/pkg/local_object_storage/metabase/select.go
+++ b/pkg/local_object_storage/metabase/select.go
@@ -59,6 +59,10 @@ func (db *DB) Select(prm SelectPrm) (res SelectRes, err error) {
 	db.modeMtx.RLock()
 	defer db.modeMtx.RUnlock()
 
+	if db.mode.NoMetabase() {
+		return res, ErrDegradedMode
+	}
+
 	if blindlyProcess(prm.filters) {
 		return res, nil
 	}
diff --git a/pkg/local_object_storage/metabase/shard_id.go b/pkg/local_object_storage/metabase/shard_id.go
index 312f3900c..5a4afe413 100644
--- a/pkg/local_object_storage/metabase/shard_id.go
+++ b/pkg/local_object_storage/metabase/shard_id.go
@@ -13,6 +13,13 @@ var (
 // ReadShardID reads shard id from db.
 // If id is missing, returns nil, nil.
 func (db *DB) ReadShardID() ([]byte, error) {
+	db.modeMtx.RLock()
+	defer db.modeMtx.RUnlock()
+
+	if db.mode.NoMetabase() {
+		return nil, ErrDegradedMode
+	}
+
 	var id []byte
 	err := db.boltDB.View(func(tx *bbolt.Tx) error {
 		b := tx.Bucket(shardInfoBucket)
@@ -26,6 +33,13 @@ func (db *DB) ReadShardID() ([]byte, error) {
 
 // WriteShardID writes shard it to db.
 func (db *DB) WriteShardID(id []byte) error {
+	db.modeMtx.RLock()
+	defer db.modeMtx.RUnlock()
+
+	if db.mode.NoMetabase() {
+		return ErrDegradedMode
+	}
+
 	return db.boltDB.Update(func(tx *bbolt.Tx) error {
 		b, err := tx.CreateBucketIfNotExists(shardInfoBucket)
 		if err != nil {
diff --git a/pkg/local_object_storage/metabase/storage_id.go b/pkg/local_object_storage/metabase/storage_id.go
index 34f5b0fbb..3fed824c5 100644
--- a/pkg/local_object_storage/metabase/storage_id.go
+++ b/pkg/local_object_storage/metabase/storage_id.go
@@ -29,6 +29,13 @@ func (r StorageIDRes) StorageID() []byte {
 // StorageID returns storage descriptor for objects from the blobstor.
 // It is put together with the object can makes get/delete operation faster.
 func (db *DB) StorageID(prm StorageIDPrm) (res StorageIDRes, err error) {
+	db.modeMtx.RLock()
+	defer db.modeMtx.RUnlock()
+
+	if db.mode.NoMetabase() {
+		return res, ErrDegradedMode
+	}
+
 	err = db.boltDB.View(func(tx *bbolt.Tx) error {
 		res.id, err = db.storageID(tx, prm.addr)
 
@@ -77,6 +84,10 @@ func (db *DB) UpdateStorageID(prm UpdateStorageIDPrm) (res UpdateStorageIDRes, e
 	db.modeMtx.RLock()
 	defer db.modeMtx.RUnlock()
 
+	if db.mode.NoMetabase() {
+		return res, ErrDegradedMode
+	}
+
 	currEpoch := db.epochState.CurrentEpoch()
 
 	err = db.boltDB.Batch(func(tx *bbolt.Tx) error {