feat(cache): add configurable JetStream KV replica count

NATS KV buckets in pkg/cache were created with the default replica
count of 1, causing inconsistent reads in clustered NATS deployments.
Add a WithReplicas option to pkg/cache and thread the replica count
from Helm values through proto config, natsconn, and bundle caches.

Signed-off-by: Jose I. Paris <jiparis@chainloop.dev>

Author: jiparis

app/controlplane/cmd/main.go

@@ -221,9 +221,15 @@ func newNatsConfig(c *conf.Bootstrap_NatsServer) *natsconn.Config {
 		return nil
 	}
 
+	replicas := int(c.GetReplicas())
+	if replicas < 1 {
+		replicas = 1
+	}
+
 	cfg := &natsconn.Config{
-		URI:  uri,
-		Name: "chainloop-controlplane",
+		URI:      uri,
+		Name:     "chainloop-controlplane",
+		Replicas: replicas,
 	}
 
 	if c.GetToken() != "" {

app/controlplane/internal/conf/controlplane/config/v1/conf.pb.go

@@ -1,5 +1,5 @@
 //
-// Copyright 2024-2025 The Chainloop Authors.
+// Copyright 2024-2026 The Chainloop Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -101,6 +101,9 @@ message Bootstrap {
       // Token based authentication
       string token = 2 [(buf.validate.field).string.min_len = 1];
     }
+    // Number of replicas for JetStream KV buckets.
+    // Defaults to 1. Set to 3 for production clusters.
+    int32 replicas = 3;
   }
 
   // External URL of the platform UI, if available

app/controlplane/internal/conf/controlplane/config/v1/conf.proto

@@ -7,7 +7,7 @@ description: Chainloop is an open source software supply chain control plane, a
 
 type: application
 # Bump the patch (not minor, not major) version on each change in the Chart Source code
-version: 1.364.0
+version: 1.364.1
 # Do not update appVersion, this is handled automatically by the release process
 appVersion: v1.90.0
 

deployment/chainloop/Chart.yaml

@@ -96,11 +96,14 @@ stringData:
         {{- end }}
     
     {{- if and .Values.controlplane.nats.enabled }}
-    nats_server: 
+    nats_server:
       uri: {{ include "controlplane.nats.connection_string" . | quote }}
       {{- if ne .Values.controlplane.nats.token "" }}
       token: {{ .Values.controlplane.nats.token | quote }}
       {{- end }}
+      {{- if .Values.controlplane.nats.replicas }}
+      replicas: {{ .Values.controlplane.nats.replicas }}
+      {{- end }}
     {{- end }}
 
     credentials_service: {{- include "chainloop.credentials_service_settings" . | indent 6 }}

deployment/chainloop/templates/controlplane/secret-config.yaml

@@ -188,11 +188,13 @@ controlplane:
   ## @param controlplane.nats.host NATS Host
   ## @param controlplane.nats.port NATS Port
   ## @param controlplane.nats.token NATS Client authentication token
+  ## @param controlplane.nats.replicas Number of JetStream KV replicas. Set to 3 for clustered NATS deployments.
   nats:
     enabled: false
     host: ""
     port: 4222
     token: ""
+    replicas: 1
 
   ## @extra controlplane.onboarding.name Name of the organization to onboard
   ## @extra controlplane.onboarding.role Role of the organization to onboard

deployment/chainloop/values.yaml

@@ -52,6 +52,7 @@ func New(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logg
 	if rc != nil {
 		opts = append(opts, cache.WithNATS(rc.Conn, bucket))
 		opts = append(opts, cache.WithReconnect(rc.Subscribe(ctx)))
+		opts = append(opts, cache.WithReplicas(rc.Replicas))
 	}
 
 	c, err := cache.New[[]byte](opts...)

pkg/cache/attestationbundle/attestationbundle.go

@@ -43,6 +43,7 @@ type Logger interface {
 type config struct {
 	ttl         time.Duration
 	maxBytes    int64
+	replicas    int
 	logger      Logger
 	natsConn    *nats.Conn
 	bucketName  string
@@ -82,6 +83,13 @@ func WithDescription(desc string) Option {
 	return func(c *config) { c.description = desc }
 }
 
+// WithReplicas sets the number of JetStream KV replicas for the NATS bucket.
+// Defaults to 1 if not set. Set to match the cluster size (e.g. 3) for
+// production NATS clusters. Ignored for in-memory backend.
+func WithReplicas(n int) Option {
+	return func(c *config) { c.replicas = n }
+}
+
 // WithReconnect provides a channel that signals NATS reconnection events.
 func WithReconnect(ch <-chan struct{}) Option {
 	return func(c *config) { c.reconnCh = ch }

pkg/cache/cache.go

@@ -67,6 +67,7 @@ func (c *natsKVCache[T]) initBucket() error {
 		Description: c.cfg.description,
 		TTL:         c.cfg.ttl,
 		MaxBytes:    c.cfg.maxBytes,
+		Replicas:    c.cfg.replicas,
 		Storage:     jetstream.MemoryStorage,
 	})
 	if err != nil {

pkg/cache/natskv.go

@@ -23,6 +23,7 @@ import (
 
 	natsserver "github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -250,6 +251,53 @@ func TestNATSKV_MaxBytesEvictsOldEntries(t *testing.T) {
 	assert.False(t, ok, "oldest entry should have been evicted")
 }
 
+func TestNATSKV_WithReplicas(t *testing.T) {
+	nc := startEmbeddedNATS(t)
+
+	tests := []struct {
+		name     string
+		replicas int
+		wantRep  int
+	}{
+		{"no WithReplicas defaults to 1", 0, 1},
+		{"explicit 1 replica", 1, 1},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			bucket := sanitizeBucketName("test-replicas-" + tt.name)
+			opts := []Option{
+				WithTTL(5 * time.Second),
+				WithNATS(nc, bucket),
+			}
+			if tt.replicas > 0 {
+				opts = append(opts, WithReplicas(tt.replicas))
+			}
+			c, err := New[string](opts...)
+			require.NoError(t, err)
+
+			// Verify replica count via the backing stream config
+			nkv := c.(*natsKVCache[string])
+			js, err := jetstream.New(nc)
+			require.NoError(t, err)
+			stream, err := js.Stream(context.Background(), "KV_"+nkv.bucket)
+			require.NoError(t, err)
+			assert.Equal(t, tt.wantRep, stream.CachedInfo().Config.Replicas)
+		})
+	}
+
+	// Replicas > 1 requires a multi-node NATS cluster. Verify that the option
+	// is actually passed through by confirming that a single-node server rejects it.
+	t.Run("replicas 3 rejected by single-node server", func(t *testing.T) {
+		_, err := New[string](
+			WithTTL(5*time.Second),
+			WithNATS(nc, "test-replicas-3"),
+			WithReplicas(3),
+		)
+		require.Error(t, err, "single-node NATS should reject replicas > 1")
+	})
+}
+
 func TestNew_WithNATSReturnsNATSBackend(t *testing.T) {
 	nc := startEmbeddedNATS(t)
 	c, err := New[string](