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++ {