From 4d7438da0c62e16fc4b1c7af1d67f117656ef9ec Mon Sep 17 00:00:00 2001 From: UniversePeak <113168673+UniversePeak@users.noreply.github.com> Date: Fri, 28 Aug 2026 12:12:16 +0800 Subject: [PATCH 1/2] fix: skip unchanged L0 retention checks --- db.go | 80 ++++++++++++++++++++++++++++++++++++++++++++++-------- db_test.go | 80 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ store.go | 3 ++ 3 files changed, 151 insertions(+), 12 deletions(-) diff --git a/db.go b/db.go index 974e2debd..ba13ec6b2 100644 --- a/db.go +++ b/db.go @@ -85,6 +85,20 @@ type DB struct { m map[int]*ltx.FileInfo } + // l0RetentionState tracks the inputs to the last successful retention + // pass. Retention can be skipped while neither the local L0 watermark nor + // the cached L1 watermark has changed. This avoids repeatedly listing the + // replica for idle databases. + l0RetentionState struct { + mu sync.Mutex + initialized bool + l1Checked bool + l0MaxTXID ltx.TXID + l1MaxTXID ltx.TXID + waitingForTime bool + nextCheckAt time.Time + } + // Cached position from the latest L0 LTX file. // nil means cache is invalid; non-nil is the cached position. pos struct { @@ -538,6 +552,10 @@ func (db *DB) ResetLocalState(ctx context.Context) error { db.maxLTXFileInfos.Unlock() db.invalidatePosCache() + db.l0RetentionState.mu.Lock() + db.l0RetentionState.initialized = false + db.l0RetentionState.l1Checked = false + db.l0RetentionState.mu.Unlock() db.Logger.Info("local state reset complete, next sync will create fresh snapshot") return nil @@ -3021,34 +3039,57 @@ func (db *DB) EnforceL0RetentionByTime(ctx context.Context) error { return nil } + // Serialize retention passes and avoid remote LIST requests when the local + // L0 watermark and cached L1 watermark are unchanged. If the previous pass + // stopped at a too-recent file, retry when that file can become eligible. + db.l0RetentionState.mu.Lock() + defer db.l0RetentionState.mu.Unlock() + + _, localMaxTXID, err := db.MaxLTX() + if err != nil { + return fmt.Errorf("fetch local l0 position: %w", err) + } + db.maxLTXFileInfos.Lock() + l1Info, l1Cached := db.maxLTXFileInfos.m[1] + var cachedL1TXID ltx.TXID + if l1Cached && l1Info != nil { + cachedL1TXID = l1Info.MaxTXID + } + db.maxLTXFileInfos.Unlock() + if db.l0RetentionState.initialized && + ((l1Cached && db.l0RetentionState.l1Checked && cachedL1TXID == db.l0RetentionState.l1MaxTXID) || + (!l1Cached && db.l0RetentionState.l1Checked && db.l0RetentionState.l1MaxTXID == 0)) && + localMaxTXID == db.l0RetentionState.l0MaxTXID && + cachedL1TXID == db.l0RetentionState.l1MaxTXID && + (!db.l0RetentionState.waitingForTime || time.Now().Before(db.l0RetentionState.nextCheckAt)) { + return nil + } + db.Logger.Debug("starting l0 retention enforcement", "retention", db.L0Retention) dbName := filepath.Base(db.Path()) // Determine the highest TXID that has been compacted into L1. - itr, err := db.Replica.Client.LTXFiles(ctx, 1, 0, false) + l1InfoValue, err := db.MaxLTXFileInfo(ctx, 1) if err != nil { return fmt.Errorf("fetch l1 files: %w", err) } - var maxL1TXID ltx.TXID - for itr.Next() { - info := itr.Item() - if info.MaxTXID > maxL1TXID { - maxL1TXID = info.MaxTXID - } - } - if err := itr.Close(); err != nil { - return fmt.Errorf("close l1 iterator: %w", err) - } + maxL1TXID := l1InfoValue.MaxTXID if maxL1TXID == 0 { internal.L0RetentionGaugeVec.WithLabelValues(dbName, "eligible").Set(0) internal.L0RetentionGaugeVec.WithLabelValues(dbName, "not_compacted").Set(0) internal.L0RetentionGaugeVec.WithLabelValues(dbName, "too_recent").Set(0) + db.l0RetentionState.initialized = true + db.l0RetentionState.l1Checked = true + db.l0RetentionState.l0MaxTXID = localMaxTXID + db.l0RetentionState.l1MaxTXID = maxL1TXID + db.l0RetentionState.waitingForTime = false + db.l0RetentionState.nextCheckAt = time.Time{} return nil } threshold := time.Now().Add(-db.L0Retention) - itr, err = db.Replica.Client.LTXFiles(ctx, 0, 0, false) + itr, err := db.Replica.Client.LTXFiles(ctx, 0, 0, false) if err != nil { return fmt.Errorf("fetch l0 files: %w", err) } @@ -3061,6 +3102,7 @@ func (db *DB) EnforceL0RetentionByTime(ctx context.Context) error { totalFiles int notCompactedCount int tooRecentCount int + nextCheckAt time.Time ) for itr.Next() { info := itr.Item() @@ -3081,6 +3123,7 @@ func (db *DB) EnforceL0RetentionByTime(ctx context.Context) error { // create gaps between retained files. VFS expects contiguous coverage. processedAll = false tooRecentCount++ + nextCheckAt = createdAt.Add(db.L0Retention) break } @@ -3115,6 +3158,12 @@ func (db *DB) EnforceL0RetentionByTime(ctx context.Context) error { "max_l1_txid", maxL1TXID) if len(deleted) == 0 { + db.l0RetentionState.initialized = true + db.l0RetentionState.l1Checked = true + db.l0RetentionState.l0MaxTXID = localMaxTXID + db.l0RetentionState.l1MaxTXID = maxL1TXID + db.l0RetentionState.waitingForTime = !nextCheckAt.IsZero() + db.l0RetentionState.nextCheckAt = nextCheckAt return nil } @@ -3135,6 +3184,13 @@ func (db *DB) EnforceL0RetentionByTime(ctx context.Context) error { db.invalidatePosCache() } + db.l0RetentionState.initialized = true + db.l0RetentionState.l1Checked = true + db.l0RetentionState.l0MaxTXID = localMaxTXID + db.l0RetentionState.l1MaxTXID = maxL1TXID + db.l0RetentionState.waitingForTime = !nextCheckAt.IsZero() + db.l0RetentionState.nextCheckAt = nextCheckAt + db.Logger.Info("l0 retention enforced", "deleted_count", len(deleted), "max_l1_txid", maxL1TXID) return nil diff --git a/db_test.go b/db_test.go index 56675cfcb..40c4b279f 100644 --- a/db_test.go +++ b/db_test.go @@ -41,6 +41,86 @@ func (c *snapshotCountingClient) writeCount() int { return c.n } +type l0RetentionCountingClient struct { + litestream.ReplicaClient + mu sync.Mutex + counts map[int]int +} + +func (c *l0RetentionCountingClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) { + c.mu.Lock() + c.counts[level]++ + c.mu.Unlock() + return c.ReplicaClient.LTXFiles(ctx, level, seek, useMetadata) +} + +func (c *l0RetentionCountingClient) count(level int) int { + c.mu.Lock() + defer c.mu.Unlock() + return c.counts[level] +} + +func TestDB_EnforceL0RetentionByTime_SkipsUnchangedDatabase(t *testing.T) { + db := testingutil.NewDB(t, filepath.Join(t.TempDir(), "db")) + db.Replica = litestream.NewReplica(db) + + client := &l0RetentionCountingClient{ + ReplicaClient: &mock.ReplicaClient{ + LTXFilesFunc: func(_ context.Context, level int, _ ltx.TXID, _ bool) (ltx.FileIterator, error) { + return ltx.NewFileInfoSliceIterator([]*ltx.FileInfo{{ + Level: level, + MinTXID: 1, + MaxTXID: 1, + CreatedAt: time.Now().Add(-time.Hour), + }}), nil + }, + }, + counts: make(map[int]int), + } + db.Replica.Client = client + db.L0Retention = 24 * time.Hour + + if err := os.MkdirAll(db.LTXLevelDir(0), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(db.LTXPath(0, 1, 1), nil, 0o644); err != nil { + t.Fatal(err) + } + + beforeL1 := client.count(1) + beforeL0 := client.count(0) + if err := db.EnforceL0RetentionByTime(t.Context()); err != nil { + t.Fatal(err) + } + firstL1 := client.count(1) + firstL0 := client.count(0) + if err := db.EnforceL0RetentionByTime(t.Context()); err != nil { + t.Fatal(err) + } + if got, want := client.count(1)-beforeL1, 1; got != want { + t.Fatalf("L1 LIST count after two unchanged passes=%d, want %d", got, want) + } + if got, want := client.count(0)-beforeL0, firstL0-beforeL0; got != want { + t.Fatalf("L0 LIST count after unchanged second pass=%d, want %d", got, want) + } + if firstL1-beforeL1 != 1 { + t.Fatalf("first pass L1 LIST count=%d, want 1", firstL1-beforeL1) + } + + if err := os.WriteFile(db.LTXPath(0, 2, 2), nil, 0o644); err != nil { + t.Fatal(err) + } + if err := db.EnforceL0RetentionByTime(t.Context()); err != nil { + t.Fatal(err) + } + if got, want := client.count(1)-beforeL1, 1; got != want { + t.Fatalf("L1 LIST count after new L0 file=%d, want %d", got, want) + } + if got, want := client.count(0)-beforeL0, firstL0-beforeL0+1; got != want { + t.Fatalf("L0 LIST count after new L0 file=%d, want %d", got, want) + } +} + func TestDB_Path(t *testing.T) { db := testingutil.NewDB(t, "/tmp/db") if got, want := db.Path(), `/tmp/db`; got != want { diff --git a/store.go b/store.go index 46f742fd1..ebf235bc3 100644 --- a/store.go +++ b/store.go @@ -493,6 +493,9 @@ func (s *Store) SetL0Retention(d time.Duration) { s.L0Retention = d for _, db := range s.dbs { db.L0Retention = d + db.l0RetentionState.mu.Lock() + db.l0RetentionState.initialized = false + db.l0RetentionState.mu.Unlock() } } From d9446b6b33cceddfefe05ef776519140e2ee4039 Mon Sep 17 00:00:00 2001 From: UniversePeak Date: Sat, 29 Aug 2026 09:19:42 +0800 Subject: [PATCH 2/2] fix: invalidate retention cache when interval changes --- db.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/db.go b/db.go index ba13ec6b2..4abd1b3ba 100644 --- a/db.go +++ b/db.go @@ -95,6 +95,7 @@ type DB struct { l1Checked bool l0MaxTXID ltx.TXID l1MaxTXID ltx.TXID + retention time.Duration waitingForTime bool nextCheckAt time.Time } @@ -3061,6 +3062,7 @@ func (db *DB) EnforceL0RetentionByTime(ctx context.Context) error { (!l1Cached && db.l0RetentionState.l1Checked && db.l0RetentionState.l1MaxTXID == 0)) && localMaxTXID == db.l0RetentionState.l0MaxTXID && cachedL1TXID == db.l0RetentionState.l1MaxTXID && + db.l0RetentionState.retention == db.L0Retention && (!db.l0RetentionState.waitingForTime || time.Now().Before(db.l0RetentionState.nextCheckAt)) { return nil } @@ -3083,6 +3085,7 @@ func (db *DB) EnforceL0RetentionByTime(ctx context.Context) error { db.l0RetentionState.l1Checked = true db.l0RetentionState.l0MaxTXID = localMaxTXID db.l0RetentionState.l1MaxTXID = maxL1TXID + db.l0RetentionState.retention = db.L0Retention db.l0RetentionState.waitingForTime = false db.l0RetentionState.nextCheckAt = time.Time{} return nil @@ -3162,6 +3165,7 @@ func (db *DB) EnforceL0RetentionByTime(ctx context.Context) error { db.l0RetentionState.l1Checked = true db.l0RetentionState.l0MaxTXID = localMaxTXID db.l0RetentionState.l1MaxTXID = maxL1TXID + db.l0RetentionState.retention = db.L0Retention db.l0RetentionState.waitingForTime = !nextCheckAt.IsZero() db.l0RetentionState.nextCheckAt = nextCheckAt return nil @@ -3188,6 +3192,7 @@ func (db *DB) EnforceL0RetentionByTime(ctx context.Context) error { db.l0RetentionState.l1Checked = true db.l0RetentionState.l0MaxTXID = localMaxTXID db.l0RetentionState.l1MaxTXID = maxL1TXID + db.l0RetentionState.retention = db.L0Retention db.l0RetentionState.waitingForTime = !nextCheckAt.IsZero() db.l0RetentionState.nextCheckAt = nextCheckAt