mirror of
https://github.com/lukaszraczylo/graphql-monitoring-proxy.git
synced 2026-10-11 05:27:47 +00:00
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.
This commit is contained in:
Vendored
+8
-3
@@ -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
|
||||
}
|
||||
|
||||
|
||||
+30
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
+7
-1
@@ -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,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user