From 2c9b0a499fe3d3e95ea630cc4aa454269e1e3845 Mon Sep 17 00:00:00 2001
From: Lukasz Raczylo <lukasz@raczylo.com>
Date: Thu, 13 Aug 2026 04:02:29 +0100
Subject: [PATCH] fix(memory, connection): two data-race/counter defects

1) connection_pool.go: GetConnectionStats read lastRecoveryAttempt (a
multi-word time.Time) without any lock while checkAndRecover wrote it
under recoveryMutex -> data race on live admin API endpoints. Snapshot it
under recoveryMutex (the writer's lock). Regression test
TestGetConnectionStatsRace drives the read path concurrent with the locked
write under -race.

2) cache/memory memory.go: Cache.Get expired path used an unconditional
Delete + atomic decrement, so concurrent Gets on the same expired entry
each decremented entryCount/memoryUsage -> permanent negative drift. Use
LoadAndDelete-with-guard (as evictOldest/evictToFreeMemory already do) so
only the goroutine that actually removes the entry decrements. Regression
test TestMemoryCacheConcurrentGetOnExpiredEntry_CountersConsistent.
---
 cache/memory/memory.go                 | 11 ++++++---
 cache/memory/memory_additional_test.go | 30 ++++++++++++++++++++++++
 connection_pool.go                     |  8 ++++++-
 connection_pool_test.go                | 32 ++++++++++++++++++++++++++
 4 files changed, 77 insertions(+), 4 deletions(-)

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