diff --git a/cache/memory/memory.go b/cache/memory/memory.go index 7e6276b..cab74f4 100644 --- a/cache/memory/memory.go +++ b/cache/memory/memory.go @@ -181,9 +181,14 @@ func (c *Cache) Get(key string) ([]byte, bool) { cacheEntry := entry.(CacheEntry) if cacheEntry.ExpiresAt.Before(time.Now()) { - c.entries.Delete(key) - atomic.AddInt64(&c.entryCount, -1) - atomic.AddInt64(&c.memoryUsage, -cacheEntry.MemorySize) + // Guard the counter decrements with LoadAndDelete so concurrent Gets on + // the same expired entry don't each decrement entryCount/memoryUsage + // (the unconditional Delete here previously double-counted). + if removed, ok := c.entries.LoadAndDelete(key); ok { + removedEntry := removed.(CacheEntry) + atomic.AddInt64(&c.entryCount, -1) + atomic.AddInt64(&c.memoryUsage, -removedEntry.MemorySize) + } return nil, false } diff --git a/cache/memory/memory_additional_test.go b/cache/memory/memory_additional_test.go index f5fb230..e5085e6 100644 --- a/cache/memory/memory_additional_test.go +++ b/cache/memory/memory_additional_test.go @@ -1,6 +1,7 @@ package libpack_cache_memory import ( + "sync" "testing" "time" @@ -88,3 +89,32 @@ func TestMemoryCacheCleanExpiredEntries(t *testing.T) { // Count should be 1 now (only the non-expired entry) assert.Equal(t, int64(1), cache.CountQueries(), "Expected count to be 1 after cleaning expired entries") } + +func TestMemoryCacheConcurrentGetOnExpiredEntry_CountersConsistent(t *testing.T) { + // Regression: concurrent Gets on the same expired entry must decrement + // entryCount exactly once. Previously Get used an unconditional Delete + + // decrement, so N concurrent Gets drove entryCount negative. + cache := NewWithSize(DefaultTestExpiration, DefaultMaxMemorySize, DefaultMaxCacheSize) + + cache.Set("k", []byte("v"), 10*time.Millisecond) + time.Sleep(50 * time.Millisecond) // let it expire + + var wg sync.WaitGroup + for range 32 { + wg.Add(1) + go func() { + defer wg.Done() + if _, ok := cache.Get("k"); ok { + t.Error("expired entry should not be returned") + } + }() + } + wg.Wait() + + if got := cache.CountQueries(); got != 0 { + t.Fatalf("entryCount = %d after concurrent expired gets, want 0", got) + } + if got := cache.GetMemoryUsage(); got != 0 { + t.Fatalf("memoryUsage = %d after concurrent expired gets, want 0", got) + } +} diff --git a/connection_pool.go b/connection_pool.go index 17adb4d..80194e8 100644 --- a/connection_pool.go +++ b/connection_pool.go @@ -259,11 +259,17 @@ func (cpm *ConnectionPoolManager) RecordConnectionFailure() { // GetConnectionStats returns current connection statistics func (cpm *ConnectionPoolManager) GetConnectionStats() map[string]any { + // lastRecoveryAttempt is written by checkAndRecover under recoveryMutex + // (time.Time is not atomic), so snapshot it under the same lock. + cpm.recoveryMutex.Lock() + lastRecovery := cpm.lastRecoveryAttempt + cpm.recoveryMutex.Unlock() + return map[string]any{ "active_connections": cpm.activeConnections.Load(), "total_connections": cpm.totalConnections.Load(), "connection_failures": cpm.connectionFailures.Load(), - "last_recovery_attempt": cpm.lastRecoveryAttempt, + "last_recovery_attempt": lastRecovery, } } diff --git a/connection_pool_test.go b/connection_pool_test.go index 8eb1f59..bfca7ee 100644 --- a/connection_pool_test.go +++ b/connection_pool_test.go @@ -262,6 +262,38 @@ func (suite *ConnectionPoolTestSuite) TestRaceConditions() { wg.Wait() } +func (suite *ConnectionPoolTestSuite) TestGetConnectionStatsRace() { + // GetConnectionStats must be safe to call concurrently with the recovery + // task, which writes lastRecoveryAttempt under recoveryMutex. Regression + // test for the data race between GetConnectionStats reading + // lastRecoveryAttempt and checkAndRecover writing it. + client := &fasthttp.Client{MaxConnsPerHost: 100} + cpm := NewConnectionPoolManager(client) + defer cpm.Shutdown() + + var wg sync.WaitGroup + for range 4 { + wg.Add(1) + go func() { + defer wg.Done() + for range 200 { + _ = cpm.GetConnectionStats() + } + }() + } + // Writer mirrors checkAndRecover's locked write of lastRecoveryAttempt. + wg.Add(1) + go func() { + defer wg.Done() + for range 200 { + cpm.recoveryMutex.Lock() + cpm.lastRecoveryAttempt = time.Now() + cpm.recoveryMutex.Unlock() + } + }() + wg.Wait() +} + func (suite *ConnectionPoolTestSuite) TestCleanupWithNilLogger() { // Test cleanup when cfg or logger is nil origCfg := cfg