fix(store): serialize snapshots and compactions per database

db.go

@@ -65,19 +65,23 @@ const (
 // with indefinite write blocking (issue #724). All checkpoints now use either
 // PASSIVE (non-blocking) or TRUNCATE (emergency only) modes.
 type DB struct {
-	mu        sync.RWMutex
-	execSem   *semaphore.Weighted
-	path      string        // part to database
-	metaPath  string        // Path to the database metadata.
-	db        *sql.DB       // target database
-	f         *os.File      // long-running db file descriptor
-	rtx       *sql.Tx       // long running read transaction
-	pageSize  int           // page size, in bytes
-	notify    chan struct{} // closes on WAL change
-	chkMu     sync.RWMutex  // checkpoint lock
-	opened    bool          // true if Open() was called and Close() not yet called
-	syncState syncState
-	syncDiag  diagState
+	mu       sync.RWMutex
+	execSem  *semaphore.Weighted
+	path     string        // part to database
+	metaPath string        // Path to the database metadata.
+	db       *sql.DB       // target database
+	f        *os.File      // long-running db file descriptor
+	rtx      *sql.Tx       // long running read transaction
+	pageSize int           // page size, in bytes
+	notify   chan struct{} // closes on WAL change
+	chkMu    sync.RWMutex  // checkpoint lock
+	// maintenanceMu serializes snapshots & compactions per database; held via
+	// TryLock by Store.CompactDB so overlapping maintenance is refused with
+	// ErrMaintenanceBusy rather than queued (issue #1477).
+	maintenanceMu sync.Mutex
+	opened        bool // true if Open() was called and Close() not yet called
+	syncState     syncState
+	syncDiag      diagState
 
 	// last file info for each level
 	maxLTXFileInfos struct {
@@ -2693,12 +2697,23 @@ func (p *snapshotReadPosition) close() {
 
 type snapshotReadCloser struct {
 	*io.PipeReader
-	pos *snapshotReadPosition
+	pos    *snapshotReadPosition
+	cancel context.CancelFunc // stops the producing goroutine
+	done   <-chan struct{}    // closed when the producing goroutine has exited
 }
 
+// Close stops the producing goroutine and waits for it to exit, so that no
+// snapshot state (WAL page map, LTX encoder page index) outlives the reader.
+// Cancelling the producer's context interrupts the WAL scan that runs before
+// the first pipe write; closing the pipe interrupts any write in progress.
+// Store.CompactDB relies on this to keep a failed snapshot's memory from
+// overlapping the next maintenance operation.
 func (r *snapshotReadCloser) Close() error {
-	defer r.pos.close()
-	return r.PipeReader.Close()
+	r.cancel()
+	err := r.PipeReader.Close()
+	<-r.done
+	r.pos.close()
+	return err
 }
 
 // SnapshotReader returns the current position of the database & a reader that contains a full database snapshot.
@@ -2812,8 +2827,12 @@ func (db *DB) snapshotReader(ctx context.Context, pos *snapshotReadPosition) (io
 
 	// Execute encoding in a separate goroutine so the caller can initialize before reading.
 	pr, pw := io.Pipe()
+	ctx, cancel := context.WithCancel(ctx)
+	done := make(chan struct{})
 	go func() {
+		defer close(done)
 		defer pos.close()
+		defer cancel() // release the derived context even if the reader is never closed
 
 		walFile, err := os.Open(db.WALPath())
 		if err != nil {
@@ -2893,7 +2912,7 @@ func (db *DB) snapshotReader(ctx context.Context, pos *snapshotReadPosition) (io
 		_ = pw.Close()
 	}()
 
-	return &snapshotReadCloser{PipeReader: pr, pos: pos}, nil
+	return &snapshotReadCloser{PipeReader: pr, pos: pos, cancel: cancel, done: done}, nil
 }
 
 func snapshotHeaderWALRange(maxOffset, frameSize int64) (offset, size int64) {

go.mod

@@ -27,7 +27,7 @@ require (
 	github.com/prometheus/client_golang v1.17.0
 	github.com/psanford/sqlite3vfs v0.0.0-20260519004904-f9180fa2acc9 // direct
 	github.com/studio-b12/gowebdav v0.11.0
-	github.com/superfly/ltx v0.5.2
+	github.com/superfly/ltx v0.5.3-0.20260827162011-d457a1ab7844
 	golang.org/x/crypto v0.52.0
 	golang.org/x/sys v0.45.0
 	google.golang.org/api v0.155.0

go.sum

@@ -270,8 +270,8 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/studio-b12/gowebdav v0.11.0 h1:qbQzq4USxY28ZYsGJUfO5jR+xkFtcnwWgitp4Zp1irU=
 github.com/studio-b12/gowebdav v0.11.0/go.mod h1:bHA7t77X/QFExdeAnDzK6vKM34kEZAcE1OX4MfiwjkE=
-github.com/superfly/ltx v0.5.2 h1:XVzytSsIlpUaOd2I7M40lDhUQIzGIdqkLBNg5M5Ax48=
-github.com/superfly/ltx v0.5.2/go.mod h1:0OtLSLHHHPa3qDrAmwkLW5pUBfj365JS+bR9FEJDrM4=
+github.com/superfly/ltx v0.5.3-0.20260827162011-d457a1ab7844 h1:4TKqDSdHlpV4Me/BoP/D7gszAPidElRDf64KBwnFg+U=
+github.com/superfly/ltx v0.5.3-0.20260827162011-d457a1ab7844/go.mod h1:0OtLSLHHHPa3qDrAmwkLW5pUBfj365JS+bR9FEJDrM4=
 github.com/tetratelabs/wazero v1.2.1 h1:J4X2hrGzJvt+wqltuvcSjHQ7ujQxA9gb6PeMs4qlUWs=
 github.com/tetratelabs/wazero v1.2.1/go.mod h1:wYx2gNRg8/WihJfSDxA1TIL8H+GkfLYm+bIfbblu9VQ=
 github.com/wasilibs/go-re2 v1.3.0 h1:LFhBNzoStM3wMie6rN2slD1cuYH2CGiHpvNL3UtcsMw=

store.go

@@ -23,6 +23,13 @@ var (
 	// re-compaction when restarting the process.
 	ErrCompactionTooEarly = errors.New("compaction too early")
 
+	// ErrMaintenanceBusy is returned when a snapshot or compaction is refused
+	// because another maintenance operation for the same database is still in
+	// flight. Maintenance is serialized per database so that concurrent
+	// operations cannot each hold a database-sized page index in memory at the
+	// same time (issue #1477); callers retry on the next monitor interval.
+	ErrMaintenanceBusy = errors.New("maintenance busy")
+
 	// ErrTxNotAvailable is returned when a transaction does not exist.
 	ErrTxNotAvailable = errors.New("transaction not available")
 
@@ -57,8 +64,17 @@ func (e *DBNotReadyError) Is(target error) bool {
 
 // Store defaults
 const (
-	DefaultSnapshotInterval  = 24 * time.Hour
-	DefaultSnapshotRetention = 24 * time.Hour
+	DefaultSnapshotInterval = 24 * time.Hour
+
+	// DefaultMaintenanceBusyRetryInterval caps how long a level monitor waits
+	// before retrying a database whose snapshot or compaction was refused with
+	// ErrMaintenanceBusy. Retries start at maintenanceBusyRetryBase and double
+	// on each consecutive busy pass up to this cap, so a momentary collision
+	// between levels resolves within a second while a long-running snapshot
+	// is not polled more than every few seconds.
+	DefaultMaintenanceBusyRetryInterval = 10 * time.Second
+	maintenanceBusyRetryBase            = time.Second
+	DefaultSnapshotRetention            = 24 * time.Hour
 
 	DefaultRetention              = 24 * time.Hour
 	DefaultRetentionCheckInterval = 1 * time.Hour
@@ -96,6 +112,10 @@ type Store struct {
 
 	// The frequency of snapshots.
 	SnapshotInterval time.Duration
+
+	// MaintenanceBusyRetryInterval bounds the delay before a level monitor
+	// retries databases that reported ErrMaintenanceBusy.
+	MaintenanceBusyRetryInterval time.Duration
 	// The duration of time that snapshots are kept before being deleted.
 	SnapshotRetention time.Duration
 
@@ -140,16 +160,17 @@ func NewStore(dbs []*DB, levels CompactionLevels) *Store {
 		dbs:    dbs,
 		levels: levels,
 
-		SnapshotInterval:         DefaultSnapshotInterval,
-		SnapshotRetention:        DefaultSnapshotRetention,
-		L0Retention:              DefaultL0Retention,
-		L0RetentionCheckInterval: DefaultL0RetentionCheckInterval,
-		CompactionMonitorEnabled: true,
-		RetentionEnabled:         true,
-		ShutdownSyncTimeout:      DefaultShutdownSyncTimeout,
-		ShutdownSyncInterval:     DefaultShutdownSyncInterval,
-		HeartbeatCheckInterval:   DefaultHeartbeatCheckInterval,
-		Logger:                   slog.Default().With(LogKeySystem, LogSystemStore),
+		SnapshotInterval:             DefaultSnapshotInterval,
+		MaintenanceBusyRetryInterval: DefaultMaintenanceBusyRetryInterval,
+		SnapshotRetention:            DefaultSnapshotRetention,
+		L0Retention:                  DefaultL0Retention,
+		L0RetentionCheckInterval:     DefaultL0RetentionCheckInterval,
+		CompactionMonitorEnabled:     true,
+		RetentionEnabled:             true,
+		ShutdownSyncTimeout:          DefaultShutdownSyncTimeout,
+		ShutdownSyncInterval:         DefaultShutdownSyncInterval,
+		HeartbeatCheckInterval:       DefaultHeartbeatCheckInterval,
+		Logger:                       slog.Default().With(LogKeySystem, LogSystemStore),
 	}
 
 	for _, db := range dbs {
@@ -566,6 +587,15 @@ func (s *Store) monitorCompactionLevel(ctx context.Context, lvl *CompactionLevel
 	timer := time.NewTimer(time.Nanosecond)
 	defer timer.Stop()
 
+	// Databases to revisit before the next regular pass: those refused with
+	// ErrMaintenanceBusy and those not yet ready. While any exist the monitor
+	// wakes early and only revisits them, leaving snapshot retention and the
+	// other databases to the regular schedule anchored at nextRegular.
+	var nextRegular time.Time
+	pending := make(map[*DB]struct{})
+	known := make(map[*DB]struct{}) // databases seen on the last regular pass
+	busyStreak := 0                 // consecutive passes that saw a busy database
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -575,33 +605,58 @@ func (s *Store) monitorCompactionLevel(ctx context.Context, lvl *CompactionLevel
 		}
 
 		now := time.Now()
-		nextDelay := time.Until(lvl.NextCompactionAt(now))
+		earlyPass := len(pending) > 0 && now.Before(nextRegular)
+		if !earlyPass {
+			nextRegular = lvl.NextCompactionAt(now)
+			known = make(map[*DB]struct{}) // rebuilt from this regular pass
+		}
 
 		var notReadyDBs []string
+		busySet := make(map[*DB]struct{})
+		notReadySet := make(map[*DB]struct{})
 
 		for _, db := range s.DBs() {
 			if !db.IsOpen() {
 				continue // skip disabled DBs
 			}
+			if earlyPass {
+				// Early pass: only revisit busy or not-ready databases, plus
+				// any registered since the last regular pass so a new database
+				// still gets its initialization retries.
+				_, revisit := pending[db]
+				_, seen := known[db]
+				if !revisit && seen {
+					continue
+				}
+			}
+			known[db] = struct{}{}
 			_, err := s.CompactDB(ctx, db, lvl)
 			switch {
 			case errors.Is(err, ErrNoCompaction), errors.Is(err, ErrCompactionTooEarly):
 				db.Logger.Debug("no compaction", "level", lvl.Level, "path", db.Path())
+			case errors.Is(err, ErrMaintenanceBusy):
+				db.Logger.Debug("maintenance busy, will retry", "level", lvl.Level, "path", db.Path(), "retry", s.MaintenanceBusyRetryInterval)
+				busySet[db] = struct{}{}
 			case errors.Is(err, ErrDBNotReady):
 				db.Logger.Debug("db not ready, skipping", "level", lvl.Level, "path", db.Path(), "error", err)
 				notReadyDBs = append(notReadyDBs, db.Path())
+				notReadySet[db] = struct{}{}
 			case err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded):
 				db.Logger.Error("compaction failed", "level", lvl.Level, "error", err)
 			}
 
-			if lvl.Level == SnapshotLevel {
+			if lvl.Level == SnapshotLevel && !earlyPass {
 				if err := s.EnforceSnapshotRetention(ctx, db); err != nil &&
 					!errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
 					db.Logger.Error("retention enforcement failed", "error", err)
 				}
 			}
 		}
 
+		// Computed after the work so a long pass cannot push the regular
+		// schedule out by its own runtime.
+		nextDelay := time.Until(nextRegular)
+
 		timedOut := !retryDeadline.IsZero() && now.After(retryDeadline)
 		if len(notReadyDBs) > 0 && !timedOut {
 			if retryDeadline.IsZero() {
@@ -618,13 +673,50 @@ func (s *Store) monitorCompactionLevel(ctx context.Context, lvl *CompactionLevel
 					"hint", "database may have corrupted local state or blocked transactions; try removing -litestream directory and restarting")
 			}
 			retryDeadline = time.Time{}
+			notReadySet = nil // give up early retries; regular passes still cover them
 		}
 
-		if nextDelay < 0 {
-			nextDelay = 0
+		pending = make(map[*DB]struct{}, len(busySet)+len(notReadySet))
+		for db := range busySet {
+			pending[db] = struct{}{}
+		}
+		for db := range notReadySet {
+			pending[db] = struct{}{}
 		}
-		timer.Reset(nextDelay)
+
+		if len(busySet) > 0 {
+			busyStreak++
+		} else {
+			busyStreak = 0
+		}
+		timer.Reset(busyRetryDelay(nextDelay, busyStreak, s.MaintenanceBusyRetryInterval))
+	}
+}
+
+// busyRetryDelay returns the delay before the next monitor pass. While a
+// database keeps reporting ErrMaintenanceBusy (busyStreak > 0) the delay is
+// capped at an exponential backoff from maintenanceBusyRetryBase up to
+// maxRetry, so the refused operation is reattempted soon after the in-flight
+// one finishes rather than after the level's full interval (a snapshot losing
+// startup contention to L1 would otherwise wait until the next snapshot
+// boundary), and a momentary collision between levels costs about a second.
+func busyRetryDelay(nextDelay time.Duration, busyStreak int, maxRetry time.Duration) time.Duration {
+	if busyStreak > 0 && maxRetry > 0 {
+		retry := maintenanceBusyRetryBase
+		for i := 1; i < busyStreak && retry < maxRetry; i++ {
+			retry *= 2
+		}
+		if retry > maxRetry {
+			retry = maxRetry
+		}
+		if nextDelay > retry {
+			nextDelay = retry
+		}
+	}
+	if nextDelay < 0 {
+		nextDelay = 0
 	}
+	return nextDelay
 }
 
 func (s *Store) monitorL0Retention(ctx context.Context) {
@@ -772,6 +864,17 @@ func (s *Store) CompactDB(ctx context.Context, db *DB, lvl *CompactionLevel) (*l
 		return nil, &DBNotReadyError{Reason: "page size not initialized"}
 	}
 
+	// Serialize maintenance per database: a snapshot and a compaction (or two
+	// compaction levels) each retain a database-sized page index, so letting
+	// them overlap multiplies peak memory with database size (issue #1477).
+	// Refuse instead of waiting so one database's long snapshot cannot stall
+	// the level monitor for every other database; the caller's next interval
+	// retries.
+	if !db.maintenanceMu.TryLock() {
+		return nil, ErrMaintenanceBusy
+	}
+	defer db.maintenanceMu.Unlock()
+
 	dstLevel := lvl.Level
 
 	// Ensure we are not re-compacting before the most recent compaction time.

store_internal_test.go

@@ -0,0 +1,31 @@
+package litestream
+
+import (
+	"testing"
+	"time"
+)
+
+func TestBusyRetryDelay(t *testing.T) {
+	for _, tt := range []struct {
+		name      string
+		nextDelay time.Duration
+		streak    int
+		maxRetry  time.Duration
+		want      time.Duration
+	}{
+		{"not busy keeps interval", time.Hour, 0, 10 * time.Second, time.Hour},
+		{"first busy pass retries after base", time.Hour, 1, 10 * time.Second, time.Second},
+		{"second busy pass doubles", time.Hour, 2, 10 * time.Second, 2 * time.Second},
+		{"fourth busy pass doubles again", time.Hour, 4, 10 * time.Second, 8 * time.Second},
+		{"backoff is capped", time.Hour, 20, 10 * time.Second, 10 * time.Second},
+		{"busy keeps shorter interval", 500 * time.Millisecond, 3, 10 * time.Second, 500 * time.Millisecond},
+		{"busy with retry disabled", time.Hour, 1, 0, time.Hour},
+		{"negative clamps to zero", -time.Second, 0, 10 * time.Second, 0},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := busyRetryDelay(tt.nextDelay, tt.streak, tt.maxRetry); got != tt.want {
+				t.Fatalf("busyRetryDelay()=%s, want %s", got, tt.want)
+			}
+		})
+	}
+}