From 6c558bc2f82129ab3b1713018e765e8dde8abd4e Mon Sep 17 00:00:00 2001 From: "Jose I. Paris" Date: Mon, 13 Apr 2026 12:28:27 +0200 Subject: [PATCH] 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 --- app/controlplane/cmd/main.go | 12 ++++- .../conf/controlplane/config/v1/conf.pb.go | 23 ++++++--- .../conf/controlplane/config/v1/conf.proto | 5 +- deployment/chainloop/Chart.yaml | 2 +- .../templates/controlplane/secret-config.yaml | 5 +- deployment/chainloop/values.yaml | 2 + .../attestationbundle/attestationbundle.go | 1 + pkg/cache/cache.go | 8 ++++ pkg/cache/natskv.go | 1 + pkg/cache/natskv_test.go | 48 +++++++++++++++++++ .../policyevalbundle/policyevalbundle.go | 1 + pkg/natsconn/natsconn.go | 10 ++-- 12 files changed, 103 insertions(+), 15 deletions(-) diff --git a/app/controlplane/cmd/main.go b/app/controlplane/cmd/main.go index 3a369b83e..5fb2805ae 100644 --- a/app/controlplane/cmd/main.go +++ b/app/controlplane/cmd/main.go @@ -221,9 +221,17 @@ func newNatsConfig(c *conf.Bootstrap_NatsServer) *natsconn.Config { return nil } + replicas := int(c.GetReplicas()) + if replicas < 1 { + replicas = 1 + } else if replicas > 3 { + replicas = 3 + } + cfg := &natsconn.Config{ - URI: uri, - Name: "chainloop-controlplane", + URI: uri, + Name: "chainloop-controlplane", + Replicas: replicas, } if c.GetToken() != "" { diff --git a/app/controlplane/internal/conf/controlplane/config/v1/conf.pb.go b/app/controlplane/internal/conf/controlplane/config/v1/conf.pb.go index 124e1302f..81d2fd82a 100644 --- a/app/controlplane/internal/conf/controlplane/config/v1/conf.pb.go +++ b/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. @@ -947,8 +947,11 @@ type Bootstrap_NatsServer struct { // // *Bootstrap_NatsServer_Token Authentication isBootstrap_NatsServer_Authentication `protobuf_oneof:"authentication"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Number of replicas for JetStream KV buckets. + // Defaults to 1. Maximum is 3. Set to 3 for production clusters. + Replicas int32 `protobuf:"varint,3,opt,name=replicas,proto3" json:"replicas,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *Bootstrap_NatsServer) Reset() { @@ -1004,6 +1007,13 @@ func (x *Bootstrap_NatsServer) GetToken() string { return "" } +func (x *Bootstrap_NatsServer) GetReplicas() int32 { + if x != nil { + return x.Replicas + } + return 0 +} + type isBootstrap_NatsServer_Authentication interface { isBootstrap_NatsServer_Authentication() } @@ -1565,7 +1575,7 @@ var File_controlplane_config_v1_conf_proto protoreflect.FileDescriptor const file_controlplane_config_v1_conf_proto_rawDesc = "" + "\n" + - "!controlplane/config/v1/conf.proto\x12\x16controlplane.config.v1\x1a\x1bbuf/validate/validate.proto\x1a#controlplane/config/v1/config.proto\x1a\x1bcredentials/v1/config.proto\x1a\x1egoogle/protobuf/duration.proto\"\x9c\x0e\n" + + "!controlplane/config/v1/conf.proto\x12\x16controlplane.config.v1\x1a\x1bbuf/validate/validate.proto\x1a#controlplane/config/v1/config.proto\x1a\x1bcredentials/v1/config.proto\x1a\x1egoogle/protobuf/duration.proto\"\xb8\x0e\n" + "\tBootstrap\x126\n" + "\x06server\x18\x01 \x01(\v2\x1e.controlplane.config.v1.ServerR\x06server\x120\n" + "\x04data\x18\x02 \x01(\v2\x1c.controlplane.config.v1.DataR\x04data\x120\n" + @@ -1601,11 +1611,12 @@ const file_controlplane_config_v1_conf_proto_rawDesc = "" + "\x04grpc\x18\x01 \x01(\v2#.controlplane.config.v1.Server.GRPCB\x06\xbaH\x03\xc8\x01\x01R\x04grpc\x12\x1a\n" + "\binsecure\x18\x02 \x01(\bR\binsecure\x12!\n" + "\fdownload_url\x18\x03 \x01(\tR\vdownloadUrl\x123\n" + - "\x16default_entry_max_size\x18\x04 \x01(\tR\x13defaultEntryMaxSize\x1aZ\n" + + "\x16default_entry_max_size\x18\x04 \x01(\tR\x13defaultEntryMaxSize\x1av\n" + "\n" + "NatsServer\x12\x19\n" + "\x03uri\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\x03uri\x12\x1f\n" + - "\x05token\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01H\x00R\x05tokenB\x10\n" + + "\x05token\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01H\x00R\x05token\x12\x1a\n" + + "\breplicas\x18\x03 \x01(\x05R\breplicasB\x10\n" + "\x0eauthentication\"O\n" + "\x17FederatedAuthentication\x12\x1a\n" + "\x03url\x18\x01 \x01(\tB\b\xbaH\x05r\x03\x88\x01\x01R\x03url\x12\x18\n" + diff --git a/app/controlplane/internal/conf/controlplane/config/v1/conf.proto b/app/controlplane/internal/conf/controlplane/config/v1/conf.proto index 1cf70681d..bb4aa0514 100644 --- a/app/controlplane/internal/conf/controlplane/config/v1/conf.proto +++ b/app/controlplane/internal/conf/controlplane/config/v1/conf.proto @@ -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. Maximum is 3. Set to 3 for production clusters. + int32 replicas = 3; } // External URL of the platform UI, if available diff --git a/deployment/chainloop/Chart.yaml b/deployment/chainloop/Chart.yaml index 296b48cec..6c6e78355 100644 --- a/deployment/chainloop/Chart.yaml +++ b/deployment/chainloop/Chart.yaml @@ -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 diff --git a/deployment/chainloop/templates/controlplane/secret-config.yaml b/deployment/chainloop/templates/controlplane/secret-config.yaml index 432d1392e..a029988bc 100644 --- a/deployment/chainloop/templates/controlplane/secret-config.yaml +++ b/deployment/chainloop/templates/controlplane/secret-config.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 }} diff --git a/deployment/chainloop/values.yaml b/deployment/chainloop/values.yaml index b9ef126ab..f8dad4b0a 100644 --- a/deployment/chainloop/values.yaml +++ b/deployment/chainloop/values.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 (1-3). 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 diff --git a/pkg/cache/attestationbundle/attestationbundle.go b/pkg/cache/attestationbundle/attestationbundle.go index f8217e9f0..b339f67cd 100644 --- a/pkg/cache/attestationbundle/attestationbundle.go +++ b/pkg/cache/attestationbundle/attestationbundle.go @@ -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...) diff --git a/pkg/cache/cache.go b/pkg/cache/cache.go index 5396afb35..c30634c51 100644 --- a/pkg/cache/cache.go +++ b/pkg/cache/cache.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 } diff --git a/pkg/cache/natskv.go b/pkg/cache/natskv.go index 5d3b3a401..a0e8fd163 100644 --- a/pkg/cache/natskv.go +++ b/pkg/cache/natskv.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 { diff --git a/pkg/cache/natskv_test.go b/pkg/cache/natskv_test.go index 1249f2b0c..6c3f4edcb 100644 --- a/pkg/cache/natskv_test.go +++ b/pkg/cache/natskv_test.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]( diff --git a/pkg/cache/policyevalbundle/policyevalbundle.go b/pkg/cache/policyevalbundle/policyevalbundle.go index b5e54bdef..2e66fa7ad 100644 --- a/pkg/cache/policyevalbundle/policyevalbundle.go +++ b/pkg/cache/policyevalbundle/policyevalbundle.go @@ -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...) diff --git a/pkg/natsconn/natsconn.go b/pkg/natsconn/natsconn.go index f030bfd2f..8ba7aecbb 100644 --- a/pkg/natsconn/natsconn.go +++ b/pkg/natsconn/natsconn.go @@ -28,15 +28,17 @@ import ( // Config holds the connection parameters for NATS. // Decoupled from protobuf config so this package can be imported externally. type Config struct { - URI string - Token string - Name string + URI string + Token string + Name string + Replicas int // JetStream KV replica count; defaults to 1 } // ReloadableConnection wraps a NATS connection and provides reconnection // notifications via a pub/sub fan-out to subscribers. type ReloadableConnection struct { *nats.Conn + Replicas int // JetStream KV replica count mu sync.RWMutex subscribers []chan struct{} logger *log.Helper @@ -52,7 +54,7 @@ func New(cfg *Config, logger log.Logger) (*ReloadableConnection, func(), error) } l := log.NewHelper(log.With(logger, "component", "natsconn")) - rc := &ReloadableConnection{logger: l} + rc := &ReloadableConnection{logger: l, Replicas: cfg.Replicas} opts := []nats.Option{ nats.MaxReconnects(-1),