diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index 96d6354fd..27a1998c4 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -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 diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index a98c7c9ea..14e92fa3c 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -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 diff --git a/frontend-modern/browser-verification.json b/frontend-modern/browser-verification.json index 10a4e4335..015b390f1 100644 --- a/frontend-modern/browser-verification.json +++ b/frontend-modern/browser-verification.json @@ -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" ] } diff --git a/frontend-modern/src/stores/__tests__/websocket-resilience.test.ts b/frontend-modern/src/stores/__tests__/websocket-resilience.test.ts index bf9f6f7e8..bfb32c94d 100644 --- a/frontend-modern/src/stores/__tests__/websocket-resilience.test.ts +++ b/frontend-modern/src/stores/__tests__/websocket-resilience.test.ts @@ -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 { diff --git a/frontend-modern/src/stores/__tests__/websocket-unified.test.ts b/frontend-modern/src/stores/__tests__/websocket-unified.test.ts index 59be54ddd..26fc6c28f 100644 --- a/frontend-modern/src/stores/__tests__/websocket-unified.test.ts +++ b/frontend-modern/src/stores/__tests__/websocket-unified.test.ts @@ -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 { diff --git a/frontend-modern/src/stores/websocket.ts b/frontend-modern/src/stores/websocket.ts index 4badb7085..c1d635fdc 100644 --- a/frontend-modern/src/stores/websocket.ts +++ b/frontend-modern/src/stores/websocket.ts @@ -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(); const asRecord = (value: unknown): Record | undefined => value && typeof value === 'object' ? (value as Record) : undefined; +const asMergePatchRecord = (value: unknown): Record | undefined => + value && typeof value === 'object' && !Array.isArray(value) + ? (value as Record) + : 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 = { ...(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(); + 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) || [])], diff --git a/frontend-modern/src/types/api.ts b/frontend-modern/src/types/api.ts index 336b142ef..21b91051d 100644 --- a/frontend-modern/src/types/api.ts +++ b/frontend-modern/src/types/api.ts @@ -33,6 +33,16 @@ export interface State { resources: Resource[]; } +export interface ResourceStateDelta { + upserts?: Array & Pick>; + removed?: string[]; + order?: string[]; +} + +export type StateUpdate = Partial & { + resourceDelta?: ResourceStateDelta; +}; + export interface PVETagStyle { colors: Record; 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 } diff --git a/internal/alerts/issue1497_test.go b/internal/alerts/issue1497_test.go new file mode 100644 index 000000000..f0617ba78 --- /dev/null +++ b/internal/alerts/issue1497_test.go @@ -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) + } +} diff --git a/internal/websocket/hub.go b/internal/websocket/hub.go index c500be3f5..a7f2b6beb 100644 --- a/internal/websocket/hub.go +++ b/internal/websocket/hub.go @@ -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 { diff --git a/internal/websocket/state_delta.go b/internal/websocket/state_delta.go new file mode 100644 index 000000000..9c863a3fb --- /dev/null +++ b/internal/websocket/state_delta.go @@ -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, ¤tValue); 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 +} diff --git a/internal/websocket/state_delta_test.go b/internal/websocket/state_delta_test.go new file mode 100644 index 000000000..dd31d41e0 --- /dev/null +++ b/internal/websocket/state_delta_test.go @@ -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") + } +}