From b689bb4fdebb5b684b1613d7d62ba950f806af0e Mon Sep 17 00:00:00 2001 From: Cory LaNou Date: Thu, 27 Aug 2026 11:03:15 -0500 Subject: [PATCH 1/2] fix(store): serialize snapshots and compactions per database Store.Open runs one goroutine per compaction level plus one for the snapshot level, and CompactDB did nothing to serialize work per database. On a large database an L9 snapshot and an L1 compaction can therefore run concurrently, and each retains a database-sized LTX page index, multiplying peak memory with database size: a 130 GiB database was OOM-killed in a 12 GiB container this way (#1477). Guard CompactDB with a per-database TryLock. A refused operation returns ErrMaintenanceBusy instead of queueing so that one database's long snapshot cannot stall the level monitor for every other database. The monitor logs the refusal and caps its next delay at MaintenanceBusyRetryInterval (10s) so the refused operation is retried soon after the in-flight one finishes; without that cap a snapshot that lost startup contention to L1 would wait for the next snapshot boundary, up to 24h. Compaction across different databases is unaffected. The snapshot reader's Close now joins the producing goroutine, so a snapshot whose replica write fails early cannot leave its WAL page map and encoder alive after CompactDB releases the lock. This bounds concurrent maintenance memory to a single operation per database. The page indexes themselves still scale with database size; shrinking them is addressed separately in superfly/ltx. Fixes #1477 --- db.go | 53 +++++++++++----- store.go | 137 ++++++++++++++++++++++++++++++++++++----- store_internal_test.go | 31 ++++++++++ store_test.go | 123 ++++++++++++++++++++++++++++++++++++ 4 files changed, 310 insertions(+), 34 deletions(-) create mode 100644 store_internal_test.go diff --git a/db.go b/db.go index 974e2debd..b44242add 100644 --- a/db.go +++ b/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) { diff --git a/store.go b/store.go index 46f742fd1..5b89d5a21 100644 --- a/store.go +++ b/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,26 +605,47 @@ 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) @@ -602,6 +653,10 @@ func (s *Store) monitorCompactionLevel(ctx context.Context, lvl *CompactionLevel } } + // 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. diff --git a/store_internal_test.go b/store_internal_test.go new file mode 100644 index 000000000..6e129c976 --- /dev/null +++ b/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) + } + }) + } +} diff --git a/store_test.go b/store_test.go index 5d8848886..31cbfdf9e 100644 --- a/store_test.go +++ b/store_test.go @@ -2,14 +2,17 @@ package litestream_test import ( "context" + "database/sql" "errors" "fmt" + "io" "path/filepath" "sync" "testing" "time" "github.com/stretchr/testify/require" + "github.com/superfly/ltx" "github.com/benbjohnson/litestream" "github.com/benbjohnson/litestream/file" @@ -578,3 +581,123 @@ func TestStore_SetRetentionEnabled(t *testing.T) { } } } + +func TestStore_CompactDB_SerializesPerDBMaintenance(t *testing.T) { + // Teardown is registered with t.Cleanup (LIFO) so that on failure the + // held snapshot is released and joined before the store and databases + // close underneath it. + db0, sqldb0 := testingutil.MustOpenDBs(t) + t.Cleanup(func() { testingutil.MustCloseDBs(t, db0, sqldb0) }) + + db1, sqldb1 := testingutil.MustOpenDBs(t) + t.Cleanup(func() { testingutil.MustCloseDBs(t, db1, sqldb1) }) + + levels := litestream.CompactionLevels{ + {Level: 0}, + {Level: 1, Interval: 1 * time.Second}, + } + s := litestream.NewStore([]*litestream.DB{db0, db1}, levels) + s.CompactionMonitorEnabled = false + if err := s.Open(t.Context()); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = s.Close(context.Background()) }) + + for _, tt := range []struct { + sqldb *sql.DB + db *litestream.DB + }{{sqldb0, db0}, {sqldb1, db1}} { + if _, err := tt.sqldb.ExecContext(t.Context(), `CREATE TABLE t (id INT);`); err != nil { + t.Fatal(err) + } + if _, err := tt.sqldb.ExecContext(t.Context(), `INSERT INTO t (id) VALUES (100)`); err != nil { + t.Fatal(err) + } else if err := tt.db.Sync(t.Context()); err != nil { + t.Fatal(err) + } else if err := tt.db.Replica.Sync(t.Context()); err != nil { + t.Fatal(err) + } + } + + // Wrap db0's replica client so its snapshot write blocks until released, + // holding db0's maintenance slot the way a long-running snapshot upload does. + started := make(chan struct{}) + release := make(chan struct{}) + var releaseOnce sync.Once + releaseSnapshot := func() { releaseOnce.Do(func() { close(release) }) } + db0.Replica.Client = &blockingSnapshotClient{ + ReplicaClient: db0.Replica.Client, + started: started, + release: release, + } + + snapshotDone := make(chan error, 1) + go func() { + _, err := s.CompactDB(context.Background(), db0, s.SnapshotLevel()) + snapshotDone <- err + }() + t.Cleanup(func() { + releaseSnapshot() + select { + case <-snapshotDone: + case <-time.After(10 * time.Second): + t.Error("snapshot goroutine did not exit after release") + } + }) + + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for snapshot to start") + } + + // A concurrent compaction for the same database must be refused while the + // snapshot is in flight — refused, not queued, so run it with a deadline. + compactDone := make(chan error, 1) + go func() { + _, err := s.CompactDB(t.Context(), db0, levels[1]) + compactDone <- err + }() + select { + case err := <-compactDone: + require.ErrorIs(t, err, litestream.ErrMaintenanceBusy) + case <-time.After(5 * time.Second): + t.Fatal("same-database compaction blocked behind the snapshot instead of returning ErrMaintenanceBusy") + } + + // A different database must not be blocked by db0's snapshot. + _, err := s.CompactDB(t.Context(), db1, levels[1]) + require.NoError(t, err) + + releaseSnapshot() + select { + case err := <-snapshotDone: + snapshotDone <- err // requeue first so cleanup's join sees completion even on failure + require.NoError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for snapshot to finish") + } + + // Once the snapshot finishes, db0 maintenance is available again. + _, err = s.CompactDB(t.Context(), db0, levels[1]) + require.NotErrorIs(t, err, litestream.ErrMaintenanceBusy) +} + +type blockingSnapshotClient struct { + litestream.ReplicaClient + started chan struct{} + release chan struct{} + once sync.Once +} + +func (c *blockingSnapshotClient) WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, r io.Reader) (*ltx.FileInfo, error) { + if level == litestream.SnapshotLevel { + c.once.Do(func() { close(c.started) }) + select { + case <-c.release: + case <-ctx.Done(): + return nil, ctx.Err() + } + } + return c.ReplicaClient.WriteLTXFile(ctx, level, minTXID, maxTXID, r) +} From 10d628d33844484a4edff339bb160e0d45d6c9df Mon Sep 17 00:00:00 2001 From: Cory LaNou Date: Thu, 27 Aug 2026 13:45:06 -0500 Subject: [PATCH 2/2] chore(deps): pin ltx to page-index branch for evaluation Pins github.com/superfly/ltx to the head of superfly/ltx#95 (v0.5.3-0.20260827162011-d457a1ab7844) so this PR builds and soaks with the chunked encoder page index, which is the other half of the #1477 fix. Replace with the tagged ltx release before merging. --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index e0387e15a..528fa67b9 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 7a4014cdf..447ce4f9a 100644 --- a/go.sum +++ b/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=