From f8f31c82d889f2d1bd793189cd3c0bf9f9e8238c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Christian=20Gonz=C3=A1lez=20Di=20Antonio?= Date: Sat, 22 Aug 2026 10:06:07 +0200 Subject: [PATCH] fix: four correctness defects found reviewing c3e in production use MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit All four were found by reviewing this library on the authorization path of a multi-tenant service, and each was reproduced live against Valkey before being fixed. Every fix has a test that was confirmed to fail when the fix is reverted. 1. A shared dependency set could have its TTL shortened Set issued an unconditional EXPIRE on each reverse-dependency set. Those sets are shared by every entry that depends on the same thing, so the last writer won even when its TTL was shorter. With jittered TTLs that is ordinary rather than rare: at 10% jitter on a 12h hard TTL, entries land in 10.8h-13.2h, so a dependency set can expire more than two hours before an entry still listed in it. An Invalidate landing in that window finds an empty set, cascades to nothing, and reports success — in the service that found this, a revoked role kept working until the dependent entry expired on its own. The expiry may now only be raised: EXPIRE NX sets it when the set has none, EXPIRE GT raises it when a longer-lived dependent joins. Both are needed; GT treats a key with no expiry as infinite and would refuse to set one, leaving the leak the TTL exists to prevent. 2. A failed encode failed the request fetchAndCache handled the two halves of a write inconsistently. A failing cache.Set was logged and swallowed, correctly, because the value had already been fetched — but a failing encode three lines earlier was returned to the caller. A serialization problem, which is purely a caching concern, became user-visible even though the expensive round trip had succeeded. The library is fail-open everywhere else; this one branch was fail-closed, and it turned "caching silently does not work" into "the service is down". Encode and wrapper-marshal failures are now logged and the fetched value is returned uncached, matching Set. 3. A payload that no longer decoded failed every read for the whole TTL A corrupt wrapper was already treated as a miss and refetched. A corrupt payload inside a valid wrapper was a hard error returned to the caller, so changing EncoderType on a warm cache — or a cached type whose shape moves during a rolling deploy where two versions share one server — broke every read of that key until its hard TTL expired, with no self-healing. A payload that cannot be decoded is now treated as a miss. 4. The invalidation cascade was O(nodes) sequential round trips The breadth-first walk issued one SMEMBERS per node, plus one more per dependent found, each as its own round trip. A wide dependency graph became hundreds of serial commands on the write path, against a server the read path deliberately fast-fails on. The walk now batches per level with DoMulti: one round trip for the level's dependency sets, one for their dependents' forward lists. Cost is now proportional to graph depth rather than node count. Verified against Valkey 9.1.1: full suite green with -race, 89.3% coverage, golangci-lint clean. Co-Authored-By: Claude Opus 5 --- correctness_test.go | 312 ++++++++++++++++++++++++++++++++++++++++++++ docs.go | 20 ++- manager.go | 166 ++++++++++++++++------- safe_manager.go | 103 +++++++++++++-- 4 files changed, 542 insertions(+), 59 deletions(-) create mode 100644 correctness_test.go diff --git a/correctness_test.go b/correctness_test.go new file mode 100644 index 0000000..359397c --- /dev/null +++ b/correctness_test.go @@ -0,0 +1,312 @@ +package c3e + +import ( + "context" + "encoding/json" + "fmt" + "testing" + "time" + + "github.com/valkey-io/valkey-go" +) + +// flushKeys removes the keys a test used, before and after it runs. +func flushKeys(t *testing.T, client valkey.Client, keys ...string) { + t.Helper() + + del := func() { + client.Do(context.Background(), client.B().Del().Key(keys...).Build()) + } + + del() + t.Cleanup(del) +} + +// TestSet_dependencyTTLNeverShrinks is the regression test for the defect that +// let a revoked role keep working for hours. +// +// A reverse-dependency set is shared by every entry that depends on the same +// thing. An unconditional EXPIRE let the last writer shorten it below the TTL of +// an entry already in the set, and jittered TTLs made that ordinary: the set +// could expire well before its dependents, after which Invalidate found nothing +// to cascade to and reported success. +func TestSet_dependencyTTLNeverShrinks(t *testing.T) { + client := realClient(t) + defer client.Close() + + ctx := context.Background() + + cm, err := NewCacheManager(client, false) + if err != nil { + t.Fatalf("NewCacheManager: %v", err) + } + + parent := CacheIdentifier{Type: "c3e_ttl_parent", ID: "1"} + long := CacheIdentifier{Type: "c3e_ttl_long", ID: "1"} + short := CacheIdentifier{Type: "c3e_ttl_short", ID: "1"} + revKey := depKey(parent) + + flushKeys(t, client, revKey, + cacheKey(long), cacheKey(short), + depsForKey(cacheKey(long)), depsForKey(cacheKey(short)), + ) + + // First dependent: a long-lived entry. NX has to set the TTL here, because + // GT alone treats a key with no expiry as infinite and would leave the set + // persistent forever. + if err := cm.Set(ctx, long, []byte(`{"x":1}`), []CacheIdentifier{parent}, time.Hour); err != nil { + t.Fatalf("Set long: %v", err) + } + + ttlAfterLong, err := client.Do(ctx, client.B().Ttl().Key(revKey).Build()).AsInt64() + if err != nil { + t.Fatalf("TTL after long: %v", err) + } + + if ttlAfterLong <= 0 { + t.Fatalf("dependency set has no TTL (%d); it would leak", ttlAfterLong) + } + + // Second dependent, deliberately short-lived. Before the fix this rewrote + // the shared set's TTL down to 3s. + if err := cm.Set(ctx, short, []byte(`{"x":2}`), []CacheIdentifier{parent}, 3*time.Second); err != nil { + t.Fatalf("Set short: %v", err) + } + + ttlAfterShort, err := client.Do(ctx, client.B().Ttl().Key(revKey).Build()).AsInt64() + if err != nil { + t.Fatalf("TTL after short: %v", err) + } + + if ttlAfterShort <= 10 { + t.Errorf("dependency set TTL was shortened to %ds by a later, shorter-lived dependent; "+ + "it must outlive every member (was %ds)", ttlAfterShort, ttlAfterLong) + } +} + +// TestSet_dependencyTTLGrows checks the other half: a longer-lived dependent +// must raise the shared set's TTL, otherwise the set still expires first. +func TestSet_dependencyTTLGrows(t *testing.T) { + client := realClient(t) + defer client.Close() + + ctx := context.Background() + + cm, err := NewCacheManager(client, false) + if err != nil { + t.Fatalf("NewCacheManager: %v", err) + } + + parent := CacheIdentifier{Type: "c3e_grow_parent", ID: "1"} + first := CacheIdentifier{Type: "c3e_grow_first", ID: "1"} + second := CacheIdentifier{Type: "c3e_grow_second", ID: "1"} + revKey := depKey(parent) + + flushKeys(t, client, revKey, + cacheKey(first), cacheKey(second), + depsForKey(cacheKey(first)), depsForKey(cacheKey(second)), + ) + + if err := cm.Set(ctx, first, []byte(`{"x":1}`), []CacheIdentifier{parent}, 30*time.Second); err != nil { + t.Fatalf("Set first: %v", err) + } + + if err := cm.Set(ctx, second, []byte(`{"x":2}`), []CacheIdentifier{parent}, time.Hour); err != nil { + t.Fatalf("Set second: %v", err) + } + + ttl, err := client.Do(ctx, client.B().Ttl().Key(revKey).Build()).AsInt64() + if err != nil { + t.Fatalf("TTL: %v", err) + } + + if ttl <= 60 { + t.Errorf("dependency set TTL is %ds; a longer-lived dependent must raise it", ttl) + } +} + +// TestGet_corruptPayloadHealsInsteadOfFailing covers a payload that no longer +// decodes inside a wrapper that still parses — what an EncoderType change on a +// warm cache, or a changed struct during a rolling deploy, produces. +// +// It used to be returned to the caller as an error, so every read of that key +// failed for the whole hard TTL. It must be treated as a miss. +func TestGet_corruptPayloadHealsInsteadOfFailing(t *testing.T) { + client := realClient(t) + defer client.Close() + + ctx := context.Background() + + cm, err := NewCacheManager(client, false) + if err != nil { + t.Fatalf("NewCacheManager: %v", err) + } + + scm, err := NewSafeCacheManager(cm, SafeCacheManagerConfig{ + HardTTL: time.Hour, + SoftTTL: 30 * time.Minute, + EncoderType: CacheEncoderTypeJSON, + QueryTimeout: 200 * time.Millisecond, + }) + if err != nil { + t.Fatalf("NewSafeCacheManager: %v", err) + } + + id := CacheIdentifier{Type: "c3e_corrupt", ID: "1"} + flushKeys(t, client, cacheKey(id), depsForKey(cacheKey(id))) + + // A valid wrapper whose payload the configured decoder cannot read. + wrapper, err := json.Marshal(CachedItem{ + Data: []byte("this is not valid json"), + RefreshAt: time.Now().Add(time.Hour).Unix(), + }) + if err != nil { + t.Fatalf("marshal wrapper: %v", err) + } + + if err := client.Do(ctx, + client.B().Set().Key(cacheKey(id)).Value(valkey.BinaryString(wrapper)).ExSeconds(3600).Build(), + ).Error(); err != nil { + t.Fatalf("seed corrupt entry: %v", err) + } + + calls := 0 + + var got map[string]any + + err = scm.Get(ctx, id, &got, func(context.Context) (any, []CacheIdentifier, error) { + calls++ + return map[string]any{"healed": true}, nil, nil + }) + if err != nil { + t.Fatalf("Get returned an error instead of refetching: %v", err) + } + + if calls != 1 { + t.Errorf("fetcher called %d times, want 1 — a corrupt payload must be treated as a miss", calls) + } + + if got["healed"] != true { + t.Errorf("got %v, want the refetched value", got) + } +} + +// TestGet_unencodableValueIsStillReturned covers the inconsistency that made a +// caching problem into a request failure: a failing Set was logged and +// swallowed, but a failing *encode* three lines earlier was returned to the +// caller — even though the source of truth had already answered. +func TestGet_unencodableValueIsStillReturned(t *testing.T) { + client := realClient(t) + defer client.Close() + + ctx := context.Background() + + cm, err := NewCacheManager(client, false) + if err != nil { + t.Fatalf("NewCacheManager: %v", err) + } + + scm, err := NewSafeCacheManager(cm, SafeCacheManagerConfig{ + HardTTL: time.Hour, + SoftTTL: 30 * time.Minute, + EncoderType: CacheEncoderTypeJSON, + QueryTimeout: 200 * time.Millisecond, + }) + if err != nil { + t.Fatalf("NewSafeCacheManager: %v", err) + } + + id := CacheIdentifier{Type: "c3e_unencodable", ID: "1"} + flushKeys(t, client, cacheKey(id), depsForKey(cacheKey(id))) + + // A channel cannot be marshalled by encoding/json or encoded by gob. + want := map[string]any{"ok": "value", "bad": make(chan int)} + + var got map[string]any + + if err := scm.Get(ctx, id, &got, func(context.Context) (any, []CacheIdentifier, error) { + return want, nil, nil + }); err != nil { + t.Fatalf("a value that cannot be cached must still be returned: %v", err) + } + + if got["ok"] != "value" { + t.Errorf("got %v, want the fetched value delivered intact", got) + } + + // And nothing should have been stored for it. + if n, err := client.Do(ctx, client.B().Exists().Key(cacheKey(id)).Build()).AsInt64(); err == nil && n != 0 { + t.Errorf("an unencodable value must not leave an entry behind") + } +} + +// TestInvalidate_cascadesAcrossLevels guards the batched breadth-first walk. The +// graph is deliberately both deep and wide, because the batching groups by level +// and a per-level bug would not show up on a single chain. +func TestInvalidate_cascadesAcrossLevels(t *testing.T) { + client := realClient(t) + defer client.Close() + + ctx := context.Background() + + cm, err := NewCacheManager(client, false) + if err != nil { + t.Fatalf("NewCacheManager: %v", err) + } + + root := CacheIdentifier{Type: "c3e_casc_root", ID: "1"} + + // level1: three entries depending on root. + // level2: two entries depending on each level1 entry. + var all []CacheIdentifier + + level1 := make([]CacheIdentifier, 0, 3) + + for i := range 3 { + level1 = append(level1, CacheIdentifier{Type: "c3e_casc_l1", ID: fmt.Sprintf("%d", i)}) + } + + keys := []string{cacheKey(root), depKey(root), depsForKey(cacheKey(root))} + + for _, l1 := range level1 { + all = append(all, l1) + keys = append(keys, cacheKey(l1), depKey(l1), depsForKey(cacheKey(l1))) + + for j := range 2 { + l2 := CacheIdentifier{Type: "c3e_casc_l2", ID: fmt.Sprintf("%s-%d", l1.ID, j)} + all = append(all, l2) + keys = append(keys, cacheKey(l2), depKey(l2), depsForKey(cacheKey(l2))) + } + } + + flushKeys(t, client, keys...) + + for _, l1 := range level1 { + if err := cm.Set(ctx, l1, []byte(`{"l":1}`), []CacheIdentifier{root}, time.Hour); err != nil { + t.Fatalf("Set l1: %v", err) + } + + for j := range 2 { + l2 := CacheIdentifier{Type: "c3e_casc_l2", ID: fmt.Sprintf("%s-%d", l1.ID, j)} + if err := cm.Set(ctx, l2, []byte(`{"l":2}`), []CacheIdentifier{l1}, time.Hour); err != nil { + t.Fatalf("Set l2: %v", err) + } + } + } + + for _, id := range all { + if n, _ := client.Do(ctx, client.B().Exists().Key(cacheKey(id)).Build()).AsInt64(); n != 1 { + t.Fatalf("precondition: %s should be cached", id) + } + } + + if err := cm.Invalidate(ctx, root); err != nil { + t.Fatalf("Invalidate: %v", err) + } + + for _, id := range all { + if n, _ := client.Do(ctx, client.B().Exists().Key(cacheKey(id)).Build()).AsInt64(); n != 0 { + t.Errorf("%s survived the cascade", id) + } + } +} diff --git a/docs.go b/docs.go index e7c693e..b1e3318 100644 --- a/docs.go +++ b/docs.go @@ -121,8 +121,18 @@ // - Dependency links only cover items already present in a cached value, so // adding a new item to a cached collection is invisible to the graph — // invalidate the collection's own key when its membership changes. -// - The dependency-tracking keys expire with the entry TTL (refreshed on each -// write), so they cannot leak when an entry expires naturally. +// - The dependency-tracking keys carry a TTL so they cannot leak when an entry +// expires naturally instead of being invalidated. A reverse-dependency set +// is shared, so its expiry may only ever be raised, never lowered: it is set +// with EXPIRE NX when the set has none and raised with EXPIRE GT when a +// longer-lived dependent joins. A set that could be shortened by the most +// recent writer would expire before entries still listed in it, and an +// Invalidate landing in that window would cascade to nothing and report +// success. Both commands are needed — GT treats a key with no expiry as +// infinite and would refuse to set one at all. +// - The cascade is breadth-first and batched one level at a time, so its cost +// is proportional to the depth of the dependency graph rather than to the +// number of nodes in it. // // # Stale-while-revalidate // @@ -134,7 +144,11 @@ // goroutine refreshes the entry. The refresh runs on a // [context.WithoutCancel] copy of the caller's context (bounded by its own // timeout) so it survives the caller returning. -// - miss — age > HardTTL or the entry is absent. The caller blocks while +// - miss — age > HardTTL, the entry is absent, or what was stored no longer +// decodes (a changed EncoderType, or a cached type whose shape moved during +// a rolling deploy). A stored value that cannot be read is treated as +// absent and refetched, because returning the decode error instead would +// fail every read of that key until its hard TTL expired. The caller blocks while // the fetcher runs (under singleflight). // - timeout / error — the cache did not answer within QueryTimeout, or // returned an error; the caller falls back to the fetcher immediately. diff --git a/manager.go b/manager.go index 3adab54..b76dc04 100644 --- a/manager.go +++ b/manager.go @@ -118,10 +118,29 @@ func (m *CacheManager) Set(ctx context.Context, identifier CacheIdentifier, data } } - // Add new dependencies (reverse-dep sets), each bounded by the entry TTL + // Add new dependencies (reverse-dep sets), each bounded by the entry TTL. + // + // A reverse-dependency set is SHARED by every entry that depends on the same + // thing, so an unconditional EXPIRE here lets the last writer shorten it — + // including below the TTL of an entry already in the set. Jittered TTLs make + // that routine rather than rare: with 10% jitter on a 12h hard TTL, entries + // land anywhere in 10.8h–13.2h, so dep:role:R can expire up to 2.4h before an + // authorization entry that depends on it. In that window Invalidate finds an + // empty set, cascades to nothing, and reports success — a revoked role keeps + // working until the dependent entry expires on its own. + // + // Two commands give the set the TTL of its longest-lived member and never + // less. NX sets an expiry only when the key has none, which covers the first + // dependent and any set whose TTL has since lapsed. GT then raises it only + // when this entry outlives what is already there. Neither can shorten it. + // + // NX is not redundant: GT treats a key with no TTL as having an infinite one + // and refuses to set it, so GT alone would leave the set persistent forever — + // exactly the leak the TTL exists to prevent. for _, newKey := range newDepKeys { cmds = append(cmds, builder.Sadd().Key(newKey).Member(cKey).Build()) - cmds = append(cmds, builder.Expire().Key(newKey).Seconds(ttlSecs).Build()) + cmds = append(cmds, builder.Expire().Key(newKey).Seconds(ttlSecs).Nx().Build()) + cmds = append(cmds, builder.Expire().Key(newKey).Seconds(ttlSecs).Gt().Build()) } // Set the cached data with TTL @@ -193,7 +212,7 @@ func (m *CacheManager) Get(ctx context.Context, identifier CacheIdentifier, ttl func (m *CacheManager) Invalidate(ctx context.Context, identifier CacheIdentifier) error { // Use a queue for breadth-first invalidation (safer than recursion) itemKey := cacheKey(identifier) - queue := []string{depKey(identifier)} + level := []string{depKey(identifier)} visited := make(map[string]bool) builder := m.client.B() @@ -221,64 +240,123 @@ func (m *CacheManager) Invalidate(ctx context.Context, identifier CacheIdentifie delCommands = append(delCommands, builder.Del().Key(itemKey).Build()) // --- Step 2: Cascade invalidation to all dependents (downstream) --- - for len(queue) > 0 { - // Pop from queue - currentDepKey := queue[0] - queue = queue[1:] + // + // Breadth-first, one level at a time, with a single round trip per lookup + // kind per level. This used to issue one SMEMBERS per node *plus* one per + // dependent, all sequentially, so a wide dependency graph turned into + // hundreds of serial round trips on the write path — against a server the + // read path deliberately fast-fails on. Batching makes the cost O(depth) + // round trips instead of O(nodes). + // + // Deleted dep sets are tracked so a dependent's forward list does not SREM + // against a set already queued for deletion, which is what the per-node + // version used the currentDepKey comparison for. + deletedDepKeys := make(map[string]bool) + + for len(level) > 0 { + // Drop anything already handled, and claim the rest for this level. + pending := level[:0:0] + + for _, depK := range level { + if visited[depK] { + continue + } - if visited[currentDepKey] { - continue + visited[depK] = true + + pending = append(pending, depK) + } + + if len(pending) == 0 { + break + } + + // One round trip: every reverse-dependency set in this level. + memberCmds := make([]valkey.Completed, 0, len(pending)) + for _, depK := range pending { + memberCmds = append(memberCmds, builder.Smembers().Key(depK).Build()) } - visited[currentDepKey] = true + dependentsOf := make(map[string][]string, len(pending)) - // Find all cache keys that depend on this entity (reverse dependencies) - smembersCmd := builder.Smembers().Key(currentDepKey).Build() - result := m.client.Do(ctx, smembersCmd) - dependents, err := result.AsStrSlice() - m.logger.Debug("cache: invalidation step", "current_dep_key", currentDepKey, "dependents_found", len(dependents)) + for i, res := range m.client.DoMulti(ctx, memberCmds...) { + dependents, err := res.AsStrSlice() + if err != nil && !valkey.IsValkeyNil(err) { + return fmt.Errorf("%w for key %s: %w", ErrGetDependents, pending[i], err) + } + + if valkey.IsValkeyNil(err) { + dependents = nil + } - if err != nil && !valkey.IsValkeyNil(err) { - return fmt.Errorf("%w for key %s: %w", ErrGetDependents, currentDepKey, err) + dependentsOf[pending[i]] = dependents } - // If key doesn't exist, treat as empty list - if valkey.IsValkeyNil(err) { - dependents = []string{} + // The dep sets themselves go away regardless of what they contained. + for _, depK := range pending { + delCommands = append(delCommands, builder.Del().Key(depK).Build()) + deletedDepKeys[depK] = true + } + + // Collect this level's dependents, de-duplicated: two dep sets in the + // same level can name the same cache entry. + seen := make(map[string]bool) + cKeys := make([]string, 0) + + for _, depK := range pending { + for _, cKey := range dependentsOf[depK] { + if seen[cKey] { + continue + } + + seen[cKey] = true + + cKeys = append(cKeys, cKey) + } + } + + if len(cKeys) == 0 { + level = nil + continue } - // Delete the reverse-dependency set - delCommands = append(delCommands, builder.Del().Key(currentDepKey).Build()) - - // Process each dependent cache entry - for _, cKey := range dependents { - // --- Clean up this dependent's forward dependencies --- - // Remove this dependent from the reverse-dependency sets of items it depends on - cKeyDepsKey := depsForKey(cKey) - cKeyDepsResult := m.client.Do(ctx, builder.Smembers().Key(cKeyDepsKey).Build()) - if cKeyDeps, err := cKeyDepsResult.AsStrSlice(); err == nil { - for _, dep := range cKeyDeps { - // Remove from the reverse dependency set - // Skip if it's the set we're currently processing (already being deleted) - if dep != currentDepKey { + m.logger.Debug("cache: invalidation level", + "dep_keys", len(pending), "dependents_found", len(cKeys)) + + // Second round trip: every dependent's forward-dependency list. + fwdCmds := make([]valkey.Completed, 0, len(cKeys)) + for _, cKey := range cKeys { + fwdCmds = append(fwdCmds, builder.Smembers().Key(depsForKey(cKey)).Build()) + } + + fwdResults := m.client.DoMulti(ctx, fwdCmds...) + + next := make([]string, 0, len(cKeys)) + + for i, cKey := range cKeys { + // Unlink this dependent from the reverse sets it belongs to, so a + // surviving set does not keep pointing at a deleted entry. Sets + // already queued for deletion need no SREM. + if fwdDeps, err := fwdResults[i].AsStrSlice(); err == nil { + for _, dep := range fwdDeps { + if !deletedDepKeys[dep] { delCommands = append(delCommands, builder.Srem().Key(dep).Member(cKey).Build()) } } } - // Delete the dependent's cache data - delCommands = append(delCommands, builder.Del().Key(cKey).Build()) - - // Delete the dependent's forward-dependency list - delCommands = append(delCommands, builder.Del().Key(cKeyDepsKey).Build()) + delCommands = append(delCommands, + builder.Del().Key(cKey).Build(), + builder.Del().Key(depsForKey(cKey)).Build(), + ) - // Queue the dependent's reverse-dependency key for cascade invalidation - // This allows us to find and invalidate anything that depends on this dependent - nextDepKey := depKeyFromCacheKey(cKey) - if !visited[nextDepKey] { - queue = append(queue, nextDepKey) + // Anything depending on this dependent belongs to the next level. + if nextDepKey := depKeyFromCacheKey(cKey); !visited[nextDepKey] { + next = append(next, nextDepKey) } } + + level = next } // Execute all deletions in a batch diff --git a/safe_manager.go b/safe_manager.go index 7e259c5..59450dc 100644 --- a/safe_manager.go +++ b/safe_manager.go @@ -7,6 +7,7 @@ import ( "encoding/json" "fmt" "log/slog" + "reflect" "time" "golang.org/x/sync/singleflight" @@ -185,7 +186,25 @@ func (m *SafeCacheManager) Get(ctx context.Context, identifier CacheIdentifier, // --- CASE A (Fresh) or B (Stale) --- // In both cases, we serve the data we have *right now*. - return m.unmarshalData(item.Data, dest) + // + // Unless it no longer decodes. A corrupt *wrapper* is already treated as + // a miss above; a payload inside a valid wrapper used to be a hard error + // returned to the caller, which is the worse half of the same problem. + // It happens whenever the stored bytes stop matching the decoder: + // EncoderType changed on a warm cache, or a cached struct changed shape + // during a rolling deploy where two binary versions share one server. + // Returning the error makes every read of that key fail for the whole + // hard TTL; treating it as a miss costs one fetch and heals the entry. + if err := m.unmarshalData(item.Data, dest); err != nil { + m.log().Warn("cache: failed to decode cached payload, refetching", + "key", cKey, "error", err) + + result = ResultMiss + + return m.blockingFetch(ctx, identifier, dest, fetcher) + } + + return nil } // 4. Handle TIMEOUT @@ -237,9 +256,20 @@ func (m *SafeCacheManager) blockingFetch(ctx context.Context, identifier CacheId return err } - // `res` is the `wrapperData` ([]byte) from fetchAndCache + fetched, ok := res.(*fetchResult) + if !ok || fetched == nil { + return fmt.Errorf("cache: unexpected fetch result %T", res) + } + + // The value could not be serialized, so there is nothing to decode. Copy it + // into dest directly rather than failing a request the source of truth + // already answered. + if fetched.wrapper == nil { + return assignFetched(dest, fetched.raw) + } + var item CachedItem - if err := json.Unmarshal(res.([]byte), &item); err != nil { + if err := json.Unmarshal(fetched.wrapper, &item); err != nil { return fmt.Errorf("cache: failed to unmarshal fetched wrapper: %w", err) } @@ -247,6 +277,35 @@ func (m *SafeCacheManager) blockingFetch(ctx context.Context, identifier CacheId return m.unmarshalData(item.Data, dest) } +// assignFetched writes src into the pointer dest, the fallback used when a +// value could not be serialized and so cannot be decoded into place. +// +// It is deliberately strict: a mismatch means the fetcher returned something +// other than what the caller asked for, which is a programming error worth +// surfacing rather than papering over with a zero value. +func assignFetched(dest, src any) error { + rv := reflect.ValueOf(dest) + if rv.Kind() != reflect.Pointer || rv.IsNil() { + return fmt.Errorf("cache: destination must be a non-nil pointer, got %T", dest) + } + + elem := rv.Elem() + + if src == nil { + elem.SetZero() + return nil + } + + sv := reflect.ValueOf(src) + if !sv.Type().AssignableTo(elem.Type()) { + return fmt.Errorf("cache: fetched %T is not assignable to %s", src, elem.Type()) + } + + elem.Set(sv) + + return nil +} + // unmarshalData decodes the data using the configured encoder type. func (m *SafeCacheManager) unmarshalData(data []byte, dest any) error { encoderType := m.cfg.EncoderType @@ -267,7 +326,17 @@ func (m *SafeCacheManager) unmarshalData(data []byte, dest any) error { // fetchAndCache is the single-flight function that does the work. // It returns the serialized wrapper (`[]byte`) to the singleflight group. -func (m *SafeCacheManager) fetchAndCache(ctx context.Context, identifier CacheIdentifier, fetcher FetcherFunc) (any, error) { +// fetchResult is what fetchAndCache hands back to singleflight. +// +// wrapper is the serialized CachedItem, present whenever serialization +// succeeded. raw is always the value the fetcher returned. When wrapper is nil +// the value could not be cached, and raw is what the caller must serve. +type fetchResult struct { + raw any + wrapper []byte +} + +func (m *SafeCacheManager) fetchAndCache(ctx context.Context, identifier CacheIdentifier, fetcher FetcherFunc) (*fetchResult, error) { // 1. Get data and dependencies from the primary source data, deps, err := fetcher(ctx) if err != nil { @@ -284,16 +353,23 @@ func (m *SafeCacheManager) fetchAndCache(ctx context.Context, identifier CacheId switch encoderType { case CacheEncoderTypeGob: serializedData, err = EncodeGob(data) - if err != nil { - return nil, fmt.Errorf("failed to encode data with gob: %w", err) - } case CacheEncoderTypeJSON: fallthrough default: serializedData, err = json.Marshal(data) - if err != nil { - return nil, fmt.Errorf("failed to marshal data: %w", err) - } + } + + // A value that cannot be serialized is a caching problem, not a request + // problem: the source of truth already answered. Failing here made a + // serialization fault user-visible even though the expensive work had + // succeeded — and because Set below already logs and swallows its own + // failure, the two halves of one write behaved differently. Hand the value + // back uncached and let the caller proceed. + if err != nil { + m.log().Warn("cache: failed to encode value, returning it uncached", + "key", cacheKey(identifier), "encoder", encoderType, "error", err) + + return &fetchResult{raw: data}, nil } // 3. Create the cache wrapper @@ -305,7 +381,10 @@ func (m *SafeCacheManager) fetchAndCache(ctx context.Context, identifier CacheId // 4. Serialize the wrapper wrapperData, err := json.Marshal(item) if err != nil { - return nil, fmt.Errorf("failed to marshal wrapper: %w", err) + m.log().Warn("cache: failed to marshal wrapper, returning value uncached", + "key", cacheKey(identifier), "error", err) + + return &fetchResult{raw: data}, nil } // 5. Apply jitter to hard TTL @@ -320,5 +399,5 @@ func (m *SafeCacheManager) fetchAndCache(ctx context.Context, identifier CacheId } // 7. Return the serialized wrapper to singleflight (for blocking waiters) - return wrapperData, nil + return &fetchResult{wrapper: wrapperData, raw: data}, nil }