internal/logs/logs.go

@@ -296,7 +296,6 @@ const (
 	WritecacheFillingFlushMarksForObjectsInDatabase                         = "filling flush marks for objects in database"
 	WritecacheFinishedUpdatingFlushMarks                                    = "finished updating flush marks"
 	WritecacheCantRemoveObjectsFromTheDatabase                              = "can't remove objects from the database"
-	WritecacheCantParseAddress                                              = "can't parse address"
 	WritecacheCantRemoveObjectFromWritecache                                = "can't remove object from write-cache"
 	WritecacheDBValueLogGCRunCompleted                                      = "value log GC run completed"
 	WritecacheBadgerObjAlreadyScheduled                                     = "object already scheduled for flush"

pkg/local_object_storage/writecache/writecachebadger/flush.go

@@ -97,6 +97,9 @@ func (c *collector) send(ctx context.Context, cancel func(), buf *z.Buffer) erro
 
 // runFlushLoop starts background workers which periodically flush objects to the blobstor.
 func (c *cache) runFlushLoop(ctx context.Context) {
+	if c.disableBackgroundFlush {
+		return
+	}
 	for i := 0; i < c.workersCount; i++ {
 		c.wg.Add(1)
 		go c.workerFlushSmall(ctx)

pkg/local_object_storage/writecache/writecachebadger/flush_test.go

@@ -28,6 +28,7 @@ func TestFlush(t *testing.T) {
 				WithMetabase(mb),
 				WithBlobstor(bs),
 				WithGCInterval(1 * time.Second),
+				WithDisableBackgroundFlush(),
 			}, opts...)...)
 	}
 

pkg/local_object_storage/writecache/writecachebadger/options.go

@@ -34,6 +34,8 @@ type options struct {
 	metrics writecache.Metrics
 	// gcInterval is the interval duration to run the GC cycle.
 	gcInterval time.Duration
+	// disableBackgroundFlush is for testing purposes only.
+	disableBackgroundFlush bool
 }
 
 // WithLogger sets logger.
@@ -108,3 +110,10 @@ func WithGCInterval(d time.Duration) Option {
 		o.gcInterval = d
 	}
 }
+
+// WithDisableBackgroundFlush disables background flush, for testing purposes only.
+func WithDisableBackgroundFlush() Option {
+	return func(o *options) {
+		o.disableBackgroundFlush = true
+	}
+}

pkg/local_object_storage/writecache/writecachebbolt/flush.go

@@ -36,6 +36,9 @@ const (
 
 // runFlushLoop starts background workers which periodically flush objects to the blobstor.
 func (c *cache) runFlushLoop(ctx context.Context) {
+	if c.disableBackgroundFlush {
+		return
+	}
 	for i := 0; i < c.workersCount; i++ {
 		c.wg.Add(1)
 		go c.workerFlushSmall(ctx)
@@ -200,7 +203,7 @@ func (c *cache) flushFSTree(ctx context.Context, ignoreErrors bool) error {
 			return err
 		}
 
-		c.deleteFromDisk(ctx, []string{sAddr})
+		c.deleteFromDisk(ctx, e.Address)
 		return nil
 	}
 

pkg/local_object_storage/writecache/writecachebbolt/flush_test.go

@@ -31,6 +31,7 @@ func TestFlush(t *testing.T) {
 				WithSmallObjectSize(smallSize),
 				WithMetabase(mb),
 				WithBlobstor(bs),
+				WithDisableBackgroundFlush(),
 			}, opts...)...)
 	}
 

pkg/local_object_storage/writecache/writecachebbolt/options.go

@@ -44,6 +44,8 @@ type options struct {
 	openFile func(string, int, fs.FileMode) (*os.File, error)
 	// metrics is metrics implementation
 	metrics writecache.Metrics
+	// disableBackgroundFlush is for testing purposes only.
+	disableBackgroundFlush bool
 }
 
 // WithLogger sets logger.
@@ -155,3 +157,10 @@ func WithMetrics(metrics writecache.Metrics) Option {
 		o.metrics = metrics
 	}
 }
+
+// WithDisableBackgroundFlush disables background flush, for testing purposes only.
+func WithDisableBackgroundFlush() Option {
+	return func(o *options) {
+		o.disableBackgroundFlush = true
+	}
+}

pkg/local_object_storage/writecache/writecachebbolt/storage.go

@@ -93,31 +93,13 @@ func (c *cache) deleteFromDB(key string) {
 	}
 }
 
-func (c *cache) deleteFromDisk(ctx context.Context, keys []string) []string {
+func (c *cache) deleteFromDisk(ctx context.Context, addr oid.Address) {
-	if len(keys) == 0 {
-		return keys
-	}
-
-	var copyIndex int
-	var addr oid.Address
-
-	for i := range keys {
-		if err := addr.DecodeString(keys[i]); err != nil {
-			c.log.Error(logs.WritecacheCantParseAddress, zap.String("address", keys[i]))
-			continue
-		}
-
 	_, err := c.fsTree.Delete(ctx, common.DeletePrm{Address: addr})
 	if err != nil && !client.IsErrObjectNotFound(err) {
 		c.log.Error(logs.WritecacheCantRemoveObjectFromWritecache, zap.Error(err))
-
-			// Save the key for the next iteration.
-			keys[copyIndex] = keys[i]
-			copyIndex++
-			continue
 	} else if err == nil {
 		storagelog.Write(c.log,
-				storagelog.AddressField(keys[i]),
+			storagelog.AddressField(addr.EncodeToString()),
 			storagelog.StorageTypeField(wcStorageType),
 			storagelog.OpField("fstree DELETE"),
 		)
@@ -125,7 +107,4 @@ func (c *cache) deleteFromDisk(ctx context.Context, keys []string) []string {
 		// counter changed by fstree
 		c.estimateCacheSize()
 	}
-	}
-
-	return keys[:copyIndex]
 }