feat(cache): unified cache abstraction with in-memory and NATS KV backends

[migmartri]

Summary

• Add new pkg/cache/ shared library with a generic Cache[T] interface, backed by in-memory LRU (default) or NATS JetStream KV (when configured)
• Migrate all four existing ad-hoc caching mechanisms (membership cache, JWT claims cache, remote policy cache, remote policy group cache) to use the unified interface
• Eliminate package-level global cache state by injecting cache instances through constructors and Wire
• Fix unbounded growth in remote policy/group caches by adding 5-minute TTL expiration

Closes #2951

[cubic-dev-ai]

7 issues found across 17 files

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="pkg/cache/natskv_test.go">

<violation number="1" location="pkg/cache/natskv_test.go:106">
P2: Do not ignore `Get` errors in the purge test; assert them so backend failures can't be hidden as cache misses.</violation>
</file>

<file name="pkg/cache/natskv.go">

<violation number="1" location="pkg/cache/natskv.go:90">
P1: `sanitizeKey` introduces key-collision risk by mapping `:` to `.`, which can make distinct input keys resolve to the same NATS KV key.</violation>

<violation number="2" location="pkg/cache/natskv.go:99">
P2: These cache methods ignore the caller context and always use `context.Background()`, so cancellation/deadline propagation is lost.</violation>
</file>

<file name="pkg/cache/cache.go">

<violation number="1" location="pkg/cache/cache.go:86">
P1: Reject non-positive TTL values; currently negative TTLs are accepted and create invalid cache expiration behavior.</violation>

<violation number="2" location="pkg/cache/cache.go:94">
P1: Fail fast when NATS is configured without a bucket name instead of silently falling back to in-memory cache.</violation>
</file>

<file name="pkg/cache/memory.go">

<violation number="1" location="pkg/cache/memory.go:32">
P2: Using `NewLRU` with size `0` disables LRU eviction and creates an unlimited-size cache.</violation>
</file>

<file name="pkg/policies/policy_groups.go">

<violation number="1" location="pkg/policies/policy_groups.go:239">
P1: `opts.GroupCache` is passed without a nil fallback, which can panic when loading chainloop policy groups from call sites that don't set `GroupCache`.</violation>
</file>

_{Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.}

[cubic-dev-ai]

1 issue found across 4 files (changes from recent commits).

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="pkg/cache/natskv.go">

<violation number="1" location="pkg/cache/natskv.go:187">
P2: Do not silently discard per-key purge errors; at least log failures to avoid reporting a successful purge when keys were not removed.</violation>
</file>

_{Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.}

[migmartri]

Code review

Found 4 issues:

1. Nil pointer dereference in ChainloopGroupLoader.Load when GroupCache is nil. attestation_init.go calls LoadPolicyGroup without setting GroupCache, so ChainloopGroupLoader is created with a nil cache. c.cache.Get(ctx, ref) on the nil interface will panic at runtime for chainloop:// scheme policy group references. The server-side path correctly initializes groupCache when nil (in NewPolicyGroupVerifier), but the CLI call site was not updated.

chainloop/pkg/policies/group_loader.go

Lines 116 to 118 in 7a2e219

	if cached, ok, _ := c.cache.Get(ctx, ref); ok {
	return cached.Group, cached.Reference, nil

chainloop/app/cli/pkg/action/attestation_init.go

Lines 353 to 358 in 7a2e219

	contractMaterials := schema.GetMaterials()
	for _, pgAtt := range schema.GetPolicyGroups() {
	group, _, err := policies.LoadPolicyGroup(ctx, pgAtt, &policies.LoadPolicyGroupOptions{
	Client: client,
	Logger: logger,
	})

2. Regression: singleflight deduplication and bounded timeouts removed from policy/group loaders. PR feat(policies): parallelize policy evaluations with bounded concurrency #2901 (commit c7f0b9b7, merged to main after this branch diverged) added singleflight.Group with 30-second bounded timeouts to ChainloopLoader.Load and ChainloopGroupLoader.Load to prevent thundering-herd on concurrent gRPC fetches. The Cache[T] abstraction provides caching but not singleflight coalescing -- multiple goroutines can race past a cache miss and each issue separate gRPC calls. Branch needs rebasing on main.

chainloop/pkg/policies/loader.go

Lines 161 to 215 in 7a2e219

	cache cache.Cache[*policyWithReference]
	}
	type policyWithReference struct {
	Policy *v1.Policy `json:"policy"`
	Reference *PolicyDescriptor `json:"reference"`
	}
	func NewChainloopLoader(client pb.AttestationServiceClient, c cache.Cache[*policyWithReference]) *ChainloopLoader {
	return &ChainloopLoader{Client: client, cache: c}
	}
	func (c *ChainloopLoader) Load(ctx context.Context, attachment *v1.PolicyAttachment) (*v1.Policy, *PolicyDescriptor, error) {
	ref := attachment.GetRef()
	if cached, ok, _ := c.cache.Get(ctx, ref); ok {
	return cached.Policy, cached.Reference, nil
	}
	if !IsProviderScheme(ref) {
	return nil, nil, fmt.Errorf("invalid policy reference %q", ref)
	}
	providerRef := ProviderParts(ref)
	resp, err := c.Client.GetPolicy(ctx, &pb.AttestationServiceGetPolicyRequest{
	Provider: providerRef.Provider,
	PolicyName: providerRef.Name,
	OrgName: providerRef.OrgName,
	})
	if err != nil {
	return nil, nil, fmt.Errorf("requesting remote policy (provider: %s, name: %s): %w", providerRef.Provider, providerRef.Name, err)
	}
	h, err := crv1.NewHash(resp.Reference.GetDigest())
	if err != nil {
	return nil, nil, fmt.Errorf("parsing digest: %w", err)
	}
	orgName := providerRef.OrgName
	// Extract organization name from URL if present
	if u, err := url.Parse(resp.Reference.GetUrl()); err == nil {
	if orgParam := u.Query().Get("org"); orgParam != "" {
	orgName = orgParam
	}
	}
	reference := policyReferenceResourceDescriptor(providerRef.Name, resp.Reference.GetUrl(), orgName, h)
	// cache result
	_ = c.cache.Set(ctx, ref, &policyWithReference{Policy: resp.GetPolicy(), Reference: reference})
	return resp.GetPolicy(), reference, nil
	}
	func unmarshallResource(raw []byte, ref string, digest string, dest proto.Message) (*PolicyDescriptor, error) {

https://github.com/chainloop-dev/chainloop/blob/7a2e219e92d584f3e78e918d1955f41089abb908/pkg/policies/group_loader.go#L109-L157

3. Regression: parallel policy evaluation (errgroup) reverted to sequential loops. PR feat(policies): parallelize policy evaluations with bounded concurrency #2901 added errgroup with bounded concurrency (max(runtime.NumCPU(), 5)) to parallelize policy evaluations in VerifyMaterial and VerifyStatement. This branch predates that merge, so evaluation is sequential. Additionally, VerifyStatement re-marshals the statement on every loop iteration instead of once before the loop.

chainloop/pkg/policies/policies.go

Lines 200 to 215 in 7a2e219

	for _, attachment := range attachments {
	ev, err := pv.evaluatePolicyAttachment(ctx, attachment, subject,
	&evalOpts{kind: material.MaterialType, name: material.GetId()},
	)
	if err != nil {
	return nil, NewPolicyError(err)
	}
	if ev != nil {
	result = append(result, ev)
	}
	}
	return result, nil
	}

chainloop/pkg/policies/policies.go

Lines 455 to 475 in 7a2e219

	result := make([]*v12.PolicyEvaluation, 0)
	policies := pv.policies.GetAttestation()
	for _, policyAtt := range policies {
	material, err := protojson.Marshal(statement)
	if err != nil {
	return nil, NewPolicyError(err)
	}
	ev, err := pv.evaluatePolicyAttachment(ctx, policyAtt, material, &evalOpts{kind: v1.CraftingSchema_Material_ATTESTATION})
	if err != nil {
	return nil, NewPolicyError(err)
	}
	if ev != nil {
	result = append(result, ev)
	}
	}
	return result, nil
	}

4. Regression: concurrency_test.go will be dropped on merge. PR feat(policies): parallelize policy evaluations with bounded concurrency #2901 added pkg/policies/concurrency_test.go to exercise race conditions under concurrent policy evaluation. This file exists on main but not on this branch. Merging without rebasing removes that test coverage.

Note: Issues 2-4 share a root cause -- the branch diverged from main before PR #2901 was merged. Rebasing on main and integrating the cache abstraction with the singleflight, errgroup, and bounded-timeout patterns from #2901 should resolve all three.

[cubic-dev-ai]

1 issue found across 7 files (changes from recent commits).

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="pkg/policies/policy_groups.go">

<violation number="1" location="pkg/policies/policy_groups.go:269">
P1: Ensure a non-nil GroupCache is always provided. Passing nil into NewChainloopGroupLoader will panic when Load calls c.cache.Get/Set.</violation>
</file>

_{Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.}