From 50af76dffb87880a5cae0722e53f5577e94cae17 Mon Sep 17 00:00:00 2001 From: Miguel Martinez Trivino Date: Thu, 9 Apr 2026 12:00:29 +0200 Subject: [PATCH 1/2] feat(cache): add 100MB MaxBytes limit to NATS KV cache buckets Cap both attestation bundle and policy evaluation bundle NATS KV buckets to 100MB to prevent unbounded memory growth. When the limit is reached, NATS automatically evicts the oldest entries. Also cleans up the unused maxSize config field by moving the defaultMaxSize constant directly into the in-memory backend where it is actually used. Signed-off-by: Miguel Martinez Trivino Signed-off-by: Miguel Martinez Trivino --- pkg/cache/attestationbundle/attestationbundle.go | 2 ++ pkg/cache/cache.go | 16 +++++++--------- pkg/cache/memory.go | 8 ++++++-- pkg/cache/natskv.go | 1 + pkg/cache/policyevalbundle/policyevalbundle.go | 2 ++ 5 files changed, 18 insertions(+), 11 deletions(-) diff --git a/pkg/cache/attestationbundle/attestationbundle.go b/pkg/cache/attestationbundle/attestationbundle.go index fca38eda6..f8217e9f0 100644 --- a/pkg/cache/attestationbundle/attestationbundle.go +++ b/pkg/cache/attestationbundle/attestationbundle.go @@ -27,6 +27,7 @@ import ( const ( ttl = 5 * 24 * time.Hour + maxBytes = 100 * 1024 * 1024 // 100 MB bucket = "chainloop-attestation-bundles" description = "Cache for attestation bundles" ) @@ -40,6 +41,7 @@ type Cache struct { func New(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (*Cache, error) { opts := []cache.Option{ cache.WithTTL(ttl), + cache.WithMaxBytes(maxBytes), cache.WithDescription(description), } diff --git a/pkg/cache/cache.go b/pkg/cache/cache.go index 1d69224e6..5396afb35 100644 --- a/pkg/cache/cache.go +++ b/pkg/cache/cache.go @@ -40,13 +40,9 @@ type Logger interface { Errorw(keyvals ...any) } -// defaultMaxSize is a sensible upper bound on in-memory cache entries -// to prevent unbounded growth. 0 means no LRU eviction (TTL-only). -const defaultMaxSize = 1000 - type config struct { ttl time.Duration - maxSize int + maxBytes int64 logger Logger natsConn *nats.Conn bucketName string @@ -75,6 +71,12 @@ func WithNATS(conn *nats.Conn, bucketName string) Option { } } +// WithMaxBytes sets the maximum total size (in bytes) for the NATS KV bucket. +// When the limit is reached, NATS discards the oldest entries. Ignored for in-memory backend. +func WithMaxBytes(n int64) Option { + return func(c *config) { c.maxBytes = n } +} + // WithDescription sets the NATS KV bucket description. Ignored for in-memory backend. func WithDescription(desc string) Option { return func(c *config) { c.description = desc } @@ -108,10 +110,6 @@ func New[T any](opts ...Option) (Cache[T], error) { return newNATSKV[T](cfg) } - if cfg.maxSize == 0 { - cfg.maxSize = defaultMaxSize - } - return newMemoryCache[T](cfg), nil } diff --git a/pkg/cache/memory.go b/pkg/cache/memory.go index 8e444f6ff..ba907a6cc 100644 --- a/pkg/cache/memory.go +++ b/pkg/cache/memory.go @@ -21,15 +21,19 @@ import ( "github.com/hashicorp/golang-lru/v2/expirable" ) +// defaultMaxSize is a sensible upper bound on in-memory cache entries +// to prevent unbounded growth. 0 means no LRU eviction (TTL-only). +const defaultMaxSize = 1000 + type memoryCache[T any] struct { lru *expirable.LRU[string, T] logger Logger } func newMemoryCache[T any](cfg *config) *memoryCache[T] { - cfg.logger.Infow("msg", "cache: using in-memory LRU backend", "ttl", cfg.ttl, "maxSize", cfg.maxSize) + cfg.logger.Infow("msg", "cache: using in-memory LRU backend", "ttl", cfg.ttl, "maxSize", defaultMaxSize) return &memoryCache[T]{ - lru: expirable.NewLRU[string, T](cfg.maxSize, nil, cfg.ttl), + lru: expirable.NewLRU[string, T](defaultMaxSize, nil, cfg.ttl), logger: cfg.logger, } } diff --git a/pkg/cache/natskv.go b/pkg/cache/natskv.go index 72e53a5dd..bb16da391 100644 --- a/pkg/cache/natskv.go +++ b/pkg/cache/natskv.go @@ -65,6 +65,7 @@ func (c *natsKVCache[T]) initBucket() error { Bucket: c.bucket, Description: c.cfg.description, TTL: c.cfg.ttl, + MaxBytes: c.cfg.maxBytes, Storage: jetstream.MemoryStorage, }) if err != nil { diff --git a/pkg/cache/policyevalbundle/policyevalbundle.go b/pkg/cache/policyevalbundle/policyevalbundle.go index fb7fa5d29..b5e54bdef 100644 --- a/pkg/cache/policyevalbundle/policyevalbundle.go +++ b/pkg/cache/policyevalbundle/policyevalbundle.go @@ -27,6 +27,7 @@ import ( const ( ttl = 24 * time.Hour + maxBytes = 100 * 1024 * 1024 // 100 MB bucket = "chainloop-policy-eval-bundles" description = "Cache for policy evaluation bundles from CAS" ) @@ -40,6 +41,7 @@ type Cache struct { func New(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (*Cache, error) { opts := []cache.Option{ cache.WithTTL(ttl), + cache.WithMaxBytes(maxBytes), cache.WithDescription(description), } From 55d77972fe18e6ae4142d9f0f5fa9ba1ed5507a9 Mon Sep 17 00:00:00 2001 From: Miguel Martinez Trivino Date: Thu, 9 Apr 2026 12:15:34 +0200 Subject: [PATCH 2/2] feat(cache): set DiscardOld policy on NATS KV backing streams NATS KV hardcodes DiscardNew on the backing stream, which rejects writes when MaxBytes is reached. For cache use-cases, update the backing stream to DiscardOld so oldest entries are evicted automatically to make room for new ones. Signed-off-by: Miguel Martinez Trivino Signed-off-by: Miguel Martinez Trivino --- pkg/cache/natskv.go | 17 +++++++++++++++++ pkg/cache/natskv_test.go | 34 ++++++++++++++++++++++++++++++++++ 2 files changed, 51 insertions(+) diff --git a/pkg/cache/natskv.go b/pkg/cache/natskv.go index bb16da391..5d3b3a401 100644 --- a/pkg/cache/natskv.go +++ b/pkg/cache/natskv.go @@ -20,6 +20,7 @@ import ( "encoding/base64" "encoding/json" "errors" + "fmt" "sync" "github.com/nats-io/nats.go" @@ -72,6 +73,22 @@ func (c *natsKVCache[T]) initBucket() error { return err } + // NATS KV hardcodes DiscardNew on the backing stream, which rejects writes + // when MaxBytes is reached. For cache use-cases we want DiscardOld so that + // the oldest entries are evicted automatically to make room for new ones. + if c.cfg.maxBytes > 0 { + streamName := fmt.Sprintf("KV_%s", c.bucket) + stream, err := js.Stream(context.Background(), streamName) + if err != nil { + return fmt.Errorf("cache: failed to get backing stream %s: %w", streamName, err) + } + cfg := stream.CachedInfo().Config + cfg.Discard = jetstream.DiscardOld + if _, err := js.UpdateStream(context.Background(), cfg); err != nil { + return fmt.Errorf("cache: failed to set DiscardOld on stream %s: %w", streamName, err) + } + } + c.mu.Lock() c.kv = kv c.mu.Unlock() diff --git a/pkg/cache/natskv_test.go b/pkg/cache/natskv_test.go index d5fa4fc2a..1249f2b0c 100644 --- a/pkg/cache/natskv_test.go +++ b/pkg/cache/natskv_test.go @@ -216,6 +216,40 @@ func TestNATSKV_NilKVGracefulDegradation(t *testing.T) { require.NoError(t, nkv.Purge(ctx)) } +func TestNATSKV_MaxBytesEvictsOldEntries(t *testing.T) { + nc := startEmbeddedNATS(t) + + // With MaxBytes set, the backing stream is updated to DiscardOld so that + // the oldest entries are evicted when the bucket is full. + const maxBytes int64 = 10 * 1024 + c, err := New[[]byte]( + WithTTL(time.Minute), + WithNATS(nc, "test-maxbytes"), + WithMaxBytes(maxBytes), + ) + require.NoError(t, err) + + ctx := context.Background() + payload := make([]byte, 1024) + + // Write 20 entries, well beyond the 10 KiB limit + for i := range 20 { + key := "key-" + strings.Repeat("x", i) + require.NoError(t, c.Set(ctx, key, payload), "Set should not fail even when bucket is full") + } + + // The latest entry should still be retrievable + lastKey := "key-" + strings.Repeat("x", 19) + _, ok, err := c.Get(ctx, lastKey) + require.NoError(t, err) + assert.True(t, ok, "most recent entry should still be in the cache") + + // The earliest entries should have been evicted + _, ok, err = c.Get(ctx, "key-") + require.NoError(t, err) + assert.False(t, ok, "oldest entry should have been evicted") +} + func TestNew_WithNATSReturnsNATSBackend(t *testing.T) { nc := startEmbeddedNATS(t) c, err := New[string](