Integrate reviewed allocation-bounded snapshot and metadata maintenance

Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
2026-09-30 14:00:19 +01:00
9 changed files with 373 additions and 72 deletions
@@ -23,6 +23,25 @@ and sort the complete canonical change table while startup and ingestion wait.
## Purpose
### Allocation-bounded snapshot and clone metadata — issue #2199
WebSocket snapshot extraction decodes every entry's ID from the encoded array
in one batched pass, preserving standard JSON identity semantics rather than
trusting source-ID hints after checking only the first entry. Missing/duplicate
IDs still reject resource keying; infrastructure and alert arrays retain the
existing whole-field fallback. The decoder-owned raw entry buffers are retained
without a redundant copy, and must remain isolated from the original encoding
and from each other. Per-entry reference decoding, malformed/escaped/duplicate
identity fuzz cases and encoded tail-change deltas pin these boundaries.
Resource clones still refresh canonical metadata on every read, including after
source, facet or tag edits. Platform membership uses a deduplicated small slice
with the existing canonical order; sensitivity scans normalized tags once,
preserving explicit public, restricted, sensitive and resource-type precedence.
The exhaustive source-combination test and clone scope/policy refresh test
protect page admission, policy routing and clone isolation. Component benchmarks
measure allocation/time changes, not installed fleet CPU or write-rate acceptance.
### Bounded generation change comparison — issue #2199
Accepted snapshot rebuilds compare the previous and replacement registry
+50
View File
@@ -1,6 +1,7 @@
package unifiedresources
import (
"reflect"
"testing"
"time"
@@ -9,6 +10,55 @@ import (
"github.com/rcourtman/pulse-go-rewrite/pkg/tlsutil"
)
func TestCloneResourceRefreshesScopeAndPolicyMetadata(t *testing.T) {
// Registry mutation paths can leave derived metadata stale until the next
// read. Keep refresh on every clone, including after source/tag edits.
original := &Resource{
ID: "app-container:lab", Type: ResourceTypeAppContainer,
Sources: []DataSource{SourceDocker, SourceAgent, SourceDocker},
PlatformScopes: []string{"stale"},
Agent: &AgentData{Hostname: "lab"},
Docker: &DockerData{HostSourceID: "proxmox-lxc-docker:lab:node:100"},
Policy: &ResourcePolicy{Sensitivity: ResourceSensitivityRestricted},
AISafeSummary: "stale",
}
for _, test := range []struct {
name string
tags []string
sensitivity ResourceSensitivity
routing ResourceRoutingScope
}{
{"public-last", []string{"backup", "PII", " PUBLIC "}, ResourceSensitivityPublic, ResourceRoutingScopeCloudSummary},
{"public-first", []string{"public", "restricted", "database"}, ResourceSensitivityPublic, ResourceRoutingScopeCloudSummary},
{"restricted-last", []string{"backup", " Customer_Data "}, ResourceSensitivityRestricted, ResourceRoutingScopeLocalOnly},
{"restricted-first", []string{" SECRETS ", "storage"}, ResourceSensitivityRestricted, ResourceRoutingScopeLocalOnly},
{"sensitive", []string{"", "unrecognised", " DaTaBaSe ", "database"}, ResourceSensitivitySensitive, ResourceRoutingScopeLocalFirst},
{"internal", []string{"prod", ""}, ResourceSensitivityInternal, ResourceRoutingScopeCloudSummary},
} {
t.Run(test.name, func(t *testing.T) {
original.Tags = test.tags
cloned := cloneResource(original)
assertStringSliceEqual(t, cloned.PlatformScopes, []string{"agent", "proxmox-pve", "docker"})
if cloned.Policy.Sensitivity != test.sensitivity || cloned.Policy.Routing.Scope != test.routing || cloned.AISafeSummary == "stale" {
t.Fatalf("clone retained stale policy: %#v, summary=%q", cloned.Policy, cloned.AISafeSummary)
}
cloned.PlatformScopes[0] = "mutated"
cloned.Policy.Sensitivity = ResourceSensitivityPublic
if !reflect.DeepEqual(original.PlatformScopes, []string{"stale"}) || original.Policy.Sensitivity != ResourceSensitivityRestricted || original.AISafeSummary != "stale" {
t.Fatal("clone modified the original derived metadata")
}
})
}
original.Sources = []DataSource{SourceTrueNAS, SourceAgent}
original.TrueNAS = &TrueNASData{Hostname: "lab"}
original.Docker = nil
cloned := cloneResource(original)
assertStringSliceEqual(t, cloned.PlatformScopes, []string{"agent", "truenas"})
if cloned.Policy.Sensitivity != ResourceSensitivitySensitive {
t.Fatalf("edited provider facet lost its default sensitivity: %#v", cloned.Policy)
}
}
// --- cloneResource: top-level isolation ---
func TestCloneResource_Nil(t *testing.T) {
@@ -0,0 +1,30 @@
package unifiedresources
import "testing"
var benchmarkSensitivity ResourceSensitivity
var benchmarkScopes []string
// Component measurements of the per-read metadata path, not installed fleet
// CPU/write attribution or release qualification thresholds.
func BenchmarkRefreshPlatformScopes(b *testing.B) {
resource := Resource{
Type: ResourceTypeAppContainer, Sources: []DataSource{SourceDocker, SourceAgent},
Agent: &AgentData{}, Docker: &DockerData{HostSourceID: "proxmox-lxc-docker:lab:node:100"},
}
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
RefreshPlatformScopes(&resource)
}
benchmarkScopes = resource.PlatformScopes
}
func BenchmarkClassifyResourceSensitivity(b *testing.B) {
resource := Resource{Type: ResourceTypeVM, Tags: []string{"prod", "web", "database", "customer-data"}}
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
benchmarkSensitivity = classifyResourceSensitivity(resource)
}
}
+42 -35
View File
@@ -28,21 +28,35 @@ func RefreshPlatformScopes(resource *Resource) {
return
}
scopes := make(map[string]struct{}, len(resource.Sources)+2)
// Clones refresh scopes on registry reads as well as writes. These small
// membership sets do not need a map on every resource read.
scopes := make([]string, 0, 4)
add := func(scope string) {
scope = strings.ToLower(strings.TrimSpace(scope))
if scope == "" {
return
}
for _, existing := range scopes {
if existing == scope {
return
}
}
scopes = append(scopes, scope)
}
for _, source := range resource.Sources {
addPlatformScope(scopes, platformScopeForSource(source))
add(platformScopeForSource(source))
}
addPlatformScopesForFacets(scopes, *resource)
addPlatformScopesForFacets(add, *resource)
if shouldAddDockerPlatformScope(*resource) {
addPlatformScope(scopes, "docker")
add("docker")
}
if resource.Docker != nil {
hostSourceID := strings.TrimSpace(resource.Docker.HostSourceID)
if strings.HasPrefix(hostSourceID, proxmoxLXCDockerHostSourcePrefix) {
addPlatformScope(scopes, "proxmox-pve")
addPlatformScope(scopes, "docker")
add("proxmox-pve")
add("docker")
}
}
@@ -74,30 +88,30 @@ func platformScopeForSource(source DataSource) string {
}
}
func addPlatformScopesForFacets(scopes map[string]struct{}, resource Resource) {
func addPlatformScopesForFacets(add func(string), resource Resource) {
if resource.Agent != nil {
addPlatformScope(scopes, "agent")
add("agent")
}
if resource.TrueNAS != nil {
addPlatformScope(scopes, "truenas")
add("truenas")
}
if resource.Proxmox != nil {
addPlatformScope(scopes, "proxmox-pve")
add("proxmox-pve")
}
if resource.PBS != nil {
addPlatformScope(scopes, "proxmox-pbs")
add("proxmox-pbs")
}
if resource.PMG != nil {
addPlatformScope(scopes, "proxmox-pmg")
add("proxmox-pmg")
}
if resource.Kubernetes != nil {
addPlatformScope(scopes, "kubernetes")
add("kubernetes")
}
if resource.VMware != nil {
addPlatformScope(scopes, "vmware-vsphere")
add("vmware-vsphere")
}
if len(AvailabilityChecksForResource(resource)) > 0 || CanonicalResourceType(resource.Type) == ResourceTypeNetworkEndpoint {
addPlatformScope(scopes, "availability")
add("availability")
}
}
@@ -111,29 +125,22 @@ func shouldAddDockerPlatformScope(resource Resource) bool {
return true
}
func addPlatformScope(scopes map[string]struct{}, scope string) {
scope = strings.ToLower(strings.TrimSpace(scope))
if scope == "" {
return
}
scopes[scope] = struct{}{}
}
func orderPlatformScopes(scopes map[string]struct{}) []string {
// orderPlatformScopes consumes the deduplicated slice, placing known scopes in
// canonical order and sorting any remaining scopes alphabetically in place.
func orderPlatformScopes(scopes []string) []string {
if len(scopes) == 0 {
return nil
}
out := make([]string, 0, len(scopes))
for _, scope := range canonicalPlatformScopeOrder {
if _, ok := scopes[scope]; ok {
out = append(out, scope)
delete(scopes, scope)
knownCount := 0
for _, canonical := range canonicalPlatformScopeOrder {
for i := knownCount; i < len(scopes); i++ {
if scopes[i] == canonical {
scopes[knownCount], scopes[i] = scopes[i], scopes[knownCount]
knownCount++
break
}
}
}
unknownStart := len(out)
for scope := range scopes {
out = append(out, scope)
}
sort.Strings(out[unknownStart:])
return out
sort.Strings(scopes[knownCount:])
return scopes
}
@@ -54,6 +54,32 @@ func TestRefreshPlatformScopes_TrueNASAppContainerKeepsOwningPlatform(t *testing
assertStringSliceEqual(t, resource.PlatformScopes, []string{"truenas"})
}
func TestRefreshPlatformScopesAllSourceCombinations(t *testing.T) {
// Include every membership combination, reverse input ordering, duplicates
// and unknown sources: changing the membership representation must not
// change page admission or canonical ordering.
sources := []DataSource{SourceAgent, SourceTrueNAS, SourceProxmox, SourcePBS, SourcePMG, SourceDocker, SourceK8s, SourceVMware, SourceAvailability}
for mask := 0; mask < 1<<len(sources); mask++ {
resource := Resource{Type: ResourceTypeVM, Sources: []DataSource{"unknown"}, PlatformScopes: []string{"stale"}}
var want []string
for i, source := range sources {
if mask&(1<<i) != 0 {
want = append(want, platformScopeForSource(source))
}
}
for i := len(sources) - 1; i >= 0; i-- {
if mask&(1<<i) != 0 {
resource.Sources = append(resource.Sources, sources[i], sources[i])
}
}
RefreshPlatformScopes(&resource)
assertStringSliceEqual(t, resource.PlatformScopes, want)
if mask == 0 && resource.PlatformScopes != nil {
t.Fatal("absent membership should remain nil")
}
}
}
func assertStringSliceEqual(t *testing.T, got, want []string) {
t.Helper()
if len(got) != len(want) {
+18 -24
View File
@@ -113,18 +113,24 @@ func RefreshCanonicalMetadataSlice(resources []Resource) []Resource {
}
func classifyResourceSensitivity(resource Resource) ResourceSensitivity {
tagSet := normalizedTagSet(resource.Tags)
switch {
case tagSet["public"]:
return ResourceSensitivityPublic
case tagSet["restricted"] || tagSet["customer-data"] || tagSet["customer_data"] ||
tagSet["pii"] || tagSet["phi"] || tagSet["pci"] || tagSet["regulated"] ||
tagSet["secret"] || tagSet["secrets"]:
return ResourceSensitivityRestricted
case tagSet["sensitive"] || tagSet["backup"] || tagSet["mail"] ||
tagSet["storage"] || tagSet["database"] || tagSet["dataset"]:
return ResourceSensitivitySensitive
// Clones classify every registry read. Scan once without building a tag
// map, preserving public > restricted > sensitive precedence independently
// of tag order, and normalising each tag only once.
tagSensitivity := ResourceSensitivityInternal
for _, tag := range resource.Tags {
switch strings.ToLower(strings.TrimSpace(tag)) {
case "public":
return ResourceSensitivityPublic
case "restricted", "customer-data", "customer_data", "pii", "phi", "pci", "regulated", "secret", "secrets":
tagSensitivity = ResourceSensitivityRestricted
case "sensitive", "backup", "mail", "storage", "database", "dataset":
if tagSensitivity != ResourceSensitivityRestricted {
tagSensitivity = ResourceSensitivitySensitive
}
}
}
if tagSensitivity != ResourceSensitivityInternal {
return tagSensitivity
}
if resource.PMG != nil {
@@ -367,18 +373,6 @@ func resourceSummaryType(resource Resource) string {
}
}
func normalizedTagSet(tags []string) map[string]bool {
out := make(map[string]bool, len(tags))
for _, tag := range tags {
trimmed := strings.ToLower(strings.TrimSpace(tag))
if trimmed == "" {
continue
}
out[trimmed] = true
}
return out
}
func resourceHasHostname(resource Resource) bool {
if canonicalHostname(resource) != "" {
return true
@@ -5,6 +5,33 @@ import (
"testing"
)
func TestClassifyResourceSensitivityTagAliasesAndPrecedence(t *testing.T) {
for _, group := range []struct {
tags []string
want ResourceSensitivity
}{
{[]string{"restricted", "customer-data", "customer_data", "pii", "phi", "pci", "regulated", "secret", "secrets"}, ResourceSensitivityRestricted},
{[]string{"sensitive", "backup", "mail", "storage", "database", "dataset"}, ResourceSensitivitySensitive},
} {
for _, tag := range group.tags {
t.Run(tag, func(t *testing.T) {
normalised := " " + strings.ToUpper(tag) + " "
for _, tags := range [][]string{{normalised, "unknown", normalised}, {"unknown", normalised}} {
if got := classifyResourceSensitivity(Resource{Type: ResourceTypeVM, Tags: tags}); got != group.want {
t.Fatalf("tags %v sensitivity = %q, want %q", tags, got, group.want)
}
}
for _, tags := range [][]string{{" PUBLIC ", normalised}, {normalised, "public"}} {
// Explicit public wins even over secret-type defaults, as before.
if got := classifyResourceSensitivity(Resource{Type: ResourceTypeK8sSecret, Tags: tags}); got != ResourceSensitivityPublic {
t.Fatalf("public precedence changed for %v: %q", tags, got)
}
}
})
}
}
}
func TestRefreshPolicyMetadata_ClassifiesRestrictedResources(t *testing.T) {
resource := Resource{
ID: "vm-100",
+21 -13
View File
@@ -56,23 +56,31 @@ func extractKeyedEntries(
if err := json.Unmarshal(encoded, &entries); err != nil {
return nil, nil, fmt.Errorf("decode state %s: %w", field, err)
}
byID := make(map[string]json.RawMessage)
// Decode all identities in one pass rather than allocating a decoder and
// throwaway struct for every entry. Keys must come from the encoded entries,
// not hints from the source value: checking one hinted ID cannot establish
// the identity of the rest, especially if a marshaler changes their order.
identities := make([]struct {
ID string `json:"id"`
}, len(entries))
if err := json.Unmarshal(encoded, &identities); err != nil {
return nil, nil, fmt.Errorf("decode state %s identity: %w", field, err)
}
byID := make(map[string]json.RawMessage, len(entries))
order := make([]string, 0, len(entries))
for _, entry := range entries {
var identity struct {
ID string `json:"id"`
}
if err := json.Unmarshal(entry, &identity); err != nil {
return nil, nil, fmt.Errorf("decode state %s identity: %w", field, err)
}
if identity.ID == "" {
for i, entry := range entries {
id := identities[i].ID
if id == "" {
return nil, nil, fmt.Errorf("state %s entry is missing id", field)
}
if _, exists := byID[identity.ID]; exists {
return nil, nil, fmt.Errorf("state %s id %q is duplicated", field, identity.ID)
if _, exists := byID[id]; exists {
return nil, nil, fmt.Errorf("state %s id %q is duplicated", field, id)
}
byID[identity.ID] = append(json.RawMessage(nil), entry...)
order = append(order, identity.ID)
// RawMessage.UnmarshalJSON already copied each entry into its own buffer.
// Retaining that buffer is safe; copying it a second time is redundant.
byID[id] = entry
order = append(order, id)
}
return byID, order, nil
}
+140
View File
@@ -1,13 +1,153 @@
package websocket
import (
"bytes"
"encoding/json"
"fmt"
"reflect"
"strings"
"testing"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
)
// Keep the previous per-entry decoder as an independent compatibility oracle.
// This includes encoding/json's escaped, case-insensitive and duplicate-key
// handling instead of assuming the first JSON field is the authoritative ID.
func extractKeyedEntriesIndividually(encoded json.RawMessage, field string) (map[string]json.RawMessage, []string, error) {
var entries []json.RawMessage
if err := json.Unmarshal(encoded, &entries); err != nil {
return nil, nil, err
}
byID := make(map[string]json.RawMessage)
order := make([]string, 0, len(entries))
for _, entry := range entries {
var identity struct {
ID string `json:"id"`
}
if err := json.Unmarshal(entry, &identity); err != nil {
return nil, nil, err
}
if identity.ID == "" {
return nil, nil, fmt.Errorf("state %s entry is missing id", field)
}
if _, exists := byID[identity.ID]; exists {
return nil, nil, fmt.Errorf("state %s id %q is duplicated", field, identity.ID)
}
byID[identity.ID] = append(json.RawMessage(nil), entry...)
order = append(order, identity.ID)
}
return byID, order, nil
}
func assertKeyedEntriesMatchPerEntryDecoding(t *testing.T, encoded []byte) {
t.Helper()
wantEntries, wantOrder, wantErr := extractKeyedEntriesIndividually(encoded, "resource")
gotEntries, gotOrder, gotErr := extractKeyedEntries(encoded, "resource")
if (wantErr == nil) != (gotErr == nil) {
t.Fatalf("error changed: individual=%v batch=%v; JSON=%q", wantErr, gotErr, encoded)
}
if wantErr == nil && (!reflect.DeepEqual(gotEntries, wantEntries) || !reflect.DeepEqual(gotOrder, wantOrder)) {
t.Fatalf("identity/order/payload changed: individual=%v/%v batch=%v/%v; JSON=%q", wantEntries, wantOrder, gotEntries, gotOrder, encoded)
}
}
func FuzzExtractKeyedEntriesMatchesPerEntryDecoding(f *testing.F) {
for _, encoded := range []string{
`null`, `[]`, `[{"id":"r-1"},{"id":"r-2"}]`,
`[{"name":"before-id","id":"尾\\\"\n\u0026","nested":{"id":"not-the-key"}}]`,
`[{"id":"r-1"},{"id":"old","id":"r-3"}]`,
`[{"ID":"case-insensitive"}]`, `[{"i\u0064":"escaped-key"}]`,
`[{"id":"r-1"},{"id":"r-1"}]`, `[{"id":"r-1"},{}]`,
`[null]`, `[{"id":null}]`, `[{"id":42}]`, `[42]`, `{}`, `[{`,
} {
f.Add([]byte(encoded))
}
f.Fuzz(assertKeyedEntriesMatchPerEntryDecoding)
}
func TestExtractKeyedEntriesOwnsEncodedBuffers(t *testing.T) {
encoded := []byte(`[{"id":"r-1","name":"alpha"},{"id":"r-2","name":"beta"}]`)
entries, order, err := extractKeyedEntries(encoded, "resource")
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(order, []string{"r-1", "r-2"}) {
t.Fatalf("order = %v", order)
}
for i := range encoded {
encoded[i] = 'x'
}
if !bytes.Equal(entries["r-1"], []byte(`{"id":"r-1","name":"alpha"}`)) ||
!bytes.Equal(entries["r-2"], []byte(`{"id":"r-2","name":"beta"}`)) {
t.Fatal("retained entries alias the original encoded buffer")
}
entries["r-1"][0] = 'x'
if entries["r-2"][0] != '{' {
t.Fatal("separate keyed entries alias each other")
}
}
// A custom marshaler may preserve the first ID while replacing/reordering the
// tail. The encoded IDs, not a sampled check of source IDs, own delta identity.
type encodedStateFixture struct{ encoded json.RawMessage }
func (s encodedStateFixture) MarshalJSON() ([]byte, error) { return s.encoded, nil }
func TestSnapshotDeltaUsesEveryEncodedTailIdentity(t *testing.T) {
previous, err := buildClientStateSnapshot(encodedStateFixture{json.RawMessage(`{
"resources":[{"id":"r-1"},{"id":"r-2"}],
"connectedInfrastructure":[{"id":"i-1"},{"id":"i-2"}],
"activeAlerts":[{"id":"a-1"},{"id":"a-2"}]
}`)})
if err != nil {
t.Fatal(err)
}
current, err := buildClientStateSnapshot(encodedStateFixture{json.RawMessage(`{
"resources":[{"id":"r-1"},{"id":"r-3"}],
"connectedInfrastructure":[{"id":"i-1"},{"id":"i-3"}],
"activeAlerts":[{"id":"a-1"},{"id":"a-3"}]
}`)})
if err != nil {
t.Fatal(err)
}
delta, err := buildClientStateDelta(previous, current)
if err != nil {
t.Fatal(err)
}
for field, prefix := range map[string]string{
resourceDeltaField: "r", infrastructureDeltaField: "i", activeAlertsDeltaField: "a",
} {
payload, ok := delta[field].(resourceDeltaPayload)
if !ok || !reflect.DeepEqual(payload.Removed, []string{prefix + "-2"}) ||
!reflect.DeepEqual(payload.Order, []string{prefix + "-1", prefix + "-3"}) ||
len(payload.Upserts) != 1 || string(payload.Upserts[0]) != `{"id":"`+prefix+`-3"}` {
t.Fatalf("%s: wrong encoded tail identity: %#v", field, delta[field])
}
}
}
func TestFrontendSnapshotValuePointerAndGenericEncodingAgree(t *testing.T) {
state := benchmarkFrontendState(100)
state.Resources[99].ID = "尾<&\"\\\n"
state.ConnectedInfrastructure = []models.ConnectedInfrastructureItemFrontend{{ID: "i-1"}, {ID: "i-2"}}
state.ActiveAlerts = []models.Alert{{ID: "a-1"}, {ID: "a-2"}}
encoded, err := json.Marshal(state)
if err != nil {
t.Fatal(err)
}
want, err := buildClientStateSnapshot(encodedStateFixture{encoded})
if err != nil {
t.Fatal(err)
}
for _, value := range []interface{}{state, &state} {
got, err := buildClientStateSnapshot(value)
if err != nil || !reflect.DeepEqual(got, want) {
t.Fatalf("%T snapshot differs from its encoding: %v", value, err)
}
}
}
func TestClientStateDeltaOmitsUnchangedResourcePayload(t *testing.T) {
previousState := models.EmptyStateFrontend()
previousState.LastUpdate = 100