[nspcc-dev#1806] writecache: Allow to ignore read errors during flush

Signed-off-by: Evgenii Stratonikov <evgeniy@morphbits.ru>

pkg/local_object_storage/engine/writecache.go

@@ -6,7 +6,8 @@ import (
 
 // FlushWriteCachePrm groups the parameters of FlushWriteCache operation.
 type FlushWriteCachePrm struct {
-	shardID *shard.ID
+	shardID      *shard.ID
+	ignoreErrors bool
 }
 
 // SetShardID is an option to set shard ID.
@@ -16,6 +17,11 @@ func (p *FlushWriteCachePrm) SetShardID(id *shard.ID) {
 	p.shardID = id
 }
 
+// SetIgnoreErrors sets errors ignore flag..
+func (p *FlushWriteCachePrm) SetIgnoreErrors(ignore bool) {
+	p.ignoreErrors = ignore
+}
+
 // FlushWriteCacheRes groups the resulting values of FlushWriteCache operation.
 type FlushWriteCacheRes struct{}
 
@@ -29,5 +35,8 @@ func (e *StorageEngine) FlushWriteCache(p FlushWriteCachePrm) (FlushWriteCacheRe
 		return FlushWriteCacheRes{}, errShardNotFound
 	}
 
-	return FlushWriteCacheRes{}, sh.FlushWriteCache()
+	var prm shard.FlushWriteCachePrm
+	prm.SetIgnoreErrors(p.ignoreErrors)
+
+	return FlushWriteCacheRes{}, sh.FlushWriteCache(prm)
 }

pkg/local_object_storage/shard/writecache.go

@@ -6,13 +6,23 @@ import (
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard/mode"
 )
 
+// FlushWriteCachePrm represents parameters of a `FlushWriteCache` operation.
+type FlushWriteCachePrm struct {
+	ignoreErrors bool
+}
+
+// SetIgnoreErrors sets the flag to ignore read-errors during flush.
+func (p *FlushWriteCachePrm) SetIgnoreErrors(ignore bool) {
+	p.ignoreErrors = ignore
+}
+
 // errWriteCacheDisabled is returned when an operation on write-cache is performed,
 // but write-cache is disabled.
 var errWriteCacheDisabled = errors.New("write-cache is disabled")
 
 // FlushWriteCache moves writecache in read-only mode and flushes all data from it.
 // After the operation writecache will remain read-only mode.
-func (s *Shard) FlushWriteCache() error {
+func (s *Shard) FlushWriteCache(p FlushWriteCachePrm) error {
 	if !s.hasWriteCache() {
 		return errWriteCacheDisabled
 	}
@@ -32,5 +42,5 @@ func (s *Shard) FlushWriteCache() error {
 		return err
 	}
 
-	return s.writeCache.Flush()
+	return s.writeCache.Flush(p.ignoreErrors)
 }

pkg/local_object_storage/writecache/flush.go

@@ -233,7 +233,7 @@ func (c *cache) flushObject(obj *object.Object) error {
 // Flush flushes all objects from the write-cache to the main storage.
 // Write-cache must be in readonly mode to ensure correctness of an operation and
 // to prevent interference with background flush workers.
-func (c *cache) Flush() error {
+func (c *cache) Flush(ignoreErrors bool) error {
 	c.modeMtx.RLock()
 	defer c.modeMtx.RUnlock()
 
@@ -242,6 +242,7 @@ func (c *cache) Flush() error {
 	}
 
 	var prm common.IteratePrm
+	prm.IgnoreErrors = ignoreErrors
 	prm.LazyHandler = func(addr oid.Address, f func() ([]byte, error)) error {
 		_, ok := c.flushed.Peek(addr.EncodeToString())
 		if ok {
@@ -250,12 +251,18 @@ func (c *cache) Flush() error {
 
 		data, err := f()
 		if err != nil {
+			if ignoreErrors {
+				return nil
+			}
 			return err
 		}
 
 		var obj object.Object
 		err = obj.Unmarshal(data)
 		if err != nil {
+			if ignoreErrors {
+				return nil
+			}
 			return err
 		}
 
@@ -279,11 +286,17 @@ func (c *cache) Flush() error {
 			}
 
 			if err := addr.DecodeString(sa); err != nil {
+				if ignoreErrors {
+					continue
+				}
 				return err
 			}
 
 			var obj object.Object
 			if err := obj.Unmarshal(data); err != nil {
+				if ignoreErrors {
+					continue
+				}
 				return err
 			}