Repository: benbjohnson/litestream

Issue:
Implement Time-Based L0 Retention for VFS Support
## Problem Statement

Currently, Litestream uses TXID-based retention for L0 files, deleting them immediately after they're compacted into L1 files. This creates a race condition for VFS operations where:

1. L0 files are deleted as soon as they're compacted into L1
2. VFS may not have sufficient time to fetch the new L1 file before the L0 files disappear
3. This can cause VFS operations to fail when trying to access recently compacted L0 files

## Current Behavior

**Location**: `db.go:1474-1481`, `db.go:1557-1602`

- L0 files are deleted via `EnforceRetentionByTXID(ctx, 0, minTXID)` after L1 compaction
- Deletion is purely TXID-based: any L0 file with `MaxTXID < minTXID` of the new L1 is deleted
- No time consideration - files could be deleted seconds after creation if compaction happens quickly
- VFS relies on `CalcRestorePlan()` which expects L0 files to be available for point-in-time restore

## Proposed Solution

Implement time-based retention for L0 files with dual criteria:

**L0 files should only be deleted when BOTH conditions are met:**
1. The file has been successfully compacted into an L1 file (MaxTXID < minL1TXID)
2. At least 5 minutes have elapsed since the file was created (configurable)

## Implementation Details

### 1. Configuration Changes

**File**: `store.go`
```go
// Add new constant
const DefaultL0Retention = 5 * time.Minute

// Add to Store struct
type Store struct {
    // ... existing fields
    L0Retention time.Duration // Time to retain L0 files after compaction
}
```

### 2. New Retention Function

**File**: `db.go`
```go
// New function similar to EnforceSnapshotRetention
func (db *DB) EnforceL0RetentionByTime(ctx context.Context) error {
    // 1. Get minimum TXID from L1 files
    // 2. Get current time - L0Retention duration
    // 3. Delete L0 files that meet BOTH criteria:
    //    - MaxTXID < minL1TXID (already compacted)
    //    - CreatedAt < retentionThreshold
}
```

### 3. Update Compaction Flow

**File**: `db.go:1474-1481`

Replace:
```go
if err := db.EnforceRetentionByTXID(ctx, 0, minTXID); err != nil {
    return nil, fmt.Errorf("cannot enforce ltx retention: %w", err)
}
```

With:
```go
if err := db.EnforceL0RetentionByTime(ctx); err != nil {
    return nil, fmt.Errorf("cannot enforce L0 time-based retention: %w", err)
}
```

### 4. Add Periodic Cleanup

**File**: `store.go`

Add L0 cleanup to the retention check interval (similar to snapshot retention):
- Run more frequently than L1 compaction interval
- Ensure VFS has adequate time to fetch L1 files

### 5. Testing Requirements

- Test that L0 files persist for at least 5 minutes after creation
- Verify L0 files are deleted after both conditions are met
- Test VFS operations during L0→L1 transition period
- Ensure restore operations work correctly with time-based retention
- Test with various database sizes and change frequencies

## Benefits

1. **VFS Stability**: Provides VFS sufficient time to fetch new L1 files before L0 files disappear
2. **Predictable Behavior**: Time-based retention is more intuitive and predictable
3. **Configurable**: Allows tuning retention period based on deployment needs
4. **Backward Compatible**: Default behavior can be maintained with configuration

## Risks and Mitigation

**Risk**: Increased storage usage from retained L0 files
- **Mitigation**: 5-minute default is minimal; make configurable for different use cases

**Risk**: Restore operations may need adjustment
- **Mitigation**: Ensure `CalcRestorePlan()` handles missing L0 files gracefully when L1 exists

## Migration Notes

- Existing deployments will need configuration update if different retention is desired
- Default 5-minute retention should work for most use cases
- Monitor disk usage after deployment to ensure L0 files aren't accumulating excessively

## Files Affected

- `db.go`: Core retention logic
- `store.go`: Configuration and constants
- `config.go`: Configuration parsing
- `replica.go`: May need updates to `CalcRestorePlan()`
- `vfs.go`: Testing to ensure compatibility
- Tests: New tests for time-based retention

## Acceptance Criteria

- [ ] L0 files are retained for at least 5 minutes after creation
- [ ] L0 files are only deleted after successful L1 compaction
- [ ] Retention period is configurable
- [ ] VFS operations work reliably during L0→L1 transitions  
- [ ] All existing tests pass
- [ ] New tests cover time-based retention scenarios
- [ ] Documentation updated with new configuration option

## References

- Current L0 deletion: `db.go:1557-1602` (`EnforceRetentionByTXID`)
- Compaction trigger: `db.go:1474-1481`
- Snapshot time-based retention (reference implementation): `db.go:1505-1555`
- VFS L0 dependency: `vfs.go:270-290` (`monitorReplicaClient`)

Discussion:

Solution patch:
diff --git a/AGENTS.md b/AGENTS.md
index 7e66ecd..7c39d80 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -769,6 +769,8 @@ go test -race -v -run TestStore_CompactDB ./...
 ### Key Configuration
 
 ```yaml
+l0-retention: 5m                 # Minimum time to keep compacted L0 files
+l0-retention-check-interval: 15s # Frequency for enforcing L0 retention
 dbs:
   - path: /path/to/db.sqlite
     replicas:
diff --git a/db.go b/db.go
index 7e8cb7c..fe741d3 100644
--- a/db.go
+++ b/db.go
@@ -106,6 +106,9 @@ type DB struct {
 	// The timeout to wait for EBUSY from SQLite.
 	BusyTimeout time.Duration
 
+	// Minimum time to retain L0 files after they have been compacted into L1.
+	L0Retention time.Duration
+
 	// Remote replica for the database.
 	// Must be set before calling Open().
 	Replica *Replica
@@ -129,6 +132,7 @@ func NewDB(path string) *DB {
 		CheckpointInterval: DefaultCheckpointInterval,
 		MonitorInterval:    DefaultMonitorInterval,
 		BusyTimeout:        DefaultBusyTimeout,
+		L0Retention:        DefaultL0Retention,
 		Logger:             slog.With("db", filepath.Base(path)),
 	}
 	db.maxLTXFileInfos.m = make(map[int]*ltx.FileInfo)
@@ -1471,12 +1475,12 @@ func (db *DB) Compact(ctx context.Context, dstLevel int) (*ltx.FileInfo, error)
 	db.maxLTXFileInfos.m[dstLevel] = info
 	db.maxLTXFileInfos.Unlock()
 
-	// If this is L1, clean up L0 files that are below the minTXID.
+	// If this is L1, clean up L0 files using the time-based retention policy.
 	if dstLevel == 1 {
-		if err := db.EnforceRetentionByTXID(ctx, 0, minTXID); err != nil {
+		if err := db.EnforceL0RetentionByTime(ctx); err != nil {
 			// Don't log context cancellation errors during shutdown
 			if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
-				db.Logger.Error("enforce L0 retention", "error", err)
+				db.Logger.Error("enforce L0 time retention", "error", err)
 			}
 		}
 	}
@@ -1554,6 +1558,95 @@ func (db *DB) EnforceSnapshotRetention(ctx context.Context, timestamp time.Time)
 	return minSnapshotTXID, nil
 }
 
+// EnforceL0RetentionByTime retains L0 files until they have been compacted into
+// L1 and have existed for at least L0Retention.
+func (db *DB) EnforceL0RetentionByTime(ctx context.Context) error {
+	if db.L0Retention <= 0 {
+		return nil
+	}
+
+	db.Logger.Debug("enforcing l0 retention", "retention", db.L0Retention)
+
+	// Determine the highest TXID that has been compacted into L1.
+	itr, err := db.Replica.Client.LTXFiles(ctx, 1, 0, false)
+	if err != nil {
+		return fmt.Errorf("fetch l1 files: %w", err)
+	}
+	var maxL1TXID ltx.TXID
+	for itr.Next() {
+		info := itr.Item()
+		if info.MaxTXID > maxL1TXID {
+			maxL1TXID = info.MaxTXID
+		}
+	}
+	if err := itr.Close(); err != nil {
+		return fmt.Errorf("close l1 iterator: %w", err)
+	}
+	if maxL1TXID == 0 {
+		return nil
+	}
+
+	threshold := time.Now().Add(-db.L0Retention)
+	itr, err = db.Replica.Client.LTXFiles(ctx, 0, 0, false)
+	if err != nil {
+		return fmt.Errorf("fetch l0 files: %w", err)
+	}
+	defer itr.Close()
+
+	var (
+		deleted      []*ltx.FileInfo
+		lastInfo     *ltx.FileInfo
+		processedAll = true
+	)
+	for itr.Next() {
+		info := itr.Item()
+		lastInfo = info
+
+		createdAt := info.CreatedAt
+		if createdAt.IsZero() {
+			if fi, err := os.Stat(db.LTXPath(0, info.MinTXID, info.MaxTXID)); err == nil {
+				createdAt = fi.ModTime().UTC()
+			} else {
+				createdAt = threshold
+			}
+		}
+
+		if createdAt.After(threshold) {
+			// L0 entries are ordered; once we reach a newer file we stop so we don't
+			// create gaps between retained files. VFS expects contiguous coverage.
+			processedAll = false
+			break
+		}
+
+		if info.MaxTXID <= maxL1TXID {
+			deleted = append(deleted, info)
+		}
+	}
+
+	// Ensure we do not delete the newest L0 file only if we processed the entire level.
+	if processedAll && len(deleted) > 0 && lastInfo != nil && deleted[len(deleted)-1] == lastInfo {
+		deleted = deleted[:len(deleted)-1]
+	}
+
+	if len(deleted) == 0 {
+		return nil
+	}
+
+	if err := db.Replica.Client.DeleteLTXFiles(ctx, deleted); err != nil {
+		return fmt.Errorf("remove expired l0 files: %w", err)
+	}
+
+	for _, info := range deleted {
+		localPath := db.LTXPath(0, info.MinTXID, info.MaxTXID)
+		db.Logger.Debug("deleting expired local l0 file", "minTXID", info.MinTXID, "maxTXID", info.MaxTXID, "path", localPath)
+		if err := os.Remove(localPath); err != nil && !os.IsNotExist(err) {
+			db.Logger.Error("failed to remove local l0 file", "path", localPath, "error", err)
+		}
+	}
+
+	return nil
+}
+
 // EnforceRetentionByTXID enforces retention so that any LTX files below
 // the target TXID are deleted. Always keep at least one file.
 func (db *DB) EnforceRetentionByTXID(ctx context.Context, level int, txID ltx.TXID) (err error) {
diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md
index b270046..05fcfe5 100644
--- a/docs/ARCHITECTURE.md
+++ b/docs/ARCHITECTURE.md
@@ -190,8 +190,10 @@ type Store struct {
     levels CompactionLevels
 
     // Configuration
-    SnapshotInterval  time.Duration
-    SnapshotRetention time.Duration
+    SnapshotInterval           time.Duration
+    SnapshotRetention          time.Duration
+    L0Retention                time.Duration
+    L0RetentionCheckInterval   time.Duration
     CompactionMonitorEnabled bool
 
     // Lifecycle
diff --git a/store.go b/store.go
index f41b096..e6b6926 100644
--- a/store.go
+++ b/store.go
@@ -32,6 +32,14 @@ const (
 
 	DefaultRetention              = 24 * time.Hour
 	DefaultRetentionCheckInterval = 1 * time.Hour
+
+	// DefaultL0Retention is the default time that L0 files are kept around
+	// after they have been compacted into L1 files.
+	DefaultL0Retention = 5 * time.Minute
+	// DefaultL0RetentionCheckInterval controls how frequently L0 retention is
+	// enforced. This interval should be more frequent than the L1 compaction
+	// interval so that VFS read replicas have time to observe new files.
+	DefaultL0RetentionCheckInterval = 15 * time.Second
 )
 
 // Store represents the top-level container for databases.
@@ -52,6 +60,11 @@ type Store struct {
 	// The duration of time that snapshots are kept before being deleted.
 	SnapshotRetention time.Duration
 
+	// The duration that L0 files are kept after being compacted into L1.
+	L0Retention time.Duration
+	// How often to check for expired L0 files.
+	L0RetentionCheckInterval time.Duration
+
 	// If true, compaction is run in the background according to compaction levels.
 	CompactionMonitorEnabled bool
 }
@@ -63,8 +76,14 @@ func NewStore(dbs []*DB, levels CompactionLevels) *Store {
 
 		SnapshotInterval:         DefaultSnapshotInterval,
 		SnapshotRetention:        DefaultSnapshotRetention,
+		L0Retention:              DefaultL0Retention,
+		L0RetentionCheckInterval: DefaultL0RetentionCheckInterval,
 		CompactionMonitorEnabled: true,
 	}
+
+	for _, db := range dbs {
+		db.L0Retention = s.L0Retention
+	}
 	s.ctx, s.cancel = context.WithCancel(context.Background())
 	return s
 }
@@ -104,6 +123,14 @@ func (s *Store) Open(ctx context.Context) error {
 		}()
 	}
 
+	if s.L0Retention > 0 && s.L0RetentionCheckInterval > 0 {
+		s.wg.Add(1)
+		go func() {
+			defer s.wg.Done()
+			s.monitorL0Retention(s.ctx)
+		}()
+	}
+
 	return nil
 }
 
@@ -127,6 +154,17 @@ func (s *Store) DBs() []*DB {
 	return slices.Clone(s.dbs)
 }
 
+// SetL0Retention updates the retention window for L0 files and propagates it to
+// all managed databases.
+func (s *Store) SetL0Retention(d time.Duration) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.L0Retention = d
+	for _, db := range s.dbs {
+		db.L0Retention = d
+	}
+}
+
 // SnapshotLevel returns a pseudo compaction level based on snapshot settings.
 func (s *Store) SnapshotLevel() *CompactionLevel {
 	return &CompactionLevel{
@@ -184,6 +222,31 @@ LOOP:
 	}
 }
 
+func (s *Store) monitorL0Retention(ctx context.Context) {
+	slog.Info("starting L0 retention monitor", "interval", s.L0RetentionCheckInterval, "retention", s.L0Retention)
+
+	ticker := time.NewTicker(s.L0RetentionCheckInterval)
+	defer ticker.Stop()
+
+LOOP:
+	for {
+		select {
+		case <-ctx.Done():
+			break LOOP
+		case <-ticker.C:
+		}
+
+		for _, db := range s.DBs() {
+			if err := db.EnforceL0RetentionByTime(ctx); err != nil {
+				if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+					continue
+				}
+				slog.Error("l0 retention enforcement failed", "path", db.Path(), "error", err)
+			}
+		}
+	}
+}
+
 // CompactDB performs a compaction or snapshot for a given database on a single destination level.
 // This function will only proceed if a compaction has not occurred before the last compaction time.
 func (s *Store) CompactDB(ctx context.Context, db *DB, lvl *CompactionLevel) (*ltx.FileInfo, error) {



Test patch:
diff --git a/cmd/litestream/main.go b/cmd/litestream/main.go
index cd8a764..69dbcb1 100644
--- a/cmd/litestream/main.go
+++ b/cmd/litestream/main.go
@@ -44,11 +44,13 @@ var errStop = errors.New("stop")
 
 // Sentinel errors for configuration validation
 var (
-	ErrInvalidSnapshotInterval   = errors.New("snapshot interval must be greater than 0")
-	ErrInvalidSnapshotRetention  = errors.New("snapshot retention must be greater than 0")
-	ErrInvalidCompactionInterval = errors.New("compaction interval must be greater than 0")
-	ErrInvalidSyncInterval       = errors.New("sync interval must be greater than 0")
-	ErrConfigFileNotFound        = errors.New("config file not found")
+	ErrInvalidSnapshotInterval         = errors.New("snapshot interval must be greater than 0")
+	ErrInvalidSnapshotRetention        = errors.New("snapshot retention must be greater than 0")
+	ErrInvalidCompactionInterval       = errors.New("compaction interval must be greater than 0")
+	ErrInvalidSyncInterval             = errors.New("sync interval must be greater than 0")
+	ErrInvalidL0Retention              = errors.New("l0 retention must be greater than 0")
+	ErrInvalidL0RetentionCheckInterval = errors.New("l0 retention check interval must be greater than 0")
+	ErrConfigFileNotFound              = errors.New("config file not found")
 )
 
 // ConfigValidationError wraps a validation error with additional context
@@ -198,6 +200,10 @@ type Config struct {
 	// Snapshot configuration
 	Snapshot SnapshotConfig `yaml:"snapshot"`
 
+	// L0 retention settings
+	L0Retention              *time.Duration `yaml:"l0-retention"`
+	L0RetentionCheckInterval *time.Duration `yaml:"l0-retention-check-interval"`
+
 	// List of databases to manage.
 	DBs []*DBConfig `yaml:"dbs"`
 
@@ -251,6 +257,8 @@ func (c *Config) propagateGlobalSettings() {
 func DefaultConfig() Config {
 	defaultSnapshotInterval := 24 * time.Hour
 	defaultSnapshotRetention := 24 * time.Hour
+	defaultL0Retention := litestream.DefaultL0Retention
+	defaultL0RetentionCheckInterval := litestream.DefaultL0RetentionCheckInterval
 	return Config{
 		Levels: []*CompactionLevelConfig{
 			{Interval: 30 * time.Second},
@@ -261,6 +269,8 @@ func DefaultConfig() Config {
 			Interval:  &defaultSnapshotInterval,
 			Retention: &defaultSnapshotRetention,
 		},
+		L0Retention:              &defaultL0Retention,
+		L0RetentionCheckInterval: &defaultL0RetentionCheckInterval,
 	}
 }
 
@@ -281,6 +291,20 @@ func (c *Config) Validate() error {
 			Value: *c.Snapshot.Retention,
 		}
 	}
+	if c.L0Retention != nil && *c.L0Retention <= 0 {
+		return &ConfigValidationError{
+			Err:   ErrInvalidL0Retention,
+			Field: "l0-retention",
+			Value: *c.L0Retention,
+		}
+	}
+	if c.L0RetentionCheckInterval != nil && *c.L0RetentionCheckInterval <= 0 {
+		return &ConfigValidationError{
+			Err:   ErrInvalidL0RetentionCheckInterval,
+			Field: "l0-retention-check-interval",
+			Value: *c.L0RetentionCheckInterval,
+		}
+	}
 
 	// Validate compaction level intervals
 	for i, level := range c.Levels {
@@ -413,6 +437,8 @@ func ParseConfig(r io.Reader, expandEnv bool) (_ Config, err error) {
 	// Save defaults before unmarshaling
 	defaultSnapshotInterval := config.Snapshot.Interval
 	defaultSnapshotRetention := config.Snapshot.Retention
+	defaultL0Retention := config.L0Retention
+	defaultL0RetentionCheckInterval := config.L0RetentionCheckInterval
 
 	if err := yaml.Unmarshal(buf, &config); err != nil {
 		return config, err
@@ -425,6 +451,12 @@ func ParseConfig(r io.Reader, expandEnv bool) (_ Config, err error) {
 	if config.Snapshot.Retention == nil {
 		config.Snapshot.Retention = defaultSnapshotRetention
 	}
+	if config.L0Retention == nil {
+		config.L0Retention = defaultL0Retention
+	}
+	if config.L0RetentionCheckInterval == nil {
+		config.L0RetentionCheckInterval = defaultL0RetentionCheckInterval
+	}
 
 	// Normalize paths.
 	for _, dbConfig := range config.DBs {
diff --git a/cmd/litestream/main_test.go b/cmd/litestream/main_test.go
index 49f90b2..51baf64 100644
--- a/cmd/litestream/main_test.go
+++ b/cmd/litestream/main_test.go
@@ -9,6 +9,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/benbjohnson/litestream"
 	main "github.com/benbjohnson/litestream/cmd/litestream"
 	"github.com/benbjohnson/litestream/file"
 	"github.com/benbjohnson/litestream/gs"
@@ -515,6 +516,67 @@ func TestParseReplicaURL_AccessPoint(t *testing.T) {
 	})
 }
 
+func TestConfig_Validate_L0Retention(t *testing.T) {
+	t.Run("ZeroRetention", func(t *testing.T) {
+		yaml := `
+l0-retention: 0s
+dbs:
+  - path: /tmp/test.db
+    replica:
+      url: file:///tmp/replica
+`
+		_, err := main.ParseConfig(strings.NewReader(yaml), false)
+		if err == nil {
+			t.Fatal("expected error for zero l0 retention")
+		}
+		if !errors.Is(err, main.ErrInvalidL0Retention) {
+			t.Errorf("expected ErrInvalidL0Retention, got %v", err)
+		}
+	})
+
+	t.Run("ZeroCheckInterval", func(t *testing.T) {
+		yaml := `
+l0-retention-check-interval: 0s
+dbs:
+  - path: /tmp/test.db
+    replica:
+      url: file:///tmp/replica
+`
+		_, err := main.ParseConfig(strings.NewReader(yaml), false)
+		if err == nil {
+			t.Fatal("expected error for zero l0 retention check interval")
+		}
+		if !errors.Is(err, main.ErrInvalidL0RetentionCheckInterval) {
+			t.Errorf("expected ErrInvalidL0RetentionCheckInterval, got %v", err)
+		}
+	})
+
+	t.Run("DefaultsApplied", func(t *testing.T) {
+		yaml := `
+dbs:
+  - path: /tmp/test.db
+    replica:
+      url: file:///tmp/replica
+`
+		config, err := main.ParseConfig(strings.NewReader(yaml), false)
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		if config.L0Retention == nil {
+			t.Fatal("expected default l0 retention to be set")
+		}
+		if *config.L0Retention != litestream.DefaultL0Retention {
+			t.Errorf("expected default l0 retention %v, got %v", litestream.DefaultL0Retention, *config.L0Retention)
+		}
+		if config.L0RetentionCheckInterval == nil {
+			t.Fatal("expected default l0 retention check interval to be set")
+		}
+		if *config.L0RetentionCheckInterval != litestream.DefaultL0RetentionCheckInterval {
+			t.Errorf("expected default l0 retention check interval %v, got %v", litestream.DefaultL0RetentionCheckInterval, *config.L0RetentionCheckInterval)
+		}
+	})
+}
+
 // TestConfig_Validate_SyncIntervals tests validation of replica sync intervals
 func TestConfig_Validate_SyncIntervals(t *testing.T) {
 	t.Run("ValidSyncInterval", func(t *testing.T) {
diff --git a/cmd/litestream/replicate.go b/cmd/litestream/replicate.go
index 4f7c33a..f3d91eb 100644
--- a/cmd/litestream/replicate.go
+++ b/cmd/litestream/replicate.go
@@ -153,6 +153,12 @@ func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
 	if c.Config.Snapshot.Retention != nil {
 		c.Store.SnapshotRetention = *c.Config.Snapshot.Retention
 	}
+	if c.Config.L0Retention != nil {
+		c.Store.SetL0Retention(*c.Config.L0Retention)
+	}
+	if c.Config.L0RetentionCheckInterval != nil {
+		c.Store.L0RetentionCheckInterval = *c.Config.L0RetentionCheckInterval
+	}
 	if err := c.Store.Open(ctx); err != nil {
 		return fmt.Errorf("cannot open store: %w", err)
 	}
diff --git a/db_test.go b/db_test.go
index b943f70..85f0a98 100644
--- a/db_test.go
+++ b/db_test.go
@@ -831,6 +831,10 @@ func TestDB_EnforceRetentionByTXID_LocalCleanup(t *testing.T) {
 		t.Fatalf("compact batch2 to L1: %v", err)
 	}
 
+	if err := db.EnforceRetentionByTXID(ctx, 0, secondCompactInfo.MinTXID); err != nil {
+		t.Fatalf("enforce retention: %v", err)
+	}
+
 	for _, lf := range firstBatchL0Files {
 		if lf.maxTXID < secondCompactInfo.MinTXID {
 			if _, err := os.Stat(lf.path); err == nil {
@@ -842,3 +846,104 @@ func TestDB_EnforceRetentionByTXID_LocalCleanup(t *testing.T) {
 		}
 	}
 }
+
+func TestDB_EnforceL0RetentionByTime(t *testing.T) {
+	ctx := context.Background()
+
+	db, sqldb := testingutil.MustOpenDBs(t)
+	defer testingutil.MustCloseDBs(t, db, sqldb)
+
+	replicaPath := filepath.Join(t.TempDir(), "replica")
+	client := file.NewReplicaClient(replicaPath)
+	db.Replica = litestream.NewReplicaWithClient(db, client)
+	db.Replica.MonitorEnabled = false
+
+	// Use a long retention initially so compaction does not immediately clean up files.
+	db.L0Retention = 30 * time.Minute
+
+	if _, err := sqldb.ExecContext(ctx, `CREATE TABLE t (id INTEGER PRIMARY KEY, val TEXT)`); err != nil {
+		t.Fatalf("create table: %v", err)
+	}
+
+	for i := 0; i < 3; i++ {
+		if _, err := sqldb.ExecContext(ctx, `INSERT INTO t (val) VALUES (?)`, fmt.Sprintf("value-%d", i)); err != nil {
+			t.Fatalf("insert %d: %v", i, err)
+		}
+		if err := db.Sync(ctx); err != nil {
+			t.Fatalf("sync db %d: %v", i, err)
+		}
+		if err := db.Replica.Sync(ctx); err != nil {
+			t.Fatalf("sync replica %d: %v", i, err)
+		}
+	}
+
+	if _, err := db.Compact(ctx, 1); err != nil {
+		t.Fatalf("compact L0 -> L1: %v", err)
+	}
+
+	itr, err := client.LTXFiles(ctx, 0, 0, false)
+	if err != nil {
+		t.Fatalf("list L0 files: %v", err)
+	}
+	l0Files, err := ltx.SliceFileIterator(itr)
+	if err != nil {
+		t.Fatalf("slice iterator: %v", err)
+	}
+	if err := itr.Close(); err != nil {
+		t.Fatalf("close iterator: %v", err)
+	}
+	if len(l0Files) < 2 {
+		t.Fatalf("expected at least two L0 files, got %d", len(l0Files))
+	}
+
+	checkExists := func(expectMissing bool) {
+		for idx, info := range l0Files {
+			remotePath := client.LTXFilePath(0, info.MinTXID, info.MaxTXID)
+			localPath := db.LTXPath(0, info.MinTXID, info.MaxTXID)
+			_, remoteErr := os.Stat(remotePath)
+			_, localErr := os.Stat(localPath)
+			if expectMissing && idx < len(l0Files)-1 {
+				if !os.IsNotExist(remoteErr) {
+					t.Fatalf("expected remote file removed: %s", remotePath)
+				}
+				if !os.IsNotExist(localErr) {
+					t.Fatalf("expected local file removed: %s", localPath)
+				}
+			}
+			if !expectMissing || idx == len(l0Files)-1 {
+				if remoteErr != nil {
+					t.Fatalf("expected remote file to exist: %s (%v)", remotePath, remoteErr)
+				}
+				if localErr != nil {
+					t.Fatalf("expected local file to exist: %s (%v)", localPath, localErr)
+				}
+			}
+		}
+	}
+
+	// Files should still exist immediately after compaction since they are new.
+	if err := db.EnforceL0RetentionByTime(ctx); err != nil {
+		t.Fatalf("enforce recent retention: %v", err)
+	}
+	checkExists(false)
+
+	// Age the files so they exceed the retention threshold.
+	oldTime := time.Now().Add(-1 * time.Hour)
+	for _, info := range l0Files {
+		remotePath := client.LTXFilePath(0, info.MinTXID, info.MaxTXID)
+		if err := os.Chtimes(remotePath, oldTime, oldTime); err != nil {
+			t.Fatalf("chtimes remote: %v", err)
+		}
+		localPath := db.LTXPath(0, info.MinTXID, info.MaxTXID)
+		if err := os.Chtimes(localPath, oldTime, oldTime); err != nil {
+			t.Fatalf("chtimes local: %v", err)
+		}
+	}
+
+	// Shorten retention so aged files qualify for deletion.
+	db.L0Retention = time.Second
+	if err := db.EnforceL0RetentionByTime(ctx); err != nil {
+		t.Fatalf("enforce aged retention: %v", err)
+	}
+	checkExists(true)
+}