From bfb4d670ab8db8b4aa7044fd437b4ba572bd96bc Mon Sep 17 00:00:00 2001 From: Miguel Martinez Trivino Date: Thu, 8 Oct 2026 12:51:04 +0200 Subject: [PATCH 1/4] feat(cas): cache blob existence checks for uploads Cache positive existence results in the Artifact CAS, namespaced by organization and backend, so repeated uploads of the same evidence skip the secrets manager and the storage backend. Backend clients loaded for uploads are reused for a short time in process memory. The cache uses NATS KV when available and an in-memory LRU otherwise, and can be turned off or tuned through the CAS config and the Helm chart. Implements R-001 to R-009 of spec issue-3543. Refs #3543 Assisted-by: Claude Code Signed-off-by: Miguel Martinez Trivino Chainloop-Trace-Sessions: 9b6de27d-6322-4cf7-8e72-5e56d858c783 --- app/artifact-cas/cmd/main_test.go | 26 ++ app/artifact-cas/cmd/wire.go | 45 ++- app/artifact-cas/cmd/wire_gen.go | 57 +++- app/artifact-cas/internal/conf/conf.pb.go | 184 +++++++---- app/artifact-cas/internal/conf/conf.proto | 11 + .../internal/service/bytestream.go | 89 ++++-- .../internal/service/bytestream_cache_test.go | 301 ++++++++++++++++++ .../internal/service/bytestream_test.go | 13 +- app/artifact-cas/internal/service/cache.go | 145 +++++++++ app/artifact-cas/internal/service/service.go | 7 + deployment/chainloop/Chart.yaml | 2 +- deployment/chainloop/README.md | 5 +- .../chainloop/templates/cas/configmap.yaml | 3 + deployment/chainloop/values.yaml | 7 + pkg/cache/cache.go | 7 + pkg/cache/casexistence/casexistence.go | 89 ++++++ pkg/cache/memory.go | 9 +- pkg/cache/memory_test.go | 23 ++ 18 files changed, 937 insertions(+), 86 deletions(-) create mode 100644 app/artifact-cas/internal/service/bytestream_cache_test.go create mode 100644 app/artifact-cas/internal/service/cache.go create mode 100644 pkg/cache/casexistence/casexistence.go diff --git a/app/artifact-cas/cmd/main_test.go b/app/artifact-cas/cmd/main_test.go index c97004007..b2c8d84fb 100644 --- a/app/artifact-cas/cmd/main_test.go +++ b/app/artifact-cas/cmd/main_test.go @@ -19,9 +19,12 @@ import ( "os" "path/filepath" "testing" + "time" "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/conf" + "github.com/go-kratos/kratos/v2/log" "github.com/stretchr/testify/require" + "google.golang.org/protobuf/types/known/durationpb" ) // TestPrepareStagingDir: the staging directory is a hard startup requirement. @@ -86,3 +89,26 @@ func TestPrepareStagingDir(t *testing.T) { }) } } + +// TestNewExistenceCache: the existence cache is on by default and an operator +// can turn it off. +func TestNewExistenceCache(t *testing.T) { + testCases := []struct { + name string + cfg *conf.Bootstrap_ExistenceCache + wantNil bool + }{ + {name: "enabled by default"}, + {name: "custom ttl", cfg: &conf.Bootstrap_ExistenceCache{Ttl: durationpb.New(time.Hour)}}, + {name: "disabled", cfg: &conf.Bootstrap_ExistenceCache{Disabled: true}, wantNil: true}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + c, cleanup, err := newExistenceCache(&conf.Bootstrap{ExistenceCache: tc.cfg}, nil, log.DefaultLogger) + require.NoError(t, err) + t.Cleanup(cleanup) + require.Equal(t, tc.wantNil, c == nil) + }) + } +} diff --git a/app/artifact-cas/cmd/wire.go b/app/artifact-cas/cmd/wire.go index a3866888e..f65516bb4 100644 --- a/app/artifact-cas/cmd/wire.go +++ b/app/artifact-cas/cmd/wire.go @@ -21,11 +21,15 @@ package main import ( + "context" + "time" + "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/conf" "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/server" "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/service" "github.com/chainloop-dev/chainloop/app/controlplane/pkg/auditor" "github.com/chainloop-dev/chainloop/pkg/blobmanager/loader" + "github.com/chainloop-dev/chainloop/pkg/cache/casexistence" "github.com/chainloop-dev/chainloop/pkg/credentials" "github.com/chainloop-dev/chainloop/pkg/natsconn" "github.com/go-kratos/kratos/v2/log" @@ -47,16 +51,53 @@ func wireApp(*conf.Bootstrap, *conf.Server, *conf.Auth, credentials.Reader, log. // publish-only: the control plane owns the chainloop-audit stream configuration auditor.NewPublishOnlyAuditLogPublisher, service.NewAuditDispatcher, + newExistenceCache, ), ) } -func serviceOpts(l log.Logger, audit *service.AuditDispatcher, bc *conf.Bootstrap) []service.NewOpt { - return []service.NewOpt{ +// backendClientTTL is how long a loaded backend client, and its credentials, is reused for uploads +const backendClientTTL = 5 * time.Minute + +func serviceOpts(l log.Logger, audit *service.AuditDispatcher, bc *conf.Bootstrap, existence *casexistence.Cache) []service.NewOpt { + opts := []service.NewOpt{ service.WithLogger(l), service.WithAuditDispatcher(audit), service.WithStagingDir(bc.GetStagingDir()), + service.WithBackendClientCache(backendClientTTL), + } + + if existence != nil { + opts = append(opts, service.WithExistenceCache(existence)) } + + return opts +} + +// newExistenceCache returns the cache of blobs known to exist in a backend, +// nil when disabled. It is shared through NATS KV when the connection is set. +// A NATS bucket that can not be set up falls back to memory: the cache only +// makes uploads faster, so it must not stop the service from starting. +func newExistenceCache(bc *conf.Bootstrap, rc *natsconn.ReloadableConnection, logger log.Logger) (*casexistence.Cache, func(), error) { + cfg := bc.GetExistenceCache() + if cfg.GetDisabled() { + return nil, func() {}, nil + } + + ttl := cfg.GetTtl().AsDuration() + ctx, cancel := context.WithCancel(context.Background()) + c, err := casexistence.New(ctx, rc, ttl, logger) + if err != nil && rc != nil { + log.NewHelper(logger).Warnw("msg", "existence cache: NATS KV unavailable, using memory", "error", err) + c, err = casexistence.New(ctx, nil, ttl, logger) + } + + if err != nil { + cancel() + return nil, nil, err + } + + return c, cancel, nil } // newNatsConfig converts the proto config to a plain natsconn.Config, nil when unset diff --git a/app/artifact-cas/cmd/wire_gen.go b/app/artifact-cas/cmd/wire_gen.go index 4b48eef04..caa7f698a 100644 --- a/app/artifact-cas/cmd/wire_gen.go +++ b/app/artifact-cas/cmd/wire_gen.go @@ -7,14 +7,17 @@ package main import ( + "context" "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/conf" "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/server" "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/service" "github.com/chainloop-dev/chainloop/app/controlplane/pkg/auditor" "github.com/chainloop-dev/chainloop/pkg/blobmanager/loader" + "github.com/chainloop-dev/chainloop/pkg/cache/casexistence" "github.com/chainloop-dev/chainloop/pkg/credentials" "github.com/chainloop-dev/chainloop/pkg/natsconn" "github.com/go-kratos/kratos/v2/log" + "time" ) import ( @@ -37,37 +40,48 @@ func wireApp(bootstrap *conf.Bootstrap, confServer *conf.Server, auth *conf.Auth return nil, nil, err } auditDispatcher := service.NewAuditDispatcher(auditLogPublisher, logger) - v := serviceOpts(logger, auditDispatcher, bootstrap) + cache, cleanup2, err := newExistenceCache(bootstrap, reloadableConnection, logger) + if err != nil { + cleanup() + return nil, nil, err + } + v := serviceOpts(logger, auditDispatcher, bootstrap, cache) byteStreamService := service.NewByteStreamService(providers, v...) resourceService := service.NewResourceService(providers, v...) validator, err := newProtoValidator() if err != nil { + cleanup2() cleanup() return nil, nil, err } grpcServer, err := server.NewGRPCServer(confServer, auth, byteStreamService, resourceService, providers, validator, logger) if err != nil { + cleanup2() cleanup() return nil, nil, err } downloadService := service.NewDownloadService(providers, v...) httpServer, err := server.NewHTTPServer(confServer, auth, downloadService, providers, logger) if err != nil { + cleanup2() cleanup() return nil, nil, err } httpMetricsServer, err := server.NewHTTPMetricsServer(confServer) if err != nil { + cleanup2() cleanup() return nil, nil, err } - tracerProvider, cleanup2, err := server.NewTracerProvider(bootstrap, logger) + tracerProvider, cleanup3, err := server.NewTracerProvider(bootstrap, logger) if err != nil { + cleanup2() cleanup() return nil, nil, err } mainApp := newApp(logger, grpcServer, httpServer, httpMetricsServer, providers, tracerProvider) return mainApp, func() { + cleanup3() cleanup2() cleanup() }, nil @@ -75,8 +89,43 @@ func wireApp(bootstrap *conf.Bootstrap, confServer *conf.Server, auth *conf.Auth // wire.go: -func serviceOpts(l log.Logger, audit *service.AuditDispatcher, bc *conf.Bootstrap) []service.NewOpt { - return []service.NewOpt{service.WithLogger(l), service.WithAuditDispatcher(audit), service.WithStagingDir(bc.GetStagingDir())} +// backendClientTTL is how long a loaded backend client, and its credentials, is reused for uploads +const backendClientTTL = 5 * time.Minute + +func serviceOpts(l log.Logger, audit *service.AuditDispatcher, bc *conf.Bootstrap, existence *casexistence.Cache) []service.NewOpt { + opts := []service.NewOpt{service.WithLogger(l), service.WithAuditDispatcher(audit), service.WithStagingDir(bc.GetStagingDir()), service.WithBackendClientCache(backendClientTTL)} + + if existence != nil { + opts = append(opts, service.WithExistenceCache(existence)) + } + + return opts +} + +// newExistenceCache returns the cache of blobs known to exist in a backend, +// nil when disabled. It is shared through NATS KV when the connection is set. +// A NATS bucket that can not be set up falls back to memory: the cache only +// makes uploads faster, so it must not stop the service from starting. +func newExistenceCache(bc *conf.Bootstrap, rc *natsconn.ReloadableConnection, logger log.Logger) (*casexistence.Cache, func(), error) { + cfg := bc.GetExistenceCache() + if cfg.GetDisabled() { + return nil, func() {}, nil + } + + ttl := cfg.GetTtl().AsDuration() + ctx, cancel := context.WithCancel(context.Background()) + c, err := casexistence.New(ctx, rc, ttl, logger) + if err != nil && rc != nil { + log.NewHelper(logger).Warnw("msg", "existence cache: NATS KV unavailable, using memory", "error", err) + c, err = casexistence.New(ctx, nil, ttl, logger) + } + + if err != nil { + cancel() + return nil, nil, err + } + + return c, cancel, nil } // newNatsConfig converts the proto config to a plain natsconn.Config, nil when unset diff --git a/app/artifact-cas/internal/conf/conf.pb.go b/app/artifact-cas/internal/conf/conf.pb.go index ac72ca7ae..3b14a245d 100644 --- a/app/artifact-cas/internal/conf/conf.pb.go +++ b/app/artifact-cas/internal/conf/conf.pb.go @@ -52,9 +52,13 @@ type Bootstrap struct { // the first byte is sent to the client. It must be a writable volume; in // production a dedicated emptyDir is mounted here (NOT tmpfs/RAM, and NOT the // /tmp secret mount). Required: when unset the service refuses to start. - StagingDir string `protobuf:"bytes,6,opt,name=staging_dir,json=stagingDir,proto3" json:"staging_dir,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + StagingDir string `protobuf:"bytes,6,opt,name=staging_dir,json=stagingDir,proto3" json:"staging_dir,omitempty"` + // Cache of the blobs known to exist in a backend, so repeated uploads of the + // same content skip the secrets manager and the backend. Shared through NATS + // KV when nats_server is set, in memory otherwise. Enabled by default. + ExistenceCache *Bootstrap_ExistenceCache `protobuf:"bytes,7,opt,name=existence_cache,json=existenceCache,proto3" json:"existence_cache,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *Bootstrap) Reset() { @@ -129,6 +133,13 @@ func (x *Bootstrap) GetStagingDir() string { return "" } +func (x *Bootstrap) GetExistenceCache() *Bootstrap_ExistenceCache { + if x != nil { + return x.ExistenceCache + } + return nil +} + type Server struct { state protoimpl.MessageState `protogen:"open.v1"` // Regular HTTP endpoint @@ -250,6 +261,60 @@ func (x *Auth) GetPublicKeyPath() string { return "" } +type Bootstrap_ExistenceCache struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Turns the cache off + Disabled bool `protobuf:"varint,1,opt,name=disabled,proto3" json:"disabled,omitempty"` + // Lifetime of an entry. Defaults to 24h. + Ttl *durationpb.Duration `protobuf:"bytes,2,opt,name=ttl,proto3" json:"ttl,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Bootstrap_ExistenceCache) Reset() { + *x = Bootstrap_ExistenceCache{} + mi := &file_conf_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Bootstrap_ExistenceCache) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Bootstrap_ExistenceCache) ProtoMessage() {} + +func (x *Bootstrap_ExistenceCache) ProtoReflect() protoreflect.Message { + mi := &file_conf_proto_msgTypes[3] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Bootstrap_ExistenceCache.ProtoReflect.Descriptor instead. +func (*Bootstrap_ExistenceCache) Descriptor() ([]byte, []int) { + return file_conf_proto_rawDescGZIP(), []int{0, 0} +} + +func (x *Bootstrap_ExistenceCache) GetDisabled() bool { + if x != nil { + return x.Disabled + } + return false +} + +func (x *Bootstrap_ExistenceCache) GetTtl() *durationpb.Duration { + if x != nil { + return x.Ttl + } + return nil +} + type Bootstrap_NatsServer struct { state protoimpl.MessageState `protogen:"open.v1"` // NATS server URI, e.g. "nats://localhost:4222" @@ -264,7 +329,7 @@ type Bootstrap_NatsServer struct { func (x *Bootstrap_NatsServer) Reset() { *x = Bootstrap_NatsServer{} - mi := &file_conf_proto_msgTypes[3] + mi := &file_conf_proto_msgTypes[4] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -276,7 +341,7 @@ func (x *Bootstrap_NatsServer) String() string { func (*Bootstrap_NatsServer) ProtoMessage() {} func (x *Bootstrap_NatsServer) ProtoReflect() protoreflect.Message { - mi := &file_conf_proto_msgTypes[3] + mi := &file_conf_proto_msgTypes[4] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -289,7 +354,7 @@ func (x *Bootstrap_NatsServer) ProtoReflect() protoreflect.Message { // Deprecated: Use Bootstrap_NatsServer.ProtoReflect.Descriptor instead. func (*Bootstrap_NatsServer) Descriptor() ([]byte, []int) { - return file_conf_proto_rawDescGZIP(), []int{0, 0} + return file_conf_proto_rawDescGZIP(), []int{0, 1} } func (x *Bootstrap_NatsServer) GetUri() string { @@ -335,7 +400,7 @@ type Bootstrap_Observability struct { func (x *Bootstrap_Observability) Reset() { *x = Bootstrap_Observability{} - mi := &file_conf_proto_msgTypes[4] + mi := &file_conf_proto_msgTypes[5] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -347,7 +412,7 @@ func (x *Bootstrap_Observability) String() string { func (*Bootstrap_Observability) ProtoMessage() {} func (x *Bootstrap_Observability) ProtoReflect() protoreflect.Message { - mi := &file_conf_proto_msgTypes[4] + mi := &file_conf_proto_msgTypes[5] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -360,7 +425,7 @@ func (x *Bootstrap_Observability) ProtoReflect() protoreflect.Message { // Deprecated: Use Bootstrap_Observability.ProtoReflect.Descriptor instead. func (*Bootstrap_Observability) Descriptor() ([]byte, []int) { - return file_conf_proto_rawDescGZIP(), []int{0, 1} + return file_conf_proto_rawDescGZIP(), []int{0, 2} } func (x *Bootstrap_Observability) GetSentry() *Bootstrap_Observability_Sentry { @@ -388,7 +453,7 @@ type Bootstrap_Observability_Sentry struct { func (x *Bootstrap_Observability_Sentry) Reset() { *x = Bootstrap_Observability_Sentry{} - mi := &file_conf_proto_msgTypes[5] + mi := &file_conf_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -400,7 +465,7 @@ func (x *Bootstrap_Observability_Sentry) String() string { func (*Bootstrap_Observability_Sentry) ProtoMessage() {} func (x *Bootstrap_Observability_Sentry) ProtoReflect() protoreflect.Message { - mi := &file_conf_proto_msgTypes[5] + mi := &file_conf_proto_msgTypes[6] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -413,7 +478,7 @@ func (x *Bootstrap_Observability_Sentry) ProtoReflect() protoreflect.Message { // Deprecated: Use Bootstrap_Observability_Sentry.ProtoReflect.Descriptor instead. func (*Bootstrap_Observability_Sentry) Descriptor() ([]byte, []int) { - return file_conf_proto_rawDescGZIP(), []int{0, 1, 0} + return file_conf_proto_rawDescGZIP(), []int{0, 2, 0} } func (x *Bootstrap_Observability_Sentry) GetDsn() string { @@ -447,7 +512,7 @@ type Bootstrap_Observability_Tracing struct { func (x *Bootstrap_Observability_Tracing) Reset() { *x = Bootstrap_Observability_Tracing{} - mi := &file_conf_proto_msgTypes[6] + mi := &file_conf_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -459,7 +524,7 @@ func (x *Bootstrap_Observability_Tracing) String() string { func (*Bootstrap_Observability_Tracing) ProtoMessage() {} func (x *Bootstrap_Observability_Tracing) ProtoReflect() protoreflect.Message { - mi := &file_conf_proto_msgTypes[6] + mi := &file_conf_proto_msgTypes[7] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -472,7 +537,7 @@ func (x *Bootstrap_Observability_Tracing) ProtoReflect() protoreflect.Message { // Deprecated: Use Bootstrap_Observability_Tracing.ProtoReflect.Descriptor instead. func (*Bootstrap_Observability_Tracing) Descriptor() ([]byte, []int) { - return file_conf_proto_rawDescGZIP(), []int{0, 1, 1} + return file_conf_proto_rawDescGZIP(), []int{0, 2, 1} } func (x *Bootstrap_Observability_Tracing) GetEnabled() bool { @@ -512,7 +577,7 @@ type Server_CORS struct { func (x *Server_CORS) Reset() { *x = Server_CORS{} - mi := &file_conf_proto_msgTypes[7] + mi := &file_conf_proto_msgTypes[8] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -524,7 +589,7 @@ func (x *Server_CORS) String() string { func (*Server_CORS) ProtoMessage() {} func (x *Server_CORS) ProtoReflect() protoreflect.Message { - mi := &file_conf_proto_msgTypes[7] + mi := &file_conf_proto_msgTypes[8] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -560,7 +625,7 @@ type Server_HTTP struct { func (x *Server_HTTP) Reset() { *x = Server_HTTP{} - mi := &file_conf_proto_msgTypes[8] + mi := &file_conf_proto_msgTypes[9] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -572,7 +637,7 @@ func (x *Server_HTTP) String() string { func (*Server_HTTP) ProtoMessage() {} func (x *Server_HTTP) ProtoReflect() protoreflect.Message { - mi := &file_conf_proto_msgTypes[8] + mi := &file_conf_proto_msgTypes[9] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -627,7 +692,7 @@ type Server_TLS struct { func (x *Server_TLS) Reset() { *x = Server_TLS{} - mi := &file_conf_proto_msgTypes[9] + mi := &file_conf_proto_msgTypes[10] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -639,7 +704,7 @@ func (x *Server_TLS) String() string { func (*Server_TLS) ProtoMessage() {} func (x *Server_TLS) ProtoReflect() protoreflect.Message { - mi := &file_conf_proto_msgTypes[9] + mi := &file_conf_proto_msgTypes[10] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -681,7 +746,7 @@ type Server_GRPC struct { func (x *Server_GRPC) Reset() { *x = Server_GRPC{} - mi := &file_conf_proto_msgTypes[10] + mi := &file_conf_proto_msgTypes[11] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -693,7 +758,7 @@ func (x *Server_GRPC) String() string { func (*Server_GRPC) ProtoMessage() {} func (x *Server_GRPC) ProtoReflect() protoreflect.Message { - mi := &file_conf_proto_msgTypes[10] + mi := &file_conf_proto_msgTypes[11] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -742,7 +807,7 @@ var File_conf_proto protoreflect.FileDescriptor const file_conf_proto_rawDesc = "" + "\n" + "\n" + - "conf.proto\x1a\x1bcredentials/v1/config.proto\x1a\x1egoogle/protobuf/duration.proto\"\xda\x05\n" + + "conf.proto\x1a\x1bcredentials/v1/config.proto\x1a\x1egoogle/protobuf/duration.proto\"\xf9\x06\n" + "\tBootstrap\x12\x1f\n" + "\x06server\x18\x01 \x01(\v2\a.ServerR\x06server\x12\x19\n" + "\x04auth\x18\x02 \x01(\v2\x05.AuthR\x04auth\x12>\n" + @@ -751,7 +816,11 @@ const file_conf_proto_rawDesc = "" + "\vnats_server\x18\x05 \x01(\v2\x15.Bootstrap.NatsServerR\n" + "natsServer\x12\x1f\n" + "\vstaging_dir\x18\x06 \x01(\tR\n" + - "stagingDir\x1aH\n" + + "stagingDir\x12B\n" + + "\x0fexistence_cache\x18\a \x01(\v2\x19.Bootstrap.ExistenceCacheR\x0eexistenceCache\x1aY\n" + + "\x0eExistenceCache\x12\x1a\n" + + "\bdisabled\x18\x01 \x01(\bR\bdisabled\x12+\n" + + "\x03ttl\x18\x02 \x01(\v2\x19.google.protobuf.DurationR\x03ttl\x1aH\n" + "\n" + "NatsServer\x12\x10\n" + "\x03uri\x18\x01 \x01(\tR\x03uri\x12\x16\n" + @@ -806,42 +875,45 @@ func file_conf_proto_rawDescGZIP() []byte { return file_conf_proto_rawDescData } -var file_conf_proto_msgTypes = make([]protoimpl.MessageInfo, 11) +var file_conf_proto_msgTypes = make([]protoimpl.MessageInfo, 12) var file_conf_proto_goTypes = []any{ (*Bootstrap)(nil), // 0: Bootstrap (*Server)(nil), // 1: Server (*Auth)(nil), // 2: Auth - (*Bootstrap_NatsServer)(nil), // 3: Bootstrap.NatsServer - (*Bootstrap_Observability)(nil), // 4: Bootstrap.Observability - (*Bootstrap_Observability_Sentry)(nil), // 5: Bootstrap.Observability.Sentry - (*Bootstrap_Observability_Tracing)(nil), // 6: Bootstrap.Observability.Tracing - (*Server_CORS)(nil), // 7: Server.CORS - (*Server_HTTP)(nil), // 8: Server.HTTP - (*Server_TLS)(nil), // 9: Server.TLS - (*Server_GRPC)(nil), // 10: Server.GRPC - (*v1.Credentials)(nil), // 11: credentials.v1.Credentials - (*durationpb.Duration)(nil), // 12: google.protobuf.Duration + (*Bootstrap_ExistenceCache)(nil), // 3: Bootstrap.ExistenceCache + (*Bootstrap_NatsServer)(nil), // 4: Bootstrap.NatsServer + (*Bootstrap_Observability)(nil), // 5: Bootstrap.Observability + (*Bootstrap_Observability_Sentry)(nil), // 6: Bootstrap.Observability.Sentry + (*Bootstrap_Observability_Tracing)(nil), // 7: Bootstrap.Observability.Tracing + (*Server_CORS)(nil), // 8: Server.CORS + (*Server_HTTP)(nil), // 9: Server.HTTP + (*Server_TLS)(nil), // 10: Server.TLS + (*Server_GRPC)(nil), // 11: Server.GRPC + (*v1.Credentials)(nil), // 12: credentials.v1.Credentials + (*durationpb.Duration)(nil), // 13: google.protobuf.Duration } var file_conf_proto_depIdxs = []int32{ 1, // 0: Bootstrap.server:type_name -> Server 2, // 1: Bootstrap.auth:type_name -> Auth - 4, // 2: Bootstrap.observability:type_name -> Bootstrap.Observability - 11, // 3: Bootstrap.credentials_service:type_name -> credentials.v1.Credentials - 3, // 4: Bootstrap.nats_server:type_name -> Bootstrap.NatsServer - 8, // 5: Server.http:type_name -> Server.HTTP - 10, // 6: Server.grpc:type_name -> Server.GRPC - 8, // 7: Server.http_metrics:type_name -> Server.HTTP - 5, // 8: Bootstrap.Observability.sentry:type_name -> Bootstrap.Observability.Sentry - 6, // 9: Bootstrap.Observability.tracing:type_name -> Bootstrap.Observability.Tracing - 12, // 10: Server.HTTP.timeout:type_name -> google.protobuf.Duration - 7, // 11: Server.HTTP.cors:type_name -> Server.CORS - 12, // 12: Server.GRPC.timeout:type_name -> google.protobuf.Duration - 9, // 13: Server.GRPC.tls_config:type_name -> Server.TLS - 14, // [14:14] is the sub-list for method output_type - 14, // [14:14] is the sub-list for method input_type - 14, // [14:14] is the sub-list for extension type_name - 14, // [14:14] is the sub-list for extension extendee - 0, // [0:14] is the sub-list for field type_name + 5, // 2: Bootstrap.observability:type_name -> Bootstrap.Observability + 12, // 3: Bootstrap.credentials_service:type_name -> credentials.v1.Credentials + 4, // 4: Bootstrap.nats_server:type_name -> Bootstrap.NatsServer + 3, // 5: Bootstrap.existence_cache:type_name -> Bootstrap.ExistenceCache + 9, // 6: Server.http:type_name -> Server.HTTP + 11, // 7: Server.grpc:type_name -> Server.GRPC + 9, // 8: Server.http_metrics:type_name -> Server.HTTP + 13, // 9: Bootstrap.ExistenceCache.ttl:type_name -> google.protobuf.Duration + 6, // 10: Bootstrap.Observability.sentry:type_name -> Bootstrap.Observability.Sentry + 7, // 11: Bootstrap.Observability.tracing:type_name -> Bootstrap.Observability.Tracing + 13, // 12: Server.HTTP.timeout:type_name -> google.protobuf.Duration + 8, // 13: Server.HTTP.cors:type_name -> Server.CORS + 13, // 14: Server.GRPC.timeout:type_name -> google.protobuf.Duration + 10, // 15: Server.GRPC.tls_config:type_name -> Server.TLS + 16, // [16:16] is the sub-list for method output_type + 16, // [16:16] is the sub-list for method input_type + 16, // [16:16] is the sub-list for extension type_name + 16, // [16:16] is the sub-list for extension extendee + 0, // [0:16] is the sub-list for field type_name } func init() { file_conf_proto_init() } @@ -849,17 +921,17 @@ func file_conf_proto_init() { if File_conf_proto != nil { return } - file_conf_proto_msgTypes[3].OneofWrappers = []any{ + file_conf_proto_msgTypes[4].OneofWrappers = []any{ (*Bootstrap_NatsServer_Token)(nil), } - file_conf_proto_msgTypes[6].OneofWrappers = []any{} + file_conf_proto_msgTypes[7].OneofWrappers = []any{} type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_conf_proto_rawDesc), len(file_conf_proto_rawDesc)), NumEnums: 0, - NumMessages: 11, + NumMessages: 12, NumExtensions: 0, NumServices: 0, }, diff --git a/app/artifact-cas/internal/conf/conf.proto b/app/artifact-cas/internal/conf/conf.proto index d9c8c5399..44ee8fef2 100644 --- a/app/artifact-cas/internal/conf/conf.proto +++ b/app/artifact-cas/internal/conf/conf.proto @@ -34,6 +34,17 @@ message Bootstrap { // production a dedicated emptyDir is mounted here (NOT tmpfs/RAM, and NOT the // /tmp secret mount). Required: when unset the service refuses to start. string staging_dir = 6; + // Cache of the blobs known to exist in a backend, so repeated uploads of the + // same content skip the secrets manager and the backend. Shared through NATS + // KV when nats_server is set, in memory otherwise. Enabled by default. + ExistenceCache existence_cache = 7; + + message ExistenceCache { + // Turns the cache off + bool disabled = 1; + // Lifetime of an entry. Defaults to 24h. + google.protobuf.Duration ttl = 2; + } message NatsServer { // NATS server URI, e.g. "nats://localhost:4222" diff --git a/app/artifact-cas/internal/service/bytestream.go b/app/artifact-cas/internal/service/bytestream.go index 6a4cbe5f7..fa59e5e0d 100644 --- a/app/artifact-cas/internal/service/bytestream.go +++ b/app/artifact-cas/internal/service/bytestream.go @@ -79,34 +79,33 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro return kerrors.BadRequest("resource name", err.Error()) } - storageBackend, err := s.loadBackend(ctx, info.BackendType, info.StoredSecretID) - if err != nil && kerrors.IsNotFound(err) { - return err - } else if err != nil { - return sl.LogAndMaskErr(err, s.log) + // A blob cached as present skips the secrets manager and the backend. The + // key is namespaced by the organization and backend of the signed token. + existenceKey := s.existenceKey(info, req.resource.Digest) + if entry, ok := s.lookupExistence(ctx, existenceKey); ok { + s.log.Infow("msg", "artifact already exists (cached)", "digest", req.resource.Digest) + s.dispatchSkippedUpload(info, req.resource, entry.Size) + return stream.SendAndClose(&bytestream.WriteResponse{}) } // We check if the file already exists even before we wait for the whole buffer to be filled - if exists, err := storageBackend.Exists(ctx, req.resource.Digest); err != nil { - return sl.LogAndMaskErr(err, s.log) - } else if exists { + storageBackend, exists, err := s.backendExists(ctx, info, req.resource.Digest) + if err != nil { + return err + } + + if exists { s.log.Infow("msg", "artifact already exists", "digest", req.resource.Digest) - if s.audit.shouldEmit(info) { - // the stored size is not known at the dedup point, look it up best-effort + if existenceKey != "" || s.audit.shouldEmit(info) { + // the stored size is not known at the dedup point, look it up best-effort. + // Only a known size is cached, so cached hits keep complete audit events. var size int64 if r, err := storageBackend.Describe(ctx, req.resource.Digest); err == nil { size = r.Size + s.storeExistence(ctx, existenceKey, size) } - s.audit.Dispatch(&events.CASArtifactUploaded{ - CASArtifactBase: &events.CASArtifactBase{ - Digest: req.resource.Digest, - SizeBytes: size, - FileName: req.resource.FileName, - BackendType: info.BackendType, - }, - Skipped: true, - }, info) + s.dispatchSkippedUpload(info, req.resource, size) } return stream.SendAndClose(&bytestream.WriteResponse{}) } @@ -137,6 +136,8 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro return status.Error(codes.InvalidArgument, err.Error()) } if backendErr, ok := errors.AsType[*backendUploadError](err); ok { + // the reused client may hold stale credentials + s.dropUploadBackend(info) return sl.LogAndMaskErr(backendErr.err, s.log) } if isClientDisconnect(err) { @@ -150,6 +151,7 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro } s.log.Infow("msg", "upload finished", "name", req.resource.FileName, "digest", req.resource.Digest, "size", committedSize) + s.storeExistence(ctx, existenceKey, committedSize) s.audit.Dispatch(&events.CASArtifactUploaded{ CASArtifactBase: &events.CASArtifactBase{ Digest: req.resource.Digest, @@ -162,6 +164,55 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro return stream.SendAndClose(&bytestream.WriteResponse{CommittedSize: committedSize}) } +// backendExists loads the backend of the token and checks if digest exists in +// it. A reused client that fails is dropped and loaded again one time, since +// its credentials may have been rotated. Returned errors are ready to be sent +// to the client. +func (s *ByteStreamService) backendExists(ctx context.Context, info *casJWT.Claims, digest string) (backend.UploaderDownloader, bool, error) { + storageBackend, cached, err := s.loadUploadBackend(ctx, info) + if err != nil { + return nil, false, s.loadBackendError(err) + } + + exists, err := storageBackend.Exists(ctx, digest) + if err != nil && cached { + s.log.Warnw("msg", "reused backend client failed, loading it again", "digest", digest, "error", err) + s.dropUploadBackend(info) + if storageBackend, _, err = s.loadUploadBackend(ctx, info); err != nil { + return nil, false, s.loadBackendError(err) + } + exists, err = storageBackend.Exists(ctx, digest) + } + + if err != nil { + s.dropUploadBackend(info) + return nil, false, sl.LogAndMaskErr(err, s.log) + } + + return storageBackend, exists, nil +} + +func (s *ByteStreamService) loadBackendError(err error) error { + if kerrors.IsNotFound(err) { + return err + } + return sl.LogAndMaskErr(err, s.log) +} + +// dispatchSkippedUpload emits the audit event of an upload skipped because the +// blob already exists +func (s *ByteStreamService) dispatchSkippedUpload(info *casJWT.Claims, resource *v1.CASResource, size int64) { + s.audit.Dispatch(&events.CASArtifactUploaded{ + CASArtifactBase: &events.CASArtifactBase{ + Digest: resource.Digest, + SizeBytes: size, + FileName: resource.FileName, + BackendType: info.BackendType, + }, + Skipped: true, + }, info) +} + // spillVerifyUpload stages an upload on local disk, verifies its digest, and // stores it. The client's bytes are streamed into a temporary file and hashed in // the same pass; the file is handed to the backend only once the hash matches the diff --git a/app/artifact-cas/internal/service/bytestream_cache_test.go b/app/artifact-cas/internal/service/bytestream_cache_test.go new file mode 100644 index 000000000..964affef6 --- /dev/null +++ b/app/artifact-cas/internal/service/bytestream_cache_test.go @@ -0,0 +1,301 @@ +// +// Copyright 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. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package service + +import ( + "context" + "errors" + "testing" + "time" + + v1 "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1" + casJWT "github.com/chainloop-dev/chainloop/internal/robotaccount/cas" + "github.com/chainloop-dev/chainloop/pkg/blobmanager/mocks" + "github.com/chainloop-dev/chainloop/pkg/cache" + "github.com/chainloop-dev/chainloop/pkg/cache/casexistence" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + "google.golang.org/genproto/googleapis/bytestream" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" +) + +const otherOrgID = "2b7d0d6e-7d3f-4a39-9a43-7c1d2f0e9b11" + +// write uploads data as resource in a single chunk +func (s *bytestreamSuite) write(ctx context.Context, resource *v1.CASResource, data []byte) (*bytestream.WriteResponse, error) { + stream, err := s.client.Write(ctx) + s.Require().NoError(err) + s.Require().NoError(stream.Send(&bytestream.WriteRequest{ + ResourceName: encodeResource(s.T(), resource), + Data: data, + })) + + return stream.CloseAndRecv() +} + +func uploaderCtx(kv ...string) context.Context { + return metadata.NewOutgoingContext(context.Background(), metadata.Pairs(append([]string{"role", string(casJWT.Uploader)}, kv...)...)) +} + +// restart replaces the server with one that also has opts +func (s *bytestreamSuite) restart(opts ...NewOpt) { + s.TearDownTest() + s.setup(opts...) +} + +// withExistenceCache restarts the server with the existence cache only, so a +// cached hit can be told apart from a reused backend client. +func (s *bytestreamSuite) withExistenceCache() { + c, err := cache.New[casexistence.Entry](cache.WithTTL(time.Hour)) + s.Require().NoError(err) + s.restart(WithExistenceCache(c)) +} + +// withClientCache restarts the server with the backend client cache only +func (s *bytestreamSuite) withClientCache() { + s.restart(WithBackendClientCache(time.Hour)) +} + +// R-001, R-006: a second upload of a blob that the backend reported as present +// makes no secrets manager or backend call, and its event keeps the size. +func (s *bytestreamSuite) TestExistenceCacheHitSkipsBackend() { + s.withExistenceCache() + s.ociBackend.On("Exists", mock.Anything, s.resource.Digest).Return(true, nil).Once() + s.ociBackend.On("Describe", mock.Anything, s.resource.Digest).Return(&v1.CASResource{ + FileName: s.resource.FileName, Digest: s.resource.Digest, Size: 1024, + }, nil).Once() + + for range 2 { + got, err := s.write(s.upCtx, s.resource, nil) + s.Require().NoError(err) + s.Equal(int64(0), got.CommittedSize) + } + + s.ociProvider.AssertNumberOfCalls(s.T(), "FromCredentials", 1) + s.ociBackend.AssertNumberOfCalls(s.T(), "Exists", 1) + s.ociBackend.AssertNumberOfCalls(s.T(), "Describe", 1) + + s.Require().Len(s.audit.published, 2) + for _, e := range s.audit.published { + info := decodeArtifactEvent(s.T(), e) + s.True(info.Skipped) + s.Equal(int64(1024), info.SizeBytes) + s.Equal(s.resource.FileName, info.FileName) + } +} + +// R-002: a successful upload is cached with its committed size. +func (s *bytestreamSuite) TestExistenceCacheUploadSuccessIsCached() { + s.withExistenceCache() + data := []byte("hello world") + resource := resourceWithDigest(data, "skynet.exe") + s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil).Once() + s.ociBackend.On("Upload", mock.Anything, mock.Anything, resource).Return(nil).Once() + + got, err := s.write(s.upCtx, resource, data) + s.Require().NoError(err) + s.Equal(int64(len(data)), got.CommittedSize) + + got, err = s.write(s.upCtx, resource, data) + s.Require().NoError(err) + s.Equal(int64(0), got.CommittedSize) + + s.ociProvider.AssertNumberOfCalls(s.T(), "FromCredentials", 1) + s.Require().Len(s.audit.published, 2) + info := decodeArtifactEvent(s.T(), s.audit.published[1]) + s.True(info.Skipped) + s.Equal(int64(len(data)), info.SizeBytes) +} + +// R-002: a "not found" answer, or a found blob without a known size, is not cached. +func (s *bytestreamSuite) TestExistenceCacheNotCached() { + data := []byte("hello world") + resource := resourceWithDigest(data, "skynet.exe") + + testCases := []struct { + name string + setup func() + code codes.Code + }{ + { + name: "not found and failed upload", + setup: func() { + s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) + s.ociBackend.On("Upload", mock.Anything, mock.Anything, resource).Return(errors.New("boom")) + }, + code: codes.Internal, + }, + { + name: "found but describe fails", + setup: func() { + s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(true, nil) + s.ociBackend.On("Describe", mock.Anything, resource.Digest).Return(nil, errors.New("boom")) + }, + code: codes.OK, + }, + } + + for _, tc := range testCases { + s.Run(tc.name, func() { + s.withExistenceCache() + tc.setup() + for range 2 { + _, err := s.write(s.upCtx, resource, data) + if tc.code == codes.OK { + s.Require().NoError(err) + } else { + assertGRPCError(s.T(), err, tc.code, "") + } + } + s.ociBackend.AssertNumberOfCalls(s.T(), "Exists", 2) + }) + } +} + +// R-003: an entry is only visible to the same organization and backend. +func (s *bytestreamSuite) TestExistenceCacheTenantNamespace() { + testCases := []struct { + name string + secondCtx context.Context + // backend the second upload must reach + second func() *mockedBackend + }{ + { + name: "different organization on shared storage", + secondCtx: uploaderCtx("org-id", otherOrgID), + second: func() *mockedBackend { return &mockedBackend{s.ociBackend, 2} }, + }, + { + name: "same organization, different backend", + secondCtx: uploaderCtx("backend-streaming", "true"), + second: func() *mockedBackend { return &mockedBackend{s.streamingBackend, 1} }, + }, + { + name: "organization missing from the token", + secondCtx: uploaderCtx("org-id", ""), + second: func() *mockedBackend { return &mockedBackend{s.ociBackend, 2} }, + }, + } + + for _, tc := range testCases { + s.Run(tc.name, func() { + s.withExistenceCache() + describe := &v1.CASResource{Digest: s.resource.Digest, Size: 1} + for _, b := range []*mockedBackend{{s.ociBackend, 0}, {s.streamingBackend, 0}} { + b.On("Exists", mock.Anything, s.resource.Digest).Maybe().Return(true, nil) + b.On("Describe", mock.Anything, s.resource.Digest).Maybe().Return(describe, nil) + } + + _, err := s.write(s.upCtx, s.resource, nil) + s.Require().NoError(err) + _, err = s.write(tc.secondCtx, s.resource, nil) + s.Require().NoError(err) + + want := tc.second() + want.AssertNumberOfCalls(s.T(), "Exists", want.calls) + }) + } +} + +// R-008: a failing cache is a miss and never fails the upload. +func (s *bytestreamSuite) TestExistenceCacheFailureIsMiss() { + s.restart(WithExistenceCache(failingCache{})) + + data := []byte("hello world") + resource := resourceWithDigest(data, "skynet.exe") + s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) + s.ociBackend.On("Upload", mock.Anything, mock.Anything, resource).Return(nil) + + for range 2 { + got, err := s.write(s.upCtx, resource, data) + s.Require().NoError(err) + s.Equal(int64(len(data)), got.CommittedSize) + } + s.ociBackend.AssertNumberOfCalls(s.T(), "Upload", 2) +} + +// R-007: a cache miss on the existence check reuses the loaded backend client. +func (s *bytestreamSuite) TestClientCacheReusesClient() { + s.withClientCache() + s.ociBackend.On("Exists", mock.Anything, mock.Anything).Return(true, nil) + + for _, d := range []string{"digest-a", "digest-b"} { + _, err := s.write(uploaderCtx("source-internal", "true"), &v1.CASResource{Digest: d}, nil) + s.Require().NoError(err) + } + s.ociProvider.AssertNumberOfCalls(s.T(), "FromCredentials", 1) + + // a client is never shared between organizations + _, err := s.write(uploaderCtx("source-internal", "true", "org-id", otherOrgID), s.resource, nil) + s.Require().NoError(err) + s.ociProvider.AssertNumberOfCalls(s.T(), "FromCredentials", 2) +} + +// D-009: an error from a reused client drops it and loads it again one time. +func (s *bytestreamSuite) TestClientCacheReloadOnError() { + s.withClientCache() + ctx := uploaderCtx("source-internal", "true") + s.ociBackend.On("Exists", mock.Anything, mock.Anything).Return(true, nil).Once() + _, err := s.write(ctx, s.resource, nil) + s.Require().NoError(err) + + s.ociBackend.On("Exists", mock.Anything, mock.Anything).Return(false, errors.New("expired credentials")).Once() + s.ociBackend.On("Exists", mock.Anything, mock.Anything).Return(true, nil).Once() + _, err = s.write(ctx, s.resource, nil) + s.Require().NoError(err) + s.ociProvider.AssertNumberOfCalls(s.T(), "FromCredentials", 2) +} + +// a fresh client that fails is not loaded again +func (s *bytestreamSuite) TestClientCacheNoReloadOfFreshClient() { + s.withClientCache() + s.ociBackend.On("Exists", mock.Anything, mock.Anything).Return(false, errors.New("boom")) + + _, err := s.write(s.upCtx, s.resource, nil) + assertGRPCError(s.T(), err, codes.Internal, "") + s.ociProvider.AssertNumberOfCalls(s.T(), "FromCredentials", 1) + s.ociBackend.AssertNumberOfCalls(s.T(), "Exists", 1) +} + +type mockedBackend struct { + *mocks.UploaderDownloader + calls int +} + +type failingCache struct{} + +var errCacheDown = errors.New("cache is down") + +func (failingCache) Get(context.Context, string) (casexistence.Entry, bool, error) { + return casexistence.Entry{}, false, errCacheDown +} +func (failingCache) Set(context.Context, string, casexistence.Entry) error { return errCacheDown } +func (failingCache) Delete(context.Context, string) error { return errCacheDown } +func (failingCache) Purge(context.Context) error { return errCacheDown } + +func TestExistenceCacheKey(t *testing.T) { + base := casexistence.Key("org", "OCI", "secret", "digest") + require.NotContains(t, base, "secret", "the secret reference must be hashed") + for _, other := range []string{ + casexistence.Key("org2", "OCI", "secret", "digest"), + casexistence.Key("org", "S3", "secret", "digest"), + casexistence.Key("org", "OCI", "secret2", "digest"), + casexistence.Key("org", "OCI", "secret", "digest2"), + } { + require.NotEqual(t, base, other) + } +} diff --git a/app/artifact-cas/internal/service/bytestream_test.go b/app/artifact-cas/internal/service/bytestream_test.go index 9eda0b4db..2c10d6dc0 100644 --- a/app/artifact-cas/internal/service/bytestream_test.go +++ b/app/artifact-cas/internal/service/bytestream_test.go @@ -425,6 +425,8 @@ type bytestreamSuite struct { srv *grpc.Server client bytestream.ByteStreamClient ociBackend *mocks.UploaderDownloader + // ociProvider builds ociBackend; tests assert how often credentials are loaded + ociProvider *mocks.Provider // streamingBackend is the backend reachable via the "backend-streaming" // metadata. streamingBackend *mocks.UploaderDownloader @@ -455,6 +457,11 @@ func (s *bytestreamSuite) TearDownSubTest() { } func (s *bytestreamSuite) SetupTest() { + s.setup() +} + +// setup starts the server with the default options plus opts +func (s *bytestreamSuite) setup(opts ...NewOpt) { const backendType = "backend-type" // 1 MB buffer l := bufconn.Listen(1 << 20) @@ -477,6 +484,9 @@ func (s *bytestreamSuite) SetupTest() { if v := md.Get("backend-streaming"); len(v) > 0 { claims.BackendType = streamingBackendType } + if v := md.Get("org-id"); len(v) > 0 { + claims.OrgID = v[0] + } if v := md.Get("source-internal"); len(v) > 0 { claims.SourceInternal = true } @@ -518,7 +528,7 @@ func (s *bytestreamSuite) SetupTest() { NewByteStreamService(backend.Providers{ backendType: ociBackendProvider, streamingBackendType: streamingBackendProvider, - }, WithLogger(log.DefaultLogger), WithAuditDispatcher(newTestDispatcher(s.audit)), WithStagingDir(s.stagingDir)), + }, append([]NewOpt{WithLogger(log.DefaultLogger), WithAuditDispatcher(newTestDispatcher(s.audit)), WithStagingDir(s.stagingDir)}, opts...)...), ) go func() { _ = server.Serve(l) @@ -534,6 +544,7 @@ func (s *bytestreamSuite) SetupTest() { s.srv = server s.conn = conn s.ociBackend = ociBackend + s.ociProvider = ociBackendProvider s.streamingBackend = streamingBackend s.client = bytestream.NewByteStreamClient(conn) s.resource = &v1.CASResource{ diff --git a/app/artifact-cas/internal/service/cache.go b/app/artifact-cas/internal/service/cache.go new file mode 100644 index 000000000..ad1ff0780 --- /dev/null +++ b/app/artifact-cas/internal/service/cache.go @@ -0,0 +1,145 @@ +// +// Copyright 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. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package service + +import ( + "context" + "strings" + "time" + + casJWT "github.com/chainloop-dev/chainloop/internal/robotaccount/cas" + backend "github.com/chainloop-dev/chainloop/pkg/blobmanager" + "github.com/chainloop-dev/chainloop/pkg/cache" + "github.com/chainloop-dev/chainloop/pkg/cache/casexistence" + "github.com/hashicorp/golang-lru/v2/expirable" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +// maxCachedClients bounds the number of backend clients kept in memory +const maxCachedClients = 1000 + +var ( + existenceCacheLookups = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "chainloop_cas_existence_cache_lookups_total", + Help: "Lookups in the cache of blobs known to exist in a backend, by result (hit or miss).", + }, []string{"result"}) + + clientCacheLookups = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "chainloop_cas_backend_client_cache_lookups_total", + Help: "Lookups in the cache of loaded backend clients, by result (hit or miss).", + }, []string{"result"}) +) + +func recordLookup(c *prometheus.CounterVec, hit bool) { + if hit { + c.WithLabelValues("hit").Inc() + } else { + c.WithLabelValues("miss").Inc() + } +} + +// WithExistenceCache enables the cache of blobs known to exist in a backend. +// Uploads of a cached digest end as "already exists" with no backend call. +func WithExistenceCache(c cache.Cache[casexistence.Entry]) NewOpt { + return func(s *commonService) { + s.existence = c + } +} + +// WithBackendClientCache enables the reuse of the backend clients loaded for +// uploads for ttl. Clients and their credentials are kept in process memory only. +func WithBackendClientCache(ttl time.Duration) NewOpt { + return func(s *commonService) { + s.clients = expirable.NewLRU[string, backend.UploaderDownloader](maxCachedClients, nil, ttl) + } +} + +// existenceKey returns the existence cache key of digest in the namespace of +// the organization and backend of the token, or "" when the upload can not be cached. +func (s *commonService) existenceKey(info *casJWT.Claims, digest string) string { + if s.existence == nil || info.OrgID == "" { + return "" + } + + return casexistence.Key(info.OrgID, info.BackendType, info.StoredSecretID, digest) +} + +// lookupExistence reports if key is cached as present. Cache errors are a miss. +func (s *commonService) lookupExistence(ctx context.Context, key string) (casexistence.Entry, bool) { + if key == "" { + return casexistence.Entry{}, false + } + + entry, ok, err := s.existence.Get(ctx, key) + if err != nil { + s.log.Warnw("msg", "existence cache lookup failed", "error", err) + ok = false + } + + recordLookup(existenceCacheLookups, ok) + return entry, ok +} + +// storeExistence caches that a blob of size is present. Best-effort. +func (s *commonService) storeExistence(ctx context.Context, key string, size int64) { + if key == "" { + return + } + + if err := s.existence.Set(ctx, key, casexistence.Entry{Size: size}); err != nil { + s.log.Warnw("msg", "existence cache store failed", "error", err) + } +} + +// clientKey identifies a backend client. It includes the organization because +// some backends (e.g. AWS-S3-ACCESS-POINT) keep per-organization session +// credentials inside the client. +func clientKey(info *casJWT.Claims) string { + return strings.Join([]string{info.BackendType, info.OrgID, info.StoredSecretID}, "\x00") +} + +// loadUploadBackend returns the backend of the token, reusing a cached client +// when the client cache is enabled. cached reports if the client was reused. +func (s *commonService) loadUploadBackend(ctx context.Context, info *casJWT.Claims) (b backend.UploaderDownloader, cached bool, err error) { + if s.clients == nil { + b, err = s.loadBackend(ctx, info.BackendType, info.StoredSecretID) + return b, false, err + } + + key := clientKey(info) + if b, ok := s.clients.Get(key); ok { + recordLookup(clientCacheLookups, true) + return b, true, nil + } + recordLookup(clientCacheLookups, false) + + b, err = s.loadBackend(ctx, info.BackendType, info.StoredSecretID) + if err != nil { + return nil, false, err + } + + s.clients.Add(key, b) + return b, false, nil +} + +// dropUploadBackend removes the cached client of the token, so the next upload +// loads the credentials again. +func (s *commonService) dropUploadBackend(info *casJWT.Claims) { + if s.clients != nil { + s.clients.Remove(clientKey(info)) + } +} diff --git a/app/artifact-cas/internal/service/service.go b/app/artifact-cas/internal/service/service.go index 8aff6e466..87e69f0e8 100644 --- a/app/artifact-cas/internal/service/service.go +++ b/app/artifact-cas/internal/service/service.go @@ -22,10 +22,13 @@ import ( "syscall" backend "github.com/chainloop-dev/chainloop/pkg/blobmanager" + "github.com/chainloop-dev/chainloop/pkg/cache" + "github.com/chainloop-dev/chainloop/pkg/cache/casexistence" "github.com/chainloop-dev/chainloop/pkg/servicelogger" kerrors "github.com/go-kratos/kratos/v2/errors" "github.com/go-kratos/kratos/v2/log" "github.com/google/wire" + "github.com/hashicorp/golang-lru/v2/expirable" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) @@ -44,6 +47,10 @@ type commonService struct { // set and writable: leaving it empty is a deployment error and transfers are // refused rather than staged somewhere unintended. stagingDir string + // existence caches the blobs known to exist in a backend, nil when disabled + existence cache.Cache[casexistence.Entry] + // clients caches the backend clients loaded for uploads, nil when disabled + clients *expirable.LRU[string, backend.UploaderDownloader] } func (s *commonService) loadBackend(ctx context.Context, providerType, secretID string) (backend.UploaderDownloader, error) { diff --git a/deployment/chainloop/Chart.yaml b/deployment/chainloop/Chart.yaml index ed2dfc767..535abbc63 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.459.0 +version: 1.459.1 # Do not update appVersion, this is handled automatically by the release process appVersion: v1.118.1 diff --git a/deployment/chainloop/README.md b/deployment/chainloop/README.md index 25eb3392b..d866737b3 100644 --- a/deployment/chainloop/README.md +++ b/deployment/chainloop/README.md @@ -794,6 +794,9 @@ Once done, you can access with [two predefined users](https://github.com/chainlo | `cas.nats.host` | NATS Host | `""` | | `cas.nats.port` | NATS Port | `4222` | | `cas.nats.token` | NATS Client authentication token | `""` | +| `cas.existenceCache` | Cache of the blobs known to exist in a backend, so repeated uploads of the same content skip the backend. Shared between replicas through NATS when cas.nats is enabled, in memory otherwise. | | +| `cas.existenceCache.enabled` | Enable the existence cache | `true` | +| `cas.existenceCache.ttl` | Lifetime of a cache entry, in seconds with an "s" suffix | `86400s` | | `cas.image.registry` | Image registry | `REGISTRY_NAME` | | `cas.image.repository` | Image repository | `REPOSITORY_NAME` | | `cas.containerPorts.http` | controlplane HTTP container port | `8000` | @@ -958,7 +961,7 @@ service_registration "kubernetes" {}` | ## License -Copyright © 2023-2025 The Chainloop Authors +Copyright © 2023-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. diff --git a/deployment/chainloop/templates/cas/configmap.yaml b/deployment/chainloop/templates/cas/configmap.yaml index fd0aa6e63..d3d977229 100644 --- a/deployment/chainloop/templates/cas/configmap.yaml +++ b/deployment/chainloop/templates/cas/configmap.yaml @@ -17,6 +17,9 @@ data: # Local directory where uploads and downloads are staged and verified against # their digest before being forwarded. Backed by the staging emptyDir. staging_dir: {{ include "chainloop.cas.staging_dir" . | quote }} + existence_cache: + disabled: {{ not .Values.cas.existenceCache.enabled }} + ttl: {{ .Values.cas.existenceCache.ttl | quote }} server: http: addr: "0.0.0.0:{{ .Values.cas.containerPorts.http }}" diff --git a/deployment/chainloop/values.yaml b/deployment/chainloop/values.yaml index 63b2cea87..a649489c2 100644 --- a/deployment/chainloop/values.yaml +++ b/deployment/chainloop/values.yaml @@ -1042,6 +1042,13 @@ cas: port: 4222 token: "" + ## @extra cas.existenceCache Cache of the blobs known to exist in a backend, so repeated uploads of the same content skip the backend. Shared between replicas through NATS when cas.nats is enabled, in memory otherwise. + ## @param cas.existenceCache.enabled Enable the existence cache + ## @param cas.existenceCache.ttl Lifetime of a cache entry, in seconds with an "s" suffix + existenceCache: + enabled: true + ttl: 86400s + ## @param cas.image.registry [default: REGISTRY_NAME] Image registry ## @param cas.image.repository [default: REPOSITORY_NAME] Image repository ## @skip cas.image.tag diff --git a/pkg/cache/cache.go b/pkg/cache/cache.go index c30634c51..e302e6bd5 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 + maxEntries int replicas int logger Logger natsConn *nats.Conn @@ -78,6 +79,12 @@ func WithMaxBytes(n int64) Option { return func(c *config) { c.maxBytes = n } } +// WithMaxEntries sets the maximum number of entries of the in-memory backend. +// Defaults to 1000. Ignored for NATS KV backend, which is bounded by WithMaxBytes. +func WithMaxEntries(n int) Option { + return func(c *config) { c.maxEntries = 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 } diff --git a/pkg/cache/casexistence/casexistence.go b/pkg/cache/casexistence/casexistence.go new file mode 100644 index 000000000..3649927ab --- /dev/null +++ b/pkg/cache/casexistence/casexistence.go @@ -0,0 +1,89 @@ +// +// Copyright 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. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package casexistence provides a typed cache of the blobs known to exist in +// a CAS backend, so repeated uploads of the same content skip the backend. +package casexistence + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "strings" + "time" + + "github.com/chainloop-dev/chainloop/pkg/cache" + "github.com/chainloop-dev/chainloop/pkg/natsconn" + "github.com/go-kratos/kratos/v2/log" +) + +const ( + // DefaultTTL is the lifetime of an entry when none is configured. + DefaultTTL = 24 * time.Hour + + maxBytes = 64 * 1024 * 1024 // 64 MB + maxEntries = 50_000 + bucket = "chainloop-cas-existence" + description = "Cache of blobs known to exist in a CAS backend" +) + +// Entry is a blob known to exist in a backend. +type Entry struct { + Size int64 `json:"size"` +} + +// Cache wraps cache.Cache[Entry] to provide a distinct type for wire disambiguation. +type Cache struct { + cache.Cache[Entry] +} + +// New creates an existence cache, backed by NATS KV when rc is set and by an +// in-memory LRU otherwise. A ttl of 0 uses DefaultTTL. +func New(ctx context.Context, rc *natsconn.ReloadableConnection, ttl time.Duration, logger log.Logger) (*Cache, error) { + if ttl <= 0 { + ttl = DefaultTTL + } + + opts := []cache.Option{ + cache.WithTTL(ttl), + cache.WithMaxBytes(maxBytes), + cache.WithMaxEntries(maxEntries), + cache.WithDescription(description), + } + + if logger != nil { + opts = append(opts, cache.WithLogger(log.NewHelper(logger))) + } + + 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[Entry](opts...) + if err != nil { + return nil, err + } + return &Cache{Cache: c}, nil +} + +// Key returns the cache key of a digest in the namespace of one organization +// and one backend. The secret reference is hashed so its path never appears in +// the shared store. +func Key(orgID, backendType, secretID, digest string) string { + secretHash := sha256.Sum256([]byte(secretID)) + return strings.Join([]string{orgID, backendType, hex.EncodeToString(secretHash[:]), digest}, "/") +} diff --git a/pkg/cache/memory.go b/pkg/cache/memory.go index ba907a6cc..ab54bf786 100644 --- a/pkg/cache/memory.go +++ b/pkg/cache/memory.go @@ -31,9 +31,14 @@ type memoryCache[T any] struct { } func newMemoryCache[T any](cfg *config) *memoryCache[T] { - cfg.logger.Infow("msg", "cache: using in-memory LRU backend", "ttl", cfg.ttl, "maxSize", defaultMaxSize) + maxSize := defaultMaxSize + if cfg.maxEntries > 0 { + maxSize = cfg.maxEntries + } + + cfg.logger.Infow("msg", "cache: using in-memory LRU backend", "ttl", cfg.ttl, "maxSize", maxSize) return &memoryCache[T]{ - lru: expirable.NewLRU[string, T](defaultMaxSize, nil, cfg.ttl), + lru: expirable.NewLRU[string, T](maxSize, nil, cfg.ttl), logger: cfg.logger, } } diff --git a/pkg/cache/memory_test.go b/pkg/cache/memory_test.go index 80abf6e74..261279cf3 100644 --- a/pkg/cache/memory_test.go +++ b/pkg/cache/memory_test.go @@ -17,6 +17,7 @@ package cache import ( "context" + "strconv" "testing" "time" @@ -105,3 +106,25 @@ func TestNew_RequiresTTL(t *testing.T) { _, err := New[string]() require.Error(t, err) } + +func TestNew_MaxEntries(t *testing.T) { + tests := []struct { + name string + opts []Option + wantStored int + }{ + {name: "default", wantStored: defaultMaxSize}, + {name: "custom", opts: []Option{WithMaxEntries(2)}, wantStored: 2}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + c, err := New[int](append(tt.opts, WithTTL(time.Minute))...) + require.NoError(t, err) + for i := range defaultMaxSize + 1 { + require.NoError(t, c.Set(context.Background(), strconv.Itoa(i), i)) + } + assert.Equal(t, tt.wantStored, c.(*memoryCache[int]).lru.Len()) + }) + } +} From cf4d05ff2c8ca287ce9535b49814bbba8e1242d5 Mon Sep 17 00:00:00 2001 From: Miguel Martinez Trivino Date: Thu, 8 Oct 2026 13:07:40 +0200 Subject: [PATCH 2/4] refactor(cas): simplify existence and client cache wiring Share one backend client cache between services, collapse the client reload retry into a loop, move the NATS to memory fallback into casexistence.New, and simplify the cache tests. Assisted-by: Claude Code Signed-off-by: Miguel Martinez Trivino Chainloop-Trace-Sessions: 9b6de27d-6322-4cf7-8e72-5e56d858c783 --- app/artifact-cas/cmd/wire.go | 10 +--- app/artifact-cas/cmd/wire_gen.go | 10 +--- .../internal/service/bytestream.go | 37 +++++------- .../internal/service/bytestream_cache_test.go | 60 ++++++++----------- app/artifact-cas/internal/service/cache.go | 4 +- pkg/cache/casexistence/casexistence.go | 16 ++++- 6 files changed, 59 insertions(+), 78 deletions(-) diff --git a/app/artifact-cas/cmd/wire.go b/app/artifact-cas/cmd/wire.go index f65516bb4..88e055f6b 100644 --- a/app/artifact-cas/cmd/wire.go +++ b/app/artifact-cas/cmd/wire.go @@ -76,22 +76,14 @@ func serviceOpts(l log.Logger, audit *service.AuditDispatcher, bc *conf.Bootstra // newExistenceCache returns the cache of blobs known to exist in a backend, // nil when disabled. It is shared through NATS KV when the connection is set. -// A NATS bucket that can not be set up falls back to memory: the cache only -// makes uploads faster, so it must not stop the service from starting. func newExistenceCache(bc *conf.Bootstrap, rc *natsconn.ReloadableConnection, logger log.Logger) (*casexistence.Cache, func(), error) { cfg := bc.GetExistenceCache() if cfg.GetDisabled() { return nil, func() {}, nil } - ttl := cfg.GetTtl().AsDuration() ctx, cancel := context.WithCancel(context.Background()) - c, err := casexistence.New(ctx, rc, ttl, logger) - if err != nil && rc != nil { - log.NewHelper(logger).Warnw("msg", "existence cache: NATS KV unavailable, using memory", "error", err) - c, err = casexistence.New(ctx, nil, ttl, logger) - } - + c, err := casexistence.New(ctx, rc, cfg.GetTtl().AsDuration(), logger) if err != nil { cancel() return nil, nil, err diff --git a/app/artifact-cas/cmd/wire_gen.go b/app/artifact-cas/cmd/wire_gen.go index caa7f698a..396d7a315 100644 --- a/app/artifact-cas/cmd/wire_gen.go +++ b/app/artifact-cas/cmd/wire_gen.go @@ -104,22 +104,14 @@ func serviceOpts(l log.Logger, audit *service.AuditDispatcher, bc *conf.Bootstra // newExistenceCache returns the cache of blobs known to exist in a backend, // nil when disabled. It is shared through NATS KV when the connection is set. -// A NATS bucket that can not be set up falls back to memory: the cache only -// makes uploads faster, so it must not stop the service from starting. func newExistenceCache(bc *conf.Bootstrap, rc *natsconn.ReloadableConnection, logger log.Logger) (*casexistence.Cache, func(), error) { cfg := bc.GetExistenceCache() if cfg.GetDisabled() { return nil, func() {}, nil } - ttl := cfg.GetTtl().AsDuration() ctx, cancel := context.WithCancel(context.Background()) - c, err := casexistence.New(ctx, rc, ttl, logger) - if err != nil && rc != nil { - log.NewHelper(logger).Warnw("msg", "existence cache: NATS KV unavailable, using memory", "error", err) - c, err = casexistence.New(ctx, nil, ttl, logger) - } - + c, err := casexistence.New(ctx, rc, cfg.GetTtl().AsDuration(), logger) if err != nil { cancel() return nil, nil, err diff --git a/app/artifact-cas/internal/service/bytestream.go b/app/artifact-cas/internal/service/bytestream.go index fa59e5e0d..f2d40576d 100644 --- a/app/artifact-cas/internal/service/bytestream.go +++ b/app/artifact-cas/internal/service/bytestream.go @@ -169,34 +169,25 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro // its credentials may have been rotated. Returned errors are ready to be sent // to the client. func (s *ByteStreamService) backendExists(ctx context.Context, info *casJWT.Claims, digest string) (backend.UploaderDownloader, bool, error) { - storageBackend, cached, err := s.loadUploadBackend(ctx, info) - if err != nil { - return nil, false, s.loadBackendError(err) - } + for attempt := 0; ; attempt++ { + storageBackend, cached, err := s.loadUploadBackend(ctx, info) + if err != nil && kerrors.IsNotFound(err) { + return nil, false, err + } else if err != nil { + return nil, false, sl.LogAndMaskErr(err, s.log) + } - exists, err := storageBackend.Exists(ctx, digest) - if err != nil && cached { - s.log.Warnw("msg", "reused backend client failed, loading it again", "digest", digest, "error", err) - s.dropUploadBackend(info) - if storageBackend, _, err = s.loadUploadBackend(ctx, info); err != nil { - return nil, false, s.loadBackendError(err) + exists, err := storageBackend.Exists(ctx, digest) + if err == nil { + return storageBackend, exists, nil } - exists, err = storageBackend.Exists(ctx, digest) - } - if err != nil { s.dropUploadBackend(info) - return nil, false, sl.LogAndMaskErr(err, s.log) - } - - return storageBackend, exists, nil -} - -func (s *ByteStreamService) loadBackendError(err error) error { - if kerrors.IsNotFound(err) { - return err + if !cached || attempt > 0 { + return nil, false, sl.LogAndMaskErr(err, s.log) + } + s.log.Warnw("msg", "reused backend client failed, loading it again", "digest", digest, "error", err) } - return sl.LogAndMaskErr(err, s.log) } // dispatchSkippedUpload emits the audit event of an upload skipped because the diff --git a/app/artifact-cas/internal/service/bytestream_cache_test.go b/app/artifact-cas/internal/service/bytestream_cache_test.go index 964affef6..075d4d21e 100644 --- a/app/artifact-cas/internal/service/bytestream_cache_test.go +++ b/app/artifact-cas/internal/service/bytestream_cache_test.go @@ -65,11 +65,6 @@ func (s *bytestreamSuite) withExistenceCache() { s.restart(WithExistenceCache(c)) } -// withClientCache restarts the server with the backend client cache only -func (s *bytestreamSuite) withClientCache() { - s.restart(WithBackendClientCache(time.Hour)) -} - // R-001, R-006: a second upload of a blob that the backend reported as present // makes no secrets manager or backend call, and its event keeps the size. func (s *bytestreamSuite) TestExistenceCacheHitSkipsBackend() { @@ -127,9 +122,9 @@ func (s *bytestreamSuite) TestExistenceCacheNotCached() { resource := resourceWithDigest(data, "skynet.exe") testCases := []struct { - name string - setup func() - code codes.Code + name string + setup func() + wantErr bool }{ { name: "not found and failed upload", @@ -137,7 +132,7 @@ func (s *bytestreamSuite) TestExistenceCacheNotCached() { s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) s.ociBackend.On("Upload", mock.Anything, mock.Anything, resource).Return(errors.New("boom")) }, - code: codes.Internal, + wantErr: true, }, { name: "found but describe fails", @@ -145,7 +140,6 @@ func (s *bytestreamSuite) TestExistenceCacheNotCached() { s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(true, nil) s.ociBackend.On("Describe", mock.Anything, resource.Digest).Return(nil, errors.New("boom")) }, - code: codes.OK, }, } @@ -155,10 +149,10 @@ func (s *bytestreamSuite) TestExistenceCacheNotCached() { tc.setup() for range 2 { _, err := s.write(s.upCtx, resource, data) - if tc.code == codes.OK { - s.Require().NoError(err) + if tc.wantErr { + assertGRPCError(s.T(), err, codes.Internal, "") } else { - assertGRPCError(s.T(), err, tc.code, "") + s.Require().NoError(err) } } s.ociBackend.AssertNumberOfCalls(s.T(), "Exists", 2) @@ -171,23 +165,24 @@ func (s *bytestreamSuite) TestExistenceCacheTenantNamespace() { testCases := []struct { name string secondCtx context.Context - // backend the second upload must reach - second func() *mockedBackend + // backend checks the two uploads must make + wantOCIExists, wantStreamingExists int }{ { - name: "different organization on shared storage", - secondCtx: uploaderCtx("org-id", otherOrgID), - second: func() *mockedBackend { return &mockedBackend{s.ociBackend, 2} }, + name: "different organization on shared storage", + secondCtx: uploaderCtx("org-id", otherOrgID), + wantOCIExists: 2, }, { - name: "same organization, different backend", - secondCtx: uploaderCtx("backend-streaming", "true"), - second: func() *mockedBackend { return &mockedBackend{s.streamingBackend, 1} }, + name: "same organization, different backend", + secondCtx: uploaderCtx("backend-streaming", "true"), + wantOCIExists: 1, + wantStreamingExists: 1, }, { - name: "organization missing from the token", - secondCtx: uploaderCtx("org-id", ""), - second: func() *mockedBackend { return &mockedBackend{s.ociBackend, 2} }, + name: "organization missing from the token", + secondCtx: uploaderCtx("org-id", ""), + wantOCIExists: 2, }, } @@ -195,7 +190,7 @@ func (s *bytestreamSuite) TestExistenceCacheTenantNamespace() { s.Run(tc.name, func() { s.withExistenceCache() describe := &v1.CASResource{Digest: s.resource.Digest, Size: 1} - for _, b := range []*mockedBackend{{s.ociBackend, 0}, {s.streamingBackend, 0}} { + for _, b := range []*mocks.UploaderDownloader{s.ociBackend, s.streamingBackend} { b.On("Exists", mock.Anything, s.resource.Digest).Maybe().Return(true, nil) b.On("Describe", mock.Anything, s.resource.Digest).Maybe().Return(describe, nil) } @@ -205,8 +200,8 @@ func (s *bytestreamSuite) TestExistenceCacheTenantNamespace() { _, err = s.write(tc.secondCtx, s.resource, nil) s.Require().NoError(err) - want := tc.second() - want.AssertNumberOfCalls(s.T(), "Exists", want.calls) + s.ociBackend.AssertNumberOfCalls(s.T(), "Exists", tc.wantOCIExists) + s.streamingBackend.AssertNumberOfCalls(s.T(), "Exists", tc.wantStreamingExists) }) } } @@ -230,7 +225,7 @@ func (s *bytestreamSuite) TestExistenceCacheFailureIsMiss() { // R-007: a cache miss on the existence check reuses the loaded backend client. func (s *bytestreamSuite) TestClientCacheReusesClient() { - s.withClientCache() + s.restart(WithBackendClientCache(time.Hour)) s.ociBackend.On("Exists", mock.Anything, mock.Anything).Return(true, nil) for _, d := range []string{"digest-a", "digest-b"} { @@ -247,7 +242,7 @@ func (s *bytestreamSuite) TestClientCacheReusesClient() { // D-009: an error from a reused client drops it and loads it again one time. func (s *bytestreamSuite) TestClientCacheReloadOnError() { - s.withClientCache() + s.restart(WithBackendClientCache(time.Hour)) ctx := uploaderCtx("source-internal", "true") s.ociBackend.On("Exists", mock.Anything, mock.Anything).Return(true, nil).Once() _, err := s.write(ctx, s.resource, nil) @@ -262,7 +257,7 @@ func (s *bytestreamSuite) TestClientCacheReloadOnError() { // a fresh client that fails is not loaded again func (s *bytestreamSuite) TestClientCacheNoReloadOfFreshClient() { - s.withClientCache() + s.restart(WithBackendClientCache(time.Hour)) s.ociBackend.On("Exists", mock.Anything, mock.Anything).Return(false, errors.New("boom")) _, err := s.write(s.upCtx, s.resource, nil) @@ -271,11 +266,6 @@ func (s *bytestreamSuite) TestClientCacheNoReloadOfFreshClient() { s.ociBackend.AssertNumberOfCalls(s.T(), "Exists", 1) } -type mockedBackend struct { - *mocks.UploaderDownloader - calls int -} - type failingCache struct{} var errCacheDown = errors.New("cache is down") diff --git a/app/artifact-cas/internal/service/cache.go b/app/artifact-cas/internal/service/cache.go index ad1ff0780..2ae16dcce 100644 --- a/app/artifact-cas/internal/service/cache.go +++ b/app/artifact-cas/internal/service/cache.go @@ -62,9 +62,11 @@ func WithExistenceCache(c cache.Cache[casexistence.Entry]) NewOpt { // WithBackendClientCache enables the reuse of the backend clients loaded for // uploads for ttl. Clients and their credentials are kept in process memory only. +// The services built with the same option share one cache. func WithBackendClientCache(ttl time.Duration) NewOpt { + clients := expirable.NewLRU[string, backend.UploaderDownloader](maxCachedClients, nil, ttl) return func(s *commonService) { - s.clients = expirable.NewLRU[string, backend.UploaderDownloader](maxCachedClients, nil, ttl) + s.clients = clients } } diff --git a/pkg/cache/casexistence/casexistence.go b/pkg/cache/casexistence/casexistence.go index 3649927ab..b704849b0 100644 --- a/pkg/cache/casexistence/casexistence.go +++ b/pkg/cache/casexistence/casexistence.go @@ -50,12 +50,26 @@ type Cache struct { } // New creates an existence cache, backed by NATS KV when rc is set and by an -// in-memory LRU otherwise. A ttl of 0 uses DefaultTTL. +// in-memory LRU otherwise. A ttl of 0 uses DefaultTTL. A NATS bucket that can +// not be set up falls back to memory: the cache only makes uploads faster, so +// it must not stop its caller from starting. func New(ctx context.Context, rc *natsconn.ReloadableConnection, ttl time.Duration, logger log.Logger) (*Cache, error) { if ttl <= 0 { ttl = DefaultTTL } + c, err := newCache(ctx, rc, ttl, logger) + if err != nil && rc != nil { + if logger != nil { + log.NewHelper(logger).Warnw("msg", "existence cache: NATS KV unavailable, using memory", "error", err) + } + c, err = newCache(ctx, nil, ttl, logger) + } + + return c, err +} + +func newCache(ctx context.Context, rc *natsconn.ReloadableConnection, ttl time.Duration, logger log.Logger) (*Cache, error) { opts := []cache.Option{ cache.WithTTL(ttl), cache.WithMaxBytes(maxBytes), From 6df0209f296673b8e4ce25995cea815422969ea5 Mon Sep 17 00:00:00 2001 From: Miguel Martinez Trivino Date: Thu, 8 Oct 2026 13:22:07 +0200 Subject: [PATCH 3/4] refactor(cas): always enable the existence cache with no configuration Remove the existence cache settings from the CAS config and the Helm chart. The cache is always on with a 24h lifetime, in NATS KV when available and in memory otherwise, also when the NATS bucket cannot be set up at startup. Record the build-time decisions in spec issue-3543 (R-010, D-010 to D-013). Assisted-by: Claude Code Signed-off-by: Miguel Martinez Trivino Chainloop-Trace-Sessions: 9b6de27d-6322-4cf7-8e72-5e56d858c783 --- app/artifact-cas/cmd/main_test.go | 26 --- app/artifact-cas/cmd/wire.go | 22 +-- app/artifact-cas/cmd/wire_gen.go | 23 +-- app/artifact-cas/internal/conf/conf.pb.go | 184 ++++++------------ app/artifact-cas/internal/conf/conf.proto | 11 -- deployment/chainloop/Chart.yaml | 2 +- deployment/chainloop/README.md | 5 +- .../chainloop/templates/cas/configmap.yaml | 3 - deployment/chainloop/values.yaml | 7 - docs/specs/issue-3543-cas-existence-cache.md | 17 +- pkg/cache/cache.go | 3 +- pkg/cache/casexistence/casexistence.go | 23 +-- pkg/cache/casexistence/casexistence_test.go | 84 ++++++++ 13 files changed, 177 insertions(+), 233 deletions(-) create mode 100644 pkg/cache/casexistence/casexistence_test.go diff --git a/app/artifact-cas/cmd/main_test.go b/app/artifact-cas/cmd/main_test.go index b2c8d84fb..c97004007 100644 --- a/app/artifact-cas/cmd/main_test.go +++ b/app/artifact-cas/cmd/main_test.go @@ -19,12 +19,9 @@ import ( "os" "path/filepath" "testing" - "time" "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/conf" - "github.com/go-kratos/kratos/v2/log" "github.com/stretchr/testify/require" - "google.golang.org/protobuf/types/known/durationpb" ) // TestPrepareStagingDir: the staging directory is a hard startup requirement. @@ -89,26 +86,3 @@ func TestPrepareStagingDir(t *testing.T) { }) } } - -// TestNewExistenceCache: the existence cache is on by default and an operator -// can turn it off. -func TestNewExistenceCache(t *testing.T) { - testCases := []struct { - name string - cfg *conf.Bootstrap_ExistenceCache - wantNil bool - }{ - {name: "enabled by default"}, - {name: "custom ttl", cfg: &conf.Bootstrap_ExistenceCache{Ttl: durationpb.New(time.Hour)}}, - {name: "disabled", cfg: &conf.Bootstrap_ExistenceCache{Disabled: true}, wantNil: true}, - } - - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - c, cleanup, err := newExistenceCache(&conf.Bootstrap{ExistenceCache: tc.cfg}, nil, log.DefaultLogger) - require.NoError(t, err) - t.Cleanup(cleanup) - require.Equal(t, tc.wantNil, c == nil) - }) - } -} diff --git a/app/artifact-cas/cmd/wire.go b/app/artifact-cas/cmd/wire.go index 88e055f6b..409723ca0 100644 --- a/app/artifact-cas/cmd/wire.go +++ b/app/artifact-cas/cmd/wire.go @@ -60,30 +60,20 @@ func wireApp(*conf.Bootstrap, *conf.Server, *conf.Auth, credentials.Reader, log. const backendClientTTL = 5 * time.Minute func serviceOpts(l log.Logger, audit *service.AuditDispatcher, bc *conf.Bootstrap, existence *casexistence.Cache) []service.NewOpt { - opts := []service.NewOpt{ + return []service.NewOpt{ service.WithLogger(l), service.WithAuditDispatcher(audit), service.WithStagingDir(bc.GetStagingDir()), service.WithBackendClientCache(backendClientTTL), + service.WithExistenceCache(existence), } - - if existence != nil { - opts = append(opts, service.WithExistenceCache(existence)) - } - - return opts } -// newExistenceCache returns the cache of blobs known to exist in a backend, -// nil when disabled. It is shared through NATS KV when the connection is set. -func newExistenceCache(bc *conf.Bootstrap, rc *natsconn.ReloadableConnection, logger log.Logger) (*casexistence.Cache, func(), error) { - cfg := bc.GetExistenceCache() - if cfg.GetDisabled() { - return nil, func() {}, nil - } - +// newExistenceCache returns the cache of blobs known to exist in a backend. +// It is shared through NATS KV when the connection is set, in memory otherwise. +func newExistenceCache(rc *natsconn.ReloadableConnection, logger log.Logger) (*casexistence.Cache, func(), error) { ctx, cancel := context.WithCancel(context.Background()) - c, err := casexistence.New(ctx, rc, cfg.GetTtl().AsDuration(), logger) + c, err := casexistence.New(ctx, rc, logger) if err != nil { cancel() return nil, nil, err diff --git a/app/artifact-cas/cmd/wire_gen.go b/app/artifact-cas/cmd/wire_gen.go index 396d7a315..aa28bcf9a 100644 --- a/app/artifact-cas/cmd/wire_gen.go +++ b/app/artifact-cas/cmd/wire_gen.go @@ -40,7 +40,7 @@ func wireApp(bootstrap *conf.Bootstrap, confServer *conf.Server, auth *conf.Auth return nil, nil, err } auditDispatcher := service.NewAuditDispatcher(auditLogPublisher, logger) - cache, cleanup2, err := newExistenceCache(bootstrap, reloadableConnection, logger) + cache, cleanup2, err := newExistenceCache(reloadableConnection, logger) if err != nil { cleanup() return nil, nil, err @@ -93,25 +93,14 @@ func wireApp(bootstrap *conf.Bootstrap, confServer *conf.Server, auth *conf.Auth const backendClientTTL = 5 * time.Minute func serviceOpts(l log.Logger, audit *service.AuditDispatcher, bc *conf.Bootstrap, existence *casexistence.Cache) []service.NewOpt { - opts := []service.NewOpt{service.WithLogger(l), service.WithAuditDispatcher(audit), service.WithStagingDir(bc.GetStagingDir()), service.WithBackendClientCache(backendClientTTL)} - - if existence != nil { - opts = append(opts, service.WithExistenceCache(existence)) - } - - return opts + return []service.NewOpt{service.WithLogger(l), service.WithAuditDispatcher(audit), service.WithStagingDir(bc.GetStagingDir()), service.WithBackendClientCache(backendClientTTL), service.WithExistenceCache(existence)} } -// newExistenceCache returns the cache of blobs known to exist in a backend, -// nil when disabled. It is shared through NATS KV when the connection is set. -func newExistenceCache(bc *conf.Bootstrap, rc *natsconn.ReloadableConnection, logger log.Logger) (*casexistence.Cache, func(), error) { - cfg := bc.GetExistenceCache() - if cfg.GetDisabled() { - return nil, func() {}, nil - } - +// newExistenceCache returns the cache of blobs known to exist in a backend. +// It is shared through NATS KV when the connection is set, in memory otherwise. +func newExistenceCache(rc *natsconn.ReloadableConnection, logger log.Logger) (*casexistence.Cache, func(), error) { ctx, cancel := context.WithCancel(context.Background()) - c, err := casexistence.New(ctx, rc, cfg.GetTtl().AsDuration(), logger) + c, err := casexistence.New(ctx, rc, logger) if err != nil { cancel() return nil, nil, err diff --git a/app/artifact-cas/internal/conf/conf.pb.go b/app/artifact-cas/internal/conf/conf.pb.go index 3b14a245d..ac72ca7ae 100644 --- a/app/artifact-cas/internal/conf/conf.pb.go +++ b/app/artifact-cas/internal/conf/conf.pb.go @@ -52,13 +52,9 @@ type Bootstrap struct { // the first byte is sent to the client. It must be a writable volume; in // production a dedicated emptyDir is mounted here (NOT tmpfs/RAM, and NOT the // /tmp secret mount). Required: when unset the service refuses to start. - StagingDir string `protobuf:"bytes,6,opt,name=staging_dir,json=stagingDir,proto3" json:"staging_dir,omitempty"` - // Cache of the blobs known to exist in a backend, so repeated uploads of the - // same content skip the secrets manager and the backend. Shared through NATS - // KV when nats_server is set, in memory otherwise. Enabled by default. - ExistenceCache *Bootstrap_ExistenceCache `protobuf:"bytes,7,opt,name=existence_cache,json=existenceCache,proto3" json:"existence_cache,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + StagingDir string `protobuf:"bytes,6,opt,name=staging_dir,json=stagingDir,proto3" json:"staging_dir,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *Bootstrap) Reset() { @@ -133,13 +129,6 @@ func (x *Bootstrap) GetStagingDir() string { return "" } -func (x *Bootstrap) GetExistenceCache() *Bootstrap_ExistenceCache { - if x != nil { - return x.ExistenceCache - } - return nil -} - type Server struct { state protoimpl.MessageState `protogen:"open.v1"` // Regular HTTP endpoint @@ -261,60 +250,6 @@ func (x *Auth) GetPublicKeyPath() string { return "" } -type Bootstrap_ExistenceCache struct { - state protoimpl.MessageState `protogen:"open.v1"` - // Turns the cache off - Disabled bool `protobuf:"varint,1,opt,name=disabled,proto3" json:"disabled,omitempty"` - // Lifetime of an entry. Defaults to 24h. - Ttl *durationpb.Duration `protobuf:"bytes,2,opt,name=ttl,proto3" json:"ttl,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache -} - -func (x *Bootstrap_ExistenceCache) Reset() { - *x = Bootstrap_ExistenceCache{} - mi := &file_conf_proto_msgTypes[3] - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - ms.StoreMessageInfo(mi) -} - -func (x *Bootstrap_ExistenceCache) String() string { - return protoimpl.X.MessageStringOf(x) -} - -func (*Bootstrap_ExistenceCache) ProtoMessage() {} - -func (x *Bootstrap_ExistenceCache) ProtoReflect() protoreflect.Message { - mi := &file_conf_proto_msgTypes[3] - if x != nil { - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - if ms.LoadMessageInfo() == nil { - ms.StoreMessageInfo(mi) - } - return ms - } - return mi.MessageOf(x) -} - -// Deprecated: Use Bootstrap_ExistenceCache.ProtoReflect.Descriptor instead. -func (*Bootstrap_ExistenceCache) Descriptor() ([]byte, []int) { - return file_conf_proto_rawDescGZIP(), []int{0, 0} -} - -func (x *Bootstrap_ExistenceCache) GetDisabled() bool { - if x != nil { - return x.Disabled - } - return false -} - -func (x *Bootstrap_ExistenceCache) GetTtl() *durationpb.Duration { - if x != nil { - return x.Ttl - } - return nil -} - type Bootstrap_NatsServer struct { state protoimpl.MessageState `protogen:"open.v1"` // NATS server URI, e.g. "nats://localhost:4222" @@ -329,7 +264,7 @@ type Bootstrap_NatsServer struct { func (x *Bootstrap_NatsServer) Reset() { *x = Bootstrap_NatsServer{} - mi := &file_conf_proto_msgTypes[4] + mi := &file_conf_proto_msgTypes[3] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -341,7 +276,7 @@ func (x *Bootstrap_NatsServer) String() string { func (*Bootstrap_NatsServer) ProtoMessage() {} func (x *Bootstrap_NatsServer) ProtoReflect() protoreflect.Message { - mi := &file_conf_proto_msgTypes[4] + mi := &file_conf_proto_msgTypes[3] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -354,7 +289,7 @@ func (x *Bootstrap_NatsServer) ProtoReflect() protoreflect.Message { // Deprecated: Use Bootstrap_NatsServer.ProtoReflect.Descriptor instead. func (*Bootstrap_NatsServer) Descriptor() ([]byte, []int) { - return file_conf_proto_rawDescGZIP(), []int{0, 1} + return file_conf_proto_rawDescGZIP(), []int{0, 0} } func (x *Bootstrap_NatsServer) GetUri() string { @@ -400,7 +335,7 @@ type Bootstrap_Observability struct { func (x *Bootstrap_Observability) Reset() { *x = Bootstrap_Observability{} - mi := &file_conf_proto_msgTypes[5] + mi := &file_conf_proto_msgTypes[4] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -412,7 +347,7 @@ func (x *Bootstrap_Observability) String() string { func (*Bootstrap_Observability) ProtoMessage() {} func (x *Bootstrap_Observability) ProtoReflect() protoreflect.Message { - mi := &file_conf_proto_msgTypes[5] + mi := &file_conf_proto_msgTypes[4] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -425,7 +360,7 @@ func (x *Bootstrap_Observability) ProtoReflect() protoreflect.Message { // Deprecated: Use Bootstrap_Observability.ProtoReflect.Descriptor instead. func (*Bootstrap_Observability) Descriptor() ([]byte, []int) { - return file_conf_proto_rawDescGZIP(), []int{0, 2} + return file_conf_proto_rawDescGZIP(), []int{0, 1} } func (x *Bootstrap_Observability) GetSentry() *Bootstrap_Observability_Sentry { @@ -453,7 +388,7 @@ type Bootstrap_Observability_Sentry struct { func (x *Bootstrap_Observability_Sentry) Reset() { *x = Bootstrap_Observability_Sentry{} - mi := &file_conf_proto_msgTypes[6] + mi := &file_conf_proto_msgTypes[5] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -465,7 +400,7 @@ func (x *Bootstrap_Observability_Sentry) String() string { func (*Bootstrap_Observability_Sentry) ProtoMessage() {} func (x *Bootstrap_Observability_Sentry) ProtoReflect() protoreflect.Message { - mi := &file_conf_proto_msgTypes[6] + mi := &file_conf_proto_msgTypes[5] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -478,7 +413,7 @@ func (x *Bootstrap_Observability_Sentry) ProtoReflect() protoreflect.Message { // Deprecated: Use Bootstrap_Observability_Sentry.ProtoReflect.Descriptor instead. func (*Bootstrap_Observability_Sentry) Descriptor() ([]byte, []int) { - return file_conf_proto_rawDescGZIP(), []int{0, 2, 0} + return file_conf_proto_rawDescGZIP(), []int{0, 1, 0} } func (x *Bootstrap_Observability_Sentry) GetDsn() string { @@ -512,7 +447,7 @@ type Bootstrap_Observability_Tracing struct { func (x *Bootstrap_Observability_Tracing) Reset() { *x = Bootstrap_Observability_Tracing{} - mi := &file_conf_proto_msgTypes[7] + mi := &file_conf_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -524,7 +459,7 @@ func (x *Bootstrap_Observability_Tracing) String() string { func (*Bootstrap_Observability_Tracing) ProtoMessage() {} func (x *Bootstrap_Observability_Tracing) ProtoReflect() protoreflect.Message { - mi := &file_conf_proto_msgTypes[7] + mi := &file_conf_proto_msgTypes[6] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -537,7 +472,7 @@ func (x *Bootstrap_Observability_Tracing) ProtoReflect() protoreflect.Message { // Deprecated: Use Bootstrap_Observability_Tracing.ProtoReflect.Descriptor instead. func (*Bootstrap_Observability_Tracing) Descriptor() ([]byte, []int) { - return file_conf_proto_rawDescGZIP(), []int{0, 2, 1} + return file_conf_proto_rawDescGZIP(), []int{0, 1, 1} } func (x *Bootstrap_Observability_Tracing) GetEnabled() bool { @@ -577,7 +512,7 @@ type Server_CORS struct { func (x *Server_CORS) Reset() { *x = Server_CORS{} - mi := &file_conf_proto_msgTypes[8] + mi := &file_conf_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -589,7 +524,7 @@ func (x *Server_CORS) String() string { func (*Server_CORS) ProtoMessage() {} func (x *Server_CORS) ProtoReflect() protoreflect.Message { - mi := &file_conf_proto_msgTypes[8] + mi := &file_conf_proto_msgTypes[7] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -625,7 +560,7 @@ type Server_HTTP struct { func (x *Server_HTTP) Reset() { *x = Server_HTTP{} - mi := &file_conf_proto_msgTypes[9] + mi := &file_conf_proto_msgTypes[8] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -637,7 +572,7 @@ func (x *Server_HTTP) String() string { func (*Server_HTTP) ProtoMessage() {} func (x *Server_HTTP) ProtoReflect() protoreflect.Message { - mi := &file_conf_proto_msgTypes[9] + mi := &file_conf_proto_msgTypes[8] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -692,7 +627,7 @@ type Server_TLS struct { func (x *Server_TLS) Reset() { *x = Server_TLS{} - mi := &file_conf_proto_msgTypes[10] + mi := &file_conf_proto_msgTypes[9] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -704,7 +639,7 @@ func (x *Server_TLS) String() string { func (*Server_TLS) ProtoMessage() {} func (x *Server_TLS) ProtoReflect() protoreflect.Message { - mi := &file_conf_proto_msgTypes[10] + mi := &file_conf_proto_msgTypes[9] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -746,7 +681,7 @@ type Server_GRPC struct { func (x *Server_GRPC) Reset() { *x = Server_GRPC{} - mi := &file_conf_proto_msgTypes[11] + mi := &file_conf_proto_msgTypes[10] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -758,7 +693,7 @@ func (x *Server_GRPC) String() string { func (*Server_GRPC) ProtoMessage() {} func (x *Server_GRPC) ProtoReflect() protoreflect.Message { - mi := &file_conf_proto_msgTypes[11] + mi := &file_conf_proto_msgTypes[10] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -807,7 +742,7 @@ var File_conf_proto protoreflect.FileDescriptor const file_conf_proto_rawDesc = "" + "\n" + "\n" + - "conf.proto\x1a\x1bcredentials/v1/config.proto\x1a\x1egoogle/protobuf/duration.proto\"\xf9\x06\n" + + "conf.proto\x1a\x1bcredentials/v1/config.proto\x1a\x1egoogle/protobuf/duration.proto\"\xda\x05\n" + "\tBootstrap\x12\x1f\n" + "\x06server\x18\x01 \x01(\v2\a.ServerR\x06server\x12\x19\n" + "\x04auth\x18\x02 \x01(\v2\x05.AuthR\x04auth\x12>\n" + @@ -816,11 +751,7 @@ const file_conf_proto_rawDesc = "" + "\vnats_server\x18\x05 \x01(\v2\x15.Bootstrap.NatsServerR\n" + "natsServer\x12\x1f\n" + "\vstaging_dir\x18\x06 \x01(\tR\n" + - "stagingDir\x12B\n" + - "\x0fexistence_cache\x18\a \x01(\v2\x19.Bootstrap.ExistenceCacheR\x0eexistenceCache\x1aY\n" + - "\x0eExistenceCache\x12\x1a\n" + - "\bdisabled\x18\x01 \x01(\bR\bdisabled\x12+\n" + - "\x03ttl\x18\x02 \x01(\v2\x19.google.protobuf.DurationR\x03ttl\x1aH\n" + + "stagingDir\x1aH\n" + "\n" + "NatsServer\x12\x10\n" + "\x03uri\x18\x01 \x01(\tR\x03uri\x12\x16\n" + @@ -875,45 +806,42 @@ func file_conf_proto_rawDescGZIP() []byte { return file_conf_proto_rawDescData } -var file_conf_proto_msgTypes = make([]protoimpl.MessageInfo, 12) +var file_conf_proto_msgTypes = make([]protoimpl.MessageInfo, 11) var file_conf_proto_goTypes = []any{ (*Bootstrap)(nil), // 0: Bootstrap (*Server)(nil), // 1: Server (*Auth)(nil), // 2: Auth - (*Bootstrap_ExistenceCache)(nil), // 3: Bootstrap.ExistenceCache - (*Bootstrap_NatsServer)(nil), // 4: Bootstrap.NatsServer - (*Bootstrap_Observability)(nil), // 5: Bootstrap.Observability - (*Bootstrap_Observability_Sentry)(nil), // 6: Bootstrap.Observability.Sentry - (*Bootstrap_Observability_Tracing)(nil), // 7: Bootstrap.Observability.Tracing - (*Server_CORS)(nil), // 8: Server.CORS - (*Server_HTTP)(nil), // 9: Server.HTTP - (*Server_TLS)(nil), // 10: Server.TLS - (*Server_GRPC)(nil), // 11: Server.GRPC - (*v1.Credentials)(nil), // 12: credentials.v1.Credentials - (*durationpb.Duration)(nil), // 13: google.protobuf.Duration + (*Bootstrap_NatsServer)(nil), // 3: Bootstrap.NatsServer + (*Bootstrap_Observability)(nil), // 4: Bootstrap.Observability + (*Bootstrap_Observability_Sentry)(nil), // 5: Bootstrap.Observability.Sentry + (*Bootstrap_Observability_Tracing)(nil), // 6: Bootstrap.Observability.Tracing + (*Server_CORS)(nil), // 7: Server.CORS + (*Server_HTTP)(nil), // 8: Server.HTTP + (*Server_TLS)(nil), // 9: Server.TLS + (*Server_GRPC)(nil), // 10: Server.GRPC + (*v1.Credentials)(nil), // 11: credentials.v1.Credentials + (*durationpb.Duration)(nil), // 12: google.protobuf.Duration } var file_conf_proto_depIdxs = []int32{ 1, // 0: Bootstrap.server:type_name -> Server 2, // 1: Bootstrap.auth:type_name -> Auth - 5, // 2: Bootstrap.observability:type_name -> Bootstrap.Observability - 12, // 3: Bootstrap.credentials_service:type_name -> credentials.v1.Credentials - 4, // 4: Bootstrap.nats_server:type_name -> Bootstrap.NatsServer - 3, // 5: Bootstrap.existence_cache:type_name -> Bootstrap.ExistenceCache - 9, // 6: Server.http:type_name -> Server.HTTP - 11, // 7: Server.grpc:type_name -> Server.GRPC - 9, // 8: Server.http_metrics:type_name -> Server.HTTP - 13, // 9: Bootstrap.ExistenceCache.ttl:type_name -> google.protobuf.Duration - 6, // 10: Bootstrap.Observability.sentry:type_name -> Bootstrap.Observability.Sentry - 7, // 11: Bootstrap.Observability.tracing:type_name -> Bootstrap.Observability.Tracing - 13, // 12: Server.HTTP.timeout:type_name -> google.protobuf.Duration - 8, // 13: Server.HTTP.cors:type_name -> Server.CORS - 13, // 14: Server.GRPC.timeout:type_name -> google.protobuf.Duration - 10, // 15: Server.GRPC.tls_config:type_name -> Server.TLS - 16, // [16:16] is the sub-list for method output_type - 16, // [16:16] is the sub-list for method input_type - 16, // [16:16] is the sub-list for extension type_name - 16, // [16:16] is the sub-list for extension extendee - 0, // [0:16] is the sub-list for field type_name + 4, // 2: Bootstrap.observability:type_name -> Bootstrap.Observability + 11, // 3: Bootstrap.credentials_service:type_name -> credentials.v1.Credentials + 3, // 4: Bootstrap.nats_server:type_name -> Bootstrap.NatsServer + 8, // 5: Server.http:type_name -> Server.HTTP + 10, // 6: Server.grpc:type_name -> Server.GRPC + 8, // 7: Server.http_metrics:type_name -> Server.HTTP + 5, // 8: Bootstrap.Observability.sentry:type_name -> Bootstrap.Observability.Sentry + 6, // 9: Bootstrap.Observability.tracing:type_name -> Bootstrap.Observability.Tracing + 12, // 10: Server.HTTP.timeout:type_name -> google.protobuf.Duration + 7, // 11: Server.HTTP.cors:type_name -> Server.CORS + 12, // 12: Server.GRPC.timeout:type_name -> google.protobuf.Duration + 9, // 13: Server.GRPC.tls_config:type_name -> Server.TLS + 14, // [14:14] is the sub-list for method output_type + 14, // [14:14] is the sub-list for method input_type + 14, // [14:14] is the sub-list for extension type_name + 14, // [14:14] is the sub-list for extension extendee + 0, // [0:14] is the sub-list for field type_name } func init() { file_conf_proto_init() } @@ -921,17 +849,17 @@ func file_conf_proto_init() { if File_conf_proto != nil { return } - file_conf_proto_msgTypes[4].OneofWrappers = []any{ + file_conf_proto_msgTypes[3].OneofWrappers = []any{ (*Bootstrap_NatsServer_Token)(nil), } - file_conf_proto_msgTypes[7].OneofWrappers = []any{} + file_conf_proto_msgTypes[6].OneofWrappers = []any{} type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_conf_proto_rawDesc), len(file_conf_proto_rawDesc)), NumEnums: 0, - NumMessages: 12, + NumMessages: 11, NumExtensions: 0, NumServices: 0, }, diff --git a/app/artifact-cas/internal/conf/conf.proto b/app/artifact-cas/internal/conf/conf.proto index 44ee8fef2..d9c8c5399 100644 --- a/app/artifact-cas/internal/conf/conf.proto +++ b/app/artifact-cas/internal/conf/conf.proto @@ -34,17 +34,6 @@ message Bootstrap { // production a dedicated emptyDir is mounted here (NOT tmpfs/RAM, and NOT the // /tmp secret mount). Required: when unset the service refuses to start. string staging_dir = 6; - // Cache of the blobs known to exist in a backend, so repeated uploads of the - // same content skip the secrets manager and the backend. Shared through NATS - // KV when nats_server is set, in memory otherwise. Enabled by default. - ExistenceCache existence_cache = 7; - - message ExistenceCache { - // Turns the cache off - bool disabled = 1; - // Lifetime of an entry. Defaults to 24h. - google.protobuf.Duration ttl = 2; - } message NatsServer { // NATS server URI, e.g. "nats://localhost:4222" diff --git a/deployment/chainloop/Chart.yaml b/deployment/chainloop/Chart.yaml index 535abbc63..ed2dfc767 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.459.1 +version: 1.459.0 # Do not update appVersion, this is handled automatically by the release process appVersion: v1.118.1 diff --git a/deployment/chainloop/README.md b/deployment/chainloop/README.md index d866737b3..25eb3392b 100644 --- a/deployment/chainloop/README.md +++ b/deployment/chainloop/README.md @@ -794,9 +794,6 @@ Once done, you can access with [two predefined users](https://github.com/chainlo | `cas.nats.host` | NATS Host | `""` | | `cas.nats.port` | NATS Port | `4222` | | `cas.nats.token` | NATS Client authentication token | `""` | -| `cas.existenceCache` | Cache of the blobs known to exist in a backend, so repeated uploads of the same content skip the backend. Shared between replicas through NATS when cas.nats is enabled, in memory otherwise. | | -| `cas.existenceCache.enabled` | Enable the existence cache | `true` | -| `cas.existenceCache.ttl` | Lifetime of a cache entry, in seconds with an "s" suffix | `86400s` | | `cas.image.registry` | Image registry | `REGISTRY_NAME` | | `cas.image.repository` | Image repository | `REPOSITORY_NAME` | | `cas.containerPorts.http` | controlplane HTTP container port | `8000` | @@ -961,7 +958,7 @@ service_registration "kubernetes" {}` | ## License -Copyright © 2023-2026 The Chainloop Authors +Copyright © 2023-2025 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. diff --git a/deployment/chainloop/templates/cas/configmap.yaml b/deployment/chainloop/templates/cas/configmap.yaml index d3d977229..fd0aa6e63 100644 --- a/deployment/chainloop/templates/cas/configmap.yaml +++ b/deployment/chainloop/templates/cas/configmap.yaml @@ -17,9 +17,6 @@ data: # Local directory where uploads and downloads are staged and verified against # their digest before being forwarded. Backed by the staging emptyDir. staging_dir: {{ include "chainloop.cas.staging_dir" . | quote }} - existence_cache: - disabled: {{ not .Values.cas.existenceCache.enabled }} - ttl: {{ .Values.cas.existenceCache.ttl | quote }} server: http: addr: "0.0.0.0:{{ .Values.cas.containerPorts.http }}" diff --git a/deployment/chainloop/values.yaml b/deployment/chainloop/values.yaml index a649489c2..63b2cea87 100644 --- a/deployment/chainloop/values.yaml +++ b/deployment/chainloop/values.yaml @@ -1042,13 +1042,6 @@ cas: port: 4222 token: "" - ## @extra cas.existenceCache Cache of the blobs known to exist in a backend, so repeated uploads of the same content skip the backend. Shared between replicas through NATS when cas.nats is enabled, in memory otherwise. - ## @param cas.existenceCache.enabled Enable the existence cache - ## @param cas.existenceCache.ttl Lifetime of a cache entry, in seconds with an "s" suffix - existenceCache: - enabled: true - ttl: 86400s - ## @param cas.image.registry [default: REGISTRY_NAME] Image registry ## @param cas.image.repository [default: REPOSITORY_NAME] Image repository ## @skip cas.image.tag diff --git a/docs/specs/issue-3543-cas-existence-cache.md b/docs/specs/issue-3543-cas-existence-cache.md index 2d743bf0c..4c64424f1 100644 --- a/docs/specs/issue-3543-cas-existence-cache.md +++ b/docs/specs/issue-3543-cas-existence-cache.md @@ -38,8 +38,8 @@ Each cache entry MUST belong to the namespace of one organization and one CAS ba - Done when: a digest that organization A uploaded is still checked against the backend when organization B uploads it, also on shared storage. - Done when: the same digest in backend A is still checked against backend B of the same organization. -### R-004: Bounded lifetime -A cached existence result MUST expire after a configurable time. The default is 24 hours. An operator MUST be able to turn the cache off. +### ~~R-004: Bounded lifetime~~ +~~A cached existence result MUST expire after a configurable time. The default is 24 hours. An operator MUST be able to turn the cache off.~~ Replaced by R-010 (D-010). ### R-005: Same authorization as today A cached hit MUST NOT bypass any check that runs today. The Artifact CAS checks the upload token, its role, and its limits before it reads the cache. @@ -56,6 +56,9 @@ If the cache is not available (for example, NATS is down), the Artifact CAS MUST ### R-009: Observability The Artifact CAS SHOULD expose metrics for cache hits and misses. +### R-010: Fixed lifetime, always on +A cached existence result MUST expire after 24 hours. The cache is always on. It has no setting in the CAS configuration or in the Helm chart. + ## Constraints - Public repository. The design must work for self-hosted deployments with and without NATS. - Backend credentials are secrets. They must not leave the Artifact CAS process. @@ -68,7 +71,7 @@ Inside the Artifact CAS, the upload handler uses two caches. **Existence cache.** Each organization has its own namespace in the cache. The key starts with the organization ID from the signed upload token. Then come the backend type, the hashed backend secret reference, and the digest. The hash keeps the secret path out of the key. The namespace is necessary also when two organizations share storage. Without it, organization B could learn that organization A stored a file with a given digest. With it, the cache tells a tenant nothing that the tenant's own backend check does not already tell. The existence cache has its own NATS bucket, separate from the other caches. The value holds the blob size. The handler reads the cache after it checks the token and reads the first chunk. On a hit, it closes the stream as "already exists", which is the response the CLI receives today. On a miss, it continues as today. When the backend says the blob exists, or when an upload succeeds, the handler writes the entry. -The existence cache uses the shared cache library that the control plane already uses for attestation and policy bundles. The Artifact CAS can have a NATS connection, which today sends only audit events. With that connection, the cache is a NATS key-value bucket that all replicas share. Without NATS, each replica has an in-memory LRU cache with expiry. The CAS selects the store one time, at startup. The entry holds only a digest and a size, never the artifact content and never credentials. An entry is less than 200 bytes, so its size is never a problem for NATS. If a NATS read or write fails at runtime, the CAS does not use the in-memory cache. It treats the failure as a miss and checks the backend (R-008). When the bucket is full, NATS deletes its oldest entries. +The existence cache uses the shared cache library that the control plane already uses for attestation and policy bundles. The Artifact CAS can have a NATS connection, which today sends only audit events. With that connection, the cache is a NATS key-value bucket that all replicas share. Without NATS, each replica has an in-memory LRU cache with expiry. The CAS selects the store one time, at startup. If the NATS bucket cannot be set up at startup, the CAS uses the in-memory cache (D-011). The entry holds only a digest and a size, never the artifact content and never credentials. An entry is less than 200 bytes, so its size is never a problem for NATS. If a NATS read or write fails at runtime, the CAS does not use the in-memory cache. It treats the failure as a miss and checks the backend (R-008). When the bucket is full, NATS deletes its oldest entries. **Client cache.** A small in-memory cache keyed by backend type and secret reference holds the loaded storage client for a few minutes. A miss in the existence cache then needs one backend call instead of a secrets manager read plus a new client plus a backend call. Credentials can change behind the same secret reference. The short lifetime limits how long an old client stays in use. If the backend rejects the old credentials, the handler drops the client and loads it again one time. @@ -113,9 +116,13 @@ sequenceDiagram | D-007 | Default lifetime of an existence entry | 24 hours, configurable | Blobs cannot be deleted through Chainloop. The only risk is a purge outside Chainloop, and 24 hours limits it. Rejected: 7 days (a longer exposure to a purge). | drafting | | D-008 | Describe endpoint | Does not read the existence cache | Only the download command uses it, and the download reads the blob from the backend next. Downloads stay without a cache. | drafting | | D-009 | Client cache lifetime | 5 minutes, and one reload on an authentication error | A short window for rotated credentials. Rejected: 1 minute (less reuse for a small gain). | drafting | +| D-010 | Configuration of the existence cache | None. Always on, with a fixed lifetime of 24 hours. The store is NATS KV when the CAS has a NATS connection, and memory otherwise. | The cache is safe by default and only makes uploads faster, so a setting adds surface with no clear use. Rejected: a configurable lifetime and an off switch in the CAS configuration and the Helm chart (D-007, R-004). | [PR #3562](https://github.com/chainloop-dev/chainloop/pull/3562) | +| D-011 | NATS bucket fails at startup | Use the in-memory cache, and log a warning | The cache is optional. It must not stop the CAS from starting, for example when the NATS credentials can only publish audit events. Rejected: fail the startup, as the control plane caches do. | [PR #3562](https://github.com/chainloop-dev/chainloop/pull/3562) | +| D-012 | Client cache key | Backend type, secret reference, and organization | One backend keeps per-organization session credentials inside its client. A client shared between organizations could sign a request with the session of another organization. Rejected: backend type and secret reference only. | [PR #3562](https://github.com/chainloop-dev/chainloop/pull/3562) | +| D-013 | Where the client cache applies | Uploads only | Uploads are the cost this spec reduces. Downloads and Describe load the backend as before. | [PR #3562](https://github.com/chainloop-dev/chainloop/pull/3562) | ## Open Questions -None at drafting time. D-007 to D-009 record the answers to the drafting questions. +None at drafting time. D-007 to D-009 record the answers to the drafting questions. D-010 to D-013 record the changes made during the build. ## Milestones 1. **Existence cache.** R-001 to R-006, R-008 and R-009. Repeated uploads skip the backend. @@ -124,6 +131,6 @@ None at drafting time. D-007 to D-009 record the answers to the drafting questio ## Risks | Risk | Mitigation | |------|------------| -| A blob is purged outside Chainloop, and the cache still says it exists. An attestation then references a missing blob. | Bounded lifetime, configurable per deployment, and a switch to turn the cache off. Downloads never use the cache. | +| A blob is purged outside Chainloop, and the cache still says it exists. An attestation then references a missing blob. | A fixed lifetime of 24 hours (R-010). Downloads never use the cache. | | Rotated credentials behind the same secret reference. The client cache keeps the old client. | Short lifetime, and one reload when the backend rejects the credentials. | | A cache entry crosses tenants, or shows that another tenant has a file. | Each organization has its own namespace, taken from the signed token. Tests cover two organizations on shared storage (R-003). | diff --git a/pkg/cache/cache.go b/pkg/cache/cache.go index e302e6bd5..772915fb8 100644 --- a/pkg/cache/cache.go +++ b/pkg/cache/cache.go @@ -80,7 +80,8 @@ func WithMaxBytes(n int64) Option { } // WithMaxEntries sets the maximum number of entries of the in-memory backend. -// Defaults to 1000. Ignored for NATS KV backend, which is bounded by WithMaxBytes. +// A value of 0 or less keeps the default of 1000. Ignored for NATS KV backend, +// which is bounded by WithMaxBytes. func WithMaxEntries(n int) Option { return func(c *config) { c.maxEntries = n } } diff --git a/pkg/cache/casexistence/casexistence.go b/pkg/cache/casexistence/casexistence.go index b704849b0..6ba6bccaf 100644 --- a/pkg/cache/casexistence/casexistence.go +++ b/pkg/cache/casexistence/casexistence.go @@ -30,9 +30,8 @@ import ( ) const ( - // DefaultTTL is the lifetime of an entry when none is configured. - DefaultTTL = 24 * time.Hour - + // ttl bounds how long an entry outlives a purge of the blob outside Chainloop + ttl = 24 * time.Hour maxBytes = 64 * 1024 * 1024 // 64 MB maxEntries = 50_000 bucket = "chainloop-cas-existence" @@ -50,26 +49,22 @@ type Cache struct { } // New creates an existence cache, backed by NATS KV when rc is set and by an -// in-memory LRU otherwise. A ttl of 0 uses DefaultTTL. A NATS bucket that can -// not be set up falls back to memory: the cache only makes uploads faster, so -// it must not stop its caller from starting. -func New(ctx context.Context, rc *natsconn.ReloadableConnection, ttl time.Duration, logger log.Logger) (*Cache, error) { - if ttl <= 0 { - ttl = DefaultTTL - } - - c, err := newCache(ctx, rc, ttl, logger) +// in-memory LRU otherwise. A NATS bucket that can not be set up falls back to +// memory: the cache only makes uploads faster, so it must not stop its caller +// from starting. +func New(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (*Cache, error) { + c, err := newCache(ctx, rc, logger) if err != nil && rc != nil { if logger != nil { log.NewHelper(logger).Warnw("msg", "existence cache: NATS KV unavailable, using memory", "error", err) } - c, err = newCache(ctx, nil, ttl, logger) + c, err = newCache(ctx, nil, logger) } return c, err } -func newCache(ctx context.Context, rc *natsconn.ReloadableConnection, ttl time.Duration, logger log.Logger) (*Cache, error) { +func newCache(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (*Cache, error) { opts := []cache.Option{ cache.WithTTL(ttl), cache.WithMaxBytes(maxBytes), diff --git a/pkg/cache/casexistence/casexistence_test.go b/pkg/cache/casexistence/casexistence_test.go new file mode 100644 index 000000000..936ff2490 --- /dev/null +++ b/pkg/cache/casexistence/casexistence_test.go @@ -0,0 +1,84 @@ +// +// Copyright 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. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package casexistence + +import ( + "context" + "testing" + "time" + + "github.com/chainloop-dev/chainloop/pkg/natsconn" + 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" +) + +func startNATS(t *testing.T, jetStream bool) *nats.Conn { + t.Helper() + ns, err := natsserver.NewServer(&natsserver.Options{ + Host: "127.0.0.1", + Port: -1, + JetStream: jetStream, + StoreDir: t.TempDir(), + }) + require.NoError(t, err) + ns.Start() + t.Cleanup(ns.Shutdown) + + require.True(t, ns.ReadyForConnections(5*time.Second)) + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + t.Cleanup(nc.Close) + return nc +} + +// TestNew: the cache uses NATS KV when it can, and falls back to memory when +// the bucket can not be set up, so its caller still starts. +func TestNew(t *testing.T) { + testCases := []struct { + name string + jetStream bool + wantBucket bool + }{ + {name: "NATS KV when JetStream is available", jetStream: true, wantBucket: true}, + {name: "memory when the bucket can not be set up", jetStream: false}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + ctx := t.Context() + nc := startNATS(t, tc.jetStream) + + c, err := New(ctx, &natsconn.ReloadableConnection{Conn: nc}, nil) + require.NoError(t, err) + + require.NoError(t, c.Set(ctx, "key", Entry{Size: 42})) + got, ok, err := c.Get(ctx, "key") + require.NoError(t, err) + assert.True(t, ok) + assert.Equal(t, int64(42), got.Size) + + if tc.wantBucket { + js, err := jetstream.New(nc) + require.NoError(t, err) + _, err = js.KeyValue(context.Background(), bucket) + assert.NoError(t, err, "the cache must be stored in its own NATS bucket") + } + }) + } +} From c6d1fc44f55d5d3457fbb191fae063f0559dde57 Mon Sep 17 00:00:00 2001 From: Miguel Martinez Trivino Date: Thu, 8 Oct 2026 13:35:02 +0200 Subject: [PATCH 4/4] chore: drop spec amendment from the implementation PR The spec amendment for issue-3543 moves to its own spec PR. Assisted-by: Claude Code Signed-off-by: Miguel Martinez Trivino Chainloop-Trace-Sessions: 9b6de27d-6322-4cf7-8e72-5e56d858c783 --- docs/specs/issue-3543-cas-existence-cache.md | 17 +++++------------ 1 file changed, 5 insertions(+), 12 deletions(-) diff --git a/docs/specs/issue-3543-cas-existence-cache.md b/docs/specs/issue-3543-cas-existence-cache.md index 4c64424f1..2d743bf0c 100644 --- a/docs/specs/issue-3543-cas-existence-cache.md +++ b/docs/specs/issue-3543-cas-existence-cache.md @@ -38,8 +38,8 @@ Each cache entry MUST belong to the namespace of one organization and one CAS ba - Done when: a digest that organization A uploaded is still checked against the backend when organization B uploads it, also on shared storage. - Done when: the same digest in backend A is still checked against backend B of the same organization. -### ~~R-004: Bounded lifetime~~ -~~A cached existence result MUST expire after a configurable time. The default is 24 hours. An operator MUST be able to turn the cache off.~~ Replaced by R-010 (D-010). +### R-004: Bounded lifetime +A cached existence result MUST expire after a configurable time. The default is 24 hours. An operator MUST be able to turn the cache off. ### R-005: Same authorization as today A cached hit MUST NOT bypass any check that runs today. The Artifact CAS checks the upload token, its role, and its limits before it reads the cache. @@ -56,9 +56,6 @@ If the cache is not available (for example, NATS is down), the Artifact CAS MUST ### R-009: Observability The Artifact CAS SHOULD expose metrics for cache hits and misses. -### R-010: Fixed lifetime, always on -A cached existence result MUST expire after 24 hours. The cache is always on. It has no setting in the CAS configuration or in the Helm chart. - ## Constraints - Public repository. The design must work for self-hosted deployments with and without NATS. - Backend credentials are secrets. They must not leave the Artifact CAS process. @@ -71,7 +68,7 @@ Inside the Artifact CAS, the upload handler uses two caches. **Existence cache.** Each organization has its own namespace in the cache. The key starts with the organization ID from the signed upload token. Then come the backend type, the hashed backend secret reference, and the digest. The hash keeps the secret path out of the key. The namespace is necessary also when two organizations share storage. Without it, organization B could learn that organization A stored a file with a given digest. With it, the cache tells a tenant nothing that the tenant's own backend check does not already tell. The existence cache has its own NATS bucket, separate from the other caches. The value holds the blob size. The handler reads the cache after it checks the token and reads the first chunk. On a hit, it closes the stream as "already exists", which is the response the CLI receives today. On a miss, it continues as today. When the backend says the blob exists, or when an upload succeeds, the handler writes the entry. -The existence cache uses the shared cache library that the control plane already uses for attestation and policy bundles. The Artifact CAS can have a NATS connection, which today sends only audit events. With that connection, the cache is a NATS key-value bucket that all replicas share. Without NATS, each replica has an in-memory LRU cache with expiry. The CAS selects the store one time, at startup. If the NATS bucket cannot be set up at startup, the CAS uses the in-memory cache (D-011). The entry holds only a digest and a size, never the artifact content and never credentials. An entry is less than 200 bytes, so its size is never a problem for NATS. If a NATS read or write fails at runtime, the CAS does not use the in-memory cache. It treats the failure as a miss and checks the backend (R-008). When the bucket is full, NATS deletes its oldest entries. +The existence cache uses the shared cache library that the control plane already uses for attestation and policy bundles. The Artifact CAS can have a NATS connection, which today sends only audit events. With that connection, the cache is a NATS key-value bucket that all replicas share. Without NATS, each replica has an in-memory LRU cache with expiry. The CAS selects the store one time, at startup. The entry holds only a digest and a size, never the artifact content and never credentials. An entry is less than 200 bytes, so its size is never a problem for NATS. If a NATS read or write fails at runtime, the CAS does not use the in-memory cache. It treats the failure as a miss and checks the backend (R-008). When the bucket is full, NATS deletes its oldest entries. **Client cache.** A small in-memory cache keyed by backend type and secret reference holds the loaded storage client for a few minutes. A miss in the existence cache then needs one backend call instead of a secrets manager read plus a new client plus a backend call. Credentials can change behind the same secret reference. The short lifetime limits how long an old client stays in use. If the backend rejects the old credentials, the handler drops the client and loads it again one time. @@ -116,13 +113,9 @@ sequenceDiagram | D-007 | Default lifetime of an existence entry | 24 hours, configurable | Blobs cannot be deleted through Chainloop. The only risk is a purge outside Chainloop, and 24 hours limits it. Rejected: 7 days (a longer exposure to a purge). | drafting | | D-008 | Describe endpoint | Does not read the existence cache | Only the download command uses it, and the download reads the blob from the backend next. Downloads stay without a cache. | drafting | | D-009 | Client cache lifetime | 5 minutes, and one reload on an authentication error | A short window for rotated credentials. Rejected: 1 minute (less reuse for a small gain). | drafting | -| D-010 | Configuration of the existence cache | None. Always on, with a fixed lifetime of 24 hours. The store is NATS KV when the CAS has a NATS connection, and memory otherwise. | The cache is safe by default and only makes uploads faster, so a setting adds surface with no clear use. Rejected: a configurable lifetime and an off switch in the CAS configuration and the Helm chart (D-007, R-004). | [PR #3562](https://github.com/chainloop-dev/chainloop/pull/3562) | -| D-011 | NATS bucket fails at startup | Use the in-memory cache, and log a warning | The cache is optional. It must not stop the CAS from starting, for example when the NATS credentials can only publish audit events. Rejected: fail the startup, as the control plane caches do. | [PR #3562](https://github.com/chainloop-dev/chainloop/pull/3562) | -| D-012 | Client cache key | Backend type, secret reference, and organization | One backend keeps per-organization session credentials inside its client. A client shared between organizations could sign a request with the session of another organization. Rejected: backend type and secret reference only. | [PR #3562](https://github.com/chainloop-dev/chainloop/pull/3562) | -| D-013 | Where the client cache applies | Uploads only | Uploads are the cost this spec reduces. Downloads and Describe load the backend as before. | [PR #3562](https://github.com/chainloop-dev/chainloop/pull/3562) | ## Open Questions -None at drafting time. D-007 to D-009 record the answers to the drafting questions. D-010 to D-013 record the changes made during the build. +None at drafting time. D-007 to D-009 record the answers to the drafting questions. ## Milestones 1. **Existence cache.** R-001 to R-006, R-008 and R-009. Repeated uploads skip the backend. @@ -131,6 +124,6 @@ None at drafting time. D-007 to D-009 record the answers to the drafting questio ## Risks | Risk | Mitigation | |------|------------| -| A blob is purged outside Chainloop, and the cache still says it exists. An attestation then references a missing blob. | A fixed lifetime of 24 hours (R-010). Downloads never use the cache. | +| A blob is purged outside Chainloop, and the cache still says it exists. An attestation then references a missing blob. | Bounded lifetime, configurable per deployment, and a switch to turn the cache off. Downloads never use the cache. | | Rotated credentials behind the same secret reference. The client cache keeps the old client. | Short lifetime, and one reload when the backend rejects the credentials. | | A cache entry crosses tenants, or shows that another tenant has a file. | Each organization has its own namespace, taken from the signed token. Tests cover two organizations on shared storage (R-003). |