fix(identityconfig): bound the installed identity-config snapshot across refreshes

targeting/identityagent/config.go

@@ -294,6 +294,17 @@ type IdentityConfigSourceConfig struct {
 	// floor: a deployment with a handful of packages may legitimately want
 	// a cap in the tens.
 	MaxConfigEntries int
+
+	// MaxSnapshotEntries bounds the snapshot the identity-config service
+	// installs, across refreshes: it caps the cumulative result of every
+	// delta the endpoint has sent over the process's life, where
+	// MaxConfigEntries caps only one response. Loaded from
+	// CONFIG_MAX_SNAPSHOT_ENTRIES; zero or negative disables the cap
+	// (the default — the previous unbounded behavior). When set, a delta
+	// that would push the installed snapshot over the cap is rejected,
+	// the prior snapshot is kept, and the refresh reports an error
+	// outcome, so a frozen snapshot is loud rather than silent.
+	MaxSnapshotEntries int
 }
 
 // minConfigMaxResponseBytes is the floor on a positive
@@ -523,6 +534,10 @@ func LoadConfigFromEnv() (Config, error) {
 	if err != nil {
 		errs = append(errs, err)
 	}
+	configMaxSnapshotEntries, err := lookupInt("CONFIG_MAX_SNAPSHOT_ENTRIES", 0)
+	if err != nil {
+		errs = append(errs, err)
+	}
 	slotIDs, err := parseTmpxSlotIDs(os.Getenv("TMPX_SLOT_IDS"))
 	if err != nil {
 		errs = append(errs, err)
@@ -639,6 +654,7 @@ func LoadConfigFromEnv() (Config, error) {
 			ExtraHeaders:       extraHeaders,
 			MaxResponseBytes:   configMaxResponseBytes,
 			MaxConfigEntries:   configMaxEntries,
+			MaxSnapshotEntries: configMaxSnapshotEntries,
 		},
 		AudienceValkey:         audienceBlock,
 		FCapValkey:             fcapBlock,

targeting/identityagent/setup.go

@@ -365,6 +365,7 @@ func buildBundle(ctx context.Context, cfg Config, recorder Recorder, logger *slo
 	}
 	configSvc, err := identityconfig.New(source, cfg.IdentityConfig.RefreshInterval,
 		identityconfig.WithStartConfig(startConfigFor(cfg.IdentityConfig)),
+		identityconfig.WithMaxSnapshotEntries(cfg.IdentityConfig.MaxSnapshotEntries),
 		identityconfig.WithLogger(logger),
 		identityconfig.WithRefreshObserver(func(outcome string) {
 			recorder.ConfigRefresh(context.Background(), outcome)

targeting/identityconfig/service.go

@@ -43,6 +43,10 @@ type Service struct {
 	logger          *slog.Logger
 	refreshObserver func(outcome string)
 
+	// maxSnapshotEntries bounds the installed snapshot (see
+	// WithMaxSnapshotEntries). Zero or negative means unlimited.
+	maxSnapshotEntries int
+
 	snap atomic.Pointer[snapshotData]
 
 	mu      sync.Mutex // serializes Start/Stop and concurrent manual refreshes
@@ -110,6 +114,23 @@ func WithLogger(logger *slog.Logger) Option {
 	}
 }
 
+// WithMaxSnapshotEntries bounds the installed snapshot: any initial load or
+// delta refresh whose merged snapshot would hold more than n entries is
+// rejected and the prior snapshot is kept. A rejected delta reports
+// RefreshOutcomeError through the refresh observer and leaves the watermark
+// unadvanced, so a later under-cap delta can still apply; a rejected
+// initial load fails Start under StartModeFailFast (and is retried under
+// StartModeRetry). This is the cap on the live snapshot, distinct from the
+// config source's per-response cap: without it, each refresh can add new
+// keys indefinitely and the snapshot grows without bound over the process's
+// life. Zero or negative disables the cap (the default), preserving the
+// previous unbounded behavior for deployments that have not opted in.
+func WithMaxSnapshotEntries(n int) Option {
+	return func(s *Service) {
+		s.maxSnapshotEntries = n
+	}
+}
+
 // WithRefreshObserver registers a callback invoked once per refresh attempt,
 // covering both the initial load (including each retry in StartModeRetry) and
 // every periodic delta refresh. The outcome argument is one of the
@@ -299,7 +320,12 @@ func (s *Service) loadAllOnce(ctx context.Context) error {
 		s.notifyRefresh(RefreshOutcomeError)
 		return fmt.Errorf("identityconfig: load all: %w", err)
 	}
-	s.snap.Store(buildSnapshotWithLogger(s.logger, snap))
+	built := buildSnapshotWithLogger(s.logger, snap)
+	if s.maxSnapshotEntries > 0 && len(built.byKey) > s.maxSnapshotEntries {
+		s.notifyRefresh(RefreshOutcomeError)
+		return fmt.Errorf("identityconfig: initial snapshot of %d entries exceeds max snapshot entries %d", len(built.byKey), s.maxSnapshotEntries)
+	}
+	s.snap.Store(built)
 	s.notifyRefresh(RefreshOutcomeSuccess)
 	return nil
 }
@@ -395,7 +421,17 @@ func (s *Service) refreshDelta(ctx context.Context) error {
 		s.notifyRefresh(RefreshOutcomeSuccess)
 		return nil
 	}
-	s.snap.Store(applyDeltaWithLogger(s.logger, current, delta))
+	candidate := applyDeltaWithLogger(s.logger, current, delta)
+	if s.maxSnapshotEntries > 0 && len(candidate.byKey) > s.maxSnapshotEntries {
+		// Keep the prior snapshot entirely, including its watermark, so
+		// the next refresh re-requests from the old watermark and a later
+		// under-cap delta can still apply. The error outcome (observer)
+		// plus the refresh loop's warning make the rejection loud: a
+		// silently frozen snapshot is its own failure mode.
+		s.notifyRefresh(RefreshOutcomeError)
+		return fmt.Errorf("identityconfig: delta snapshot of %d entries exceeds max snapshot entries %d; keeping prior snapshot", len(candidate.byKey), s.maxSnapshotEntries)
+	}
+	s.snap.Store(candidate)
 	s.notifyRefresh(RefreshOutcomeSuccess)
 	return nil
 }

targeting/identityconfig/service_test.go

@@ -676,3 +676,84 @@ func TestService_LenReflectsSnapshot(t *testing.T) {
 	assert.Equal(t, 3, svc.Len(), "Len should still reflect every entry after an upsert+remove delta")
 	assert.Nil(t, svc.Get("https://seller.example/agent", "pkg-1"), "removed entry should not be readable")
 }
+
+func TestService_MaxSnapshotEntriesRejectsOverCapDelta(t *testing.T) {
+	src := newMemorySource()
+	src.put("https://seller.example/agent", "pkg-1", nil, time.Unix(100, 0))
+	src.put("https://seller.example/agent", "pkg-2", nil, time.Unix(100, 0))
+
+	// A delta that would grow the snapshot to 3 entries against a cap of 2.
+	src.queueDelta(Delta{
+		Upserted:      []Entry{{Key: Key{SellerAgentURL: "https://seller.example/agent", PackageID: "pkg-3"}, TargetSegments: nil}},
+		LastUpdatedAt: time.Unix(200, 0),
+	})
+	// A later under-cap delta: removes pkg-1, so the snapshot stays at 2.
+	src.queueDelta(Delta{
+		Removed:       []Key{{SellerAgentURL: "https://seller.example/agent", PackageID: "pkg-1"}},
+		LastUpdatedAt: time.Unix(300, 0),
+	})
+
+	rec := &outcomeRecorder{}
+	svc, err := New(src, 10*time.Millisecond,
+		WithMaxSnapshotEntries(2), WithRefreshObserver(rec.record))
+	require.NoError(t, err)
+	require.NoError(t, svc.Start(context.Background()))
+	defer svc.Stop()
+
+	require.Eventually(t, func() bool {
+		return slices.Contains(rec.snapshot(), RefreshOutcomeError)
+	}, time.Second, 5*time.Millisecond, "over-cap delta should report an error outcome")
+
+	// The prior snapshot is kept whole: the over-cap entry never lands.
+	// (Len is not asserted here — the next 10ms tick may already have
+	// applied the queued under-cap delta by the time this runs.)
+	_, present := svc.Lookup("https://seller.example/agent", "pkg-3")
+	assert.False(t, present, "over-cap upsert must not land")
+
+	// The service is not wedged: the later under-cap delta still applies,
+	// and it re-requests from the unadvanced watermark.
+	require.Eventually(t, func() bool {
+		_, present := svc.Lookup("https://seller.example/agent", "pkg-1")
+		return !present
+	}, time.Second, 5*time.Millisecond, "later under-cap delta should still apply")
+	assert.Equal(t, 1, svc.Len())
+}
+
+func TestService_MaxSnapshotEntriesAllowsExactlyAtCap(t *testing.T) {
+	src := newMemorySource()
+	src.put("https://seller.example/agent", "pkg-1", nil, time.Unix(100, 0))
+	src.queueDelta(Delta{
+		Upserted:      []Entry{{Key: Key{SellerAgentURL: "https://seller.example/agent", PackageID: "pkg-2"}, TargetSegments: nil}},
+		LastUpdatedAt: time.Unix(200, 0),
+	})
+
+	rec := &outcomeRecorder{}
+	svc, err := New(src, 10*time.Millisecond,
+		WithMaxSnapshotEntries(2), WithRefreshObserver(rec.record))
+	require.NoError(t, err)
+	require.NoError(t, svc.Start(context.Background()))
+	defer svc.Stop()
+
+	require.Eventually(t, func() bool {
+		_, present := svc.Lookup("https://seller.example/agent", "pkg-2")
+		return present
+	}, time.Second, 5*time.Millisecond, "delta landing exactly at the cap should install")
+	assert.Equal(t, 2, svc.Len())
+	assert.NotContains(t, rec.snapshot(), RefreshOutcomeError)
+}
+
+func TestService_MaxSnapshotEntriesRejectsOverCapInitialLoad(t *testing.T) {
+	src := newMemorySource()
+	src.put("https://seller.example/agent", "pkg-1", nil, time.Unix(100, 0))
+	src.put("https://seller.example/agent", "pkg-2", nil, time.Unix(100, 0))
+
+	rec := &outcomeRecorder{}
+	svc, err := New(src, time.Hour,
+		WithMaxSnapshotEntries(1), WithRefreshObserver(rec.record))
+	require.NoError(t, err)
+
+	require.Error(t, svc.Start(context.Background()),
+		"fail-fast initial load over the cap should fail loudly, not install a partial snapshot")
+	assert.Equal(t, []string{RefreshOutcomeError}, rec.snapshot())
+	assert.Equal(t, 0, svc.Len(), "no snapshot should be installed on a rejected initial load")
+}