From 4e5e0622e2587801800fd2ae3d15be86eccfc235 Mon Sep 17 00:00:00 2001
From: Injun Song <ijsong@gmail.com>
Date: Fri, 14 Jul 2023 09:47:22 +0900
Subject: [PATCH] feat(storagenode): add --trim-delay to set a delay before the
 deletion of log entries

This change adds a new flag --trim-delay to the storage node. It sets the delay before storage
removes logs. If the value is zero, Trim will delay the removal of prefix log entries until writing
additional log entries.
---
 cmd/varlogsn/cli.go         |  1 +
 cmd/varlogsn/flags.go       |  5 +++++
 cmd/varlogsn/varlogsn.go    |  3 +++
 internal/storage/config.go  | 10 ++++++++++
 internal/storage/storage.go |  1 +
 5 files changed, 20 insertions(+)

diff --git a/cmd/varlogsn/cli.go b/cmd/varlogsn/cli.go
index c7f779b70..a8ea3e73d 100644
--- a/cmd/varlogsn/cli.go
+++ b/cmd/varlogsn/cli.go
@@ -75,6 +75,7 @@ func newStartCommand() *cli.Command {
 			flagStorageMaxConcurrentCompaction,
 			flagStorageMetricsLogInterval,
 			flagStorageVerbose.BoolFlag(),
+			flagStorageTrimDelay,
 
 			// logger options
 			flags.LogDir,
diff --git a/cmd/varlogsn/flags.go b/cmd/varlogsn/flags.go
index df40fb1a6..18359b191 100644
--- a/cmd/varlogsn/flags.go
+++ b/cmd/varlogsn/flags.go
@@ -195,6 +195,11 @@ var (
 		EnvVars: []string{"STORAGE_METRICS_LOG_INTERVAL"},
 		Value:   storage.DefaultMetricsLogInterval,
 	}
+	flagStorageTrimDelay = &cli.DurationFlag{
+		Name:    "storage-trim-delay",
+		EnvVars: []string{"STORAGE_TRIM_DELAY"},
+		Usage:   "Delay before deletion of log entries caused by Trim operation. If zero, lazy deletion waits for other log entries to be appended.",
+	}
 
 	// flags for telemetry.
 	flagExporterType = flags.FlagDesc{
diff --git a/cmd/varlogsn/varlogsn.go b/cmd/varlogsn/varlogsn.go
index 4913f9108..5b934a51c 100644
--- a/cmd/varlogsn/varlogsn.go
+++ b/cmd/varlogsn/varlogsn.go
@@ -334,6 +334,9 @@ func parseStorageOptions(c *cli.Context) (opts []storage.Option, err error) {
 	if c.Bool(flagStorageVerbose.Name) {
 		opts = append(opts, storage.WithVerboseLogging())
 	}
+	if name := flagStorageTrimDelay.Name; c.IsSet(name) {
+		opts = append(opts, storage.WithTrimDelay(c.Duration(name)))
+	}
 	return opts, nil
 }
 
diff --git a/internal/storage/config.go b/internal/storage/config.go
index 4b6096020..d5cf05440 100644
--- a/internal/storage/config.go
+++ b/internal/storage/config.go
@@ -143,6 +143,7 @@ type config struct {
 	commitDBConfig     dbConfig
 	verbose            bool
 	metricsLogInterval time.Duration
+	trimDelay          time.Duration
 	logger             *zap.Logger
 	readOnly           bool
 }
@@ -246,6 +247,15 @@ func WithLogger(logger *zap.Logger) Option {
 	})
 }
 
+// WithTrimDelay sets the delay before storage removes logs. If the value is
+// zero, Trim will delay the removal of prefix log entries until writing
+// additional log entries.
+func WithTrimDelay(trimDelay time.Duration) Option {
+	return newFuncOption(func(cfg *config) {
+		cfg.trimDelay = trimDelay
+	})
+}
+
 // ReadOnly makes storage read-only. It is helpful only for testing. Usually,
 // users do not have to call it.
 func ReadOnly() Option {
diff --git a/internal/storage/storage.go b/internal/storage/storage.go
index e9e34d846..8ffda3ba0 100644
--- a/internal/storage/storage.go
+++ b/internal/storage/storage.go
@@ -112,6 +112,7 @@ func (s *Storage) newDB(path string, cfg *dbConfig) (*pebble.DB, error) {
 		MaxConcurrentCompactions:    func() int { return cfg.maxConcurrentCompaction },
 		Levels:                      make([]pebble.LevelOptions, 7),
 		ErrorIfExists:               false,
+		FlushDelayDeleteRange:       s.trimDelay,
 	}
 	pebbleOpts.Levels[0].TargetFileSize = cfg.l0TargetFileSize
 	for i := 0; i < len(pebbleOpts.Levels); i++ {