Stabilise label-derived tags without hiding metadata changes

Docker and Kubernetes label maps became ordered tag arrays in random map iteration order, forcing unchanged resources into connected-dashboard deltas (#2441, found while profiling #2199). Sort only the deduplicated label-derived list. Preserve existing label values, source maps, first-choice image references and identity ordering. Exercise all seven affected Docker adapters and twenty-six Kubernetes kinds through registry rebuilds, monitor projection, per-client queues and receiver replay, including real replace/add/remove/clear/restore changes, tenant isolation and a thousand-pod heartbeat.

Contract-Neutral: Deterministic ordering of existing label-derived tag sets only; no public JSON shape, canonical ownership or ordered identity-list change.
Change-source: pulse-maintainer

internal/unifiedresources/adapters.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"net"
 	"net/url"
+	"slices"
 	"strconv"
 	"strings"
 	"time"
@@ -3781,7 +3782,12 @@ func labelsToTags(labels map[string]string) []string {
 		}
 		out = append(out, key+":"+v)
 	}
-	return uniqueStrings(out)
+	// Labels are a set, but tags are sent as an ordered JSON array. Stabilise
+	// only this map-derived list so rebuilds do not create spurious socket
+	// deltas. uniqueStrings also serves ordered identity lists; do not sort it.
+	out = uniqueStrings(out)
+	slices.Sort(out)
+	return out
 }
 
 func convertInterfaces(interfaces []models.HostNetworkInterface) []NetworkInterface {

internal/unifiedresources/label_tags_test.go

@@ -0,0 +1,67 @@
+package unifiedresources
+
+import (
+	"maps"
+	"reflect"
+	"testing"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/models"
+)
+
+func TestLabelTagsStableAcrossMapIterations(t *testing.T) {
+	labels := map[string]string{
+		"app": "api", "env": "production", "team": "ops", "tier": "backend",
+		"region": "west", "release": "stable", "managed": "yes", "version": "1",
+	}
+	before := maps.Clone(labels)
+	want := []string{"app:api", "env:production", "managed:yes", "region:west", "release:stable", "team:ops", "tier:backend", "version:1"}
+	for i := 0; i < 256; i++ {
+		if got := labelsToTags(labels); !reflect.DeepEqual(got, want) {
+			t.Fatalf("iteration %d: tags = %v, want stable %v", i, got, want)
+		}
+	}
+	if !reflect.DeepEqual(labels, before) {
+		t.Fatal("tag conversion mutated source labels")
+	}
+}
+
+func TestLabelTagsRetainExistingValueSemantics(t *testing.T) {
+	for _, tc := range []struct {
+		name   string
+		labels map[string]string
+		want   []string
+	}{
+		{"nil", nil, nil},
+		{"empty", map[string]string{}, nil},
+		{"blank-key", map[string]string{" ": "ignored"}, []string{}},
+		{"bare-keys", map[string]string{" z ": "\t", "a": ""}, []string{"a", "z"}},
+		{"deduplicated-trimmed-keys", map[string]string{" a": "v", "a ": "v"}, []string{"a:v"}},
+		{"case-colon-and-value-spaces", map[string]string{"Z": "https://example.invalid/a:b", "a": " Mixed Case "}, []string{"Z:https://example.invalid/a:b", "a: Mixed Case"}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := labelsToTags(tc.labels); !reflect.DeepEqual(got, tc.want) {
+				t.Fatalf("tags = %#v, want %#v", got, tc.want)
+			}
+		})
+	}
+}
+
+func TestLabelTagsDoNotSortOrderedIdentityInputs(t *testing.T) {
+	values := []string{" z ", "a", "z", "b", ""}
+	before := append([]string(nil), values...)
+	if got := uniqueStrings(values); !reflect.DeepEqual(got, []string{"z", "a", "b"}) {
+		t.Fatalf("first-occurrence order changed: %v", got)
+	}
+	if !reflect.DeepEqual(values, before) {
+		t.Fatal("deduplication changed the input list")
+	}
+
+	image, identity := resourceFromDockerImage(models.DockerImage{
+		ID: "sha256:example", RepoTags: []string{"z:last", "a:first"},
+		Labels: map[string]string{"z": "last", "a": "first"},
+	}, models.DockerHost{Hostname: "runtime-host"})
+	if image.Name != "z:last" || !reflect.DeepEqual(image.Docker.RepoTags, []string{"z:last", "a:first"}) ||
+		!reflect.DeepEqual(identity.Hostnames, []string{"z:last", "runtime-host:z:last"}) {
+		t.Fatalf("label sorting reordered image references or identity: %+v / %+v", image, identity)
+	}
+}

internal/websocket/broadcast_projection_export_test.go

@@ -32,3 +32,39 @@ func NewBroadcastProjectionProbeForTest(getState func(string) interface{}, recip
 		return bytes, nil
 	}, nil
 }
+
+// NewBroadcastProjectionCaptureForTest retains the actual per-client frames,
+// including a nil frame for a quiet or unrelated tenant. It runs the production
+// current-state getter, projection, delta baseline and queue path without timers
+// or a network socket, and is available only in the test binary.
+func NewBroadcastProjectionCaptureForTest(getState func(string) interface{}, orgIDs []string) (func(string) ([][]byte, error), error) {
+	hub := NewHub(getState)
+	clients := make([]*Client, len(orgIDs))
+	for i, orgID := range orgIDs {
+		client := &Client{hub: hub, orgID: orgID, send: make(chan []byte, 1)}
+		if _, sent, err := client.queueFullState("initialState", getState(orgID)); err != nil || !sent {
+			return nil, fmt.Errorf("initial state: sent=%v error=%v", sent, err)
+		}
+		<-client.send
+		clients[i] = client
+		hub.clients[client] = true
+		if hub.clientsByTenant[orgID] == nil {
+			hub.clientsByTenant[orgID] = make(map[*Client]bool)
+		}
+		hub.clientsByTenant[orgID][client] = true
+	}
+	return func(orgID string) ([][]byte, error) {
+		hub.dispatchStateBroadcast(&Message{Type: "rawData", Data: stateBroadcastRequest{}}, orgID)
+		frames := make([][]byte, len(clients))
+		for i, client := range clients {
+			if !hub.clients[client] {
+				return nil, fmt.Errorf("broadcast removed client %d", i)
+			}
+			select {
+			case frames[i] = <-client.send:
+			default:
+			}
+		}
+		return frames, nil
+	}, nil
+}

internal/websocket/label_tags_connected_test.go

@@ -0,0 +1,329 @@
+package websocket_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"maps"
+	"reflect"
+	"slices"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/models"
+	"github.com/rcourtman/pulse-go-rewrite/internal/monitoring"
+	"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
+	"github.com/rcourtman/pulse-go-rewrite/internal/websocket"
+)
+
+// These are source-native inventories, not pre-sorted ResourceFrontend fixtures.
+// Every label-derived Docker adapter and Kubernetes kind passes through a real
+// registry generation, monitor projection, wire snapshot and client delta queue.
+func labelTagInventory(now time.Time, labels map[string]string) models.StateSnapshot {
+	cluster := models.KubernetesCluster{ID: "label-cluster", Name: "label-cluster", Status: "online", LastSeen: now}
+	// The Kubernetes inventory has many kinds with the same UID/name/labels
+	// envelope. Include each one declared by models; the explicit count below
+	// prevents an empty or partial fixture from silently satisfying the test.
+	value := reflect.ValueOf(&cluster).Elem()
+	for i := 0; i < value.NumField(); i++ {
+		field := value.Field(i)
+		if field.Kind() != reflect.Slice || field.Type().Elem().Kind() != reflect.Struct {
+			continue
+		}
+		entry := reflect.New(field.Type().Elem()).Elem()
+		labelField := entry.FieldByName("Labels")
+		if !labelField.IsValid() {
+			continue
+		}
+		name := "label-" + strings.ToLower(value.Type().Field(i).Name)
+		for key, text := range map[string]string{"UID": "uid-" + name, "Name": name, "Namespace": "workloads", "Phase": "Running"} {
+			if target := entry.FieldByName(key); target.IsValid() && target.Kind() == reflect.String {
+				target.SetString(text)
+			}
+		}
+		if value.Type().Field(i).Name == "Namespaces" {
+			entry.FieldByName("Name").SetString("workloads")
+			entry.FieldByName("Phase").SetString("Active")
+		}
+		labelField.Set(reflect.ValueOf(labels))
+		field.Set(reflect.Append(field, entry))
+	}
+	return models.StateSnapshot{
+		LastUpdate: now,
+		DockerHosts: []models.DockerHost{{
+			ID: "label-docker", Hostname: "label-docker", Status: "online", LastSeen: now,
+			Swarm:    &models.DockerSwarmInfo{ClusterID: "label-swarm", ClusterName: "label-swarm"},
+			Images:   []models.DockerImage{{ID: "sha256:label-image", RepoTags: []string{"z:latest", "a:old"}, Labels: labels}},
+			Volumes:  []models.DockerVolume{{Name: "label-volume", Labels: labels}},
+			Networks: []models.DockerNetwork{{ID: "label-network", Name: "label-network", Labels: labels}},
+			Services: []models.DockerService{{ID: "label-service", Name: "label-service", Labels: labels}},
+			Nodes:    []models.DockerNode{{ID: "label-node", Hostname: "swarm-node", Labels: labels}},
+			Secrets:  []models.DockerSecret{{ID: "label-secret", Name: "label-secret", Labels: labels}},
+			Configs:  []models.DockerConfig{{ID: "label-config", Name: "label-config", Labels: labels}},
+		}},
+		KubernetesClusters: []models.KubernetesCluster{cluster},
+	}
+}
+
+func labelTagValues() map[string]string {
+	return map[string]string{"app": "api", "env": "production", "team": "ops", "tier": "backend", "region": "west", "release": "stable", "managed": "yes", "version": "1"}
+}
+
+type labelTagProjection struct {
+	adapter *unifiedresources.MonitorAdapter
+	store   *unifiedresources.MemoryStore
+	monitor *monitoring.Monitor
+}
+
+func newLabelTagProjection(snapshot models.StateSnapshot) *labelTagProjection {
+	store := unifiedresources.NewMemoryStore()
+	adapter := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(store))
+	adapter.PopulateFromSnapshot(snapshot)
+	monitor := &monitoring.Monitor{}
+	// Completed ingest is the baseline: a dashboard read must not replace it
+	// with this test Monitor's empty live polling state.
+	monitor.SetResourceStore(&broadcastReadStore{adapter})
+	return &labelTagProjection{adapter: adapter, store: store, monitor: monitor}
+}
+
+func (p *labelTagProjection) state() models.StateFrontend {
+	return p.monitor.BuildBroadcastFrontendState()
+}
+
+func (p *labelTagProjection) changes(t *testing.T) int {
+	t.Helper()
+	count, err := p.store.CountRecentChanges("", time.Time{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return count
+}
+
+type labelTagWireDelta struct {
+	Upserts []map[string]any `json:"upserts"`
+	Removed []string         `json:"removed"`
+	Order   []string         `json:"order"`
+}
+
+func labelTagFrameDelta(t *testing.T, frame []byte) labelTagWireDelta {
+	t.Helper()
+	if len(frame) == 0 {
+		return labelTagWireDelta{}
+	}
+	var message struct {
+		Type string                     `json:"type"`
+		Data map[string]json.RawMessage `json:"data"`
+	}
+	if err := json.Unmarshal(frame, &message); err != nil || message.Type != "rawData" {
+		t.Fatalf("invalid state frame: %s / %v", frame, err)
+	}
+	if _, full := message.Data["resources"]; full {
+		t.Fatal("connected client unexpectedly received a full resource array")
+	}
+	var delta labelTagWireDelta
+	if encoded, ok := message.Data["resourceDelta"]; ok {
+		if err := json.Unmarshal(encoded, &delta); err != nil {
+			t.Fatal(err)
+		}
+	}
+	return delta
+}
+
+func labelTagResourceMaps(t *testing.T, state models.StateFrontend) map[string]map[string]any {
+	t.Helper()
+	encoded, err := json.Marshal(state.Resources)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var resources []map[string]any
+	if err := json.Unmarshal(encoded, &resources); err != nil {
+		t.Fatal(err)
+	}
+	out := make(map[string]map[string]any, len(resources))
+	for _, resource := range resources {
+		out[resource["id"].(string)] = resource
+	}
+	return out
+}
+
+// Independent RFC 7396 receiver oracle; do not reuse the production diff builder.
+func applyLabelTagMergePatch(previous map[string]any, patch map[string]any) map[string]any {
+	out := maps.Clone(previous)
+	if out == nil {
+		out = make(map[string]any)
+	}
+	for key, value := range patch {
+		if value == nil {
+			delete(out, key)
+		} else if object, ok := value.(map[string]any); ok {
+			old, _ := out[key].(map[string]any)
+			out[key] = applyLabelTagMergePatch(old, object)
+		} else {
+			out[key] = value
+		}
+	}
+	return out
+}
+
+func TestLabelTagsConnectedRebuildsAreQuiet(t *testing.T) {
+	labels := labelTagValues()
+	before := maps.Clone(labels)
+	seed := labelTagInventory(time.Now().UTC(), labels)
+	projection := newLabelTagProjection(seed)
+	tagged := 0
+	for _, resource := range projection.state().Resources {
+		if len(resource.Tags) == len(labels) {
+			tagged++
+		}
+	}
+	if tagged != 33 { // Seven Docker adapters and twenty-six Kubernetes kinds.
+		t.Fatalf("fixture covers %d label-derived resources, want 33", tagged)
+	}
+	broadcast, err := websocket.NewBroadcastProjectionCaptureForTest(func(string) interface{} { return projection.state() }, []string{"default", "default"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	changes := projection.changes(t)
+	var tagPatches, upserts, wireBytes int
+	for i := 0; i < 32; i++ {
+		projection.adapter.PopulateFromSnapshot(seed)
+		frames, err := broadcast("default")
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !bytes.Equal(frames[0], frames[1]) {
+			t.Fatal("equal client baselines received different frames")
+		}
+		for _, frame := range frames {
+			wireBytes += len(frame)
+			delta := labelTagFrameDelta(t, frame)
+			if len(delta.Removed) != 0 || len(delta.Order) != 0 {
+				t.Fatal("equivalent rebuild changed resource identity or order")
+			}
+			upserts += len(delta.Upserts)
+			for _, patch := range delta.Upserts {
+				if _, ok := patch["tags"]; ok {
+					tagPatches++
+				}
+			}
+		}
+	}
+	t.Logf("33 label-derived resources, 32 rebuilds, two clients: tag patches=%d, all resource upserts=%d, queued wire bytes=%d", tagPatches, upserts, wireBytes)
+	if tagPatches != 0 || upserts != 0 {
+		t.Fatalf("unchanged label sets created %d tag patches / %d resource upserts", tagPatches, upserts)
+	}
+	if projection.changes(t) != changes || !reflect.DeepEqual(labels, before) {
+		t.Fatal("equivalent rebuild changed history or mutated source labels")
+	}
+}
+
+func TestLabelTagsConnectedChangesAndClearsRemainVisible(t *testing.T) {
+	labels := labelTagValues()
+	now := time.Now().UTC()
+	projection := newLabelTagProjection(labelTagInventory(now, labels))
+	otherTenant := newLabelTagProjection(labelTagInventory(now, map[string]string{"private": "other-tenant"}))
+	broadcast, err := websocket.NewBroadcastProjectionCaptureForTest(func(org string) interface{} {
+		if org == "other" {
+			return otherTenant.state()
+		}
+		return projection.state()
+	}, []string{"default", "default", "other"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	received := labelTagResourceMaps(t, projection.state())
+	changes := projection.changes(t)
+	for _, name := range []string{"replace", "add", "remove", "clear", "restore"} {
+		t.Run(name, func(t *testing.T) {
+			labels = maps.Clone(labels)
+			switch name {
+			case "replace":
+				labels["env"] = "staging"
+			case "add":
+				labels["new"] = "visible"
+			case "remove":
+				delete(labels, "app")
+			case "clear":
+				labels = nil
+			case "restore":
+				labels = labelTagValues()
+			}
+			projection.adapter.PopulateFromSnapshot(labelTagInventory(now, labels))
+			frames, err := broadcast("default")
+			if err != nil {
+				t.Fatal(err)
+			}
+			if len(frames[0]) == 0 || !bytes.Equal(frames[0], frames[1]) || len(frames[2]) != 0 {
+				t.Fatal("change was lost, differed per client or reached another tenant")
+			}
+			delta := labelTagFrameDelta(t, frames[0])
+			if len(delta.Upserts) != 33 || len(delta.Removed) != 0 || len(delta.Order) != 0 {
+				t.Fatalf("metadata edit changed identity/order or missed a resource: %+v", delta)
+			}
+			for _, patch := range delta.Upserts {
+				if _, changed := patch["tags"]; !changed {
+					t.Fatalf("real label change omitted tags: %v", patch)
+				}
+				id := patch["id"].(string)
+				received[id] = applyLabelTagMergePatch(received[id], patch)
+			}
+			want := labelTagResourceMaps(t, projection.state())
+			if !reflect.DeepEqual(received, want) {
+				t.Fatal("receiver replay differs from the full current projection (including labels/tags clears)")
+			}
+			for _, resource := range projection.state().Resources {
+				if !slices.IsSorted(resource.Tags) {
+					t.Fatalf("changed tags are not sorted for %s: %v", resource.ID, resource.Tags)
+				}
+				if name == "clear" && (len(resource.Tags) != 0 || len(resource.Labels) != 0) {
+					t.Fatalf("cleared metadata remained on %s", resource.ID)
+				}
+			}
+			changes += 33
+			if got := projection.changes(t); got != changes {
+				t.Fatalf("history lost or duplicated real changes: %d, want %d", got, changes)
+			}
+		})
+	}
+}
+
+func TestLabelTagsFleetHeartbeatDeltas(t *testing.T) {
+	now := time.Now().UTC()
+	labels := labelTagValues()
+	seed := models.StateSnapshot{LastUpdate: now, KubernetesClusters: []models.KubernetesCluster{{ID: "fleet", Name: "fleet", Status: "online", LastSeen: now}}}
+	for i := 0; i < 1000; i++ {
+		seed.KubernetesClusters[0].Pods = append(seed.KubernetesClusters[0].Pods, models.KubernetesPod{
+			UID: fmt.Sprintf("pod-%04d", i), Name: fmt.Sprintf("pod-%04d", i), Namespace: "workloads", Phase: "Running", Labels: labels,
+		})
+	}
+	projection := newLabelTagProjection(seed)
+	broadcast, err := websocket.NewBroadcastProjectionCaptureForTest(func(string) interface{} { return projection.state() }, []string{"default"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var tagPatches, wireBytes int
+	for i := 0; i < 8; i++ {
+		// Advance observed freshness as a real heartbeat does, not labels.
+		seed.LastUpdate = now.Add(time.Duration(i+1) * time.Second)
+		seed.KubernetesClusters[0].LastSeen = seed.LastUpdate
+		projection.adapter.PopulateFromSnapshot(seed)
+		frames, err := broadcast("default")
+		if err != nil || len(frames[0]) == 0 {
+			t.Fatalf("fresh heartbeat was not queued: %v", err)
+		}
+		wireBytes += len(frames[0])
+		for _, patch := range labelTagFrameDelta(t, frames[0]).Upserts {
+			if _, changed := patch["tags"]; changed {
+				tagPatches++
+			}
+			if _, changed := patch["labels"]; changed {
+				t.Fatal("heartbeat changed label values")
+			}
+		}
+	}
+	t.Logf("1,000 pods, eight fresh heartbeats, one client: false tag patches=%d, queued wire bytes=%d (includes real freshness deltas)", tagPatches, wireBytes)
+	if tagPatches != 0 {
+		t.Fatalf("unchanged fleet labels created %d tag patches", tagPatches)
+	}
+}