From 731979a482aa6237725582e98002ee0a96420360 Mon Sep 17 00:00:00 2001
From: Evgenii Stratonikov <e.stratonikov@yadro.com>
Date: Wed, 5 Apr 2023 10:46:59 +0300
Subject: [PATCH] [#211] engine: Unify shard mode checks for tree operations

All operations must ensure the shard is not in a degraded mode.
Write operations must also ensure the shard is not in a read-only mode.

Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>
---
 pkg/local_object_storage/shard/tree.go | 75 ++++++++++++++++++++++++++
 1 file changed, 75 insertions(+)

diff --git a/pkg/local_object_storage/shard/tree.go b/pkg/local_object_storage/shard/tree.go
index 684c92e6..db07c001 100644
--- a/pkg/local_object_storage/shard/tree.go
+++ b/pkg/local_object_storage/shard/tree.go
@@ -23,6 +23,9 @@ func (s *Shard) TreeMove(d pilorama.CIDDescriptor, treeID string, m *pilorama.Mo
 	if s.info.Mode.ReadOnly() {
 		return nil, ErrReadOnlyMode
 	}
+	if s.info.Mode.NoMetabase() {
+		return nil, ErrDegradedMode
+	}
 	return s.pilorama.TreeMove(d, treeID, m)
 }
 
@@ -38,6 +41,9 @@ func (s *Shard) TreeAddByPath(d pilorama.CIDDescriptor, treeID string, attr stri
 	if s.info.Mode.ReadOnly() {
 		return nil, ErrReadOnlyMode
 	}
+	if s.info.Mode.NoMetabase() {
+		return nil, ErrDegradedMode
+	}
 	return s.pilorama.TreeAddByPath(d, treeID, attr, path, meta)
 }
 
@@ -53,6 +59,9 @@ func (s *Shard) TreeApply(cnr cidSDK.ID, treeID string, m *pilorama.Move, backgr
 	if s.info.Mode.ReadOnly() {
 		return ErrReadOnlyMode
 	}
+	if s.info.Mode.NoMetabase() {
+		return ErrDegradedMode
+	}
 	return s.pilorama.TreeApply(cnr, treeID, m, backgroundSync)
 }
 
@@ -61,6 +70,13 @@ func (s *Shard) TreeGetByPath(cid cidSDK.ID, treeID string, attr string, path []
 	if s.pilorama == nil {
 		return nil, ErrPiloramaDisabled
 	}
+
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.NoMetabase() {
+		return nil, ErrDegradedMode
+	}
 	return s.pilorama.TreeGetByPath(cid, treeID, attr, path, latest)
 }
 
@@ -69,6 +85,13 @@ func (s *Shard) TreeGetMeta(cid cidSDK.ID, treeID string, nodeID pilorama.Node)
 	if s.pilorama == nil {
 		return pilorama.Meta{}, 0, ErrPiloramaDisabled
 	}
+
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.NoMetabase() {
+		return pilorama.Meta{}, 0, ErrDegradedMode
+	}
 	return s.pilorama.TreeGetMeta(cid, treeID, nodeID)
 }
 
@@ -77,6 +100,13 @@ func (s *Shard) TreeGetChildren(cid cidSDK.ID, treeID string, nodeID pilorama.No
 	if s.pilorama == nil {
 		return nil, ErrPiloramaDisabled
 	}
+
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.NoMetabase() {
+		return nil, ErrDegradedMode
+	}
 	return s.pilorama.TreeGetChildren(cid, treeID, nodeID)
 }
 
@@ -85,6 +115,13 @@ func (s *Shard) TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64) (pilor
 	if s.pilorama == nil {
 		return pilorama.Move{}, ErrPiloramaDisabled
 	}
+
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.NoMetabase() {
+		return pilorama.Move{}, ErrDegradedMode
+	}
 	return s.pilorama.TreeGetOpLog(cid, treeID, height)
 }
 
@@ -93,6 +130,13 @@ func (s *Shard) TreeDrop(cid cidSDK.ID, treeID string) error {
 	if s.pilorama == nil {
 		return ErrPiloramaDisabled
 	}
+
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.NoMetabase() {
+		return ErrDegradedMode
+	}
 	return s.pilorama.TreeDrop(cid, treeID)
 }
 
@@ -101,6 +145,13 @@ func (s *Shard) TreeList(cid cidSDK.ID) ([]string, error) {
 	if s.pilorama == nil {
 		return nil, ErrPiloramaDisabled
 	}
+
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.NoMetabase() {
+		return nil, ErrDegradedMode
+	}
 	return s.pilorama.TreeList(cid)
 }
 
@@ -109,6 +160,13 @@ func (s *Shard) TreeExists(cid cidSDK.ID, treeID string) (bool, error) {
 	if s.pilorama == nil {
 		return false, ErrPiloramaDisabled
 	}
+
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.NoMetabase() {
+		return false, ErrDegradedMode
+	}
 	return s.pilorama.TreeExists(cid, treeID)
 }
 
@@ -117,6 +175,16 @@ func (s *Shard) TreeUpdateLastSyncHeight(cid cidSDK.ID, treeID string, height ui
 	if s.pilorama == nil {
 		return ErrPiloramaDisabled
 	}
+
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.ReadOnly() {
+		return ErrReadOnlyMode
+	}
+	if s.info.Mode.NoMetabase() {
+		return ErrDegradedMode
+	}
 	return s.pilorama.TreeUpdateLastSyncHeight(cid, treeID, height)
 }
 
@@ -125,5 +193,12 @@ func (s *Shard) TreeLastSyncHeight(cid cidSDK.ID, treeID string) (uint64, error)
 	if s.pilorama == nil {
 		return 0, ErrPiloramaDisabled
 	}
+
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	if s.info.Mode.NoMetabase() {
+		return 0, ErrDegradedMode
+	}
 	return s.pilorama.TreeLastSyncHeight(cid, treeID)
 }