diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 38491007a..9cc302b38 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -3335,6 +3335,18 @@ for provider-read breadcrumbs such as VMware tasks and events, plus the change model instead of introducing a second event shape, and `RecordChange` must stay idempotent by canonical change ID so poller refreshes and replayed supplemental snapshots do not duplicate resource history. +Change emission over registry rebuilds must diff relationships by edge +identity only: canonical source, canonical target, type, and active state, +order-insensitive (`relationshipsEquivalent` in `change_emission.go`). +Volatile provenance fields (`ObservedAt`, `LastSeenAt`, `Confidence`, +`Metadata`, `Discoverer`) refresh on every rebuild cycle and must never +count as a relationship change; comparing rebuilt slices structurally +emitted a no-op `relationship_change` row per relationship-bearing resource +per cycle and grew `resource_changes` without bound (issue #1496, demo +outage 2026-07-08). Retention pruning must also enforce a hard row cap on +`resource_changes` (`maxResourceChangesRows` in `store.go`) so a +pathological writer cannot grow the table unbounded inside the time-based +retention window. Action plans in `actions.go` still keep stale-plan protection to the canonical `resourceVersion`, `policyVersion`, and `planHash` fields, so stale execution checks stay in the shared resource action model rather than provider-local diff --git a/internal/unifiedresources/change_emission.go b/internal/unifiedresources/change_emission.go index dcd71bb5d..b1f1abb55 100644 --- a/internal/unifiedresources/change_emission.go +++ b/internal/unifiedresources/change_emission.go @@ -125,7 +125,7 @@ func buildResourceChange(before Resource, beforeOK bool, after Resource, afterOK change.From = resourceStateSummary(before) change.To = resourceStateSummary(after) change.Reason = "resource state changed" - case !equalStringPtr(before.ParentID, after.ParentID) || !reflect.DeepEqual(before.Relationships, after.Relationships): + case !equalStringPtr(before.ParentID, after.ParentID) || !relationshipsEquivalent(before.Relationships, after.Relationships): change.Kind = ChangeRelationship change.From = resourceRelationSummary(before) change.To = resourceRelationSummary(after) @@ -168,7 +168,7 @@ func resourceChangedFields(before, after Resource) []string { if !equalStringPtr(before.ParentID, after.ParentID) { changed = append(changed, "parentId") } - if !reflect.DeepEqual(before.Relationships, after.Relationships) { + if !relationshipsEquivalent(before.Relationships, after.Relationships) { changed = append(changed, "relationships") } if !reflect.DeepEqual(before.Capabilities, after.Capabilities) { @@ -205,6 +205,44 @@ func resourceChangedFields(before, after Resource) []string { return changed } +// relationshipsEquivalent reports whether two relationship sets describe the +// same edges. Registry rebuilds reconstruct every relationship with fresh +// ObservedAt/LastSeenAt stamps and metadata maps, so comparing with +// reflect.DeepEqual emitted a no-op relationship_change row for every +// relationship-bearing resource on every rebuild cycle (the unbounded +// unified_resources.db growth behind issue #1496). Only edge identity — +// canonical source, canonical target, type, and active state — counts as +// change. +func relationshipsEquivalent(a, b []ResourceRelationship) bool { + if len(a) != len(b) { + return false + } + if len(a) == 0 { + return true + } + edgeKeys := func(relationships []ResourceRelationship) []string { + keys := make([]string, 0, len(relationships)) + for _, relationship := range relationships { + keys = append(keys, strings.Join([]string{ + CanonicalResourceID(relationship.SourceID), + CanonicalResourceID(relationship.TargetID), + string(relationship.Type), + fmt.Sprintf("%t", relationship.Active), + }, "\x1f")) + } + sort.Strings(keys) + return keys + } + aKeys := edgeKeys(a) + bKeys := edgeKeys(b) + for i := range aKeys { + if aKeys[i] != bKeys[i] { + return false + } + } + return true +} + func dockerCommandChanged(before, after Resource) bool { var beforeCommand, afterCommand any if before.Docker != nil { diff --git a/internal/unifiedresources/change_emission_test.go b/internal/unifiedresources/change_emission_test.go index 83005f728..53b35e666 100644 --- a/internal/unifiedresources/change_emission_test.go +++ b/internal/unifiedresources/change_emission_test.go @@ -18,6 +18,90 @@ func TestBuildResourceChange_ReturnsNilWhenUnchanged(t *testing.T) { } } +func TestBuildResourceChange_IgnoresVolatileRelationshipFields(t *testing.T) { + baseTime := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC) + before := Resource{ + ID: "app-container:1", + Type: ResourceTypeAppContainer, + Name: "web", + Status: StatusOnline, + Relationships: []ResourceRelationship{ + { + SourceID: "app-container:1", TargetID: "docker-network:1", Type: RelAttachedTo, + Confidence: 1, Active: true, Discoverer: "docker_adapter", + ObservedAt: baseTime, LastSeenAt: baseTime, + Metadata: map[string]any{"network": "bridge"}, + }, + { + SourceID: "app-container:1", TargetID: "agent:1", Type: RelRunsOn, + Confidence: 1, Active: true, Discoverer: "docker_adapter", + ObservedAt: baseTime, LastSeenAt: baseTime, + }, + }, + } + // Registry rebuilds reconstruct the same edges with fresh timestamps, + // fresh metadata maps, and possibly different ordering. + after := before + rebuiltTime := baseTime.Add(time.Hour) + after.Relationships = []ResourceRelationship{ + { + SourceID: "app-container:1", TargetID: "agent:1", Type: RelRunsOn, + Confidence: 1, Active: true, Discoverer: "docker_adapter", + ObservedAt: rebuiltTime, LastSeenAt: rebuiltTime, + }, + { + SourceID: "app-container:1", TargetID: "docker-network:1", Type: RelAttachedTo, + Confidence: 1, Active: true, Discoverer: "docker_adapter", + ObservedAt: rebuiltTime, LastSeenAt: rebuiltTime, + Metadata: map[string]any{"network": "bridge"}, + }, + } + + if change := buildResourceChange(before, true, after, true, rebuiltTime, nil, SourcePulseDiff, ""); change != nil { + t.Fatalf("expected nil change for rebuilt-but-identical relationships, got %+v", change) + } +} + +func TestBuildResourceChange_DetectsRealRelationshipChanges(t *testing.T) { + edge := ResourceRelationship{ + SourceID: "app-container:1", TargetID: "docker-network:1", Type: RelAttachedTo, + Confidence: 1, Active: true, Discoverer: "docker_adapter", + } + before := Resource{ + ID: "app-container:1", + Type: ResourceTypeAppContainer, + Name: "web", + Status: StatusOnline, + Relationships: []ResourceRelationship{edge}, + } + + retargeted := edge + retargeted.TargetID = "docker-network:2" + deactivated := edge + deactivated.Active = false + + cases := []struct { + name string + after []ResourceRelationship + }{ + {"edge added", []ResourceRelationship{edge, {SourceID: "app-container:1", TargetID: "agent:1", Type: RelRunsOn, Confidence: 1, Active: true}}}, + {"edge removed", nil}, + {"edge retargeted", []ResourceRelationship{retargeted}}, + {"edge deactivated", []ResourceRelationship{deactivated}}, + } + for _, tc := range cases { + after := before + after.Relationships = tc.after + change := buildResourceChange(before, true, after, true, time.Now().UTC(), nil, SourcePulseDiff, "") + if change == nil { + t.Fatalf("%s: expected relationship change, got nil", tc.name) + } + if change.Kind != ChangeRelationship { + t.Fatalf("%s: Kind = %q, want %q", tc.name, change.Kind, ChangeRelationship) + } + } +} + func TestBuildResourceChange_ClassifiesStateTransition(t *testing.T) { before := Resource{ ID: "vm:1", diff --git a/internal/unifiedresources/code_standards_test.go b/internal/unifiedresources/code_standards_test.go index ee1c7962c..f0ea38d42 100644 --- a/internal/unifiedresources/code_standards_test.go +++ b/internal/unifiedresources/code_standards_test.go @@ -1153,8 +1153,9 @@ func TestResourceChangeEmissionCoversRelationshipAndCapabilityChanges(t *testing "change.RelatedResources = relatedResourceIDs(change.ResourceID, before, after)", "case resourceRestartChanged(before, after):", "case resourceIncidentChanged(before, after):", - "if !reflect.DeepEqual(before.Relationships, after.Relationships) {", + "if !relationshipsEquivalent(before.Relationships, after.Relationships) {", "changed = append(changed, \"relationships\")", + "func relationshipsEquivalent(a, b []ResourceRelationship) bool {", "if resourceIncidentChanged(before, after) {", "changed = append(changed, \"incidents\")", "if dockerRestartChanged(before, after) {", diff --git a/internal/unifiedresources/store.go b/internal/unifiedresources/store.go index 83293039e..61b69eaae 100644 --- a/internal/unifiedresources/store.go +++ b/internal/unifiedresources/store.go @@ -955,6 +955,12 @@ const ( retentionInterval = 1 * time.Hour initialRetentionDelay = 30 * time.Second maxUnifiedReclaimPages = 50000 + // maxResourceChangesRows bounds resource_changes even inside the + // retention window: time-based pruning alone cannot contain a + // pathological writer (the demo hit 1.6M rows in three days), and an + // unbounded table starves the store's single connection until prunes + // themselves fail with SQLITE_BUSY. + maxResourceChangesRows = 200000 ) // migrateAutoVacuum ensures the database uses incremental auto-vacuum so that @@ -1038,6 +1044,24 @@ func (s *SQLiteResourceStore) startRetentionLoop() chan struct{} { return stop } +// capResourceChanges deletes the oldest resource_changes rows beyond limit, +// keeping the newest rows by observed_at. +func (s *SQLiteResourceStore) capResourceChanges(limit int) (int64, error) { + res, err := s.db.Exec( + `DELETE FROM resource_changes WHERE rowid IN ( + SELECT rowid FROM resource_changes + ORDER BY observed_at DESC + LIMIT -1 OFFSET ? + )`, + limit, + ) + if err != nil { + return 0, err + } + affected, _ := res.RowsAffected() + return affected, nil +} + func (s *SQLiteResourceStore) pruneOldRecords() { now := time.Now() changesCutoff := now.Add(-resourceChangesRetention) @@ -1059,6 +1083,12 @@ func (s *SQLiteResourceStore) pruneOldRecords() { totalDeleted += affected } + if affected, err := s.capResourceChanges(maxResourceChangesRows); err != nil { + log.Printf("unifiedresources: failed to cap resource_changes: %v", err) + } else if affected > 0 { + totalDeleted += affected + } + res, err = s.db.Exec( `DELETE FROM action_audits WHERE created_at < ?`, auditsCutoff.UTC().Format(tsFmt), diff --git a/internal/unifiedresources/store_test.go b/internal/unifiedresources/store_test.go index 880d9aa4e..6e411d7f6 100644 --- a/internal/unifiedresources/store_test.go +++ b/internal/unifiedresources/store_test.go @@ -2498,3 +2498,45 @@ func TestPruneOldRecords_DeletesExpiredChangesAndAudits(t *testing.T) { t.Fatalf("expected only audit-recent to survive, got %d: %+v", len(auditResults), auditResults) } } + +func TestCapResourceChanges_KeepsNewestRows(t *testing.T) { + store := newTestStore(t) + + base := time.Now().Add(-10 * time.Hour) + for i := 0; i < 10; i++ { + change := ResourceChange{ + ID: fmt.Sprintf("change-%d", i), + ObservedAt: base.Add(time.Duration(i) * time.Hour), + ResourceID: "vm:100", + Kind: ChangeStateTransition, + SourceType: SourcePulseDiff, + Confidence: ConfidenceHigh, + } + if err := store.RecordChange(change); err != nil { + t.Fatalf("RecordChange(%s): %v", change.ID, err) + } + } + + deleted, err := store.capResourceChanges(3) + if err != nil { + t.Fatalf("capResourceChanges: %v", err) + } + if deleted != 7 { + t.Fatalf("deleted = %d, want 7", deleted) + } + + remaining, err := store.GetRecentChanges("vm:100", time.Time{}, 0) + if err != nil { + t.Fatalf("GetRecentChanges after cap: %v", err) + } + if len(remaining) != 3 { + t.Fatalf("expected 3 surviving rows, got %d", len(remaining)) + } + for _, change := range remaining { + switch change.ID { + case "change-7", "change-8", "change-9": + default: + t.Fatalf("unexpected survivor %s; want the newest three", change.ID) + } + } +}