From 953719904442f09a7e77801803ac74a636924e96 Mon Sep 17 00:00:00 2001 From: Aaron Asuncion Date: Fri, 18 Sep 2026 09:48:50 +0800 Subject: [PATCH 1/8] versionedcache: a cache that reloads when a Redis version key moves Holds one dataset in memory and reloads it only when a short version string published beside that dataset changes. A cycle that finds the version unchanged costs one read of that string instead of reading the whole dataset. - Cache[T] is built on otter v2, which holds the value, times the interval and runs one refresh at a time. The version comparison and what each cycle reports are this package's. - ReadVersion says where the version lives, with VersionInKey and VersionInHashField for the two Redis shapes. A nil one means the dataset carries no version, so the data reloads every interval. - Refreshes run in the background, so a warm cache is answered from memory and Get returns an error only when nothing has ever loaded. - Options.MaxStaleness reads the data again once a version has stopped moving for some reason other than the data being unchanged. Six hours by default. - Six outcomes reported through Options.WhenRefreshed, so a caller can tell an unchanged version from a reload, a staleness read and a failure. - New panics on settings that build a cache which then never refreshes properly, since it has no error to return. - A panicking loader, version read or reporter becomes a failed cycle rather than ending the process, and a source that is failing is read once an interval however much traffic keeps arriving. Co-Authored-By: Claude Opus 5 (1M context) --- versionedcache/README.md | 108 +++ versionedcache/cache.go | 347 ++++++++ versionedcache/cache_internal_test.go | 42 + versionedcache/cache_test.go | 1054 +++++++++++++++++++++++++ versionedcache/go.mod | 20 + versionedcache/go.sum | 32 + versionedcache/otter_adapters.go | 36 + versionedcache/refresh_outcome.go | 33 + 8 files changed, 1672 insertions(+) create mode 100644 versionedcache/README.md create mode 100644 versionedcache/cache.go create mode 100644 versionedcache/cache_internal_test.go create mode 100644 versionedcache/cache_test.go create mode 100644 versionedcache/go.mod create mode 100644 versionedcache/go.sum create mode 100644 versionedcache/otter_adapters.go create mode 100644 versionedcache/refresh_outcome.go diff --git a/versionedcache/README.md b/versionedcache/README.md new file mode 100644 index 0000000..560aacb --- /dev/null +++ b/versionedcache/README.md @@ -0,0 +1,108 @@ +# versionedcache + +Holds a dataset in memory and reloads it only when a small version published +beside that dataset says the data changed. A cycle that finds the version unchanged costs one +read of a short string instead of reading the whole dataset. + +Holding the value, timing the interval and running one refresh at a time are +[otter](https://github.com/maypok86/otter)'s, configured here as refresh-after-write. What this +package adds is the version comparison that decides whether a cycle reads the data at all, and +the outcome it reports for each cycle. + +The dataset's owner publishes the version, writing it after the data, so a value is never +advertised for data that has not landed. This package only reads it. + +## Use + +`New` takes a `ReadVersion`, which says where the version lives. Two are provided. + +**One field of a hash**, read with `HGET`. Use this when the publisher keeps the versions of +several datasets in a single hash. Each cache reads its own field with its own `HGET`, so the +shared hash is the publisher's layout rather than a batched read. + +```go +settings := versionedcache.New( + versionedcache.VersionInHashField(rdb, "app:versions", "settings"), + func(ctx context.Context) (map[string]Config, error) { + return readTheWholeHash(ctx, rdb) + }, + 30*time.Second, + versionedcache.Options{ + WhenRefreshed: func(outcome versionedcache.RefreshOutcome, err error) { + metrics.Count("config.refresh", "outcome:"+string(outcome)) + }, + }, +) + +configs, err := settings.Get(ctx) +``` + +`New` refuses a cache that would be quietly wrong for the life of the process, and panics on a +nil loader, a `checkEvery` that is not positive, or a negative `Options.MaxStaleness`. There is no +error to return, and each of those builds a cache that compiles and then never refreshes properly. + +**A string key of its own**, read with `GET`. Use this when the version has no hash to sit in. + +```go +versionedcache.New( + versionedcache.VersionInKey(rdb, "app:settings:version"), + loadData, 30*time.Second, versionedcache.Options{}, +) +``` + +**No version at all**: pass `nil`. The data then reloads every interval, which is what every +caller did before this package existed. A dataset small enough not to be worth gating wants +this, and a cache built that way needs no Redis behind it at all. + +A version that is configured but not published reads the same way. A key or field that is not +there means nobody publishes one, so deleting it is an off switch that needs no deploy. + +**A version that stops moving** is not proof that the data is unchanged: a writer can change +rows and die before it publishes, or a bulk import can publish once at the very end. So data +kept on an unchanged version for `Options.MaxStaleness` is read again at the next cycle anyway, +and that read restarts the clock; a cycle that skipped does not. Zero means +`DefaultMaxStaleness`, six hours. Set it against how often the writer runs, and keep it longer +than the check interval, or every cycle reloads. + +`Get` returns `T` by value, so it copies whatever you store on every call. Hold a map, a slice +or a pointer when the dataset is large. + +## What each cycle reports + +`Options.WhenRefreshed` is called once per cycle, from inside the load and a moment before the +value that cycle produced is stored. The outcome carries the whole classification, so a caller's +metric tag is `string(outcome)` with nothing to branch on. + +| Outcome | Meaning | +| --- | --- | +| `version_unchanged` | the version matched, so nothing was read | +| `reloaded_after_change` | the version moved, so the data was read again | +| `reloaded_without_version` | there was no version to compare, so the data was read again | +| `reloaded_at_max_staleness` | the version had not moved but the data had been kept for `MaxStaleness`, so it was read again | +| `refresh_failed` | the read failed and the previous value is still being served | +| `first_load_failed` | the read failed with nothing ever loaded, so `Get` returned the error | + +The `error` argument carries detail for the log line, never the classification. + +## Three things that surprise people + +**A refresh runs in the background.** The call that finds the interval elapsed is answered with +the data the cache already holds, and starts the refresh; a dataset that changed reaches a later +call. Nothing waits on Redis once the cache is warm, and the worst-case staleness is the interval +plus one refresh. The refresh also runs with the caller's cancellation stripped, so a deadline on the +`Get` that started it does not cut it short; the Redis client's own timeouts bound it instead. + +**Get returns an error only when it has never loaded.** Once a load has succeeded, a failure +to read either the version or the data keeps the previous value and `Get` returns it with a +nil error. The failure reaches you through `Options.WhenRefreshed`, which is where your log +line and metric belong. A Redis outage therefore does not empty a warm cache, and while one lasts +the source is still read only once an interval however much traffic arrives. + +The first load is the exception on both counts: it runs on the calling goroutine, and callers +arriving while it is still running wait for it and are handed what it produced. Only when that +load fails does `Get` return an error, and the next call tries again straight away. + +**The version is compared as exact text, never as an ordering.** Any different value counts +as changed — a newer one, an older one restored from a backup, or a different shape +entirely. That means the publisher only has to make the value change; it does not have to +make it increase. diff --git a/versionedcache/cache.go b/versionedcache/cache.go new file mode 100644 index 0000000..82c5cdf --- /dev/null +++ b/versionedcache/cache.go @@ -0,0 +1,347 @@ +// Package versionedcache holds a dataset in memory and reloads it only when a small version +// published beside that dataset says the data changed. A cycle that finds the version +// unchanged costs one read of a short string instead of reading the whole dataset. +// +// Where the version lives is the caller's choice, handed in as a ReadVersion. VersionInKey and +// VersionInHashField cover the two Redis shapes. A nil ReadVersion means the dataset carries no +// version at all, so the data reloads every interval. +package versionedcache + +import ( + "context" + "errors" + "fmt" + "runtime/debug" + "sync/atomic" + "time" + + "github.com/maypok86/otter/v2" + "github.com/redis/go-redis/v9" +) + +// ReadVersion reports the version published for a dataset right now. published is false when +// there is nothing to compare against, which tells the cache to reload rather than to fail. +type ReadVersion func(ctx context.Context) (version string, published bool, err error) + +// VersionInKey reads a version that has a Redis string key to itself, with GET. +func VersionInKey(client redis.UniversalClient, key string) ReadVersion { + return func(ctx context.Context) (string, bool, error) { + version, err := client.Get(ctx, key).Result() + return publishedVersion(version, err, key) + } +} + +// VersionInHashField reads one field of a Redis hash, with HGET. Use it when a publisher keeps +// the versions of several datasets in one hash; each cache still reads its own field. +func VersionInHashField(client redis.UniversalClient, hashKey, field string) ReadVersion { + location := hashKey + " field " + field + return func(ctx context.Context) (string, bool, error) { + version, err := client.HGet(ctx, hashKey, field).Result() + return publishedVersion(version, err, location) + } +} + +// publishedVersion turns one Redis reply into the answer ReadVersion owes the cache. A key or +// field that is not there is not a failure: it means nobody publishes one, so the data reloads. +func publishedVersion(version string, err error, location string) (string, bool, error) { + switch { + case errors.Is(err, redis.Nil): + return "", false, nil + case err != nil: + return "", false, fmt.Errorf("read version from %s: %w", location, err) + } + return version, true, nil +} + +// Options are the parts of a cache a caller may leave out. The zero value is fine: nothing +// is reported and the real clock is used. +type Options struct { + // WhenRefreshed is called once per refresh cycle with what the cycle did and any error + // that stopped it. This is where a caller emits its own metric and log line. + WhenRefreshed func(RefreshOutcome, error) + + // MaxStaleness is how long the data may be kept on an unchanged version before it is read + // again anyway, at the next cycle. Zero means DefaultMaxStaleness. It is the way out when a + // version stops moving for some reason other than the data being unchanged, such as a + // writer that changed rows and died before publishing. + MaxStaleness time.Duration + + // Now replaces the clock. Tests set it; leave it nil in production. + Now func() time.Time +} + +// DefaultMaxStaleness is the limit a cache runs on when Options.MaxStaleness is left zero. +const DefaultMaxStaleness = 6 * time.Hour + +// theDataset is the key the one dataset is stored under. A cache holds a single dataset, so +// the key never varies; the store underneath is keyed and needs one. +const theDataset = "dataset" + +// cacheEntry is one loaded dataset and the version it came with. They are kept together so a +// cycle that finds the version unchanged can prove the data it keeps was loaded under it. +type cacheEntry[T any] struct { + data T + version string + // published says a version was actually read for this entry. Without it the empty string is + // both "nobody publishes one" and a legitimate published value, and the two compare equal. + published bool + // loadedAt is when the data was last read. A cycle that keeps the data on an unchanged + // version leaves it alone, so the staleness limit counts from the read, not the check. + loadedAt time.Time +} + +// Cache holds one dataset together with the version it was loaded under, so a cycle that skips +// the reload can never serve a version whose data the cache does not hold. Holding the value, +// timing the interval and electing one refresher are otter's; the version comparison and what +// each cycle reports are here. +type Cache[T any] struct { + cached *otter.Cache[string, cacheEntry[T]] + cycle *refreshCycle[T] +} + +// New builds a cache that reloads its data only when readVersion reports a different version, +// or when the data has been kept past Options.MaxStaleness on a version that has not moved. +// +// A nil readVersion means the dataset carries no version, so the data reloads every checkEvery, +// which is what every caller did before this package existed. +// +// New panics on a nil loadData, a checkEvery that is not positive, or a negative +// Options.MaxStaleness. Each of those builds a cache that compiles and is then quietly wrong for +// the life of the process, so it is refused here rather than at the first cycle. +func New[T any]( + readVersion ReadVersion, + loadData func(context.Context) (T, error), + checkEvery time.Duration, + options Options, +) *Cache[T] { + if loadData == nil { + panic("versionedcache: loadData must not be nil") + } + // Both of these otherwise produce a cache that compiles, loads once and is quietly wrong: + // otter arms no refresh for an interval that is not positive, and a negative staleness limit + // makes every cycle read the whole dataset. + if checkEvery <= 0 { + panic("versionedcache: checkEvery must be greater than zero") + } + if options.MaxStaleness < 0 { + panic("versionedcache: Options.MaxStaleness must not be negative") + } + now := options.Now + if now == nil { + now = time.Now + } + maxStaleness := options.MaxStaleness + if maxStaleness == 0 { + maxStaleness = DefaultMaxStaleness + } + store := &otter.Options[string, cacheEntry[T]]{ + RefreshCalculator: refreshEvery[T]{interval: checkEvery}, + // Every failure is already reported through WhenRefreshed, where the caller's log line + // and metric live. otter's own logger writes to slog, which is not this service's file. + Logger: &otter.NoopLogger{}, + } + if options.Now != nil { + store.Clock = &clockFrom{now: options.Now} + } + return &Cache[T]{ + cached: otter.Must(store), + cycle: &refreshCycle[T]{ + readVersion: readVersion, + loadData: loadData, + interval: checkEvery, + maxStaleness: maxStaleness, + whenRefreshed: options.WhenRefreshed, + now: now, + }, + } +} + +// Get returns the cached data, and starts a refresh once the interval has elapsed. The refresh +// runs in the background: this call is answered with the data the cache already holds, and a +// changed dataset reaches a later call. It returns an error only when nothing has ever loaded. +func (c *Cache[T]) Get(ctx context.Context) (T, error) { + entry, err := c.cached.Get(ctx, theDataset, c.cycle) + if err != nil { + var nothing T + return nothing, err + } + return entry.data, nil +} + +// refreshCycle is one cache's refresh: the first read, and every cycle after it. The store +// calls Load with nothing cached and Reload with the data the cache is already holding. +type refreshCycle[T any] struct { + readVersion ReadVersion + loadData func(context.Context) (T, error) + interval time.Duration + maxStaleness time.Duration + whenRefreshed func(RefreshOutcome, error) + now func() time.Time + + // startedAt is when the last cycle began, as Unix nanoseconds, and zero before the first one. + // startCycle below is the only writer. + startedAt atomic.Int64 + + // newest is what the last finished cycle produced. The store hands Reload the value the + // cache held when that call's Get ran, which is out of date for a cycle that was queued + // behind another one, and comparing against an out-of-date version re-reads a dataset the + // cycle in front has just read. + newest atomic.Pointer[cacheEntry[T]] +} + +// Load is the first read. A failure has no previous value to fall back to, so the error travels +// back to Get, which is how a caller learns it is running on its own defaults. +func (c *refreshCycle[T]) Load(ctx context.Context, _ string) (cacheEntry[T], error) { + version, published, err := c.currentVersion(ctx) + if err != nil { + return c.failed(FirstLoadFailed, err) + } + + data, err := c.loadDataNow(ctx) + if err != nil { + return c.failed(FirstLoadFailed, err) + } + + c.report(reloadReason(published, false), nil) + return c.loadedNow(data, version, published), nil +} + +// Reload is every cycle after the first. A cycle that cannot read reports the failure and hands +// the error back: the cache keeps the data it already holds, and refreshEvery below waits out +// the interval before the next attempt. +func (c *refreshCycle[T]) Reload(ctx context.Context, _ string, cached cacheEntry[T]) (cacheEntry[T], error) { + if newest := c.newest.Load(); newest != nil { + cached = *newest + } + + if !c.startCycle() { + return cached, nil + } + + // The version is read before the data on purpose. A cycle that overlaps a writer is then + // stamped with the older version and reloads again next time, rather than advertising + // data it does not hold. + version, published, err := c.currentVersion(ctx) + if err != nil { + return c.failed(RefreshFailed, err) + } + + versionUnchanged := published && cached.published && cached.version == version + if versionUnchanged && !c.keptPastMaxStaleness(cached) { + c.report(VersionUnchanged, nil) + return cached, nil + } + + data, err := c.loadDataNow(ctx) + if err != nil { + return c.failed(RefreshFailed, err) + } + + c.report(reloadReason(published, versionUnchanged), nil) + return c.loadedNow(data, version, published), nil +} + +// startCycle claims the current interval, so that of the callers who find the data due only the +// first goes on to read. The store puts the data back on the clock when a cycle finishes, not when +// it starts, so while a read is failing the data stays due for as long as that read takes and every +// call arriving meanwhile starts a cycle of its own — a source already known to be down is then +// read continuously rather than once an interval. +// +// A cycle that loses this claim reports nothing, because none ran: the caller is counting cycles, +// and a call that did no work is not one. +func (c *refreshCycle[T]) startCycle() bool { + now := c.now().UnixNano() + for { + started := c.startedAt.Load() + if started != 0 && now-started < int64(c.interval) { + return false + } + if c.startedAt.CompareAndSwap(started, now) { + return true + } + } +} + +// failed reports a cycle that could not read and hands the error back. A refresh keeps whatever +// the cache already holds; a first load has nothing to keep, so the error reaches Get instead. +func (c *refreshCycle[T]) failed(outcome RefreshOutcome, err error) (cacheEntry[T], error) { + c.report(outcome, err) + var nothing cacheEntry[T] + if errors.Is(err, otter.ErrNotFound) { + // The store reads that sentinel as the dataset being gone and drops what it holds, which + // would empty a warm cache after one failed read. Flatten the chain rather than pass it on: + // a loader reporting it means the read failed, not that the data no longer exists. + return nothing, fmt.Errorf("versionedcache: %s", err) + } + return nothing, err +} + +// currentVersion asks the caller's ReadVersion, treating a cache that has none as a dataset +// nobody publishes a version for: nothing to compare, so reload. +func (c *refreshCycle[T]) currentVersion(ctx context.Context) (version string, published bool, err error) { + if c.readVersion == nil { + return "", false, nil + } + defer func() { + if r := recover(); r != nil { + version, published, err = "", false, recovered("ReadVersion", r) + } + }() + return c.readVersion(ctx) +} + +// loadDataNow reads the dataset, turning a panic into an error. otter starts a background reload +// on a goroutine of its own and re-panics whatever the loader panicked with, so an unrecovered +// panic there ends the process instead of reporting one failed cycle. +func (c *refreshCycle[T]) loadDataNow(ctx context.Context) (data T, err error) { + defer func() { + if r := recover(); r != nil { + var nothing T + data, err = nothing, recovered("loadData", r) + } + }() + return c.loadData(ctx) +} + +// recovered carries the stack into the error, because nothing else will print it: the cycle +// reports through WhenRefreshed and the goroutine it ran on ends quietly. +func recovered(what string, panicked any) error { + return fmt.Errorf("versionedcache: %s panicked: %v\n%s", what, panicked, debug.Stack()) +} + +// keptPastMaxStaleness says whether the data has been kept for MaxStaleness or longer since it +// was last read. It is asked only on a cycle, so the read lands on the first cycle past the limit. +func (c *refreshCycle[T]) keptPastMaxStaleness(cached cacheEntry[T]) bool { + return c.now().Sub(cached.loadedAt) >= c.maxStaleness +} + +// loadedNow is the entry for data read this cycle, with the staleness clock started again. +func (c *refreshCycle[T]) loadedNow(data T, version string, published bool) cacheEntry[T] { + loaded := cacheEntry[T]{data: data, version: version, published: published, loadedAt: c.now()} + c.newest.Store(&loaded) + return loaded +} + +// reloadReason names why a cycle read the data: nothing to compare against, a version that +// moved, or one that stayed put for longer than the data may be kept. +func reloadReason(published, versionUnchanged bool) RefreshOutcome { + switch { + case !published: + return ReloadedWithoutVersion + case versionUnchanged: + return ReloadedAtMaxStaleness + default: + return ReloadedAfterChange + } +} + +// report hands the cycle's outcome to the caller. It runs on the store's refresh goroutine, like +// the two reads above it, so a panic here would end the process. There is nowhere to report a +// reporter that failed, so the report is dropped and the next cycle still runs. +func (c *refreshCycle[T]) report(outcome RefreshOutcome, err error) { + if c.whenRefreshed == nil { + return + } + defer func() { _ = recover() }() + c.whenRefreshed(outcome, err) +} diff --git a/versionedcache/cache_internal_test.go b/versionedcache/cache_internal_test.go new file mode 100644 index 0000000..0927fb4 --- /dev/null +++ b/versionedcache/cache_internal_test.go @@ -0,0 +1,42 @@ +package versionedcache + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// otter hands Reload the value the cache held when the refreshing Get ran, captured when the +// refresh was scheduled rather than when it runs. A cycle that starts while another is still +// finishing therefore sees an out-of-date entry, and comparing against its version re-reads the +// dataset the cycle in front has just read. The cycle is driven directly here because that +// window is a few instructions wide inside the library and cannot be forced from outside. +func TestAReloadHandedAnOutOfDateEntryComparesAgainstTheLastCycleInstead(t *testing.T) { + ctx := context.Background() + now := time.Now() + reads := 0 + + cycle := &refreshCycle[string]{ + readVersion: func(context.Context) (string, bool, error) { return "v2", true, nil }, + loadData: func(context.Context) (string, error) { reads++; return "data for v2", nil }, + maxStaleness: time.Hour, + now: func() time.Time { return now }, + } + + outOfDate := cacheEntry[string]{data: "data for v1", version: "v1", published: true, loadedAt: now} + + // The version has moved, so this cycle reads the dataset and records what it loaded. + loaded, err := cycle.Reload(ctx, theDataset, outOfDate) + require.NoError(t, err) + require.Equal(t, "v2", loaded.version) + require.Equal(t, 1, reads) + + // The next cycle is handed the same out-of-date entry, which is what a queued one gets. + kept, err := cycle.Reload(ctx, theDataset, outOfDate) + require.NoError(t, err) + assert.Equal(t, 1, reads, "a cycle must compare against what the last finished cycle loaded") + assert.Equal(t, "data for v2", kept.data) +} diff --git a/versionedcache/cache_test.go b/versionedcache/cache_test.go new file mode 100644 index 0000000..55248c2 --- /dev/null +++ b/versionedcache/cache_test.go @@ -0,0 +1,1054 @@ +package versionedcache_test + +import ( + "context" + "errors" + "fmt" + "slices" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/maypok86/otter/v2" + "github.com/redis/go-redis/v9" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/wego/pkg/versionedcache" +) + +// fakeClock lets a test move time forward without sleeping. Guarded by a mutex +// because the cache reads it from its refresh goroutines under -race. +type fakeClock struct { + mutex sync.Mutex + now time.Time +} + +// newFakeClock starts from the real clock, so no test carries a typed-in date. +func newFakeClock() *fakeClock { + return &fakeClock{now: time.Now()} +} + +func (c *fakeClock) Now() time.Time { + c.mutex.Lock() + defer c.mutex.Unlock() + return c.now +} + +func (c *fakeClock) Advance(by time.Duration) { + c.mutex.Lock() + defer c.mutex.Unlock() + c.now = c.now.Add(by) +} + +// answers is the data the loader hands back, and a count of how often it was asked. Guarded: +// the loader runs on a refresh goroutine while the test changes what it should answer. +type answers struct { + mutex sync.Mutex + value string + reads atomic.Int64 +} + +func newAnswers(value string) *answers { + return &answers{value: value} +} + +func (a *answers) set(value string) { + a.mutex.Lock() + defer a.mutex.Unlock() + a.value = value +} + +func (a *answers) load(context.Context) (string, error) { + a.reads.Add(1) + a.mutex.Lock() + defer a.mutex.Unlock() + return a.value, nil +} + +func (a *answers) timesRead() int { + return int(a.reads.Load()) +} + +// cycles records what every refresh cycle reported and lets a test wait for one. Every cycle +// after the first load runs in the background, so a test waits for it rather than guessing +// how long it takes. +type cycles struct { + t *testing.T + mutex sync.Mutex + outcomes []versionedcache.RefreshOutcome + errs []error +} + +func newCycles(t *testing.T) *cycles { + return &cycles{t: t} +} + +func (c *cycles) record(outcome versionedcache.RefreshOutcome, err error) { + c.mutex.Lock() + defer c.mutex.Unlock() + c.outcomes = append(c.outcomes, outcome) + c.errs = append(c.errs, err) +} + +// waitUntil blocks until this many cycles have finished reporting. Counted rather than awaited +// one at a time, so the cycles a test has already seen cannot be mistaken for the one it awaits. +func (c *cycles) waitUntil(count int) { + c.t.Helper() + require.Eventually(c.t, func() bool { return c.count() >= count }, + 5*time.Second, 5*time.Millisecond, "waited for %d refresh cycles", count) +} + +func (c *cycles) count() int { + c.mutex.Lock() + defer c.mutex.Unlock() + return len(c.outcomes) +} + +func (c *cycles) seen() []versionedcache.RefreshOutcome { + c.mutex.Lock() + defer c.mutex.Unlock() + return slices.Clone(c.outcomes) +} + +func (c *cycles) lastError() error { + c.mutex.Lock() + defer c.mutex.Unlock() + if len(c.errs) == 0 { + return nil + } + return c.errs[len(c.errs)-1] +} + +func (c *cycles) failures() int { + c.mutex.Lock() + defer c.mutex.Unlock() + failed := 0 + for _, err := range c.errs { + if err != nil { + failed++ + } + } + return failed +} + +// refreshCycle asks for the data and waits for the background cycle that the elapsed interval +// starts. The call itself is answered with whatever the cache already holds. +func refreshCycle(t *testing.T, cache *versionedcache.Cache[string], reported *cycles) { + t.Helper() + sofar := reported.count() + _, err := cache.Get(context.Background()) + require.NoError(t, err) + reported.waitUntil(sofar + 1) +} + +// assertServes waits for the cache to be answering with want. A cycle reports from inside the +// load, a moment before the value it loaded is stored, so a read straight after the cycle can +// still be answered with the old one. Read the counts a cycle changed before calling this: a +// read landing in that moment starts a further cycle of its own. +func assertServes(t *testing.T, cache *versionedcache.Cache[string], want string) { + t.Helper() + require.Eventually(t, func() bool { + served, err := cache.Get(context.Background()) + return err == nil && served == want + }, 2*time.Second, 20*time.Millisecond, "the cache never served %q", want) +} + +func newTestClient(t *testing.T) (*miniredis.Miniredis, *redis.Client) { + t.Helper() + server := miniredis.RunT(t) + client := redis.NewClient(&redis.Options{Addr: server.Addr()}) + t.Cleanup(func() { _ = client.Close() }) + return server, client +} + +const exampleVersionKey = "example:version" + +const ( + exampleVersionHashKey = "example:versions" + exampleVersionField = "example" +) + +// versionPlace is one of the two places a version can live. A behaviour that has to hold +// wherever it lives is written once here and run for each place, so the two cannot drift apart. +type versionPlace struct { + name string + publish func(*testing.T, *miniredis.Miniredis, string) + build func(*redis.Client, func(context.Context) (string, error), versionedcache.Options) *versionedcache.Cache[string] +} + +func versionPlaces() []versionPlace { + return []versionPlace{ + { + name: "a string key of its own", + publish: func(t *testing.T, server *miniredis.Miniredis, version string) { + require.NoError(t, server.Set(exampleVersionKey, version)) + }, + build: func(client *redis.Client, loadData func(context.Context) (string, error), + options versionedcache.Options) *versionedcache.Cache[string] { + return versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), loadData, time.Minute, options) + }, + }, + { + name: "one field of a shared hash", + publish: func(_ *testing.T, server *miniredis.Miniredis, version string) { + server.HSet(exampleVersionHashKey, exampleVersionField, version) + }, + build: func(client *redis.Client, loadData func(context.Context) (string, error), + options versionedcache.Options) *versionedcache.Cache[string] { + return versionedcache.New(versionedcache.VersionInHashField(client, + exampleVersionHashKey, exampleVersionField), loadData, time.Minute, options) + }, + }, + } +} + +func TestAnUnchangedVersionDoesNotReadTheData(t *testing.T) { + for _, place := range versionPlaces() { + t.Run(place.name, func(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + data := newAnswers("first") + reported := newCycles(t) + place.publish(t, server, "20260912.1") + + cache := place.build(client, data.load, + versionedcache.Options{Now: clock.Now, WhenRefreshed: reported.record}) + + served, err := cache.Get(context.Background()) + require.NoError(t, err) + assert.Equal(t, "first", served) + assert.Equal(t, 1, data.timesRead()) + + // The data behind the version changed but the version did not: the cache must not see it. + data.set("second") + clock.Advance(2 * time.Minute) + + refreshCycle(t, cache, reported) + + served, err = cache.Get(context.Background()) + require.NoError(t, err) + assert.Equal(t, "first", served) + assert.Equal(t, 1, data.timesRead(), "an unchanged version must not read the data") + assert.Equal(t, versionedcache.VersionUnchanged, reported.seen()[1]) + }) + } +} + +func TestADifferentVersionReloadsTheData(t *testing.T) { + // The comparison is exact text and never an ordering, so a version that moves backwards + // counts as changed exactly as much as one that moves forwards. + changes := []struct{ name, from, to string }{ + {"a newer version", "20260912.1", "20260912.2"}, + {"an older version restored from a backup", "20260912.9", "20260101.1"}, + } + for _, place := range versionPlaces() { + for _, change := range changes { + t.Run(place.name+", "+change.name, func(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + data := newAnswers("first") + reported := newCycles(t) + place.publish(t, server, change.from) + + cache := place.build(client, data.load, + versionedcache.Options{Now: clock.Now, WhenRefreshed: reported.record}) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + data.set("second") + place.publish(t, server, change.to) + clock.Advance(2 * time.Minute) + + refreshCycle(t, cache, reported) + assert.Equal(t, 2, data.timesRead()) + assertServes(t, cache, "second") + }) + } + } +} + +// Every one of these reads the same as a version nobody publishes, which is the behaviour every +// caller had before this package existed. Deleting the version is an off switch needing no deploy. +func TestWithNothingToCompareTheDataReloadsEveryInterval(t *testing.T) { + inAHash := func(client *redis.Client, loadData func(context.Context) (string, error), + options versionedcache.Options) *versionedcache.Cache[string] { + return versionedcache.New(versionedcache.VersionInHashField(client, exampleVersionHashKey, + exampleVersionField), loadData, time.Minute, options) + } + cases := []struct { + name string + seed func(*miniredis.Miniredis) + build func(*redis.Client, func(context.Context) (string, error), versionedcache.Options) *versionedcache.Cache[string] + }{ + { + name: "no version key was configured", + build: func(_ *redis.Client, loadData func(context.Context) (string, error), + options versionedcache.Options) *versionedcache.Cache[string] { + return versionedcache.New(nil, loadData, time.Minute, options) + }, + }, + { + name: "the key is configured but nobody has published it", + build: func(client *redis.Client, loadData func(context.Context) (string, error), + options versionedcache.Options) *versionedcache.Cache[string] { + return versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), loadData, time.Minute, options) + }, + }, + { + name: "the hash carries somebody else's field but not this one", + seed: func(server *miniredis.Miniredis) { + server.HSet(exampleVersionHashKey, "someoneElse", "1789380316123456789") + }, + build: inAHash, + }, + { + name: "the hash is not there at all", + build: inAHash, + }, + } + for _, testCase := range cases { + t.Run(testCase.name, func(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + data := newAnswers("first") + if testCase.seed != nil { + testCase.seed(server) + } + reported := newCycles(t) + + cache := testCase.build(client, data.load, + versionedcache.Options{Now: clock.Now, WhenRefreshed: reported.record}) + + served, err := cache.Get(context.Background()) + require.NoError(t, err) + assert.Equal(t, "first", served) + + data.set("second") + clock.Advance(2 * time.Minute) + + refreshCycle(t, cache, reported) + assert.Equal(t, 2, data.timesRead(), "with no version published the cache reloads every interval") + assert.Equal(t, []versionedcache.RefreshOutcome{ + versionedcache.ReloadedWithoutVersion, versionedcache.ReloadedWithoutVersion, + }, reported.seen()) + assertServes(t, cache, "second") + }) + } +} + +func TestDataIsServedFromMemoryInsideTheInterval(t *testing.T) { + // A cache with no version to check needs no redis at all. + clock := newFakeClock() + data := newAnswers("first") + + cache := versionedcache.New(nil, data.load, time.Minute, versionedcache.Options{Now: clock.Now}) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + data.set("second") + clock.Advance(30 * time.Second) + + served, err := cache.Get(context.Background()) + require.NoError(t, err) + assert.Equal(t, "first", served, "inside the interval the cache must not reload") + assert.Equal(t, 1, data.timesRead()) +} + +func TestOutcomesTellTheThreeCasesApart(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + data := newAnswers("first") + reported := newCycles(t) + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), data.load, time.Minute, versionedcache.Options{ + Now: clock.Now, + WhenRefreshed: reported.record, + }) + + // No version published yet. + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + // A version appears. + require.NoError(t, server.Set(exampleVersionKey, "20260912.1")) + clock.Advance(2 * time.Minute) + refreshCycle(t, cache, reported) + + // The same version again. + clock.Advance(2 * time.Minute) + refreshCycle(t, cache, reported) + + assert.Equal(t, []versionedcache.RefreshOutcome{ + versionedcache.ReloadedWithoutVersion, + versionedcache.ReloadedAfterChange, + versionedcache.VersionUnchanged, + }, reported.seen()) +} + +func TestFailedVersionReadKeepsTheCachedData(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + data := newAnswers("first") + reported := newCycles(t) + require.NoError(t, server.Set(exampleVersionKey, "20260912.1")) + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), data.load, time.Minute, versionedcache.Options{ + Now: clock.Now, + WhenRefreshed: reported.record, + }) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + // Redis goes away. + server.Close() + clock.Advance(2 * time.Minute) + + refreshCycle(t, cache, reported) + + served, err := cache.Get(context.Background()) + require.NoError(t, err, "a warm cache keeps serving through a Redis outage") + assert.Equal(t, "first", served) + assert.Equal(t, 1, data.timesRead(), "the data must not be read when the version could not be") + assert.Error(t, reported.lastError(), "the failure is reported even though Get succeeded") +} + +func TestFailedDataReadKeepsTheCachedData(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + reported := newCycles(t) + require.NoError(t, server.Set(exampleVersionKey, "20260912.1")) + + var failNextLoad atomic.Bool + loadData := func(context.Context) (string, error) { + if failNextLoad.Load() { + return "", errors.New("the hash could not be read") + } + return "first", nil + } + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), loadData, time.Minute, versionedcache.Options{ + Now: clock.Now, + WhenRefreshed: reported.record, + }) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + failNextLoad.Store(true) + require.NoError(t, server.Set(exampleVersionKey, "20260912.2")) + clock.Advance(2 * time.Minute) + + refreshCycle(t, cache, reported) + + served, err := cache.Get(context.Background()) + require.NoError(t, err) + assert.Equal(t, "first", served) + assert.Equal(t, versionedcache.RefreshFailed, reported.seen()[1]) + assert.Error(t, reported.lastError()) +} + +// The version is read before the data, so a redis that is down fails the first load before the +// loader is ever called. The caller gets the error and runs on its own defaults. +func TestAFirstLoadThatCannotReadTheVersionReturnsTheError(t *testing.T) { + server, client := newTestClient(t) + reported := newCycles(t) + var loads atomic.Int64 + loadData := func(context.Context) (string, error) { + loads.Add(1) + return "never reached", nil + } + + server.Close() + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), loadData, time.Minute, + versionedcache.Options{WhenRefreshed: reported.record}) + + data, err := cache.Get(context.Background()) + assert.Error(t, err) + assert.Empty(t, data) + assert.Equal(t, int64(0), loads.Load(), "the data is not read when the version could not be") + assert.Equal(t, []versionedcache.RefreshOutcome{versionedcache.FirstLoadFailed}, reported.seen()) +} + +func TestFirstLoadFailureIsReturnedToTheCaller(t *testing.T) { + _, client := newTestClient(t) + loadData := func(context.Context) (string, error) { + return "", errors.New("the hash could not be read") + } + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), loadData, time.Minute, versionedcache.Options{}) + + data, err := cache.Get(context.Background()) + assert.Error(t, err, "with nothing cached there is nothing to fall back to") + assert.Empty(t, data) +} + +// Nothing has ever loaded, so a caller arriving while the first load is running has no previous +// value to fall back to. It waits for that load rather than being turned away, and however many +// callers arrive the data is read once. +func TestCallersArrivingDuringTheFirstLoadAllGetIt(t *testing.T) { + server, client := newTestClient(t) + require.NoError(t, server.Set(exampleVersionKey, "20260912.1")) + + firstLoadStarted := make(chan struct{}) + releaseFirstLoad := make(chan struct{}) + var loads atomic.Int64 + loadData := func(context.Context) (string, error) { + if loads.Add(1) == 1 { + close(firstLoadStarted) + <-releaseFirstLoad + } + return "first", nil + } + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), loadData, time.Minute, versionedcache.Options{}) + + var callers sync.WaitGroup + for i := 0; i < 4; i++ { + callers.Add(1) + go func() { + defer callers.Done() + data, err := cache.Get(context.Background()) + assert.NoError(t, err) + assert.Equal(t, "first", data) + }() + } + + <-firstLoadStarted + close(releaseFirstLoad) + callers.Wait() + + assert.Equal(t, int64(1), loads.Load(), "however many callers arrive, the data is read once") +} + +// A failed cycle waits out the interval like a successful one, so an unreachable Redis is tried +// once a cycle rather than on every call. +func TestAFailedCycleRestartsTheInterval(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + data := newAnswers("first") + reported := newCycles(t) + require.NoError(t, server.Set(exampleVersionKey, "20260912.1")) + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), data.load, time.Minute, versionedcache.Options{ + Now: clock.Now, + WhenRefreshed: reported.record, + }) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + server.Close() + clock.Advance(2 * time.Minute) + + refreshCycle(t, cache, reported) + require.Equal(t, 1, reported.failures()) + + served, err := cache.Get(context.Background()) + require.NoError(t, err) + assert.Equal(t, "first", served) + + for i := 0; i < 5; i++ { + served, err = cache.Get(context.Background()) + require.NoError(t, err) + assert.Equal(t, "first", served) + } + assert.Equal(t, 1, reported.failures(), "calls inside the interval must not reach for redis again") + + clock.Advance(2 * time.Minute) + refreshCycle(t, cache, reported) + assert.Equal(t, 2, reported.failures(), "the next interval tries again") + assert.Equal(t, 1, data.timesRead(), "the data was never re-read while the version could not be") +} + +// A failed cycle is put back on the clock, so calls inside the interval start no further refresh. +// The store's own calculator leaves a failed reload still due, which would send every call that +// follows straight back to a source already known to be down. The loader here fails at once, so an +// extra cycle would report well inside the window below. +func TestAFailedCycleIsPutBackOnTheClock(t *testing.T) { + clock := newFakeClock() + reported := newCycles(t) + var failNow atomic.Bool + loadData := func(context.Context) (string, error) { + if failNow.Load() { + return "", errors.New("the dataset could not be read") + } + return "first", nil + } + + cache := versionedcache.New(nil, loadData, time.Minute, + versionedcache.Options{Now: clock.Now, WhenRefreshed: reported.record}) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + failNow.Store(true) + clock.Advance(2 * time.Minute) + refreshCycle(t, cache, reported) + require.Equal(t, 1, reported.failures()) + + // The clock does not move again, so every call below is inside the interval the failed cycle + // restarted. None of them may start a cycle of its own. + assert.Never(t, func() bool { + served, getErr := cache.Get(context.Background()) + assert.NoError(t, getErr) + assert.Equal(t, "first", served) + return reported.failures() > 1 + }, 300*time.Millisecond, 10*time.Millisecond, "a failed cycle must wait out the interval like a good one") +} + +// The cold case is the exception: with nothing to serve there is no interval to wait out. +func TestACacheThatHasNeverLoadedRetriesEveryCall(t *testing.T) { + _, client := newTestClient(t) + clock := newFakeClock() + var attempts atomic.Int64 + loadData := func(context.Context) (string, error) { + attempts.Add(1) + return "", errors.New("the hash could not be read") + } + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), loadData, time.Minute, + versionedcache.Options{Now: clock.Now}) + + // The clock never moves, so an interval that applied here would stop the later calls. + for call := 1; call <= 3; call++ { + _, err := cache.Get(context.Background()) + require.Error(t, err) + assert.Equal(t, int64(call), attempts.Load()) + } +} + +// A cycle that fails must leave the newest value in place rather than writing back whatever the +// cache was holding when that cycle started. +func TestAFailedCycleLeavesTheNewestValueInPlace(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + data := newAnswers("v1data") + reported := newCycles(t) + require.NoError(t, server.Set(exampleVersionKey, "v1")) + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), data.load, time.Minute, versionedcache.Options{ + Now: clock.Now, + WhenRefreshed: reported.record, + }) + + warm, err := cache.Get(context.Background()) + require.NoError(t, err) + require.Equal(t, "v1data", warm) + + data.set("v2data") + require.NoError(t, server.Set(exampleVersionKey, "v2")) + clock.Advance(2 * time.Minute) + + refreshCycle(t, cache, reported) + assertServes(t, cache, "v2data") + + // Redis goes away, so the next cycle fails with v2 already held. + server.Close() + clock.Advance(2 * time.Minute) + + refreshCycle(t, cache, reported) + require.Equal(t, 1, reported.failures()) + + served, err := cache.Get(context.Background()) + require.NoError(t, err) + assert.Equal(t, "v2data", served, "a failed cycle must not write an older snapshot back") +} + +func TestConcurrentGetsCauseOneDataRead(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + reported := newCycles(t) + require.NoError(t, server.Set(exampleVersionKey, "20260912.1")) + + var loads atomic.Int64 + refreshIsRunning := make(chan struct{}) + releaseRefresh := make(chan struct{}) + loadData := func(context.Context) (string, error) { + if loads.Add(1) == 2 { + // Hold the refresh open so the other callers arrive while it is still running. + close(refreshIsRunning) + <-releaseRefresh + return "second", nil + } + return "first", nil + } + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), loadData, time.Minute, versionedcache.Options{ + Now: clock.Now, + WhenRefreshed: reported.record, + }) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + require.NoError(t, server.Set(exampleVersionKey, "20260912.2")) + clock.Advance(2 * time.Minute) + + // The call that trips the interval starts the refresh and is answered from memory. + served, err := cache.Get(context.Background()) + require.NoError(t, err) + assert.Equal(t, "first", served) + <-refreshIsRunning + + for i := 0; i < 4; i++ { + served, err = cache.Get(context.Background()) + require.NoError(t, err) + assert.Equal(t, "first", served, "callers arriving mid-refresh are served the held value") + } + + close(releaseRefresh) + reported.waitUntil(2) + assert.Equal(t, int64(2), loads.Load(), "one warm-up load plus one refresh, not six") + assertServes(t, cache, "second") +} + +// the whole point of one hash for every dataset: a version moving for somebody else's namespace +// must not send this reader off to read its own dataset again +func TestAVersionMovingInAnotherFieldIsIgnored(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + data := newAnswers("first") + reported := newCycles(t) + server.HSet(exampleVersionHashKey, exampleVersionField, "1789380316123456789") + + cache := versionedcache.New( + versionedcache.VersionInHashField(client, exampleVersionHashKey, exampleVersionField), + data.load, time.Minute, versionedcache.Options{Now: clock.Now, WhenRefreshed: reported.record}) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + server.HSet(exampleVersionHashKey, "someoneElse", "1789380341772904118") + clock.Advance(2 * time.Minute) + + refreshCycle(t, cache, reported) + assert.Equal(t, 1, data.timesRead()) +} + +// A version that stops moving is not proof the data is unchanged: a writer can change rows and die +// before publishing. So data kept on an unchanged version for MaxStaleness is read again anyway, and +// only a read restarts that clock, never a check that skipped. +func TestDataKeptPastMaxStalenessIsReadAgainOnAnUnchangedVersion(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + data := newAnswers("first") + reported := newCycles(t) + require.NoError(t, server.Set(exampleVersionKey, "20260912.1")) + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), data.load, time.Minute, + versionedcache.Options{Now: clock.Now, WhenRefreshed: reported.record, MaxStaleness: 10 * time.Minute}) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + // The rows change under a version that never moves. + data.set("second") + + clock.Advance(6 * time.Minute) + refreshCycle(t, cache, reported) + assert.Equal(t, 1, data.timesRead(), "inside the limit an unchanged version still skips the read") + + // Eleven minutes since the read, five since the check that skipped: only the read counts. + clock.Advance(5 * time.Minute) + refreshCycle(t, cache, reported) + assert.Equal(t, 2, data.timesRead(), "past the limit the data is read though the version has not moved") + assert.Equal(t, versionedcache.ReloadedAtMaxStaleness, reported.seen()[2]) + assertServes(t, cache, "second") + + // That read restarted the clock, so the next cycle skips again. + clock.Advance(2 * time.Minute) + refreshCycle(t, cache, reported) + assert.Equal(t, 2, data.timesRead(), "a read restarts the staleness clock") + assert.Equal(t, versionedcache.VersionUnchanged, reported.seen()[3]) +} + +func TestMaxStalenessDefaultsToSixHours(t *testing.T) { + assert.Equal(t, 6*time.Hour, versionedcache.DefaultMaxStaleness) + + server, client := newTestClient(t) + clock := newFakeClock() + data := newAnswers("first") + reported := newCycles(t) + require.NoError(t, server.Set(exampleVersionKey, "20260912.1")) + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), data.load, time.Minute, + versionedcache.Options{Now: clock.Now, WhenRefreshed: reported.record}) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + clock.Advance(6*time.Hour - time.Minute) + refreshCycle(t, cache, reported) + assert.Equal(t, 1, data.timesRead(), "a minute short of six hours is still inside the default") + + clock.Advance(2 * time.Minute) + refreshCycle(t, cache, reported) + assert.Equal(t, 2, data.timesRead(), "six hours on an unchanged version reads the data again") + assert.Equal(t, versionedcache.ReloadedAtMaxStaleness, reported.seen()[2]) +} + +// A panic in the loader has to become a failed cycle. otter runs a background reload on a +// goroutine it starts itself and re-panics whatever the loader panicked with, so without a +// recover the process ends on the refresh timer instead of reporting one bad cycle. +func TestALoaderThatPanicsIsReportedAsAFailedCycle(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + reported := newCycles(t) + require.NoError(t, server.Set(exampleVersionKey, "v1")) + + var reads atomic.Int64 + loadData := func(context.Context) (string, error) { + if reads.Add(1) > 1 { + panic("boom from loadData") + } + return "first", nil + } + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), loadData, time.Minute, + versionedcache.Options{Now: clock.Now, WhenRefreshed: reported.record}) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + require.NoError(t, server.Set(exampleVersionKey, "v2")) + clock.Advance(2 * time.Minute) + refreshCycle(t, cache, reported) + + assert.Equal(t, versionedcache.RefreshFailed, reported.seen()[1]) + require.Error(t, reported.lastError()) + assert.Contains(t, reported.lastError().Error(), "loadData panicked") + + served, err := cache.Get(context.Background()) + require.NoError(t, err) + assert.Equal(t, "first", served, "a panicking reload keeps the data the cache already held") +} + +// Same for the version read, which runs on the same background goroutine. +func TestAVersionReadThatPanicsIsReportedAsAFailedCycle(t *testing.T) { + clock := newFakeClock() + reported := newCycles(t) + + var checks atomic.Int64 + readVersion := func(context.Context) (string, bool, error) { + if checks.Add(1) > 1 { + panic("boom from ReadVersion") + } + return "v1", true, nil + } + + cache := versionedcache.New(readVersion, newAnswers("first").load, time.Minute, + versionedcache.Options{Now: clock.Now, WhenRefreshed: reported.record}) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + clock.Advance(2 * time.Minute) + refreshCycle(t, cache, reported) + + assert.Equal(t, versionedcache.RefreshFailed, reported.seen()[1]) + require.Error(t, reported.lastError()) + assert.Contains(t, reported.lastError().Error(), "ReadVersion panicked") +} + +// Each of these builds a cache that compiles, loads once and is quietly wrong for the rest of +// the process. New has no error return, so refusing them has to happen before the first tag. +func TestNewRejectsSettingsThatWouldNeverRefresh(t *testing.T) { + _, client := newTestClient(t) + readVersion := versionedcache.VersionInKey(client, exampleVersionKey) + loadData := newAnswers("first").load + + cases := []struct { + name string + build func() + }{ + {"an interval of zero", func() { + versionedcache.New(readVersion, loadData, 0, versionedcache.Options{}) + }}, + {"a negative interval", func() { + versionedcache.New(readVersion, loadData, -time.Second, versionedcache.Options{}) + }}, + {"a negative staleness limit", func() { + versionedcache.New(readVersion, loadData, time.Minute, versionedcache.Options{MaxStaleness: -time.Second}) + }}, + {"no loader", func() { + versionedcache.New[string](readVersion, nil, time.Minute, versionedcache.Options{}) + }}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + assert.Panics(t, c.build) + }) + } +} + +// The empty string is what a missing key reads as, and it is also a value a publisher can write. +// An entry loaded while nothing was published must not compare equal to a published empty one. +func TestAPublishedEmptyVersionIsNotMistakenForNothingPublished(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + data := newAnswers("first") + reported := newCycles(t) + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), data.load, time.Minute, + versionedcache.Options{Now: clock.Now, WhenRefreshed: reported.record}) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + require.Equal(t, versionedcache.ReloadedWithoutVersion, reported.seen()[0]) + + // The publisher starts publishing, with an empty value, and changes the data with it. + data.set("second") + require.NoError(t, server.Set(exampleVersionKey, "")) + clock.Advance(2 * time.Minute) + refreshCycle(t, cache, reported) + + assert.Equal(t, 2, data.timesRead(), "an empty published version is a version, not the absence of one") + assertServes(t, cache, "second") +} + +// A first load with a version already published has nothing to compare against, so it counts as +// a change. This is a live dashboard tag value, so the outcome is asserted rather than assumed. +func TestAFirstLoadWithAVersionPublishedReportsReloadedAfterChange(t *testing.T) { + server, client := newTestClient(t) + reported := newCycles(t) + require.NoError(t, server.Set(exampleVersionKey, "v1")) + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), + newAnswers("first").load, time.Minute, versionedcache.Options{WhenRefreshed: reported.record}) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + require.Len(t, reported.seen(), 1) + assert.Equal(t, versionedcache.ReloadedAfterChange, reported.seen()[0]) +} + +// A caller's reporter is the third callback that runs on the store's refresh goroutine, so a +// panic in it would end the process the same way a panicking loader would. +func TestAReporterThatPanicsDoesNotEndTheProcess(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + require.NoError(t, server.Set(exampleVersionKey, "20260912.1")) + + var reports atomic.Int64 + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), + newAnswers("first").load, time.Minute, + versionedcache.Options{Now: clock.Now, WhenRefreshed: func(versionedcache.RefreshOutcome, error) { + if reports.Add(1) == 2 { + panic("the caller's reporter panicked") + } + }}) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + clock.Advance(2 * time.Minute) + _, err = cache.Get(context.Background()) + require.NoError(t, err) + require.Eventually(t, func() bool { return reports.Load() >= 2 }, + 5*time.Second, 5*time.Millisecond, "waited for the cycle that panics") + + clock.Advance(2 * time.Minute) + _, err = cache.Get(context.Background()) + require.NoError(t, err) + assert.Eventually(t, func() bool { return reports.Load() >= 3 }, + 5*time.Second, 5*time.Millisecond, "the cycle after the panicking one must still run") +} + +// The interval is what keeps a source that is down from being read on every call. The store puts +// the data back on the clock only once a cycle finishes, so without a claim of our own every call +// arriving while a slow failure is in flight starts a cycle of its own. +func TestAFailingSourceIsReadOncePerIntervalNotOncePerGet(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + require.NoError(t, server.Set(exampleVersionKey, "20260912.1")) + + var versionReads atomic.Int64 + var failing atomic.Bool + published := versionedcache.VersionInKey(client, exampleVersionKey) + counted := func(ctx context.Context) (string, bool, error) { + versionReads.Add(1) + if failing.Load() { + // A real outage times out rather than answering at once, and the window that failure + // leaves open is the whole point of this test. + time.Sleep(2 * time.Millisecond) + return "", false, errors.New("the source is down") + } + return published(ctx) + } + + cache := versionedcache.New(counted, newAnswers("first").load, time.Minute, + versionedcache.Options{Now: clock.Now}) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + failing.Store(true) + readsBefore := versionReads.Load() + clock.Advance(2 * time.Minute) + + // Live traffic arriving while the source is down, with the clock held still. It has to keep + // arriving: the calls that pile up behind one failing read are folded into it, and the cycle + // that should not happen is the one a later call starts once that read has failed. + stop := make(chan struct{}) + var callers sync.WaitGroup + for range 50 { + callers.Add(1) + go func() { + defer callers.Done() + for { + select { + case <-stop: + return + default: + served, err := cache.Get(context.Background()) + assert.NoError(t, err) + assert.Equal(t, "first", served) + } + } + }() + } + time.Sleep(300 * time.Millisecond) + close(stop) + callers.Wait() + + assert.Equal(t, int64(1), versionReads.Load()-readsBefore, + "one interval elapsed, so the source is read once however long the traffic keeps arriving") +} + +// otter reads its own ErrNotFound as the dataset being gone and drops what it holds. A loader that +// wraps that sentinel would therefore empty a warm cache on one failed read. +func TestAFailedReadReportingNotFoundKeepsTheWarmData(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + reported := newCycles(t) + require.NoError(t, server.Set(exampleVersionKey, "20260912.1")) + + var failing atomic.Bool + loadData := func(context.Context) (string, error) { + if failing.Load() { + return "", fmt.Errorf("read the dataset: %w", otter.ErrNotFound) + } + return "first", nil + } + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), + loadData, time.Minute, versionedcache.Options{Now: clock.Now, WhenRefreshed: reported.record}) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + failing.Store(true) + require.NoError(t, server.Set(exampleVersionKey, "20260912.2")) + clock.Advance(2 * time.Minute) + refreshCycle(t, cache, reported) + + assert.Equal(t, versionedcache.RefreshFailed, reported.seen()[len(reported.seen())-1]) + assertServes(t, cache, "first") +} diff --git a/versionedcache/go.mod b/versionedcache/go.mod new file mode 100644 index 0000000..9112379 --- /dev/null +++ b/versionedcache/go.mod @@ -0,0 +1,20 @@ +module github.com/wego/pkg/versionedcache + +go 1.25.0 + +require ( + github.com/alicebob/miniredis/v2 v2.39.0 + github.com/maypok86/otter/v2 v2.3.0 + github.com/redis/go-redis/v9 v9.22.0 + github.com/stretchr/testify v1.11.1 +) + +require ( + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/yuin/gopher-lua v1.1.1 // indirect + go.uber.org/atomic v1.11.0 // indirect + golang.org/x/sys v0.30.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/versionedcache/go.sum b/versionedcache/go.sum new file mode 100644 index 0000000..3c07d31 --- /dev/null +++ b/versionedcache/go.sum @@ -0,0 +1,32 @@ +github.com/alicebob/miniredis/v2 v2.39.0 h1:M7WbmV5BmV56L8KTG0rw6vEQ+woTOghpDgin2xv4A0g= +github.com/alicebob/miniredis/v2 v2.39.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= +github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +github.com/maypok86/otter/v2 v2.3.0 h1:8H8AVVFUSzJwIegKwv1uF5aGitTY+AIrtktg7OcLs8w= +github.com/maypok86/otter/v2 v2.3.0/go.mod h1:XgIdlpmL6jYz882/CAx1E4C1ukfgDKSaw4mWq59+7l8= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/redis/go-redis/v9 v9.22.0 h1:laDvpYXTJtZLloinw1fA5Kqd6HAEH2XKxOkG/PDq2F0= +github.com/redis/go-redis/v9 v9.22.0/go.mod h1:y2g0Wj8rQvuK0ELM+oxSudcLtC09JScs98I/X9gRWY4= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= +github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= +golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc= +golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/versionedcache/otter_adapters.go b/versionedcache/otter_adapters.go new file mode 100644 index 0000000..f14e447 --- /dev/null +++ b/versionedcache/otter_adapters.go @@ -0,0 +1,36 @@ +package versionedcache + +import ( + "time" + + "github.com/maypok86/otter/v2" +) + +// refreshEvery puts the dataset back on the clock after every cycle, a failed one included. +// otter's own RefreshWriting leaves a failed reload still due, which would send the very next +// call back to a source already known to be down. +type refreshEvery[T any] struct{ interval time.Duration } + +func (r refreshEvery[T]) RefreshAfterCreate(otter.Entry[string, cacheEntry[T]]) time.Duration { + return r.interval +} + +func (r refreshEvery[T]) RefreshAfterUpdate(otter.Entry[string, cacheEntry[T]], cacheEntry[T]) time.Duration { + return r.interval +} + +func (r refreshEvery[T]) RefreshAfterReload(otter.Entry[string, cacheEntry[T]], cacheEntry[T]) time.Duration { + return r.interval +} + +func (r refreshEvery[T]) RefreshAfterReloadFailure(otter.Entry[string, cacheEntry[T]], error) time.Duration { + return r.interval +} + +// clockFrom lets Options.Now drive the interval. Tick is only asked for by a cache that expires +// entries early, which this one never configures, so the real ticker is right there. +type clockFrom struct{ now func() time.Time } + +func (c *clockFrom) NowNano() int64 { return c.now().UnixNano() } + +func (c *clockFrom) Tick(every time.Duration) <-chan time.Time { return time.Tick(every) } diff --git a/versionedcache/refresh_outcome.go b/versionedcache/refresh_outcome.go new file mode 100644 index 0000000..701b0dc --- /dev/null +++ b/versionedcache/refresh_outcome.go @@ -0,0 +1,33 @@ +package versionedcache + +// RefreshOutcome says what one refresh cycle did. Callers turn it into a metric tag; +// this package deliberately depends on no metrics library of its own. +type RefreshOutcome string + +const ( + // VersionUnchanged means the published version matched the one the cached data was + // loaded under, so the data was not read. This is the cycle the package exists for. + VersionUnchanged RefreshOutcome = "version_unchanged" + + // ReloadedAfterChange means the version differed from the one the cached data was + // loaded under, so the data was read again. A first load with a version published + // reports this too, because there is nothing cached to compare against. + ReloadedAfterChange RefreshOutcome = "reloaded_after_change" + + // ReloadedWithoutVersion means there was no version to check — either the caller gave + // no ReadVersion, or nothing is published — so the data was read again. + ReloadedWithoutVersion RefreshOutcome = "reloaded_without_version" + + // ReloadedAtMaxStaleness means the version had not moved but the data had been kept for + // Options.MaxStaleness, so it was read again in case the version stopped moving for some + // reason other than the data being unchanged. + ReloadedAtMaxStaleness RefreshOutcome = "reloaded_at_max_staleness" + + // RefreshFailed means reading the version or the data failed and the previous value is + // still being served. The cache is warm, so the caller is running on slightly old data. + RefreshFailed RefreshOutcome = "refresh_failed" + + // FirstLoadFailed means that same read failed with nothing ever loaded. There is + // nothing to serve, Get returned the error, and the caller is on its own defaults. + FirstLoadFailed RefreshOutcome = "first_load_failed" +) From 4499eed331049fe3f2245f7184ed37a4655510cf Mon Sep 17 00:00:00 2001 From: Aaron Asuncion <131644118+aaron-wego@users.noreply.github.com> Date: Fri, 18 Sep 2026 10:25:57 +0800 Subject: [PATCH 2/8] Update versionedcache/README.md Co-authored-by: lei-wego <84778813+lei-wego@users.noreply.github.com> --- versionedcache/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/versionedcache/README.md b/versionedcache/README.md index 560aacb..99f57bd 100644 --- a/versionedcache/README.md +++ b/versionedcache/README.md @@ -76,7 +76,7 @@ metric tag is `string(outcome)` with nothing to branch on. | Outcome | Meaning | | --- | --- | | `version_unchanged` | the version matched, so nothing was read | -| `reloaded_after_change` | the version moved, so the data was read again | +| `reloaded_after_change` | the version moved, so the data was read again, and a first load with a version published reports this too | | `reloaded_without_version` | there was no version to compare, so the data was read again | | `reloaded_at_max_staleness` | the version had not moved but the data had been kept for `MaxStaleness`, so it was read again | | `refresh_failed` | the read failed and the previous value is still being served | From bcb2ad3f66ed01a6317ecf7a522ea35638c42aaa Mon Sep 17 00:00:00 2001 From: Aaron Asuncion Date: Fri, 18 Sep 2026 13:22:50 +0800 Subject: [PATCH 3/8] versionedcache: count the refresh interval on a clock that cannot go backwards The interval claim that came with the refresh gate read wall time through UnixNano, which a clock correction moves backwards. The store's own clock counts from a monotonic reading and goes on asking for cycles regardless, so a correction would have held every cycle back until wall time caught up. - keep the last cycle's start as a time.Time and compare with Sub, so the gate counts elapsed time the way the store's clock does - read a negative gap as a clock that was reset, which covers a caller whose Now carries no monotonic reading of its own - add an internal test for a backward step, which fails without the guard - say in the Options.WhenRefreshed docs that calling Get from there never returns, since it runs inside the cycle - note in the README that a call arriving during a refresh waits on a goroutine - shorten the comments added over the last two rounds and drop the jargon Co-Authored-By: Claude Opus 5 (1M context) --- versionedcache/README.md | 2 + versionedcache/cache.go | 58 ++++++++++++++------------- versionedcache/cache_internal_test.go | 14 +++++++ 3 files changed, 47 insertions(+), 27 deletions(-) diff --git a/versionedcache/README.md b/versionedcache/README.md index 99f57bd..134ffcc 100644 --- a/versionedcache/README.md +++ b/versionedcache/README.md @@ -91,6 +91,8 @@ the data the cache already holds, and starts the refresh; a dataset that changed call. Nothing waits on Redis once the cache is warm, and the worst-case staleness is the interval plus one refresh. The refresh also runs with the caller's cancellation stripped, so a deadline on the `Get` that started it does not cut it short; the Redis client's own timeouts bound it instead. +Each call arriving while a refresh is still running starts a goroutine that waits for it, so a slow +Redis is a short-lived spike in goroutines bounded by that client's timeouts. **Get returns an error only when it has never loaded.** Once a load has succeeded, a failure to read either the version or the data keeps the previous value and `Get` returns it with a diff --git a/versionedcache/cache.go b/versionedcache/cache.go index 82c5cdf..65a0932 100644 --- a/versionedcache/cache.go +++ b/versionedcache/cache.go @@ -58,6 +58,9 @@ func publishedVersion(version string, err error, location string) (string, bool, type Options struct { // WhenRefreshed is called once per refresh cycle with what the cycle did and any error // that stopped it. This is where a caller emits its own metric and log line. + // + // It runs inside the cycle, so calling Get on this cache from here waits on the cycle it is + // itself part of and never returns. A panic here is caught and the report dropped. WhenRefreshed func(RefreshOutcome, error) // MaxStaleness is how long the data may be kept on an unchanged version before it is read @@ -106,8 +109,8 @@ type Cache[T any] struct { // which is what every caller did before this package existed. // // New panics on a nil loadData, a checkEvery that is not positive, or a negative -// Options.MaxStaleness. Each of those builds a cache that compiles and is then quietly wrong for -// the life of the process, so it is refused here rather than at the first cycle. +// Options.MaxStaleness. Each of those builds a cache that compiles and then never refreshes +// properly, so it is refused here rather than at the first cycle. func New[T any]( readVersion ReadVersion, loadData func(context.Context) (T, error), @@ -117,9 +120,8 @@ func New[T any]( if loadData == nil { panic("versionedcache: loadData must not be nil") } - // Both of these otherwise produce a cache that compiles, loads once and is quietly wrong: - // otter arms no refresh for an interval that is not positive, and a negative staleness limit - // makes every cycle read the whole dataset. + // Without these: an interval that is not positive never refreshes at all, and a negative + // staleness limit reads the whole dataset every cycle. if checkEvery <= 0 { panic("versionedcache: checkEvery must be greater than zero") } @@ -178,9 +180,10 @@ type refreshCycle[T any] struct { whenRefreshed func(RefreshOutcome, error) now func() time.Time - // startedAt is when the last cycle began, as Unix nanoseconds, and zero before the first one. - // startCycle below is the only writer. - startedAt atomic.Int64 + // startedAt is when the last cycle began, nil before the first one. Kept as a time.Time so the + // comparison below counts elapsed time the way the store's own clock does, and a clock set + // backwards does not stall every cycle after it. + startedAt atomic.Pointer[time.Time] // newest is what the last finished cycle produced. The store hands Reload the value the // cache held when that call's Get ran, which is out of date for a cycle that was queued @@ -241,22 +244,24 @@ func (c *refreshCycle[T]) Reload(ctx context.Context, _ string, cached cacheEntr return c.loadedNow(data, version, published), nil } -// startCycle claims the current interval, so that of the callers who find the data due only the -// first goes on to read. The store puts the data back on the clock when a cycle finishes, not when -// it starts, so while a read is failing the data stays due for as long as that read takes and every -// call arriving meanwhile starts a cycle of its own — a source already known to be down is then -// read continuously rather than once an interval. +// startCycle claims the interval, so only the first caller that finds the data due goes on to read +// it. The store puts the data back on the clock when a cycle finishes, not when it starts, so a +// slow failing read leaves it due meanwhile and every call arriving starts its own cycle — reading +// a source already known to be down over and over. // -// A cycle that loses this claim reports nothing, because none ran: the caller is counting cycles, -// and a call that did no work is not one. +// A cycle that loses the claim reports nothing, because none ran. func (c *refreshCycle[T]) startCycle() bool { - now := c.now().UnixNano() + now := c.now() for { started := c.startedAt.Load() - if started != 0 && now-started < int64(c.interval) { - return false + // A negative gap means the clock was set backwards. Run the cycle rather than wait for + // the clock to reach a time it has already been. + if started != nil { + if since := now.Sub(*started); since >= 0 && since < c.interval { + return false + } } - if c.startedAt.CompareAndSwap(started, now) { + if c.startedAt.CompareAndSwap(started, &now) { return true } } @@ -268,9 +273,8 @@ func (c *refreshCycle[T]) failed(outcome RefreshOutcome, err error) (cacheEntry[ c.report(outcome, err) var nothing cacheEntry[T] if errors.Is(err, otter.ErrNotFound) { - // The store reads that sentinel as the dataset being gone and drops what it holds, which - // would empty a warm cache after one failed read. Flatten the chain rather than pass it on: - // a loader reporting it means the read failed, not that the data no longer exists. + // The store reads that error as the dataset being gone and throws away what it holds. A + // loader reporting it means the read failed, so flatten it rather than pass it on. return nothing, fmt.Errorf("versionedcache: %s", err) } return nothing, err @@ -290,9 +294,9 @@ func (c *refreshCycle[T]) currentVersion(ctx context.Context) (version string, p return c.readVersion(ctx) } -// loadDataNow reads the dataset, turning a panic into an error. otter starts a background reload -// on a goroutine of its own and re-panics whatever the loader panicked with, so an unrecovered -// panic there ends the process instead of reporting one failed cycle. +// loadDataNow reads the dataset, turning a panic into an error. The store runs a background reload +// on a goroutine of its own and re-raises whatever the loader panicked with, so a panic left alone +// there ends the process instead of failing one cycle. func (c *refreshCycle[T]) loadDataNow(ctx context.Context) (data T, err error) { defer func() { if r := recover(); r != nil { @@ -336,8 +340,8 @@ func reloadReason(published, versionUnchanged bool) RefreshOutcome { } // report hands the cycle's outcome to the caller. It runs on the store's refresh goroutine, like -// the two reads above it, so a panic here would end the process. There is nowhere to report a -// reporter that failed, so the report is dropped and the next cycle still runs. +// the two reads above, so a panic here would end the process. Nothing can report a broken +// reporter, so the report is dropped and the next cycle still runs. func (c *refreshCycle[T]) report(outcome RefreshOutcome, err error) { if c.whenRefreshed == nil { return diff --git a/versionedcache/cache_internal_test.go b/versionedcache/cache_internal_test.go index 0927fb4..df43238 100644 --- a/versionedcache/cache_internal_test.go +++ b/versionedcache/cache_internal_test.go @@ -40,3 +40,17 @@ func TestAReloadHandedAnOutOfDateEntryComparesAgainstTheLastCycleInstead(t *test assert.Equal(t, 1, reads, "a cycle must compare against what the last finished cycle loaded") assert.Equal(t, "data for v2", kept.data) } + +// The store's clock never goes backwards, so it keeps asking for cycles however the wall clock +// moves. Counting the interval on wall time would stall them for the whole of a clock correction. +// Driven directly because Options.Now is the store's clock too, so a test cannot move just one. +func TestCyclesAreNotHeldBackByAClockThatMovesBackwards(t *testing.T) { + now := time.Now() + cycle := &refreshCycle[string]{interval: time.Minute, now: func() time.Time { return now }} + + require.True(t, cycle.startCycle(), "the first cycle has nothing to wait behind") + require.False(t, cycle.startCycle(), "a second inside the interval waits for it") + + now = now.Add(-time.Hour) + assert.True(t, cycle.startCycle(), "a clock that moved back must not hold the cycle back with it") +} From ef69a0f4dacada9c5a9a51cd9555b9e77d2c5cd7 Mon Sep 17 00:00:00 2001 From: Aaron Asuncion Date: Fri, 18 Sep 2026 14:31:21 +0800 Subject: [PATCH 4/8] versionedcache: read the clock inside the claim loop, not before it A caller that read the clock before the loop and then lost the swap compared its own earlier reading against the winner's stamp. That gap comes back negative, which the guard reads as a clock that moved, so the interval was claimed a second time and the stamp went backwards. Sixty-four callers entering together reached two claims; reading the clock inside the loop puts that back to one, because a caller that retries reads a time at or after the winner's, while a clock that really moved still reads negative. Not reachable through Get today, since the store runs one reload per key at a time, but startCycle claims the property for itself and nothing here rests on the store being what holds it up. - build the backward-clock test's fixture with Round(0) so it carries no monotonic reading, which is the case the guard exists for; a fixture built by adding to time.Now keeps one and models a state the real clock cannot reach, so the test passed without exercising the guard - name that test for the clock it uses Co-Authored-By: Claude Opus 5 (1M context) --- versionedcache/cache.go | 8 +++++--- versionedcache/cache_internal_test.go | 12 +++++++----- 2 files changed, 12 insertions(+), 8 deletions(-) diff --git a/versionedcache/cache.go b/versionedcache/cache.go index 65a0932..3e48bf9 100644 --- a/versionedcache/cache.go +++ b/versionedcache/cache.go @@ -251,11 +251,13 @@ func (c *refreshCycle[T]) Reload(ctx context.Context, _ string, cached cacheEntr // // A cycle that loses the claim reports nothing, because none ran. func (c *refreshCycle[T]) startCycle() bool { - now := c.now() for { started := c.startedAt.Load() - // A negative gap means the clock was set backwards. Run the cycle rather than wait for - // the clock to reach a time it has already been. + now := c.now() + // A negative gap means the clock was set backwards. Run the cycle rather than wait for the + // clock to reach a time it has already been. Reading the clock inside the loop is what + // makes that safe: a caller that loses the swap reads it again, so a negative gap is only + // ever a clock that really moved, not a caller that sampled early and lost. if started != nil { if since := now.Sub(*started); since >= 0 && since < c.interval { return false diff --git a/versionedcache/cache_internal_test.go b/versionedcache/cache_internal_test.go index df43238..71573b0 100644 --- a/versionedcache/cache_internal_test.go +++ b/versionedcache/cache_internal_test.go @@ -41,11 +41,13 @@ func TestAReloadHandedAnOutOfDateEntryComparesAgainstTheLastCycleInstead(t *test assert.Equal(t, "data for v2", kept.data) } -// The store's clock never goes backwards, so it keeps asking for cycles however the wall clock -// moves. Counting the interval on wall time would stall them for the whole of a clock correction. -// Driven directly because Options.Now is the store's clock too, so a test cannot move just one. -func TestCyclesAreNotHeldBackByAClockThatMovesBackwards(t *testing.T) { - now := time.Now() +// A caller's clock with no monotonic reading, which is the case the guard exists for: time.Now +// carries one and Sub counts on it whatever the wall clock does, so only a clock without one can +// report time going backwards. Driven directly because Options.Now is the store's clock too, so a +// test cannot move just one of them. +func TestCyclesAreNotHeldBackByAClockWithNoMonotonicReading(t *testing.T) { + // Off the real clock, as every fixture here is, with Round(0) dropping the monotonic reading. + now := time.Now().Round(0) cycle := &refreshCycle[string]{interval: time.Minute, now: func() time.Time { return now }} require.True(t, cycle.startCycle(), "the first cycle has nothing to wait behind") From 582c1742989c55a9e6e5a673c7dab73bf728cc38 Mon Sep 17 00:00:00 2001 From: Aaron Asuncion Date: Fri, 18 Sep 2026 15:07:04 +0800 Subject: [PATCH 5/8] versionedcache: test that a cycle losing the claim swap re-reads the clock The claim loop had no test that entered it twice, which is why both of its regressions landed with the suite green: hoisting the clock read back above the loop still passes every other test in the package. This drives the lost-swap window through the injected clock, so it needs no goroutines and depends on no timing: the first read stamps a later start, as a cycle that won the swap would, and returns an earlier time to the caller that lost it. That caller has to read the clock again rather than compare its own stale sample, which would look like a clock that had moved backwards. Co-Authored-By: Claude Opus 5 (1M context) --- versionedcache/cache_internal_test.go | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/versionedcache/cache_internal_test.go b/versionedcache/cache_internal_test.go index 71573b0..e89a65d 100644 --- a/versionedcache/cache_internal_test.go +++ b/versionedcache/cache_internal_test.go @@ -56,3 +56,26 @@ func TestCyclesAreNotHeldBackByAClockWithNoMonotonicReading(t *testing.T) { now = now.Add(-time.Hour) assert.True(t, cycle.startCycle(), "a clock that moved back must not hold the cycle back with it") } + +// startCycle's contract is that only one cycle holds the interval. A caller that loses the swap +// re-reads the clock, so the gap it compares is against a start that really is in the past. +func TestOnlyOneCycleHoldsTheIntervalWhenAnotherClaimsItMidSwap(t *testing.T) { + base := time.Now().Round(0) + cycle := &refreshCycle[string]{interval: time.Minute} + + reads := 0 + cycle.now = func() time.Time { + reads++ + if reads == 1 { + // Another cycle reads the clock a second later and wins the claim, which is the + // window this caller's swap has to survive. + won := base.Add(2 * time.Second) + cycle.startedAt.Store(&won) + return base.Add(1 * time.Second) + } + return base.Add(3 * time.Second) + } + + assert.False(t, cycle.startCycle(), + "a cycle that lost the swap must re-read the clock, not read its own stale sample as a clock that moved backwards") +} From 66301efc69e853c4f94acb339ac9d94be6556da2 Mon Sep 17 00:00:00 2001 From: Aaron Asuncion Date: Tue, 22 Sep 2026 10:28:25 +0800 Subject: [PATCH 6/8] fs-997-versionedcache MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - narrow the WhenRefreshed godoc: the Get deadlock is the first load only, since a later cycle is answered from memory - reject an Options.MaxStaleness shorter than checkEvery in New — every cycle lands past the limit, so the whole dataset reloads and the version decides nothing - check that limit after the zero default is filled in, so a check interval longer than six hours is caught too - drop the separate negative-MaxStaleness guard — the new check already covers it - say in the README that Get copies a map or slice header only, so callers share the cache's backing store and must treat it as read-only - add two cases to the New guard test for the short limit and the long interval - pin the six RefreshOutcome strings, which are the metric contract nothing else asserts - assert the failed version read names the key it could not read, instead of just any error Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Uhe5yXCw4Knkz9fEZdqDth --- versionedcache/README.md | 9 ++++++--- versionedcache/cache.go | 26 ++++++++++++++++---------- versionedcache/cache_test.go | 21 ++++++++++++++++++++- 3 files changed, 42 insertions(+), 14 deletions(-) diff --git a/versionedcache/README.md b/versionedcache/README.md index 134ffcc..eb38c74 100644 --- a/versionedcache/README.md +++ b/versionedcache/README.md @@ -61,11 +61,14 @@ there means nobody publishes one, so deleting it is an off switch that needs no rows and die before it publishes, or a bulk import can publish once at the very end. So data kept on an unchanged version for `Options.MaxStaleness` is read again at the next cycle anyway, and that read restarts the clock; a cycle that skipped does not. Zero means -`DefaultMaxStaleness`, six hours. Set it against how often the writer runs, and keep it longer -than the check interval, or every cycle reloads. +`DefaultMaxStaleness`, six hours. Set it against how often the writer runs. It has to be at +least the check interval — a shorter one puts every cycle past the limit, so the whole dataset +is read every time and the version decides nothing, and `New` refuses it. `Get` returns `T` by value, so it copies whatever you store on every call. Hold a map, a slice -or a pointer when the dataset is large. +or a pointer when the dataset is large. For those shapes the copy is the header alone, so every +caller shares the backing store the cache holds: treat what `Get` returns as read-only, and copy +it before mutating. ## What each cycle reports diff --git a/versionedcache/cache.go b/versionedcache/cache.go index 3e48bf9..83ba5de 100644 --- a/versionedcache/cache.go +++ b/versionedcache/cache.go @@ -59,8 +59,9 @@ type Options struct { // WhenRefreshed is called once per refresh cycle with what the cycle did and any error // that stopped it. This is where a caller emits its own metric and log line. // - // It runs inside the cycle, so calling Get on this cache from here waits on the cycle it is - // itself part of and never returns. A panic here is caught and the report dropped. + // It runs inside the cycle, so calling Get on this cache from here during the first load + // waits on the load it is itself part of and never returns. On a later cycle that Get is + // answered from memory instead. A panic here is caught and the report dropped. WhenRefreshed func(RefreshOutcome, error) // MaxStaleness is how long the data may be kept on an unchanged version before it is read @@ -108,9 +109,10 @@ type Cache[T any] struct { // A nil readVersion means the dataset carries no version, so the data reloads every checkEvery, // which is what every caller did before this package existed. // -// New panics on a nil loadData, a checkEvery that is not positive, or a negative -// Options.MaxStaleness. Each of those builds a cache that compiles and then never refreshes -// properly, so it is refused here rather than at the first cycle. +// New panics on a nil loadData, a checkEvery that is not positive, or an Options.MaxStaleness +// shorter than checkEvery — including a negative one, and including the DefaultMaxStaleness a +// zero takes. Each of those builds a cache that compiles and then never refreshes properly, so +// it is refused here rather than at the first cycle. func New[T any]( readVersion ReadVersion, loadData func(context.Context) (T, error), @@ -120,14 +122,10 @@ func New[T any]( if loadData == nil { panic("versionedcache: loadData must not be nil") } - // Without these: an interval that is not positive never refreshes at all, and a negative - // staleness limit reads the whole dataset every cycle. + // An interval that is not positive never refreshes at all. if checkEvery <= 0 { panic("versionedcache: checkEvery must be greater than zero") } - if options.MaxStaleness < 0 { - panic("versionedcache: Options.MaxStaleness must not be negative") - } now := options.Now if now == nil { now = time.Now @@ -136,6 +134,14 @@ func New[T any]( if maxStaleness == 0 { maxStaleness = DefaultMaxStaleness } + // A limit the interval always clears puts every cycle past it, so the whole dataset is read + // every time and the version decides nothing. Checked after the default is filled in, because + // a zero left with a check interval longer than six hours lands in the same place. + if maxStaleness < checkEvery { + panic(fmt.Sprintf( + "versionedcache: Options.MaxStaleness (%s) must not be shorter than checkEvery (%s)", + maxStaleness, checkEvery)) + } store := &otter.Options[string, cacheEntry[T]]{ RefreshCalculator: refreshEvery[T]{interval: checkEvery}, // Every failure is already reported through WhenRefreshed, where the caller's log line diff --git a/versionedcache/cache_test.go b/versionedcache/cache_test.go index 55248c2..b097b47 100644 --- a/versionedcache/cache_test.go +++ b/versionedcache/cache_test.go @@ -414,7 +414,8 @@ func TestFailedVersionReadKeepsTheCachedData(t *testing.T) { require.NoError(t, err, "a warm cache keeps serving through a Redis outage") assert.Equal(t, "first", served) assert.Equal(t, 1, data.timesRead(), "the data must not be read when the version could not be") - assert.Error(t, reported.lastError(), "the failure is reported even though Get succeeded") + assert.ErrorContains(t, reported.lastError(), exampleVersionKey, + "the failure is reported even though Get succeeded, and names the key it could not read") } func TestFailedDataReadKeepsTheCachedData(t *testing.T) { @@ -876,6 +877,12 @@ func TestNewRejectsSettingsThatWouldNeverRefresh(t *testing.T) { {"a negative staleness limit", func() { versionedcache.New(readVersion, loadData, time.Minute, versionedcache.Options{MaxStaleness: -time.Second}) }}, + {"a staleness limit shorter than the interval", func() { + versionedcache.New(readVersion, loadData, time.Minute, versionedcache.Options{MaxStaleness: 30 * time.Second}) + }}, + {"an interval longer than the staleness limit a zero takes", func() { + versionedcache.New(readVersion, loadData, versionedcache.DefaultMaxStaleness+time.Minute, versionedcache.Options{}) + }}, {"no loader", func() { versionedcache.New[string](readVersion, nil, time.Minute, versionedcache.Options{}) }}, @@ -887,6 +894,18 @@ func TestNewRejectsSettingsThatWouldNeverRefresh(t *testing.T) { } } +// A consumer tags its metric with string(outcome), so these six literals are the contract and +// renaming one silently renames someone's dashboard series. Nothing else in the suite asserts a +// literal value, so they are pinned here. +func TestTheOutcomeStringsAreTheMetricContract(t *testing.T) { + assert.Equal(t, "version_unchanged", string(versionedcache.VersionUnchanged)) + assert.Equal(t, "reloaded_after_change", string(versionedcache.ReloadedAfterChange)) + assert.Equal(t, "reloaded_without_version", string(versionedcache.ReloadedWithoutVersion)) + assert.Equal(t, "reloaded_at_max_staleness", string(versionedcache.ReloadedAtMaxStaleness)) + assert.Equal(t, "refresh_failed", string(versionedcache.RefreshFailed)) + assert.Equal(t, "first_load_failed", string(versionedcache.FirstLoadFailed)) +} + // The empty string is what a missing key reads as, and it is also a value a publisher can write. // An entry loaded while nothing was published must not compare equal to a published empty one. func TestAPublishedEmptyVersionIsNotMistakenForNothingPublished(t *testing.T) { From b2280d7a2bc77d367996a87a9ef0c485093be290 Mon Sep 17 00:00:00 2001 From: Aaron Asuncion Date: Tue, 22 Sep 2026 12:03:49 +0800 Subject: [PATCH 7/8] fs-997-versionedcache MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - say in the README and on New that loadData gets no deadline, so its timeout has to go inside it — otter strips the deadline along with the cancellation, so nothing else bounds it - fix the README's Use section, which still listed the negative-only MaxStaleness guard that 66301ef replaced - drop "the zero value is fine" from the Options godoc — a zero MaxStaleness fails the floor once checkEvery is longer than six hours - reword the startCycle comment: the store already deduplicates a slow read, so what the guard bounds is a quick one that leaves the data due again straight away - add a failed version read to the versionPlaces table, so the hash shape's error path is covered instead of only the string key's - add a test for deleting a published empty version, the off switch the README advertises - report an outcome in the first-load failure test, so FirstLoadFailed is no longer unasserted on the data path - add a test that a background reload sees neither the caller's cancellation nor a deadline - fail fast when dialling a closed miniredis instead of retrying for 1.7s — suite goes from 10.5s to 5.1s under -race Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Uhe5yXCw4Knkz9fEZdqDth --- versionedcache/README.md | 14 ++-- versionedcache/cache.go | 13 ++-- versionedcache/cache_test.go | 128 +++++++++++++++++++++++++++++++++-- 3 files changed, 139 insertions(+), 16 deletions(-) diff --git a/versionedcache/README.md b/versionedcache/README.md index eb38c74..59ef84d 100644 --- a/versionedcache/README.md +++ b/versionedcache/README.md @@ -38,8 +38,10 @@ configs, err := settings.Get(ctx) ``` `New` refuses a cache that would be quietly wrong for the life of the process, and panics on a -nil loader, a `checkEvery` that is not positive, or a negative `Options.MaxStaleness`. There is no -error to return, and each of those builds a cache that compiles and then never refreshes properly. +nil loader, a `checkEvery` that is not positive, or an `Options.MaxStaleness` shorter than +`checkEvery`, including a negative one and including the six-hour default a zero takes. There is +no error to return, and each of those builds a cache that compiles and then never refreshes +properly. **A string key of its own**, read with `GET`. Use this when the version has no hash to sit in. @@ -92,10 +94,10 @@ The `error` argument carries detail for the log line, never the classification. **A refresh runs in the background.** The call that finds the interval elapsed is answered with the data the cache already holds, and starts the refresh; a dataset that changed reaches a later call. Nothing waits on Redis once the cache is warm, and the worst-case staleness is the interval -plus one refresh. The refresh also runs with the caller's cancellation stripped, so a deadline on the -`Get` that started it does not cut it short; the Redis client's own timeouts bound it instead. -Each call arriving while a refresh is still running starts a goroutine that waits for it, so a slow -Redis is a short-lived spike in goroutines bounded by that client's timeouts. +plus one refresh. The refresh runs with the caller's cancellation and deadline stripped, so a +deadline on the `Get` that started it does not cut it short. The Redis client bounds its own +reads, but nothing bounds `loadData`: put any timeout it needs inside it, or a load that hangs +parks every arriving `Get` on a goroutine and reports no outcome at all. **Get returns an error only when it has never loaded.** Once a load has succeeded, a failure to read either the version or the data keeps the previous value and `Get` returns it with a diff --git a/versionedcache/cache.go b/versionedcache/cache.go index 83ba5de..375686f 100644 --- a/versionedcache/cache.go +++ b/versionedcache/cache.go @@ -53,8 +53,8 @@ func publishedVersion(version string, err error, location string) (string, bool, return version, true, nil } -// Options are the parts of a cache a caller may leave out. The zero value is fine: nothing -// is reported and the real clock is used. +// Options are the parts of a cache a caller may leave out: nothing is reported, the real clock +// is used, and MaxStaleness takes DefaultMaxStaleness. type Options struct { // WhenRefreshed is called once per refresh cycle with what the cycle did and any error // that stopped it. This is where a caller emits its own metric and log line. @@ -109,6 +109,9 @@ type Cache[T any] struct { // A nil readVersion means the dataset carries no version, so the data reloads every checkEvery, // which is what every caller did before this package existed. // +// loadData is handed a context with no deadline and no cancellation, because a refresh must +// outlive the Get that started it. Put any timeout it needs inside it. +// // New panics on a nil loadData, a checkEvery that is not positive, or an Options.MaxStaleness // shorter than checkEvery — including a negative one, and including the DefaultMaxStaleness a // zero takes. Each of those builds a cache that compiles and then never refreshes properly, so @@ -251,9 +254,9 @@ func (c *refreshCycle[T]) Reload(ctx context.Context, _ string, cached cacheEntr } // startCycle claims the interval, so only the first caller that finds the data due goes on to read -// it. The store puts the data back on the clock when a cycle finishes, not when it starts, so a -// slow failing read leaves it due meanwhile and every call arriving starts its own cycle — reading -// a source already known to be down over and over. +// it. A slow read is already deduplicated by the store, so what this bounds is a quick one: the +// store does not re-check freshness when a cycle ends, so the next call arriving finds the data +// still due and starts another, over and over — reading a source that may already be down. // // A cycle that loses the claim reports nothing, because none ran. func (c *refreshCycle[T]) startCycle() bool { diff --git a/versionedcache/cache_test.go b/versionedcache/cache_test.go index b097b47..e3fd720 100644 --- a/versionedcache/cache_test.go +++ b/versionedcache/cache_test.go @@ -158,7 +158,13 @@ func assertServes(t *testing.T, cache *versionedcache.Cache[string], want string func newTestClient(t *testing.T) (*miniredis.Miniredis, *redis.Client) { t.Helper() server := miniredis.RunT(t) - client := redis.NewClient(&redis.Options{Addr: server.Addr()}) + // Stock options retry three times and wait five seconds to dial, so every test that closes + // the server pays 1.7s of backoff, against a 5s budget in waitUntil. Fail fast instead. + client := redis.NewClient(&redis.Options{ + Addr: server.Addr(), + MaxRetries: -1, + DialTimeout: 100 * time.Millisecond, + }) t.Cleanup(func() { _ = client.Close() }) return server, client } @@ -173,7 +179,9 @@ const ( // versionPlace is one of the two places a version can live. A behaviour that has to hold // wherever it lives is written once here and run for each place, so the two cannot drift apart. type versionPlace struct { - name string + name string + // named is how an error from this place says where it was reading. + named string publish func(*testing.T, *miniredis.Miniredis, string) build func(*redis.Client, func(context.Context) (string, error), versionedcache.Options) *versionedcache.Cache[string] } @@ -181,7 +189,8 @@ type versionPlace struct { func versionPlaces() []versionPlace { return []versionPlace{ { - name: "a string key of its own", + name: "a string key of its own", + named: exampleVersionKey, publish: func(t *testing.T, server *miniredis.Miniredis, version string) { require.NoError(t, server.Set(exampleVersionKey, version)) }, @@ -191,7 +200,8 @@ func versionPlaces() []versionPlace { }, }, { - name: "one field of a shared hash", + name: "one field of a shared hash", + named: exampleVersionHashKey + " field " + exampleVersionField, publish: func(_ *testing.T, server *miniredis.Miniredis, version string) { server.HSet(exampleVersionHashKey, exampleVersionField, version) }, @@ -478,15 +488,20 @@ func TestAFirstLoadThatCannotReadTheVersionReturnsTheError(t *testing.T) { func TestFirstLoadFailureIsReturnedToTheCaller(t *testing.T) { _, client := newTestClient(t) + reported := newCycles(t) loadData := func(context.Context) (string, error) { return "", errors.New("the hash could not be read") } - cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), loadData, time.Minute, versionedcache.Options{}) + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), loadData, time.Minute, + versionedcache.Options{WhenRefreshed: reported.record}) data, err := cache.Get(context.Background()) assert.Error(t, err, "with nothing cached there is nothing to fall back to") assert.Empty(t, data) + // The first load runs on this goroutine, so the report is already in by now. + assert.Equal(t, []versionedcache.RefreshOutcome{versionedcache.FirstLoadFailed}, reported.seen(), + "nothing is cached, so this is not a warm cache serving slightly old data") } // Nothing has ever loaded, so a caller arriving while the first load is running has no previous @@ -894,6 +909,109 @@ func TestNewRejectsSettingsThatWouldNeverRefresh(t *testing.T) { } } +// A background reload must not inherit the cancellation of the Get that started it, or a caller +// whose request ends would cut the refresh short for everyone. otter strips it today, and nothing +// in the suite held that down, so a store swap that stopped stripping would ship green. +func TestABackgroundReloadDoesNotInheritTheCallersCancellation(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + reported := newCycles(t) + require.NoError(t, server.Set(exampleVersionKey, "20260912.1")) + + var loads atomic.Int64 + cancelled := make(chan struct{}) + reloadSaw := make(chan context.Context, 4) + loadData := func(ctx context.Context) (string, error) { + if loads.Add(1) > 1 { + // Read the context only once the Get that started this cycle is cancelled. + <-cancelled + reloadSaw <- ctx + } + return "data", nil + } + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), loadData, time.Minute, + versionedcache.Options{Now: clock.Now, WhenRefreshed: reported.record}) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + // A version that moved, so the next cycle reaches the loader at all. + require.NoError(t, server.Set(exampleVersionKey, "20260912.2")) + clock.Advance(2 * time.Minute) + + ctx, cancel := context.WithCancel(context.Background()) + _, err = cache.Get(ctx) // answered from memory; the reload runs behind it + require.NoError(t, err) + cancel() + close(cancelled) + + select { + case reloadCtx := <-reloadSaw: + assert.NoError(t, reloadCtx.Err(), "the reload must outlive the Get that started it") + _, hasDeadline := reloadCtx.Deadline() + assert.False(t, hasDeadline, "and it carries no deadline, so loadData has to bound itself") + case <-time.After(5 * time.Second): + t.Fatal("the reload never reached the loader") + } +} + +// Every failure test hardcoded the string-key shape, so a hash read that swallowed its errors +// stayed green and a Redis outage on a hash-shaped cache looked like "no version published". +func TestAFailedVersionReadIsReportedWhereverTheVersionLives(t *testing.T) { + for _, place := range versionPlaces() { + t.Run(place.name, func(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + data := newAnswers("first") + reported := newCycles(t) + place.publish(t, server, "20260912.1") + + cache := place.build(client, data.load, + versionedcache.Options{Now: clock.Now, WhenRefreshed: reported.record}) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + // Redis goes away. + server.Close() + clock.Advance(2 * time.Minute) + refreshCycle(t, cache, reported) + + assert.Equal(t, versionedcache.RefreshFailed, reported.seen()[1], + "an outage is a failed cycle, not a dataset with no version") + assert.ErrorContains(t, reported.lastError(), place.named, + "the error names where it was reading") + assert.Equal(t, 1, data.timesRead(), "the data must not be read when the version could not be") + }) + } +} + +// Deleting the version is the off switch the README advertises. Data loaded on a published empty +// version must not compare equal to nothing published, or the switch is ignored for a full cycle. +func TestDeletingAPublishedEmptyVersionReadsTheDataAgain(t *testing.T) { + server, client := newTestClient(t) + clock := newFakeClock() + data := newAnswers("first") + reported := newCycles(t) + require.NoError(t, server.Set(exampleVersionKey, "")) + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), data.load, time.Minute, + versionedcache.Options{Now: clock.Now, WhenRefreshed: reported.record}) + + _, err := cache.Get(context.Background()) + require.NoError(t, err) + + server.Del(exampleVersionKey) + data.set("second") + clock.Advance(2 * time.Minute) + refreshCycle(t, cache, reported) + + assert.Equal(t, 2, data.timesRead(), "nothing published cannot match a published empty version") + assert.Equal(t, versionedcache.ReloadedWithoutVersion, reported.seen()[1]) + assertServes(t, cache, "second") +} + // A consumer tags its metric with string(outcome), so these six literals are the contract and // renaming one silently renames someone's dashboard series. Nothing else in the suite asserts a // literal value, so they are pinned here. From 6bbcb416fd0d959d82ba0c19ee55895303368a02 Mon Sep 17 00:00:00 2001 From: Aaron Asuncion Date: Tue, 22 Sep 2026 12:30:32 +0800 Subject: [PATCH 8/8] fs-997-versionedcache - correct New's godoc: only a background refresh gets a stripped context, the first load runs on the calling goroutine under that caller's deadline - add a test for that first-load half, so the asymmetry the godoc promises is asserted at both ends - reword the startCycle comment again: the store does re-arm the entry when a cycle finishes, failed ones included, so what the guard bounds is the window before that lands - say on the MaxStaleness field that New panics when the value in force is shorter than checkEvery, the six-hour default included Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Uhe5yXCw4Knkz9fEZdqDth --- versionedcache/cache.go | 16 ++++++++++------ versionedcache/cache_test.go | 24 ++++++++++++++++++++++++ 2 files changed, 34 insertions(+), 6 deletions(-) diff --git a/versionedcache/cache.go b/versionedcache/cache.go index 375686f..955e0e8 100644 --- a/versionedcache/cache.go +++ b/versionedcache/cache.go @@ -65,7 +65,8 @@ type Options struct { WhenRefreshed func(RefreshOutcome, error) // MaxStaleness is how long the data may be kept on an unchanged version before it is read - // again anyway, at the next cycle. Zero means DefaultMaxStaleness. It is the way out when a + // again anyway, at the next cycle. Zero means DefaultMaxStaleness, and New panics when the + // value in force is shorter than checkEvery, that default included. It is the way out when a // version stops moving for some reason other than the data being unchanged, such as a // writer that changed rows and died before publishing. MaxStaleness time.Duration @@ -109,8 +110,10 @@ type Cache[T any] struct { // A nil readVersion means the dataset carries no version, so the data reloads every checkEvery, // which is what every caller did before this package existed. // -// loadData is handed a context with no deadline and no cancellation, because a refresh must -// outlive the Get that started it. Put any timeout it needs inside it. +// A background refresh hands loadData a context with no deadline and no cancellation, because +// it must outlive the Get that started it. The first load is the exception: it runs on the +// calling goroutine under that caller's context, deadline and all. Put any timeout loadData +// needs inside it, and give the first Get a budget the load can finish within. // // New panics on a nil loadData, a checkEvery that is not positive, or an Options.MaxStaleness // shorter than checkEvery — including a negative one, and including the DefaultMaxStaleness a @@ -254,9 +257,10 @@ func (c *refreshCycle[T]) Reload(ctx context.Context, _ string, cached cacheEntr } // startCycle claims the interval, so only the first caller that finds the data due goes on to read -// it. A slow read is already deduplicated by the store, so what this bounds is a quick one: the -// store does not re-check freshness when a cycle ends, so the next call arriving finds the data -// still due and starts another, over and over — reading a source that may already be down. +// it. A read still in flight is already deduplicated by the store; what this bounds is the window +// after one ends. The store re-arms the entry when a cycle finishes, failed ones included, but +// calls arriving before that lands still find the data due and each start a cycle of their own, +// hammering a source that may already be down. // // A cycle that loses the claim reports nothing, because none ran. func (c *refreshCycle[T]) startCycle() bool { diff --git a/versionedcache/cache_test.go b/versionedcache/cache_test.go index e3fd720..cbb4311 100644 --- a/versionedcache/cache_test.go +++ b/versionedcache/cache_test.go @@ -909,6 +909,30 @@ func TestNewRejectsSettingsThatWouldNeverRefresh(t *testing.T) { } } +// New's godoc promises the two loads differ: this one runs on the calling goroutine, so it +// carries that caller's deadline. The background half is the test below. +func TestTheFirstLoadRunsUnderTheCallersDeadline(t *testing.T) { + server, client := newTestClient(t) + require.NoError(t, server.Set(exampleVersionKey, "20260912.1")) + + hasDeadline := make(chan bool, 4) + loadData := func(ctx context.Context) (string, error) { + _, has := ctx.Deadline() + hasDeadline <- has + return "data", nil + } + + cache := versionedcache.New(versionedcache.VersionInKey(client, exampleVersionKey), loadData, + time.Minute, versionedcache.Options{}) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _, err := cache.Get(ctx) + require.NoError(t, err) + + assert.True(t, <-hasDeadline, "a first Get needs a budget its load can finish within") +} + // A background reload must not inherit the cancellation of the Get that started it, or a caller // whose request ends would cut the refresh short for everyone. otter strips it today, and nothing // in the suite held that down, so a store swap that stopped stripping would ship green.