Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 0 additions & 7 deletions internal/datastore/proxy/schemacaching/standardcache.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 {
Expand Down
8 changes: 0 additions & 8 deletions pkg/cache/cache.go
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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) {
Expand Down
1 change: 0 additions & 1 deletion pkg/cache/cache_otter.go
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
48 changes: 39 additions & 9 deletions pkg/cache/cache_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)

Expand All @@ -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.
Expand Down Expand Up @@ -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)
Expand All @@ -86,15 +84,13 @@ 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)

// 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)
Expand All @@ -120,16 +116,16 @@ 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)
require.True(t, ok)
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)
Expand All @@ -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()
})
})
}
2 changes: 0 additions & 2 deletions pkg/datalayer/datalayer_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
1 change: 0 additions & 1 deletion pkg/datalayer/hashcache.go
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
4 changes: 0 additions & 4 deletions pkg/datalayer/hashcache_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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"))
Expand Down
Loading