gsadmin/pulse
1
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-09-10 02:25:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge metrics-only resource deltas by patch shape

Browse Source 
The 2026-08-25 evening perf pass re-profiled the remaining frontend
costs on the 50-node mock rig after d5440ff43. Idle main-thread burn on
a throttled phone (~10.5s of long tasks per 30s on the worst-case mock)
and residual warm tab-entry cost both decomposed into per-changed-row
work that ignores what actually changed: every metrics tick deep-cloned,
re-canonicalized, and fully re-merged each patched row, and both store
commits then deep-unwrapped the merged rows again through whole-row
keyed reconciles.

The server's resource delta is a JSON merge patch, so the change shape
is already known. applyResourceStateDelta now records the top-level
keys each patch touched (platformData expanded one level), the
connection store publishes them with the resource revision and unions
them across the bounded history and the hidden-tab deferral set, and
mergeCanonicalResourceDeltaSnapshot takes a fast path for changed
non-host rows whose keys stay within the pass-through metric fields,
the proxmox facet mirror, and the four platformData metric mirror
leaves: the previous display row with just those subtrees cloned in
(a manual plain-data clone — structuredClone's per-invocation setup
dominates at this size). Both commit sites write fast rows as per-key
subtree patches (nested reconcile for records, direct sets for
primitives and platformData leaves) instead of whole-row reconciles,
and the connection store commits aligned ticks per index. Any other
change shape — additions, removals, repeated patches in one frame,
structural keys, agent rows (host coalescing) — keeps the full
clone-canonicalize-merge path, and the fast output is pinned
content-equivalent to it in resourceStateAdapters tests; a websocket
store test pins the per-key change shapes, the meta-history union, and
its unknown-shape contamination.

Measured on the pulse-dev rig (VM-to-VM, 50-node/1500-resource mock,
worst-case RandomMetrics, vs two d5440ff43 baseline runs): mobile-4x
idle long tasks 10.8/10.2s -> 3.9s per 30s; warm Alerts entry 5-rep
median long-task total 3.85s -> 1.52s; mobile warm Overview return
3.8/1.7s -> 1.0s; desktop Backups entry settle 1.06/1.20s -> 0.71s;
warm Storage/Overview sub-tab switch medians flat within rep spread;
desktop cold load and /api/state size unchanged.

Browser-verified on the rig build of this tree: live ticks render
through the fast path with per-cell DOM-vs-store consistency at two
samples 30s apart, zero console errors, desktop and 390px/4x mobile.
This commit is contained in:
rcourtman
2026-08-25 15:39:24 +01:00
parent 6e5571997e
commit 054ef7de13
11 changed files with 939 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/release-control/v6/internal/subsystems/alerts.md
+10 -2
View File
@@ -302,10 +302,18 @@ remains free to hydrate without waiting for the retired connection's request
to settle.
While the document is hidden, that same connection-scoped baseline must keep
accepting resource deltas without reconciling the visible resource store on
every message. The store accumulates changed resource IDs and performs one
canonical catch-up reconciliation on `visibilitychange` to visible. Alert and
every message. The store accumulates changed resource IDs (and their per-key
change shapes, unioned across the hidden ticks with unknown-shape
contamination) and performs one canonical catch-up reconciliation on
`visibilitychange` to visible. Alert and
resolved-alert truth continues to update normally while hidden; this resource
optimization must not defer, clear, or reinterpret alert lifecycle state.
The visible-tick resource commit may apply metrics-only rows as per-key
subtree writes instead of whole-row reconciles, but alert truth is outside
that fast path entirely: active-alert and resolved-alert stores keep their
own commit path, resource `alerts` facets are not in the fast-path
allow-list, and a row whose patch touches alert-relevant structure always
takes the full canonical merge.
Operational evidence and lifecycle identity are typed through
`internal/operationaltrust`. Evidence envelopes distinguish completeness,
confidence, permissions, freshness, correlation, and bounded provider detail.
docs/release-control/v6/internal/subsystems/performance-and-scalability.md
+18 -1
View File
@@ -226,7 +226,24 @@ 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
reserved for uncovered gaps and full-snapshot commits.
Per-tick reconciliation work must additionally scale with the patch, not the
row: the server's resource delta is a JSON merge patch, and each
reconciliation records which top-level keys every patch touched
(`platformData` expanded one level). A changed non-host row whose recorded
keys are confined to the pass-through metric fields (`cpu`, `memory`, `disk`,
`network`, `diskIO`, `temperature`, `uptime`, `lastSeen`, `status`), the
`proxmox` facet mirror, and the four `platformData` metric mirror leaves
(`diskRead`, `diskWrite`, `networkIn`, `networkOut`) is merged as the previous
display row with only those subtrees cloned in — no full-row deep clone,
re-canonicalization, or full canonical merge — and commits to the resource
stores as per-key subtree writes instead of a whole-row keyed reconcile, so
neither commit path deep-walks the unchanged remainder of the row. The
changed-key shapes ride the revision history (and the hidden-tab deferral
set) with unknown-shape contamination: rows added, removed, patched twice in
one frame, or patched outside the allow-list fall back to the full
clone-canonicalize-merge path, and the fast output must remain
content-equivalent to that path. 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
docs/release-control/v6/internal/subsystems/storage-recovery.md
+10
View File
@@ -1138,6 +1138,16 @@ recovery scope, or a storage/recovery-owned secret source.
reorder the REST-first hydration guarantee above, and a skipped store read
must never leave a storage/recovery projection behind the shared cache's
applied revision.
The same invisibility bound covers the per-key fast merge: a changed row
whose recorded patch keys stay within the metric fast-path allow-list may
be merged as the previous display row with only the patched subtrees
cloned in, and committed to instance projections as per-key subtree writes
— but the result must stay content-equivalent to the full
clone-canonicalize-merge. The `storage` facet, structural `platformData`
keys, and identity fields are all outside the allow-list, so a patch
touching any of them forces the full merge path, and no
storage/recovery-visible field may change value, presence, or merge
semantics because of the fast path.
Shared chart transports in `internal/api/chartapi/service.go` must follow the same
rule in mock mode: `/api/storage-charts` and adjacent infrastructure chart
payloads must read through `GetUnifiedReadStateOrSnapshot()` so storage and
docs/release-control/v6/internal/subsystems/unified-resources.md
+15 -1
View File
@@ -1192,7 +1192,21 @@ AI-only summary payloads, or page-local heuristics.
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
fall back to keyed full reconciliation.
Each reconciliation also records the per-resource top-level keys its merge
patches touched (`platformData` expanded one level), published with the
revision and unioned across the history window and the hidden-tab deferral
set with unknown-shape contamination. A changed non-host row whose recorded
keys stay within the pass-through metric fields, the `proxmox` facet
mirror, and the `platformData` metric mirror leaves takes a fast merge
path: the previous display row with only the patched subtrees cloned in,
bypassing the full clone-canonicalize-merge, and committing to the
connection store and instance projections as per-key subtree writes rather
than whole-row keyed reconciles. The fast output must stay
content-equivalent to the full path (facet keeps, deletion semantics, and
default-policy synthesis included), must never adopt raw-baseline subtrees
by reference, and any row outside the allow-list — including agent rows,
whose output can depend on host coalescing — must take the full path. Route-prefetch and route-realtime
activation are separate:
a prefetched hidden surface may retain REST data without subscribing its full
projection to every realtime tick, and activation catches up from the shared
frontend-modern/browser-verification.json
+39 -15
View File
@@ -1,21 +1,44 @@
{
"version": 1,
"base_sha": "d5440ff43efd4964cabdcf3ae0beea0306cb94d7",
"verified_at": "2026-08-25T14:14:16Z",
"base_sha": "6e5571997e8b94b0a698c4c5c7486548c5ef7d42",
"verified_at": "2026-08-25T14:27:03Z",
"result": "passed",
"changed_paths": [
"frontend-modern/src/features/storageBackups/diskPresentation.ts"
"frontend-modern/src/hooks/useUnifiedResources.ts",
"frontend-modern/src/stores/websocket-global.ts",
"frontend-modern/src/stores/websocket.ts",
"frontend-modern/src/utils/resourceStateAdapters.ts"
],
"content_sha256": {
"frontend-modern/src/features/storageBackups/diskPresentation.ts": "c84bcda4c2f41c581a4985abbc0bb60df31d2196f5722a81dafcd39bda8f02a8"
"frontend-modern/src/hooks/useUnifiedResources.ts": "7cad439395da8c5ccb281abe4b38313d63d45f87d38add8574120be5d1ab76a8",
"frontend-modern/src/stores/websocket-global.ts": "644e5bccf8612259188d8cf3fd8b89e7b30fcc326bc717b69745f0b0c5ae5d6c",
"frontend-modern/src/stores/websocket.ts": "e59b92a87db8f373129e7321e0bc06222ad59c57d25703aaf8f8ffc1d07b8fad",
"frontend-modern/src/utils/resourceStateAdapters.ts": "099acb8c55827396ef0722b6b5afdd8d742e3811434ca498de4604ad05edb8fa"
},
"routes": [
"/proxmox/storage"
"/proxmox/overview",
"/proxmox/storage",
"/proxmox/backups",
"/proxmox/replication",
"/proxmox/ceph",
"/docker",
"/kubernetes",
"/truenas",
"/vmware",
"/standalone",
"/alerts",
"/ai",
"/actions",
"/settings"
],
"viewports": [
{
"width": 1440,
"height": 900
},
{
"width": 1280,
"height": 800
"height": 720
},
{
"width": 390,
@@ -23,16 +46,17 @@
}
],
"states": [
"Physical Disks default inventory",
"provider-scoped search results",
"expanded physical-disk detail",
"restored default inventory after clearing search"
"live realtime metric ticks flowing through the patched-keys fast merge on the 50-node/1500-resource mock estate (pulse-dev rig build of this exact tree on :7696)",
"per-cell DOM-vs-store consistency for workloads-table CPU cells across cohort updates (11/12 rows within 1% at two samples 30s apart; 12th is a group header without a CPU cell)",
"warm tab re-entry catch-up through the changed-key history (Alerts, Proxmox, Storage/Overview sub-tab cycles)",
"30s idle observation on Proxmox Overview at desktop and 390px/4x-throttle mobile",
"nodes / workloads / storage tables rendered and live under realtime deltas; zero console errors across all probes on both viewport classes"
],
"interactions": [
"opened the Physical Disks view",
"filtered disks by owning Proxmox node",
"expanded and collapsed a physical-disk detail",
"cleared search and restored the full inventory",
"verified no horizontal table overflow at 390px"
"login",
"desktop platform-tab rail navigation across all platforms and utility tabs",
"mobile bottom-bar navigation including platform-switcher and More sheets",
"Proxmox sub-tab cycle (Storage, Backups, Replication, Ceph, Overview) with repeated warm switches",
"repeated warm Alerts entry (5-rep A/B harness)"
]
}
frontend-modern/src/hooks/useUnifiedResources.ts
+101 -25
View File
@@ -37,9 +37,12 @@ import { canonicalizeFrontendResourceType } from '@/utils/resourceTypeCompat';
import { getPreferredNormalizedPlatformId } from '@/utils/resourceIdentity';
import { getExplicitResourceClusterName } from '@/utils/agentResources';
import {
buildFastResourceStorePatchOps,
getFastResourceMergePatchKeys,
mergeCanonicalResource,
mergeCanonicalResourceDeltaSnapshot,
mergeCanonicalResourceSnapshot,
type ResourceChangedKeys,
} from '@/utils/resourceStateAdapters';
import { RESOURCE_METADATA_CHANGED_EVENT } from '@/utils/resourceMetadataEvents';
import {
@@ -1621,6 +1624,7 @@ export function useUnifiedResources(options?: UseUnifiedResourcesOptions) {
const resourceChange = wsStore.resourceChange?.() ?? {
version: Number(lastUpdateToken) || 0,
changedIds: null,
changedKeys: null,
};
// For normal page loads, keep the first paint on the canonical REST contract.
// Only explicit websocket-first consumers are allowed to render directly
@@ -1638,16 +1642,33 @@ export function useUnifiedResources(options?: UseUnifiedResourcesOptions) {
// 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;
// The per-key change shapes ride along so metrics-only rows keep fast-path
// eligibility across the whole catch-up window.
type CatchUpMeta = {
changedIds: ReadonlySet<string>;
changedKeys: ResourceChangedKeys | null;
};
const allEntryCatchUpIds =
const resolveCatchUpMeta = (sinceVersion: number): CatchUpMeta | null => {
if (sinceVersion <= 0 || resourceChange.version <= sinceVersion) return null;
if (resourceChange.version === sinceVersion + 1) {
return resourceChange.changedIds
? {
changedIds: resourceChange.changedIds,
changedKeys: resourceChange.changedKeys ?? null,
}
: null;
}
if (wsStore.changedResourceMetaSince) {
return wsStore.changedResourceMetaSince(sinceVersion);
}
const catchUpIds = wsStore.changedResourceIdsSince?.(sinceVersion) ?? null;
return catchUpIds ? { changedIds: catchUpIds, changedKeys: null } : null;
};
const allEntryCatchUp =
allResourcesEntry.realtimeVersion > 0
? resolveCatchUpIds(allResourcesEntry.realtimeVersion)
? resolveCatchUpMeta(allResourcesEntry.realtimeVersion)
: null;
const canApplyIncrementally = allEntryCatchUpIds !== null;
const canApplyIncrementally = allEntryCatchUp !== null;
const realtimeSnapshotAlreadyApplied =
resourceChange.version > 0 &&
allResourcesEntry.hasSnapshot &&
@@ -1660,27 +1681,47 @@ export function useUnifiedResources(options?: UseUnifiedResourcesOptions) {
// store read entirely when this realtime version is already applied.
// untrack keeps the per-item id/type reads from registering thousands of
// fine-grained dependencies; the effect is driven by resourceChange().
const readWsResources = (changedIds: ReadonlySet<string> | null): Resource[] =>
const readWsResources = (catchUp: CatchUpMeta | null): Resource[] =>
untrack(() => {
const stored = wsStore.state.resources;
if (!Array.isArray(stored)) return [];
if (changedIds === null) return unwrap(stored) as Resource[];
const cachedIds = new Set(allResourcesEntry.resources.map((resource) => resource.id));
return stored.map((resource) =>
resource.type === 'agent' || changedIds.has(resource.id) || !cachedIds.has(resource.id)
? (unwrap(resource) as Resource)
: (resource as Resource),
if (catchUp === null) return unwrap(stored) as Resource[];
const cachedById = new Map(
allResourcesEntry.resources.map((resource) => [resource.id, resource] as const),
);
return stored.map((resource) => {
if (resource.type === 'agent' || !cachedById.has(resource.id)) {
return unwrap(resource) as Resource;
}
if (!catchUp.changedIds.has(resource.id)) {
return resource as Resource;
}
// Fast-path rows are read per patched key inside the merge (an O(1)
// proxy escape per subtree); only full-merge rows need the deep
// unwrap that structuredClone requires.
return getFastResourceMergePatchKeys(
catchUp.changedKeys ?? undefined,
resource.id,
cachedById.get(resource.id),
)
? (resource as Resource)
: (unwrap(resource) as Resource);
});
});
const mergedWsResources = realtimeSnapshotAlreadyApplied
? allResourcesEntry.resources
: canApplyIncrementally
? mergeCanonicalResourceDeltaSnapshot(
readWsResources(allEntryCatchUpIds),
allResourcesEntry.resources,
allEntryCatchUpIds!,
? untrack(() =>
mergeCanonicalResourceDeltaSnapshot(
readWsResources(allEntryCatchUp),
allResourcesEntry.resources,
allEntryCatchUp!.changedIds,
allEntryCatchUp!.changedKeys ?? undefined,
),
)
: mergeCanonicalResourceSnapshot(readWsResources(null), allResourcesEntry.resources);
: untrack(() =>
mergeCanonicalResourceSnapshot(readWsResources(null), allResourcesEntry.resources),
);
const projectedResources = filterCanonicalUnifiedResources(
mergedWsResources,
query,
@@ -1717,27 +1758,30 @@ export function useUnifiedResources(options?: UseUnifiedResourcesOptions) {
allResourcesEntry.realtimeVersion = resourceChange.version;
}
const cacheEntryCatchUpIds =
const cacheEntryCatchUp =
cacheEntry.hasSnapshot &&
cacheEntry.realtimeVersion > 0 &&
resolvedProjectedResources === projectedResources
? resolveCatchUpIds(cacheEntry.realtimeVersion)
? resolveCatchUpMeta(cacheEntry.realtimeVersion)
: null;
const canPatchProjectionIncrementally = cacheEntryCatchUpIds !== null;
const canPatchProjectionIncrementally = cacheEntryCatchUp !== null;
const changedResourceTouchesAgent =
canPatchProjectionIncrementally &&
mergedWsResources.some(
(resource) => resource.type === 'agent' && cacheEntryCatchUpIds!.has(resource.id),
(resource) => resource.type === 'agent' && cacheEntryCatchUp!.changedIds.has(resource.id),
);
const incrementalPatchIndices = canPatchProjectionIncrementally
? resolveIncrementalResourcePatchIndices(
resources as unknown as Resource[],
resolvedProjectedResources,
cacheEntryCatchUpIds!,
cacheEntryCatchUp!.changedIds,
changedResourceTouchesAgent,
)
: null;
// Captured before the cache write below replaces it: the fast-commit
// eligibility check needs the row the instance store currently mirrors.
const previousProjectedResources = cacheEntry.resources;
setUnifiedResourcesCache(cacheEntry, resolvedProjectedResources, now);
cacheEntry.lastFetchAt = now;
cacheEntry.realtimeVersion = resourceChange.version;
@@ -1746,7 +1790,39 @@ export function useUnifiedResources(options?: UseUnifiedResourcesOptions) {
setResources(reconcile(resolvedProjectedResources, { key: 'id' }));
} else {
incrementalPatchIndices.forEach((index) => {
setResources(index, reconcile(resolvedProjectedResources[index], { key: 'id' }));
const nextRow = resolvedProjectedResources[index];
const previousRow = previousProjectedResources[index];
const fastKeys =
previousRow && previousRow.id === nextRow.id
? getFastResourceMergePatchKeys(
cacheEntryCatchUp!.changedKeys ?? undefined,
nextRow.id,
previousRow,
)
: null;
if (fastKeys) {
// A fast-path row differs from the store row only in its patched
// subtrees; committing those per key avoids the full-row reconcile
// whose unwrap pass deep-walks every field.
for (const op of buildFastResourceStorePatchOps(nextRow, fastKeys)) {
if (op.leaf !== undefined) {
setResources(
index,
op.key as keyof Resource,
op.leaf as never,
(op.mode === 'reconcile' ? reconcile(op.value) : op.value) as never,
);
} else {
setResources(
index,
op.key as keyof Resource,
(op.mode === 'reconcile' ? reconcile(op.value) : op.value) as never,
);
}
}
} else {
setResources(index, reconcile(nextRow, { key: 'id' }));
}
});
}
setPolicyPosture(cacheEntry.policyPosture);
frontend-modern/src/stores/__tests__/websocket-unified.test.ts
+107
View File
@@ -368,6 +368,113 @@ describe('websocket store unified resource contract', () => {
}
});
it('commits metrics-only deltas with per-key change shapes and a catch-up meta history', async () => {
const { store, dispose } = await createStoreHarness();
try {
await waitForOpenTick();
emitMessage({
type: 'initialState',
data: {
connectedInfrastructure: [],
resources: [
{
id: 'vm-1',
type: 'vm',
name: 'vm-1',
status: 'running',
lastSeen: 100,
cpu: { current: 10 },
memory: { current: 20, total: 1024, used: 256 },
tags: ['prod'],
platformData: {
sources: ['proxmox-pve'],
vmid: 100,
diskRead: 5,
networkIn: 100,
},
},
{ id: 'vm-2', type: 'vm', name: 'vm-2', status: 'running', lastSeen: 100 },
],
lastUpdate: 100,
activeAlerts: [],
recentlyResolved: [],
},
});
const baseVersion = store.resourceChange().version;
// Metrics-only tick: aligned commit, per-key change shape recorded with
// platformData expanded into its leaves.
emitMessage({
type: 'rawData',
data: {
lastUpdate: 200,
resourceDelta: {
upserts: [
{
id: 'vm-1',
lastSeen: 200,
cpu: { current: 55 },
platformData: { diskRead: 9, networkIn: 300 },
},
],
},
},
});
const row = store.state.resources[0];
expect(row?.cpu?.current).toBe(55);
expect(row?.lastSeen).toBe(200);
expect(row?.tags).toEqual(['prod']);
expect(row?.platformData).toMatchObject({ vmid: 100, diskRead: 9, networkIn: 300 });
expect(store.resourceChange().changedKeys?.get('vm-1')).toEqual([
'lastSeen',
'cpu',
'platformData.diskRead',
'platformData.networkIn',
]);
// Second tick with a structural key; the catch-up meta must union both
// ticks per id.
emitMessage({
type: 'rawData',
data: {
lastUpdate: 300,
resourceDelta: {
upserts: [
{ id: 'vm-1', cpu: { current: 60 } },
{ id: 'vm-2', tags: ['edge'] },
],
},
},
});
const meta = store.changedResourceMetaSince(baseVersion);
expect(meta).not.toBeNull();
expect(meta?.changedIds).toEqual(new Set(['vm-1', 'vm-2']));
expect(meta?.changedKeys.get('vm-1')).toEqual([
'lastSeen',
'cpu',
'platformData.diskRead',
'platformData.networkIn',
]);
expect(meta?.changedKeys.get('vm-2')).toEqual(['tags']);
// A removal marks the row's change shape unknown.
emitMessage({
type: 'rawData',
data: {
lastUpdate: 400,
resourceDelta: { removed: ['vm-2'] },
},
});
expect(store.changedResourceMetaSince(baseVersion)?.changedKeys.get('vm-2')).toBeNull();
expect(store.state.resources.map((resource) => resource.id)).toEqual(['vm-1']);
} finally {
dispose();
}
});
it('expands capabilitiesRef through the catalog and synthesizes omitted default policies', async () => {
const { store, dispose } = await createStoreHarness();
try {
frontend-modern/src/stores/websocket-global.ts
+3 -1
View File
@@ -25,7 +25,8 @@ const createNoopWebSocketStore = (): ReturnType<typeof createWebSocketStore> =>
const [resourceChange] = createSignal<{
version: number;
changedIds: ReadonlySet<string> | null;
}>({ version: 0, changedIds: null });
changedKeys: import('@/utils/resourceStateAdapters').ResourceChangedKeys | null;
}>({ version: 0, changedIds: null, changedKeys: null });
const [state] = createStore<State>({
connectedInfrastructure: [],
metrics: [],
@@ -64,6 +65,7 @@ const createNoopWebSocketStore = (): ReturnType<typeof createWebSocketStore> =>
updateProgress,
resourceChange,
changedResourceIdsSince: () => null,
changedResourceMetaSince: () => null,
shutdown: () => {},
reconnect: () => {},
switchUrl: () => {},
frontend-modern/src/stores/websocket.ts
+175 -16
View File
@@ -20,8 +20,12 @@ import {
isAppContainerDiscoveryResourceType,
} from '@/utils/discoveryTarget';
import {
buildFastResourceStorePatchOps,
getFastResourceMergePatchKeys,
mergeCanonicalResourceDeltaSnapshot,
mergeCanonicalResourceSnapshot,
unionResourceChangedKeys,
type ResourceChangedKeys,
} from '@/utils/resourceStateAdapters';
import { apiFetchJSON } from '@/utils/apiClient';
@@ -95,24 +99,58 @@ const applyJSONMergePatch = (current: unknown, patch: unknown): unknown => {
return result;
};
// Top-level keys a merge patch touches, with platformData expanded one level
// so the fast merge path can reason about the metric mirror leaves. Returns
// null when the patch replaces or deletes platformData wholesale — the change
// shape is then too coarse for a per-key fast path.
const collectResourcePatchKeys = (patch: Record<string, unknown>): readonly string[] | null => {
const keys: string[] = [];
for (const key of Object.keys(patch)) {
if (key === 'id') continue;
if (key === 'platformData') {
const platformData = asRecord(patch.platformData);
if (!platformData) return null;
for (const leaf of Object.keys(platformData)) keys.push(`platformData.${leaf}`);
continue;
}
keys.push(key);
}
return keys;
};
const applyResourceStateDelta = (
current: readonly Resource[],
delta: { upserts?: unknown; removed?: unknown; order?: unknown },
): { resources: Resource[]; changedIds: Set<string> } => {
): {
resources: Resource[];
changedIds: Set<string>;
changedKeys: Map<string, readonly string[] | null>;
} => {
const resourcesById = new Map(current.map((resource) => [resource.id, resource] as const));
const removed = Array.isArray(delta.removed)
? new Set(delta.removed.filter((id): id is string => typeof id === 'string'))
: new Set<string>();
removed.forEach((id) => resourcesById.delete(id));
const changedIds = new Set(removed);
const changedKeys = new Map<string, readonly string[] | null>();
removed.forEach((id) => changedKeys.set(id, null));
const addedIds: string[] = [];
if (Array.isArray(delta.upserts)) {
delta.upserts.forEach((patch) => {
const id = asString(asRecord(patch)?.id);
if (!id) return;
const patchRecord = asRecord(patch);
const id = asString(patchRecord?.id);
if (!id || !patchRecord) return;
changedIds.add(id);
if (!resourcesById.has(id)) addedIds.push(id);
const isNewRow = !resourcesById.has(id);
if (isNewRow) addedIds.push(id);
if (isNewRow || changedKeys.has(id)) {
// Added rows and repeated patches for one id in a single frame have no
// usable per-key shape; force the full merge path.
changedKeys.set(id, null);
} else {
changedKeys.set(id, collectResourcePatchKeys(patchRecord));
}
const next = applyJSONMergePatch(resourcesById.get(id), patch) as Resource;
if (next.id === id) resourcesById.set(id, next);
});
@@ -132,7 +170,7 @@ const applyResourceStateDelta = (
resourcesById.forEach((resource, id) => {
if (!included.has(id)) ordered.push(resource);
});
return { resources: ordered, changedIds };
return { resources: ordered, changedIds, changedKeys };
};
const parseTimestampMs = (value: unknown): number | null => {
@@ -252,15 +290,28 @@ export function createWebSocketStore(url: string) {
const [resourceChange, setResourceChange] = createSignal<{
version: number;
changedIds: ReadonlySet<string> | null;
}>({ version: 0, changedIds: null });
changedKeys: ResourceChangedKeys | null;
}>({ version: 0, changedIds: null, changedKeys: 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> }[] = [];
let resourceChangeHistory: {
version: number;
changedIds: ReadonlySet<string>;
changedKeys: ResourceChangedKeys | null;
}[] = [];
const changedResourceIdsSince = (sinceVersion: number): ReadonlySet<string> | null => {
const changedResourceIdsSince = (sinceVersion: number): ReadonlySet<string> | null =>
changedResourceMetaSince(sinceVersion)?.changedIds ?? null;
// Union of changed ids and their per-key change shapes across the history
// window (sinceVersion, current]. changedKeys entries degrade to null for a
// row whenever any covered tick could not describe its change shape.
const changedResourceMetaSince = (
sinceVersion: number,
): { changedIds: ReadonlySet<string>; changedKeys: ResourceChangedKeys } | null => {
if (sinceVersion >= resourceChangeVersion) return null;
if (resourceChangeHistory.length === 0) return null;
if (resourceChangeHistory[0].version > sinceVersion + 1) return null;
@@ -268,11 +319,19 @@ export function createWebSocketStore(url: string) {
return null;
}
const union = new Set<string>();
const keysUnion = new Map<string, readonly string[] | null>();
for (const entry of resourceChangeHistory) {
if (entry.version <= sinceVersion) continue;
entry.changedIds.forEach((id) => union.add(id));
entry.changedIds.forEach((id) => {
union.add(id);
const tickKeys = entry.changedKeys ? (entry.changedKeys.get(id) ?? null) : null;
keysUnion.set(
id,
keysUnion.has(id) ? unionResourceChangedKeys(keysUnion.get(id), tickKeys) : tickKeys,
);
});
}
return union;
return { changedIds: union, changedKeys: keysUnion };
};
// Track alerts with pending acknowledgment changes to prevent race conditions
@@ -362,6 +421,9 @@ export function createWebSocketStore(url: string) {
// and Solid's reconcile mutates adopted objects in place.
let rawServerResources: Resource[] | null = null;
const deferredResourceIds = new Set<string>();
// Per-key change shapes for deferred ids, unioned across the hidden ticks so
// the resume merge can still take the fast path for metrics-only rows.
const deferredResourceKeys = new Map<string, readonly string[] | null>();
// 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.
@@ -456,6 +518,7 @@ export function createWebSocketStore(url: string) {
const resetConnectionBaseline = () => {
rawServerResources = null;
deferredResourceIds.clear();
deferredResourceKeys.clear();
lastFullStateRecoveryAt = 0;
oversizedSnapshotObserved = false;
// A recovery request from a retired connection may still settle, but its
@@ -500,21 +563,84 @@ export function createWebSocketStore(url: string) {
});
};
const commitResources = (nextResources: Resource[], changedResourceIds?: ReadonlySet<string>) => {
const commitResources = (
nextResources: Resource[],
changedResourceIds?: ReadonlySet<string>,
changedResourceKeys?: ResourceChangedKeys,
) => {
logger.debug('[WebSocket] Updating resources', {
count: nextResources.length,
changedCount: changedResourceIds?.size ?? nextResources.length,
});
setState('resources', reconcile(nextResources, { key: 'id' }));
// A metrics tick leaves row count and order untouched, so it can commit as
// per-index writes instead of a whole-array reconcile (whose unwrap pass
// deep-walks every merged row). Delta-merge pass-through rows keep their
// store proxy identity, so the alignment scan skips them by equality.
const currentResources = state.resources;
let alignedPatchIndices: number[] | null = null;
if (
changedResourceIds &&
Array.isArray(currentResources) &&
currentResources.length === nextResources.length
) {
alignedPatchIndices = [];
for (let index = 0; index < nextResources.length; index += 1) {
const nextRow = nextResources[index];
const currentRow = currentResources[index];
if (nextRow === currentRow) continue;
if (!nextRow || !currentRow || nextRow.id !== currentRow.id) {
alignedPatchIndices = null;
break;
}
alignedPatchIndices.push(index);
}
}
if (alignedPatchIndices) {
for (const index of alignedPatchIndices) {
const row = nextResources[index];
const fastKeys = getFastResourceMergePatchKeys(
changedResourceKeys,
row.id,
currentResources[index],
);
if (fastKeys) {
for (const op of buildFastResourceStorePatchOps(row, fastKeys)) {
if (op.leaf !== undefined) {
setState(
'resources',
index,
op.key as keyof Resource,
op.leaf as never,
(op.mode === 'reconcile' ? reconcile(op.value) : op.value) as never,
);
} else {
setState(
'resources',
index,
op.key as keyof Resource,
(op.mode === 'reconcile' ? reconcile(op.value) : op.value) as never,
);
}
}
} else {
setState('resources', index, reconcile(row, { key: 'id' }));
}
}
} else {
setState('resources', reconcile(nextResources, { key: 'id' }));
}
const committedChangedIds = changedResourceIds ? new Set(changedResourceIds) : null;
const committedChangedKeys = changedResourceKeys ?? null;
setResourceChange({
version: ++resourceChangeVersion,
changedIds: committedChangedIds,
changedKeys: committedChangedKeys,
});
if (committedChangedIds) {
resourceChangeHistory.push({
version: resourceChangeVersion,
changedIds: committedChangedIds,
changedKeys: committedChangedKeys,
});
if (resourceChangeHistory.length > RESOURCE_CHANGE_HISTORY_LIMIT) {
resourceChangeHistory.splice(
@@ -538,13 +664,16 @@ export function createWebSocketStore(url: string) {
}
const changedResourceIds = new Set(deferredResourceIds);
const changedResourceKeys = new Map(deferredResourceKeys);
deferredResourceIds.clear();
deferredResourceKeys.clear();
const nextResources = mergeCanonicalResourceDeltaSnapshot(
rawServerResources,
state.resources,
changedResourceIds,
changedResourceKeys,
);
batch(() => commitResources(nextResources, changedResourceIds));
batch(() => commitResources(nextResources, changedResourceIds, changedResourceKeys));
};
if (typeof document !== 'undefined') {
@@ -714,6 +843,7 @@ export function createWebSocketStore(url: string) {
// Handle unified resources
let nextResources: Resource[] | undefined;
let changedResourceIds: ReadonlySet<string> | undefined;
let changedResourceKeys: ResourceChangedKeys | undefined;
if (message.data.resources !== undefined) {
deferredResourceIds.clear();
if (Array.isArray(message.data.resources)) {
@@ -761,14 +891,38 @@ export function createWebSocketStore(url: string) {
rawServerResources = appliedDelta.resources;
changedResourceIds = appliedDelta.changedIds;
if (typeof document !== 'undefined' && document.visibilityState === 'hidden') {
changedResourceIds.forEach((id) => deferredResourceIds.add(id));
changedResourceIds.forEach((id) => {
const tickKeys = appliedDelta.changedKeys.get(id) ?? null;
deferredResourceKeys.set(
id,
deferredResourceIds.has(id)
? unionResourceChangedKeys(deferredResourceKeys.get(id), tickKeys)
: tickKeys,
);
deferredResourceIds.add(id);
});
} else {
const combinedKeys = appliedDelta.changedKeys;
deferredResourceIds.forEach((id) => {
combinedKeys.set(
id,
combinedKeys.has(id)
? unionResourceChangedKeys(
combinedKeys.get(id),
deferredResourceKeys.get(id) ?? null,
)
: (deferredResourceKeys.get(id) ?? null),
);
});
changedResourceIds = new Set([...deferredResourceIds, ...changedResourceIds]);
deferredResourceIds.clear();
deferredResourceKeys.clear();
changedResourceKeys = combinedKeys;
nextResources = mergeCanonicalResourceDeltaSnapshot(
rawServerResources,
state.resources,
changedResourceIds,
changedResourceKeys,
);
}
} else {
@@ -780,7 +934,7 @@ export function createWebSocketStore(url: string) {
}
}
if (nextResources !== undefined) {
commitResources(nextResources, changedResourceIds);
commitResources(nextResources, changedResourceIds, changedResourceKeys);
}
// Sync active alerts from state
if (message.data.activeAlerts !== undefined) {
@@ -1248,6 +1402,7 @@ export function createWebSocketStore(url: string) {
updateProgress,
resourceChange,
changedResourceIdsSince,
changedResourceMetaSince,
shutdown,
reconnect: () => {
if (isDisposed) return;
@@ -1273,7 +1428,11 @@ export function createWebSocketStore(url: string) {
setReconnecting(false);
setInitialDataReceived(false);
setUpdateProgress(null);
setResourceChange({ version: ++resourceChangeVersion, changedIds: null });
setResourceChange({
version: ++resourceChangeVersion,
changedIds: null,
changedKeys: null,
});
resourceChangeHistory = [];
setState(reconcile(createInitialState()));
setActiveAlerts(reconcile({}));
frontend-modern/src/utils/__tests__/resourceStateAdapters.test.ts
+242
View File
@@ -1,11 +1,14 @@
import { describe, expect, it } from 'vitest';
import {
buildFastResourceStorePatchOps,
getFastResourceMergePatchKeys,
mergeCanonicalResourceDeltaSnapshot,
mergeCanonicalResourceSnapshot,
nodeFromResource,
pbsInstanceFromResource,
pmgInstanceFromResource,
unionResourceChangedKeys,
} from '../resourceStateAdapters';
import type { Resource } from '@/types/resource';
@@ -1168,3 +1171,242 @@ describe('incremental canonical resource snapshots', () => {
expect(result[0]?.sources).toEqual(expect.arrayContaining(['agent', 'proxmox-pve']));
});
});
describe('fast merge path for metrics-only delta patches', () => {
// Mirrors the wire shape of a PVE guest metrics tick captured from the mock
// estate: top-level metric subtrees, the proxmox facet mirror, and the four
// platformData metric mirror leaves.
const createPveGuestRaw = (): Resource =>
({
id: 'vm-fast-1',
type: 'vm',
name: 'vm-fast-1',
displayName: 'VM Fast 1',
platformId: 'pve-node-1',
platformType: 'proxmox-pve',
sourceType: 'api',
status: 'running',
lastSeen: 1700000000000,
uptime: 1000,
cpu: { current: 10 },
memory: { current: 20, total: 1024, used: 256, free: 768 },
disk: { current: 30, total: 2048, used: 512, free: 1536 },
network: { rxBytes: 100, txBytes: 200 },
diskIO: { readRate: 5, writeRate: 7 },
tags: ['prod'],
canonicalIdentity: { canonicalId: 'vm-fast-1', aliases: ['vm/100'] },
proxmox: {
vmid: 100,
nodeName: 'pve-node-1',
memory: { free: 768, usage: 20, used: 256 },
uptime: 1000,
},
platformData: {
sources: ['proxmox-pve'],
instance: 'pve-node-1',
vmid: 100,
diskRead: 5,
diskWrite: 7,
networkIn: 100,
networkOut: 200,
proxmox: { vmid: 100, nodeName: 'pve-node-1' },
},
}) as unknown as Resource;
const METRICS_PATCH_KEYS = [
'cpu',
'uptime',
'lastSeen',
'memory',
'network',
'diskIO',
'proxmox',
'platformData.diskRead',
'platformData.diskWrite',
'platformData.networkIn',
'platformData.networkOut',
] as const;
const applyMetricsTick = (raw: Resource): Resource => {
const next = structuredClone(raw) as unknown as Record<string, unknown>;
next.cpu = { current: 55 };
next.uptime = 1002;
next.lastSeen = 1700000002000;
next.memory = { current: 40, total: 1024, used: 400, free: 624 };
next.network = { rxBytes: 300, txBytes: 400 };
next.diskIO = { readRate: 9, writeRate: 11 };
next.proxmox = {
...(next.proxmox as Record<string, unknown>),
memory: { free: 624, usage: 40, used: 400 },
uptime: 1002,
};
next.platformData = {
...(next.platformData as Record<string, unknown>),
diskRead: 9,
diskWrite: 11,
networkIn: 300,
networkOut: 400,
};
return next as unknown as Resource;
};
const seedDisplayRows = (raw: Resource[]): Resource[] =>
mergeCanonicalResourceDeltaSnapshot(
raw.map((row) => structuredClone(row)),
[],
new Set(raw.map((row) => row.id)),
);
it('produces the same merged row as the full path for a metrics-only patch', () => {
const raw = createPveGuestRaw();
const display = seedDisplayRows([raw]);
const patched = applyMetricsTick(raw);
const changedIds = new Set(['vm-fast-1']);
const changedKeys = new Map([['vm-fast-1', [...METRICS_PATCH_KEYS]]]);
const fast = mergeCanonicalResourceDeltaSnapshot([patched], display, changedIds, changedKeys);
const slow = mergeCanonicalResourceDeltaSnapshot(
[structuredClone(patched)],
display,
changedIds,
);
expect(fast[0]).toEqual(slow[0]);
expect(fast[0]?.cpu?.current).toBe(55);
expect(fast[0]?.uptime).toBe(1002);
expect((fast[0]?.platformData as Record<string, unknown>)?.diskRead).toBe(9);
expect((fast[0]?.proxmox as Record<string, unknown>)?.uptime).toBe(1002);
});
it('keeps unpatched subtree identity on the fast path', () => {
const raw = createPveGuestRaw();
const display = seedDisplayRows([raw]);
const patched = applyMetricsTick(raw);
const changedKeys = new Map([['vm-fast-1', [...METRICS_PATCH_KEYS]]]);
const fast = mergeCanonicalResourceDeltaSnapshot(
[patched],
display,
new Set(['vm-fast-1']),
changedKeys,
);
expect(fast[0]).not.toBe(display[0]);
expect(fast[0]?.canonicalIdentity).toBe(display[0]?.canonicalIdentity);
expect(fast[0]?.disk).toBe(display[0]?.disk);
expect(fast[0]?.tags).toBe(display[0]?.tags);
// platformData is rebuilt for the patched leaves but its untouched nested
// records keep identity so downstream reconciles short-circuit on them.
expect((fast[0]?.platformData as Record<string, unknown>)?.proxmox).toBe(
(display[0]?.platformData as Record<string, unknown>)?.proxmox,
);
// The fast row must not adopt the raw row's subtrees: reconcile mutates
// adopted objects in place and would corrupt the raw delta baseline.
expect(fast[0]?.cpu).not.toBe(patched.cpu);
expect(fast[0]?.proxmox).not.toBe(patched.proxmox);
});
it('falls back to the full merge when a patch touches an ineligible key', () => {
const raw = createPveGuestRaw();
const display = seedDisplayRows([raw]);
const patched = {
...applyMetricsTick(raw),
tags: ['prod', 'new-tag'],
} as unknown as Resource;
const changedKeys = new Map([['vm-fast-1', [...METRICS_PATCH_KEYS, 'tags']]]);
expect(getFastResourceMergePatchKeys(changedKeys, 'vm-fast-1', display[0])).toBeNull();
const result = mergeCanonicalResourceDeltaSnapshot(
[patched],
display,
new Set(['vm-fast-1']),
changedKeys,
);
expect(result[0]?.tags).toEqual(['prod', 'new-tag']);
expect(result[0]?.cpu?.current).toBe(55);
});
it('falls back to the full merge for platformData leaves outside the metric mirrors', () => {
const display = seedDisplayRows([createPveGuestRaw()]);
const changedKeys = new Map([['vm-fast-1', ['cpu', 'platformData.instance']]]);
expect(getFastResourceMergePatchKeys(changedKeys, 'vm-fast-1', display[0])).toBeNull();
});
it('treats unknown change shapes and agent rows as ineligible', () => {
const display = seedDisplayRows([createPveGuestRaw()]);
expect(
getFastResourceMergePatchKeys(new Map([['vm-fast-1', null]]), 'vm-fast-1', display[0]),
).toBeNull();
expect(getFastResourceMergePatchKeys(undefined, 'vm-fast-1', display[0])).toBeNull();
expect(
getFastResourceMergePatchKeys(new Map([['vm-fast-1', ['cpu']]]), 'vm-fast-1', undefined),
).toBeNull();
const agent = { ...createPveGuestRaw(), type: 'agent' } as unknown as Resource;
expect(
getFastResourceMergePatchKeys(new Map([['vm-fast-1', ['cpu']]]), 'vm-fast-1', agent),
).toBeNull();
});
it('keeps the existing value when a fast key was deleted from the raw row', () => {
const raw = createPveGuestRaw();
const display = seedDisplayRows([raw]);
const patched = structuredClone(raw) as unknown as Record<string, unknown>;
delete patched.diskIO;
const changedKeys = new Map([['vm-fast-1', ['diskIO']]]);
const fast = mergeCanonicalResourceDeltaSnapshot(
[patched as unknown as Resource],
display,
new Set(['vm-fast-1']),
changedKeys,
);
const slow = mergeCanonicalResourceDeltaSnapshot(
[structuredClone(patched) as unknown as Resource],
display,
new Set(['vm-fast-1']),
);
expect(fast[0]?.diskIO).toEqual(display[0]?.diskIO);
expect(fast[0]).toEqual(slow[0]);
});
it('unions changed keys across ticks with null contamination', () => {
expect(unionResourceChangedKeys(['cpu'], ['memory', 'cpu'])).toEqual(['cpu', 'memory']);
expect(unionResourceChangedKeys(['cpu'], null)).toBeNull();
expect(unionResourceChangedKeys(null, ['cpu'])).toBeNull();
expect(unionResourceChangedKeys(undefined, ['cpu'])).toEqual(['cpu']);
expect(unionResourceChangedKeys(['cpu'], undefined)).toEqual(['cpu']);
});
it('builds per-key store patch ops for a fast row', () => {
const raw = createPveGuestRaw();
const display = seedDisplayRows([raw]);
const patched = applyMetricsTick(raw);
const changedKeys = new Map([['vm-fast-1', [...METRICS_PATCH_KEYS]]]);
const fast = mergeCanonicalResourceDeltaSnapshot(
[patched],
display,
new Set(['vm-fast-1']),
changedKeys,
);
const ops = buildFastResourceStorePatchOps(fast[0]!, [...METRICS_PATCH_KEYS]);
const byTarget = new Map(ops.map((op) => [op.leaf ? `${op.key}.${op.leaf}` : op.key, op]));
expect(byTarget.get('cpu')?.mode).toBe('reconcile');
expect(byTarget.get('uptime')).toEqual({ key: 'uptime', value: 1002, mode: 'set' });
expect(byTarget.get('platformData.diskRead')).toEqual({
key: 'platformData',
leaf: 'diskRead',
value: 9,
mode: 'set',
});
expect(byTarget.get('proxmox')?.mode).toBe('reconcile');
// No op may reference platformData wholesale; only leaf writes are allowed
// so unpatched platformData subtrees never get touched in the store.
expect(ops.every((op) => op.key !== 'platformData' || op.leaf !== undefined)).toBe(true);
});
});
frontend-modern/src/utils/resourceStateAdapters.ts
+219 -1
View File
@@ -20,6 +20,8 @@ import type {
PMGSpamBucket,
Temperature,
} from '@/types/api';
import { $RAW } from 'solid-js/store';
import type { Resource } from '@/types/resource';
import {
getActionableAgentIdFromResource,
@@ -836,16 +838,218 @@ export const mergeCanonicalResourceSnapshot = (
);
};
// Per-resource top-level keys touched by a server delta's JSON merge patches.
// `platformData` is expanded one level into `platformData.<leaf>` entries. A
// `null` value means the change shape is unknown (row added/removed, whole
// subtree replaced, deferred ticks) and the row must take the full merge path.
export type ResourceChangedKeys = ReadonlyMap<string, readonly string[] | null>;
// Union of two per-resource changed-key lists across ticks. Unknown (`null`)
// contaminates: once a tick could not describe a row's change shape, no later
// tick can restore fast-path eligibility for that row.
export const unionResourceChangedKeys = (
first: readonly string[] | null | undefined,
second: readonly string[] | null | undefined,
): readonly string[] | null => {
if (first === undefined) return second ?? null;
if (second === undefined) return first ?? null;
if (first === null || second === null) return null;
return Array.from(new Set([...first, ...second]));
};
// Top-level Resource fields the canonical merge takes verbatim from the
// incoming row (plain `...incoming` spread, no special handling) and that the
// canonicalization pass never reads. A patch confined to these fields cannot
// change platform/source resolution, facet keeps, or identity merging, so the
// merged display row is the previous display row with just these subtrees
// replaced. `diskIO` is special-cased in the merge (`incoming ?? existing`)
// but a merge-patch either sets it (incoming wins) or deletes it (existing
// survives both paths), so it stays equivalent.
const FAST_MERGE_TOP_KEYS = new Set([
'cpu',
'memory',
'disk',
'network',
'diskIO',
'temperature',
'uptime',
'lastSeen',
'status',
]);
const FAST_MERGE_PLATFORM_DATA_PREFIX = 'platformData.';
// platformData leaves the metric mirror writes touch every tick. None of them
// are read by canonicalizeLegacyPlatformData or the source-list derivation, so
// patching them cannot alter canonicalization output beyond the leaf values.
const FAST_MERGE_PLATFORM_DATA_KEYS = new Set(['diskRead', 'diskWrite', 'networkIn', 'networkOut']);
// O(1) escape from a Solid store proxy; identity for plain values. Never use
// store `unwrap` here: it deep-walks plain objects, which is the cost this
// path exists to avoid.
const rawStoreValue = <T>(value: T): T =>
(value != null && ((value as { [$RAW]?: T })[$RAW] as T)) || value;
// Returns the validated changed-key list when `id`'s row can skip the full
// clone+canonicalize+merge, or null when it must take the full path. The same
// predicate gates the per-key store commits, so it must stay conservative:
// anything it accepts is asserted to leave every other field of the merged
// display row untouched.
export const getFastResourceMergePatchKeys = (
changedKeys: ResourceChangedKeys | undefined,
id: string,
existing: Resource | undefined,
): readonly string[] | null => {
if (!changedKeys || !existing) return null;
// Agent rows can join host-coalescing groups; their merged output is not a
// per-row function of the patch.
if (existing.type === 'agent') return null;
const keys = changedKeys.get(id);
if (!keys || keys.length === 0) return null;
let touchesPlatformData = false;
for (const key of keys) {
if (FAST_MERGE_TOP_KEYS.has(key) || key === 'proxmox') continue;
if (key.startsWith(FAST_MERGE_PLATFORM_DATA_PREFIX)) {
if (!FAST_MERGE_PLATFORM_DATA_KEYS.has(key.slice(FAST_MERGE_PLATFORM_DATA_PREFIX.length))) {
return null;
}
touchesPlatformData = true;
continue;
}
return null;
}
if (
touchesPlatformData &&
!asRecord(rawStoreValue(existing as unknown as JsonRecord).platformData)
) {
return null;
}
return keys;
};
// Patched subtrees are JSON-derived plain data and typically tiny (a handful
// of numeric leaves). A manual walk clones them several times faster than
// structuredClone, whose per-invocation setup dominates at this size; symbol
// keys (Solid's internal store markers on raw nodes) are skipped by design.
const clonePlainValue = <T>(value: T): T => {
if (value === null || typeof value !== 'object') return value;
if (Array.isArray(value)) return value.map(clonePlainValue) as unknown as T;
const source = value as Record<string, unknown>;
const out: Record<string, unknown> = {};
for (const key of Object.keys(source)) out[key] = clonePlainValue(source[key]);
return out as T;
};
const cloneFastPatchValue = (value: unknown): unknown => clonePlainValue(value);
// Fast-path counterpart of mergeCanonicalResource for rows whose patch passed
// getFastResourceMergePatchKeys: the previous display row with the patched
// subtrees cloned in. Content-equivalent to the full path; unpatched subtrees
// keep their object identity so downstream reconciles short-circuit on them.
const applyFastResourceMergePatch = (
incoming: Resource,
existing: Resource,
keys: readonly string[],
): Resource => {
const rawIncoming = rawStoreValue(incoming as unknown as JsonRecord);
const rawExisting = rawStoreValue(existing as unknown as JsonRecord);
const next: JsonRecord = { ...rawExisting };
let platformDataLeaves: string[] | null = null;
for (const key of keys) {
if (key.startsWith(FAST_MERGE_PLATFORM_DATA_PREFIX)) {
(platformDataLeaves ??= []).push(key.slice(FAST_MERGE_PLATFORM_DATA_PREFIX.length));
continue;
}
// A merge-patch deletion removed the key from the raw row; the full merge's
// `...incoming` spread would keep the existing display value, so keep it.
if (!(key in rawIncoming)) continue;
const value = rawIncoming[key];
if (key === 'proxmox') {
const incomingFacet = asRecord(value);
if (!incomingFacet) continue;
const cloned = clonePlainValue(incomingFacet);
const existingFacet = asRecord(next.proxmox);
// Mirror mergeCanonicalSourceFacet: the facet merges with the existing
// one only while the (unchanged) source list keeps it authoritative.
next.proxmox =
existingFacet &&
shouldKeepSourceFacet(
getCanonicalSourceList(existing, existing.platformData),
'proxmox-pve',
)
? { ...existingFacet, ...cloned }
: cloned;
continue;
}
next[key] = cloneFastPatchValue(value);
}
if (platformDataLeaves) {
const incomingPlatformData = asRecord(rawIncoming.platformData);
const nextPlatformData: JsonRecord = { ...(asRecord(next.platformData) ?? {}) };
for (const leaf of platformDataLeaves) {
if (!incomingPlatformData || !(leaf in incomingPlatformData)) continue;
nextPlatformData[leaf] = cloneFastPatchValue(incomingPlatformData[leaf]);
}
next.platformData = nextPlatformData;
}
return next as unknown as Resource;
};
export type FastResourceStorePatchOp = {
key: string;
// Set for platformData leaf writes; the value then targets platformData[leaf].
leaf?: string;
value: unknown;
// Records diff via a nested reconcile at the key path; primitives replace.
mode: 'set' | 'reconcile';
};
// Store-commit counterpart of the fast merge: instead of a full-row reconcile
// (whose unwrap deep-walks every subtree), a fast row commits as a handful of
// per-key writes. Callers apply `reconcile` ops with a subtree reconcile and
// `set` ops as plain path sets.
export const buildFastResourceStorePatchOps = (
row: Resource,
keys: readonly string[],
): FastResourceStorePatchOp[] => {
const record = rawStoreValue(row as unknown as JsonRecord);
const ops: FastResourceStorePatchOp[] = [];
const platformData = asRecord(record.platformData);
for (const key of keys) {
if (key.startsWith(FAST_MERGE_PLATFORM_DATA_PREFIX)) {
const leaf = key.slice(FAST_MERGE_PLATFORM_DATA_PREFIX.length);
if (!platformData || !(leaf in platformData)) continue;
const value = platformData[leaf];
ops.push({
key: 'platformData',
leaf,
value,
mode: value !== null && typeof value === 'object' ? 'reconcile' : 'set',
});
continue;
}
if (!(key in record)) continue;
const value = record[key];
ops.push({
key,
value,
mode: value !== null && typeof value === 'object' ? 'reconcile' : 'set',
});
}
return ops;
};
// Incremental counterpart to mergeCanonicalResourceSnapshot. Server delta
// application preserves the raw object identity of untouched resources, so
// only changed rows and the small host-coalescing set need to be cloned and
// canonicalized. Non-host resources outside the delta retain their exact display
// objects, preventing an estate-wide reactive invalidation on every metrics
// tick while keeping the full-snapshot compatibility semantics intact.
// tick while keeping the full-snapshot compatibility semantics intact. Rows
// whose per-key change shape passes getFastResourceMergePatchKeys skip the
// clone+canonicalize+merge entirely and patch the previous display row.
export const mergeCanonicalResourceDeltaSnapshot = (
incoming: Resource[],
existing: Resource[],
changedIds: ReadonlySet<string>,
changedKeys?: ResourceChangedKeys,
): Resource[] => {
if (incoming.length === 0) {
return [];
@@ -877,6 +1081,9 @@ export const mergeCanonicalResourceDeltaSnapshot = (
});
const SKIP = Symbol('skip');
// Fast-path outputs are already fully merged display rows; they must bypass
// the final mergeCanonicalResource pass.
const fastMergedRows = new Set<Resource>();
const prepared = incoming
.map((resource, index) => {
const hostKey = hostKeys[index];
@@ -895,6 +1102,14 @@ export const mergeCanonicalResourceDeltaSnapshot = (
if (!mustRefresh) {
return existingResource;
}
if (existingResource !== undefined) {
const fastKeys = getFastResourceMergePatchKeys(changedKeys, resource.id, existingResource);
if (fastKeys) {
const fastRow = applyFastResourceMergePatch(resource, existingResource, fastKeys);
fastMergedRows.add(fastRow);
return fastRow;
}
}
return canonicalizeRealtimeResource(structuredClone(resource), {
synthesizePlatformScopes: false,
});
@@ -907,6 +1122,9 @@ export const mergeCanonicalResourceDeltaSnapshot = (
if (resource === existingResource) {
return existingResource;
}
if (fastMergedRows.has(resource)) {
return resource;
}
return mergeCanonicalResource(resource, existingResource);
});
};