From f6f327b5dc5c77f8a96f4df59db45df2b858aed9 Mon Sep 17 00:00:00 2001
From: Anton Nikiforov <an.nikiforov@yadro.com>
Date: Tue, 20 Feb 2024 17:24:57 +0300
Subject: [PATCH] [#959] writecache: Avoid manipulation with cache in
 `DEGRADED` mode

Signed-off-by: Anton Nikiforov <an.nikiforov@yadro.com>
---
 pkg/local_object_storage/writecache/delete.go    |  4 ++++
 pkg/local_object_storage/writecache/flush.go     |  5 ++++-
 pkg/local_object_storage/writecache/get.go       | 16 ++++++++++++++++
 pkg/local_object_storage/writecache/mode.go      |  6 ++++++
 pkg/local_object_storage/writecache/put.go       |  3 +++
 .../writecache/writecache.go                     |  2 ++
 6 files changed, 35 insertions(+), 1 deletion(-)

diff --git a/pkg/local_object_storage/writecache/delete.go b/pkg/local_object_storage/writecache/delete.go
index cedb9f4ae..b1a0511ee 100644
--- a/pkg/local_object_storage/writecache/delete.go
+++ b/pkg/local_object_storage/writecache/delete.go
@@ -19,6 +19,7 @@ import (
 //
 // Returns an error of type apistatus.ObjectNotFound if object is missing in write-cache.
 // Returns ErrNotInitialized if write-cache has not been initialized yet.
+// Returns ErrDegraded if write-cache is in DEGRADED mode.
 func (c *cache) Delete(ctx context.Context, addr oid.Address) error {
 	ctx, span := tracing.StartSpanFromContext(ctx, "writecache.Delete",
 		trace.WithAttributes(
@@ -40,6 +41,9 @@ func (c *cache) Delete(ctx context.Context, addr oid.Address) error {
 	if c.readOnly() {
 		return ErrReadOnly
 	}
+	if c.noMetabase() {
+		return ErrDegraded
+	}
 
 	saddr := addr.EncodeToString()
 
diff --git a/pkg/local_object_storage/writecache/flush.go b/pkg/local_object_storage/writecache/flush.go
index c4e4b3b34..c3e19c4d2 100644
--- a/pkg/local_object_storage/writecache/flush.go
+++ b/pkg/local_object_storage/writecache/flush.go
@@ -155,7 +155,7 @@ func (c *cache) workerFlushBig(ctx context.Context) {
 		select {
 		case <-tick.C:
 			c.modeMtx.RLock()
-			if c.readOnly() {
+			if c.readOnly() || c.noMetabase() {
 				c.modeMtx.RUnlock()
 				break
 			}
@@ -281,6 +281,9 @@ func (c *cache) Flush(ctx context.Context, ignoreErrors, seal bool) error {
 
 	c.modeMtx.Lock() // exclusive lock to not to conflict with background flush
 	defer c.modeMtx.Unlock()
+	if c.noMetabase() {
+		return ErrDegraded
+	}
 
 	if err := c.flush(ctx, ignoreErrors); err != nil {
 		return err
diff --git a/pkg/local_object_storage/writecache/get.go b/pkg/local_object_storage/writecache/get.go
index 5194446ec..bf26833bd 100644
--- a/pkg/local_object_storage/writecache/get.go
+++ b/pkg/local_object_storage/writecache/get.go
@@ -29,6 +29,14 @@ func (c *cache) Get(ctx context.Context, addr oid.Address) (*objectSDK.Object, e
 		))
 	defer span.End()
 
+	if !c.modeMtx.TryRLock() {
+		return nil, ErrNotInitialized
+	}
+	defer c.modeMtx.RUnlock()
+	if c.mode.NoMetabase() {
+		return nil, ErrDegraded
+	}
+
 	obj, err := c.getInternal(ctx, saddr, addr)
 	return obj, metaerr.Wrap(err)
 }
@@ -71,6 +79,14 @@ func (c *cache) Head(ctx context.Context, addr oid.Address) (*objectSDK.Object,
 		))
 	defer span.End()
 
+	if !c.modeMtx.TryRLock() {
+		return nil, ErrNotInitialized
+	}
+	defer c.modeMtx.RUnlock()
+	if c.mode.NoMetabase() {
+		return nil, ErrDegraded
+	}
+
 	obj, err := c.getInternal(ctx, saddr, addr)
 	if err != nil {
 		return nil, metaerr.Wrap(err)
diff --git a/pkg/local_object_storage/writecache/mode.go b/pkg/local_object_storage/writecache/mode.go
index 7c6439fe9..a10e593aa 100644
--- a/pkg/local_object_storage/writecache/mode.go
+++ b/pkg/local_object_storage/writecache/mode.go
@@ -76,3 +76,9 @@ func (c *cache) setMode(ctx context.Context, m mode.Mode, ignoreErrors bool) err
 func (c *cache) readOnly() bool {
 	return c.mode.ReadOnly()
 }
+
+// noMetabase returns true if c is operating without the metabase.
+// `c.modeMtx` must be taken.
+func (c *cache) noMetabase() bool {
+	return c.mode.NoMetabase()
+}
diff --git a/pkg/local_object_storage/writecache/put.go b/pkg/local_object_storage/writecache/put.go
index 8b6c09e9a..0e419f95b 100644
--- a/pkg/local_object_storage/writecache/put.go
+++ b/pkg/local_object_storage/writecache/put.go
@@ -41,6 +41,9 @@ func (c *cache) Put(ctx context.Context, prm common.PutPrm) (common.PutRes, erro
 	if c.readOnly() {
 		return common.PutRes{}, ErrReadOnly
 	}
+	if c.noMetabase() {
+		return common.PutRes{}, ErrDegraded
+	}
 
 	sz := uint64(len(prm.RawData))
 	if sz > c.maxObjectSize {
diff --git a/pkg/local_object_storage/writecache/writecache.go b/pkg/local_object_storage/writecache/writecache.go
index 112594ec0..71dba61cf 100644
--- a/pkg/local_object_storage/writecache/writecache.go
+++ b/pkg/local_object_storage/writecache/writecache.go
@@ -59,6 +59,8 @@ type Metabase interface {
 var (
 	// ErrReadOnly is returned when Put/Write is performed in a read-only mode.
 	ErrReadOnly = logicerr.New("write-cache is in read-only mode")
+	// ErrDegraded is returned when writecache is in degraded mode.
+	ErrDegraded = logicerr.New("write-cache is in degraded mode")
 	// ErrNotInitialized is returned when write-cache is initializing.
 	ErrNotInitialized = logicerr.New("write-cache is not initialized yet")
 	// ErrBigObject is returned when object is too big to be placed in cache.
-- 
2.45.2