fix race

Author: giortzisg

internal/telemetry/buffer.go

@@ -8,7 +8,7 @@ import (
 
 const defaultCapacity = 100
 
-// Buffer is a thread-safe ring buffer with overflow policies
+// Buffer is a thread-safe ring buffer with overflow policies.
 type Buffer[T any] struct {
 	mu       sync.RWMutex
 	items    []T
@@ -46,7 +46,7 @@ func (b *Buffer[T]) SetDroppedCallback(callback func(item T, reason string)) {
 	b.onDropped = callback
 }
 
-// Offer adds an item to the buffer, returns false if dropped due to overflow
+// Offer adds an item to the buffer, returns false if dropped due to overflow.
 func (b *Buffer[T]) Offer(item T) bool {
 	atomic.AddInt64(&b.offered, 1)
 
@@ -89,7 +89,7 @@ func (b *Buffer[T]) Offer(item T) bool {
 	}
 }
 
-// Poll removes and returns the oldest item, false if empty
+// Poll removes and returns the oldest item, false if empty.
 func (b *Buffer[T]) Poll() (T, bool) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
@@ -251,14 +251,15 @@ func (b *Buffer[T]) Clear() {
 func (b *Buffer[T]) GetMetrics() BufferMetrics {
 	b.mu.RLock()
 	size := b.size
+	util := float64(b.size) / float64(b.capacity)
 	b.mu.RUnlock()
 
 	return BufferMetrics{
 		Category:      b.category,
 		Priority:      b.priority,
 		Capacity:      b.capacity,
 		Size:          size,
-		Utilization:   b.Utilization(),
+		Utilization:   util,
 		OfferedCount:  b.OfferedCount(),
 		DroppedCount:  b.DroppedCount(),
 		AcceptedCount: b.AcceptedCount(),

internal/telemetry/buffer_test.go

@@ -2,6 +2,7 @@ package telemetry
 
 import (
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 )
@@ -348,7 +349,7 @@ func TestBufferStressTest(t *testing.T) {
 
 	// Start consumers
 	wg.Add(numConsumers)
-	consumedCount := int64(0)
+	var consumedCount int64
 	for i := 0; i < numConsumers; i++ {
 		go func() {
 			defer wg.Done()
@@ -361,13 +362,13 @@ func TestBufferStressTest(t *testing.T) {
 						if !ok {
 							break
 						}
-						consumedCount++
+						atomic.AddInt64(&consumedCount, 1)
 					}
 					return
 				default:
 					_, ok := buffer.Poll()
 					if ok {
-						consumedCount++
+						atomic.AddInt64(&consumedCount, 1)
 					}
 				}
 			}
@@ -380,13 +381,13 @@ func TestBufferStressTest(t *testing.T) {
 	wg.Wait()
 
 	t.Logf("Stress test results: offered=%d, dropped=%d, consumed=%d",
-		buffer.OfferedCount(), buffer.DroppedCount(), consumedCount)
+		buffer.OfferedCount(), buffer.DroppedCount(), atomic.LoadInt64(&consumedCount))
 
 	// Basic sanity checks
 	if buffer.OfferedCount() <= 0 {
 		t.Error("Expected some items to be offered")
 	}
-	if consumedCount <= 0 {
+	if atomic.LoadInt64(&consumedCount) <= 0 {
 		t.Error("Expected some items to be consumed")
 	}
 }