mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-09-10 02:25:56 +00:00
Slim the resource stream and catch up realtime gaps incrementally
The 2026-08-25 audit follow-up profiled the remaining frontend costs on the 50-node mock rig. Entering a tab more than one websocket tick behind the shared cache fell off the incremental path onto a full-estate deep-unwrap merge — a 2.1s unwrap dominated warm Alerts entry at 4x throttle — and the client-facing resource stream re-shipped ~3KB of mostly-static metadata per resource on every snapshot, REST recovery, and reconnect (governed gap resource-payload-static-metadata). The connection store now keeps a bounded per-tick changed-id history and serves unions from it, so an instance that mounts or resumes a few revisions behind catches up with a delta merge over the union instead of remerging the estate. Uncovered gaps and full-snapshot commits still fall back to keyed full reconciliation. Broadcast payloads dedupe the estate's distinct capability blobs into a content-addressed state-level capabilityCatalog referenced per resource by capabilitiesRef; default-posture resources (internal sensitivity, cloud-summary routing, no redactions) omit policy and aiSafeSummary; and canonicalIdentity.aliases no longer duplicates supersededIds. Ingestion reverses all three: refs expand through the catalog as per-row clones (reconcile mutates adopted objects in place), missing policies synthesize the default posture so a policy:null transition patch cannot strand a stale governed policy, and identity alias resolution consults supersededIds explicitly. Slimming edits only the per-broadcast copy, never stored monitor state, and the delta engine's generic top-level field diff carries catalog changes in the same frame as the first resource referencing a new entry. Rig A/B on the pinned 50-node mock (same VM as the morning baseline, single runs, +-30% variance; RandomMetrics mutates nearly every row per tick, so this estate is the worst case for the catch-up union): mobile 4x Alerts entry settles 3.9s -> 2.0s with long tasks 5.5s -> 3.3s, idle-30s long tasks drop 12.6s -> 10.8s mobile and 3.4s -> 2.8s desktop, remaining steps flat within variance. Sparse-delta estates gain more, because their catch-up unions stay small. /api/state on the pinned mock estate drops 4.75MB -> 4.09MB (-13.9%); the capability catalog is 7 entries totalling 2.5KB where 946 resources previously inlined the blobs, and 806 default-posture resources shed policy and aiSafeSummary. Browser-verified against a live build of this tree: byte-identical UI outcomes vs the unmodified baseline across overview/storage/docker/alerts probes at both viewports, with docker container lifecycle actions enabled through catalog-expanded capabilities. pulse-mobile and pulse-enterprise do not read the slimmed fields; the AI runtime consumes the internal model, which is unchanged. Contracts for monitoring, api-contracts, unified-resources, performance-and-scalability, alerts, agent-lifecycle, and storage-recovery are amended with the slimmed wire shape, the ingestion synthesis obligations, the catch-up history, and superseded-id resolution.
This commit is contained in:
rcourtman
+35
@@ -50,3 +50,38 @@ contract updates and cross-client verification, not an opportunistic patch.
|
||||
start no longer fetches and compiles all ~3.1 MB of lazy chunks up front.
|
||||
- Per-tab scoped hydration and realtime gating on the Proxmox surface
|
||||
(`17bb2b3b7`, Performance lane).
|
||||
|
||||
## 2026-08-25 partial remediation (interactive session, governed claim on this gap)
|
||||
|
||||
Landed on `main` (same-day follow-up to the tick-pipeline fixes in
|
||||
`7bac525af`):
|
||||
|
||||
- **Capabilities catalog**: broadcast payloads dedupe the estate's distinct
|
||||
capability blobs into a content-addressed state-level `capabilityCatalog`
|
||||
referenced per resource via `capabilitiesRef`; the websocket store expands
|
||||
refs back to inline `capabilities` at ingestion. On the 50-node mock the
|
||||
catalog is 7 entries / 2.5KB where 946 resources previously inlined the
|
||||
blobs.
|
||||
- **Audience-scoped policy metadata**: default-posture resources (internal
|
||||
sensitivity, cloud-summary routing, no redactions) omit `policy` and
|
||||
`aiSafeSummary` from the stream; ingestion synthesizes the default posture
|
||||
so consumer semantics are unchanged. 806 of 1,509 mock resources shed both
|
||||
fields.
|
||||
- **Alias dedupe**: broadcast `canonicalIdentity.aliases` no longer
|
||||
duplicates `supersededIds`; client identity resolution now consults
|
||||
`supersededIds` explicitly (alert overrides, thresholds, and workload
|
||||
matching already did).
|
||||
|
||||
Measured: `/api/state` 4.75MB -> 4.09MB (-13.9%) at the pinned mock estate.
|
||||
Cross-client check: pulse-mobile and pulse-enterprise contain no reads of
|
||||
`capabilities`/`canonicalIdentity`/`aiSafeSummary`/resource `policy` from
|
||||
this stream; the AI runtime consumes the internal resource model, which is
|
||||
untouched.
|
||||
|
||||
**Remaining residual (why the gap stays open):** the canonicalIdentity alias
|
||||
vocabulary itself (~0.8MB) still ships inline because superseded-spelling
|
||||
resolution is load-bearing client-side while
|
||||
`host-identity-fork-heal-on-reenrollment` remains open; moving identity
|
||||
history behind a detail endpoint needs the alert-override/threshold matching
|
||||
reworked onto an on-demand lookup. Platform source payloads (~1.4MB) are
|
||||
live data, not static metadata, and are out of scope for this gap.
|
||||
|
||||
@@ -65,6 +65,12 @@ or rename correlations.
|
||||
websocket payloads. Carrying plural availability facets through that serializer
|
||||
is an adjacent monitoring/API projection and does not change agent enrollment,
|
||||
report admission, removal, update, profile, or command authority.
|
||||
The serializer's broadcast slimming is likewise an adjacent projection with an
|
||||
agent-identity obligation: superseded canonical ids produced by re-enrollment
|
||||
or identity re-derivation must remain resolvable to the client through
|
||||
`canonicalIdentity.supersededIds` even though they are no longer duplicated
|
||||
into the broadcast alias list, so alert overrides and identity consumers keep
|
||||
following a forked or re-enrolled host across its retired spellings.
|
||||
That shared monitor constructor may also copy the persisted alert schedule's
|
||||
initial notification target into the notification manager. This adjacent
|
||||
alerts/notifications wiring grants no agent enrollment, reporting, removal,
|
||||
|
||||
@@ -247,6 +247,13 @@ default construction path still restores.
|
||||
## Shared Boundaries
|
||||
|
||||
1. `frontend-modern/src/stores/websocket.ts` shared with `performance-and-scalability`: the connection-owned realtime store is both the canonical alert truth boundary and the fleet-scale resource reconciliation hot path.
|
||||
That shared store normalizes slimmed broadcast resources at ingestion —
|
||||
expanding `capabilitiesRef` through the state `capabilityCatalog` and
|
||||
synthesizing the default policy posture for resources published without one
|
||||
— before any alert consumer resolves alert-to-resource identity or policy
|
||||
display. Alert override and threshold identity matching must consult
|
||||
`canonicalIdentity.supersededIds` alongside aliases, because broadcast
|
||||
aliases no longer duplicate superseded canonical ids.
|
||||
2. `internal/operationaltrust/contracts.go` shared with `notifications`: the operational trust contract is jointly consumed by canonical alert lifecycle ownership and notification delivery linkage without making delivery state operational truth.
|
||||
3. `internal/proxmoxidentity/backup_identity.go` shared with `monitoring`, `storage-recovery`: Proxmox PBS backup subject identity is a shared runtime boundary for monitoring backup freshness, backup-age alert attribution, and recovery-point guest mapping.
|
||||
Alert multiline field presentation is shared with frontend-primitives:
|
||||
|
||||
@@ -2580,6 +2580,18 @@ a new API state machine, queue contract, or verification-accounting field.
|
||||
host-agent row for the same machine must leave the API boundary as one
|
||||
hybrid `agent` resource with merged source facets, not as duplicate rows
|
||||
that disappear only after REST reconciliation.
|
||||
Realtime `/api/state` and websocket state payloads carry deduped static
|
||||
metadata: the state-level `capabilityCatalog` maps content-addressed ids to
|
||||
the estate's distinct capability blobs and each resource references its set
|
||||
via `capabilitiesRef` instead of inlining the blob per resource; a resource
|
||||
published without an inline `policy` is, by contract, at the default posture
|
||||
(internal sensitivity, cloud-summary routing, no redactions) and its
|
||||
`aiSafeSummary` is withheld with it, while non-default postures keep both
|
||||
inline; and `canonicalIdentity.aliases` no longer duplicates
|
||||
`supersededIds`, which stay the canonical superseded-spelling carrier for
|
||||
identity resolution. Because the websocket delta engine diffs top-level
|
||||
state fields generically, a catalog change must ride the same frame as the
|
||||
first resource referencing the new entry.
|
||||
Realtime `/api/state` and websocket state snapshots also own Proxmox tag
|
||||
presentation payloads. `pveTagStyles` is keyed by Proxmox instance and
|
||||
carries the parsed datacenter `tag-style` color overrides plus
|
||||
|
||||
@@ -1503,6 +1503,18 @@ canonical unified-resource metrics include `diskRead` or `diskWrite`,
|
||||
`ResourceFrontend.diskIO` through the shared resource converter, so
|
||||
`/api/state` and websocket consumers read disk throughput from the same
|
||||
freshness-gated resource metrics contract as CPU, memory, disk, and network.
|
||||
That same broadcast projection owns client-payload static-metadata slimming
|
||||
(governed gap `resource-payload-static-metadata`): identical resource
|
||||
capability blobs are deduped into the state-level `capabilityCatalog` under
|
||||
content-addressed ids and referenced per resource via `capabilitiesRef`
|
||||
instead of being inlined on every row; resources whose derived policy posture
|
||||
is the default (internal sensitivity, cloud-summary routing, no redactions)
|
||||
omit `policy` and `aiSafeSummary` from the broadcast, which ingestion
|
||||
synthesizes back, while non-default postures keep both inline; and broadcast
|
||||
`canonicalIdentity.aliases` entries that duplicate `supersededIds` are
|
||||
dropped, with the superseded ids still shipped for identity resolution.
|
||||
Slimming edits only the per-broadcast copy refreshed by
|
||||
`RefreshCanonicalMetadata`, never stored monitor state.
|
||||
Unraid ingest must preserve the agent's native disk topology fields through the
|
||||
monitoring model and read-state projection. `internal/monitoring/monitor_agents.go`
|
||||
and `internal/monitoring/monitor.go` must carry model, transport, filesystem,
|
||||
|
||||
@@ -222,7 +222,17 @@ cache once, then projects route filters from the already-canonical result;
|
||||
route-scoped consumers must not independently canonicalize the full estate. A
|
||||
hook instance observing an already-applied revision must not deep-unwrap the
|
||||
realtime store, and the merging instance dereferences raw store subtrees only
|
||||
for rows the delta merge will clone. Workload table rows derived from canonical
|
||||
for rows the delta merge will clone. The connection store keeps a bounded
|
||||
per-revision changed-id history so an instance that mounts or resumes several
|
||||
revisions behind the shared cache catches up through the unioned changed-id
|
||||
set as a delta merge instead of a full-estate remerge; the full fallback is
|
||||
reserved for uncovered gaps and full-snapshot commits. The broadcast payload
|
||||
itself is a scalability surface: duplicated capability blobs travel once
|
||||
through the state-level `capabilityCatalog` and per-resource
|
||||
`capabilitiesRef`, default-posture policy and AI-safe prose are omitted and
|
||||
synthesized at ingestion, and superseded canonical ids are not duplicated
|
||||
into the alias list, so full snapshots, REST recovery, and reconnects do not
|
||||
re-ship static metadata per resource. 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
|
||||
|
||||
@@ -1126,9 +1126,11 @@ recovery scope, or a storage/recovery-owned secret source.
|
||||
must invalidate stale in-flight REST refreshes before their errors or
|
||||
request-guard cleanup can leak into the active resource snapshot.
|
||||
The shared hook's realtime tick path may read the connection store lazily —
|
||||
skipping revisions the shared canonical cache already holds and
|
||||
dereferencing raw store subtrees only for rows the delta merge will clone —
|
||||
but that optimization must stay invisible to storage/recovery consumers: it
|
||||
skipping revisions the shared canonical cache already holds, catching up
|
||||
multi-revision gaps through the store's bounded changed-id history union,
|
||||
and dereferencing raw store subtrees only for rows the delta merge will
|
||||
clone — but that optimization must stay invisible to storage/recovery
|
||||
consumers: it
|
||||
must not change which canonical fields a merged row carries, must not
|
||||
reorder the REST-first hydration guarantee above, and a skipped store read
|
||||
must never leave a storage/recovery projection behind the shared cache's
|
||||
|
||||
@@ -1180,8 +1180,13 @@ AI-only summary payloads, or page-local heuristics.
|
||||
instance dereferences raw store subtrees only for rows the delta merge will
|
||||
clone (flagged ids, host-merge members, and ids absent from the shared
|
||||
cache). A sequential revision with unchanged route membership
|
||||
patches only the changed row indices plus the bounded agent coalescing set;
|
||||
initial hydration, missed revisions, additions, removals, or reorderings
|
||||
patches only the changed row indices plus the bounded agent coalescing set.
|
||||
The connection store retains a bounded per-revision changed-id history; an
|
||||
instance that resumes several revisions behind the shared cache must catch
|
||||
up through the unioned changed-id set as an incremental delta merge whenever
|
||||
the history covers the gap, so tab entry and re-entry do not deep-unwrap or
|
||||
remerge the full estate. Only initial hydration, uncovered revision gaps,
|
||||
full-snapshot commits, additions, removals, or reorderings
|
||||
fall back to keyed full reconciliation. Route-prefetch and route-realtime
|
||||
activation are separate:
|
||||
a prefetched hidden surface may retain REST data without subscribing its full
|
||||
@@ -1189,6 +1194,16 @@ AI-only summary payloads, or page-local heuristics.
|
||||
cache. Richer REST-only facets are promoted into that cache before thinner
|
||||
realtime deltas are applied, so the optimization cannot discard disk I/O,
|
||||
PBS, policy, or provider metadata.
|
||||
Broadcast payload slimming is reversed at the connection-store ingestion
|
||||
boundary, before any canonical merge or consumer read: `capabilitiesRef` is
|
||||
expanded into per-row inline `capabilities` through the state payload's
|
||||
`capabilityCatalog` (per-row clones, because store reconciliation mutates
|
||||
adopted objects in place), and a resource arriving without a policy is given
|
||||
a synthesized default posture (internal sensitivity, cloud-summary routing,
|
||||
no redactions) so a posture transition patched as `policy: null` cannot
|
||||
leave a stale governed policy behind. Client identity-alias resolution must
|
||||
consult `canonicalIdentity.supersededIds` explicitly, because broadcast
|
||||
aliases no longer duplicate superseded canonical ids.
|
||||
That same unified-resource owner also defines the canonical transport
|
||||
projection for operator-facing resources: `/api/resources` and websocket
|
||||
`state.resources` must share `ContractResourceType`, canonical display
|
||||
|
||||
@@ -1,26 +1,35 @@
|
||||
{
|
||||
"version": 1,
|
||||
"base_sha": "7bac525af355bc5c8eb6ff2db8f9ce116697013f",
|
||||
"verified_at": "2026-08-25T10:44:58Z",
|
||||
"base_sha": "51b7357285b0ac7020212aff0a67b8cd5974757c",
|
||||
"verified_at": "2026-08-25T11:20:27Z",
|
||||
"result": "passed",
|
||||
"changed_paths": [
|
||||
"frontend-modern/src/components/shared/horizontalRailVisibilityModel.ts",
|
||||
"frontend-modern/src/components/shared/useActiveHorizontalRailItemVisibility.ts",
|
||||
"frontend-modern/src/features/platformPage/sharedPlatformPage.tsx"
|
||||
"frontend-modern/src/hooks/useUnifiedResources.ts",
|
||||
"frontend-modern/src/stores/websocket-global.ts",
|
||||
"frontend-modern/src/stores/websocket.ts",
|
||||
"frontend-modern/src/types/api.ts",
|
||||
"frontend-modern/src/types/resource.ts",
|
||||
"frontend-modern/src/utils/resourceIdentity.ts"
|
||||
],
|
||||
"content_sha256": {
|
||||
"frontend-modern/src/components/shared/horizontalRailVisibilityModel.ts": "1fa3ad5d3cfffd0ccf78ced497d87564602cb124103175c17ba6c3e4b9531c4f",
|
||||
"frontend-modern/src/components/shared/useActiveHorizontalRailItemVisibility.ts": "add047b223402743fbabf5d7c9b867f160174ab2e48baf53c0c60509caf01510",
|
||||
"frontend-modern/src/features/platformPage/sharedPlatformPage.tsx": "3507374c3c9c236a95ee50793f2462b16d71680f13c1f4c864efc3ea88159659"
|
||||
"frontend-modern/src/hooks/useUnifiedResources.ts": "5b63ab55ce28b72216561acf7b818c55179e70acfc2a35155cd6391490a13f54",
|
||||
"frontend-modern/src/stores/websocket-global.ts": "59d2ddc55882f98c72b5c893ce16c250ba5c0eeb20e31419e23f651ab72c00b5",
|
||||
"frontend-modern/src/stores/websocket.ts": "56c7fa58dc5f79855b6ab4a9c9c36c85f00a554a76ea981b593e175a4547dba1",
|
||||
"frontend-modern/src/types/api.ts": "88a40975498e284b3c704b384a6a2ab52bd1fd19e6fd1de522953d570f2afa25",
|
||||
"frontend-modern/src/types/resource.ts": "9b73aa4df0b734f37f19f1b607872442ea988f5212d6bdb2a59be4b613d83249",
|
||||
"frontend-modern/src/utils/resourceIdentity.ts": "f0a0203a2e5de11afcffa1ae70382c9e55249e9a5651b0fb9611b3f0f196fb51"
|
||||
},
|
||||
"routes": [
|
||||
"/login",
|
||||
"/proxmox/overview",
|
||||
"/proxmox/replication"
|
||||
"/proxmox/storage",
|
||||
"/docker",
|
||||
"/alerts"
|
||||
],
|
||||
"viewports": [
|
||||
{
|
||||
"width": 1440,
|
||||
"height": 900
|
||||
"width": 1280,
|
||||
"height": 800
|
||||
},
|
||||
{
|
||||
"width": 390,
|
||||
@@ -28,16 +37,20 @@
|
||||
}
|
||||
],
|
||||
"states": [
|
||||
"Proxmox overview tab rail with all six destinations at desktop width",
|
||||
"Proxmox overview tab rail initially clipped at phone width with the right affordance visible",
|
||||
"phone tab rail manually scrolled to its 207px end position with the left affordance visible",
|
||||
"manually scrolled phone rail after a same-width viewport-height resize",
|
||||
"Replication selected from the manually explored rail with the active label clear of the left overlay"
|
||||
"authenticated session against a dev build of this exact diff serving the 50-node mock estate with the slimmed broadcast payload (capabilityCatalog with 7 entries, zero inline capabilities, 806 default-posture policies and aiSafeSummary omitted, aliases deduped against supersededIds)",
|
||||
"Proxmox overview table populated at both viewports with live 2s websocket ticks applying",
|
||||
"Proxmox storage sub-tab populated (sensitive-posture resources keep inline policy)",
|
||||
"Docker workloads table with an expanded container row and its lifecycle actions group",
|
||||
"container lifecycle menu open: Start disabled with running reason, Stop and Restart enabled via catalog-expanded capabilities",
|
||||
"Alerts utility surface rendering the active-alert list",
|
||||
"console error stream captured on both builds: only the pre-existing CSP inline-style warnings, no new errors"
|
||||
],
|
||||
"interactions": [
|
||||
"native horizontal wheel/swipe projection across the phone-width Proxmox section rail",
|
||||
"phone viewport resize after manual rail exploration preserved scrollLeft at 207px",
|
||||
"selected Replication from the revealed rail and confirmed route navigation plus active-tab visibility",
|
||||
"repeated the final phone flow after the active-label clearance adjustment: swipe to 190px, resize without snap-back, then select Replication at a safe 55px viewport offset"
|
||||
"logged in and waited for landing hydration",
|
||||
"switched platform tabs to Docker and sub-tabs to Storage and back to Overview",
|
||||
"expanded a docker container row and opened its container-actions menu, reading each action's enabled/disabled state and aria-label",
|
||||
"navigated to Alerts and back",
|
||||
"fetched /api/state on the slimmed build and asserted capabilityCatalog presence, per-resource capabilitiesRef, omitted default policies, and the 4.75MB to 4.09MB size drop",
|
||||
"ran the identical probe script against the unmodified baseline build and confirmed byte-identical UI outcomes (rows, drawer, action states, console errors)"
|
||||
]
|
||||
}
|
||||
|
||||
@@ -145,6 +145,7 @@ describe('useUnifiedResources', () => {
|
||||
version: number;
|
||||
changedIds: ReadonlySet<string> | null;
|
||||
}) => unknown;
|
||||
let wsStoreMock: Record<string, unknown>;
|
||||
let useUnifiedResources: UseUnifiedResourcesModule['useUnifiedResources'];
|
||||
let useStorageRecoveryResources: UseUnifiedResourcesModule['useStorageRecoveryResources'];
|
||||
let resetUnifiedResourcesCacheForTests: UseUnifiedResourcesModule['__resetUnifiedResourcesCacheForTests'];
|
||||
@@ -199,6 +200,7 @@ describe('useUnifiedResources', () => {
|
||||
setWsState = _setWsState;
|
||||
|
||||
const wsStore = { connected, initialDataReceived, state, resourceChange };
|
||||
wsStoreMock = wsStore as unknown as Record<string, unknown>;
|
||||
|
||||
vi.doMock('@/utils/apiClient', () => ({
|
||||
apiFetch: apiFetchMock,
|
||||
@@ -303,6 +305,65 @@ describe('useUnifiedResources', () => {
|
||||
dispose();
|
||||
});
|
||||
|
||||
it('catches up multi-tick version gaps through the changed-id history instead of a full remerge', async () => {
|
||||
// Distinct nested cpu objects: the shared wsResource base would otherwise
|
||||
// alias one cpu object across both rows and the two writes below.
|
||||
const vm1 = createWsResource({
|
||||
id: 'vm-1',
|
||||
name: 'vm-1',
|
||||
displayName: 'vm-1',
|
||||
type: 'vm',
|
||||
cpu: { current: 15 },
|
||||
});
|
||||
const vm2 = createWsResource({
|
||||
id: 'vm-2',
|
||||
name: 'vm-2',
|
||||
displayName: 'vm-2',
|
||||
type: 'vm',
|
||||
cpu: { current: 15 },
|
||||
});
|
||||
setWsState('resources', [vm1, vm2]);
|
||||
setWsResourceChange({ version: 1, changedIds: null });
|
||||
|
||||
let dispose = () => {};
|
||||
let result: ReturnType<UseUnifiedResourcesModule['useUnifiedResources']> | undefined;
|
||||
createRoot((d) => {
|
||||
dispose = d;
|
||||
result = useUnifiedResources({
|
||||
query: '',
|
||||
cacheKey: 'all-resources',
|
||||
initialHydration: 'prefer-ws',
|
||||
});
|
||||
});
|
||||
|
||||
await waitForResourceCount(() => result!.resources().length);
|
||||
|
||||
const vm2RowBefore = result!.resources().find((resource) => resource.id === 'vm-2');
|
||||
|
||||
// Two ticks land while no instance applies them (version 1 -> 3). The
|
||||
// store's history union names only vm-1, so the delta path must re-merge
|
||||
// vm-1 and keep the unflagged vm-2 row by identity — a full remerge would
|
||||
// rebuild it.
|
||||
const historySpy = vi.fn().mockReturnValue(new Set(['vm-1']));
|
||||
wsStoreMock.changedResourceIdsSince = historySpy;
|
||||
batch(() => {
|
||||
setWsState('resources', 0, 'cpu', 'current', 88);
|
||||
setWsState('resources', 1, 'cpu', 'current', 99);
|
||||
setWsResourceChange({ version: 3, changedIds: new Set(['vm-1']) });
|
||||
setWsState('lastUpdate', 1738843203000);
|
||||
});
|
||||
|
||||
await waitForValue(
|
||||
() => result!.resources().find((resource) => resource.id === 'vm-1')?.cpu?.current,
|
||||
88,
|
||||
);
|
||||
expect(historySpy).toHaveBeenCalledWith(1);
|
||||
const vm2RowAfter = result!.resources().find((resource) => resource.id === 'vm-2');
|
||||
expect(vm2RowAfter).toBe(vm2RowBefore);
|
||||
expect(apiFetchMock).not.toHaveBeenCalled();
|
||||
dispose();
|
||||
});
|
||||
|
||||
it('hydrates realtime store entries that arrive without being flagged changed', async () => {
|
||||
// The incremental path only unwraps raw subtrees for entries it expects to
|
||||
// clone (changed ids, agents, and ids missing from the merge cache). An
|
||||
|
||||
@@ -1633,10 +1633,21 @@ export function useUnifiedResources(options?: UseUnifiedResourcesOptions) {
|
||||
const allResourcesEntry = getUnifiedResourcesCacheEntry(
|
||||
buildScopedUnifiedResourcesCacheKey(ALL_RESOURCES_CACHE_KEY, currentOrgScope),
|
||||
);
|
||||
const canApplyIncrementally =
|
||||
resourceChange.changedIds !== null &&
|
||||
allResourcesEntry.realtimeVersion > 0 &&
|
||||
resourceChange.version === allResourcesEntry.realtimeVersion + 1;
|
||||
// An instance that mounts or resumes more than one tick behind the shared
|
||||
// cache used to fall off the incremental path onto a full-estate remerge —
|
||||
// the dominant Alerts-entry / tab re-entry cost in the 2026-08-25 profile.
|
||||
// Catch up through the store's bounded changed-id history instead; a
|
||||
// single-tick gap keeps using the latest delta without touching history.
|
||||
const resolveCatchUpIds = (sinceVersion: number): ReadonlySet<string> | null => {
|
||||
if (sinceVersion <= 0 || resourceChange.version <= sinceVersion) return null;
|
||||
if (resourceChange.version === sinceVersion + 1) return resourceChange.changedIds;
|
||||
return wsStore.changedResourceIdsSince?.(sinceVersion) ?? null;
|
||||
};
|
||||
const allEntryCatchUpIds =
|
||||
allResourcesEntry.realtimeVersion > 0
|
||||
? resolveCatchUpIds(allResourcesEntry.realtimeVersion)
|
||||
: null;
|
||||
const canApplyIncrementally = allEntryCatchUpIds !== null;
|
||||
const realtimeSnapshotAlreadyApplied =
|
||||
resourceChange.version > 0 &&
|
||||
allResourcesEntry.hasSnapshot &&
|
||||
@@ -1665,9 +1676,9 @@ export function useUnifiedResources(options?: UseUnifiedResourcesOptions) {
|
||||
? allResourcesEntry.resources
|
||||
: canApplyIncrementally
|
||||
? mergeCanonicalResourceDeltaSnapshot(
|
||||
readWsResources(resourceChange.changedIds),
|
||||
readWsResources(allEntryCatchUpIds),
|
||||
allResourcesEntry.resources,
|
||||
resourceChange.changedIds!,
|
||||
allEntryCatchUpIds!,
|
||||
)
|
||||
: mergeCanonicalResourceSnapshot(readWsResources(null), allResourcesEntry.resources);
|
||||
const projectedResources = filterCanonicalUnifiedResources(
|
||||
@@ -1706,22 +1717,23 @@ export function useUnifiedResources(options?: UseUnifiedResourcesOptions) {
|
||||
allResourcesEntry.realtimeVersion = resourceChange.version;
|
||||
}
|
||||
|
||||
const canPatchProjectionIncrementally =
|
||||
resourceChange.changedIds !== null &&
|
||||
const cacheEntryCatchUpIds =
|
||||
cacheEntry.hasSnapshot &&
|
||||
cacheEntry.realtimeVersion > 0 &&
|
||||
resourceChange.version === cacheEntry.realtimeVersion + 1 &&
|
||||
resolvedProjectedResources === projectedResources;
|
||||
resolvedProjectedResources === projectedResources
|
||||
? resolveCatchUpIds(cacheEntry.realtimeVersion)
|
||||
: null;
|
||||
const canPatchProjectionIncrementally = cacheEntryCatchUpIds !== null;
|
||||
const changedResourceTouchesAgent =
|
||||
canPatchProjectionIncrementally &&
|
||||
mergedWsResources.some(
|
||||
(resource) => resource.type === 'agent' && resourceChange.changedIds!.has(resource.id),
|
||||
(resource) => resource.type === 'agent' && cacheEntryCatchUpIds!.has(resource.id),
|
||||
);
|
||||
const incrementalPatchIndices = canPatchProjectionIncrementally
|
||||
? resolveIncrementalResourcePatchIndices(
|
||||
resources as unknown as Resource[],
|
||||
resolvedProjectedResources,
|
||||
resourceChange.changedIds!,
|
||||
cacheEntryCatchUpIds!,
|
||||
changedResourceTouchesAgent,
|
||||
)
|
||||
: null;
|
||||
|
||||
@@ -368,6 +368,167 @@ describe('websocket store unified resource contract', () => {
|
||||
}
|
||||
});
|
||||
|
||||
it('expands capabilitiesRef through the catalog and synthesizes omitted default policies', async () => {
|
||||
const { store, dispose } = await createStoreHarness();
|
||||
try {
|
||||
await waitForOpenTick();
|
||||
|
||||
emitMessage({
|
||||
type: 'initialState',
|
||||
data: {
|
||||
connectedInfrastructure: [],
|
||||
capabilityCatalog: {
|
||||
cap1: [{ name: 'restart', type: 'common', description: 'Restart the guest' }],
|
||||
},
|
||||
resources: [
|
||||
{
|
||||
id: 'vm-1',
|
||||
type: 'vm',
|
||||
name: 'vm-1',
|
||||
status: 'running',
|
||||
capabilitiesRef: 'cap1',
|
||||
},
|
||||
{
|
||||
id: 'storage-1',
|
||||
type: 'storage',
|
||||
name: 'tank',
|
||||
status: 'online',
|
||||
policy: { sensitivity: 'sensitive', routing: { scope: 'local-first', redact: [] } },
|
||||
},
|
||||
],
|
||||
lastUpdate: 100,
|
||||
activeAlerts: [],
|
||||
recentlyResolved: [],
|
||||
},
|
||||
});
|
||||
|
||||
const vm = store.state.resources.find((resource) => resource.id === 'vm-1');
|
||||
expect(vm?.capabilities?.map((capability) => capability.name)).toEqual(['restart']);
|
||||
// Omitted policy means default posture; ingestion synthesizes it so every
|
||||
// policy consumer keeps seeing an inline policy.
|
||||
expect(vm?.policy).toEqual({
|
||||
sensitivity: 'internal',
|
||||
routing: { scope: 'cloud-summary', redact: [] },
|
||||
});
|
||||
const storage = store.state.resources.find((resource) => resource.id === 'storage-1');
|
||||
expect(storage?.policy?.sensitivity).toBe('sensitive');
|
||||
|
||||
// A later delta can move a resource's ref; the catalog change rides the
|
||||
// same frame and the patched row re-expands.
|
||||
emitMessage({
|
||||
type: 'rawData',
|
||||
data: {
|
||||
lastUpdate: 200,
|
||||
capabilityCatalog: {
|
||||
cap1: [{ name: 'restart', type: 'common', description: 'Restart the guest' }],
|
||||
cap2: [{ name: 'stop', type: 'common', description: 'Stop the guest' }],
|
||||
},
|
||||
resourceDelta: {
|
||||
upserts: [{ id: 'vm-1', capabilitiesRef: 'cap2' }],
|
||||
removed: [],
|
||||
order: ['vm-1', 'storage-1'],
|
||||
},
|
||||
},
|
||||
});
|
||||
const vmAfter = store.state.resources.find((resource) => resource.id === 'vm-1');
|
||||
expect(vmAfter?.capabilities?.map((capability) => capability.name)).toEqual(['stop']);
|
||||
|
||||
// A posture transition back to default nulls the policy in the patch;
|
||||
// ingestion re-synthesizes the default instead of keeping the stale one.
|
||||
emitMessage({
|
||||
type: 'rawData',
|
||||
data: {
|
||||
lastUpdate: 300,
|
||||
resourceDelta: {
|
||||
upserts: [{ id: 'storage-1', policy: null }],
|
||||
removed: [],
|
||||
order: ['vm-1', 'storage-1'],
|
||||
},
|
||||
},
|
||||
});
|
||||
const storageAfter = store.state.resources.find((resource) => resource.id === 'storage-1');
|
||||
expect(storageAfter?.policy).toEqual({
|
||||
sensitivity: 'internal',
|
||||
routing: { scope: 'cloud-summary', redact: [] },
|
||||
});
|
||||
} finally {
|
||||
dispose();
|
||||
}
|
||||
});
|
||||
|
||||
it('serves unions from the bounded changed-id history so late consumers catch up incrementally', 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' },
|
||||
{ id: 'vm-1', type: 'vm', name: 'vm-1', status: 'running' },
|
||||
{ id: 'vm-2', type: 'vm', name: 'vm-2', status: 'running' },
|
||||
],
|
||||
lastUpdate: 100,
|
||||
activeAlerts: [],
|
||||
recentlyResolved: [],
|
||||
},
|
||||
});
|
||||
// version 1: full snapshot — not coverable by the history.
|
||||
expect(store.changedResourceIdsSince(0)).toBeNull();
|
||||
|
||||
emitMessage({
|
||||
type: 'rawData',
|
||||
data: {
|
||||
lastUpdate: 200,
|
||||
resourceDelta: {
|
||||
upserts: [{ id: 'vm-1', lastSeen: 200 }],
|
||||
removed: [],
|
||||
order: ['agent-1', 'vm-1', 'vm-2'],
|
||||
},
|
||||
},
|
||||
});
|
||||
emitMessage({
|
||||
type: 'rawData',
|
||||
data: {
|
||||
lastUpdate: 300,
|
||||
resourceDelta: {
|
||||
upserts: [{ id: 'vm-2', lastSeen: 300 }],
|
||||
removed: [],
|
||||
order: ['agent-1', 'vm-1', 'vm-2'],
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
// versions 2 and 3 are delta commits: a consumer still on version 1 gets
|
||||
// the union, one on version 2 gets just the last delta, and a current
|
||||
// consumer gets null (nothing to catch up).
|
||||
expect(store.changedResourceIdsSince(1)).toEqual(new Set(['vm-1', 'vm-2']));
|
||||
expect(store.changedResourceIdsSince(2)).toEqual(new Set(['vm-2']));
|
||||
expect(store.changedResourceIdsSince(3)).toBeNull();
|
||||
// version 1 was a full snapshot, so a span reaching before it is not
|
||||
// coverable and must force the caller onto the full-merge path.
|
||||
expect(store.changedResourceIdsSince(0)).toBeNull();
|
||||
|
||||
emitMessage({
|
||||
type: 'initialState',
|
||||
data: {
|
||||
connectedInfrastructure: [],
|
||||
resources: [{ id: 'agent-1', type: 'agent', name: 'agent-1', status: 'online' }],
|
||||
lastUpdate: 400,
|
||||
activeAlerts: [],
|
||||
recentlyResolved: [],
|
||||
},
|
||||
});
|
||||
// A later full snapshot invalidates the accumulated history.
|
||||
expect(store.changedResourceIdsSince(3)).toBeNull();
|
||||
expect(store.changedResourceIdsSince(2)).toBeNull();
|
||||
} finally {
|
||||
dispose();
|
||||
}
|
||||
});
|
||||
|
||||
it('keeps canonically merged hosts consistent with full snapshots across deltas (#1601)', async () => {
|
||||
const { store, dispose } = await createStoreHarness();
|
||||
try {
|
||||
|
||||
@@ -63,6 +63,7 @@ const createNoopWebSocketStore = (): ReturnType<typeof createWebSocketStore> =>
|
||||
initialDataReceived,
|
||||
updateProgress,
|
||||
resourceChange,
|
||||
changedResourceIdsSince: () => null,
|
||||
shutdown: () => {},
|
||||
reconnect: () => {},
|
||||
switchUrl: () => {},
|
||||
|
||||
@@ -7,7 +7,8 @@ import type {
|
||||
ResolvedAlert,
|
||||
ConnectedInfrastructureItem,
|
||||
} from '@/types/api';
|
||||
import type { Resource } from '@/types/resource';
|
||||
import type { Resource, ResourceCapability } from '@/types/resource';
|
||||
import { createDefaultResourcePolicy } from '@/types/resource';
|
||||
import { logger } from '@/utils/logger';
|
||||
import { POLLING_INTERVALS, WEBSOCKET } from '@/constants';
|
||||
import { notificationStore } from './notifications';
|
||||
@@ -252,6 +253,27 @@ export function createWebSocketStore(url: string) {
|
||||
version: number;
|
||||
changedIds: ReadonlySet<string> | null;
|
||||
}>({ version: 0, changedIds: null });
|
||||
// Bounded per-tick changed-id history so a consumer that mounted or resumed
|
||||
// a few ticks behind the live version can still catch up with a delta merge
|
||||
// instead of a full-estate remerge. A full-snapshot commit (changedIds null)
|
||||
// invalidates the whole span, so the history resets there.
|
||||
const RESOURCE_CHANGE_HISTORY_LIMIT = 30;
|
||||
let resourceChangeHistory: { version: number; changedIds: ReadonlySet<string> }[] = [];
|
||||
|
||||
const changedResourceIdsSince = (sinceVersion: number): ReadonlySet<string> | null => {
|
||||
if (sinceVersion >= resourceChangeVersion) return null;
|
||||
if (resourceChangeHistory.length === 0) return null;
|
||||
if (resourceChangeHistory[0].version > sinceVersion + 1) return null;
|
||||
if (resourceChangeHistory[resourceChangeHistory.length - 1].version !== resourceChangeVersion) {
|
||||
return null;
|
||||
}
|
||||
const union = new Set<string>();
|
||||
for (const entry of resourceChangeHistory) {
|
||||
if (entry.version <= sinceVersion) continue;
|
||||
entry.changedIds.forEach((id) => union.add(id));
|
||||
}
|
||||
return union;
|
||||
};
|
||||
|
||||
// Track alerts with pending acknowledgment changes to prevent race conditions
|
||||
const pendingAckChanges = new Map<string, { ack: boolean; previousAckTime?: string }>();
|
||||
@@ -340,6 +362,34 @@ export function createWebSocketStore(url: string) {
|
||||
// and Solid's reconcile mutates adopted objects in place.
|
||||
let rawServerResources: Resource[] | null = null;
|
||||
const deferredResourceIds = new Set<string>();
|
||||
// Latest capabilityCatalog from the state payload. Broadcast resources carry
|
||||
// capabilitiesRef instead of inline capability blobs; ingestion expands the
|
||||
// ref back into `capabilities` so consumers keep the inline shape.
|
||||
let capabilityCatalog: Record<string, ResourceCapability[]> = {};
|
||||
// Records which catalog ref an expanded capabilities array was cloned from,
|
||||
// so a patched row re-expands only when its ref actually changed.
|
||||
const capabilitiesExpandedFromRef = new WeakMap<object, string>();
|
||||
|
||||
const hydrateSlimResource = (resource: Resource): void => {
|
||||
if (!resource || typeof resource !== 'object') return;
|
||||
const ref = resource.capabilitiesRef;
|
||||
if (ref) {
|
||||
const current = resource.capabilities as unknown as object | undefined;
|
||||
if (!current || capabilitiesExpandedFromRef.get(current) !== ref) {
|
||||
const entry = capabilityCatalog[ref];
|
||||
if (entry) {
|
||||
// Per-row clone: reconcile mutates adopted objects in place, so rows
|
||||
// must not share one materialized capabilities array.
|
||||
const expanded = structuredClone(entry);
|
||||
capabilitiesExpandedFromRef.set(expanded as unknown as object, ref);
|
||||
resource.capabilities = expanded;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (!resource.policy) {
|
||||
resource.policy = createDefaultResourcePolicy();
|
||||
}
|
||||
};
|
||||
let lastFullStateRecoveryAt = 0;
|
||||
// Set once the server has sent a full snapshot too large for the inbound
|
||||
// guard. Asking for another one over the socket would only reproduce the same
|
||||
@@ -456,10 +506,25 @@ export function createWebSocketStore(url: string) {
|
||||
changedCount: changedResourceIds?.size ?? nextResources.length,
|
||||
});
|
||||
setState('resources', reconcile(nextResources, { key: 'id' }));
|
||||
const committedChangedIds = changedResourceIds ? new Set(changedResourceIds) : null;
|
||||
setResourceChange({
|
||||
version: ++resourceChangeVersion,
|
||||
changedIds: changedResourceIds ? new Set(changedResourceIds) : null,
|
||||
changedIds: committedChangedIds,
|
||||
});
|
||||
if (committedChangedIds) {
|
||||
resourceChangeHistory.push({
|
||||
version: resourceChangeVersion,
|
||||
changedIds: committedChangedIds,
|
||||
});
|
||||
if (resourceChangeHistory.length > RESOURCE_CHANGE_HISTORY_LIMIT) {
|
||||
resourceChangeHistory.splice(
|
||||
0,
|
||||
resourceChangeHistory.length - RESOURCE_CHANGE_HISTORY_LIMIT,
|
||||
);
|
||||
}
|
||||
} else {
|
||||
resourceChangeHistory = [];
|
||||
}
|
||||
syncContainerCommands(nextResources, changedResourceIds);
|
||||
};
|
||||
|
||||
@@ -615,6 +680,13 @@ export function createWebSocketStore(url: string) {
|
||||
setInitialDataReceived(true);
|
||||
}
|
||||
|
||||
// Adopt the capability catalog before any resource handling below:
|
||||
// the delta engine ships catalog changes in the same frame as the
|
||||
// first resource that references a new entry.
|
||||
if (message.data.capabilityCatalog !== undefined) {
|
||||
capabilityCatalog = message.data.capabilityCatalog ?? {};
|
||||
}
|
||||
|
||||
// Canonical resource contract:
|
||||
// `state.resources` is the authoritative frontend model.
|
||||
// `state.connectedInfrastructure` is the authoritative reporting projection.
|
||||
@@ -645,6 +717,10 @@ export function createWebSocketStore(url: string) {
|
||||
if (message.data.resources !== undefined) {
|
||||
deferredResourceIds.clear();
|
||||
if (Array.isArray(message.data.resources)) {
|
||||
// Expand capabilitiesRef and synthesize omitted default
|
||||
// policies before the rows feed the baseline clone and the
|
||||
// canonical merge.
|
||||
message.data.resources.forEach(hydrateSlimResource);
|
||||
// Only a full payload actually delivered on this socket can
|
||||
// establish its server-owned delta baseline. REST recovery is
|
||||
// an independently built display snapshot and stays baseline-
|
||||
@@ -678,6 +754,10 @@ export function createWebSocketStore(url: string) {
|
||||
rawServerResources,
|
||||
message.data.resourceDelta,
|
||||
);
|
||||
// Patched rows are fresh objects: re-synthesize a nulled-out
|
||||
// default policy and re-expand capabilities when a patch moved
|
||||
// the ref. Unpatched rows no-op.
|
||||
appliedDelta.resources.forEach(hydrateSlimResource);
|
||||
rawServerResources = appliedDelta.resources;
|
||||
changedResourceIds = appliedDelta.changedIds;
|
||||
if (typeof document !== 'undefined' && document.visibilityState === 'hidden') {
|
||||
@@ -1167,6 +1247,7 @@ export function createWebSocketStore(url: string) {
|
||||
initialDataReceived,
|
||||
updateProgress,
|
||||
resourceChange,
|
||||
changedResourceIdsSince,
|
||||
shutdown,
|
||||
reconnect: () => {
|
||||
if (isDisposed) return;
|
||||
@@ -1193,6 +1274,7 @@ export function createWebSocketStore(url: string) {
|
||||
setInitialDataReceived(false);
|
||||
setUpdateProgress(null);
|
||||
setResourceChange({ version: ++resourceChangeVersion, changedIds: null });
|
||||
resourceChangeHistory = [];
|
||||
setState(reconcile(createInitialState()));
|
||||
setActiveAlerts(reconcile({}));
|
||||
setRecentlyResolved(reconcile({}));
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
// Properly typed TypeScript interfaces for Pulse API
|
||||
|
||||
import type { Resource } from './resource';
|
||||
import type { Resource, ResourceCapability } from './resource';
|
||||
import type { EvidenceEnvelope, LifecycleTransition, OperationalRecord } from './operationalTrust';
|
||||
|
||||
export interface APITokenRecord {
|
||||
@@ -31,6 +31,9 @@ export interface State {
|
||||
pveTagStyles?: Record<string, PVETagStyle>;
|
||||
// Unified resources (canonical resource model)
|
||||
resources: Resource[];
|
||||
// Distinct capability blobs referenced by Resource.capabilitiesRef; the
|
||||
// websocket store expands refs into inline capabilities at ingestion.
|
||||
capabilityCatalog?: Record<string, ResourceCapability[]> | null;
|
||||
}
|
||||
|
||||
export interface ResourceStateDelta {
|
||||
|
||||
@@ -221,6 +221,17 @@ export const requiresGovernedResourceDisplay = (policy?: ResourcePolicy | null):
|
||||
if (!policy) return false;
|
||||
return policy.routing.scope === 'local-only' || (policy.routing.redact?.length ?? 0) > 0;
|
||||
};
|
||||
|
||||
// The posture the server omits from broadcast payloads: default-posture
|
||||
// resources carry no inline policy on the wire (see the backend's
|
||||
// isDefaultBroadcastResourcePolicy), and the websocket store synthesizes it at
|
||||
// ingestion so every policy consumer keeps seeing an inline policy. Fresh
|
||||
// objects per resource — a shared instance would be corrupted in place by
|
||||
// store reconciliation when one resource's posture later changes.
|
||||
export const createDefaultResourcePolicy = (): ResourcePolicy => ({
|
||||
sensitivity: 'internal',
|
||||
routing: { scope: 'cloud-summary', redact: [] },
|
||||
});
|
||||
export type ResourceChangeConfidence = 'high' | 'medium' | 'low';
|
||||
export type ResourceChangeKind =
|
||||
| 'state_transition'
|
||||
@@ -1587,6 +1598,10 @@ export interface Resource {
|
||||
policy?: ResourcePolicy;
|
||||
aiSafeSummary?: string;
|
||||
capabilities?: ResourceCapability[];
|
||||
// Reference into the state payload's capabilityCatalog; the websocket store
|
||||
// expands it into `capabilities` at ingestion so consumers keep reading the
|
||||
// inline shape.
|
||||
capabilitiesRef?: string;
|
||||
actionReadiness?: ResourceActionReadiness[];
|
||||
sourceStatus?: Record<string, ResourceSourceStatus>;
|
||||
relationships?: ResourceRelationship[];
|
||||
|
||||
@@ -89,6 +89,24 @@ describe('resourceIdentity', () => {
|
||||
]);
|
||||
});
|
||||
|
||||
it('resolves superseded canonical ids even though broadcast aliases no longer carry them', () => {
|
||||
// Slimmed broadcast payloads stop duplicating supersededIds into the
|
||||
// alias list, so the identity resolver has to consult them explicitly for
|
||||
// alert overrides and lookups keyed by a retired spelling.
|
||||
const aliases = getResourceIdentityAliases(
|
||||
makeResource({
|
||||
canonicalIdentity: {
|
||||
primaryId: 'agent:core2026',
|
||||
aliases: ['agent:core2026', 'core2026'],
|
||||
supersededIds: ['agent-core2026-a1b2c3', 'agent-core2026-d4e5f6'],
|
||||
},
|
||||
}),
|
||||
);
|
||||
expect(aliases).toContain('agent-core2026-a1b2c3');
|
||||
expect(aliases).toContain('agent-core2026-d4e5f6');
|
||||
expect(aliases).toContain('agent:core2026');
|
||||
});
|
||||
|
||||
it('uses actionable linked identities before falling back to unified ids', () => {
|
||||
expect(
|
||||
getPrimaryResourceIdentity(
|
||||
|
||||
@@ -138,9 +138,17 @@ export const getResourceIdentityAliases = (resource: Resource): string[] => {
|
||||
.map((alias) => asTrimmedString(alias))
|
||||
.filter((alias): alias is string => Boolean(alias))
|
||||
: [];
|
||||
// Broadcast payloads no longer duplicate superseded canonical ids into the
|
||||
// alias list, so identity resolution has to consult them explicitly.
|
||||
const canonicalSupersededIds = Array.isArray(canonical?.supersededIds)
|
||||
? canonical.supersededIds
|
||||
.map((id) => asTrimmedString(id))
|
||||
.filter((id): id is string => Boolean(id))
|
||||
: [];
|
||||
|
||||
const raw = [
|
||||
...canonicalAliases,
|
||||
...canonicalSupersededIds,
|
||||
resource.metricsTarget?.resourceId,
|
||||
resource.discoveryTarget?.agentId,
|
||||
resource.discoveryTarget?.resourceId,
|
||||
|
||||
@@ -1189,6 +1189,7 @@ type ResourceConvertInput struct {
|
||||
Policy json.RawMessage
|
||||
AISafeSummary string
|
||||
Capabilities json.RawMessage
|
||||
CapabilitiesRef string
|
||||
Relationships json.RawMessage
|
||||
RecentChanges json.RawMessage
|
||||
FacetCounts json.RawMessage
|
||||
@@ -1272,6 +1273,7 @@ func ConvertResourceToFrontend(input ResourceConvertInput) ResourceFrontend {
|
||||
Policy: input.Policy,
|
||||
AISafeSummary: input.AISafeSummary,
|
||||
Capabilities: input.Capabilities,
|
||||
CapabilitiesRef: input.CapabilitiesRef,
|
||||
Relationships: input.Relationships,
|
||||
RecentChanges: input.RecentChanges,
|
||||
FacetCounts: input.FacetCounts,
|
||||
|
||||
@@ -964,6 +964,13 @@ type StateFrontend struct {
|
||||
// Unified resources - the new way to access all monitored entities
|
||||
Resources []ResourceFrontend `json:"resources"`
|
||||
ConnectedInfrastructure []ConnectedInfrastructureItemFrontend `json:"connectedInfrastructure"`
|
||||
// CapabilityCatalog maps content-hash ids to capability blobs referenced by
|
||||
// ResourceFrontend.CapabilitiesRef. Estates carry only a handful of
|
||||
// distinct capability sets, so the catalog replaces per-resource inline
|
||||
// duplication in broadcast payloads. The websocket delta engine diffs
|
||||
// top-level state fields generically, so catalog changes ride the same
|
||||
// frame as the resource refs that need them.
|
||||
CapabilityCatalog map[string]json.RawMessage `json:"capabilityCatalog,omitempty"`
|
||||
}
|
||||
|
||||
// EmptyStateFrontend returns a canonical empty frontend state with stable
|
||||
@@ -1064,12 +1071,18 @@ type ResourceFrontend struct {
|
||||
// Identity for deduplication
|
||||
Identity *ResourceIdentityFrontend `json:"identity,omitempty"`
|
||||
|
||||
DiscoveryTarget json.RawMessage `json:"discoveryTarget,omitempty"`
|
||||
MetricsTarget json.RawMessage `json:"metricsTarget,omitempty"`
|
||||
Canonical json.RawMessage `json:"canonicalIdentity,omitempty"`
|
||||
Policy json.RawMessage `json:"policy,omitempty"`
|
||||
AISafeSummary string `json:"aiSafeSummary,omitempty"`
|
||||
Capabilities json.RawMessage `json:"capabilities,omitempty"`
|
||||
DiscoveryTarget json.RawMessage `json:"discoveryTarget,omitempty"`
|
||||
MetricsTarget json.RawMessage `json:"metricsTarget,omitempty"`
|
||||
Canonical json.RawMessage `json:"canonicalIdentity,omitempty"`
|
||||
Policy json.RawMessage `json:"policy,omitempty"`
|
||||
AISafeSummary string `json:"aiSafeSummary,omitempty"`
|
||||
Capabilities json.RawMessage `json:"capabilities,omitempty"`
|
||||
// CapabilitiesRef references an entry in StateFrontend.CapabilityCatalog.
|
||||
// Broadcast payloads dedupe the small set of distinct capability blobs
|
||||
// through the catalog instead of inlining ~3KB per resource; consumers
|
||||
// resolve the ref at ingestion. Inline Capabilities remains valid for
|
||||
// producers that do not carry a catalog.
|
||||
CapabilitiesRef string `json:"capabilitiesRef,omitempty"`
|
||||
Relationships json.RawMessage `json:"relationships,omitempty"`
|
||||
RecentChanges json.RawMessage `json:"recentChanges,omitempty"`
|
||||
FacetCounts json.RawMessage `json:"facetCounts,omitempty"`
|
||||
|
||||
@@ -604,7 +604,7 @@ func TestBroadcastResourceProjectionCoalescesSplitHostIdentities(t *testing.T) {
|
||||
for _, snippet := range []string{
|
||||
"metricsTargetResolver := broadcastMetricsTargetResolver(unifiedView.readState)",
|
||||
"broadcastResources := unifiedresources.CoalescePresentationHostResources(unifiedView.resources)",
|
||||
"frontendState.Resources = convertResourcesForBroadcast(broadcastResources, metricsTargetResolver)",
|
||||
"frontendState.Resources, frontendState.CapabilityCatalog = convertResourcesForBroadcast(broadcastResources, metricsTargetResolver)",
|
||||
"frontendState.ConnectedInfrastructure = buildConnectedInfrastructure(broadcastResources, snapshot)",
|
||||
"func attachBroadcastMetricsTargets(",
|
||||
"allResources = unifiedresources.CoalescePresentationHostResources(allResources)",
|
||||
|
||||
@@ -87,7 +87,7 @@ func TestIssue1597UnifiedGuestCPUFeedsDashboardDetailsAndAlerts(t *testing.T) {
|
||||
})
|
||||
|
||||
adapter := unifiedresources.NewMonitorAdapter(registry)
|
||||
frontend := convertResourcesForBroadcast(adapter.GetAll(), adapter)
|
||||
frontend, _ := convertResourcesForBroadcast(adapter.GetAll(), adapter)
|
||||
if len(frontend) != 1 {
|
||||
t.Fatalf("dashboard resource count = %d, want one unified guest", len(frontend))
|
||||
}
|
||||
|
||||
@@ -2,6 +2,8 @@ package monitoring
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
stderrors "errors"
|
||||
"fmt"
|
||||
@@ -4165,7 +4167,7 @@ func (m *Monitor) buildBroadcastFrontendStateFromSnapshot(snapshot models.StateS
|
||||
metricsTargetResolver := broadcastMetricsTargetResolver(unifiedView.readState)
|
||||
broadcastResources := unifiedresources.CoalescePresentationHostResources(unifiedView.resources)
|
||||
broadcastResources = m.applyPersistedMetadataToUnifiedResources(broadcastResources)
|
||||
frontendState.Resources = convertResourcesForBroadcast(broadcastResources, metricsTargetResolver)
|
||||
frontendState.Resources, frontendState.CapabilityCatalog = convertResourcesForBroadcast(broadcastResources, metricsTargetResolver)
|
||||
frontendState.ConnectedInfrastructure = buildConnectedInfrastructure(broadcastResources, snapshot)
|
||||
if !unifiedView.freshness.IsZero() {
|
||||
frontendState.LastUpdate = unifiedView.freshness.UnixMilli()
|
||||
@@ -5603,10 +5605,11 @@ func (m *Monitor) getResourcesForBroadcast() []models.ResourceFrontend {
|
||||
m.mu.RLock()
|
||||
store := m.resourceStore
|
||||
m.mu.RUnlock()
|
||||
return convertResourcesForBroadcast(
|
||||
resources, _ := convertResourcesForBroadcast(
|
||||
m.applyPersistedMetadataToUnifiedResources(m.getUnifiedResourcesForBroadcast()),
|
||||
broadcastMetricsTargetResolver(store),
|
||||
)
|
||||
return resources
|
||||
}
|
||||
|
||||
func (m *Monitor) applyPersistedMetadataToUnifiedResources(resources []unifiedresources.Resource) []unifiedresources.Resource {
|
||||
@@ -5808,13 +5811,15 @@ func (m *Monitor) dockerAppContainerCustomURL(
|
||||
return "", false
|
||||
}
|
||||
|
||||
// convertResourcesForBroadcast converts unified resources into the frontend payload shape.
|
||||
// convertResourcesForBroadcast converts unified resources into the frontend
|
||||
// payload shape. Distinct capability blobs are deduped into the returned
|
||||
// catalog and referenced per resource via capabilitiesRef.
|
||||
func convertResourcesForBroadcast(
|
||||
allResources []unifiedresources.Resource,
|
||||
metricsTargetResolvers ...MetricsTargetResourceStore,
|
||||
) []models.ResourceFrontend {
|
||||
) ([]models.ResourceFrontend, map[string]json.RawMessage) {
|
||||
if len(allResources) == 0 {
|
||||
return []models.ResourceFrontend{}
|
||||
return []models.ResourceFrontend{}, nil
|
||||
}
|
||||
allResources = attachBroadcastMetricsTargets(
|
||||
allResources,
|
||||
@@ -5828,8 +5833,15 @@ func convertResourcesForBroadcast(
|
||||
}
|
||||
|
||||
converted := make([]broadcastResource, 0, len(allResources))
|
||||
capabilityCatalog := make(map[string]json.RawMessage)
|
||||
for _, r := range allResources {
|
||||
input := monitorResourceToConvertInput(r)
|
||||
if len(input.Capabilities) > 0 {
|
||||
id := capabilityCatalogID(input.Capabilities)
|
||||
capabilityCatalog[id] = input.Capabilities
|
||||
input.CapabilitiesRef = id
|
||||
input.Capabilities = nil
|
||||
}
|
||||
sortKey := strings.ToLower(input.DisplayName)
|
||||
if sortKey == "" {
|
||||
sortKey = strings.ToLower(input.Name)
|
||||
@@ -5852,7 +5864,10 @@ func convertResourcesForBroadcast(
|
||||
for i, resource := range converted {
|
||||
result[i] = models.ConvertResourceToFrontend(resource.input)
|
||||
}
|
||||
return result
|
||||
if len(capabilityCatalog) == 0 {
|
||||
capabilityCatalog = nil
|
||||
}
|
||||
return result, capabilityCatalog
|
||||
}
|
||||
|
||||
func broadcastMetricsTargetResolver(source interface{}) MetricsTargetResourceStore {
|
||||
@@ -5920,6 +5935,7 @@ func monitorResourceToConvertInput(resource unifiedresources.Resource) models.Re
|
||||
resource.MetricsTarget = monitorMetricsTarget(resource, resourceType)
|
||||
}
|
||||
unifiedresources.RefreshCanonicalMetadata(&resource)
|
||||
slimResourceForBroadcast(&resource)
|
||||
name, displayName := monitorFrontendNames(resource, resourceType)
|
||||
platformID := monitorPlatformID(resource, resourceType)
|
||||
|
||||
@@ -6007,6 +6023,59 @@ func monitorRawJSON(value interface{}) json.RawMessage {
|
||||
return encoded
|
||||
}
|
||||
|
||||
// slimResourceForBroadcast trims static metadata the client-facing stream was
|
||||
// re-shipping on every resource (governed gap resource-payload-static-metadata).
|
||||
// It only edits the freshly refreshed per-broadcast copy, never stored state:
|
||||
// - Aliases duplicating superseded canonical ids are dropped; the ids remain
|
||||
// available in canonicalIdentity.supersededIds, which identity consumers
|
||||
// also resolve against.
|
||||
// - Default-posture policy and its AI-safe summary are omitted; consumers
|
||||
// treat a missing policy as the default posture, and the summary is only
|
||||
// rendered for non-default postures.
|
||||
func slimResourceForBroadcast(resource *unifiedresources.Resource) {
|
||||
if resource == nil {
|
||||
return
|
||||
}
|
||||
if canonical := resource.Canonical; canonical != nil && len(canonical.SupersededIDs) > 0 &&
|
||||
len(canonical.Aliases) > 0 {
|
||||
superseded := make(map[string]struct{}, len(canonical.SupersededIDs))
|
||||
for _, id := range canonical.SupersededIDs {
|
||||
superseded[strings.ToLower(strings.TrimSpace(id))] = struct{}{}
|
||||
}
|
||||
aliases := make([]string, 0, len(canonical.Aliases))
|
||||
for _, alias := range canonical.Aliases {
|
||||
if _, dup := superseded[strings.ToLower(strings.TrimSpace(alias))]; dup {
|
||||
continue
|
||||
}
|
||||
aliases = append(aliases, alias)
|
||||
}
|
||||
canonical.Aliases = aliases
|
||||
}
|
||||
if isDefaultBroadcastResourcePolicy(resource.Policy) {
|
||||
resource.Policy = nil
|
||||
resource.AISafeSummary = ""
|
||||
}
|
||||
}
|
||||
|
||||
// isDefaultBroadcastResourcePolicy mirrors the frontend's
|
||||
// hasDefaultResourcePolicyPosture: internal sensitivity, cloud-summary
|
||||
// routing, and no redactions is the posture consumers assume when the
|
||||
// broadcast omits the policy entirely.
|
||||
func isDefaultBroadcastResourcePolicy(policy *unifiedresources.ResourcePolicy) bool {
|
||||
return policy != nil &&
|
||||
policy.Sensitivity == unifiedresources.ResourceSensitivityInternal &&
|
||||
policy.Routing.Scope == unifiedresources.ResourceRoutingScopeCloudSummary &&
|
||||
len(policy.Routing.Redact) == 0
|
||||
}
|
||||
|
||||
// capabilityCatalogID derives the stable catalog id for one marshaled
|
||||
// capabilities blob. Content-addressed ids keep the catalog and refs stable
|
||||
// across restarts and reconnects, so they never churn websocket deltas.
|
||||
func capabilityCatalogID(encoded json.RawMessage) string {
|
||||
digest := sha256.Sum256(encoded)
|
||||
return hex.EncodeToString(digest[:6])
|
||||
}
|
||||
|
||||
func monitorDiscoveryTarget(resource unifiedresources.Resource, resourceType string) *unifiedresources.DiscoveryTarget {
|
||||
switch resourceType {
|
||||
case "agent", "docker-host":
|
||||
|
||||
@@ -104,6 +104,51 @@ func TestMonitoringBroadcastCarriesEveryAvailabilityProjection(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestBroadcastSlimmingKeepsHostAgentSupersededIdentityResolvable(t *testing.T) {
|
||||
// A re-enrolled host carries retired canonical spellings. Broadcast
|
||||
// slimming stops duplicating them into the alias list, but they must stay
|
||||
// resolvable through supersededIds so alert overrides and identity
|
||||
// consumers keep following the host across its retired ids.
|
||||
input := monitorResourceToConvertInput(unifiedresources.Resource{
|
||||
ID: "agent-core2026",
|
||||
Type: unifiedresources.ResourceTypeAgent,
|
||||
Name: "core2026",
|
||||
Status: unifiedresources.StatusOnline,
|
||||
Sources: []unifiedresources.DataSource{unifiedresources.SourceAgent},
|
||||
SupersededCanonicalIDs: []string{"agent-core2026-a1b2c3", "agent-core2026-d4e5f6"},
|
||||
Agent: &unifiedresources.AgentData{
|
||||
AgentID: "core2026",
|
||||
Hostname: "core2026",
|
||||
},
|
||||
})
|
||||
|
||||
var canonical struct {
|
||||
Aliases []string `json:"aliases"`
|
||||
SupersededIDs []string `json:"supersededIds"`
|
||||
}
|
||||
if err := json.Unmarshal(input.Canonical, &canonical); err != nil {
|
||||
t.Fatalf("unmarshal canonicalIdentity: %v (payload %s)", err, input.Canonical)
|
||||
}
|
||||
if len(canonical.SupersededIDs) != 2 {
|
||||
t.Fatalf("supersededIds = %v, want both retired spellings", canonical.SupersededIDs)
|
||||
}
|
||||
for _, alias := range canonical.Aliases {
|
||||
if alias == "agent-core2026-a1b2c3" || alias == "agent-core2026-d4e5f6" {
|
||||
t.Fatalf("aliases must not duplicate superseded ids, got %v", canonical.Aliases)
|
||||
}
|
||||
}
|
||||
// The live alias vocabulary the host is reachable by must survive the trim.
|
||||
aliasSet := make(map[string]struct{}, len(canonical.Aliases))
|
||||
for _, alias := range canonical.Aliases {
|
||||
aliasSet[alias] = struct{}{}
|
||||
}
|
||||
for _, required := range []string{"agent:core2026", "core2026"} {
|
||||
if _, ok := aliasSet[required]; !ok {
|
||||
t.Fatalf("aliases = %v, want %q retained", canonical.Aliases, required)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestHostAgentCorrelationKeepsSameNamedProxmoxProvidersDistinct(t *testing.T) {
|
||||
now := time.Now()
|
||||
state := &models.State{
|
||||
|
||||
@@ -134,7 +134,7 @@ func TestConvertResourcesForBroadcastCoalescesSplitHostResources(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
frontend := convertResourcesForBroadcast(resources)
|
||||
frontend, _ := convertResourcesForBroadcast(resources)
|
||||
if len(frontend) != 1 {
|
||||
t.Fatalf("expected split host resources to coalesce into 1 frontend resource, got %d: %#v", len(frontend), frontend)
|
||||
}
|
||||
@@ -218,7 +218,7 @@ func TestConvertResourcesForBroadcastAttachesResolvedStorageMetricsTarget(t *tes
|
||||
})
|
||||
adapter := unifiedresources.NewMonitorAdapter(registry)
|
||||
|
||||
frontend := convertResourcesForBroadcast(adapter.GetAll(), adapter)
|
||||
frontend, _ := convertResourcesForBroadcast(adapter.GetAll(), adapter)
|
||||
|
||||
var storagePayload struct {
|
||||
ID string `json:"id"`
|
||||
@@ -338,7 +338,7 @@ func TestConvertResourcesForBroadcastKeepsLinkedGuestCPUAndHistoryTargetAligned(
|
||||
})
|
||||
|
||||
adapter := unifiedresources.NewMonitorAdapter(registry)
|
||||
frontend := convertResourcesForBroadcast(adapter.GetAll(), adapter)
|
||||
frontend, _ := convertResourcesForBroadcast(adapter.GetAll(), adapter)
|
||||
if len(frontend) != 1 {
|
||||
t.Fatalf("broadcast resources = %d, want one merged guest: %+v", len(frontend), frontend)
|
||||
}
|
||||
@@ -686,7 +686,8 @@ func TestMonitorBuildBroadcastFrontendStateUsesCanonicalMockUnifiedResources(t *
|
||||
// pin an exact merge count. Require the broadcast count to fall
|
||||
// within a 5% tolerance of the convert-pipeline count so future
|
||||
// fixture bumps stay green as long as the merge contract is honored.
|
||||
expectedBroadcastCount := len(convertResourcesForBroadcast(expectedResources))
|
||||
expectedBroadcastResources, _ := convertResourcesForBroadcast(expectedResources)
|
||||
expectedBroadcastCount := len(expectedBroadcastResources)
|
||||
tolerance := expectedBroadcastCount / 20
|
||||
if tolerance < 5 {
|
||||
tolerance = 5
|
||||
@@ -783,7 +784,7 @@ func TestConvertResourcesForBroadcastIncludesCanonicalHealthContext(t *testing.T
|
||||
},
|
||||
}
|
||||
|
||||
frontend := convertResourcesForBroadcast(resources)
|
||||
frontend, _ := convertResourcesForBroadcast(resources)
|
||||
if len(frontend) != 2 {
|
||||
t.Fatalf("expected 2 frontend resources, got %d", len(frontend))
|
||||
}
|
||||
@@ -1000,3 +1001,117 @@ func TestMonitorMonitoredSystemUsageDoesNotDoubleCountLinkedProxmoxHostContinuit
|
||||
t.Fatalf("MonitoredSystemUsage().Count = %d, want 1", usage.Count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBroadcastPayloadSlimsStaticMetadata(t *testing.T) {
|
||||
sharedCapabilities := []unifiedresources.ResourceCapability{{
|
||||
Name: "restart",
|
||||
Type: unifiedresources.CapabilityTypeCommon,
|
||||
Description: "Restart the guest",
|
||||
MinimumApprovalLevel: unifiedresources.ApprovalAdmin,
|
||||
}}
|
||||
resources := []unifiedresources.Resource{
|
||||
{
|
||||
ID: "vm-100",
|
||||
Type: unifiedresources.ResourceTypeVM,
|
||||
Name: "web-01",
|
||||
Status: unifiedresources.StatusOnline,
|
||||
LastSeen: time.Date(2026, 8, 25, 10, 0, 0, 0, time.UTC),
|
||||
Sources: []unifiedresources.DataSource{unifiedresources.SourceProxmox},
|
||||
SupersededCanonicalIDs: []string{"vm-superseded-hash-1", "vm-superseded-hash-2"},
|
||||
Capabilities: sharedCapabilities,
|
||||
},
|
||||
{
|
||||
ID: "vm-101",
|
||||
Type: unifiedresources.ResourceTypeVM,
|
||||
Name: "web-02",
|
||||
Status: unifiedresources.StatusOnline,
|
||||
LastSeen: time.Date(2026, 8, 25, 10, 0, 0, 0, time.UTC),
|
||||
Sources: []unifiedresources.DataSource{unifiedresources.SourceProxmox},
|
||||
Capabilities: sharedCapabilities,
|
||||
},
|
||||
{
|
||||
ID: "storage-1",
|
||||
Type: unifiedresources.ResourceTypeStorage,
|
||||
Name: "tank",
|
||||
Status: unifiedresources.StatusOnline,
|
||||
LastSeen: time.Date(2026, 8, 25, 10, 0, 0, 0, time.UTC),
|
||||
Sources: []unifiedresources.DataSource{unifiedresources.SourceProxmox},
|
||||
},
|
||||
}
|
||||
|
||||
frontend, catalog := convertResourcesForBroadcast(resources)
|
||||
if len(frontend) != 3 {
|
||||
t.Fatalf("expected 3 frontend resources, got %d", len(frontend))
|
||||
}
|
||||
byID := make(map[string]map[string]json.RawMessage, len(frontend))
|
||||
refs := make(map[string]string, len(frontend))
|
||||
for _, resource := range frontend {
|
||||
encoded, err := json.Marshal(resource)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal %s: %v", resource.ID, err)
|
||||
}
|
||||
fields := map[string]json.RawMessage{}
|
||||
if err := json.Unmarshal(encoded, &fields); err != nil {
|
||||
t.Fatalf("unmarshal %s: %v", resource.ID, err)
|
||||
}
|
||||
byID[resource.ID] = fields
|
||||
refs[resource.ID] = resource.CapabilitiesRef
|
||||
}
|
||||
|
||||
// Default-posture workloads must not re-ship policy or AI-safe prose.
|
||||
vm := byID["vm-100"]
|
||||
if _, ok := vm["policy"]; ok {
|
||||
t.Fatalf("default-posture vm must omit policy, got %s", vm["policy"])
|
||||
}
|
||||
if _, ok := vm["aiSafeSummary"]; ok {
|
||||
t.Fatalf("default-posture vm must omit aiSafeSummary, got %s", vm["aiSafeSummary"])
|
||||
}
|
||||
// Sensitive storage keeps its governed posture inline.
|
||||
storage := byID["storage-1"]
|
||||
var storagePolicy struct {
|
||||
Sensitivity string `json:"sensitivity"`
|
||||
}
|
||||
if err := json.Unmarshal(storage["policy"], &storagePolicy); err != nil {
|
||||
t.Fatalf("sensitive storage must keep policy inline: %v (payload %s)", err, storage["policy"])
|
||||
}
|
||||
if storagePolicy.Sensitivity != "sensitive" {
|
||||
t.Fatalf("storage sensitivity = %q", storagePolicy.Sensitivity)
|
||||
}
|
||||
|
||||
// Superseded ids stay resolvable via supersededIds but are no longer
|
||||
// duplicated into the alias list.
|
||||
var canonical struct {
|
||||
Aliases []string `json:"aliases"`
|
||||
SupersededIDs []string `json:"supersededIds"`
|
||||
}
|
||||
if err := json.Unmarshal(vm["canonicalIdentity"], &canonical); err != nil {
|
||||
t.Fatalf("unmarshal canonicalIdentity: %v", err)
|
||||
}
|
||||
if len(canonical.SupersededIDs) != 2 {
|
||||
t.Fatalf("supersededIds = %v", canonical.SupersededIDs)
|
||||
}
|
||||
for _, alias := range canonical.Aliases {
|
||||
if alias == "vm-superseded-hash-1" || alias == "vm-superseded-hash-2" {
|
||||
t.Fatalf("aliases must not duplicate superseded ids, got %v", canonical.Aliases)
|
||||
}
|
||||
}
|
||||
|
||||
// Identical capability sets collapse onto one catalog entry.
|
||||
if refs["vm-100"] == "" || refs["vm-100"] != refs["vm-101"] {
|
||||
t.Fatalf("expected shared capabilitiesRef, got %q vs %q", refs["vm-100"], refs["vm-101"])
|
||||
}
|
||||
if _, ok := vm["capabilities"]; ok {
|
||||
t.Fatalf("catalog broadcast must not inline capabilities, got %s", vm["capabilities"])
|
||||
}
|
||||
entry, ok := catalog[refs["vm-100"]]
|
||||
if !ok {
|
||||
t.Fatalf("catalog missing ref %q (catalog %v)", refs["vm-100"], catalog)
|
||||
}
|
||||
var decoded []unifiedresources.ResourceCapability
|
||||
if err := json.Unmarshal(entry, &decoded); err != nil {
|
||||
t.Fatalf("decode catalog entry: %v", err)
|
||||
}
|
||||
if len(decoded) != 1 || decoded[0].Name != "restart" {
|
||||
t.Fatalf("catalog entry = %+v", decoded)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user