From c36f5c755fa3a06d843e2ebc4d6911dcdfdfed44 Mon Sep 17 00:00:00 2001 From: Jakub Novak Date: Sat, 14 Feb 2026 17:34:20 -0800 Subject: [PATCH 1/5] feat(api): refactor template cache to redis --- .../internal/cache/templates/alias_cache.go | 8 ++ .../api/internal/cache/templates/cache.go | 132 +++++++++--------- .../internal/cache/templates/cache_test.go | 49 ++++++- .../cache/templates/template_build.go | 2 +- .../api/internal/handlers/sandbox_kill.go | 2 +- .../handlers/snapshot_template_create.go | 3 +- packages/api/internal/handlers/store.go | 2 +- .../internal/handlers/template_alias_test.go | 10 +- .../api/internal/handlers/template_delete.go | 3 +- .../api/internal/handlers/template_tags.go | 5 +- .../api/internal/handlers/template_update.go | 2 +- .../template-manager/create_template.go | 2 +- packages/shared/pkg/cache/redis.go | 70 +++++++++- packages/shared/pkg/cache/redis_test.go | 56 +++++++- 14 files changed, 254 insertions(+), 92 deletions(-) diff --git a/packages/api/internal/cache/templates/alias_cache.go b/packages/api/internal/cache/templates/alias_cache.go index 65d63e634d..f8de4df40d 100644 --- a/packages/api/internal/cache/templates/alias_cache.go +++ b/packages/api/internal/cache/templates/alias_cache.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "time" "github.com/google/uuid" "go.opentelemetry.io/otel/attribute" @@ -16,6 +17,13 @@ import ( "github.com/e2b-dev/infra/packages/shared/pkg/id" ) +const ( + templateInfoExpiration = 5 * time.Minute + refreshInterval = 1 * time.Minute + refreshTimeout = 30 * time.Second + callbackTimeout = 30 * time.Second +) + // AliasInfo holds resolved alias information type AliasInfo struct { TemplateID string diff --git a/packages/api/internal/cache/templates/cache.go b/packages/api/internal/cache/templates/cache.go index 4b594a5ddb..4baae7b3c6 100644 --- a/packages/api/internal/cache/templates/cache.go +++ b/packages/api/internal/cache/templates/cache.go @@ -4,10 +4,10 @@ import ( "context" "errors" "fmt" - "strings" "time" "github.com/google/uuid" + "github.com/redis/go-redis/v9" "go.opentelemetry.io/otel" "github.com/e2b-dev/infra/packages/api/internal/api" @@ -21,48 +21,54 @@ import ( ) const ( - templateInfoExpiration = 5 * time.Minute - refreshInterval = 1 * time.Minute - refreshTimeout = 30 * time.Second - callbackTimeout = 30 * time.Second + templateCacheTTL = 5 * time.Minute + templateCacheRefreshInterval = 1 * time.Minute + templateCacheTimeout = 2 * time.Second + + templateCacheKeyPrefix = "template:info" ) var tracer = otel.Tracer("github.com/e2b-dev/infra/packages/api/internal/cache/templates") func buildCacheKey(templateID, tag string) string { - return templateID + id.TagSeparator + tag + // Wrap templateID in {} so it becomes a Redis hash tag — all keys for the + // same template land on the same hash slot in Redis Cluster, enabling + // pipelined prefix deletion in InvalidateAllTags. + return fmt.Sprintf("{%s}:%s", templateID, tag) } // TemplateInfo holds cached template with build information type TemplateInfo struct { - Template *api.Template - TeamID uuid.UUID - ClusterID uuid.UUID - Build *queries.EnvBuild - Tag string + Template *api.Template `json:"template"` + TeamID uuid.UUID `json:"team_id"` + ClusterID uuid.UUID `json:"cluster_id"` + Build *queries.EnvBuild `json:"build"` + Tag string `json:"tag"` } // TemplateCache caches template+build by templateID:tag. // This is a simple lookup layer - resolution happens in AliasCache. type TemplateCache struct { - cache *cache.MemoryCache[*TemplateInfo] + cache *cache.RedisCache[*TemplateInfo] db *sqlcdb.Client aliasCache *AliasCache } -func NewTemplateCache(db *sqlcdb.Client) *TemplateCache { - config := cache.Config[*TemplateInfo]{ - TTL: templateInfoExpiration, - RefreshInterval: refreshInterval, - RefreshTimeout: refreshTimeout, - CallbackTimeout: callbackTimeout, +func NewTemplateCache(db *sqlcdb.Client, redisClient redis.UniversalClient) *TemplateCache { + redisCache := cache.NewRedisCache[*TemplateInfo](cache.RedisConfig[*TemplateInfo]{ + TTL: templateCacheTTL, + RefreshInterval: templateCacheRefreshInterval, + RedisTimeout: templateCacheTimeout, + RedisClient: redisClient, + RedisPrefix: templateCacheKeyPrefix, + ExtractKeyFunc: func(value *TemplateInfo) string { return buildCacheKey(value.Template.TemplateID, value.Tag) }, - } + }) return &TemplateCache{ - cache: cache.NewMemoryCache(config), + cache: redisCache, db: db, aliasCache: NewAliasCache(db), } @@ -114,9 +120,7 @@ func (c *TemplateCache) getByID(ctx context.Context, templateID string, tag *str } cacheKey := buildCacheKey(templateID, tagValue) - info, err := c.cache.GetOrSet(ctx, cacheKey, func(ctx context.Context, _ string) (*TemplateInfo, error) { - return c.fetchTemplateWithBuild(ctx, templateID, tag) - }) + info, err := c.cache.GetOrSet(ctx, cacheKey, c.fetchTemplateWithBuild(templateID, tag)) if err != nil { return nil, err } @@ -124,62 +128,58 @@ func (c *TemplateCache) getByID(ctx context.Context, templateID string, tag *str return info, nil } -func (c *TemplateCache) fetchTemplateWithBuild(ctx context.Context, templateID string, tag *string) (*TemplateInfo, error) { - ctx, span := tracer.Start(ctx, "fetch template with build") - defer span.End() - - result, err := c.db.GetTemplateWithBuildByTag(ctx, queries.GetTemplateWithBuildByTagParams{ - TemplateID: templateID, - Tag: tag, - }) - if err != nil { - if dberrors.IsNotFoundError(err) { - return nil, ErrTemplateNotFound +func (c *TemplateCache) fetchTemplateWithBuild(templateID string, tag *string) func(context.Context, string) (*TemplateInfo, error) { + return func(ctx context.Context, _ string) (*TemplateInfo, error) { + ctx, span := tracer.Start(ctx, "fetch template with build") + defer span.End() + + result, err := c.db.GetTemplateWithBuildByTag(ctx, queries.GetTemplateWithBuildByTagParams{ + TemplateID: templateID, + Tag: tag, + }) + if err != nil { + if dberrors.IsNotFoundError(err) { + return nil, ErrTemplateNotFound + } + + return nil, fmt.Errorf("fetching template with build: %w", err) } - return nil, fmt.Errorf("fetching template with build: %w", err) + build := &result.EnvBuild + template := result.Env + clusterID := clusters.WithClusterFallback(template.ClusterID) + + tagValue := sharedUtils.DerefOrDefault(tag, id.DefaultTag) + + return &TemplateInfo{ + Template: &api.Template{ + TemplateID: template.ID, + BuildID: build.ID.String(), + Public: template.Public, + Aliases: result.Aliases, + Names: result.Names, + }, + TeamID: template.TeamID, + ClusterID: clusterID, + Build: build, + Tag: tagValue, + }, nil } - - build := &result.EnvBuild - template := result.Env - clusterID := clusters.WithClusterFallback(template.ClusterID) - - tagValue := sharedUtils.DerefOrDefault(tag, id.DefaultTag) - - return &TemplateInfo{ - Template: &api.Template{ - TemplateID: template.ID, - BuildID: build.ID.String(), - Public: template.Public, - Aliases: result.Aliases, - Names: result.Names, - }, - TeamID: template.TeamID, - ClusterID: clusterID, - Build: build, - Tag: tagValue, - }, nil } -func (c *TemplateCache) Invalidate(templateID string, tag *string) { +func (c *TemplateCache) Invalidate(ctx context.Context, templateID string, tag *string) { tagValue := id.DefaultTag if tag != nil { tagValue = *tag } - c.cache.Delete(buildCacheKey(templateID, tagValue)) + cacheKey := buildCacheKey(templateID, tagValue) + c.cache.Delete(ctx, cacheKey) } // InvalidateAllTags invalidates the cache for the given templateID across all tags -func (c *TemplateCache) InvalidateAllTags(templateID string) []string { - keys := make([]string, 0) +func (c *TemplateCache) InvalidateAllTags(ctx context.Context, templateID string) []string { pattern := buildCacheKey(templateID, "") - - for _, key := range c.cache.Keys() { - if strings.HasPrefix(key, pattern) { - keys = append(keys, key) - c.cache.Delete(key) - } - } + keys := c.cache.DeleteByPrefix(ctx, pattern) c.aliasCache.InvalidateByTemplateID(templateID) diff --git a/packages/api/internal/cache/templates/cache_test.go b/packages/api/internal/cache/templates/cache_test.go index f20be569c4..2ac95e5ee8 100644 --- a/packages/api/internal/cache/templates/cache_test.go +++ b/packages/api/internal/cache/templates/cache_test.go @@ -7,6 +7,8 @@ import ( "github.com/stretchr/testify/require" "github.com/e2b-dev/infra/packages/db/pkg/testutils" + "github.com/e2b-dev/infra/packages/shared/pkg/consts" + redis_utils "github.com/e2b-dev/infra/packages/shared/pkg/redis" ) // TestAliasCacheResolve_BareAliasInTeamNamespace tests that a bare alias @@ -361,6 +363,7 @@ func TestAliasCache_InvalidateByTemplateID(t *testing.T) { func TestTemplateCache_InvalidateDoesNotInvalidateAliases(t *testing.T) { t.Parallel() db := testutils.SetupDatabase(t) + redis := redis_utils.SetupInstance(t) ctx := t.Context() teamID := testutils.CreateTestTeam(t, db) @@ -369,7 +372,7 @@ func TestTemplateCache_InvalidateDoesNotInvalidateAliases(t *testing.T) { testutils.CreateTestTemplateAliasWithName(t, db, templateID, "alias-for-template", &teamSlug) - cache := NewTemplateCache(db.SqlcClient) + cache := NewTemplateCache(db.SqlcClient, redis) defer cache.Close(ctx) // Resolve alias to populate alias cache @@ -378,7 +381,7 @@ func TestTemplateCache_InvalidateDoesNotInvalidateAliases(t *testing.T) { require.NotNil(t, info1) // Invalidate the template (should NOT invalidate alias cache) - cache.Invalidate(templateID, nil) + cache.Invalidate(ctx, templateID, nil) // Next resolve should return the same cached pointer info2, err := cache.ResolveAlias(ctx, "alias-for-template", teamSlug) @@ -387,11 +390,49 @@ func TestTemplateCache_InvalidateDoesNotInvalidateAliases(t *testing.T) { assert.Same(t, info1, info2) } +// TestTemplateCache_InvalidateAllTagsDeletesRedisEntries tests that +// InvalidateAllTags deletes Redis entries. +func TestTemplateCache_InvalidateAllTagsDeletesRedisEntries(t *testing.T) { + t.Parallel() + db := testutils.SetupDatabase(t) + redisClient := redis_utils.SetupInstance(t) + ctx := t.Context() + + teamID := testutils.CreateTestTeam(t, db) + templateID := testutils.CreateTestTemplate(t, db, teamID) + buildID := testutils.CreateTestBuild(t, ctx, db, templateID, "ready") + testutils.CreateTestBuildAssignment(t, ctx, db, templateID, buildID, "default") + + tc := NewTemplateCache(db.SqlcClient, redisClient) + defer tc.Close(ctx) + + // Populate the cache (this backfills into Redis via the callback) + _, _, err := tc.Get(ctx, templateID, nil, teamID, consts.LocalClusterID) + require.NoError(t, err) + + // Verify the entry exists in Redis + cacheKey := buildCacheKey(templateID, "default") + redisKey := tc.cache.RedisKey(cacheKey) + val, err := redisClient.Get(ctx, redisKey).Result() + require.NoError(t, err) + require.NotEmpty(t, val) + + // InvalidateAllTags should delete the entry from Redis + keys := tc.InvalidateAllTags(ctx, templateID) + require.NotEmpty(t, keys, "should have deleted at least one key") + + // Verify Redis entry is gone + exists, err := redisClient.Exists(ctx, redisKey).Result() + require.NoError(t, err) + assert.Equal(t, int64(0), exists, "Redis entry should be deleted after InvalidateAllTags") +} + // TestTemplateCache_InvalidateAllTagsAlsoInvalidatesAliases tests that // TemplateCache.InvalidateAllTags also invalidates the alias cache entries func TestTemplateCache_InvalidateAllTagsAlsoInvalidatesAliases(t *testing.T) { t.Parallel() db := testutils.SetupDatabase(t) + redis := redis_utils.SetupInstance(t) ctx := t.Context() teamID := testutils.CreateTestTeam(t, db) @@ -400,7 +441,7 @@ func TestTemplateCache_InvalidateAllTagsAlsoInvalidatesAliases(t *testing.T) { testutils.CreateTestTemplateAliasWithName(t, db, templateID, "alias-all-tags", &teamSlug) - cache := NewTemplateCache(db.SqlcClient) + cache := NewTemplateCache(db.SqlcClient, redis) defer cache.Close(ctx) // Resolve alias to populate alias cache @@ -409,7 +450,7 @@ func TestTemplateCache_InvalidateAllTagsAlsoInvalidatesAliases(t *testing.T) { require.NotNil(t, info1) // Invalidate all tags (should also invalidate alias cache) - cache.InvalidateAllTags(templateID) + cache.InvalidateAllTags(ctx, templateID) // Next resolve should return a different pointer (fresh fetch) info2, err := cache.ResolveAlias(ctx, "alias-all-tags", teamSlug) diff --git a/packages/api/internal/cache/templates/template_build.go b/packages/api/internal/cache/templates/template_build.go index f339fac273..31dbd51b31 100644 --- a/packages/api/internal/cache/templates/template_build.go +++ b/packages/api/internal/cache/templates/template_build.go @@ -72,7 +72,7 @@ type TemplatesBuildCache struct { } func NewTemplateBuildCache(db *sqlcdb.Client, redisClient redis.UniversalClient) *TemplatesBuildCache { - rc := cache.NewRedisCache[TemplateBuildInfo](cache.RedisConfig{ + rc := cache.NewRedisCache[TemplateBuildInfo](cache.RedisConfig[TemplateBuildInfo]{ TTL: buildCacheTTL, RefreshInterval: buildCacheRefreshInterval, RedisTimeout: buildCacheTimeout, diff --git a/packages/api/internal/handlers/sandbox_kill.go b/packages/api/internal/handlers/sandbox_kill.go index 4d8ff4b8fe..419888f2d9 100644 --- a/packages/api/internal/handlers/sandbox_kill.go +++ b/packages/api/internal/handlers/sandbox_kill.go @@ -67,7 +67,7 @@ func (a *APIStore) deleteSnapshot(ctx context.Context, sandboxID string, teamID } }(context.WithoutCancel(ctx)) - a.templateCache.InvalidateAllTags(snapshot.TemplateID) + a.templateCache.InvalidateAllTags(context.WithoutCancel(ctx), snapshot.TemplateID) return nil } diff --git a/packages/api/internal/handlers/snapshot_template_create.go b/packages/api/internal/handlers/snapshot_template_create.go index 260ba8b0b0..49e937a1ef 100644 --- a/packages/api/internal/handlers/snapshot_template_create.go +++ b/packages/api/internal/handlers/snapshot_template_create.go @@ -1,6 +1,7 @@ package handlers import ( + "context" "errors" "fmt" "net/http" @@ -127,7 +128,7 @@ func (a *APIStore) PostSandboxesSandboxIDSnapshots(c *gin.Context, sandboxID api a.templateCache.InvalidateAlias(opts.Namespace, *opts.Alias) } - a.templateCache.Invalidate(result.TemplateID, &opts.Tag) + a.templateCache.Invalidate(context.WithoutCancel(ctx), result.TemplateID, &opts.Tag) // Use namespace/alias when a name was provided, otherwise fall back to the raw template ID snapshotID := id.WithTag(result.TemplateID, opts.Tag) diff --git a/packages/api/internal/handlers/store.go b/packages/api/internal/handlers/store.go index ee40d0f70d..2ee1dc7092 100644 --- a/packages/api/internal/handlers/store.go +++ b/packages/api/internal/handlers/store.go @@ -143,7 +143,7 @@ func NewAPIStore(ctx context.Context, tel *telemetry.Client, config cfg.Config) } authCache := authcache.NewTeamAuthCache() - templateCache := templatecache.NewTemplateCache(sqlcDB) + templateCache := templatecache.NewTemplateCache(sqlcDB, redisClient) templateSpawnCounter := utils.NewTemplateSpawnCounter(ctx, time.Minute, sqlcDB) templateBuildsCache := templatecache.NewTemplateBuildCache(sqlcDB, redisClient) diff --git a/packages/api/internal/handlers/template_alias_test.go b/packages/api/internal/handlers/template_alias_test.go index e513c76358..492a166888 100644 --- a/packages/api/internal/handlers/template_alias_test.go +++ b/packages/api/internal/handlers/template_alias_test.go @@ -16,18 +16,20 @@ import ( "github.com/e2b-dev/infra/packages/api/internal/db/types" authqueries "github.com/e2b-dev/infra/packages/db/pkg/auth/queries" "github.com/e2b-dev/infra/packages/db/pkg/testutils" + redis_utils "github.com/e2b-dev/infra/packages/shared/pkg/redis" ) func TestQueryNotExistingTemplateAlias(t *testing.T) { t.Parallel() testDB := testutils.SetupDatabase(t) + redis := redis_utils.SetupInstance(t) ctx := t.Context() store := &APIStore{ sqlcDB: testDB.SqlcClient, authDB: testDB.AuthDb, - templateCache: templatecache.NewTemplateCache(testDB.SqlcClient), + templateCache: templatecache.NewTemplateCache(testDB.SqlcClient, redis), } alias := "non-existing-template-alias" @@ -58,6 +60,7 @@ func TestQueryExistingTemplateAlias(t *testing.T) { t.Parallel() testDB := testutils.SetupDatabase(t) + redis := redis_utils.SetupInstance(t) ctx := t.Context() teamID := testutils.CreateTestTeam(t, testDB) @@ -70,7 +73,7 @@ func TestQueryExistingTemplateAlias(t *testing.T) { store := &APIStore{ sqlcDB: testDB.SqlcClient, authDB: testDB.AuthDb, - templateCache: templatecache.NewTemplateCache(testDB.SqlcClient), + templateCache: templatecache.NewTemplateCache(testDB.SqlcClient, redis), } w := httptest.NewRecorder() @@ -103,6 +106,7 @@ func TestQueryExistingTemplateAliasAsNotOwnerTeam(t *testing.T) { t.Parallel() testDB := testutils.SetupDatabase(t) + redis := redis_utils.SetupInstance(t) ctx := t.Context() ownerTeamID := testutils.CreateTestTeam(t, testDB) @@ -117,7 +121,7 @@ func TestQueryExistingTemplateAliasAsNotOwnerTeam(t *testing.T) { store := &APIStore{ sqlcDB: testDB.SqlcClient, authDB: testDB.AuthDb, - templateCache: templatecache.NewTemplateCache(testDB.SqlcClient), + templateCache: templatecache.NewTemplateCache(testDB.SqlcClient, redis), } w := httptest.NewRecorder() diff --git a/packages/api/internal/handlers/template_delete.go b/packages/api/internal/handlers/template_delete.go index ed8fd62707..cf64d969c4 100644 --- a/packages/api/internal/handlers/template_delete.go +++ b/packages/api/internal/handlers/template_delete.go @@ -1,6 +1,7 @@ package handlers import ( + "context" "fmt" "net/http" @@ -76,7 +77,7 @@ func (a *APIStore) DeleteTemplatesTemplateID(c *gin.Context, aliasOrTemplateID a return } - a.templateCache.InvalidateAllTags(templateID) + a.templateCache.InvalidateAllTags(context.WithoutCancel(ctx), templateID) telemetry.ReportEvent(ctx, "deleted template from db") diff --git a/packages/api/internal/handlers/template_tags.go b/packages/api/internal/handlers/template_tags.go index 632c13c86b..c4f3776479 100644 --- a/packages/api/internal/handlers/template_tags.go +++ b/packages/api/internal/handlers/template_tags.go @@ -1,6 +1,7 @@ package handlers import ( + "context" "fmt" "net/http" "slices" @@ -150,7 +151,7 @@ func (a *APIStore) PostTemplatesTags(c *gin.Context) { } for _, tag := range tags { - a.templateCache.Invalidate(template.ID, &tag) + a.templateCache.Invalidate(context.WithoutCancel(ctx), template.ID, &tag) } telemetry.ReportEvent(ctx, "assigned template tag") @@ -273,7 +274,7 @@ func (a *APIStore) DeleteTemplatesTags(c *gin.Context) { } for _, tag := range tags { - a.templateCache.Invalidate(aliasInfo.TemplateID, &tag) + a.templateCache.Invalidate(context.WithoutCancel(ctx), aliasInfo.TemplateID, &tag) } telemetry.ReportEvent(ctx, "deleted template tags") diff --git a/packages/api/internal/handlers/template_update.go b/packages/api/internal/handlers/template_update.go index cb5c5642a8..1640b4247e 100644 --- a/packages/api/internal/handlers/template_update.go +++ b/packages/api/internal/handlers/template_update.go @@ -128,7 +128,7 @@ func (a *APIStore) updateTemplate(ctx context.Context, c *gin.Context, aliasOrTe } // Invalidate cache immediately after successful DB update - a.templateCache.InvalidateAllTags(aliasInfo.TemplateID) + a.templateCache.InvalidateAllTags(context.WithoutCancel(ctx), aliasInfo.TemplateID) // For backward compatibility with older CLIs (v1 endpoint), also create a non-namespaced alias // when publishing a template, so older CLIs can still find it by bare alias name diff --git a/packages/api/internal/template-manager/create_template.go b/packages/api/internal/template-manager/create_template.go index 54d80052fa..30ee52402b 100644 --- a/packages/api/internal/template-manager/create_template.go +++ b/packages/api/internal/template-manager/create_template.go @@ -193,7 +193,7 @@ func (tm *TemplateManager) CreateTemplate( telemetry.ReportEvent(ctx, "build status sync completed") // Invalidate the cache - invalidatedKeys := tm.templateCache.InvalidateAllTags(templateID) + invalidatedKeys := tm.templateCache.InvalidateAllTags(context.WithoutCancel(ctx), templateID) telemetry.ReportEvent(ctx, "invalidated template cache", attribute.StringSlice("invalidated_keys", invalidatedKeys)) }(context.WithoutCancel(ctx)) diff --git a/packages/shared/pkg/cache/redis.go b/packages/shared/pkg/cache/redis.go index 6959f4dcb1..4b2e79b86e 100644 --- a/packages/shared/pkg/cache/redis.go +++ b/packages/shared/pkg/cache/redis.go @@ -19,10 +19,12 @@ const ( // redisNoPTTL is the value returned by go-redis PTTL when a key has no expiration. // Redis returns the integer -1; go-redis converts it to -1 * time.Nanosecond. redisNoPTTL = -1 * time.Nanosecond + + redisScanCount = 100 ) // RedisConfig holds the configuration for a RedisCache. -type RedisConfig struct { +type RedisConfig[V any] struct { RedisClient redis.UniversalClient TTL time.Duration // RefreshInterval triggers a background refresh of a Redis entry @@ -33,17 +35,21 @@ type RedisConfig struct { RefreshTimeout time.Duration RedisTimeout time.Duration // default 2s RedisPrefix string // e.g. "template:build" + // ExtractKeyFunc is an optional function to extract a key from the value. + // When set, the key used for Redis storage is derived from the fetched value + // rather than the original lookup key. + ExtractKeyFunc ExtractKeyFunc[V] } // RedisCache is a generic two-tier cache: Redis + user callback. type RedisCache[V any] struct { - config RedisConfig + config RedisConfig[V] fetchGroup singleflight.Group redisRefresh singleflight.Group } // NewRedisCache creates a new RedisCache with the given configuration. -func NewRedisCache[V any](config RedisConfig) *RedisCache[V] { +func NewRedisCache[V any](config RedisConfig[V]) *RedisCache[V] { if config.RedisTimeout == 0 { config.RedisTimeout = 2 * time.Second } @@ -92,7 +98,11 @@ func (rc *RedisCache[V]) GetOrSet(ctx context.Context, key string, dataCallback } // Backfill into Redis - rc.setInRedis(ctx, key, v) + storeKey := key + if rc.config.ExtractKeyFunc != nil { + storeKey = rc.config.ExtractKeyFunc(v) + } + rc.setInRedis(ctx, storeKey, v) return result{value: v}, nil }) @@ -112,6 +122,52 @@ func (rc *RedisCache[V]) Delete(ctx context.Context, key string) { rc.deleteFromRedis(ctx, key) } +// DeleteByPrefix removes all keys matching the given prefix from Redis. +// Uses SCAN (not KEYS) to avoid blocking Redis on large keyspaces. +// Returns the list of deleted cache keys (without the Redis prefix). +func (rc *RedisCache[V]) DeleteByPrefix(ctx context.Context, prefix string) []string { + redisPattern := fmt.Sprintf("%s:%s*", rc.config.RedisPrefix, prefix) + + var deleted []string + var cursor uint64 + for { + scanCtx, cancel := context.WithTimeout(ctx, rc.config.RedisTimeout) + keys, nextCursor, err := rc.config.RedisClient.Scan(scanCtx, cursor, redisPattern, redisScanCount).Result() + cancel() + if err != nil { + logger.L().Warn(ctx, "RedisCache: SCAN error", + zap.String("pattern", redisPattern), + zap.Error(err)) + + break + } + + if len(keys) > 0 { + pipeCtx, cancel := context.WithTimeout(ctx, rc.config.RedisTimeout) + pipe := rc.config.RedisClient.Pipeline() + for _, redisKey := range keys { + pipe.Del(ctx, redisKey) + cacheKey := redisKey[len(rc.config.RedisPrefix)+1:] + deleted = append(deleted, cacheKey) + } + + if _, err := pipe.Exec(pipeCtx); err != nil { + logger.L().Warn(ctx, "RedisCache: pipeline DEL error", + zap.String("pattern", redisPattern), + zap.Error(err)) + } + cancel() + } + + cursor = nextCursor + if cursor == 0 { + break + } + } + + return deleted +} + // Close is a no-op (no background goroutines to stop). func (rc *RedisCache[V]) Close(_ context.Context) error { return nil @@ -175,7 +231,11 @@ func (rc *RedisCache[V]) refreshRedis(ctx context.Context, key string, dataCallb return nil, nil } - rc.setInRedis(ctx, key, value) + storeKey := key + if rc.config.ExtractKeyFunc != nil { + storeKey = rc.config.ExtractKeyFunc(value) + } + rc.setInRedis(ctx, storeKey, value) return nil, nil }) diff --git a/packages/shared/pkg/cache/redis_test.go b/packages/shared/pkg/cache/redis_test.go index 3055bd3b39..1c3cca774c 100644 --- a/packages/shared/pkg/cache/redis_test.go +++ b/packages/shared/pkg/cache/redis_test.go @@ -24,7 +24,7 @@ type testValue struct { func newTestRedisCache(t *testing.T, redisClient redis.UniversalClient) *RedisCache[testValue] { t.Helper() - return NewRedisCache[testValue](RedisConfig{ + return NewRedisCache[testValue](RedisConfig[testValue]{ TTL: 30 * time.Second, RedisClient: redisClient, RedisPrefix: fmt.Sprintf("test:%s", t.Name()), @@ -105,7 +105,7 @@ func TestRedisCache_RedisErrorFallthrough(t *testing.T) { }) defer badClient.Close() - rc := NewRedisCache[testValue](RedisConfig{ + rc := NewRedisCache[testValue](RedisConfig[testValue]{ TTL: 30 * time.Second, RedisClient: badClient, RedisPrefix: "test:bad", @@ -216,7 +216,7 @@ func TestRedisCache_RedisRefresh_TriggeredWhenStale(t *testing.T) { redisTTL := 10 * time.Second refreshInterval := 100 * time.Millisecond - rc := NewRedisCache[testValue](RedisConfig{ + rc := NewRedisCache[testValue](RedisConfig[testValue]{ TTL: redisTTL, RefreshInterval: refreshInterval, RefreshTimeout: 5 * time.Second, @@ -270,7 +270,7 @@ func TestRedisCache_RedisRefresh_UpdatesRedis(t *testing.T) { redisTTL := 10 * time.Second refreshInterval := 100 * time.Millisecond - rc := NewRedisCache[testValue](RedisConfig{ + rc := NewRedisCache[testValue](RedisConfig[testValue]{ TTL: redisTTL, RefreshInterval: refreshInterval, RefreshTimeout: 5 * time.Second, @@ -315,7 +315,7 @@ func TestRedisCache_RedisRefresh_ErrorKeepsStaleValue(t *testing.T) { redisTTL := 10 * time.Second refreshInterval := 100 * time.Millisecond - rc := NewRedisCache[testValue](RedisConfig{ + rc := NewRedisCache[testValue](RedisConfig[testValue]{ TTL: redisTTL, RefreshInterval: refreshInterval, RefreshTimeout: 5 * time.Second, @@ -391,3 +391,49 @@ func TestRedisCache_RedisRefresh_Disabled(t *testing.T) { // Callback should not have been called (Redis hit, no refresh) assert.Equal(t, int32(0), callCount.Load()) } + +func TestRedisCache_DeleteByPrefix(t *testing.T) { + t.Parallel() + redisClient := redis_utils.SetupInstance(t) + rc := newTestRedisCache(t, redisClient) + defer rc.Close(t.Context()) + + // Populate keys with a shared prefix and one without + rc.Set(t.Context(), "team1:sandbox-a", testValue{ID: "a", Name: "A"}) + rc.Set(t.Context(), "team1:sandbox-b", testValue{ID: "b", Name: "B"}) + rc.Set(t.Context(), "team2:sandbox-c", testValue{ID: "c", Name: "C"}) + + // Delete by prefix + deleted := rc.DeleteByPrefix(t.Context(), "team1:") + + // Should have deleted exactly the two team1 keys + assert.Len(t, deleted, 2) + assert.ElementsMatch(t, []string{"team1:sandbox-a", "team1:sandbox-b"}, deleted) + + // Verify team1 keys are gone + _, err := redisClient.Get(t.Context(), rc.RedisKey("team1:sandbox-a")).Result() + assert.ErrorIs(t, err, redis.Nil) + _, err = redisClient.Get(t.Context(), rc.RedisKey("team1:sandbox-b")).Result() + assert.ErrorIs(t, err, redis.Nil) + + // Verify team2 key still exists + _, err = redisClient.Get(t.Context(), rc.RedisKey("team2:sandbox-c")).Result() + assert.NoError(t, err) +} + +func TestRedisCache_DeleteByPrefix_NoMatches(t *testing.T) { + t.Parallel() + redisClient := redis_utils.SetupInstance(t) + rc := newTestRedisCache(t, redisClient) + defer rc.Close(t.Context()) + + rc.Set(t.Context(), "key1", testValue{ID: "1", Name: "One"}) + + deleted := rc.DeleteByPrefix(t.Context(), "nonexistent:") + + assert.Empty(t, deleted) + + // Original key untouched + _, err := redisClient.Get(t.Context(), rc.RedisKey("key1")).Result() + assert.NoError(t, err) +} From c4df734dba20e19a36f2f2a40074eda691d3d717 Mon Sep 17 00:00:00 2001 From: Jakub Novak Date: Wed, 18 Feb 2026 12:15:37 -0800 Subject: [PATCH 2/5] fix: lint --- packages/shared/pkg/cache/redis_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/shared/pkg/cache/redis_test.go b/packages/shared/pkg/cache/redis_test.go index 1c3cca774c..226a0eda2f 100644 --- a/packages/shared/pkg/cache/redis_test.go +++ b/packages/shared/pkg/cache/redis_test.go @@ -412,9 +412,9 @@ func TestRedisCache_DeleteByPrefix(t *testing.T) { // Verify team1 keys are gone _, err := redisClient.Get(t.Context(), rc.RedisKey("team1:sandbox-a")).Result() - assert.ErrorIs(t, err, redis.Nil) + require.ErrorIs(t, err, redis.Nil) _, err = redisClient.Get(t.Context(), rc.RedisKey("team1:sandbox-b")).Result() - assert.ErrorIs(t, err, redis.Nil) + require.ErrorIs(t, err, redis.Nil) // Verify team2 key still exists _, err = redisClient.Get(t.Context(), rc.RedisKey("team2:sandbox-c")).Result() From 38bd719c98848267540d995e346768f6d906d1d4 Mon Sep 17 00:00:00 2001 From: Jakub Novak Date: Wed, 18 Feb 2026 12:36:28 -0800 Subject: [PATCH 3/5] chore: refactor invalidate all keys --- packages/shared/pkg/cache/redis.go | 55 ++++++++++++++++++------------ 1 file changed, 34 insertions(+), 21 deletions(-) diff --git a/packages/shared/pkg/cache/redis.go b/packages/shared/pkg/cache/redis.go index 4b2e79b86e..6198783fd5 100644 --- a/packages/shared/pkg/cache/redis.go +++ b/packages/shared/pkg/cache/redis.go @@ -124,40 +124,53 @@ func (rc *RedisCache[V]) Delete(ctx context.Context, key string) { // DeleteByPrefix removes all keys matching the given prefix from Redis. // Uses SCAN (not KEYS) to avoid blocking Redis on large keyspaces. +// Keys are collected first, then deleted after SCAN completes so that +// the keyspace is not mutated during cursor iteration. // Returns the list of deleted cache keys (without the Redis prefix). func (rc *RedisCache[V]) DeleteByPrefix(ctx context.Context, prefix string) []string { redisPattern := fmt.Sprintf("%s:%s*", rc.config.RedisPrefix, prefix) - var deleted []string + // Phase 1: collect all matching keys without mutating the keyspace. + allKeys := rc.collectKeys(ctx, prefix) + if len(allKeys) == 0 { + return nil + } + + // Phase 2: delete all collected keys. + deleted := make([]string, 0, len(allKeys)) + pipeCtx, cancel := context.WithTimeout(ctx, rc.config.RedisTimeout) + defer cancel() + pipe := rc.config.RedisClient.Pipeline() + for _, redisKey := range allKeys { + pipe.Del(pipeCtx, redisKey) + cacheKey := redisKey[len(rc.config.RedisPrefix)+1:] + deleted = append(deleted, cacheKey) + } + + if _, err := pipe.Exec(pipeCtx); err != nil { + logger.L().Warn(ctx, "RedisCache: pipeline DEL error", + zap.String("pattern", redisPattern), + zap.Error(err)) + } + + return deleted +} + +func (rc *RedisCache[V]) collectKeys(ctx context.Context, prefix string) []string { + var allKeys []string var cursor uint64 for { scanCtx, cancel := context.WithTimeout(ctx, rc.config.RedisTimeout) - keys, nextCursor, err := rc.config.RedisClient.Scan(scanCtx, cursor, redisPattern, redisScanCount).Result() + keys, nextCursor, err := rc.config.RedisClient.Scan(scanCtx, cursor, fmt.Sprintf("%s:%s*", rc.config.RedisPrefix, prefix), 1000).Result() cancel() if err != nil { logger.L().Warn(ctx, "RedisCache: SCAN error", - zap.String("pattern", redisPattern), + zap.String("pattern", fmt.Sprintf("%s:%s*", rc.config.RedisPrefix, prefix)), zap.Error(err)) - break } - if len(keys) > 0 { - pipeCtx, cancel := context.WithTimeout(ctx, rc.config.RedisTimeout) - pipe := rc.config.RedisClient.Pipeline() - for _, redisKey := range keys { - pipe.Del(ctx, redisKey) - cacheKey := redisKey[len(rc.config.RedisPrefix)+1:] - deleted = append(deleted, cacheKey) - } - - if _, err := pipe.Exec(pipeCtx); err != nil { - logger.L().Warn(ctx, "RedisCache: pipeline DEL error", - zap.String("pattern", redisPattern), - zap.Error(err)) - } - cancel() - } + allKeys = append(allKeys, keys...) cursor = nextCursor if cursor == 0 { @@ -165,7 +178,7 @@ func (rc *RedisCache[V]) DeleteByPrefix(ctx context.Context, prefix string) []st } } - return deleted + return allKeys } // Close is a no-op (no background goroutines to stop). From 95b9ca3db21b8514a138e79ac9819d99fee711f8 Mon Sep 17 00:00:00 2001 From: Jakub Novak Date: Wed, 18 Feb 2026 12:45:30 -0800 Subject: [PATCH 4/5] fix: collected deleted after delation --- packages/shared/pkg/cache/redis.go | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/packages/shared/pkg/cache/redis.go b/packages/shared/pkg/cache/redis.go index 6198783fd5..9bcdc5aaff 100644 --- a/packages/shared/pkg/cache/redis.go +++ b/packages/shared/pkg/cache/redis.go @@ -137,20 +137,24 @@ func (rc *RedisCache[V]) DeleteByPrefix(ctx context.Context, prefix string) []st } // Phase 2: delete all collected keys. - deleted := make([]string, 0, len(allKeys)) pipeCtx, cancel := context.WithTimeout(ctx, rc.config.RedisTimeout) defer cancel() pipe := rc.config.RedisClient.Pipeline() for _, redisKey := range allKeys { pipe.Del(pipeCtx, redisKey) - cacheKey := redisKey[len(rc.config.RedisPrefix)+1:] - deleted = append(deleted, cacheKey) } if _, err := pipe.Exec(pipeCtx); err != nil { logger.L().Warn(ctx, "RedisCache: pipeline DEL error", zap.String("pattern", redisPattern), zap.Error(err)) + + return nil + } + + deleted := make([]string, 0, len(allKeys)) + for _, redisKey := range allKeys { + deleted = append(deleted, redisKey[len(rc.config.RedisPrefix)+1:]) } return deleted @@ -161,7 +165,7 @@ func (rc *RedisCache[V]) collectKeys(ctx context.Context, prefix string) []strin var cursor uint64 for { scanCtx, cancel := context.WithTimeout(ctx, rc.config.RedisTimeout) - keys, nextCursor, err := rc.config.RedisClient.Scan(scanCtx, cursor, fmt.Sprintf("%s:%s*", rc.config.RedisPrefix, prefix), 1000).Result() + keys, nextCursor, err := rc.config.RedisClient.Scan(scanCtx, cursor, fmt.Sprintf("%s:%s*", rc.config.RedisPrefix, prefix), redisScanCount).Result() cancel() if err != nil { logger.L().Warn(ctx, "RedisCache: SCAN error", From 0669d9295a2f75d3094d9dca497623366030a22d Mon Sep 17 00:00:00 2001 From: Jakub Novak Date: Wed, 18 Feb 2026 13:01:29 -0800 Subject: [PATCH 5/5] fix: lint --- packages/shared/pkg/cache/redis.go | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/shared/pkg/cache/redis.go b/packages/shared/pkg/cache/redis.go index 9bcdc5aaff..7449d93e9c 100644 --- a/packages/shared/pkg/cache/redis.go +++ b/packages/shared/pkg/cache/redis.go @@ -171,6 +171,7 @@ func (rc *RedisCache[V]) collectKeys(ctx context.Context, prefix string) []strin logger.L().Warn(ctx, "RedisCache: SCAN error", zap.String("pattern", fmt.Sprintf("%s:%s*", rc.config.RedisPrefix, prefix)), zap.Error(err)) + break }