cache/stage: add cooldown time for cache (#6435)

Signed-off-by: jiefenghuang <[email protected]>

Author: jiefenghuang

pkg/chunk/disk_cache.go

@@ -99,6 +99,10 @@ type cacheStore struct {
 
 	state     dcState
 	stateLock sync.Mutex
+
+	// newBlockCooldown reduces the initial access time for newly cached staged blocks.
+	// This helps prevent a surge of writes from evicting active read blocks.
+	stagedBlockCooldown time.Duration
 }
 
 func newCacheStore(m *cacheManagerMetrics, dir string, cacheSize, maxItems int64, pendingPages int, config *Config, uploader func(key, path string, force bool) bool) *cacheStore {
@@ -115,22 +119,23 @@ func newCacheStore(m *cacheManagerMetrics, dir string, cacheSize, maxItems int64
 		keyIndex, _ = NewKeyIndex(config)
 	}
 	c := &cacheStore{
-		m:             m,
-		dir:           dir,
-		mode:          config.CacheMode,
-		capacity:      cacheSize,
-		maxItems:      maxItems,
-		maxStageWrite: config.MaxStageWrite,
-		freeRatio:     config.FreeSpace,
-		checksum:      config.CacheChecksum,
-		hashPrefix:    config.HashPrefix,
-		scanInterval:  config.CacheScanInterval,
-		cacheExpire:   config.CacheExpire,
-		keys:          keyIndex,
-		pending:       make(chan pendingFile, pendingPages),
-		pages:         make(map[string]*Page),
-		uploader:      uploader,
-		opTs:          make(map[time.Duration]func() error),
+		m:                   m,
+		dir:                 dir,
+		mode:                config.CacheMode,
+		capacity:            cacheSize,
+		maxItems:            maxItems,
+		maxStageWrite:       config.MaxStageWrite,
+		freeRatio:           config.FreeSpace,
+		checksum:            config.CacheChecksum,
+		hashPrefix:          config.HashPrefix,
+		scanInterval:        config.CacheScanInterval,
+		cacheExpire:         config.CacheExpire,
+		keys:                keyIndex,
+		pending:             make(chan pendingFile, pendingPages),
+		pages:               make(map[string]*Page),
+		uploader:            uploader,
+		opTs:                make(map[time.Duration]func() error),
+		stagedBlockCooldown: config.CacheExpire / 2,
 	}
 	c.stateLock = sync.Mutex{}
 	if config.Writeback {
@@ -736,9 +741,6 @@ func (cache *cacheStore) add(key string, size int32, atime uint32) {
 			cache.used -= int64(iter.size + 4096)
 		}
 		iter.size = size
-		if atime > iter.atime {
-			iter.atime = atime
-		}
 	}
 	cache.keys.add(k, *iter) // add or update
 	if size > 0 {
@@ -769,7 +771,7 @@ func (cache *cacheStore) stage(key string, data []byte) (string, error) {
 			path := cache.cachePath(key)
 			cache.createDir(filepath.Dir(path))
 			if err = os.Link(stagingPath, path); err == nil {
-				cache.add(key, -int32(len(data)), uint32(time.Now().Unix()))
+				cache.add(key, -int32(len(data)), uint32(time.Now().Add(-cache.stagedBlockCooldown).Unix()))
 			} else {
 				logger.Warnf("link %s to %s failed: %s", stagingPath, path, err)
 			}

pkg/chunk/disk_cache_test.go

@@ -591,3 +591,25 @@ func TestLruEviction(t *testing.T) {
 		require.Equal(t, 0, len(le.lruHeap), "LRU heap should be empty by cleanupFull after setting maxItems to 1")
 	})
 }
+
+func TestCooldownAtimeOnWriteFixedOnLoad(t *testing.T) {
+	dir := t.TempDir()
+	conf := defaultConf
+	conf.CacheExpire = time.Hour
+	m := new(cacheManagerMetrics)
+	m.initMetrics()
+	cache := newCacheStore(m, dir, 1<<30, 1000, 1, &conf, nil)
+	key := "0_0_4"
+
+	now := time.Now()
+	path, err := cache.stage(key, []byte("test"))
+	require.NoError(t, err)
+	require.NotEmpty(t, path)
+	require.LessOrEqual(t, cache.keys.peekAtime(cache.getCacheKey(key)), uint32(now.Add(-conf.CacheExpire/2).Unix())) // should have bias in atime
+
+	rc, err := cache.load(key)
+	require.NoError(t, err)
+	require.NotNil(t, rc)
+	defer rc.Close()
+	require.GreaterOrEqual(t, cache.keys.peekAtime(cache.getCacheKey(key)), uint32(now.Unix())) // bias should have been fixed on load
+}