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 }