Repository: benbjohnson/litestream
Pull request #650 (2025-06-20 19:33:02)

Multi-level compaction
This pull request adds a new `Store` type that manages the compactions & snapshots for individual `DB` objects. This has been moved from the `DB` object itself as we want to support a large number of databases for a single Litestream process and we need to be able to limit the number of replica client interactions at a given time.

The multi-level restore will be implemented in a separate PR.


Patch:
diff --git a/abs/replica_client.go b/abs/replica_client.go
index 0353260..feaef16 100644
--- a/abs/replica_client.go
+++ b/abs/replica_client.go
@@ -91,11 +91,11 @@ func (c *ReplicaClient) Init(ctx context.Context) (err error) {
 }
 
 // LTXFiles returns an iterator over all available LTX files.
-func (c *ReplicaClient) LTXFiles(ctx context.Context, level int) (ltx.FileIterator, error) {
+func (c *ReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID) (ltx.FileIterator, error) {
 	if err := c.Init(ctx); err != nil {
 		return nil, err
 	}
-	return newLTXFileIterator(ctx, c, level), nil
+	return newLTXFileIterator(ctx, c, level, seek), nil
 }
 
 // WriteLTXFile writes an LTX file to remote storage.
@@ -206,6 +206,7 @@ func (c *ReplicaClient) DeleteAll(ctx context.Context) error {
 type ltxFileIterator struct {
 	client *ReplicaClient
 	level  int
+	seek   ltx.TXID
 
 	ch     chan ltx.FileInfo
 	g      errgroup.Group
@@ -216,10 +217,11 @@ type ltxFileIterator struct {
 	err  error
 }
 
-func newLTXFileIterator(ctx context.Context, client *ReplicaClient, level int) *ltxFileIterator {
+func newLTXFileIterator(ctx context.Context, client *ReplicaClient, level int, seek ltx.TXID) *ltxFileIterator {
 	itr := &ltxFileIterator{
 		client: client,
 		level:  level,
+		seek:   seek,
 		ch:     make(chan ltx.FileInfo),
 	}
 
@@ -234,12 +236,16 @@ func (itr *ltxFileIterator) fetch() error {
 	defer close(itr.ch)
 
 	dir := litestream.LTXLevelDir(itr.client.Path, itr.level)
+	prefix := dir + "/"
+	if itr.seek != 0 {
+		prefix += itr.seek.String()
+	}
 
 	var marker azblob.Marker
 	for marker.NotDone() {
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
 
-		resp, err := itr.client.containerURL.ListBlobsFlatSegment(itr.ctx, marker, azblob.ListBlobsSegmentOptions{Prefix: dir + "/"})
+		resp, err := itr.client.containerURL.ListBlobsFlatSegment(itr.ctx, marker, azblob.ListBlobsSegmentOptions{Prefix: prefix})
 		if err != nil {
 			return err
 		}
diff --git a/compaction_level.go b/compaction_level.go
new file mode 100644
index 0000000..9f135fb
--- /dev/null
+++ b/compaction_level.go
@@ -0,0 +1,106 @@
+package litestream
+
+import (
+	"fmt"
+	"time"
+)
+
+// SnapshotLevel represents the level which full snapshots are held.
+const SnapshotLevel = 9
+
+// CompactionLevel represents a single part of a multi-level compaction.
+// Each level merges LTX files from the previous level into larger time granularities.
+type CompactionLevel struct {
+	// The numeric level. Must match the index in the list of levels.
+	Level int
+
+	// 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.
+// Returns the current time if it is exactly a multiple of the level interval.
+func (lvl *CompactionLevel) PrevCompactionAt(now time.Time) time.Time {
+	return now.Truncate(lvl.Interval).UTC()
+}
+
+// NextCompactionAt returns the time until the next compaction occurs.
+// Returns the current time if it is exactly a multiple of the level interval.
+func (lvl *CompactionLevel) NextCompactionAt(now time.Time) time.Time {
+	return lvl.PrevCompactionAt(now).Add(lvl.Interval)
+}
+
+// CompactionLevels represents a sorted slice of non-snapshot compaction levels.
+type CompactionLevels []*CompactionLevel
+
+// Level returns the compaction level at the given index.
+// Returns an error if the index is a snapshot level or is out of bounds.
+func (a CompactionLevels) Level(level int) (*CompactionLevel, error) {
+	if level == SnapshotLevel {
+		return nil, fmt.Errorf("invalid argument, snapshot level")
+	}
+	if level < 0 || level > a.MaxLevel() {
+		return nil, fmt.Errorf("level out of bounds: %d", level)
+	}
+	return a[level], nil
+}
+
+// MaxLevel return the highest non-snapshot compaction level.
+func (a CompactionLevels) MaxLevel() int {
+	return len(a) - 1
+}
+
+// Validate returns an error if the levels are invalid.
+func (a CompactionLevels) Validate() error {
+	if len(a) == 0 {
+		return fmt.Errorf("at least one compaction level is required")
+	}
+
+	for i, lvl := range a {
+		if i != lvl.Level {
+			return fmt.Errorf("compaction level number out of order: %d, expected %d", lvl.Level, i)
+		} else if lvl.Level > SnapshotLevel-1 {
+			return fmt.Errorf("compaction level cannot exceed %d", SnapshotLevel-1)
+		}
+
+		if lvl.Level == 0 && lvl.Interval != 0 {
+			return fmt.Errorf("cannot set interval on compaction level zero")
+		}
+
+		if lvl.Level != 0 && lvl.Interval <= 0 {
+			return fmt.Errorf("interval required for level %d", lvl.Level)
+		}
+	}
+	return nil
+}
+
+// IsValidLevel returns true if level is a valid compaction level number.
+func (a CompactionLevels) IsValidLevel(level int) bool {
+	if level == SnapshotLevel {
+		return true
+	}
+	return level >= 0 && level < len(a)
+}
+
+// PrevLevel returns the previous compaction level.
+// Returns -1 if there is no previous level.
+func (a CompactionLevels) PrevLevel(level int) int {
+	if level == SnapshotLevel {
+		return a.MaxLevel()
+	}
+	return level - 1
+}
+
+// NextLevel returns the next compaction level.
+// Returns -1 if there is no next level.
+func (a CompactionLevels) NextLevel(level int) int {
+	if level == SnapshotLevel {
+		return -1
+	} else if level == a.MaxLevel() {
+		return SnapshotLevel
+	}
+	return level + 1
+}
diff --git a/db.go b/db.go
index 729ceea..9fc9fd1 100644
--- a/db.go
+++ b/db.go
@@ -43,7 +43,13 @@ type DB struct {
 	rtx      *sql.Tx       // long running read transaction
 	pageSize int           // page size, in bytes
 	notify   chan struct{} // closes on WAL change
-	chkMu    sync.Mutex    // checkpoint lock
+	chkMu    sync.RWMutex  // checkpoint lock
+
+	// last file info for each level
+	maxLTXFileInfos struct {
+		sync.Mutex
+		m map[int]*ltx.FileInfo
+	}
 
 	fileInfo os.FileInfo // db info cached during init
 	dirInfo  os.FileInfo // parent dir info cached during init
@@ -123,6 +129,7 @@ func NewDB(path string) *DB {
 		BusyTimeout:        DefaultBusyTimeout,
 		Logger:             slog.With("db", filepath.Base(path)),
 	}
+	db.maxLTXFileInfos.m = make(map[int]*ltx.FileInfo)
 
 	db.dbSizeGauge = dbSizeGaugeVec.WithLabelValues(db.path)
 	db.walSizeGauge = walSizeGaugeVec.WithLabelValues(db.path)
@@ -748,8 +755,8 @@ func (db *DB) sync(ctx context.Context, checkpointing bool, info syncInfo) error
 
 	// Prevent internal checkpoints during sync. Ignore if already in a checkpoint.
 	if !checkpointing {
-		db.chkMu.Lock()
-		defer db.chkMu.Unlock()
+		db.chkMu.RLock()
+		defer db.chkMu.RUnlock()
 	}
 
 	fi, err := db.f.Stat()
@@ -873,7 +880,7 @@ func (db *DB) sync(ctx context.Context, checkpointing bool, info syncInfo) error
 	return nil
 }
 
-func (db *DB) writeLTXFromDB(_ context.Context, enc *ltx.Encoder, walFile *os.File, commit uint32, pageMap map[uint32]int64) error {
+func (db *DB) writeLTXFromDB(ctx context.Context, enc *ltx.Encoder, walFile *os.File, commit uint32, pageMap map[uint32]int64) error {
 	lockPgno := ltx.LockPgno(uint32(db.pageSize))
 	data := make([]byte, db.pageSize)
 
@@ -882,9 +889,16 @@ func (db *DB) writeLTXFromDB(_ context.Context, enc *ltx.Encoder, walFile *os.Fi
 			continue
 		}
 
+		// Check if the caller has canceled during processing.
+		select {
+		case <-ctx.Done():
+			return context.Cause(ctx)
+		default:
+		}
+
 		// If page exists in the WAL, read from there.
 		if offset, ok := pageMap[pgno]; ok {
-			db.Logger.Debug("encode page from wal", "offset", offset, "pgno", pgno)
+			db.Logger.Debug("encode page from wal", "txid", enc.Header().MinTXID, "offset", offset, "pgno", pgno)
 
 			if n, err := walFile.ReadAt(data, offset+WALFrameHeaderSize); err != nil {
 				return fmt.Errorf("read page %d @ %d: %w", pgno, offset, err)
@@ -925,7 +939,7 @@ func (db *DB) writeLTXFromWAL(_ context.Context, enc *ltx.Encoder, walFile *os.F
 	for _, pgno := range pgnos {
 		offset := pageMap[pgno]
 
-		db.Logger.Debug("encode page from wal", "offset", offset, "pgno", pgno)
+		db.Logger.Debug("encode page from wal", "txid", enc.Header().MinTXID, "offset", offset, "pgno", pgno)
 
 		// Read source page using page map.
 		if n, err := walFile.ReadAt(data, offset+WALFrameHeaderSize); err != nil {
@@ -1057,6 +1071,183 @@ func (db *DB) execCheckpoint(mode string) (err error) {
 	return nil
 }
 
+// SnapshotReader returns the current position of the database & a reader that contains a full database snapshot.
+func (db *DB) SnapshotReader(ctx context.Context) (ltx.Pos, io.Reader, error) {
+	pos, err := db.Pos()
+	if err != nil {
+		return pos, nil, fmt.Errorf("pos: %w", err)
+	}
+
+	db.Logger.Debug("snapshot", "txid", pos.TXID.String())
+
+	// Prevent internal checkpoints during sync.
+	db.chkMu.RLock()
+	defer db.chkMu.RUnlock()
+
+	// TODO(ltx): Read database size from database header.
+
+	fi, err := db.f.Stat()
+	if err != nil {
+		return pos, nil, err
+	}
+	commit := uint32(fi.Size() / int64(db.pageSize))
+
+	// Execute encoding in a separate goroutine so the caller can initialize before reading.
+	pr, pw := io.Pipe()
+	go func() {
+		walFile, err := os.Open(db.WALPath())
+		if err != nil {
+			pw.CloseWithError(err)
+			return
+		}
+		defer walFile.Close()
+
+		rd, err := NewWALReader(walFile, db.Logger)
+		if err != nil {
+			pw.CloseWithError(fmt.Errorf("new wal reader: %w", err))
+			return
+		}
+
+		// Build a mapping of changed page numbers and their latest content.
+		pageMap, sz, walCommit, err := rd.PageMap()
+		if err != nil {
+			pw.CloseWithError(fmt.Errorf("page map: %w", err))
+			return
+		}
+		if walCommit > 0 {
+			commit = walCommit
+		}
+
+		db.Logger.Debug("encode snapshot header",
+			"txid", pos.TXID.String(),
+			"commit", commit,
+			"walOffset", rd.Offset(),
+			"walSize", sz,
+			"salt1", rd.salt1,
+			"salt2", rd.salt2)
+
+		enc := ltx.NewEncoder(pw)
+		if err := enc.EncodeHeader(ltx.Header{
+			Version:   ltx.Version,
+			Flags:     ltx.HeaderFlagNoChecksum | ltx.HeaderFlagCompressLZ4,
+			PageSize:  uint32(db.pageSize),
+			Commit:    commit,
+			MinTXID:   1,
+			MaxTXID:   pos.TXID,
+			Timestamp: time.Now().UnixMilli(),
+			WALOffset: rd.Offset(),
+			WALSize:   sz,
+			WALSalt1:  rd.salt1,
+			WALSalt2:  rd.salt2,
+		}); err != nil {
+			pw.CloseWithError(fmt.Errorf("encode ltx snapshot header: %w", err))
+			return
+		}
+
+		if err := db.writeLTXFromDB(ctx, enc, walFile, commit, pageMap); err != nil {
+			pw.CloseWithError(fmt.Errorf("write snapshot ltx: %w", err))
+			return
+		}
+
+		if err := enc.Close(); err != nil {
+			pw.CloseWithError(fmt.Errorf("close ltx snapshot encoder: %w", err))
+			return
+		}
+		_ = pw.Close()
+	}()
+
+	return pos, pr, nil
+}
+
+// Compact performs a compaction of the LTX file at the previous level into dstLevel.
+// Returns metadata for the newly written compaction file. Returns ErrNoCompaction
+// if no new files are available to be compacted.
+func (db *DB) Compact(ctx context.Context, dstLevel int) (*ltx.FileInfo, error) {
+	srcLevel := dstLevel - 1
+
+	prevMaxInfo, err := db.Replica.MaxLTXFileInfo(ctx, dstLevel)
+	if err != nil {
+		return nil, fmt.Errorf("cannot determine max ltx file for destination level: %w", err)
+	}
+	seekTXID := prevMaxInfo.MaxTXID + 1
+
+	// Collect files after last compaction.
+	itr, err := db.Replica.Client.LTXFiles(ctx, srcLevel, seekTXID)
+	if err != nil {
+		return nil, fmt.Errorf("source ltx files after %s: %w", seekTXID, err)
+	}
+	defer itr.Close()
+
+	// Ensure all readers are closed by the end, even if an error occurs.
+	var rdrs []io.Reader
+	defer func() {
+		for _, rd := range rdrs {
+			_ = rd.(io.Closer).Close()
+		}
+	}()
+
+	// Build a list of input files to compact from.
+	var minTXID, maxTXID ltx.TXID
+	for itr.Next() {
+		info := itr.Item()
+
+		// Track TXID bounds of all files being compacted.
+		if minTXID == 0 || info.MinTXID < minTXID {
+			minTXID = info.MinTXID
+		}
+		if maxTXID == 0 || info.MaxTXID > maxTXID {
+			maxTXID = info.MaxTXID
+		}
+
+		f, err := db.Replica.Client.OpenLTXFile(ctx, info.Level, info.MinTXID, info.MaxTXID)
+		if err != nil {
+			return nil, fmt.Errorf("open ltx file: %w", err)
+		}
+		rdrs = append(rdrs, f)
+	}
+	if len(rdrs) == 0 {
+		return nil, ErrNoCompaction
+	}
+
+	// Stream compaction to destination in level.
+	pr, pw := io.Pipe()
+	go func() {
+		c := ltx.NewCompactor(pw, rdrs)
+		c.HeaderFlags = ltx.HeaderFlagNoChecksum | ltx.HeaderFlagCompressLZ4
+		_ = pw.CloseWithError(c.Compact(ctx))
+	}()
+
+	info, err := db.Replica.Client.WriteLTXFile(ctx, dstLevel, minTXID, maxTXID, pr)
+	if err != nil {
+		return nil, fmt.Errorf("write ltx file: %w", err)
+	}
+
+	// Cache last metadata for the level.
+	db.maxLTXFileInfos.Lock()
+	db.maxLTXFileInfos.m[dstLevel] = info
+	db.maxLTXFileInfos.Unlock()
+
+	return info, nil
+}
+
+// SnapshotDB writes a snapshot to the replica for the current position of the database.
+func (db *DB) Snapshot(ctx context.Context) (*ltx.FileInfo, error) {
+	pos, r, err := db.SnapshotReader(ctx)
+	if err != nil {
+		return nil, err
+	}
+	info, err := db.Replica.Client.WriteLTXFile(ctx, SnapshotLevel, 1, pos.TXID, r)
+	if err != nil {
+		return info, err
+	}
+
+	db.maxLTXFileInfos.Lock()
+	db.maxLTXFileInfos.m[SnapshotLevel] = info
+	db.maxLTXFileInfos.Unlock()
+
+	return info, nil
+}
+
 // monitor runs in a separate goroutine and monitors the database & WAL.
 func (db *DB) monitor() {
 	ticker := time.NewTicker(db.MonitorInterval)
@@ -1116,6 +1307,26 @@ func (db *DB) CRC64(ctx context.Context) (uint64, ltx.Pos, error) {
 	return h.Sum64(), pos, nil
 }
 
+// MaxLTXFileInfo returns the metadata for the last LTX file in a level.
+// If cached, it will returned the local copy. Otherwise, it fetches from the replica.
+func (db *DB) MaxLTXFileInfo(ctx context.Context, level int) (ltx.FileInfo, error) {
+	db.maxLTXFileInfos.Lock()
+	defer db.maxLTXFileInfos.Unlock()
+
+	info, ok := db.maxLTXFileInfos.m[level]
+	if ok {
+		return *info, nil
+	}
+
+	remoteInfo, err := db.Replica.MaxLTXFileInfo(ctx, level)
+	if err != nil {
+		return ltx.FileInfo{}, fmt.Errorf("cannot determine L%d max ltx file for %q: %w", level, db.Path(), err)
+	}
+
+	db.maxLTXFileInfos.m[level] = &remoteInfo
+	return remoteInfo, nil
+}
+
 // DefaultRestoreParallelism is the default parallelism when downloading WAL files.
 const DefaultRestoreParallelism = 8
 
diff --git a/file/replica_client.go b/file/replica_client.go
index be57231..8f5cff6 100644
--- a/file/replica_client.go
+++ b/file/replica_client.go
@@ -59,7 +59,7 @@ func (c *ReplicaClient) LTXFilePath(level int, minTXID, maxTXID ltx.TXID) string
 }
 
 // LTXFiles returns an iterator over all available LTX files.
-func (c *ReplicaClient) LTXFiles(ctx context.Context, level int) (ltx.FileIterator, error) {
+func (c *ReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID) (ltx.FileIterator, error) {
 	f, err := os.Open(c.LTXLevelDir(level))
 	if os.IsNotExist(err) {
 		return ltx.NewFileInfoSliceIterator(nil), nil
@@ -79,6 +79,8 @@ func (c *ReplicaClient) LTXFiles(ctx context.Context, level int) (ltx.FileIterat
 		minTXID, maxTXID, err := ltx.ParseFilename(fi.Name())
 		if err != nil {
 			continue
+		} else if minTXID < seek {
+			continue
 		}
 
 		infos = append(infos, &ltx.FileInfo{
@@ -101,7 +103,6 @@ func (c *ReplicaClient) OpenLTXFile(ctx context.Context, level int, minTXID, max
 
 // WriteLTXFile writes an LTX file to disk.
 func (c *ReplicaClient) WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, rd io.Reader) (info *ltx.FileInfo, err error) {
-
 	var fileInfo, dirInfo os.FileInfo
 	if db := c.db(); db != nil {
 		fileInfo, dirInfo = db.FileInfo(), db.DirInfo()
@@ -126,12 +127,9 @@ func (c *ReplicaClient) WriteLTXFile(ctx context.Context, level int, minTXID, ma
 	if err := f.Sync(); err != nil {
 		return nil, err
 	}
-	if err := f.Close(); err != nil {
-		return nil, err
-	}
 
 	// Build metadata.
-	fi, err := os.Stat(filename + ".tmp")
+	fi, err := f.Stat()
 	if err != nil {
 		return nil, err
 	}
@@ -143,6 +141,10 @@ func (c *ReplicaClient) WriteLTXFile(ctx context.Context, level int, minTXID, ma
 		CreatedAt: fi.ModTime().UTC(),
 	}
 
+	if err := f.Close(); err != nil {
+		return nil, err
+	}
+
 	// Move LTX file to final path when it has been written & synced to disk.
 	if err := os.Rename(filename+".tmp", filename); err != nil {
 		return nil, err
diff --git a/gcs/replica_client.go b/gcs/replica_client.go
index f4ba7f5..b8289dd 100644
--- a/gcs/replica_client.go
+++ b/gcs/replica_client.go
@@ -89,12 +89,18 @@ func (c *ReplicaClient) DeleteAll(ctx context.Context) error {
 }
 
 // LTXFiles returns an iterator over all available LTX files for a level.
-func (c *ReplicaClient) LTXFiles(ctx context.Context, level int) (ltx.FileIterator, error) {
+func (c *ReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID) (ltx.FileIterator, error) {
 	if err := c.Init(ctx); err != nil {
 		return nil, err
 	}
+
 	dir := litestream.LTXLevelDir(c.Path, level)
-	return newLTXFileIterator(c.bkt.Objects(ctx, &storage.Query{Prefix: dir + "/"}), level), nil
+	prefix := dir + "/"
+	if seek != 0 {
+		prefix += seek.String()
+	}
+
+	return newLTXFileIterator(c.bkt.Objects(ctx, &storage.Query{Prefix: prefix}), level), nil
 }
 
 // WriteLTXFile writes an LTX file from rd to a remote path.
diff --git a/mock/replica_client.go b/mock/replica_client.go
index 2182560..2716897 100644
--- a/mock/replica_client.go
+++ b/mock/replica_client.go
@@ -12,7 +12,7 @@ var _ litestream.ReplicaClient = (*ReplicaClient)(nil)
 
 type ReplicaClient struct {
 	DeleteAllFunc      func(ctx context.Context) error
-	LTXFilesFunc       func(ctx context.Context, level int) (ltx.FileIterator, error)
+	LTXFilesFunc       func(ctx context.Context, level int, seek ltx.TXID) (ltx.FileIterator, error)
 	OpenLTXFileFunc    func(ctx context.Context, level int, minTXID, maxTXID ltx.TXID) (io.ReadCloser, error)
 	WriteLTXFileFunc   func(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, r io.Reader) (*ltx.FileInfo, error)
 	DeleteLTXFilesFunc func(ctx context.Context, a []*ltx.FileInfo) error
@@ -24,8 +24,8 @@ func (c *ReplicaClient) DeleteAll(ctx context.Context) error {
 	return c.DeleteAllFunc(ctx)
 }
 
-func (c *ReplicaClient) LTXFiles(ctx context.Context, level int) (ltx.FileIterator, error) {
-	return c.LTXFilesFunc(ctx, level)
+func (c *ReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID) (ltx.FileIterator, error) {
+	return c.LTXFilesFunc(ctx, level, seek)
 }
 
 func (c *ReplicaClient) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID) (io.ReadCloser, error) {
diff --git a/replica.go b/replica.go
index c2bb7c6..081936f 100644
--- a/replica.go
+++ b/replica.go
@@ -28,8 +28,7 @@ const (
 // The replica manages periodic synchronization and maintaining the current
 // replica position.
 type Replica struct {
-	db   *DB
-	name string
+	db *DB
 
 	mu  sync.RWMutex
 	pos ltx.Pos // current replicated position
@@ -65,10 +64,9 @@ type Replica struct {
 	AgeRecipients []age.Recipient
 }
 
-func NewReplica(db *DB, name string) *Replica {
+func NewReplica(db *DB) *Replica {
 	r := &Replica{
 		db:     db,
-		name:   name,
 		cancel: func() {},
 
 		SyncInterval:           DefaultSyncInterval,
@@ -80,21 +78,13 @@ func NewReplica(db *DB, name string) *Replica {
 	return r
 }
 
-// Name returns the name of the replica.
-func (r *Replica) Name() string {
-	if r.name == "" && r.Client != nil {
-		return r.Client.Type()
-	}
-	return r.name
-}
-
 // Logger returns the DB sub-logger for this replica.
 func (r *Replica) Logger() *slog.Logger {
 	logger := slog.Default()
 	if r.db != nil {
 		logger = r.db.Logger
 	}
-	return logger.With("replica", r.Name())
+	return logger.With("replica", r.Client.Type())
 }
 
 // DB returns a reference to the database the replica is attached to, if any.
@@ -204,28 +194,29 @@ func (r *Replica) uploadLTXFile(ctx context.Context, level int, minTXID, maxTXID
 
 // calcPos returns the last position saved to the replica for level 0.
 func (r *Replica) calcPos(ctx context.Context) (pos ltx.Pos, err error) {
-	_, maxTXID, err := r.maxLTXFile(ctx, 0)
+	info, err := r.MaxLTXFileInfo(ctx, 0)
 	if err != nil {
 		return pos, fmt.Errorf("max ltx file: %w", err)
 	}
-	return ltx.Pos{TXID: maxTXID}, nil
+	return ltx.Pos{TXID: info.MaxTXID}, nil
 }
 
-// maxLTXFile returns the min & max TXID of the last LTX file for a given level.
-func (r *Replica) maxLTXFile(ctx context.Context, level int) (minTXID, maxTXID ltx.TXID, err error) {
-	itr, err := r.Client.LTXFiles(ctx, level)
+// MaxLTXFileInfo returns metadata about the last LTX file for a given level.
+// Retuns nil if no files exist for the level.
+func (r *Replica) MaxLTXFileInfo(ctx context.Context, level int) (info ltx.FileInfo, err error) {
+	itr, err := r.Client.LTXFiles(ctx, level, 0)
 	if err != nil {
-		return 0, 0, err
+		return info, err
 	}
 	defer itr.Close()
 
 	for itr.Next() {
-		info := itr.Item()
-		if info.MaxTXID > maxTXID {
-			minTXID, maxTXID = info.MinTXID, info.MaxTXID
+		item := itr.Item()
+		if item.MaxTXID > info.MaxTXID {
+			info = *item
 		}
 	}
-	return minTXID, maxTXID, itr.Close()
+	return info, itr.Close()
 }
 
 // Pos returns the current replicated position.
@@ -387,7 +378,7 @@ func (r *Replica) retainer(ctx context.Context) {
 func (r *Replica) validator(ctx context.Context) {
 	// Initialize counters since validation occurs infrequently.
 	for _, status := range []string{"ok", "error"} {
-		replicaValidationTotalCounterVec.WithLabelValues(r.db.Path(), r.Name(), status).Add(0)
+		replicaValidationTotalCounterVec.WithLabelValues(r.db.Path(), status).Add(0)
 	}
 
 	// Exit validation if interval is not set.
@@ -437,9 +428,8 @@ func (r *Replica) Validate(ctx context.Context) error {
 
 	restorePath := filepath.Join(tmpdir, "replica")
 	if err := r.Restore(ctx, RestoreOptions{
-		OutputPath:  restorePath,
-		ReplicaName: r.Name(),
-		TXID:        pos.TXID - 1,
+		OutputPath: restorePath,
+		TXID:       pos.TXID - 1,
 	}); err != nil {
 		return fmt.Errorf("cannot restore: %w", err)
 	}
@@ -468,11 +458,11 @@ func (r *Replica) Validate(ctx context.Context) error {
 
 	// Validate checksums match.
 	if mismatch {
-		replicaValidationTotalCounterVec.WithLabelValues(r.db.Path(), r.Name(), "error").Inc()
+		replicaValidationTotalCounterVec.WithLabelValues(r.db.Path(), "error").Inc()
 		return ErrChecksumMismatch
 	}
 
-	replicaValidationTotalCounterVec.WithLabelValues(r.db.Path(), r.Name(), "ok").Inc()
+	replicaValidationTotalCounterVec.WithLabelValues(r.db.Path(), "ok").Inc()
 
 	if err := os.RemoveAll(tmpdir); err != nil {
 		return fmt.Errorf("cannot remove temporary validation directory: %w", err)
@@ -528,7 +518,7 @@ func (r *Replica) waitForReplica(ctx context.Context, pos ltx.Pos) error {
 func (r *Replica) CreatedAt(ctx context.Context) (time.Time, error) {
 	var min time.Time
 
-	itr, err := r.Client.LTXFiles(ctx, 0)
+	itr, err := r.Client.LTXFiles(ctx, 0, 0)
 	if err != nil {
 		return min, err
 	}
@@ -543,7 +533,7 @@ func (r *Replica) CreatedAt(ctx context.Context) (time.Time, error) {
 // TimeBounds returns the creation time & last updated time.
 // Returns zero time if LTX files exist.
 func (r *Replica) TimeBounds(ctx context.Context) (createdAt, updatedAt time.Time, err error) {
-	itr, err := r.Client.LTXFiles(ctx, 0)
+	itr, err := r.Client.LTXFiles(ctx, 0, 0)
 	if err != nil {
 		return createdAt, updatedAt, err
 	}
@@ -601,7 +591,7 @@ func (r *Replica) Restore(ctx context.Context, opt RestoreOptions) (err error) {
 	}
 
 	// Fetch every LTX file from TXID 1 to target TXID or timestamp.
-	itr, err := r.Client.LTXFiles(ctx, 0)
+	itr, err := r.Client.LTXFiles(ctx, 0, 0)
 	if err != nil {
 		return fmt.Errorf("cannot list ltx files: %w", err)
 	}
@@ -680,5 +670,5 @@ var (
 		Subsystem: "replica",
 		Name:      "validation_total",
 		Help:      "The number of validations performed",
-	}, []string{"db", "name", "status"})
+	}, []string{"db", "status"})
 )
diff --git a/replica_client.go b/replica_client.go
index 04b83ba..f1ce44a 100644
--- a/replica_client.go
+++ b/replica_client.go
@@ -13,7 +13,8 @@ type ReplicaClient interface {
 	Type() string
 
 	// LTXFiles returns an iterator of all LTX files on the replica for a given level.
-	LTXFiles(ctx context.Context, level int) (ltx.FileIterator, error)
+	// If seek is specified, the iterator start from the given TXID or the next available if not found.
+	LTXFiles(ctx context.Context, level int, seek ltx.TXID) (ltx.FileIterator, error)
 
 	// OpenLTXFile returns a reader that contains an LTX file at a given TXID.
 	// Returns an os.ErrNotFound error if the LTX file does not exist.
diff --git a/s3/replica_client.go b/s3/replica_client.go
index c85167c..6d19ee1 100644
--- a/s3/replica_client.go
+++ b/s3/replica_client.go
@@ -197,11 +197,11 @@ func (c *ReplicaClient) DeleteAll(ctx context.Context) error {
 }
 
 // LTXFiles returns an iterator over all available LTX files.
-func (c *ReplicaClient) LTXFiles(ctx context.Context, level int) (ltx.FileIterator, error) {
+func (c *ReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID) (ltx.FileIterator, error) {
 	if err := c.Init(ctx); err != nil {
 		return nil, err
 	}
-	return newLTXFileIterator(ctx, c, level), nil
+	return newLTXFileIterator(ctx, c, level, seek), nil
 }
 
 // WriteLTXFile writes an LTX file from rd into a remote location.
@@ -298,6 +298,7 @@ func (c *ReplicaClient) DeleteLTXFiles(ctx context.Context, a []*ltx.FileInfo) e
 type ltxFileIterator struct {
 	client *ReplicaClient
 	level  int
+	seek   ltx.TXID
 
 	ch     chan *ltx.FileInfo
 	g      errgroup.Group
@@ -308,10 +309,11 @@ type ltxFileIterator struct {
 	err  error
 }
 
-func newLTXFileIterator(ctx context.Context, client *ReplicaClient, level int) *ltxFileIterator {
+func newLTXFileIterator(ctx context.Context, client *ReplicaClient, level int, seek ltx.TXID) *ltxFileIterator {
 	itr := &ltxFileIterator{
 		client: client,
 		level:  level,
+		seek:   seek,
 		ch:     make(chan *ltx.FileInfo),
 	}
 
@@ -326,10 +328,14 @@ func (itr *ltxFileIterator) fetch() error {
 	defer close(itr.ch)
 
 	dir := litestream.LTXLevelDir(itr.client.Path, itr.level)
+	prefix := dir + "/"
+	if itr.seek != 0 {
+		prefix += itr.seek.String()
+	}
 
 	return itr.client.s3.ListObjectsPagesWithContext(itr.ctx, &s3.ListObjectsInput{
 		Bucket:    aws.String(itr.client.Bucket),
-		Prefix:    aws.String(dir + "/"),
+		Prefix:    aws.String(prefix),
 		Delimiter: aws.String("/"),
 	}, func(page *s3.ListObjectsOutput, lastPage bool) bool {
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
diff --git a/sftp/replica_client.go b/sftp/replica_client.go
index 781fad0..0211980 100644
--- a/sftp/replica_client.go
+++ b/sftp/replica_client.go
@@ -155,7 +155,7 @@ func (c *ReplicaClient) DeleteAll(ctx context.Context) (err error) {
 }
 
 // LTXFiles returns an iterator over all available LTX files for a level.
-func (c *ReplicaClient) LTXFiles(ctx context.Context, level int) (_ ltx.FileIterator, err error) {
+func (c *ReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID) (_ ltx.FileIterator, err error) {
 	defer func() { c.resetOnConnError(err) }()
 
 	sftpClient, err := c.Init(ctx)
@@ -177,6 +177,8 @@ func (c *ReplicaClient) LTXFiles(ctx context.Context, level int) (_ ltx.FileIter
 		minTXID, maxTXID, err := ltx.ParseFilename(path.Base(fi.Name()))
 		if err != nil {
 			continue
+		} else if minTXID < seek {
+			continue
 		}
 
 		infos = append(infos, &ltx.FileInfo{
diff --git a/store.go b/store.go
new file mode 100644
index 0000000..7062029
--- /dev/null
+++ b/store.go
@@ -0,0 +1,211 @@
+package litestream
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"slices"
+	"sync"
+	"time"
+
+	"github.com/superfly/ltx"
+)
+
+var (
+	// ErrNoCompaction is returned when no new files are available from the previous level.
+	ErrNoCompaction = errors.New("no compaction")
+
+	// ErrCompactionTooEarly is returned when a compaction is attempted too soon
+	// since the last compaction time. This is used to prevent frequent
+	// re-compaction when restarting the process.
+	ErrCompactionTooEarly = errors.New("compaction too early")
+)
+
+// Store defaults
+const (
+	DefaultSnapshotInterval  = 24 * time.Hour
+	DefaultSnapshotRetention = 24 * time.Hour
+)
+
+// Store represents the top-level container for databases.
+//
+// It manages async background tasks like compactions so that the system
+// is not overloaded by too many concurrent tasks.
+type Store struct {
+	mu     sync.Mutex
+	dbs    []*DB
+	levels CompactionLevels
+
+	wg     sync.WaitGroup
+	ctx    context.Context
+	cancel func()
+
+	// The frequency of snapshots.
+	SnapshotInterval time.Duration
+	// The duration of time that snapshots are kept before being deleted.
+	SnapshotRetention time.Duration
+
+	// If true, compaction is run in the background according to compaction levels.
+	CompactionMonitorEnabled bool
+}
+
+func NewStore(dbs []*DB, levels CompactionLevels) *Store {
+	s := &Store{
+		dbs:    dbs,
+		levels: levels,
+
+		SnapshotInterval:         DefaultSnapshotInterval,
+		SnapshotRetention:        DefaultSnapshotRetention,
+		CompactionMonitorEnabled: true,
+	}
+	s.ctx, s.cancel = context.WithCancel(context.Background())
+	return s
+}
+
+func (s *Store) Open(ctx context.Context) error {
+	if err := s.levels.Validate(); err != nil {
+		return err
+	}
+
+	for _, db := range s.dbs {
+		if err := db.Open(); err != nil {
+			return err
+		}
+	}
+
+	// Start monitors for compactions & snapshots.
+	if s.CompactionMonitorEnabled {
+		for _, lvl := range s.levels {
+			lvl := lvl
+			if lvl.Level == 0 {
+				continue
+			}
+			go func() {
+				defer s.wg.Done()
+				s.monitorCompactionLevel(s.ctx, lvl)
+			}()
+		}
+
+		go func() {
+			defer s.wg.Done()
+			s.monitorCompactionLevel(s.ctx, s.SnapshotLevel())
+		}()
+	}
+
+	return nil
+}
+
+func (s *Store) Close() (err error) {
+	for _, db := range s.dbs {
+		if e := db.Close(context.Background()); e != nil && err == nil {
+			err = e
+		}
+	}
+
+	// Cancel and wait for background tasks to complete.
+	s.cancel()
+	s.wg.Wait()
+
+	return err
+}
+
+func (s *Store) DBs() []*DB {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return slices.Clone(s.dbs)
+}
+
+// SnapshotLevel returns a pseudo compaction level based on snapshot settings.
+func (s *Store) SnapshotLevel() *CompactionLevel {
+	return &CompactionLevel{
+		Level:     SnapshotLevel,
+		Interval:  s.SnapshotInterval,
+		Retention: s.SnapshotRetention,
+	}
+}
+
+func (s *Store) monitorCompactionLevel(ctx context.Context, lvl *CompactionLevel) {
+	// Start first compaction immediately to check for any missed compactions from shutdown
+	timer := time.NewTimer(time.Nanosecond)
+
+LOOP:
+	for {
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			break LOOP
+
+		case <-timer.C:
+			// Reset timer before we start compactions so we don't delay it
+			// from long compactions.
+			timer = time.NewTimer(time.Until(lvl.NextCompactionAt(time.Now())))
+
+			for _, db := range s.DBs() {
+				if _, err := s.CompactDB(ctx, db, lvl); errors.Is(err, ErrNoCompaction) {
+					slog.Debug("no compaction", "path", db.Path())
+					continue
+				} else if errors.Is(err, ErrCompactionTooEarly) {
+					slog.Debug("recently compacted, skipping", "path", db.Path())
+					continue
+				} else if err != nil {
+					slog.Error("compaction failed", "error", err)
+					time.Sleep(1 * time.Second) // wait so we don't rack up S3 charges
+				}
+			}
+		}
+	}
+}
+
+// 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) {
+	dstLevel := lvl.Level
+
+	// Ensure we are not re-compacting before the most recent compaction time.
+	prevCompactionAt := lvl.PrevCompactionAt(time.Now())
+	dstInfo, err := db.MaxLTXFileInfo(ctx, dstLevel)
+	if err != nil {
+		return nil, fmt.Errorf("fetch dst level info: %w", err)
+	} else if dstInfo.CreatedAt.After(prevCompactionAt) {
+		return nil, ErrCompactionTooEarly
+	}
+
+	// Shortcut if this is a snapshot since we are not pulling from a previous level.
+	if dstLevel == SnapshotLevel {
+		info, err := db.Snapshot(ctx)
+		if err != nil {
+			return info, err
+		}
+		slog.InfoContext(ctx, "snapshot complete", "txid", info.MaxTXID.String(), "size", info.Size)
+		return info, nil
+	}
+
+	// Fetch latest LTX files for both the source & destination so we can see if we need to make progress.
+	srcLevel := s.levels.PrevLevel(dstLevel)
+	srcInfo, err := db.MaxLTXFileInfo(ctx, srcLevel)
+	if err != nil {
+		return nil, fmt.Errorf("fetch src level info: %w", err)
+	}
+
+	// Skip if there are no new files to compact.
+	if srcInfo.MaxTXID <= dstInfo.MinTXID {
+		return nil, ErrNoCompaction
+	}
+
+	info, err := db.Compact(ctx, dstLevel)
+	if err != nil {
+		return info, err
+	}
+
+	slog.InfoContext(ctx, "compaction complete",
+		"level", dstLevel,
+		slog.Group("txid",
+			"min", info.MinTXID.String(),
+			"max", info.MaxTXID.String(),
+		),
+		"size", info.Size,
+	)
+
+	return info, nil
+}


Test patch:
diff --git a/cmd/litestream/databases.go b/cmd/litestream/databases.go
index a2b22a5..86f85bb 100644
--- a/cmd/litestream/databases.go
+++ b/cmd/litestream/databases.go
@@ -35,7 +35,7 @@ func (c *DatabasesCommand) Run(ctx context.Context, args []string) (err error) {
 	w := tabwriter.NewWriter(os.Stdout, 0, 8, 2, ' ', 0)
 	defer w.Flush()
 
-	fmt.Fprintln(w, "path\treplicas")
+	fmt.Fprintln(w, "path\treplica")
 	for _, dbConfig := range config.DBs {
 		db, err := NewDBFromConfig(dbConfig)
 		if err != nil {
@@ -44,7 +44,7 @@ func (c *DatabasesCommand) Run(ctx context.Context, args []string) (err error) {
 
 		fmt.Fprintf(w, "%s\t%s\n",
 			db.Path(),
-			db.Replica.Name())
+			db.Replica.Client.Type())
 	}
 
 	return nil
diff --git a/cmd/litestream/ltx.go b/cmd/litestream/ltx.go
index 107157b..6cfaeff 100644
--- a/cmd/litestream/ltx.go
+++ b/cmd/litestream/ltx.go
@@ -70,7 +70,7 @@ func (c *LTXCommand) Run(ctx context.Context, args []string) (err error) {
 	defer w.Flush()
 
 	fmt.Fprintln(w, "min_txid\tmax_txid\tsize\tcreated")
-	itr, err := r.Client.LTXFiles(ctx, 0)
+	itr, err := r.Client.LTXFiles(ctx, 0, 0)
 	if err != nil {
 		return err
 	}
diff --git a/cmd/litestream/main.go b/cmd/litestream/main.go
index 88b4358..4b5508d 100644
--- a/cmd/litestream/main.go
+++ b/cmd/litestream/main.go
@@ -153,6 +153,10 @@ type Config struct {
 	// Bind address for serving metrics.
 	Addr string `yaml:"addr"`
 
+	// List of stages in a multi-level compaction.
+	// Only includes L1 through the last non-snapshot level.
+	Levels []*CompactionLevelConfig `yaml:"levels"`
+
 	// List of databases to manage.
 	DBs []*DBConfig `yaml:"dbs"`
 
@@ -194,6 +198,23 @@ func DefaultConfig() Config {
 	return Config{}
 }
 
+// CompactionLevels returns a full list of compaction levels include L0.
+func (c *Config) CompactionLevels() litestream.CompactionLevels {
+	levels := litestream.CompactionLevels{
+		{Level: 0},
+	}
+
+	for i, lvl := range c.Levels {
+		levels = append(levels, &litestream.CompactionLevel{
+			Level:     i + 1,
+			Interval:  lvl.Interval,
+			Retention: lvl.Retention,
+		})
+	}
+
+	return levels
+}
+
 // DBConfig returns database configuration by path.
 func (c *Config) DBConfig(path string) *DBConfig {
 	for _, dbConfig := range c.DBs {
@@ -275,6 +296,12 @@ func ReadConfigFile(filename string, expandEnv bool) (_ Config, err error) {
 	return config, nil
 }
 
+// CompactionLevelConfig the configuration for a single level of compaction.
+type CompactionLevelConfig struct {
+	Interval  time.Duration `yaml:"interval"`
+	Retention time.Duration `yaml:"retention"`
+}
+
 // DBConfig represents the configuration for a single database.
 type DBConfig struct {
 	Path               string         `yaml:"path"`
@@ -391,7 +418,7 @@ func NewReplicaFromConfig(c *ReplicaConfig, db *litestream.DB) (_ *litestream.Re
 	}
 
 	// Build replica.
-	r := litestream.NewReplica(db, c.Name)
+	r := litestream.NewReplica(db)
 	if v := c.Retention; v != nil {
 		r.Retention = *v
 	}
@@ -482,7 +509,7 @@ func newFileReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_
 }
 
 // newS3ReplicaClientFromConfig returns a new instance of s3.ReplicaClient built from config.
-func newS3ReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *s3.ReplicaClient, err error) {
+func newS3ReplicaClientFromConfig(c *ReplicaConfig, _ *litestream.Replica) (_ *s3.ReplicaClient, err error) {
 	// Ensure URL & constituent parts are not both specified.
 	if c.URL != "" && c.Path != "" {
 		return nil, fmt.Errorf("cannot specify url & path for s3 replica")
@@ -545,7 +572,7 @@ func newS3ReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *s
 }
 
 // newGCSReplicaClientFromConfig returns a new instance of gcs.ReplicaClient built from config.
-func newGCSReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *gcs.ReplicaClient, err error) {
+func newGCSReplicaClientFromConfig(c *ReplicaConfig, _ *litestream.Replica) (_ *gcs.ReplicaClient, err error) {
 	// Ensure URL & constituent parts are not both specified.
 	if c.URL != "" && c.Path != "" {
 		return nil, fmt.Errorf("cannot specify url & path for gcs replica")
@@ -584,7 +611,7 @@ func newGCSReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *
 }
 
 // newABSReplicaClientFromConfig returns a new instance of abs.ReplicaClient built from config.
-func newABSReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *abs.ReplicaClient, err error) {
+func newABSReplicaClientFromConfig(c *ReplicaConfig, _ *litestream.Replica) (_ *abs.ReplicaClient, err error) {
 	// Ensure URL & constituent parts are not both specified.
 	if c.URL != "" && c.Path != "" {
 		return nil, fmt.Errorf("cannot specify url & path for abs replica")
@@ -627,7 +654,7 @@ func newABSReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *
 }
 
 // newSFTPReplicaClientFromConfig returns a new instance of sftp.ReplicaClient built from config.
-func newSFTPReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *sftp.ReplicaClient, err error) {
+func newSFTPReplicaClientFromConfig(c *ReplicaConfig, _ *litestream.Replica) (_ *sftp.ReplicaClient, err error) {
 	// Ensure URL & constituent parts are not both specified.
 	if c.URL != "" && c.Path != "" {
 		return nil, fmt.Errorf("cannot specify url & path for sftp replica")
diff --git a/cmd/litestream/replicate.go b/cmd/litestream/replicate.go
index 64123b9..1904a8b 100644
--- a/cmd/litestream/replicate.go
+++ b/cmd/litestream/replicate.go
@@ -28,8 +28,8 @@ type ReplicateCommand struct {
 
 	Config Config
 
-	// List of managed databases specified in the config.
-	DBs []*litestream.DB
+	// Manages the set of databases & compaction levels.
+	Store *litestream.Store
 }
 
 func NewReplicateCommand() *ReplicateCommand {
@@ -92,24 +92,23 @@ func (c *ReplicateCommand) Run() (err error) {
 		slog.Error("no databases specified in configuration")
 	}
 
+	var dbs []*litestream.DB
 	for _, dbConfig := range c.Config.DBs {
 		db, err := NewDBFromConfig(dbConfig)
 		if err != nil {
 			return err
 		}
-
-		// Open database & attach to program.
-		if err := db.Open(); err != nil {
-			return err
-		}
-		c.DBs = append(c.DBs, db)
+		dbs = append(dbs, db)
 	}
 
+	levels := c.Config.CompactionLevels()
+	c.Store = litestream.NewStore(dbs, levels)
+
 	// Notify user that initialization is done.
-	for _, db := range c.DBs {
+	for _, db := range c.Store.DBs() {
 		r := db.Replica
 		slog.Info("initialized db", "path", db.Path())
-		slog := slog.With("name", r.Name(), "type", r.Client.Type(), "sync-interval", r.SyncInterval)
+		slog := slog.With("type", r.Client.Type(), "sync-interval", r.SyncInterval)
 		switch client := r.Client.(type) {
 		case *file.ReplicaClient:
 			slog.Info("replicating to", "path", client.Path())
@@ -166,13 +165,8 @@ func (c *ReplicateCommand) Run() (err error) {
 
 // Close closes all open databases.
 func (c *ReplicateCommand) Close() (err error) {
-	for _, db := range c.DBs {
-		if e := db.Close(context.Background()); e != nil {
-			db.Logger.Error("error closing db", "error", e)
-			if err == nil {
-				err = e
-			}
-		}
+	if e := c.Store.Close(); e != nil {
+		slog.Error("failed to close database", "error", e)
 	}
 	return err
 }
diff --git a/cmd/litestream/restore.go b/cmd/litestream/restore.go
index 8b8d0f8..9598d1d 100644
--- a/cmd/litestream/restore.go
+++ b/cmd/litestream/restore.go
@@ -107,7 +107,7 @@ func (c *RestoreCommand) loadFromURL(ctx context.Context, replicaURL string, ifD
 }
 
 // loadFromConfig returns a replica & updates the restore options from a DB reference.
-func (c *RestoreCommand) loadFromConfig(ctx context.Context, dbPath, configPath string, expandEnv, ifDBNotExists bool, opt *litestream.RestoreOptions) (*litestream.Replica, error) {
+func (c *RestoreCommand) loadFromConfig(_ context.Context, dbPath, configPath string, expandEnv, ifDBNotExists bool, opt *litestream.RestoreOptions) (*litestream.Replica, error) {
 	// Load configuration.
 	config, err := ReadConfigFile(configPath, expandEnv)
 	if err != nil {
diff --git a/db_test.go b/db_test.go
index 16955cb..c1c6ffa 100644
--- a/db_test.go
+++ b/db_test.go
@@ -3,6 +3,7 @@ package litestream_test
 import (
 	"context"
 	"database/sql"
+	"hash/crc64"
 	"log/slog"
 	"os"
 	"path/filepath"
@@ -340,6 +341,158 @@ func TestDB_Sync(t *testing.T) {
 	})
 }
 
+func TestDB_Compact(t *testing.T) {
+	// Ensure that raw L0 transactions can be compacted into the first level.
+	t.Run("L1", func(t *testing.T) {
+		db, sqldb := MustOpenDBs(t)
+		defer MustCloseDBs(t, db, sqldb)
+		db.Replica = litestream.NewReplica(db)
+		db.Replica.Client = NewFileReplicaClient(t)
+
+		if err := db.Sync(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+
+		if _, err := sqldb.Exec(`CREATE TABLE t (id INT);`); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := sqldb.Exec(`INSERT INTO t (id) VALUES (100)`); err != nil {
+			t.Fatal(err)
+		}
+
+		if err := db.Sync(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+
+		if err := db.Replica.Sync(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+
+		info, err := db.Compact(t.Context(), 1)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if got, want := info.Level, 1; got != want {
+			t.Fatalf("Level=%v, want %v", got, want)
+		}
+		if got, want := info.MinTXID, ltx.TXID(1); got != want {
+			t.Fatalf("MinTXID=%s, want %s", got, want)
+		}
+		if got, want := info.MaxTXID, ltx.TXID(2); got != want {
+			t.Fatalf("MaxTXID=%s, want %s", got, want)
+		}
+		if info.Size == 0 {
+			t.Fatalf("expected non-zero size")
+		}
+	})
+
+	// Ensure that higher level compactions pull from the correct levels.
+	t.Run("L2+", func(t *testing.T) {
+		db, sqldb := MustOpenDBs(t)
+		defer MustCloseDBs(t, db, sqldb)
+
+		if err := db.Sync(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+
+		if _, err := sqldb.Exec(`CREATE TABLE t (id INT);`); err != nil {
+			t.Fatal(err)
+		}
+
+		// TXID 2
+		if _, err := sqldb.Exec(`INSERT INTO t (id) VALUES (100)`); err != nil {
+			t.Fatal(err)
+		} else if err := db.Sync(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+
+		if err := db.Replica.Sync(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+
+		// Compact to L1:1-2
+		if info, err := db.Compact(t.Context(), 1); err != nil {
+			t.Fatal(err)
+		} else if got, want := ltx.FormatFilename(info.MinTXID, info.MaxTXID), `0000000000000001-0000000000000002.ltx`; got != want {
+			t.Fatalf("Filename=%s, want %s", got, want)
+		}
+
+		// TXID 3
+		if _, err := sqldb.Exec(`INSERT INTO t (id) VALUES (100)`); err != nil {
+			t.Fatal(err)
+		} else if err := db.Sync(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+
+		if err := db.Replica.Sync(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+
+		// Compact to L1:3-3
+		if info, err := db.Compact(t.Context(), 1); err != nil {
+			t.Fatal(err)
+		} else if got, want := ltx.FormatFilename(info.MinTXID, info.MaxTXID), `0000000000000003-0000000000000003.ltx`; got != want {
+			t.Fatalf("Filename=%s, want %s", got, want)
+		}
+
+		// Compact to L2:1-3
+		if info, err := db.Compact(t.Context(), 2); err != nil {
+			t.Fatal(err)
+		} else if got, want := info.Level, 2; got != want {
+			t.Fatalf("Level=%v, want %v", got, want)
+		} else if got, want := ltx.FormatFilename(info.MinTXID, info.MaxTXID), `0000000000000001-0000000000000003.ltx`; got != want {
+			t.Fatalf("Filename=%s, want %s", got, want)
+		}
+	})
+}
+
+func TestDB_Snapshot(t *testing.T) {
+	db, sqldb := MustOpenDBs(t)
+	defer MustCloseDBs(t, db, sqldb)
+	db.Replica = litestream.NewReplica(db)
+	db.Replica.Client = NewFileReplicaClient(t)
+
+	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)
+	}
+
+	if _, err := sqldb.Exec(`INSERT INTO t (id) VALUES (100)`); err != nil {
+		t.Fatal(err)
+	} else if err := db.Sync(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+
+	info, err := db.Snapshot(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got, want := ltx.FormatFilename(info.MinTXID, info.MaxTXID), `0000000000000001-0000000000000002.ltx`; got != want {
+		t.Fatalf("Filename=%s, want %s", got, want)
+	}
+
+	// Calculate local checksum
+	chksum0, _, err := db.CRC64(t.Context())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Fetch remote LTX snapshot file and ensure it matches the checksum of the local database.
+	rc, err := db.Replica.Client.OpenLTXFile(t.Context(), litestream.SnapshotLevel, 1, 2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rc.Close()
+
+	h := crc64.New(crc64.MakeTable(crc64.ISO))
+	if err := ltx.NewDecoder(rc).DecodeDatabaseTo(h); err != nil {
+		t.Fatal(err)
+	} else if got, want := h.Sum64(), chksum0; got != want {
+		t.Fatal("snapshot checksum mismatch")
+	}
+}
+
 func newDB(tb testing.TB, path string) *litestream.DB {
 	tb.Helper()
 	tb.Logf("db=%s", path)
@@ -375,6 +528,9 @@ func MustOpenDBAt(tb testing.TB, path string) *litestream.DB {
 	tb.Helper()
 	db := newDB(tb, path)
 	db.MonitorInterval = 0 // disable background goroutine
+	db.Replica = litestream.NewReplica(db)
+	db.Replica.Client = NewFileReplicaClient(tb)
+	db.Replica.MonitorEnabled = false // disable background goroutine
 	if err := db.Open(); err != nil {
 		tb.Fatal(err)
 	}
@@ -384,7 +540,7 @@ func MustOpenDBAt(tb testing.TB, path string) *litestream.DB {
 // MustCloseDB closes db and removes its parent directory.
 func MustCloseDB(tb testing.TB, db *litestream.DB) {
 	tb.Helper()
-	if err := db.Close(context.Background()); err != nil && !strings.Contains(err.Error(), `database is closed`) {
+	if err := db.Close(context.Background()); err != nil && !strings.Contains(err.Error(), `database is closed`) && !strings.Contains(err.Error(), `file already closed`) {
 		tb.Fatal(err)
 	} else if err := os.RemoveAll(filepath.Dir(db.Path())); err != nil {
 		tb.Fatal(err)
diff --git a/replica_client_test.go b/replica_client_test.go
index ef53c9b..fa73e51 100644
--- a/replica_client_test.go
+++ b/replica_client_test.go
@@ -80,7 +80,7 @@ func TestReplicaClient_LTX(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		itr, err := c.LTXFiles(context.Background(), 0)
+		itr, err := c.LTXFiles(context.Background(), 0, 0)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -115,7 +115,7 @@ func TestReplicaClient_LTX(t *testing.T) {
 	RunWithReplicaClient(t, "NoWALs", func(t *testing.T, c litestream.ReplicaClient) {
 		t.Parallel()
 
-		itr, err := c.LTXFiles(context.Background(), 0)
+		itr, err := c.LTXFiles(context.Background(), 0, 0)
 		if err != nil {
 			t.Fatal(err)
 		}
diff --git a/replica_test.go b/replica_test.go
index ccf68c8..4ac5005 100644
--- a/replica_test.go
+++ b/replica_test.go
@@ -6,25 +6,9 @@ import (
 
 	"github.com/benbjohnson/litestream"
 	"github.com/benbjohnson/litestream/file"
-	"github.com/benbjohnson/litestream/mock"
 	"github.com/superfly/ltx"
 )
 
-func TestReplica_Name(t *testing.T) {
-	t.Run("WithName", func(t *testing.T) {
-		if got, want := litestream.NewReplica(nil, "NAME").Name(), "NAME"; got != want {
-			t.Fatalf("Name()=%v, want %v", got, want)
-		}
-	})
-	t.Run("WithoutName", func(t *testing.T) {
-		r := litestream.NewReplica(nil, "")
-		r.Client = &mock.ReplicaClient{}
-		if got, want := r.Name(), "mock"; got != want {
-			t.Fatalf("Name()=%v, want %v", got, want)
-		}
-	})
-}
-
 func TestReplica_Sync(t *testing.T) {
 	db, sqldb := MustOpenDBs(t)
 	defer MustCloseDBs(t, db, sqldb)
@@ -45,7 +29,7 @@ func TestReplica_Sync(t *testing.T) {
 	t.Logf("position after sync: %s", dpos.String())
 
 	c := file.NewReplicaClient(t.TempDir())
-	r := litestream.NewReplica(db, "")
+	r := litestream.NewReplica(db)
 	c.Replica, r.Client = r, c
 
 	if err := r.Sync(context.Background()); err != nil {
diff --git a/store_test.go b/store_test.go
new file mode 100644
index 0000000..fe9b9d5
--- /dev/null
+++ b/store_test.go
@@ -0,0 +1,94 @@
+package litestream_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/benbjohnson/litestream"
+)
+
+func TestStore_CompactDB(t *testing.T) {
+	t.Run("L1", func(t *testing.T) {
+		db0, sqldb0 := MustOpenDBs(t)
+		defer MustCloseDBs(t, db0, sqldb0)
+
+		db1, sqldb1 := MustOpenDBs(t)
+		defer MustCloseDBs(t, db1, sqldb1)
+
+		levels := litestream.CompactionLevels{
+			{Level: 0},
+			{Level: 1, Interval: 1 * time.Second},
+			{Level: 2, Interval: 500 * time.Millisecond},
+		}
+		s := litestream.NewStore([]*litestream.DB{db0, db1}, levels)
+		s.CompactionMonitorEnabled = false
+		if err := s.Open(t.Context()); err != nil {
+			t.Fatal(err)
+		}
+		defer s.Close()
+
+		if _, err := sqldb0.Exec(`CREATE TABLE t (id INT);`); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := sqldb0.Exec(`INSERT INTO t (id) VALUES (100)`); err != nil {
+			t.Fatal(err)
+		} else if err := db0.Sync(context.Background()); err != nil {
+			t.Fatal(err)
+		} else if err := db0.Replica.Sync(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+
+		if _, err := s.CompactDB(t.Context(), db0, levels[1]); err != nil {
+			t.Fatal(err)
+		}
+
+		// Re-compacting immediately should return an error that it's too soon.
+		if _, err := s.CompactDB(t.Context(), db0, levels[1]); err != litestream.ErrCompactionTooEarly {
+			t.Fatalf("unexpected error: %s", err)
+		}
+
+		// Re-compacting after the interval should show that there is nothing to compact.
+		time.Sleep(levels[1].Interval)
+		if _, err := s.CompactDB(t.Context(), db0, levels[1]); err != litestream.ErrNoCompaction {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("Snapshot", func(t *testing.T) {
+		db0, sqldb0 := MustOpenDBs(t)
+		defer MustCloseDBs(t, db0, sqldb0)
+
+		levels := litestream.CompactionLevels{
+			{Level: 0},
+			{Level: 1, Interval: 100 * time.Millisecond},
+			{Level: 2, Interval: 500 * time.Millisecond},
+		}
+		s := litestream.NewStore([]*litestream.DB{db0}, levels)
+		s.CompactionMonitorEnabled = false
+		if err := s.Open(t.Context()); err != nil {
+			t.Fatal(err)
+		}
+		defer s.Close()
+
+		if _, err := sqldb0.Exec(`CREATE TABLE t (id INT);`); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := sqldb0.Exec(`INSERT INTO t (id) VALUES (100)`); err != nil {
+			t.Fatal(err)
+		} else if err := db0.Sync(context.Background()); err != nil {
+			t.Fatal(err)
+		} else if err := db0.Replica.Sync(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+
+		if _, err := s.CompactDB(t.Context(), db0, s.SnapshotLevel()); err != nil {
+			t.Fatal(err)
+		}
+
+		// Re-compacting immediately should return an error that there's nothing to compact.
+		if _, err := s.CompactDB(t.Context(), db0, s.SnapshotLevel()); err != litestream.ErrCompactionTooEarly {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+}