fix(cache): make NATS KV bucket init idempotent and fail-open

The control plane panicked on startup when UpdateStream on the KV
backing stream hit the NATS default request timeout. initBucket now
skips UpdateStream when Discard is already DiscardOld and logs a
warning instead of returning an error when the JetStream metadata
calls fail, matching the fail-open pattern used by the cache's
runtime operations.

Signed-off-by: Miguel Martinez <miguel@chainloop.dev>

Author: migmartri

pkg/cache/natskv.go

@@ -22,11 +22,14 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
+const streamUpdateTimeout = 5 * time.Second
+
 type natsKVCache[T any] struct {
 	mu     sync.RWMutex
 	kv     jetstream.KeyValue
@@ -74,28 +77,42 @@ 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()
+
+	// NATS KV hardcodes DiscardNew; we want DiscardOld so the cache evicts
+	// oldest entries at MaxBytes. Fail-open to avoid crashing on NATS slowness.
+	if c.cfg.maxBytes > 0 {
+		c.ensureDiscardOld(js)
+	}
+
 	return nil
 }
 
+func (c *natsKVCache[T]) ensureDiscardOld(js jetstream.JetStream) {
+	streamName := fmt.Sprintf("KV_%s", c.bucket)
+
+	ctx, cancel := context.WithTimeout(context.Background(), streamUpdateTimeout)
+	defer cancel()
+
+	stream, err := js.Stream(ctx, streamName)
+	if err != nil {
+		c.logger.Warnw("msg", "cache: failed to get backing stream, skipping discard policy update", "stream", streamName, "error", err)
+		return
+	}
+
+	cfg := stream.CachedInfo().Config
+	if cfg.Discard == jetstream.DiscardOld {
+		return
+	}
+	cfg.Discard = jetstream.DiscardOld
+
+	if _, err := js.UpdateStream(ctx, cfg); err != nil {
+		c.logger.Warnw("msg", "cache: failed to set DiscardOld on stream, continuing without it", "stream", streamName, "error", err)
+	}
+}
+
 func (c *natsKVCache[T]) watchReconnect(ch <-chan struct{}) {
 	for range ch {
 		c.logger.Infow("msg", "cache: NATS reconnected, reinitializing bucket", "bucket", c.bucket)

pkg/cache/natskv_test.go

@@ -251,6 +251,73 @@ func TestNATSKV_MaxBytesEvictsOldEntries(t *testing.T) {
 	assert.False(t, ok, "oldest entry should have been evicted")
 }
 
+func TestNATSKV_InitBucketIsIdempotent(t *testing.T) {
+	// Simulates a pod restart: first New() sets DiscardOld on the backing
+	// stream, second New() against the same NATS sees it's already correct
+	// and must not issue another UpdateStream.
+	nc := startEmbeddedNATS(t)
+	bucket := "test-idempotent"
+
+	c1, err := New[[]byte](
+		WithTTL(time.Minute),
+		WithNATS(nc, bucket),
+		WithMaxBytes(10*1024),
+	)
+	require.NoError(t, err)
+	_ = c1
+
+	js, err := jetstream.New(nc)
+	require.NoError(t, err)
+	streamName := "KV_" + bucket
+	stream, err := js.Stream(context.Background(), streamName)
+	require.NoError(t, err)
+	info, err := stream.Info(context.Background())
+	require.NoError(t, err)
+	firstUpdate := info.Created
+	firstDiscard := info.Config.Discard
+	require.Equal(t, jetstream.DiscardOld, firstDiscard, "first init must set DiscardOld")
+
+	// Second init against the same bucket must be a no-op for the update path.
+	c2, err := New[[]byte](
+		WithTTL(time.Minute),
+		WithNATS(nc, bucket),
+		WithMaxBytes(10*1024),
+	)
+	require.NoError(t, err)
+	_ = c2
+
+	info2, err := stream.Info(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, firstUpdate, info2.Created, "stream should not have been recreated")
+	assert.Equal(t, jetstream.DiscardOld, info2.Config.Discard)
+}
+
+func TestNATSKV_InitBucketFailsOpenOnStreamUpdateError(t *testing.T) {
+	// If the JetStream metadata calls time out or fail, init must not crash
+	// the process — the runtime Get/Set already fail-open, so the cache
+	// simply degrades. We exercise this by cancelling the context in the
+	// middle of ensureDiscardOld via a NATS connection that's already closed.
+	nc := startEmbeddedNATS(t)
+	bucket := "test-failopen"
+
+	c, err := New[[]byte](
+		WithTTL(time.Minute),
+		WithNATS(nc, bucket),
+		WithMaxBytes(10*1024),
+	)
+	require.NoError(t, err)
+
+	// Now close the NATS connection and call ensureDiscardOld directly —
+	// it must return without panicking or returning an error.
+	nc.Close()
+	nkv := c.(*natsKVCache[[]byte])
+	js, err := jetstream.New(nkv.conn)
+	require.NoError(t, err)
+	require.NotPanics(t, func() {
+		nkv.ensureDiscardOld(js)
+	})
+}
+
 func TestNATSKV_WithReplicas(t *testing.T) {
 	nc := startEmbeddedNATS(t)