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) { diff --git a/go-redis/go-redis.go b/go-redis/go-redis.go index b401727..220e5c4 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 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.") + + 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 !walkFn(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. @@ -201,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() @@ -233,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 c701dfc..b200604 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" ) @@ -14,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) } @@ -167,3 +171,117 @@ 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(".*") +} + +func TestRedis_SetMultiLevel_MappingTTL(t *testing.T) { + client, _ := getRedisInstance() + client.DeleteMany(".*") + + inspector := baseRedis.NewClient(&baseRedis.Options{Addr: redisAddr}) + + defer func() { + _ = 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(".*") +} 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=