From 5f4ed5ac3aed0c61d5f669cf1e3b70d57cfaea13 Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Thu, 8 Oct 2026 05:57:00 +0000 Subject: [PATCH 1/2] test: reject mismatched LRU access buckets Signed-off-by: Perfloop Agent --- storage/access_updater_test.go | 214 +++++++++++++++++++++++++++++++++ 1 file changed, 214 insertions(+) diff --git a/storage/access_updater_test.go b/storage/access_updater_test.go index ce37f80b..4876f764 100644 --- a/storage/access_updater_test.go +++ b/storage/access_updater_test.go @@ -385,6 +385,220 @@ func TestAccessUpdater_ReadBumpOutranksEarlierWrites(t *testing.T) { "a read bump must sort after a key written before the read, so eviction reaches it later:\n bumped: %s\n written: %s", bumped, written) } +func TestAccessUpdater_DelayedReadDoesNotUndoNewerOverwrite(t *testing.T) { + const ( + targetKey = "overwritten" + middleKey = "middle" + laterKey = "later" + ) + value := bytes.Repeat([]byte("x"), 128) + middleValue := bytes.Repeat([]byte("m"), len(value)) + + s, err := NewStorageWithConfig(&StorageConfig{ + DiskPath: t.TempDir(), + InlineThreshold: 1024, + CompactThreshold: 4096, + SegmentSize: 16 * 1024 * 1024, + FdCacheSize: 1000, + MaxDiskUsage: 1 << 20, + EvictionPolicy: EvictionPolicyLRU, + CleanupInterval: time.Hour, + }) + require.NoError(t, err) + defer s.Close() + require.NotNil(t, s.accessUpdater, "LRU with a disk cap must run the access updater") + + // Hold the automatic notification admitted by Get until after the overwrite. + // The replacement worker's long interval prevents a timer flush while the + // notification is held; Flush below still exercises the worker path. + s.accessUpdater.Stop() + updater := newAccessUpdater(s, 16, time.Hour, DefaultAccessUpdateDelay) + s.accessUpdater = updater + + require.NoError(t, s.Put(targetKey, bytes.NewReader(value), 0)) + require.NoError(t, s.Put(middleKey, bytes.NewReader(middleValue), 0)) + require.NoError(t, s.Put(laterKey, bytes.NewReader([]byte("l")), 0)) + + reader, found, err := s.Get(targetKey, 0, 0) + require.NoError(t, err) + require.True(t, found, "the initial read must find the inline value") + got, err := io.ReadAll(reader) + require.NoError(t, err) + require.Equal(t, value, got) + + var readUpdate accessUpdate + select { + case readUpdate = <-updater.updates: + case <-time.After(time.Second): + t.Fatal("Get did not admit an automatic access update") + } + require.Equal(t, targetKey, readUpdate.key) + + // Put a live, equal-sized key between the read time and the overwrite time. + middleTime := time.Unix(readUpdate.time.Unix()+1, 0) + require.True(t, readUpdate.time.Before(middleTime)) + s.SetAccessTime(middleKey, middleTime.Unix()) + updater.Start() + require.Equal(t, 1, updater.Flush(), "the explicit middle timestamp should be persisted") + + indexTime := func(key string) time.Time { + entry, ok := readAccessIndex(t, s, key) + require.True(t, ok, "%s should have a current LRU entry", key) + indexedKey, accessTime, err := keys.ParseBucketedAccessKey([]byte(entry)) + require.NoError(t, err) + require.Equal(t, key, indexedKey) + return accessTime + } + keyExists := func(key string) bool { + ro := grocksdb.NewDefaultReadOptions() + defer ro.Destroy() + slice, err := s.meta.Handle().Get(ro, keys.MakeMetadataKey(key)) + require.NoError(t, err) + defer slice.Free() + return slice.Exists() + } + indexEntryExists := func(entry string) bool { + ro := grocksdb.NewDefaultReadOptions() + defer ro.Destroy() + slice, err := s.meta.Handle().Get(ro, []byte(entry)) + require.NoError(t, err) + defer slice.Free() + return slice.Exists() + } + + require.True(t, indexTime(middleKey).Equal(middleTime)) + require.Eventually(t, func() bool { + return time.Now().Unix() > middleTime.Unix() + }, 5*time.Second, time.Millisecond, "the overwrite must use a timestamp in a later second") + require.NoError(t, s.Put(targetKey, bytes.NewReader(value), 0)) + writeEntry, writeEntryExists := readAccessIndex(t, s, targetKey) + require.True(t, writeEntryExists) + writeTime := indexTime(targetKey) + require.True(t, middleTime.Before(writeTime)) + require.Greater(t, writeTime.Unix(), readUpdate.time.Unix(), "read and overwrite timestamps must differ by seconds") + + // A later explicit index update shares the flush batch. Its persisted time + // proves the batch write succeeded and puts it after both eviction candidates. + laterTime := time.Unix(writeTime.Unix()+1, 0) + require.True(t, writeTime.Before(laterTime)) + updater.updates <- readUpdate + s.SetAccessTime(laterKey, laterTime.Unix()) + updater.Flush() + require.True(t, indexTime(laterKey).Equal(laterTime), + "the later index update must prove the flush batch committed") + currentEntry, currentIndexExists := readAccessIndex(t, s, targetKey) + assert.True(t, currentIndexExists && currentEntry == writeEntry && indexEntryExists(currentEntry), + "automatic LRU update must not undo recency established by a completed overwrite") + + // Walk the ordinary LRU index for one value's bytes. The old read time sorts + // before middleTime on the broken implementation, so it evicts targetKey. + evicted := s.cleaner.evictByIndex(lruEvictionIndex(), int64(len(value))) + evictionHolds := evicted == 1 && keyExists(targetKey) && !keyExists(middleKey) + require.True(t, evictionHolds, + "ordinary LRU eviction should remove the intermediate key and retain the overwritten key") + + // A later admitted automatic read still advances the persisted timestamp. + updater.accessTimeLRU.Remove(targetKey) + reader, found, err = s.Get(targetKey, 0, 0) + require.NoError(t, err) + require.True(t, found, "the overwritten value must survive the LRU eviction") + got, err = io.ReadAll(reader) + require.NoError(t, err) + require.Equal(t, value, got) + require.Equal(t, 1, updater.Flush()) + require.True(t, indexTime(targetKey).After(writeTime), "a newer automatic read should advance recency") + + // An explicit SetAccessTime is not an automatic read and may deliberately + // move the index to an older timestamp. Remove only the throttle cache entry + // so this control tests flush semantics rather than the read-delay gate. + updater.accessTimeLRU.Remove(targetKey) + explicitTime := time.Unix(readUpdate.time.Unix()-60, 0) + s.SetAccessTime(targetKey, explicitTime.Unix()) + require.Equal(t, 1, updater.Flush()) + require.True(t, indexTime(targetKey).Equal(explicitTime), "explicit SetAccessTime may set an older timestamp") +} + +func TestAccessUpdater_RewritesMalformedCurrentIndex(t *testing.T) { + prefixLen := len(keys.AccessBucketPrefix) + bucketEnd := prefixLen + len(keys.AccessBucketFormat) + timestampStart := bucketEnd + 1 + cases := []struct { + name string + mutate func([]byte) + }{ + {name: "prefix", mutate: func(entry []byte) { entry[0] = '?' }}, + {name: "bucket byte", mutate: func(entry []byte) { entry[prefixLen] = 'X' }}, + {name: "invalid bucket date", mutate: func(entry []byte) { copy(entry[prefixLen:bucketEnd], "2023023000") }}, + {name: "bucket timestamp mismatch", mutate: func(entry []byte) { + lastBucketDigit := bucketEnd - 1 + if entry[lastBucketDigit] == '0' { + entry[lastBucketDigit] = '1' + } else { + entry[lastBucketDigit] = '0' + } + }}, + {name: "timestamp separator", mutate: func(entry []byte) { entry[timestampStart+19] = '?' }}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + storage, cleanup := createTestStorage(t, 3600, 1024, 4096, 16*1024*1024, 1000, 1<<20) + defer cleanup() + require.NotNil(t, storage.accessUpdater) + + storage.accessUpdater.Stop() + updater := newAccessUpdater(storage, 16, time.Hour, DefaultAccessUpdateDelay) + storage.accessUpdater = updater + + const key = "malformed" + value := []byte("value") + require.NoError(t, storage.Put(key, bytes.NewReader(value), 0)) + + futureTime := time.Now().Add(time.Hour) + malformedEntry := keys.MakeBucketedAccessKey(key, futureTime) + tc.mutate(malformedEntry) + wo := grocksdb.NewDefaultWriteOptions() + defer wo.Destroy() + require.NoError(t, storage.meta.Handle().Put(wo, malformedEntry, []byte{})) + require.NoError(t, storage.meta.Handle().Put(wo, keys.MakeBucketedAccessIndexKey(key), malformedEntry)) + + reader, found, err := storage.Get(key, 0, 0) + require.NoError(t, err) + require.True(t, found) + got, err := io.ReadAll(reader) + require.NoError(t, err) + require.Equal(t, value, got) + + var update accessUpdate + select { + case update = <-updater.updates: + case <-time.After(time.Second): + t.Fatal("Get did not admit an automatic access update") + } + require.Equal(t, key, update.key) + require.True(t, futureTime.After(update.time)) + updater.updates <- update + updater.Start() + require.Equal(t, 1, updater.Flush()) + currentEntry, ok := readAccessIndex(t, storage, key) + require.True(t, ok) + require.NotEqual(t, string(malformedEntry), currentEntry, + "malformed current entries must follow the existing refresh path") + indexedKey, accessTime, err := keys.ParseBucketedAccessKey([]byte(currentEntry)) + require.NoError(t, err) + require.Equal(t, key, indexedKey) + require.True(t, accessTime.Equal(update.time)) + + ro := grocksdb.NewDefaultReadOptions() + defer ro.Destroy() + oldEntry, err := storage.meta.Handle().Get(ro, malformedEntry) + require.NoError(t, err) + defer oldEntry.Free() + require.False(t, oldEntry.Exists(), "the malformed ordered row should be replaced") + }) + } +} + func benchmarkStorageGet(b *testing.B, storage *Storage, key string) { b.Helper() From 8b49514db43a4043f4c972639f727acac473420d Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Thu, 8 Oct 2026 05:58:10 +0000 Subject: [PATCH 2/2] fix: keep automatic LRU reads from undoing writes Signed-off-by: Perfloop Agent --- storage/access_updater.go | 59 ++++++++++++++++++++++++++++++++------- 1 file changed, 49 insertions(+), 10 deletions(-) diff --git a/storage/access_updater.go b/storage/access_updater.go index 12904d5a..30578ef7 100644 --- a/storage/access_updater.go +++ b/storage/access_updater.go @@ -20,8 +20,9 @@ import ( // the second would sort BEFORE keys written later in the same second and lose // to them at eviction (issue #257). type accessUpdate struct { - key string - time time.Time + key string + time time.Time + automatic bool // automatic reads must not replace a newer persisted recency } // accessUpdater handles asynchronous batched updates of access times for LRU tracking @@ -76,10 +77,19 @@ func (a *accessUpdater) Stop() { a.wg.Wait() } -// Update queues an access time update (non-blocking) +// Update queues an explicit access time (non-blocking). func (a *accessUpdater) Update(key string, accessTime time.Time) { + a.enqueue(accessUpdate{key: key, time: accessTime}) +} + +// UpdateNow queues an automatic access time update with current time (non-blocking). +func (a *accessUpdater) UpdateNow(key string) { + a.enqueue(accessUpdate{key: key, time: time.Now(), automatic: true}) +} + +func (a *accessUpdater) enqueue(update accessUpdate) { select { - case a.updates <- accessUpdate{key: key, time: accessTime}: + case a.updates <- update: // Update queued successfully metrics.LRUAccessUpdates.Inc() default: @@ -88,11 +98,6 @@ func (a *accessUpdater) Update(key string, accessTime time.Time) { } } -// UpdateNow queues an access time update with current time (non-blocking) -func (a *accessUpdater) UpdateNow(key string) { - a.Update(key, time.Now()) -} - // Flush forces all pending updates to be written to RocksDB immediately. // It synchronizes with the background goroutine to ensure that all updates // currently buffered in the `updates` channel are processed and flushed in a @@ -178,6 +183,31 @@ func (a *accessUpdater) timeGateUpdate(update accessUpdate) { } } +// parseWellFormedBucketedAccessKey validates the full stored key format. +func parseWellFormedBucketedAccessKey(bucketedKey []byte) (string, time.Time, bool) { + prefixLen := len(keys.AccessBucketPrefix) + bucketEnd := prefixLen + len(keys.AccessBucketFormat) + timestampStart := bucketEnd + 1 + timestampEnd := timestampStart + 19 + if len(bucketedKey) < timestampEnd+1 || string(bucketedKey[:prefixLen]) != keys.AccessBucketPrefix || + bucketedKey[bucketEnd] != '/' || bucketedKey[timestampEnd] != '/' { + return "", time.Time{}, false + } + for i, digit := range bucketedKey[timestampStart:timestampEnd] { + if i == 0 && digit == '-' { + continue + } + if digit < '0' || digit > '9' { + return "", time.Time{}, false + } + } + key, accessTime, err := keys.ParseBucketedAccessKey(bucketedKey) + if err != nil || string(bucketedKey[:timestampStart]) != keys.GetBucketedAccessKey(accessTime) { + return "", time.Time{}, false + } + return key, accessTime, true +} + // flushBatch writes a batch of access updates to RocksDB and returns its size. func (a *accessUpdater) flushBatch() int { a.batchMutex.Lock() @@ -204,8 +234,17 @@ func (a *accessUpdater) flushBatch() int { bucketIndexKey := keys.MakeBucketedAccessIndexKey(key) slice, err := a.storage.meta.Handle().Get(ro, bucketIndexKey) if err == nil && slice.Exists() { - // Delete the old bucketed entry oldBucketKey := slice.Data() + if update.automatic { + indexedKey, currentTime, wellFormed := parseWellFormedBucketedAccessKey(oldBucketKey) + if wellFormed && indexedKey == key && currentTime.After(update.time) { + // A completed write or newer read already published recency after this read. + slice.Free() + continue + } + } + + // Delete the old bucketed entry writeBatch.Delete(oldBucketKey) slice.Free() }