Skip to content
Merged
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
107 changes: 92 additions & 15 deletions README.md
Original file line number Diff line number Diff line change
@@ -1,12 +1,14 @@
# Update Cache

This Go library provides an in-memory cache optimized for highly concurrent access. It provides a `Get(key)`method to
This Go library provides an in-memory cache optimized for highly concurrent access. It provides a `Get(key)` method to
look up a key-value pair within the cache. In case of cache misses it uses a pre-defined `UpdateFunc` to update the
cache entry. Features:

* Thread-safe `Get()` method to access key value pairs
* Thread-safe `Get()` method to access key value pairs, reporting whether the value was a cache hit, freshly updated
or served stale
* On concurrent calls to `Get()` for an uncached key, the `UpdateFunc` will only be called for the first caller. All
subsequent calls will block and then receive the cached result.
subsequent calls will block and then receive the cached result. Should the `UpdateFunc` fail, the next waiting
caller retries it.
* The `UpdateFunc` can decide whether errors should be cached or not. Either way `Get()` will return this error to the
caller transparently.

Expand Down Expand Up @@ -51,8 +53,8 @@ func main() {
for _ = range 3 {
wg.Add(1)
go func() {
valid, hit, _ := cache.Get(context.Background(), "my_secret_key")
fmt.Printf("requested key from cache, valid: %t, hit: %t \n", valid, hit)
valid, state, _ := cache.Get(context.Background(), "my_secret_key")
fmt.Printf("requested key from cache, valid: %t, state: %s \n", valid, state)
wg.Done()
}()
}
Expand All @@ -61,14 +63,14 @@ func main() {
```

Running this will result in all 3 goroutines getting the requested key, but only the first call will trigger a cache
update:
update (as long as it succeeds; a failed update is retried by the next waiting caller):

```bash
$ go run example.go
updated api key
requested key from cache, valid: true, hit: false
requested key from cache, valid: true, hit: true
requested key from cache, valid: true, hit: true
requested key from cache, valid: true, state: fresh
requested key from cache, valid: true, state: hit
requested key from cache, valid: true, state: hit
```

### Caching Errors
Expand Down Expand Up @@ -103,11 +105,11 @@ func main() {

cache := concache.NewUpdateCache(5*time.Minute, updateUserFunc)

_, hit, err := cache.Get(context.Background(), "someone")
fmt.Printf("hit: %t, err: %s \n", hit, err)
_, state, err := cache.Get(context.Background(), "someone")
fmt.Printf("state: %s, err: %s \n", state, err)

_, hit, err = cache.Get(context.Background(), "someone")
fmt.Printf("hit: %t, err: %s \n", hit, err)
_, state, err = cache.Get(context.Background(), "someone")
fmt.Printf("state: %s, err: %s \n", state, err)
}
```

Expand All @@ -116,8 +118,8 @@ The second call to Get will now return the cached error instead of executing the
```bash
$ go run example.go
updated user
hit: false, err: user id "someone" not found on database
hit: true, err: user id "someone" not found on database
state: fresh, err: user id "someone" not found on database
state: hit, err: user id "someone" not found on database
```

To **not** cache any errors (because of temporary issues such as connection losses to a database), just return the error
Expand All @@ -129,3 +131,78 @@ updateUserFunc := func (ctx context.Context, userId string) (concache.EntryUpdat
return concache.EntryUpdate[string]{}, errors.New("temporary connection issues")
}
```

## Bounding the cache

By default the cache grows with the key space. When keys come from open-ended or untrusted input,
bound it:

```golang
cache := concache.NewUpdateCache(5*time.Minute, updateFunc, concache.WithMaxEntries(50_000))
```

Once the limit is reached, storing a new entry drops the least recently stored one. Note that
`Prune()` still has to be called periodically to release entries that expired while the cache sat
below the limit:

```golang
go func() {
for range time.Tick(time.Minute) {
cache.Prune()
}
}()
```

## Serving stale values while the source is down

`Get` returns an error whenever the entry is expired and the `UpdateFunc` fails, which means a
temporary outage of the underlying data source turns into an outage for the caller. When serving the
previous answer is better than serving none, allow stale reads and check the returned state:

```golang
cache := concache.NewUpdateCache(time.Minute, updateFunc, concache.WithMaxStaleAge(time.Hour))

value, state, err := cache.Get(ctx, "my_key")
switch state {
case concache.StateHit, concache.StateFresh:
// value is current
case concache.StateStale:
// value is the previous answer, err explains why it could not be refreshed
case concache.StateMiss:
// nothing usable, err explains why
}
```

An entry may be served stale until `maxStaleAge` past its expiry, after which it is dropped. That
bound matters: it decides how long the cache keeps handing out an answer that may no longer be true.

A stale value always comes with the error that prevented the refresh, so callers that treat any
error as fatal keep failing exactly as they would without the option.

## Per-entry TTL

An `UpdateFunc` can override the cache's ttl for the entry it just produced by setting
`EntryUpdate.TTL`. This is useful when the data source tells you how long its answer is good for:

```golang
return concache.EntryUpdate[Response]{Value: response, TTL: response.CacheFor}, nil
```

A zero `TTL` uses the cache default.

## Supplying the update function per call

`GetWith` takes the `UpdateFunc` as an argument, for callers whose key cannot carry everything
the update needs — for example when the key is a hash of a request rather than the request itself:

```golang
cache := concache.NewUpdateCache[Response](time.Minute, nil, concache.WithMaxStaleAge(time.Hour))

value, state, err := cache.GetWith(ctx, hashOf(request), func(ctx context.Context, _ string) (concache.EntryUpdate[Response], error) {
return load(ctx, request)
})
```

Concurrent calls on the same key are still serialized. A successful update is shared with all
waiting callers, so only the first update function runs. A failed update is not cached, so the next
waiting caller runs its own update function instead.
159 changes: 159 additions & 0 deletions bounds_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,159 @@
package concache

import (
"context"
"strconv"
"sync"
"testing"
"time"

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)

func echoCache(t *testing.T, options ...Option) *UpdateCache[string] {

t.Helper()

return NewUpdateCache(time.Minute, func(_ context.Context, key string) (EntryUpdate[string], error) {
return EntryUpdate[string]{Value: key}, nil
}, options...)
}

func TestWithMaxEntries_EvictsLeastRecentlyStored(t *testing.T) {

cache := echoCache(t, WithMaxEntries(2))

for _, key := range []string{"a", "b", "c"} {
_, _, err := cache.Get(context.Background(), key)
require.NoError(t, err)
}

assert.Equal(t, 2, cache.Len())
assert.False(t, hasEntry(cache, "a"), "the least recently stored entry must be evicted")
assert.True(t, hasEntry(cache, "b"))
assert.True(t, hasEntry(cache, "c"))
}

func TestWithMaxEntries_RewriteRefreshesPosition(t *testing.T) {

cache := echoCache(t, WithMaxEntries(2))
cache.now = func() time.Time { return time.Now() }

putEntry(cache, "a", "a", time.Now().Add(-time.Hour)) // expired, so the next Get rewrites it
putEntry(cache, "b", "b", time.Now().Add(time.Hour))

_, _, err := cache.Get(context.Background(), "a")
require.NoError(t, err)

_, _, err = cache.Get(context.Background(), "c")
require.NoError(t, err)

assert.True(t, hasEntry(cache, "a"), "the rewritten entry must not be the eviction candidate")
assert.False(t, hasEntry(cache, "b"))
}

func TestWithMaxEntries_UnboundedByDefault(t *testing.T) {

cache := echoCache(t)

for i := 0; i < 200; i++ {
_, _, err := cache.Get(context.Background(), strconv.Itoa(i))
require.NoError(t, err)
}

assert.Equal(t, 200, cache.Len())
}

func TestEntryUpdate_TTLOverridesTheDefault(t *testing.T) {

now := time.Now()
perKeyTTL := map[string]time.Duration{"short": time.Second, "long": time.Hour}

cache := NewUpdateCache(time.Minute, func(_ context.Context, key string) (EntryUpdate[string], error) {
return EntryUpdate[string]{Value: key, TTL: perKeyTTL[key]}, nil
})
cache.now = func() time.Time { return now }

for _, key := range []string{"short", "long", "default"} {
_, _, err := cache.Get(context.Background(), key)
require.NoError(t, err)
}

assert.Equal(t, now.Add(time.Second), entryOf(cache, "short").ExpiresAt)
assert.Equal(t, now.Add(time.Hour), entryOf(cache, "long").ExpiresAt)
assert.Equal(t, now.Add(time.Minute), entryOf(cache, "default").ExpiresAt, "a zero ttl must fall back to the cache default")
}

func TestGetWith_UsesThePerCallUpdateFunc(t *testing.T) {

cache := NewUpdateCache[string](time.Minute, nil)

value, state, err := cache.GetWith(context.Background(), "key", func(_ context.Context, _ string) (EntryUpdate[string], error) {
return EntryUpdate[string]{Value: "from the caller"}, nil
})

require.NoError(t, err)
assert.Equal(t, StateFresh, state)
assert.Equal(t, "from the caller", value)

// The value is cached like any other, so a later read does not need an update function at all.
value, state, err = cache.Get(context.Background(), "key")
require.NoError(t, err)
assert.Equal(t, StateHit, state)
assert.Equal(t, "from the caller", value)
}

func TestGetWith_SingleFlightPerKey(t *testing.T) {

cache := NewUpdateCache[string](time.Minute, nil)

calls := 0
callsLock := sync.Mutex{}

wg := sync.WaitGroup{}
for i := 0; i < 10; i++ {
wg.Add(1)
go func() {
defer wg.Done()

_, _, err := cache.GetWith(context.Background(), "key", func(_ context.Context, _ string) (EntryUpdate[string], error) {
callsLock.Lock()
calls++
callsLock.Unlock()

time.Sleep(100 * time.Millisecond)

return EntryUpdate[string]{Value: "value"}, nil
})
assert.NoError(t, err)
}()
}
wg.Wait()

callsLock.Lock()
defer callsLock.Unlock()
assert.Equal(t, 1, calls, "concurrent readers of one key must share a single update")
}

func TestGet_WithoutAnUpdateFunc(t *testing.T) {

cache := NewUpdateCache[string](time.Minute, nil)

_, state, err := cache.Get(context.Background(), "key")

assert.Equal(t, StateMiss, state)
assert.ErrorIs(t, err, ErrNoUpdateFunc)
}

func TestWithMaxEntries_NegativeIsUnbounded(t *testing.T) {

cache := echoCache(t, WithMaxEntries(-1))

for i := 0; i < 200; i++ {
_, _, err := cache.Get(context.Background(), strconv.Itoa(i))
require.NoError(t, err)
}

assert.Equal(t, 200, cache.Len())
}
Loading
Loading