From 86dc757e93b6da13919ca9a28c818c7094e3760d Mon Sep 17 00:00:00 2001 From: Mohammed Al Sahaf Date: Wed, 8 Jul 2026 01:50:00 +0300 Subject: [PATCH 1/5] feat(core): add MappingWalker for streaming mapping iteration Storer.MapKeys forces implementations to materialize every mapping key and value in a single map. On large deployments the mapping index can reach hundreds of MB, so any caller doing periodic maintenance pays that allocation on every run. MappingWalker is an optional interface that lets a storer stream mapping entries in bounded batches instead. Callers fall back to MapKeys when the storer doesn't implement it. Also expose Lz4WriterPool so downstream storers can reuse lz4 writers. Writers are safe to pool once Close has flushed the frame; readers must never be pooled this way because they escape through http.Response.Body. Signed-off-by: Mohammed Al Sahaf --- core/registered.go | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/core/registered.go b/core/registered.go index 6b6ccad..e9a8ff1 100644 --- a/core/registered.go +++ b/core/registered.go @@ -3,8 +3,25 @@ package core import ( "fmt" "sync" + + "github.com/pierrec/lz4/v4" ) +// Lz4WriterPool pools lz4 writers, which are safe to reuse once Close has +// flushed the frame. Callers must Reset the writer before use and only +// return it to the pool after Close. Readers must never be pooled this way: +// a pooled reader escapes through http.Response.Body and would be recycled +// while another goroutine still reads from it. +var Lz4WriterPool = sync.Pool{New: func() any { return lz4.NewWriter(nil) }} + +// MappingWalker is an optional interface a Storer can implement to stream +// mapping entries in bounded batches instead of materializing the whole +// mapping index in memory like MapKeys does. The walk stops early when fn +// returns false. The key passed to fn is stripped of the given prefix. +type MappingWalker interface { + WalkMappings(prefix string, fn func(key string, value []byte) bool) error +} + var registered = sync.Map{} func RegisterStorage(s Storer) { From 38c8bd3bff0accc46c403f14ebe8119651d18c2a Mon Sep 17 00:00:00 2001 From: Mohammed Al Sahaf Date: Wed, 8 Jul 2026 01:53:50 +0300 Subject: [PATCH 2/5] perf(go-redis): stream mapping keys in bounded MGET batches MapKeys collected every key from SCAN into one slice, then issued a single MGET for all of them and copied every value into a map. With a large mapping index this materializes the entire index in memory at once (observed: 325 MB live from a single caller in a production heap profile). Implement MappingWalker with SCAN + MGET in batches of 100 keys and rewire MapKeys through it, so even the compatibility path no longer issues one unbounded MGET. --- go-redis/go-redis.go | 78 +++++++++++++++++++++++++++++++-------- go-redis/go-redis_test.go | 52 ++++++++++++++++++++++++++ go.work.sum | 10 ++--- 3 files changed, 120 insertions(+), 20 deletions(-) diff --git a/go-redis/go-redis.go b/go-redis/go-redis.go index b401727..14d6efe 100644 --- a/go-redis/go-redis.go +++ b/go-redis/go-redis.go @@ -155,31 +155,79 @@ func (provider *Redis) ListKeys() []string { // MapKeys method returns the list of existing keys. func (provider *Redis) MapKeys(prefix string) map[string]string { mapKeys := map[string]string{} - keys := []string{} - iter := provider.inClient.Scan(provider.ctx, 0, prefix+"*", 100).Iterator() - for iter.Next(provider.ctx) { - keys = append(keys, iter.Val()) - } + _ = provider.WalkMappings(prefix, func(key string, value []byte) bool { + mapKeys[key] = string(value) - if err := iter.Err(); err != nil { - return mapKeys + return true + }) + + return mapKeys +} + +const mappingBatchSize = 100 + +// WalkMappings streams the keys matching the prefix and their values in +// bounded batches so the whole mapping index is never loaded in memory at +// once. The walk stops early when fn returns false. +func (provider *Redis) WalkMappings(prefix string, fn func(key string, value []byte) bool) error { + if provider.reconnecting { + provider.logger.Error("Impossible to walk the redis mappings while reconnecting.") + + return errors.New("reconnecting error") } - vals, err := provider.inClient.MGet(provider.ctx, keys...).Result() - if err != nil { - return mapKeys + batch := make([]string, 0, mappingBatchSize) + + flush := func() (bool, error) { + if len(batch) == 0 { + return true, nil + } + + vals, err := provider.inClient.MGet(provider.ctx, batch...).Result() + if err != nil { + return false, err + } + + for idx, item := range batch { + if idx >= len(vals) || vals[idx] == nil { + continue + } + + value, ok := vals[idx].(string) + if !ok { + continue + } + + k, _ := strings.CutPrefix(item, prefix) + if !fn(k, []byte(value)) { + return false, nil + } + } + + batch = batch[:0] + + return true, nil } - for idx, item := range keys { - k, _ := strings.CutPrefix(item, prefix) + iter := provider.inClient.Scan(provider.ctx, 0, prefix+"*", mappingBatchSize).Iterator() + for iter.Next(provider.ctx) { + batch = append(batch, iter.Val()) - if vals[idx] != nil { - mapKeys[k] = vals[idx].(string) + if len(batch) >= mappingBatchSize { + if cont, err := flush(); err != nil || !cont { + return err + } } } - return mapKeys + if err := iter.Err(); err != nil { + return err + } + + _, err := flush() + + return err } // GetMultiLevel tries to load the key and check if one of linked keys is a fresh/stale candidate. diff --git a/go-redis/go-redis_test.go b/go-redis/go-redis_test.go index c701dfc..dc84fa5 100644 --- a/go-redis/go-redis_test.go +++ b/go-redis/go-redis_test.go @@ -167,3 +167,55 @@ func TestRedis_DeleteMany(t *testing.T) { t.Errorf("The map should be empty, %d given", len(client.MapKeys(""))) } } + +func TestRedis_WalkMappings(t *testing.T) { + client, _ := getRedisInstance() + client.DeleteMany(".*") + + walker, ok := client.(core.MappingWalker) + if !ok { + t.Fatal("The go-redis storer should implement core.MappingWalker") + } + + prefix := "WALK_MAPPINGS_PREFIX_" + // Use more keys than one batch to cover the batch boundary. + count := 250 + + for i := range count { + _ = client.Set(fmt.Sprintf("%s%d", prefix, i), fmt.Appendf(nil, "Hello from %d", i), time.Minute) + } + + values := map[string]string{} + if err := walker.WalkMappings(prefix, func(key string, value []byte) bool { + values[key] = string(value) + + return true + }); err != nil { + t.Errorf("The walk shouldn't error, %v given", err) + } + + if len(values) != count { + t.Errorf("The walk should visit %d entries, %d given", count, len(values)) + } + + for k, v := range values { + if v != "Hello from "+k { + t.Errorf("Expected Hello from %s, %s given", k, v) + } + } + + visited := 0 + if err := walker.WalkMappings(prefix, func(key string, value []byte) bool { + visited++ + + return false + }); err != nil { + t.Errorf("The walk shouldn't error, %v given", err) + } + + if visited != 1 { + t.Errorf("The walk should stop after the first entry, %d visited", visited) + } + + client.DeleteMany(".*") +} diff --git a/go.work.sum b/go.work.sum index db20f50..6b6fe91 100644 --- a/go.work.sum +++ b/go.work.sum @@ -652,6 +652,7 @@ github.com/google/pprof v0.0.0-20230207041349-798e818bf904 h1:4/hN5RUoecvl+RmJRE github.com/google/pprof v0.0.0-20230207041349-798e818bf904/go.mod h1:uglQLonpP8qtYCYyzA+8c/9qtqgA3qsXGYqCPKARAFg= github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad h1:a6HEuzUHeKH6hwfN/ZoQgRgVIWFJljSWa/zetS2WTvg= github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad/go.mod h1:vavhavw2zAxS5dIdcRluK6cSGGPlZynqzFM8NdvU144= +github.com/google/pprof v0.0.0-20250403155104-27863c87afa6/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA= github.com/google/renameio v0.1.0 h1:GOZbcHa3HfsPKPlmyPyN2KEohoMXOhdMbHrvbpl2QaA= github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 h1:El6M4kTTCOh6aBiKaUGG7oYTSPP8MxqL4YI3kZKwcP4= github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510/go.mod h1:pupxD2MaaD3pAXIBCelhxNneeOaAeabZDe5s4K6zSpQ= @@ -905,6 +906,7 @@ github.com/nunnatsa/ginkgolinter v0.21.0/go.mod h1:QlzY9UP9zaqu58FjYxhp9bnjuwXwG github.com/onsi/ginkgo/v2 v2.0.0 h1:CcuG/HvWNkkaqCUpJifQY8z7qEMBJya6aLPx6ftGyjQ= github.com/onsi/ginkgo/v2 v2.22.1 h1:QW7tbJAUDyVDVOM5dFa7qaybo+CRfR7bemlQUN6Z8aM= github.com/onsi/ginkgo/v2 v2.22.1/go.mod h1:S6aTpoRsSq2cZOd+pssHAlKW/Q/jZt6cPrPlnj4a1xM= +github.com/onsi/ginkgo/v2 v2.25.3/go.mod h1:43uiyQC4Ed2tkOzLsEYm7hnrb7UJTWHYNsuy3bG/snE= github.com/onsi/gomega v1.30.0/go.mod h1:9sxs+SwGrKI0+PWe4Fxa9tFQQBG5xSsSbMXOI8PPpoQ= github.com/onsi/gomega v1.36.1/go.mod h1:PvZbdDc8J6XJEpDK4HCuRBm8a6Fzp9/DmhC9C7yFlog= github.com/op/go-logging v0.0.0-20160315200505-970db520ece7 h1:lDH9UUVJtmYCjyT0CI4q8xvlXPxeZ0gYCVvWbmPlp88= @@ -1103,8 +1105,7 @@ github.com/yuin/goldmark v1.7.16 h1:n+CJdUxaFMiDUNnWC3dMWCIQJSkxH4uz3ZwQBkAlVNE= github.com/yuin/goldmark v1.7.16/go.mod h1:ip/1k0VRfGynBgxOz0yCqHrbZXhcjxyuS66Brc7iBKg= github.com/yuin/goldmark-highlighting/v2 v2.0.0-20230729083705-37449abec8cc h1:+IAOyRda+RLrxa1WC7umKOZRsGq4QrFFMYApOeHzQwQ= github.com/yuin/goldmark-highlighting/v2 v2.0.0-20230729083705-37449abec8cc/go.mod h1:ovIvrum6DQJA4QsJSovrkC4saKHQVs7TvcaeO8AIl5I= -github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0= -github.com/zeebo/xxh3 v1.0.2/go.mod h1:5NWz9Sef7zIDm2JHfFlcQvNekmcEl9ekUZQQKCYaDcA= +github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0= github.com/zenazn/goji v0.9.0 h1:RSQQAbXGArQ0dIDEq+PI6WqN6if+5KHu6x2Cx/GXLTQ= gitlab.com/bosi/decorder v0.4.2 h1:qbQaV3zgwnBZ4zPMhGLW4KZe7A7NwxEhJx39R3shffo= gitlab.com/bosi/decorder v0.4.2/go.mod h1:muuhHoaJkA9QLcYHq4Mj8FJUwDZ+EirSHRiaTcTf6T8= @@ -1246,7 +1247,6 @@ golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4= golang.org/x/net v0.38.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8= golang.org/x/net v0.40.0/go.mod h1:y0hY0exeL2Pku80/zKK7tpntoX23cqL3Oa6njdgRtds= golang.org/x/net v0.42.0/go.mod h1:FF1RA5d3u7nAYA4z2TkclSCKh68eSXtiFwcWQpPXdt8= -golang.org/x/net v0.48.0/go.mod h1:+ndRgGjkh8FGtu1w1FGbEC31if4VrNVMuKTgcAAnQRY= golang.org/x/net v0.49.0/go.mod h1:/ysNB2EvaqvesRkuLAyjI1ycPZlQHM3q01F02UY/MV8= golang.org/x/net v0.50.0/go.mod h1:UgoSli3F/pBgdJBHCTc+tp3gmrU4XswgGRgtnwWTfyM= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= @@ -1286,7 +1286,6 @@ golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= golang.org/x/sys v0.34.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= -golang.org/x/sys v0.39.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/sys v0.40.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2 h1:IRJeR9r1pYWsHKTRe/IInb7lYvbBVIqOgsX/u0mbOWY= golang.org/x/telemetry v0.0.0-20240521205824-bda55230c457 h1:zf5N6UOrA487eEFacMePxjXAJctxKmyjKUsjA11Uzuk= @@ -1302,7 +1301,6 @@ golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ= golang.org/x/text v0.17.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY= golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4= golang.org/x/text v0.28.0/go.mod h1:U8nCwOR8jO/marOQ0QbDiOngZVEBB7MAiitBuMjXiNU= -golang.org/x/text v0.32.0/go.mod h1:o/rUWzghvpD5TXrTIBuJU77MTaN0ljMWE47kxGJQ7jY= golang.org/x/tools v0.6.0 h1:BOw41kyTf3PuCW1pVQf8+Cyg8pMlkYB1oo9iJ6D/lKM= golang.org/x/tools v0.16.1/go.mod h1:kYVVN6I1mBNoB1OX+noeBjbRk4IUEPa7JJ+TJMEooJ0= golang.org/x/tools v0.19.0/go.mod h1:qoJWxmGSIBmAeriMx19ogtrEPrGtDbPK634QFIcLAhc= @@ -1310,6 +1308,7 @@ golang.org/x/tools v0.22.0/go.mod h1:aCwcsjqvq7Yqt6TNyX7QMU2enbQ/Gt0bo6krSeEri+c golang.org/x/tools v0.26.0/go.mod h1:TPVVj70c7JJ3WCazhD8OdXcZg/og+b9+tH/KxylGwH0= golang.org/x/tools v0.28.0/go.mod h1:dcIOrVd3mfQKTgrDVQHqCPMWy6lnhfhtX3hLXYVLfRw= golang.org/x/tools v0.34.0/go.mod h1:pAP9OwEaY1CAW3HOmg3hLZC5Z0CCmzjAF2UQMSqNARg= +golang.org/x/tools v0.36.0/go.mod h1:WBDiHKJK8YgLHlcQPYQzNCkUxUypCaa5ZegCVutKm+s= golang.org/x/tools v0.40.0/go.mod h1:Ik/tzLRlbscWpqqMRjyWYDisX8bG13FrdXp3o4Sr9lc= golang.org/x/tools v0.41.0/go.mod h1:XSY6eDqxVNiYgezAVqqCeihT4j1U2CCsqvH3WhQpnlg= golang.org/x/tools/go/expect v0.1.1-deprecated h1:jpBZDwmgPhXsKZC6WhL20P4b/wmnpsEAGHaNy0n/rJM= @@ -1391,6 +1390,7 @@ google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/ google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU= google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc= +gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo= gopkg.in/errgo.v2 v2.1.0 h1:0vLT13EuvQ0hNvakwLuFZ/jYrLp5F3kcWHXdRggjCE8= gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4= From bcbbc98ee13fe7a34bd195a81c7956e0fe0e7b90 Mon Sep 17 00:00:00 2001 From: Mohammed Al Sahaf Date: Wed, 8 Jul 2026 01:55:15 +0300 Subject: [PATCH 3/5] fix(go-redis): bound mapping key TTL and shrink lz4 block size SetMultiLevel stored the mapping key with duration -1, which maps to KeepTTL: mapping keys never expired and grew by one entry per varied key forever. The index could only shrink via the eviction job, and its size was unbounded between runs. Give the mapping key a TTL of max(existing TTL, duration + stale) so it always outlives the longest-lived entry it references, never shortens an expiration owned by a longer-lived entry, and converts legacy unbounded keys to bounded ones on their next update. Also configure the lz4 writer with 64 KB blocks instead of the 4 MB default. Every compression and decompression churned 4 MB pooled blocks even for tiny payloads. Readers pick the block size up from the frame header, so old entries remain readable and new entries are cheap on both paths. --- go-redis/go-redis.go | 22 +++++++++++++- go-redis/go-redis_test.go | 60 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 81 insertions(+), 1 deletion(-) diff --git a/go-redis/go-redis.go b/go-redis/go-redis.go index 14d6efe..ee67691 100644 --- a/go-redis/go-redis.go +++ b/go-redis/go-redis.go @@ -249,6 +249,16 @@ func (provider *Redis) SetMultiLevel(baseKey, variedKey string, value []byte, va compressed := new(bytes.Buffer) writer := lz4.NewWriter(compressed) + // The lz4 default block size is 4 MB, which makes every compression and + // later decompression of the value churn 4 MB pooled blocks even for tiny + // payloads. Cached bodies are usually far smaller, so use the smallest + // block size. Readers pick the block size up from the frame header. + if err := writer.Apply(lz4.BlockSizeOption(lz4.Block64Kb)); err != nil { + provider.logger.Errorf("Impossible to configure the compressor for key %s into Redis, %v", variedKey, err) + + return err + } + if _, err := writer.Write(value); err != nil { _ = writer.Close() @@ -281,7 +291,17 @@ func (provider *Redis) SetMultiLevel(baseKey, variedKey string, value []byte, va return err } - if err = provider.Set(mappingKey, val, -1); err != nil { + // Bound the mapping key lifetime instead of storing it forever: it only + // needs to outlive the longest-lived entry it references. Never shorten + // an expiration owned by a longer-lived entry; TTL returns a negative + // value for missing keys or keys without expiration, so legacy unbounded + // mapping keys become bounded on their next update. + mappingTTL := duration + provider.stale + if remaining := provider.inClient.TTL(provider.ctx, mappingKey).Val(); remaining > mappingTTL { + mappingTTL = remaining + } + + if err = provider.inClient.Set(provider.ctx, mappingKey, val, mappingTTL).Err(); err != nil { provider.logger.Errorf("Impossible to set value into Redis, %v", err) } diff --git a/go-redis/go-redis_test.go b/go-redis/go-redis_test.go index dc84fa5..f4f3ccc 100644 --- a/go-redis/go-redis_test.go +++ b/go-redis/go-redis_test.go @@ -1,12 +1,15 @@ package redis_test import ( + "context" "fmt" + "net/http" "testing" "time" "github.com/darkweak/storages/core" redis "github.com/darkweak/storages/go-redis" + baseRedis "github.com/redis/go-redis/v9" "go.uber.org/zap" ) @@ -219,3 +222,60 @@ func TestRedis_WalkMappings(t *testing.T) { client.DeleteMany(".*") } + +func TestRedis_SetMultiLevel_MappingTTL(t *testing.T) { + client, _ := getRedisInstance() + client.DeleteMany(".*") + + inspector := baseRedis.NewClient(&baseRedis.Options{Addr: "localhost:6379"}) + defer inspector.Close() + + ctx := context.Background() + mappingKey := core.MappingKeyPrefix + "base" + + if err := client.SetMultiLevel("base", "varied-short", []byte("value"), http.Header{}, "", 10*time.Second, "varied-short"); err != nil { + t.Errorf("Impossible to store the value, %v given", err) + } + + ttl := inspector.TTL(ctx, mappingKey).Val() + if ttl <= 0 || ttl > 10*time.Second { + t.Errorf("The mapping key should expire within the entry lifetime, %v given", ttl) + } + + if err := client.SetMultiLevel("base", "varied-long", []byte("value"), http.Header{}, "", time.Hour, "varied-long"); err != nil { + t.Errorf("Impossible to store the value, %v given", err) + } + + ttl = inspector.TTL(ctx, mappingKey).Val() + if ttl <= 10*time.Second || ttl > time.Hour { + t.Errorf("The mapping key expiration should be extended by the longer-lived entry, %v given", ttl) + } + + // A shorter-lived entry must not shorten the mapping key lifetime owned + // by the longer-lived one. + if err := client.SetMultiLevel("base", "varied-shorter", []byte("value"), http.Header{}, "", 5*time.Second, "varied-shorter"); err != nil { + t.Errorf("Impossible to store the value, %v given", err) + } + + ttl = inspector.TTL(ctx, mappingKey).Val() + if ttl <= 10*time.Second || ttl > time.Hour { + t.Errorf("The mapping key expiration shouldn't be shortened, %v given", ttl) + } + + // Legacy mapping keys stored without expiration must become bounded on + // their next update. + if err := inspector.Set(ctx, mappingKey, inspector.Get(ctx, mappingKey).Val(), 0).Err(); err != nil { + t.Errorf("Impossible to remove the mapping key expiration, %v given", err) + } + + if err := client.SetMultiLevel("base", "varied-migrated", []byte("value"), http.Header{}, "", 30*time.Second, "varied-migrated"); err != nil { + t.Errorf("Impossible to store the value, %v given", err) + } + + ttl = inspector.TTL(ctx, mappingKey).Val() + if ttl <= 0 || ttl > 30*time.Second { + t.Errorf("The unbounded mapping key should become bounded, %v given", ttl) + } + + client.DeleteMany(".*") +} From 6f41a049943ab583c11fc7194fa6a040343ca994 Mon Sep 17 00:00:00 2001 From: Mohammed Al Sahaf Date: Wed, 8 Jul 2026 02:07:56 +0300 Subject: [PATCH 4/5] style(go-redis): satisfy errcheck, goconst, varnamelen and wsl linters --- go-redis/go-redis.go | 6 +++--- go-redis/go-redis_test.go | 14 ++++++++++---- 2 files changed, 13 insertions(+), 7 deletions(-) diff --git a/go-redis/go-redis.go b/go-redis/go-redis.go index ee67691..220e5c4 100644 --- a/go-redis/go-redis.go +++ b/go-redis/go-redis.go @@ -169,8 +169,8 @@ const mappingBatchSize = 100 // WalkMappings streams the keys matching the prefix and their values in // bounded batches so the whole mapping index is never loaded in memory at -// once. The walk stops early when fn returns false. -func (provider *Redis) WalkMappings(prefix string, fn func(key string, value []byte) bool) error { +// once. The walk stops early when walkFn returns false. +func (provider *Redis) WalkMappings(prefix string, walkFn func(key string, value []byte) bool) error { if provider.reconnecting { provider.logger.Error("Impossible to walk the redis mappings while reconnecting.") @@ -200,7 +200,7 @@ func (provider *Redis) WalkMappings(prefix string, fn func(key string, value []b } k, _ := strings.CutPrefix(item, prefix) - if !fn(k, []byte(value)) { + if !walkFn(k, []byte(value)) { return false, nil } } diff --git a/go-redis/go-redis_test.go b/go-redis/go-redis_test.go index f4f3ccc..b200604 100644 --- a/go-redis/go-redis_test.go +++ b/go-redis/go-redis_test.go @@ -17,15 +17,16 @@ const ( byteKey = "MyByteKey" nonExistentKey = "NonExistentKey" baseValue = "My first data" + redisAddr = "localhost:6379" ) func getRedisInstance() (core.Storer, error) { - return redis.Factory(core.CacheProvider{URL: "localhost:6379"}, zap.NewNop().Sugar(), 0) + return redis.Factory(core.CacheProvider{URL: redisAddr}, zap.NewNop().Sugar(), 0) } func getRedisConfigurationInstance() (core.Storer, error) { return redis.Factory(core.CacheProvider{Configuration: map[string]interface{}{ - "Addrs": []string{"localhost:6379"}, + "Addrs": []string{redisAddr}, }}, zap.NewNop().Sugar(), 0) } @@ -189,6 +190,7 @@ func TestRedis_WalkMappings(t *testing.T) { } values := map[string]string{} + if err := walker.WalkMappings(prefix, func(key string, value []byte) bool { values[key] = string(value) @@ -208,6 +210,7 @@ func TestRedis_WalkMappings(t *testing.T) { } visited := 0 + if err := walker.WalkMappings(prefix, func(key string, value []byte) bool { visited++ @@ -227,8 +230,11 @@ func TestRedis_SetMultiLevel_MappingTTL(t *testing.T) { client, _ := getRedisInstance() client.DeleteMany(".*") - inspector := baseRedis.NewClient(&baseRedis.Options{Addr: "localhost:6379"}) - defer inspector.Close() + inspector := baseRedis.NewClient(&baseRedis.Options{Addr: redisAddr}) + + defer func() { + _ = inspector.Close() + }() ctx := context.Background() mappingKey := core.MappingKeyPrefix + "base" From 53a85089a1bb07e35c79ae2146c89d7cb3a2c1fe Mon Sep 17 00:00:00 2001 From: Mohammed Al Sahaf Date: Thu, 20 Aug 2026 12:50:04 +0300 Subject: [PATCH 5/5] feat(go-redis): store surrogate tags as native sets Surrogate tags were stored as one comma-joined string per tag. Every stored response reread the whole value, appended one key and rewrote it, so tag values grew without bound (~720 KB single reads observed in a production allocs profile) and each write cost O(value size). Add a SetStorer optional interface to core and implement it with native Redis sets: SADD deduplicates members without reading the value back, SMEMBERS serves purges, and a SCAN-based WalkSets streams tags for listings. Legacy string values are migrated to sets transparently on first write and remain readable until then. A positive duration bounds the set lifetime without ever shortening a longer remaining one, so legacy unbounded tags become bounded too. --- core/registered.go | 17 +++++ go-redis/go-redis.go | 110 ++++++++++++++++++++++++++++ go-redis/go-redis_test.go | 150 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 277 insertions(+) diff --git a/core/registered.go b/core/registered.go index e9a8ff1..3b3bdaa 100644 --- a/core/registered.go +++ b/core/registered.go @@ -3,6 +3,7 @@ package core import ( "fmt" "sync" + "time" "github.com/pierrec/lz4/v4" ) @@ -22,6 +23,22 @@ type MappingWalker interface { WalkMappings(prefix string, fn func(key string, value []byte) bool) error } +// SetStorer is an optional interface a Storer can implement to store a set +// of members under a key natively, instead of a separator-joined string +// value that must be fully read and rewritten on every addition. +type SetStorer interface { + // AddToSet adds members to the set stored at key, extending the set + // lifetime to at least the given duration when it is positive, without + // shortening a longer remaining lifetime. + AddToSet(key string, members []string, duration time.Duration) error + // GetSet returns all members of the set stored at key. + GetSet(key string) []string + // WalkSets visits every set whose key matches the prefix. The walk stops + // early when fn returns false. The key passed to fn is stripped of the + // given prefix. + WalkSets(prefix string, fn func(key string, members []string) bool) error +} + var registered = sync.Map{} func RegisterStorage(s Storer) { diff --git a/go-redis/go-redis.go b/go-redis/go-redis.go index 220e5c4..11e0bdf 100644 --- a/go-redis/go-redis.go +++ b/go-redis/go-redis.go @@ -230,6 +230,101 @@ func (provider *Redis) WalkMappings(prefix string, walkFn func(key string, value return err } +// AddToSet stores members in the native set at key, migrating any legacy +// string value first. A positive duration extends the set lifetime without +// ever shortening a longer remaining one, and bounds legacy unbounded keys. +func (provider *Redis) AddToSet(key string, members []string, duration time.Duration) error { + if provider.reconnecting { + provider.logger.Error("Impossible to add to the redis set while reconnecting.") + + return errors.New("reconnecting error") + } + + legacy := provider.legacySetMembers(key) + + values := make([]interface{}, 0, len(members)+len(legacy)) + for _, member := range members { + values = append(values, member) + } + + for _, member := range legacy { + values = append(values, member) + } + + expire := time.Duration(0) + + if duration > 0 { + if remaining := provider.inClient.TTL(provider.ctx, key).Val(); remaining < duration { + expire = duration + } + } + + _, err := provider.inClient.TxPipelined(provider.ctx, func(pipe redis.Pipeliner) error { + if len(legacy) > 0 { + pipe.Del(provider.ctx, key) + } + + pipe.SAdd(provider.ctx, key, values...) + + if expire > 0 { + pipe.Expire(provider.ctx, key, expire) + } + + return nil + }) + if err != nil { + provider.logger.Errorf("Impossible to add members to the set %s into Redis, %v", key, err) + } + + return err +} + +// GetSet returns all members of the set stored at key, supporting sets that +// are still stored in the legacy string format. +func (provider *Redis) GetSet(key string) []string { + if provider.reconnecting { + provider.logger.Error("Impossible to get the redis set while reconnecting.") + + return nil + } + + if legacy := provider.legacySetMembers(key); len(legacy) > 0 { + return legacy + } + + members, err := provider.inClient.SMembers(provider.ctx, key).Result() + if err != nil { + return nil + } + + return members +} + +// WalkSets visits every set whose key matches the prefix. The walk stops +// early when walkFn returns false. +func (provider *Redis) WalkSets(prefix string, walkFn func(key string, members []string) bool) error { + if provider.reconnecting { + provider.logger.Error("Impossible to walk the redis sets while reconnecting.") + + return errors.New("reconnecting error") + } + + iter := provider.inClient.Scan(provider.ctx, 0, prefix+"*", mappingBatchSize).Iterator() + for iter.Next(provider.ctx) { + members := provider.GetSet(iter.Val()) + if len(members) == 0 { + continue + } + + key, _ := strings.CutPrefix(iter.Val(), prefix) + if !walkFn(key, members) { + return nil + } + } + + return iter.Err() +} + // GetMultiLevel tries to load the key and check if one of linked keys is a fresh/stale candidate. func (provider *Redis) GetMultiLevel(key string, req *http.Request, validator *core.Revalidator) (fresh *http.Response, stale *http.Response) { b, e := provider.inClient.Get(provider.ctx, provider.hashtags+core.MappingKeyPrefix+key).Bytes() @@ -438,3 +533,18 @@ func (provider *Redis) Reconnect() { provider.Reconnect() } } + +// legacySetMembers returns the members of a set that is still stored in the +// legacy format: a single comma-joined string value. +func (provider *Redis) legacySetMembers(key string) []string { + if keyType, _ := provider.inClient.Type(provider.ctx, key).Result(); keyType != "string" { + return nil + } + + value, err := provider.inClient.Get(provider.ctx, key).Result() + if err != nil || value == "" { + return nil + } + + return strings.Split(value, ",") +} diff --git a/go-redis/go-redis_test.go b/go-redis/go-redis_test.go index b200604..f59b059 100644 --- a/go-redis/go-redis_test.go +++ b/go-redis/go-redis_test.go @@ -285,3 +285,153 @@ func TestRedis_SetMultiLevel_MappingTTL(t *testing.T) { client.DeleteMany(".*") } + +func TestRedis_Sets(t *testing.T) { + client, _ := getRedisInstance() + client.DeleteMany(".*") + + setStorer, ok := client.(core.SetStorer) + if !ok { + t.Fatal("The go-redis storer should implement core.SetStorer") + } + + key := "SURROGATE_test" + + if err := setStorer.AddToSet(key, []string{"key1", "key2"}, time.Minute); err != nil { + t.Errorf("The set addition shouldn't error, %v given", err) + } + + // Duplicated members must be stored once. + if err := setStorer.AddToSet(key, []string{"key2", "key3"}, time.Minute); err != nil { + t.Errorf("The set addition shouldn't error, %v given", err) + } + + members := setStorer.GetSet(key) + if len(members) != 3 { + t.Errorf("The set should contain 3 members, %d given", len(members)) + } + + inspector := baseRedis.NewClient(&baseRedis.Options{Addr: redisAddr}) + + defer func() { + _ = inspector.Close() + }() + + ctx := context.Background() + + ttl := inspector.TTL(ctx, key).Val() + if ttl <= 0 || ttl > time.Minute { + t.Errorf("The set should expire within the given duration, %v given", ttl) + } + + // A shorter lifetime must not shorten the remaining one. + if err := setStorer.AddToSet(key, []string{"key4"}, time.Second); err != nil { + t.Errorf("The set addition shouldn't error, %v given", err) + } + + ttl = inspector.TTL(ctx, key).Val() + if ttl <= time.Second { + t.Errorf("The set expiration shouldn't be shortened, %v given", ttl) + } + + client.DeleteMany(".*") +} + +func TestRedis_Sets_LegacyStringMigration(t *testing.T) { + client, _ := getRedisInstance() + client.DeleteMany(".*") + + setStorer, _ := client.(core.SetStorer) + key := "SURROGATE_legacy" + + // Legacy format: comma-joined string without expiration. + if err := client.Set(key, []byte("old1,old2"), -1); err != nil { + t.Errorf("Impossible to store the legacy value, %v given", err) + } + + members := setStorer.GetSet(key) + if len(members) != 2 { + t.Errorf("The legacy value should expose 2 members, %d given", len(members)) + } + + if err := setStorer.AddToSet(key, []string{"new1"}, time.Minute); err != nil { + t.Errorf("The set addition shouldn't error, %v given", err) + } + + members = setStorer.GetSet(key) + if len(members) != 3 { + t.Errorf("The migrated set should contain 3 members, %d given", len(members)) + } + + inspector := baseRedis.NewClient(&baseRedis.Options{Addr: redisAddr}) + + defer func() { + _ = inspector.Close() + }() + + ctx := context.Background() + + if keyType := inspector.Type(ctx, key).Val(); keyType != "set" { + t.Errorf("The legacy value should be migrated to a native set, %s given", keyType) + } + + ttl := inspector.TTL(ctx, key).Val() + if ttl <= 0 || ttl > time.Minute { + t.Errorf("The migrated set should become bounded, %v given", ttl) + } + + client.DeleteMany(".*") +} + +func TestRedis_WalkSets(t *testing.T) { + client, _ := getRedisInstance() + client.DeleteMany(".*") + + setStorer, _ := client.(core.SetStorer) + prefix := "SURROGATE_" + + for i := range 5 { + if err := setStorer.AddToSet(fmt.Sprintf("%s%d", prefix, i), []string{fmt.Sprintf("key%d", i)}, time.Minute); err != nil { + t.Errorf("The set addition shouldn't error, %v given", err) + } + } + + // An unrelated string key must not be visited. + _ = client.Set("unrelated", []byte("value"), time.Minute) + + sets := map[string][]string{} + + if err := setStorer.WalkSets(prefix, func(key string, members []string) bool { + sets[key] = members + + return true + }); err != nil { + t.Errorf("The walk shouldn't error, %v given", err) + } + + if len(sets) != 5 { + t.Errorf("The walk should visit 5 sets, %d given", len(sets)) + } + + for k, members := range sets { + if len(members) != 1 || members[0] != "key"+k { + t.Errorf("Expected [key%s], %v given", k, members) + } + } + + visited := 0 + + if err := setStorer.WalkSets(prefix, func(key string, members []string) bool { + visited++ + + return false + }); err != nil { + t.Errorf("The walk shouldn't error, %v given", err) + } + + if visited != 1 { + t.Errorf("The walk should stop after the first set, %d visited", visited) + } + + client.DeleteMany(".*") +}