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/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= 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) +}