Revert "wip: reap in progress guard"

This reverts commit dfc33cc.

Signed-off-by: Chris Randles <randles.chris@gmail.com>

Author: crandles

pkg/cache/index/client.go

@@ -125,16 +125,15 @@ type IndexedClient struct {
 	LastFlush atomicx.Time `msg:"LastFlush,extension"`
 
 	// internal index configuration
-	name           string               `msg:"-"`
-	cacheProvider  string               `msg:"-"`
-	options        atomic.Value         `msg:"-"`
-	ico            IndexedClientOptions `msg:"-"`
-	lastWrite      atomicx.Time         `msg:"-"`
-	isClosing      atomic.Bool
-	cancel         context.CancelFunc
-	flusherExited  atomic.Bool
-	reaperExited   atomic.Bool
-	reapInProgress atomic.Bool
+	name          string               `msg:"-"`
+	cacheProvider string               `msg:"-"`
+	options       atomic.Value         `msg:"-"`
+	ico           IndexedClientOptions `msg:"-"`
+	lastWrite     atomicx.Time         `msg:"-"`
+	isClosing     atomic.Bool
+	cancel        context.CancelFunc
+	flusherExited atomic.Bool
+	reaperExited  atomic.Bool
 
 	// used only in tests: fields to interact with client goroutines
 	forceFlush chan bool
@@ -185,23 +184,6 @@ func (idx *IndexedClient) updateIndex(cacheKey string, size int64, la, lw, e tim
 	metrics.ObserveCacheSizeChange(idx.name, idx.cacheProvider, cacheSize, count)
 	idx.lastWrite.Store(time.Now())
 	idx.Objects.Store(cacheKey, obj)
-
-	// Inline eviction guard: trigger async reaping if approaching max size
-	// to prevent large overshoot between periodic reaper cycles.
-	// Only trigger if reaper goroutine is running (ReapInterval > 0) and
-	// no reap is already in progress.
-	opts := idx.options.Load().(*options.Options)
-	if opts.ReapInterval > 0 && opts.MaxSizeBytes > 0 && cacheSize > 0 && !idx.reapInProgress.Load() {
-		threshold := int64(float64(opts.MaxSizeBytes) * 0.9) // 90% threshold
-		if cacheSize > threshold {
-			select {
-			case idx.forceReap <- true:
-				// successfully triggered async reap
-			default:
-				// reap already in progress or channel full, skip
-			}
-		}
-	}
 }
 
 func (idx *IndexedClient) StoreReference(cacheKey string, data cache.ReferenceObject, ttl time.Duration) error {
@@ -425,12 +407,6 @@ REAPER:
 // reap makes a single iteration through the cache index to to find and remove expired elements
 // and evict least-recently-accessed elements to maintain the Maximum allowed Cache Size
 func (idx *IndexedClient) reap() {
-	// Set flag to prevent concurrent reaps
-	if !idx.reapInProgress.CompareAndSwap(false, true) {
-		return // reap already in progress, skip
-	}
-	defer idx.reapInProgress.Store(false)
-
 	cacheSize := max(atomic.LoadInt64(&idx.CacheSize), 0)
 	removals := make([]string, 0)
 	remainders := make(objectsAtime, 0, cacheSize)

pkg/encoding/zstd/zstd_test.go

@@ -19,7 +19,6 @@ package zstd
 import (
 	"bytes"
 	"net/http/httptest"
-	"sync"
 	"testing"
 )
 
@@ -76,83 +75,3 @@ func TestNewEncoder(t *testing.T) {
 		t.Error("expected non-nil encoder")
 	}
 }
-
-// TestConcurrentEncodeDecode verifies that the shared commonEncoder and
-// commonDecoder are safe for concurrent use via EncodeAll/DecodeAll.
-// If this test fails or produces data corruption, the shared instances
-// must be replaced with sync.Pool.
-func TestConcurrentEncodeDecode(t *testing.T) {
-	// Create test data with varying patterns to detect corruption
-	testData := [][]byte{
-		bytes.Repeat([]byte("test data 1"), 100),
-		bytes.Repeat([]byte("different pattern 2"), 150),
-		bytes.Repeat([]byte("yet another test 3"), 200),
-		bytes.Repeat([]byte("final test pattern 4"), 250),
-	}
-
-	const goroutines = 100
-	const iterations = 10
-
-	var wg sync.WaitGroup
-	errors := make(chan error, goroutines*iterations*2)
-
-	// Launch concurrent encode/decode operations
-	for i := 0; i < goroutines; i++ {
-		dataIndex := i % len(testData)
-		testBytes := testData[dataIndex]
-
-		wg.Add(2)
-
-		// Concurrent encode
-		go func(data []byte, idx int) {
-			defer wg.Done()
-			for j := 0; j < iterations; j++ {
-				encoded, err := Encode(data)
-				if err != nil {
-					errors <- err
-					return
-				}
-				// Verify we can decode what we just encoded
-				decoded, err := Decode(encoded)
-				if err != nil {
-					errors <- err
-					return
-				}
-				if !bytes.Equal(decoded, data) {
-					t.Errorf("data corruption: goroutine %d iteration %d", idx, j)
-					return
-				}
-			}
-		}(testBytes, i)
-
-		// Concurrent decode (of pre-encoded data)
-		go func(data []byte, idx int) {
-			defer wg.Done()
-			// Pre-encode the data
-			encoded, err := Encode(data)
-			if err != nil {
-				errors <- err
-				return
-			}
-			for j := 0; j < iterations; j++ {
-				decoded, err := Decode(encoded)
-				if err != nil {
-					errors <- err
-					return
-				}
-				if !bytes.Equal(decoded, data) {
-					t.Errorf("data corruption: goroutine %d iteration %d", idx, j)
-					return
-				}
-			}
-		}(testBytes, i)
-	}
-
-	wg.Wait()
-	close(errors)
-
-	// Check for any errors
-	for err := range errors {
-		t.Error(err)
-	}
-}