perf(go-redis): bound memory usage of the mapping index (streaming eviction, mapping TTL, smaller lz4 blocks)

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) {

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)
 	}
 

go-redis/go-redis_test.go

@@ -1,28 +1,32 @@
 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"
 )
 
 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(".*")
+}

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,14 +1301,14 @@ 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=
 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=