From d79f7e545bdd477e9d8cdaeccc9a65c3e8a67f56 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Mon, 5 Oct 2026 02:50:40 +0100 Subject: [PATCH] Pin bulk registry metadata freshness and allocation ownership Retain the pre-change fresh-clone oracle for every derived field, incremental ingestion and detached reads. Expose repeated canonical derivation in a 1000-resource bulk read before altering production code. Refs: #2199 Contract-Neutral: Add independent regression controls and paired measurements; runtime and public contracts are unchanged. Change-source: pulse-maintainer --- .../registry_materialization_test.go | 192 ++++++++++++++++++ 1 file changed, 192 insertions(+) create mode 100644 internal/unifiedresources/registry_materialization_test.go diff --git a/internal/unifiedresources/registry_materialization_test.go b/internal/unifiedresources/registry_materialization_test.go new file mode 100644 index 0000000000..ff0b6feef0 --- /dev/null +++ b/internal/unifiedresources/registry_materialization_test.go @@ -0,0 +1,192 @@ +package unifiedresources + +import ( + "encoding/json" + "fmt" + "reflect" + "sync" + "testing" + "time" +) + +// This is the pre-materialization List implementation, retained independently +// of the optimized copy. It always derives metadata from the live resource. +func freshRegistryListForMetadataTest(rr *ResourceRegistry) []Resource { + rr.mu.RLock() + defer rr.mu.RUnlock() + out := make([]Resource, 0, len(rr.resources)) + for _, r := range rr.resources { + out = append(out, cloneResource(r)) + } + sortResourcesByName(out) + return out +} + +func assertRegistryMetadataOracle(t *testing.T, rr *ResourceRegistry) []Resource { + t.Helper() + actual := rr.List() + expected := freshRegistryListForMetadataTest(rr) + if !reflect.DeepEqual(actual, expected) { + for i := range expected { + if i >= len(actual) || !reflect.DeepEqual(actual[i], expected[i]) { + t.Fatalf("List differs from fresh clone at resource %q", expected[i].ID) + } + } + t.Fatal("List changed inventory") + } + actualJSON, err := json.Marshal(actual) + if err != nil { + t.Fatal(err) + } + expectedJSON, err := json.Marshal(expected) + if err != nil { + t.Fatal(err) + } + if string(actualJSON) != string(expectedJSON) { + t.Fatal("List changed canonical wire content") + } + return actual +} + +func metadataOracleFixture() []Resource { + now := time.Now().UTC() + return []Resource{ + {ID: "agent:lab", Type: ResourceTypeAgent, Name: "lab", LastSeen: now, Sources: []DataSource{SourceAgent, SourceProxmox}, Agent: &AgentData{AgentID: "lab", Hostname: "lab.example.test"}, Proxmox: &ProxmoxData{NodeName: "lab"}, Tags: []string{"PII"}, Identity: ResourceIdentity{IPAddresses: []string{"192.0.2.1"}}, SupersededCanonicalIDs: []string{"agent:old-lab"}}, + {ID: "vm:lab", Type: ResourceTypeVM, Name: "guest", LastSeen: now, Sources: []DataSource{SourceProxmox}, Proxmox: &ProxmoxData{VMID: 100, NodeName: "lab"}, Metrics: &ResourceMetrics{CPU: &MetricValue{Percent: 1.25, Source: SourceProxmox}}}, + {ID: "container:lab", Type: ResourceTypeSystemContainer, Name: "ct", LastSeen: now, Sources: []DataSource{SourceProxmox}, Proxmox: &ProxmoxData{VMID: 101, NodeName: "lab"}}, + {ID: "docker:lab", Type: ResourceTypeAgent, Name: "docker", LastSeen: now, Sources: []DataSource{SourceDocker}, Docker: &DockerData{Hostname: "docker.example.test", HostSourceID: "runtime"}}, + {ID: "app:lab", Type: ResourceTypeAppContainer, Name: "app", LastSeen: now, Sources: []DataSource{SourceDocker}, Docker: &DockerData{ContainerID: "container-id", HostSourceID: "runtime"}}, + {ID: "storage:lab", Type: ResourceTypeStorage, Name: "storage", LastSeen: now, Sources: []DataSource{SourceProxmox}, Storage: &StorageMeta{Path: "/srv/synthetic"}}, + {ID: "pbs:lab", Type: ResourceTypePBS, Name: "backup", LastSeen: now, Sources: []DataSource{SourcePBS}, PBS: &PBSData{InstanceID: "pbs-lab", Hostname: "backup.example.test"}}, + {ID: "pmg:lab", Type: ResourceTypePMG, Name: "mail", LastSeen: now, Sources: []DataSource{SourcePMG}, PMG: &PMGData{InstanceID: "pmg-lab", Hostname: "mail.example.test"}}, + {ID: "k8s:lab", Type: ResourceTypeK8sCluster, Name: "cluster", LastSeen: now, Sources: []DataSource{SourceK8s}, Kubernetes: &K8sData{ClusterID: "cluster-lab"}}, + {ID: "k8s:node", Type: ResourceTypeK8sNode, Name: "node", LastSeen: now, Sources: []DataSource{SourceK8s}, Kubernetes: &K8sData{ClusterID: "cluster-lab"}}, + {ID: "k8s:pod", Type: ResourceTypePod, Name: "pod", LastSeen: now, Sources: []DataSource{SourceK8s}, Kubernetes: &K8sData{ClusterID: "cluster-lab"}}, + {ID: "k8s:deploy", Type: ResourceTypeK8sDeployment, Name: "deploy", LastSeen: now, Sources: []DataSource{SourceK8s}, Kubernetes: &K8sData{ClusterID: "cluster-lab"}}, + {ID: "k8s:secret", Type: ResourceTypeK8sSecret, Name: "synthetic-secret-metadata", LastSeen: now, Sources: []DataSource{SourceK8s}}, + {ID: "k8s:role", Type: ResourceTypeK8sRole, Name: "role", LastSeen: now, Sources: []DataSource{SourceK8s}}, + {ID: "tn:lab", Type: ResourceTypeAgent, Name: "nas", LastSeen: now, Sources: []DataSource{SourceTrueNAS}, TrueNAS: &TrueNASData{Hostname: "nas.example.test"}}, + {ID: "vmware:lab", Type: ResourceTypeAgent, Name: "esxi", LastSeen: now, Sources: []DataSource{SourceVMware}, VMware: &VMwareData{ConnectionID: "vsphere-lab", ManagedObjectID: "host-1"}}, + {ID: "disk:lab", Type: ResourceTypePhysicalDisk, Name: "disk", LastSeen: now, Sources: []DataSource{SourceAgent}, PhysicalDisk: &PhysicalDiskMeta{}}, + {ID: "ceph:lab", Type: ResourceTypeCeph, Name: "ceph", LastSeen: now, Sources: []DataSource{SourceProxmox}, Ceph: &CephMeta{}}, + } +} + +func TestRegistryMaterializedMetadataMatchesFreshClones(t *testing.T) { + rr := NewRegistry(nil) + rr.IngestResources(metadataOracleFixture()) + original := assertRegistryMetadataOracle(t, rr) + if len(original) != 18 { + t.Fatalf("fixture lost resources: %d", len(original)) + } + // Low-level record ingestion is observable before its batch epilogue. It + // must invalidate metadata even when the typed cache was previously clean. + rr.Workloads() + id := rr.ingestRecord(SourceAgent, "lab", Resource{Type: ResourceTypeAgent, Name: "renamed", Status: StatusOnline, LastSeen: time.Now().UTC(), Agent: &AgentData{AgentID: "lab", Hostname: "renamed.example.test"}, Tags: []string{"public"}}, ResourceIdentity{}, false) + assertRegistryMetadataOracle(t, rr) + rr.retainSupersededCanonicalIDs(id, []string{"agent:new-retired-era"}) + assertRegistryMetadataOracle(t, rr) + rr.IngestRecords(SourceAgent, []IngestRecord{{SourceID: "lab", Resource: Resource{Type: ResourceTypeAgent, Name: "restricted", Status: StatusOffline, LastSeen: time.Now().UTC(), Tags: []string{"customer-data"}, Agent: &AgentData{AgentID: "lab", Hostname: "renamed.example.test"}}}}) + assertRegistryMetadataOracle(t, rr) + // Seeding normalized input must not trust caller-supplied policy or scopes. + rr.IngestResources([]Resource{{ID: "storage:lab", Type: ResourceTypeStorage, Name: "moved", Status: StatusOnline, LastSeen: time.Now().UTC(), Sources: []DataSource{SourceTrueNAS}, TrueNAS: &TrueNASData{Hostname: "new-nas.example.test"}, Tags: []string{"public"}, PlatformScopes: []string{"invented"}, AISafeSummary: "stale"}}) + assertRegistryMetadataOracle(t, rr) + rr.MarkStale(time.Now().UTC().Add(24*time.Hour), nil) + stale := assertRegistryMetadataOracle(t, rr) + if reflect.DeepEqual(stale, original) { + t.Fatal("mutation fixture did not change any output") + } +} + +func TestRegistryMaterializedMetadataDetachesDerivedFields(t *testing.T) { + rr := NewRegistry(nil) + rr.IngestResources(metadataOracleFixture()) + before := assertRegistryMetadataOracle(t, rr) + actual := rr.List() + for i := range actual { + r := &actual[i] + if r.Canonical != nil { + r.Canonical.DisplayName = "changed" + if len(r.Canonical.Aliases) > 0 { + r.Canonical.Aliases[0] = "changed" + } + if len(r.Canonical.SupersededIDs) > 0 { + r.Canonical.SupersededIDs[0] = "changed" + } + } + if r.Policy != nil { + r.Policy.Sensitivity = ResourceSensitivityPublic + if len(r.Policy.Routing.Redact) > 0 { + r.Policy.Routing.Redact[0] = "changed" + } + } + if len(r.PlatformScopes) > 0 { + r.PlatformScopes[0] = "changed" + } + } + if got := assertRegistryMetadataOracle(t, rr); !reflect.DeepEqual(got, before) { + t.Fatal("List result mutated stored canonical metadata") + } +} + +func TestRegistryMaterializedMetadataConcurrentReads(t *testing.T) { + rr := NewRegistry(nil) + rr.IngestResources(metadataOracleFixture()) + var wg sync.WaitGroup + for reader := 0; reader < 4; reader++ { + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 40; i++ { + for _, r := range rr.List() { + fresh := cloneResource(&r) + if !reflect.DeepEqual(fresh, r) { + t.Error("concurrent List retained stale derived metadata") + return + } + } + rr.Workloads() + } + }() + } + for i := 0; i < 40; i++ { + rr.IngestRecords(SourceAgent, []IngestRecord{{SourceID: "live", Resource: Resource{Type: ResourceTypeAgent, Name: fmt.Sprintf("host-%d", i), Status: StatusOnline, LastSeen: time.Now().UTC(), Tags: []string{"customer-data"}, Agent: &AgentData{AgentID: "live"}}}}) + } + wg.Wait() + assertRegistryMetadataOracle(t, rr) +} + +var materializedMetadataAllocationSink []Resource + +func TestRegistryMaterializedMetadataAllocationBudget(t *testing.T) { + rr := NewRegistry(nil) + rr.IngestSnapshot(benchmarkVMState(1000)) + assertRegistryMetadataOracle(t, rr) + before := testing.AllocsPerRun(3, func() { materializedMetadataAllocationSink = freshRegistryListForMetadataTest(rr) }) + after := testing.AllocsPerRun(3, func() { materializedMetadataAllocationSink = rr.List() }) + t.Logf("1000-resource bulk read allocations: always-refresh=%.0f materialized=%.0f", before, after) + if after >= before*0.7 { + t.Fatalf("bulk reads still repeat canonical derivation: %.0f >= 70%% of %.0f", after, before) + } +} + +// Paired same-process measurements are source costs, not fleet CPU/RSS. The +// immutable fresh-clone oracle and current implementation use the same input. +func BenchmarkRegistryMetadataBulkReads(b *testing.B) { + for _, count := range []int{1, 1000} { + rr := NewRegistry(nil) + rr.IngestSnapshot(benchmarkVMState(count)) + rr.List() + for _, read := range []struct { + name string + fn func() []Resource + }{{"fresh-clone", func() []Resource { return freshRegistryListForMetadataTest(rr) }}, {"materialized", rr.List}} { + b.Run(fmt.Sprintf("resources-%d/%s", count, read.name), func(b *testing.B) { + b.ReportAllocs() + for i := 0; i < b.N; i++ { + materializedMetadataAllocationSink = read.fn() + } + }) + } + } +}