fix(websocket): send resource deltas after initial state

Refs #1601
Refs #1665
Refs #1497
This commit is contained in:
rcourtman
2026-08-06 00:45:27 +01:00
parent 662bb4f194
commit 962d297803
11 changed files with 759 additions and 52 deletions
@@ -216,6 +216,7 @@ acknowledgement, history, and recovery reuse the normal alert pipeline.
73. `frontend-modern/src/utils/alertsActivation.ts`
74. `internal/operationaltrust/contracts.go`
75. `internal/alerts/operational_contract.go`
76. `internal/alerts/issue1497_test.go`
## Shared Boundaries
@@ -88,6 +88,11 @@ an enabled external-probe assignment.
11. `pkg/pulsecli/root.go`
12. `frontend-modern/src/types/api.ts`
12a. `frontend-modern/src/types/operationalTrust.ts`
12b. `frontend-modern/src/stores/websocket.ts`
12c. `frontend-modern/src/stores/__tests__/websocket-unified.test.ts`
12d. `internal/websocket/hub.go`
12e. `internal/websocket/state_delta.go`
12f. `internal/websocket/state_delta_test.go`
13. `frontend-modern/src/types/actionAudit.ts`
14. `frontend-modern/src/api/actionAudit.ts`
14a. `frontend-modern/src/api/patrolAttention.ts`
@@ -291,12 +296,20 @@ not a parallel API payload contract. `internal/api/agent_handlers_base.go` and
or retain full frontend-state payloads at the handler boundary. The WebSocket
hub owns tenant-aware state resolution after coalescing, through the same state
getter that backs the canonical `/api/state` payload. The hub must serialize
whole-state resolution and JSON construction across coalesced broadcasts,
initial-client delivery, and explicit client data requests; cancel delayed or
queued work when its client leaves; and join every state producer before
closing client channels. Reconnect, request, or invalidation churn must not
multiply concurrent clones of that canonical payload or race shutdown sends
against channel closure. A client's lifecycle cancellation signal must be
whole-state resolution across coalesced broadcasts, initial-client delivery,
and explicit client data requests. Initial hydration and explicit resync return
the complete canonical state; subsequent invalidation broadcasts must derive
ordered, per-client resource merge patches from that same resolved state,
including explicit removals and resource order, instead of repeatedly
serialising the whole resource graph. A delta may be sent only after that
client has accepted its full baseline, and the baseline may advance only after
the ordered client queue accepts the delta. The frontend must apply JSON Merge
Patch deletion semantics, preserve unchanged resource fields, and retain
support for complete `rawData` snapshots during resync. The hub must cancel
delayed or queued work when its client leaves and join every state producer
before closing client channels. Reconnect, request, or invalidation churn must
not multiply concurrent clones of that canonical payload or race shutdown
sends against channel closure. A client's lifecycle cancellation signal must be
initialized and read through one synchronized owner before registration,
requested-state work, delayed initial-state delivery, or disconnect may race
over it; every path must observe the same stable signal. Ordinary unregister
+28 -24
View File
@@ -1,38 +1,42 @@
{
"version": 1,
"base_sha": "1b8bb4e91c905de19931634ab79e9e95960f811a",
"verified_at": "2026-08-05T23:31:48Z",
"base_sha": "662bb4f1946606e51122bfe06fd2d82aa33c6a15",
"verified_at": "2026-08-05T23:44:24Z",
"result": "passed",
"changed_paths": ["frontend-modern/src/components/shared/InlineDetailTableRow.tsx"],
"changed_paths": [
"frontend-modern/src/stores/websocket.ts",
"frontend-modern/src/types/api.ts"
],
"content_sha256": {
"frontend-modern/src/components/shared/InlineDetailTableRow.tsx": "aeb0e3d0d5946d5dc5929d2d5ce9ae1a39d1c48fc8cc35db2332059d577d5c9e"
"frontend-modern/src/stores/websocket.ts": "056eb325939f188c6095273441fd3b48c0f02cdfb4dd0a5b044c43b73868301a",
"frontend-modern/src/types/api.ts": "895ab5aa5ee5e1162008be6c8da83377169941d25b1655d73a1042d91076697b"
},
"routes": [
"/standalone/machines",
"/proxmox/backups?view=coverage",
"/proxmox/ceph",
"/proxmox/mail"
"/proxmox/overview",
"/truenas/overview"
],
"viewports": [
{ "width": 1920, "height": 1080 },
{ "width": 1280, "height": 820 },
{ "width": 768, "height": 820 },
{ "width": 390, "height": 820 },
{ "width": 320, "height": 820 }
{
"width": 1280,
"height": 720
},
{
"width": 390,
"height": 844
}
],
"states": [
"collapsed and expanded Agent machine resource details",
"collapsed and expanded Proxmox backup coverage evidence",
"auto-opened and closed single-cluster Ceph details",
"collapsed, expanded, and closed Mail Gateway details",
"desktop, tablet, narrow-phone, and minimum-width responsive table columns"
"full initial WebSocket state hydration",
"live resource merge-patch refresh after the initial baseline",
"full WebSocket resync after a page reload",
"Proxmox node disk-array rendering after a delta refresh",
"TrueNAS system and health-alert rendering after desktop and narrow delta refreshes"
],
"interactions": [
"opened and closed the Agent machine resource drawer and confirmed focus returned to its disclosure",
"expanded and collapsed backup coverage evidence and confirmed focus remained on its disclosure",
"closed the auto-opened Ceph cluster drawer and confirmed focus returned to its disclosure",
"opened and closed the Mail Gateway drawer and confirmed focus returned to its disclosure",
"measured every visible table at all five viewports and confirmed table width matched its container with zero document overflow",
"visually inspected the final Mail Gateway table at 1920 and 320 pixels"
"signed in to the current local build and waited for the initial Proxmox resource workspace to hydrate",
"waited through a live refresh and confirmed changing metrics rendered without disconnecting or logging browser errors",
"opened the TrueNAS workspace and confirmed its system and alert projections remained intact after delta application",
"reloaded the TrueNAS route and confirmed the complete resync restored a connected, populated workspace",
"repeated TrueNAS initial hydration and live refresh at 390 by 844 and visually inspected the responsive result"
]
}
@@ -162,6 +162,57 @@ describe('websocket store resilience', () => {
}
});
it('preserves active alert state when a resource-only delta arrives', async () => {
const { store, dispose } = await createStoreHarness();
try {
vi.advanceTimersByTime(1);
const alert = {
id: 'agent:host-1-cpu',
type: 'cpu',
level: 'warning',
resourceId: 'agent:host-1',
resourceName: 'host-1',
node: 'host-1',
instance: 'host-1',
message: 'CPU above threshold',
value: 90,
threshold: 80,
startTime: '2026-05-14T07:59:00Z',
acknowledged: false,
};
currentInstance!.onmessage?.({
data: JSON.stringify({
type: 'initialState',
data: {
connectedInfrastructure: [],
resources: [{ id: 'agent:host-1', type: 'agent', name: 'host-1' }],
activeAlerts: [alert],
recentlyResolved: [],
lastUpdate: Date.now(),
},
}),
} as MessageEvent);
currentInstance!.onmessage?.({
data: JSON.stringify({
type: 'rawData',
data: {
resourceDelta: {
upserts: [{ id: 'agent:host-1', cpu: { current: 91 } }],
},
lastUpdate: Date.now() + 1_000,
},
}),
} as MessageEvent);
expect(store.activeAlerts[alert.id]).toMatchObject(alert);
expect(store.state.activeAlerts).toHaveLength(1);
expect(store.state.resources[0]?.cpu?.current).toBe(91);
} finally {
dispose();
}
});
it('manual reconnect avoids duplicate reconnect scheduling', async () => {
const { store, dispose } = await createStoreHarness();
try {
@@ -254,6 +254,72 @@ describe('websocket store unified resource contract', () => {
}
});
it('applies resource merge patches without replacing unchanged resource data', async () => {
const { store, dispose } = await createStoreHarness();
try {
await waitForOpenTick();
emitMessage({
type: 'initialState',
data: {
connectedInfrastructure: [],
resources: [
{
id: 'agent-1',
type: 'agent',
name: 'agent-1',
status: 'online',
lastSeen: 100,
cpu: { current: 10 },
platformData: {
agent: { osName: 'Debian', agentVersion: '6.2.0' },
disks: [{ name: 'sda', usage: 10 }],
},
},
{ id: 'vm-1', type: 'vm', name: 'vm-1', status: 'running' },
],
lastUpdate: 100,
activeAlerts: [],
recentlyResolved: [],
},
});
emitMessage({
type: 'rawData',
data: {
lastUpdate: 200,
resourceDelta: {
upserts: [
{
id: 'agent-1',
lastSeen: 200,
cpu: { current: 42 },
status: null,
platformData: { disks: [{ name: 'sda', usage: 20 }] },
},
{ id: 'vm-2', type: 'vm', name: 'vm-2', status: 'running' },
],
removed: ['vm-1'],
order: ['vm-2', 'agent-1'],
},
},
});
expect(store.state.resources.map((resource) => resource.id)).toEqual(['vm-2', 'agent-1']);
const agent = store.state.resources[1];
expect(agent?.cpu?.current).toBe(42);
expect(agent?.lastSeen).toBe(200);
expect(agent?.platformData).toMatchObject({
agent: { osName: 'Debian', agentVersion: '6.2.0' },
disks: [{ name: 'sda', usage: 20 }],
});
expect(agent).not.toHaveProperty('status');
expect(store.state.lastUpdate).toBe(200);
} finally {
dispose();
}
});
it('preserves connected infrastructure when raw updates omit that projection', async () => {
const { store, dispose } = await createStoreHarness();
try {
+81 -4
View File
@@ -7,6 +7,7 @@ import type {
ResolvedAlert,
ConnectedInfrastructureItem,
} from '@/types/api';
import type { Resource } from '@/types/resource';
import { logger } from '@/utils/logger';
import { POLLING_INTERVALS, WEBSOCKET } from '@/constants';
import { notificationStore } from './notifications';
@@ -40,9 +41,75 @@ const shownAutoRegisterNotifications = new Map<string, number>();
const asRecord = (value: unknown): Record<string, unknown> | undefined =>
value && typeof value === 'object' ? (value as Record<string, unknown>) : undefined;
const asMergePatchRecord = (value: unknown): Record<string, unknown> | undefined =>
value && typeof value === 'object' && !Array.isArray(value)
? (value as Record<string, unknown>)
: undefined;
const asString = (value: unknown): string | undefined =>
typeof value === 'string' && value.trim().length > 0 ? value.trim() : undefined;
const isInboundPayloadWithinLimit = (payload: string): boolean => {
// JSON state is overwhelmingly ASCII. Avoid allocating a second multi-megabyte
// Blob for normal messages, while retaining an exact UTF-8 check for strings
// large enough that non-ASCII code points could cross the byte limit.
if (payload.length <= Math.floor(MAX_INBOUND_WEBSOCKET_MESSAGE_BYTES / 3)) return true;
if (payload.length > MAX_INBOUND_WEBSOCKET_MESSAGE_BYTES) return false;
return new TextEncoder().encode(payload).byteLength <= MAX_INBOUND_WEBSOCKET_MESSAGE_BYTES;
};
const applyJSONMergePatch = (current: unknown, patch: unknown): unknown => {
const patchRecord = asMergePatchRecord(patch);
if (!patchRecord) return patch;
const result: Record<string, unknown> = { ...(asMergePatchRecord(current) ?? {}) };
Object.entries(patchRecord).forEach(([key, value]) => {
if (value === null) {
delete result[key];
return;
}
result[key] = asMergePatchRecord(value) ? applyJSONMergePatch(result[key], value) : value;
});
return result;
};
const applyResourceStateDelta = (
current: readonly Resource[],
delta: { upserts?: unknown; removed?: unknown; order?: unknown },
): Resource[] => {
const resourcesById = new Map(current.map((resource) => [resource.id, resource] as const));
const removed = Array.isArray(delta.removed)
? new Set(delta.removed.filter((id): id is string => typeof id === 'string'))
: new Set<string>();
removed.forEach((id) => resourcesById.delete(id));
const addedIds: string[] = [];
if (Array.isArray(delta.upserts)) {
delta.upserts.forEach((patch) => {
const id = asString(asRecord(patch)?.id);
if (!id) return;
if (!resourcesById.has(id)) addedIds.push(id);
const next = applyJSONMergePatch(resourcesById.get(id), patch) as Resource;
if (next.id === id) resourcesById.set(id, next);
});
}
const requestedOrder = Array.isArray(delta.order)
? delta.order.filter((id): id is string => typeof id === 'string')
: undefined;
const order = requestedOrder ?? [
...current.map((resource) => resource.id).filter((id) => !removed.has(id)),
...addedIds,
];
const ordered = order
.map((id) => resourcesById.get(id))
.filter((resource): resource is Resource => Boolean(resource));
const included = new Set(ordered.map((resource) => resource.id));
resourcesById.forEach((resource, id) => {
if (!included.has(id)) ordered.push(resource);
});
return ordered;
};
const parseTimestampMs = (value: unknown): number | null => {
if (typeof value === 'number' && Number.isFinite(value)) {
return value < 1_000_000_000_000 ? value * 1000 : value;
@@ -407,10 +474,9 @@ export function createWebSocketStore(url: string) {
return;
}
const payloadSizeBytes = new Blob([event.data]).size;
if (payloadSizeBytes > MAX_INBOUND_WEBSOCKET_MESSAGE_BYTES) {
if (!isInboundPayloadWithinLimit(event.data)) {
logger.warn('Ignoring oversized WebSocket payload', {
sizeBytes: payloadSizeBytes,
characterCount: event.data.length,
maxBytes: MAX_INBOUND_WEBSOCKET_MESSAGE_BYTES,
});
return;
@@ -466,10 +532,21 @@ export function createWebSocketStore(url: string) {
setState('pveTagStyles', message.data.pveTagStyles ?? {});
}
// Handle unified resources
let nextResources: Resource[] | undefined;
if (message.data.resources !== undefined) {
const nextResources = Array.isArray(message.data.resources)
nextResources = Array.isArray(message.data.resources)
? mergeCanonicalResourceSnapshot(message.data.resources, state.resources)
: [];
} else if (
'resourceDelta' in message.data &&
message.data.resourceDelta !== undefined
) {
nextResources = applyResourceStateDelta(
state.resources,
message.data.resourceDelta,
);
}
if (nextResources !== undefined) {
logger.debug('[WebSocket] Updating resources', {
count: nextResources.length,
types: [...new Set(nextResources.map((resource) => resource.type) || [])],
+11 -1
View File
@@ -33,6 +33,16 @@ export interface State {
resources: Resource[];
}
export interface ResourceStateDelta {
upserts?: Array<Partial<Resource> & Pick<Resource, 'id'>>;
removed?: string[];
order?: string[];
}
export type StateUpdate = Partial<State> & {
resourceDelta?: ResourceStateDelta;
};
export interface PVETagStyle {
colors: Record<string, string>;
caseSensitive: boolean;
@@ -1333,7 +1343,7 @@ export interface Incident {
// WebSocket message types
export type WSMessage =
| { type: 'initialState'; data: State }
| { type: 'rawData'; data: State }
| { type: 'rawData'; data: StateUpdate }
| { type: 'error'; error: string }
| { type: 'ping'; data?: unknown }
| { type: 'pong'; data?: unknown }
+80
View File
@@ -0,0 +1,80 @@
package alerts
import (
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
)
func TestCheckHostHonoursCanonicalResourceIntentGrace(t *testing.T) {
m := newTestManager(t)
start := time.Date(2026, 8, 6, 0, 0, 0, 0, time.UTC)
now := start
var tick time.Duration
m.now = func() time.Time { return now }
m.intentClock = func() time.Duration { return tick }
m.mu.Lock()
m.config.TimeThresholds = map[string]int{}
m.config.AgentDefaults = ThresholdConfig{
CPU: &HysteresisThreshold{Trigger: 80, Clear: 75},
}
m.mu.Unlock()
document := NewAlertIntentPolicyDocument()
document.Resources["agent-canonical"] = map[string]AlertIntentRule{
MetricAlertIntentSignal("cpu"): {GraceSeconds: intPointer(240)},
}
if err := m.LoadIntentPolicies(document); err != nil {
t.Fatal(err)
}
m.SetResourceIntentIdentityResolver(func(resourceID string) (string, bool) {
if resourceID == "agent:host-1" {
return "agent-canonical", true
}
return "", false
})
host := models.Host{
ID: "host-1",
Hostname: "host-1",
CPUUsage: 90,
Status: "online",
LastSeen: start,
}
alertID := canonicalMetricStateID(hostResourceID(host.ID), "cpu")
m.CheckHost(host)
m.mu.RLock()
_, exists := m.activeAlerts[alertID]
m.mu.RUnlock()
if exists {
t.Fatal("alert activated before the canonical resource grace period started")
}
now = start.Add(239 * time.Second)
tick = 239 * time.Second
m.CheckHost(host)
m.mu.RLock()
_, exists = m.activeAlerts[alertID]
m.mu.RUnlock()
if exists {
t.Fatal("alert activated before the canonical resource grace period elapsed")
}
now = start.Add(240 * time.Second)
tick = 240 * time.Second
m.CheckHost(host)
m.mu.RLock()
alert := m.activeAlerts[alertID]
if alert == nil {
m.mu.RUnlock()
t.Fatal("alert did not activate when the canonical resource grace period elapsed")
}
alertStart := alert.StartTime
m.mu.RUnlock()
if !alertStart.Equal(start) {
t.Fatalf("alert StartTime = %s, want first match %s", alertStart, start)
}
}
+115 -17
View File
@@ -273,6 +273,8 @@ type Client struct {
lifecycleMu sync.Mutex
lifecycleDone chan struct{}
lifecycleOnce sync.Once
stateMu sync.Mutex
stateSnapshot *clientStateSnapshot
}
func (c *Client) lifecycleSignal() chan struct{} {
@@ -320,6 +322,57 @@ func (c *Client) safeSend(data []byte) (sent bool) {
}
}
func (c *Client) queueFullState(messageType string, state interface{}) (int, bool, error) {
snapshot, err := buildClientStateSnapshot(state)
if err != nil {
return 0, false, err
}
data, err := json.Marshal(Message{Type: messageType, Data: state})
if err != nil {
return 0, false, err
}
c.stateMu.Lock()
defer c.stateMu.Unlock()
if !c.safeSend(data) {
return len(data), false, nil
}
c.stateSnapshot = snapshot
return len(data), true, nil
}
func (c *Client) queueStateDelta(current *clientStateSnapshot) (int, bool, bool, error) {
if current == nil {
return 0, false, false, fmt.Errorf("current state snapshot is nil")
}
c.stateMu.Lock()
defer c.stateMu.Unlock()
if c.stateSnapshot == nil {
// Initial hydration owns the baseline. A client still waiting for that
// payload must not receive a delta based on state it has never seen.
return 0, false, false, nil
}
delta, err := buildClientStateDelta(c.stateSnapshot, current)
if err != nil {
return 0, true, false, err
}
if len(delta) == 0 {
c.stateSnapshot = current
return 0, false, true, nil
}
data, err := json.Marshal(Message{Type: "rawData", Data: delta})
if err != nil {
return 0, true, false, err
}
if !c.safeSend(data) {
return len(data), true, false, nil
}
c.stateSnapshot = current
return len(data), true, true, nil
}
// cloneAlertData returns a broadcast-safe copy of alert data to avoid data races when
// downstream sanitization/encoding happens concurrently with alert manager mutations.
func cloneAlertData(alert interface{}) interface{} {
@@ -753,15 +806,7 @@ func (h *Hub) sendInitialState(client *Client) {
log.Debug().Str("client", client.id).Msg("about to get state")
stateData := h.getStateForClient(client)
log.Debug().Str("client", client.id).Interface("stateType", fmt.Sprintf("%T", stateData)).Msg("got state for initial message")
initialMsg := Message{
Type: "initialState",
Data: sanitizeData(h.prepareStateForBroadcast(stateData)),
}
data, err = json.Marshal(initialMsg)
if err != nil {
log.Error().Err(err).Str("client", client.id).Msg("failed to marshal initial state")
return
}
stateData = h.prepareStateForBroadcast(stateData)
h.mu.RLock()
_, stillRegistered := h.clients[client]
@@ -771,8 +816,14 @@ func (h *Hub) sendInitialState(client *Client) {
return
}
log.Info().Str("client", client.id).Int("dataLen", len(data)).Int("dataKB", len(data)/1024).Msg("sending initial state to client")
if client.safeSend(data) {
dataLen, sent, err := client.queueFullState("initialState", stateData)
if err != nil {
log.Error().Err(err).Str("client", client.id).Msg("failed to marshal initial state")
return
}
log.Info().Str("client", client.id).Int("dataLen", dataLen).Int("dataKB", dataLen/1024).Msg("sending initial state to client")
if sent {
log.Info().Str("client", client.id).Msg("initial state sent successfully")
} else {
log.Warn().Str("client", client.id).Msg("client closed or buffer full, skipping initial state")
@@ -789,16 +840,15 @@ func (h *Hub) sendRequestedState(client *Client) {
}
defer h.releaseStateBuildSlot()
stateMsg := Message{
Type: "rawData",
Data: sanitizeData(h.prepareStateForBroadcast(h.getStateForClient(client))),
}
data, err := json.Marshal(stateMsg)
_, sent, err := client.queueFullState(
"rawData",
h.prepareStateForBroadcast(h.getStateForClient(client)),
)
if err != nil {
log.Error().Err(err).Str("client", client.id).Msg("failed to marshal state for requestData")
return
}
if !client.safeSend(data) {
if !sent {
log.Warn().Str("client", client.id).Msg("Failed to queue requestData state response; client channel closed or full")
}
}
@@ -1126,6 +1176,14 @@ func (h *Hub) dispatchStateBroadcast(pending *Message, orgID string) {
return
}
defer h.releaseStateBuildSlot()
if _, currentStateRequest := h.messageHasCurrentStateRequest(*pending); currentStateRequest {
prepared, ok := h.prepareMessageForMarshal(*pending, orgID)
if !ok {
return
}
h.dispatchCurrentStateSnapshot(prepared.Data, orgID)
return
}
data, ok := h.marshalBroadcastMessage(*pending, orgID)
if !ok {
return
@@ -1137,6 +1195,46 @@ func (h *Hub) dispatchStateBroadcast(pending *Message, orgID string) {
h.dispatchToTenantClients(orgID, data, "Client send channel full, dropping tenant coalesced message and closing connection")
}
func (h *Hub) dispatchCurrentStateSnapshot(state interface{}, orgID string) {
snapshot, err := buildClientStateSnapshot(state)
if err != nil {
log.Error().Err(err).Str("org_id", normalizeOrgID(orgID)).Msg("Failed to build WebSocket state delta snapshot")
return
}
h.mu.RLock()
clients := make([]*Client, 0, len(h.clients))
if orgID == "" {
for client := range h.clients {
clients = append(clients, client)
}
} else if tenantClients := h.clientsByTenant[orgID]; tenantClients != nil {
for client := range tenantClients {
clients = append(clients, client)
}
}
h.mu.RUnlock()
for _, client := range clients {
dataLen, attempted, sent, queueErr := client.queueStateDelta(snapshot)
if queueErr != nil {
log.Error().Err(queueErr).Str("client", client.id).Str("org_id", client.orgID).Msg("Failed to marshal WebSocket state delta")
continue
}
if !attempted || sent {
if sent {
log.Debug().Str("client", client.id).Int("dataLen", dataLen).Msg("queued WebSocket state delta")
}
continue
}
h.mu.Lock()
if h.removeClientLocked(client) {
log.Warn().Str("client", client.id).Str("org_id", client.orgID).Msg("Client send channel full, dropping state delta and closing connection")
}
h.mu.Unlock()
}
}
func (h *Hub) runStateBroadcastWorker() {
defer close(h.stateBroadcastDone)
for {
+175
View File
@@ -0,0 +1,175 @@
package websocket
import (
"bytes"
"encoding/json"
"fmt"
"reflect"
"sort"
)
const resourceDeltaField = "resourceDelta"
type clientStateSnapshot struct {
fields map[string]json.RawMessage
resources map[string]json.RawMessage
resourceOrder []string
}
type resourceDeltaPayload struct {
Upserts []json.RawMessage `json:"upserts,omitempty"`
Removed []string `json:"removed,omitempty"`
Order []string `json:"order,omitempty"`
}
func buildClientStateSnapshot(state interface{}) (*clientStateSnapshot, error) {
encoded, err := json.Marshal(state)
if err != nil {
return nil, fmt.Errorf("marshal state snapshot: %w", err)
}
fields := make(map[string]json.RawMessage)
if err := json.Unmarshal(encoded, &fields); err != nil {
return nil, fmt.Errorf("decode state snapshot: %w", err)
}
resources := make(map[string]json.RawMessage)
resourceOrder := make([]string, 0)
if encodedResources, ok := fields["resources"]; ok {
var entries []json.RawMessage
if err := json.Unmarshal(encodedResources, &entries); err != nil {
return nil, fmt.Errorf("decode state resources: %w", err)
}
for _, entry := range entries {
var identity struct {
ID string `json:"id"`
}
if err := json.Unmarshal(entry, &identity); err != nil {
return nil, fmt.Errorf("decode state resource identity: %w", err)
}
if identity.ID == "" {
return nil, fmt.Errorf("state resource is missing id")
}
if _, exists := resources[identity.ID]; exists {
return nil, fmt.Errorf("state resource id %q is duplicated", identity.ID)
}
resources[identity.ID] = append(json.RawMessage(nil), entry...)
resourceOrder = append(resourceOrder, identity.ID)
}
delete(fields, "resources")
}
return &clientStateSnapshot{
fields: fields,
resources: resources,
resourceOrder: resourceOrder,
}, nil
}
func buildClientStateDelta(previous, current *clientStateSnapshot) (map[string]interface{}, error) {
if previous == nil || current == nil {
return nil, fmt.Errorf("state delta requires previous and current snapshots")
}
delta := make(map[string]interface{})
for key, currentValue := range current.fields {
if previousValue, ok := previous.fields[key]; !ok || !bytes.Equal(previousValue, currentValue) {
delta[key] = currentValue
}
}
for key := range previous.fields {
if _, ok := current.fields[key]; !ok {
delta[key] = nil
}
}
resourceDelta := resourceDeltaPayload{}
for _, id := range current.resourceOrder {
currentResource := current.resources[id]
previousResource, exists := previous.resources[id]
if !exists {
resourceDelta.Upserts = append(resourceDelta.Upserts, currentResource)
continue
}
if bytes.Equal(previousResource, currentResource) {
continue
}
patch, err := createJSONMergePatch(previousResource, currentResource)
if err != nil {
return nil, fmt.Errorf("build resource %q patch: %w", id, err)
}
resourceDelta.Upserts = append(resourceDelta.Upserts, patch)
}
for id := range previous.resources {
if _, exists := current.resources[id]; !exists {
resourceDelta.Removed = append(resourceDelta.Removed, id)
}
}
sort.Strings(resourceDelta.Removed)
if !reflect.DeepEqual(previous.resourceOrder, current.resourceOrder) {
resourceDelta.Order = append([]string(nil), current.resourceOrder...)
}
if len(resourceDelta.Upserts) > 0 || len(resourceDelta.Removed) > 0 || len(resourceDelta.Order) > 0 {
delta[resourceDeltaField] = resourceDelta
}
return delta, nil
}
func createJSONMergePatch(previous, current json.RawMessage) (json.RawMessage, error) {
var previousValue interface{}
if err := json.Unmarshal(previous, &previousValue); err != nil {
return nil, err
}
var currentValue interface{}
if err := json.Unmarshal(current, &currentValue); err != nil {
return nil, err
}
patchValue, changed := diffJSONMergeValue(previousValue, currentValue)
if !changed {
return json.RawMessage(`{}`), nil
}
if patchObject, ok := patchValue.(map[string]interface{}); ok {
if currentObject, ok := currentValue.(map[string]interface{}); ok {
if id, ok := currentObject["id"]; ok {
patchObject["id"] = id
}
}
}
patch, err := json.Marshal(patchValue)
if err != nil {
return nil, err
}
return patch, nil
}
func diffJSONMergeValue(previous, current interface{}) (interface{}, bool) {
if reflect.DeepEqual(previous, current) {
return nil, false
}
previousObject, previousIsObject := previous.(map[string]interface{})
currentObject, currentIsObject := current.(map[string]interface{})
if !previousIsObject || !currentIsObject {
return current, true
}
patch := make(map[string]interface{})
for key := range previousObject {
if _, exists := currentObject[key]; !exists {
patch[key] = nil
}
}
for key, currentValue := range currentObject {
previousValue, exists := previousObject[key]
if !exists {
patch[key] = currentValue
continue
}
if nestedPatch, changed := diffJSONMergeValue(previousValue, currentValue); changed {
patch[key] = nestedPatch
}
}
return patch, len(patch) > 0
}
+132
View File
@@ -0,0 +1,132 @@
package websocket
import (
"encoding/json"
"strings"
"testing"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
)
func TestClientStateDeltaOmitsUnchangedResourcePayload(t *testing.T) {
previousState := models.EmptyStateFrontend()
previousState.LastUpdate = 100
previousState.Resources = []models.ResourceFrontend{{
ID: "agent:host-1",
Type: "agent",
Name: "host-1",
DisplayName: "host-1",
Status: "online",
LastSeen: 100,
CPU: &models.ResourceMetricFrontend{Current: 40},
PlatformData: json.RawMessage(`{"static":"` + strings.Repeat("x", 16*1024) + `"}`),
}}
currentState := previousState
currentState.LastUpdate = 200
currentState.Resources = append([]models.ResourceFrontend(nil), previousState.Resources...)
currentState.Resources[0].LastSeen = 200
currentState.Resources[0].CPU = &models.ResourceMetricFrontend{Current: 41}
previous, err := buildClientStateSnapshot(previousState)
if err != nil {
t.Fatal(err)
}
current, err := buildClientStateSnapshot(currentState)
if err != nil {
t.Fatal(err)
}
delta, err := buildClientStateDelta(previous, current)
if err != nil {
t.Fatal(err)
}
encoded, err := json.Marshal(delta)
if err != nil {
t.Fatal(err)
}
if strings.Contains(string(encoded), strings.Repeat("x", 128)) {
t.Fatal("state delta repeated unchanged platform data")
}
if !strings.Contains(string(encoded), `"current":41`) || !strings.Contains(string(encoded), `"lastSeen":200`) {
t.Fatalf("state delta omitted changed telemetry: %s", encoded)
}
if len(encoded) >= 1024 {
t.Fatalf("state delta = %d bytes, want a bounded telemetry patch", len(encoded))
}
}
func TestClientStateDeltaTracksResourceRemovalAndOrder(t *testing.T) {
previousState := models.EmptyStateFrontend()
previousState.Resources = []models.ResourceFrontend{
{ID: "a", Type: "agent", Name: "a", DisplayName: "a"},
{ID: "b", Type: "agent", Name: "b", DisplayName: "b"},
}
currentState := models.EmptyStateFrontend()
currentState.Resources = []models.ResourceFrontend{
{ID: "c", Type: "agent", Name: "c", DisplayName: "c"},
{ID: "a", Type: "agent", Name: "a", DisplayName: "a"},
}
previous, err := buildClientStateSnapshot(previousState)
if err != nil {
t.Fatal(err)
}
current, err := buildClientStateSnapshot(currentState)
if err != nil {
t.Fatal(err)
}
delta, err := buildClientStateDelta(previous, current)
if err != nil {
t.Fatal(err)
}
resources, ok := delta[resourceDeltaField].(resourceDeltaPayload)
if !ok {
t.Fatalf("resource delta type = %T", delta[resourceDeltaField])
}
if len(resources.Removed) != 1 || resources.Removed[0] != "b" {
t.Fatalf("removed = %#v, want [b]", resources.Removed)
}
if len(resources.Order) != 2 || resources.Order[0] != "c" || resources.Order[1] != "a" {
t.Fatalf("order = %#v, want [c a]", resources.Order)
}
if len(resources.Upserts) != 1 || !strings.Contains(string(resources.Upserts[0]), `"id":"c"`) {
t.Fatalf("upserts = %s, want full resource c", resources.Upserts)
}
}
func TestClientStateBaselineAdvancesOnlyAfterDeltaIsQueued(t *testing.T) {
initial := models.EmptyStateFrontend()
initial.LastUpdate = 100
initial.Resources = []models.ResourceFrontend{{
ID: "agent:host-1", Type: "agent", Name: "host-1", DisplayName: "host-1",
CPU: &models.ResourceMetricFrontend{Current: 10},
}}
current := initial
current.LastUpdate = 200
current.Resources = append([]models.ResourceFrontend(nil), initial.Resources...)
current.Resources[0].CPU = &models.ResourceMetricFrontend{Current: 20}
client := &Client{send: make(chan []byte, 1)}
if _, sent, err := client.queueFullState("initialState", initial); err != nil || !sent {
t.Fatalf("queueFullState() sent=%v error=%v", sent, err)
}
initialBaseline := client.stateSnapshot
currentSnapshot, err := buildClientStateSnapshot(current)
if err != nil {
t.Fatal(err)
}
if _, attempted, sent, err := client.queueStateDelta(currentSnapshot); err != nil || !attempted || sent {
t.Fatalf("full-channel queueStateDelta() attempted=%v sent=%v error=%v", attempted, sent, err)
}
if client.stateSnapshot != initialBaseline {
t.Fatal("client baseline advanced after the delta queue rejected the payload")
}
<-client.send
if _, attempted, sent, err := client.queueStateDelta(currentSnapshot); err != nil || !attempted || !sent {
t.Fatalf("retry queueStateDelta() attempted=%v sent=%v error=%v", attempted, sent, err)
}
if client.stateSnapshot != currentSnapshot {
t.Fatal("client baseline did not advance after the delta was queued")
}
}