diff --git a/cmd/varlogsn/cli.go b/cmd/varlogsn/cli.go
index ce11b5b53..49c15c406 100644
--- a/cmd/varlogsn/cli.go
+++ b/cmd/varlogsn/cli.go
@@ -73,6 +73,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 14c98037b..18359b191 100644
--- a/cmd/varlogsn/flags.go
+++ b/cmd/varlogsn/flags.go
@@ -195,4 +195,36 @@ 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{
+		Name:  "exporter-type",
+		Usage: "exporter type: stdout, otlp or noop",
+		Envs:  []string{"EXPORTER_TYPE"},
+	}
+	flagExporterStopTimeout = flags.FlagDesc{
+		Name:  "expoter-stop-timeout",
+		Usage: "timeout for stopping exporter",
+		Envs:  []string{"EXPORTER_STOP_TIMEOUT"},
+	}
+	flagStdoutExporterPrettyPrint = flags.FlagDesc{
+		Name:  "exporter-pretty-print",
+		Usage: "pretty print when using stdout exporter",
+		Envs:  []string{"EXPORTER_PRETTY_PRINT"},
+	}
+	flagOTLPExporterInsecure = flags.FlagDesc{
+		Name:  "exporter-otlp-insecure",
+		Usage: "disable client transport security for the OTLP exporter",
+		Envs:  []string{"EXPORTER_OTLP_INSECURE"},
+	}
+	flagOTLPExporterEndpoint = flags.FlagDesc{
+		Name:  "exporter-otlp-endpoint",
+		Usage: "the endpoint that exporter connects",
+		Envs:  []string{"EXPORTER_OTLP_ENDPOINT"},
+	}
 )
diff --git a/cmd/varlogsn/varlogsn.go b/cmd/varlogsn/varlogsn.go
index 78e98fc70..bbab5640f 100644
--- a/cmd/varlogsn/varlogsn.go
+++ b/cmd/varlogsn/varlogsn.go
@@ -277,6 +277,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++ {