diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index ede9eaeb9..b9ba7e2fd 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -6774,10 +6774,13 @@ Over the state WebSocket that projection now travels the same per-client keyed delta transport as resources: once a full payload establishes the client's baseline, `connectedInfrastructureDelta` carries id-keyed JSON merge patches plus removed ids and order instead of re-shipping the whole -projection on every broadcast. An entry that cannot key by id drops the -whole field back to plain whole-payload diffing, and a client without a -baseline ignores the delta and heals on the next full payload, so the keyed -path can never corrupt the projection. +projection on every broadcast. `activeAlerts` rides the identical transport +as `activeAlertsDelta`, with one boundary held fixed: the client applies +alert deltas immediately on arrival — alert lifecycle truth never waits for +the operator-input idle gate that defers resource work. An entry that cannot +key by id drops the whole field back to plain whole-payload diffing, and a +client without a baseline ignores the delta and heals on the next full +payload, so the keyed path can never corrupt either array. That same install-command payload continuity now also applies when auth is optional: copied install and upgrade commands must omit token arguments entirely on token-optional Pulse instances rather than serializing a fake diff --git a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md index 4d9c241db..6ad9e75dc 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -257,7 +257,10 @@ id-keyed merge patches for changed items only (at 50-node scale, timestamp patches of a few hundred bytes instead of the full ~55KB projection every tick), and the client applies them to an isolated baseline then reconciles the store with reference-stable untouched items so the per-tick deep walk of -the whole projection is gone. Workload table rows derived from canonical +the whole projection is gone. Active alerts travel the same keyed transport +(`activeAlertsDelta`, replacing the ~37KB whole-array re-ship whenever any +alert changed) while their application stays immediate per the alerts +subsystem boundary. Workload table rows derived from canonical snapshots reuse the previous row object whenever the serialized row is unchanged, and a refresh that changes nothing returns the previous row array itself, so per-tick row identity churn stays bounded to guests whose data diff --git a/frontend-modern/browser-verification.json b/frontend-modern/browser-verification.json index 9465bf368..1dfcd8cdd 100644 --- a/frontend-modern/browser-verification.json +++ b/frontend-modern/browser-verification.json @@ -1,17 +1,16 @@ { "version": 1, - "base_sha": "20503086ffc1eca6a4bb2121fe2c2ebd67160a0b", - "verified_at": "2026-08-26T17:08:31Z", + "base_sha": "f3d34346528fb981631238a394672d375d7092e9", + "verified_at": "2026-08-26T17:23:36Z", "result": "passed", "changed_paths": [ "frontend-modern/src/stores/websocket.ts" ], "content_sha256": { - "frontend-modern/src/stores/websocket.ts": "c648fbef42e36c2ae8502d258a6da2ab009f3d6bb9718e36192ba05085c3b760" + "frontend-modern/src/stores/websocket.ts": "a1f05a8bbe44d9e99349d9f3651304823458a04fc20949ecc305f5bc22bab9af" }, "routes": [ - "/proxmox/overview", - "/settings/infrastructure" + "/proxmox/overview" ], "viewports": [ { @@ -24,17 +23,15 @@ } ], "states": [ - "isolated test backend built from this change serving the 50-node/929-guest mock estate over the state websocket", - "fresh socket connection: initialState ships the full connectedInfrastructure projection once to establish the client baseline", - "steady realtime ticks carrying connectedInfrastructureDelta merge patches with zero full projection re-ships", - "Connected systems settings surface rendering the delta-maintained projection with live per-item freshness", - "narrow 700px viewport rendering the same projection surface" + "isolated test backend built from this change serving the mock estate with churning alerts (13 active / 62 unacknowledged)", + "fresh socket: initialState ships the full activeAlerts array once to establish the client baseline", + "steady realtime ticks carrying activeAlertsDelta merge patches with zero full alert-array re-ships", + "alert badge and estate rendering live throughout on desktop and narrow viewports" ], "interactions": [ - "raw websocket capture: ticks carry connectedInfrastructureDelta at 0.7-4KB of id+lastSeen merge patches instead of the ~55-60KB full projection previously re-shipped every tick; the full array shipped exactly once at initialState", - "client message counting over 30s: 3 delta ticks consumed, 0 full projection re-ships, 140 workload rows mounted, no console errors", - "Connected systems surface observed live at /settings/infrastructure: 71 systems with per-item freshness reading 1-5s ago, values that now only travel as merge patches, updating across ticks", - "unit-verified reference stability: untouched projection items keep their store identity across a delta sync, so reconcile no longer deep-walks the whole projection per tick", - "narrow 700px emulated viewport rendered the projection surface with live freshness text" + "raw websocket capture over 40s: 4 activeAlertsDelta payloads at 184-1981 bytes replacing the previous ~37-75KB whole-array re-ship whenever any alert changed; zero full re-ships after initialState", + "client consumed all delta ticks with 140 workload rows mounted and no console errors", + "unit-verified: alert level escalation, new-alert arrival, and resolution removal propagate to the store immediately even while operator input is active, encoding the alerts-subsystem immediacy boundary", + "narrow 700px emulated viewport rendered the alerts badge and estate live" ] } diff --git a/frontend-modern/src/stores/__tests__/websocket-unified.test.ts b/frontend-modern/src/stores/__tests__/websocket-unified.test.ts index 63d2b970f..7055caf4a 100644 --- a/frontend-modern/src/stores/__tests__/websocket-unified.test.ts +++ b/frontend-modern/src/stores/__tests__/websocket-unified.test.ts @@ -1133,6 +1133,76 @@ describe('websocket store unified resource contract', () => { } }); + it('applies alert deltas immediately even while operator input is active', async () => { + const { store, dispose } = await createStoreHarness(); + try { + await waitForOpenTick(); + emitMessage({ + type: 'initialState', + data: { + resources: [], + lastUpdate: 100, + activeAlerts: [ + { id: 'alert-1', type: 'cpu', level: 'warning', resourceId: 'vm-1', value: 85 }, + ], + recentlyResolved: [], + }, + }); + expect(store.state.activeAlerts).toHaveLength(1); + expect(store.state.activeAlerts[0]?.level).toBe('warning'); + + // Alert lifecycle truth does not wait for input idle: a delta arriving + // mid-gesture escalates and adds alerts immediately. + window.dispatchEvent(new Event('scroll')); + emitMessage({ + type: 'rawData', + data: { + lastUpdate: 200, + activeAlertsDelta: { + upserts: [ + { id: 'alert-1', level: 'critical', value: 97 }, + { id: 'alert-2', type: 'memory', level: 'warning', resourceId: 'vm-2', value: 91 }, + ], + }, + }, + }); + const byId = new Map(store.state.activeAlerts.map((alert) => [alert.id, alert])); + expect(byId.get('alert-1')?.level).toBe('critical'); + expect(byId.get('alert-1')?.value).toBe(97); + expect(byId.get('alert-1')?.resourceId).toBe('vm-1'); + expect(byId.get('alert-2')?.level).toBe('warning'); + + // Resolution rides the same payload and is equally immediate. + emitMessage({ + type: 'rawData', + data: { + lastUpdate: 300, + activeAlertsDelta: { removed: ['alert-1'] }, + }, + }); + expect(store.state.activeAlerts.map((alert) => alert.id)).toEqual(['alert-2']); + } finally { + dispose(); + } + }); + + it('ignores alert deltas that arrive without an alert baseline', async () => { + const { store, dispose } = await createStoreHarness(); + try { + await waitForOpenTick(); + emitMessage({ + type: 'rawData', + data: { + lastUpdate: 200, + activeAlertsDelta: { upserts: [{ id: 'alert-1', level: 'warning' }] }, + }, + }); + expect(store.state.activeAlerts).toEqual([]); + } finally { + dispose(); + } + }); + it('drops scroll-queued deltas when a full snapshot supersedes their baseline', async () => { const { store, dispose } = await createStoreHarness(); try { diff --git a/frontend-modern/src/stores/websocket.ts b/frontend-modern/src/stores/websocket.ts index 465fb9173..354a2425e 100644 --- a/frontend-modern/src/stores/websocket.ts +++ b/frontend-modern/src/stores/websocket.ts @@ -464,6 +464,10 @@ export function createWebSocketStore(url: string) { let rawConnectedInfrastructure: ConnectedInfrastructureItem[] | null = null; const pendingInfrastructureIds = new Set(); let pendingInfrastructureFull = false; + // Active alerts ride the same keyed delta transport, but their application + // is never gated: alert lifecycle updates stay immediate during operator + // input (alerts subsystem contract), so this baseline only slims the wire. + let rawActiveAlerts: Alert[] | null = null; // lastUpdate is the realtime tick token every downstream consumer keys on // (unified resource projections, workload remaps). It defers with the rest // so a mid-interaction tick triggers no derived recomputation at all. @@ -569,6 +573,7 @@ export function createWebSocketStore(url: string) { rawConnectedInfrastructure = null; pendingInfrastructureIds.clear(); pendingInfrastructureFull = false; + rawActiveAlerts = null; deferredLastUpdate = null; lastFullStateRecoveryAt = 0; oversizedSnapshotObserved = false; @@ -1156,15 +1161,41 @@ export function createWebSocketStore(url: string) { } // Sync active alerts from state if (message.data.activeAlerts !== undefined) { + const alertsArray = + message.data.activeAlerts && Array.isArray(message.data.activeAlerts) + ? (message.data.activeAlerts as Alert[]) + : []; + // A full payload re-establishes the keyed alert baseline. + rawActiveAlerts = structuredClone(alertsArray); const newAlerts: Record = {}; - if (message.data.activeAlerts && Array.isArray(message.data.activeAlerts)) { - message.data.activeAlerts.forEach((alert: Alert) => { - newAlerts[alert.id] = alert; - }); - } + alertsArray.forEach((alert: Alert) => { + newAlerts[alert.id] = alert; + }); lastActiveAlertsPayload = newAlerts; applyActiveAlerts(alertsEnabled ? newAlerts : {}); + } else if ( + 'activeAlertsDelta' in message.data && + message.data.activeAlertsDelta !== undefined + ) { + if (rawActiveAlerts !== null) { + const appliedAlerts = applyKeyedStateDelta( + rawActiveAlerts, + message.data.activeAlertsDelta as KeyedStateDelta, + ); + rawActiveAlerts = appliedAlerts.entries; + // Cloned on the way out so the alerts store never adopts + // baseline-owned objects. Applied immediately, input-active or + // not: alert lifecycle truth does not wait for idle. + const newAlerts: Record = {}; + for (const alert of rawActiveAlerts) { + newAlerts[alert.id] = structuredClone(alert); + } + lastActiveAlertsPayload = newAlerts; + applyActiveAlerts(alertsEnabled ? newAlerts : {}); + } + // No baseline: leave current alerts; the next full payload + // (reconnect or snapshot) re-establishes it. } // Sync recently resolved alerts if (message.data.recentlyResolved !== undefined) { diff --git a/internal/websocket/state_delta.go b/internal/websocket/state_delta.go index 2e73506d6..4cc37294d 100644 --- a/internal/websocket/state_delta.go +++ b/internal/websocket/state_delta.go @@ -11,20 +11,35 @@ import ( const resourceDeltaField = "resourceDelta" const infrastructureField = "connectedInfrastructure" const infrastructureDeltaField = "connectedInfrastructureDelta" +const activeAlertsField = "activeAlerts" +const activeAlertsDeltaField = "activeAlertsDelta" + +// Keyed delta fields are id-keyed arrays that travel as per-item merge +// patches once a client baseline exists, instead of re-shipping the whole +// array on every broadcast. When a payload cannot be keyed (an entry without +// an id), it stays in fields and diffs whole, so the keyed path can never +// corrupt the projection. +var keyedDeltaFields = []struct { + field string + deltaField string +}{ + {infrastructureField, infrastructureDeltaField}, + {activeAlertsField, activeAlertsDeltaField}, +} + +type keyedFieldSnapshot struct { + entries map[string]json.RawMessage + order []string + raw json.RawMessage +} type clientStateSnapshot struct { fields map[string]json.RawMessage resources map[string]json.RawMessage resourceOrder []string - // The connected-infrastructure projection is keyed the same way as - // resources so per-item merge patches replace re-shipping the whole - // projection on every broadcast. When the payload cannot be keyed (an - // entry without an id), it stays in fields and diffs whole, so the - // keyed path can never corrupt the projection. - infrastructure map[string]json.RawMessage - infrastructureOrder []string - infrastructureRaw json.RawMessage - infrastructureKeyed bool + // Keyed snapshots by field name; a field is present only when every + // entry keyed by id. + keyed map[string]*keyedFieldSnapshot } type resourceDeltaPayload struct { @@ -87,20 +102,24 @@ func buildClientStateSnapshot(state interface{}) (*clientStateSnapshot, error) { fields: fields, resources: resources, resourceOrder: resourceOrder, + keyed: make(map[string]*keyedFieldSnapshot), } - if encodedInfrastructure, ok := fields[infrastructureField]; ok { - infrastructure, infrastructureOrder, keyErr := extractKeyedEntries( - encodedInfrastructure, - infrastructureField, - ) - if keyErr == nil { - snapshot.infrastructure = infrastructure - snapshot.infrastructureOrder = infrastructureOrder - snapshot.infrastructureRaw = append(json.RawMessage(nil), encodedInfrastructure...) - snapshot.infrastructureKeyed = true - delete(fields, infrastructureField) + for _, keyedField := range keyedDeltaFields { + encodedField, ok := fields[keyedField.field] + if !ok { + continue } + entries, order, keyErr := extractKeyedEntries(encodedField, keyedField.field) + if keyErr != nil { + continue + } + snapshot.keyed[keyedField.field] = &keyedFieldSnapshot{ + entries: entries, + order: order, + raw: append(json.RawMessage(nil), encodedField...), + } + delete(fields, keyedField.field) } return snapshot, nil @@ -175,35 +194,40 @@ func buildClientStateDelta(previous, current *clientStateSnapshot) (map[string]i delta[resourceDeltaField] = resourceDelta } - switch { - case previous.infrastructureKeyed && current.infrastructureKeyed: - infrastructureDelta, err := buildKeyedArrayDelta( - previous.infrastructure, - current.infrastructure, - previous.infrastructureOrder, - current.infrastructureOrder, - infrastructureField, - ) - if err != nil { - return nil, err - } - if !infrastructureDelta.isEmpty() { - delta[infrastructureDeltaField] = infrastructureDelta - } - case current.infrastructureKeyed: - // The previous snapshot carried the projection as a plain field (or - // not at all): fall back to shipping the full array once so the - // client re-adopts a clean baseline. - if previousRaw, ok := previous.fields[infrastructureField]; !ok || - !bytes.Equal(previousRaw, current.infrastructureRaw) { - delta[infrastructureField] = current.infrastructureRaw - } - case previous.infrastructureKeyed: - // The projection left the keyed path. If the current snapshot still - // carries it as a plain field, the fields loop above already shipped - // it whole; if it vanished entirely, clear it like any removed field. - if _, ok := current.fields[infrastructureField]; !ok { - delta[infrastructureField] = nil + for _, keyedField := range keyedDeltaFields { + previousSnapshot := previous.keyed[keyedField.field] + currentSnapshot := current.keyed[keyedField.field] + switch { + case previousSnapshot != nil && currentSnapshot != nil: + keyedDelta, err := buildKeyedArrayDelta( + previousSnapshot.entries, + currentSnapshot.entries, + previousSnapshot.order, + currentSnapshot.order, + keyedField.field, + ) + if err != nil { + return nil, err + } + if !keyedDelta.isEmpty() { + delta[keyedField.deltaField] = keyedDelta + } + case currentSnapshot != nil: + // The previous snapshot carried the array as a plain field (or + // not at all): fall back to shipping the full array once so the + // client re-adopts a clean baseline. + if previousRaw, ok := previous.fields[keyedField.field]; !ok || + !bytes.Equal(previousRaw, currentSnapshot.raw) { + delta[keyedField.field] = currentSnapshot.raw + } + case previousSnapshot != nil: + // The array left the keyed path. If the current snapshot still + // carries it as a plain field, the fields loop above already + // shipped it whole; if it vanished entirely, clear it like any + // removed field. + if _, ok := current.fields[keyedField.field]; !ok { + delta[keyedField.field] = nil + } } } diff --git a/internal/websocket/state_delta_test.go b/internal/websocket/state_delta_test.go index d4040c117..33dddcf11 100644 --- a/internal/websocket/state_delta_test.go +++ b/internal/websocket/state_delta_test.go @@ -200,7 +200,7 @@ func TestClientStateDeltaFallsBackToFullInfrastructureWhenUnkeyed(t *testing.T) if err != nil { t.Fatal(err) } - if previous.infrastructureKeyed { + if previous.keyed[infrastructureField] != nil { t.Fatal("id-less projection must not key") } current, err := buildClientStateSnapshot(currentState) @@ -251,6 +251,89 @@ func TestClientStateDeltaShipsFullInfrastructureWhenBaselineWasUnkeyed(t *testin } } +func TestClientStateDeltaShipsActiveAlertsAsKeyedPatches(t *testing.T) { + previousState := models.EmptyStateFrontend() + previousState.ActiveAlerts = []models.Alert{ + { + ID: "alert-1", Type: "cpu", Level: "warning", ResourceID: "vm-1", + Message: strings.Repeat("m", 4*1024), Value: 85, + }, + {ID: "alert-2", Type: "memory", Level: "warning", ResourceID: "vm-2", Value: 70}, + } + currentState := previousState + currentState.ActiveAlerts = append([]models.Alert(nil), previousState.ActiveAlerts...) + currentState.ActiveAlerts[0].Level = "critical" + currentState.ActiveAlerts[0].Value = 97 + + 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) + } + if _, ok := delta[activeAlertsField]; ok { + t.Fatal("state delta re-shipped the full activeAlerts array") + } + payload, ok := delta[activeAlertsDeltaField].(resourceDeltaPayload) + if !ok { + t.Fatalf("alerts delta type = %T", delta[activeAlertsDeltaField]) + } + if len(payload.Upserts) != 1 { + t.Fatalf("upserts = %d, want 1", len(payload.Upserts)) + } + patch := string(payload.Upserts[0]) + if !strings.Contains(patch, `"id":"alert-1"`) || !strings.Contains(patch, `"level":"critical"`) { + t.Fatalf("patch = %s, want id + level", patch) + } + if strings.Contains(patch, strings.Repeat("m", 128)) { + t.Fatal("patch re-shipped the unchanged alert message") + } + if len(patch) >= 256 { + t.Fatalf("patch = %d bytes, want a bounded lifecycle patch", len(patch)) + } +} + +func TestClientStateDeltaTracksResolvedAlertRemoval(t *testing.T) { + previousState := models.EmptyStateFrontend() + previousState.ActiveAlerts = []models.Alert{ + {ID: "alert-1", Type: "cpu", Level: "warning", ResourceID: "vm-1"}, + {ID: "alert-2", Type: "memory", Level: "warning", ResourceID: "vm-2"}, + } + currentState := models.EmptyStateFrontend() + currentState.ActiveAlerts = []models.Alert{ + {ID: "alert-2", Type: "memory", Level: "warning", ResourceID: "vm-2"}, + } + + 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) + } + payload, ok := delta[activeAlertsDeltaField].(resourceDeltaPayload) + if !ok { + t.Fatalf("alerts delta type = %T", delta[activeAlertsDeltaField]) + } + if len(payload.Removed) != 1 || payload.Removed[0] != "alert-1" { + t.Fatalf("removed = %#v, want [alert-1]", payload.Removed) + } + if len(payload.Upserts) != 0 { + t.Fatalf("upserts = %d, want none", len(payload.Upserts)) + } +} + func TestClientStateBaselineAdvancesOnlyAfterDeltaIsQueued(t *testing.T) { initial := models.EmptyStateFrontend() initial.LastUpdate = 100