diff --git a/internal/datastore/proxy/schemacaching/standardcache.go b/internal/datastore/proxy/schemacaching/standardcache.go index b5a51d9a5..c862b5ce9 100644 --- a/internal/datastore/proxy/schemacaching/standardcache.go +++ b/internal/datastore/proxy/schemacaching/standardcache.go @@ -168,10 +168,6 @@ func listAndCache[T schemaDefinition]( entry := &cacheEntry{def.Definition, def.LastWrittenRevision, estimatedDefinitionSize, err} r.p.c.Set(cache.StringKey(cacheRevisionKey), entry, entry.Size()) } - - // We have to call wait here or else Ristretto may not have the key(s) - // available to a subsequent caller. - r.p.c.Wait() } return foundDefs, nil @@ -203,9 +199,6 @@ func readAndCache[T schemaDefinition]( entry := &cacheEntry{loaded, updatedRev, estimatedDefinitionSize, err} r.p.c.Set(cache.StringKey(cacheRevisionKey), entry, entry.Size()) - // We have to call wait here or else Ristretto may not have the key - // available to a subsequent caller. - r.p.c.Wait() return entry, nil }) if err != nil { diff --git a/pkg/cache/cache.go b/pkg/cache/cache.go index 7977a48c3..c489f9609 100644 --- a/pkg/cache/cache.go +++ b/pkg/cache/cache.go @@ -75,13 +75,6 @@ type Cache[K KeyString, V any] interface { // implementation, so writes are best-effort. Set(key K, entry V, cost int64) bool - // Wait blocks until buffered Set calls have been processed by the - // underlying implementation. Required for read-your-own-writes - // semantics with implementations that buffer writes (e.g. Ristretto); - // a no-op on the current Otter-backed implementation, which applies - // writes synchronously. - Wait() - // Close stops the cache's background workers (if any) and tears down // associated metrics registration, if one was set up. Close() @@ -118,7 +111,6 @@ var _ Cache[StringKey, any] = (*noopCache[StringKey, any])(nil) func (no *noopCache[K, V]) Get(_ K) (V, bool) { return *new(V), false } func (no *noopCache[K, V]) GetTTL() time.Duration { return time.Duration(0) } func (no *noopCache[K, V]) Set(_ K, _ V, _ int64) bool { return false } -func (no *noopCache[K, V]) Wait() {} func (no *noopCache[K, V]) Close() {} func (no *noopCache[K, V]) GetMetrics() Metrics { return &noopMetrics{} } func (no *noopCache[K, V]) MarshalZerologObject(e *zerolog.Event) { diff --git a/pkg/cache/cache_otter.go b/pkg/cache/cache_otter.go index fc7180a23..b52cc1f48 100644 --- a/pkg/cache/cache_otter.go +++ b/pkg/cache/cache_otter.go @@ -189,7 +189,6 @@ func (wtc *otterCache[K, V]) Set(key K, value V, cost int64) bool { return true } -func (wtc *otterCache[K, V]) Wait() {} func (wtc *otterCache[K, V]) Close() { // Stops the pending goroutine that Otter spins off wtc.cache.StopAllGoroutines() diff --git a/pkg/cache/cache_test.go b/pkg/cache/cache_test.go index a95dbf8ea..3e0a16a2d 100644 --- a/pkg/cache/cache_test.go +++ b/pkg/cache/cache_test.go @@ -3,9 +3,11 @@ package cache import ( "math" "testing" + "testing/synctest" "time" "github.com/prometheus/client_golang/prometheus" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -32,7 +34,6 @@ func TestCostAddedIncludesKey(t *testing.T) { const key = "some-key" const payloadCost = 10 require.True(t, cache.Set(StringKey(key), "value", payloadCost)) - cache.Wait() // costAdded must reflect the full entry weight (payload + key bytes), not // just the caller-supplied payload cost. @@ -68,9 +69,6 @@ func TestCacheWithMetrics(t *testing.T) { require.True(t, ok) } - // Wait for all sets to be processed - cache.Wait() - // Verify all entries for _, entry := range entries { retrieved, found := cache.Get(entry.key) @@ -86,7 +84,6 @@ func TestCacheWithMetrics(t *testing.T) { ok := cache.Set(StringKey("metric-key-1"), "value1", 10) require.True(t, ok) - cache.Wait() val, found := cache.Get("metric-key-1") require.True(t, found) require.Equal(t, "value1", val) @@ -94,7 +91,6 @@ func TestCacheWithMetrics(t *testing.T) { // same key set, diff value ok = cache.Set(StringKey("metric-key-1"), "value2", 10) require.True(t, ok) - cache.Wait() val, found = cache.Get("metric-key-1") require.True(t, found) require.Equal(t, "value2", val) @@ -120,7 +116,9 @@ func TestCacheWithMetrics(t *testing.T) { t.Run("GetMetrics", func(t *testing.T) { cache, err := NewOtterCacheWithMetrics[StringKey, string](prometheus.NewRegistry(), "test-otter", config) require.NoError(t, err) - defer cache.Close() + t.Cleanup(func() { + cache.Close() + }) // Set some values ok := cache.Set(StringKey("metric-key-1"), "value1", 10) @@ -128,8 +126,6 @@ func TestCacheWithMetrics(t *testing.T) { ok = cache.Set(StringKey("metric-key-2"), "value2", 20) require.True(t, ok) - cache.Wait() - // Perform some gets (hits and misses) _, ok = cache.Get(StringKey("metric-key-1")) // hit require.True(t, ok) @@ -151,4 +147,38 @@ func TestCacheWithMetrics(t *testing.T) { costAdded := metrics.CostAdded() require.GreaterOrEqual(t, costAdded, uint64(10), "expected cost to be tracked") }) + + t.Run("TTL Behavior", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + cache, err := NewOtterCacheWithMetrics[StringKey, string](prometheus.NewRegistry(), "test-otter", &Config{ + MaxCost: 1000, + // set a lower TTL + DefaultTTL: 1 * time.Minute, + }) + //nolint:testifylint // we're in a goroutine + if !assert.NoError(t, err) { + return + } + + // Set and get a key + ok := cache.Set(StringKey("a key"), "a value", 10) + assert.True(t, ok) + + // retrieve the key + retrieved, found := cache.Get(StringKey("a key")) + assert.True(t, found, "expected key %s to be found", "a key") + assert.Equal(t, "a value", retrieved, "expected value for key %s to match", "a key") + + // wait for a bit + time.Sleep(3 * time.Minute) + + // Retrieve the original key again; we're expecting not to find it. + _, found = cache.Get(StringKey("a key")) + assert.False(t, found, "expected key %s to be found", "a key") + + cache.Close() + + synctest.Wait() + }) + }) } diff --git a/pkg/datalayer/datalayer_test.go b/pkg/datalayer/datalayer_test.go index b4c10516e..3d46ed2de 100644 --- a/pkg/datalayer/datalayer_test.go +++ b/pkg/datalayer/datalayer_test.go @@ -916,8 +916,6 @@ func (c *testSchemaCache) cost(key SchemaCacheKey) int64 { return c.lastCosts[key] } -func (c *testSchemaCache) Wait() {} - // countingDatastore wraps a datastore.Datastore and counts ReadStoredSchema calls. type countingDatastore struct { datastore.Datastore diff --git a/pkg/datalayer/hashcache.go b/pkg/datalayer/hashcache.go index 4fdcfeaf4..442d714dd 100644 --- a/pkg/datalayer/hashcache.go +++ b/pkg/datalayer/hashcache.go @@ -36,7 +36,6 @@ func (k SchemaCacheKey) KeyString() string { return string(k) } type SchemaCache interface { Get(key SchemaCacheKey) (*datastore.ReadOnlyStoredSchema, bool) Set(key SchemaCacheKey, entry *datastore.ReadOnlyStoredSchema, cost int64) bool - Wait() } // latestSchemaEntry holds the most recent schema entry for fast-path lookups. diff --git a/pkg/datalayer/hashcache_test.go b/pkg/datalayer/hashcache_test.go index 873049da3..6a6acc25f 100644 --- a/pkg/datalayer/hashcache_test.go +++ b/pkg/datalayer/hashcache_test.go @@ -39,8 +39,6 @@ func TestSchemaHashCache_BasicGetSet(t *testing.T) { err = shc.Set(SchemaHash("hash1"), schema) require.NoError(t, err) - shc.cache.Wait() - retrieved, err = shc.get(SchemaHash("hash1")) require.NoError(t, err) require.NotNil(t, retrieved) @@ -280,8 +278,6 @@ func TestSchemaHashCache_SlowPathCacheHit(t *testing.T) { err = shc.Set(SchemaHash("hash2"), schema2) require.NoError(t, err) - shc.cache.Wait() - // Get hash1 — latest is hash2, so fast path misses, // but backing cache should have it (slow path hit, line 95) retrieved, err := shc.get(SchemaHash("hash1"))