Implement retention enforcement (#661)

Co-authored-by: Copilot <[email protected]>

Author: benbjohnson

cmd/litestream/main.go

@@ -223,9 +223,8 @@ func (c *Config) CompactionLevels() litestream.CompactionLevels {
 
 	for i, lvl := range c.Levels {
 		levels = append(levels, &litestream.CompactionLevel{
-			Level:     i + 1,
-			Interval:  lvl.Interval,
-			Retention: lvl.Retention,
+			Level:    i + 1,
+			Interval: lvl.Interval,
 		})
 	}
 
@@ -397,14 +396,12 @@ func NewDBFromConfig(dbc *DBConfig) (*litestream.DB, error) {
 
 // ReplicaConfig represents the configuration for a single replica in a database.
 type ReplicaConfig struct {
-	Type                   string         `yaml:"type"` // "file", "s3"
-	Name                   string         `yaml:"name"` // Deprecated
-	Path                   string         `yaml:"path"`
-	URL                    string         `yaml:"url"`
-	Retention              *time.Duration `yaml:"retention"`
-	RetentionCheckInterval *time.Duration `yaml:"retention-check-interval"`
-	SyncInterval           *time.Duration `yaml:"sync-interval"`
-	ValidationInterval     *time.Duration `yaml:"validation-interval"`
+	Type               string         `yaml:"type"` // "file", "s3"
+	Name               string         `yaml:"name"` // Deprecated
+	Path               string         `yaml:"path"`
+	URL                string         `yaml:"url"`
+	SyncInterval       *time.Duration `yaml:"sync-interval"`
+	ValidationInterval *time.Duration `yaml:"validation-interval"`
 
 	// S3 settings
 	AccessKeyID     string `yaml:"access-key-id"`
@@ -441,18 +438,9 @@ func NewReplicaFromConfig(c *ReplicaConfig, db *litestream.DB) (_ *litestream.Re
 
 	// Build replica.
 	r := litestream.NewReplica(db)
-	if v := c.Retention; v != nil {
-		r.Retention = *v
-	}
-	if v := c.RetentionCheckInterval; v != nil {
-		r.RetentionCheckInterval = *v
-	}
 	if v := c.SyncInterval; v != nil {
 		r.SyncInterval = *v
 	}
-	if v := c.ValidationInterval; v != nil {
-		r.ValidationInterval = *v
-	}
 	for _, str := range c.Age.Identities {
 		identities, err := age.ParseIdentities(strings.NewReader(str))
 		if err != nil {

compaction_level.go

@@ -16,9 +16,6 @@ type CompactionLevel struct {
 
 	// The frequency that the level is compacted from the previous level.
 	Interval time.Duration
-
-	// The duration that files in this level are stored.
-	Retention time.Duration
 }
 
 // PrevCompactionAt returns the time when the last compaction occurred.

db.go

@@ -1263,6 +1263,13 @@ 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 dstLevel == 1 {
+		if err := db.EnforceRetentionByTXID(ctx, 0, maxTXID); err != nil {
+			db.Logger.Error("enforce L0 retention", "error", err)
+		}
+	}
+
 	return info, nil
 }
 
@@ -1284,6 +1291,91 @@ func (db *DB) Snapshot(ctx context.Context) (*ltx.FileInfo, error) {
 	return info, nil
 }
 
+// EnforceSnapshotRetention enforces retention of the snapshot level in the database by timestamp.
+func (db *DB) EnforceSnapshotRetention(ctx context.Context, timestamp time.Time) (minSnapshotTXID ltx.TXID, err error) {
+	db.Logger.Debug("enforcing snapshot retention", "timestamp", timestamp)
+
+	itr, err := db.Replica.Client.LTXFiles(ctx, SnapshotLevel, 0)
+	if err != nil {
+		return 0, fmt.Errorf("fetch ltx files: %w", err)
+	}
+	defer itr.Close()
+
+	var deleted []*ltx.FileInfo
+	var lastInfo *ltx.FileInfo
+	for itr.Next() {
+		info := itr.Item()
+		lastInfo = info
+
+		// If this snapshot is before the retention timestamp, mark it for deletion.
+		if info.CreatedAt.Before(timestamp) {
+			deleted = append(deleted, info)
+			continue
+		}
+
+		// Track the lowest snapshot TXID so we can enforce retention in lower levels.
+		// This is only tracked for snapshots not marked for deletion.
+		if minSnapshotTXID == 0 || info.MaxTXID < minSnapshotTXID {
+			minSnapshotTXID = info.MaxTXID
+		}
+	}
+
+	// If this is the snapshot level, we need to ensure that at least one snapshot exists.
+	if len(deleted) > 0 && deleted[len(deleted)-1] == lastInfo {
+		deleted = deleted[:len(deleted)-1]
+	}
+
+	// Remove all files marked for deletion.
+	for _, info := range deleted {
+		db.Logger.Info("deleting ltx file", "level", SnapshotLevel, "minTXID", info.MinTXID, "maxTXID", info.MaxTXID)
+	}
+	if err := db.Replica.Client.DeleteLTXFiles(ctx, deleted); err != nil {
+		return 0, fmt.Errorf("remove ltx files: %w", err)
+	}
+
+	return minSnapshotTXID, 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) {
+	db.Logger.Debug("enforcing retention", "level", level, "txid", txID)
+
+	itr, err := db.Replica.Client.LTXFiles(ctx, level, 0)
+	if err != nil {
+		return fmt.Errorf("fetch ltx files: %w", err)
+	}
+	defer itr.Close()
+
+	var deleted []*ltx.FileInfo
+	var lastInfo *ltx.FileInfo
+	for itr.Next() {
+		info := itr.Item()
+		lastInfo = info
+
+		// If this file's maxTXID is below the target TXID, mark it for deletion.
+		if info.MaxTXID < txID {
+			deleted = append(deleted, info)
+			continue
+		}
+	}
+
+	// Ensure we don't delete the last file.
+	if len(deleted) > 0 && deleted[len(deleted)-1] == lastInfo {
+		deleted = deleted[:len(deleted)-1]
+	}
+
+	// Remove all files marked for deletion.
+	for _, info := range deleted {
+		db.Logger.Info("deleting ltx file", "level", level, "minTXID", info.MinTXID, "maxTXID", info.MaxTXID)
+	}
+	if err := db.Replica.Client.DeleteLTXFiles(ctx, deleted); err != nil {
+		return fmt.Errorf("remove ltx files: %w", err)
+	}
+
+	return nil
+}
+
 // monitor runs in a separate goroutine and monitors the database & WAL.
 func (db *DB) monitor() {
 	ticker := time.NewTicker(db.MonitorInterval)

db_test.go

@@ -497,6 +497,80 @@ func TestDB_Snapshot(t *testing.T) {
 	}
 }
 
+func TestDB_EnforceRetention(t *testing.T) {
+	db, sqldb := MustOpenDBs(t)
+	defer MustCloseDBs(t, db, sqldb)
+	db.Replica = litestream.NewReplica(db)
+	db.Replica.Client = NewFileReplicaClient(t)
+
+	// Create table and sync initial state
+	if _, err := sqldb.Exec(`CREATE TABLE t (id INT);`); err != nil {
+		t.Fatal(err)
+	} else if err := db.Sync(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+
+	// Create multiple snapshots with delays to test retention
+	for i := 0; i < 3; i++ {
+		if _, err := sqldb.Exec(`INSERT INTO t (id) VALUES (?)`, i); err != nil {
+			t.Fatal(err)
+		} else if err := db.Sync(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+
+		if _, err := db.Snapshot(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+
+		// Sleep between snapshots to create time differences
+		time.Sleep(100 * time.Millisecond)
+	}
+
+	// Get list of snapshots before retention
+	itr, err := db.Replica.Client.LTXFiles(context.Background(), litestream.SnapshotLevel, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var beforeCount int
+	for itr.Next() {
+		beforeCount++
+	}
+	itr.Close()
+
+	if beforeCount != 3 {
+		t.Fatalf("expected 3 snapshots before retention, got %d", beforeCount)
+	}
+
+	// Enforce retention to remove older snapshots
+	retentionTime := time.Now().Add(-150 * time.Millisecond)
+	if minSnapshotTXID, err := db.EnforceSnapshotRetention(context.Background(), retentionTime); err != nil {
+		t.Fatal(err)
+	} else if got, want := minSnapshotTXID, ltx.TXID(4); got != want {
+		t.Fatalf("MinSnapshotTXID=%s, want %s", got, want)
+	}
+
+	// Verify snapshots after retention
+	itr, err = db.Replica.Client.LTXFiles(context.Background(), litestream.SnapshotLevel, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var afterCount int
+	for itr.Next() {
+		afterCount++
+	}
+	itr.Close()
+
+	// Should have at least one snapshot remaining
+	if afterCount < 1 {
+		t.Fatal("expected at least 1 snapshot after retention")
+	}
+
+	// Should have fewer snapshots than before
+	if afterCount >= beforeCount {
+		t.Fatalf("expected fewer snapshots after retention, before=%d after=%d", beforeCount, afterCount)
+	}
+}
+
 func newDB(tb testing.TB, path string) *litestream.DB {
 	tb.Helper()
 	tb.Logf("db=%s", path)

go.mod

@@ -15,7 +15,7 @@ require (
 	github.com/mattn/go-sqlite3 v1.14.19
 	github.com/pkg/sftp v1.13.6
 	github.com/prometheus/client_golang v1.17.0
-	github.com/superfly/ltx v0.3.17
+	github.com/superfly/ltx v0.3.18
 	golang.org/x/crypto v0.17.0
 	golang.org/x/sync v0.5.0
 	golang.org/x/sys v0.15.0

go.sum

@@ -166,8 +166,8 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
-github.com/superfly/ltx v0.3.17 h1:7cFPsrPYxgfGlqyhyKQtE9tx6mLZmcBUOrCUx1J6ju8=
-github.com/superfly/ltx v0.3.17/go.mod h1:Nf50QAIXU/ET4ua3AuQ2fh31MbgNQZA7r/DYx6Os77s=
+github.com/superfly/ltx v0.3.18 h1:sAIww45DNoTvFD1fRfrhDTFKnKGca5/hGS4esSIFnfM=
+github.com/superfly/ltx v0.3.18/go.mod h1:Nf50QAIXU/ET4ua3AuQ2fh31MbgNQZA7r/DYx6Os77s=
 github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4=
 github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=