= (props) => {
aria-describedby={descriptionId}
onMouseMove={chart.handleMouseMove}
onMouseLeave={chart.handleMouseLeave}
+ onPointerDown={chart.handlePointerDown}
+ onPointerMove={chart.handlePointerMove}
+ onPointerUp={chart.handlePointerUp}
+ onPointerCancel={chart.handlePointerCancel}
/>
- {refreshStatus()} {accessibleDescription()} Use Left and Right arrow keys to inspect
- readings, Home and End for the first and last reading, and Escape to clear inspection.
+ {refreshStatus()} {accessibleDescription()} Tap the chart to inspect a reading, or use
+ Left and Right arrow keys, Home and End for the first and last reading, and Escape to
+ clear inspection.
{chart.keyboardInspecting() && chart.hoveredPoint()
diff --git a/frontend-modern/src/components/shared/__tests__/HistoryChart.test.tsx b/frontend-modern/src/components/shared/__tests__/HistoryChart.test.tsx
index d2eb0a84b9..8729c8a2ba 100644
--- a/frontend-modern/src/components/shared/__tests__/HistoryChart.test.tsx
+++ b/frontend-modern/src/components/shared/__tests__/HistoryChart.test.tsx
@@ -15,6 +15,63 @@ import {
HISTORY_CHART_RANGES,
} from '@/components/shared/historyChartModel';
+function touchPointer(canvas: Element, type: string, fields: Partial = {}) {
+ const event = Object.assign(new Event(type, { bubbles: true, cancelable: true }), {
+ pointerId: 1,
+ pointerType: 'touch',
+ isPrimary: true,
+ clientX: 60,
+ clientY: 60,
+ ...fields,
+ });
+ fireEvent(canvas, event);
+ // Inspection must never consume native page panning or pinch zoom.
+ expect(event.defaultPrevented).toBe(false);
+}
+
+function mountTouchHistory() {
+ const [target, setTarget] = createSignal('disk-a');
+ const [data, setData] = createSignal(
+ [10, 20, 30].map((value, index) => ({
+ timestamp: (index + 1) * 1000,
+ value,
+ min: value,
+ max: value,
+ })),
+ );
+ const view = render(() => (
+
+ ));
+ const canvas = view.container.querySelector('canvas')!;
+ vi.spyOn(canvas, 'getBoundingClientRect').mockReturnValue({
+ x: 0,
+ y: 0,
+ left: 0,
+ top: 0,
+ right: 400,
+ bottom: 200,
+ width: 400,
+ height: 200,
+ toJSON: () => ({}),
+ });
+ return {
+ ...view,
+ canvas,
+ setTarget,
+ data,
+ setData,
+ tooltip: () => view.container.querySelector('[data-history-chart-tooltip]'),
+ announcement: view.container.querySelector('[aria-live="polite"]')!,
+ };
+}
+
if (typeof globalThis.ResizeObserver === 'undefined') {
globalThis.ResizeObserver = class ResizeObserver {
observe() {}
@@ -181,6 +238,142 @@ describe('HistoryChart', () => {
expect(announcement.textContent).toBe('');
});
+ it('inspects the tapped sample without compatibility focus selecting the latest reading', () => {
+ const view = mountTouchHistory();
+ touchPointer(view.canvas, 'pointerdown');
+ touchPointer(view.canvas, 'pointerup');
+ fireEvent.focus(view.canvas);
+ expect(view.tooltip()).toHaveTextContent('10.0%');
+ expect(view.announcement.textContent).toBe('');
+ fireEvent.mouseMove(view.canvas, { clientX: 370 });
+ fireEvent.mouseLeave(view.canvas);
+ expect(view.tooltip()).toHaveTextContent('10.0%');
+ touchPointer(view.canvas, 'pointerdown', { clientX: 200 });
+ touchPointer(view.canvas, 'pointerup', { clientX: 200 });
+ expect(view.tooltip()).toHaveTextContent('20.0%');
+ fireEvent.keyDown(view.canvas, { key: 'ArrowRight' });
+ expect(view.announcement).toHaveTextContent('30.0%');
+ fireEvent.keyDown(view.canvas, { key: 'Escape' });
+ expect(view.tooltip()).toBeNull();
+ fireEvent.blur(view.canvas);
+ fireEvent.focus(view.canvas);
+ expect(view.announcement).toHaveTextContent('30.0%');
+ });
+
+ it.each(['moved', 'release-displaced', 'cancelled', 'second-finger', 'different-pointer'])(
+ 'does not turn a %s touch into inspection',
+ (gesture) => {
+ const view = mountTouchHistory();
+ touchPointer(view.canvas, 'pointerdown');
+ if (gesture === 'moved') {
+ touchPointer(view.canvas, 'pointermove', { clientY: 90 });
+ // Returning to the origin is still a drag, not a new tap.
+ touchPointer(view.canvas, 'pointermove');
+ } else if (gesture === 'cancelled') {
+ touchPointer(view.canvas, 'pointercancel');
+ } else if (gesture === 'second-finger') {
+ touchPointer(view.canvas, 'pointerdown', { pointerId: 2, isPrimary: false });
+ }
+ touchPointer(view.canvas, 'pointerup', {
+ ...(gesture === 'release-displaced' ? { clientY: 90 } : {}),
+ ...(gesture === 'different-pointer' ? { pointerId: 2 } : {}),
+ });
+ fireEvent.focus(view.canvas);
+ expect(view.tooltip()).toBeNull();
+ expect(view.announcement.textContent).toBe('');
+ },
+ );
+
+ it.each([{ clientX: 0 }, { clientX: 400 }, { clientY: -1 }, { clientY: 201 }])(
+ 'does not inspect a tap outside the plot: %j',
+ (position) => {
+ const view = mountTouchHistory();
+ touchPointer(view.canvas, 'pointerdown', position);
+ touchPointer(view.canvas, 'pointerup', position);
+ expect(view.tooltip()).toBeNull();
+ },
+ );
+
+ it('invalidates an in-flight touch when the resource changes', () => {
+ const view = mountTouchHistory();
+ touchPointer(view.canvas, 'pointerdown');
+ view.setTarget('disk-b');
+ touchPointer(view.canvas, 'pointerup');
+ fireEvent.focus(view.canvas);
+ expect(view.tooltip()).toBeNull();
+ expect(view.announcement.textContent).toBe('');
+ touchPointer(view.canvas, 'pointerdown', { clientX: 200 });
+ touchPointer(view.canvas, 'pointerup', { clientX: 200 });
+ expect(view.tooltip()).toHaveTextContent('20.0%');
+ });
+
+ it('keeps the tapped time on refresh and distinguishes a lone zero from empty history', () => {
+ const view = mountTouchHistory();
+ touchPointer(view.canvas, 'pointerdown');
+ touchPointer(view.canvas, 'pointerup');
+ view.setData(view.data().map((point) => ({ ...point, value: point.value + 1 })));
+ expect(view.tooltip()).toHaveTextContent('11.0%');
+ view.setData([{ timestamp: 1000, value: 0, min: 0, max: 0 }]);
+ touchPointer(view.canvas, 'pointerdown', { clientX: 200 });
+ touchPointer(view.canvas, 'pointerup', { clientX: 200 });
+ expect(view.tooltip()).toHaveTextContent('0.0%');
+ view.setData([]);
+ touchPointer(view.canvas, 'pointerdown');
+ touchPointer(view.canvas, 'pointerup');
+ fireEvent.focus(view.canvas);
+ expect(view.tooltip()).toBeNull();
+ expect(view.announcement.textContent).toBe('');
+ });
+
+ it('leaves mouse hover available after touch inspection', () => {
+ const view = mountTouchHistory();
+ touchPointer(view.canvas, 'pointerdown');
+ touchPointer(view.canvas, 'pointerup');
+ touchPointer(view.canvas, 'pointermove', { pointerType: 'mouse', clientX: 200 });
+ fireEvent.mouseMove(view.canvas, { clientX: 200 });
+ expect(view.tooltip()).toHaveTextContent('20.0%');
+ fireEvent.mouseLeave(view.canvas);
+ expect(view.tooltip()).toBeNull();
+ });
+
+ it('keeps a new group chart tap when the previously focused chart blurs', () => {
+ const points = [10, 20, 30].map((value, index) => ({
+ timestamp: (index + 1) * 1000,
+ value,
+ min: value,
+ max: value,
+ }));
+ const { container } = render(() => (
+
+
+
+
+ ));
+ const [previous, next] = container.querySelectorAll('canvas');
+ vi.spyOn(next, 'getBoundingClientRect').mockReturnValue({
+ x: 0,
+ y: 0,
+ left: 0,
+ top: 0,
+ right: 400,
+ bottom: 200,
+ width: 400,
+ height: 200,
+ toJSON: () => ({}),
+ });
+ fireEvent.focus(previous);
+ touchPointer(next, 'pointerdown');
+ touchPointer(next, 'pointerup');
+ fireEvent.blur(previous);
+ fireEvent.focus(next);
+ const tooltips = container.querySelectorAll('[data-history-chart-tooltip]');
+ expect(tooltips).toHaveLength(2);
+ for (const tooltip of tooltips) expect(tooltip).toHaveTextContent('10.0%');
+ for (const live of container.querySelectorAll('[aria-live="polite"]')) {
+ expect(live.textContent).toBe('');
+ }
+ });
+
it('renders the default history label', () => {
render(() => );
diff --git a/frontend-modern/src/components/shared/useHistoryChartState.ts b/frontend-modern/src/components/shared/useHistoryChartState.ts
index 93d36881d5..2ccfa69c54 100644
--- a/frontend-modern/src/components/shared/useHistoryChartState.ts
+++ b/frontend-modern/src/components/shared/useHistoryChartState.ts
@@ -48,6 +48,9 @@ export function useHistoryChartState(
const chartHeight = createMemo(() => props.height || 200);
let chartLeftInset = HISTORY_CHART_MIN_LEFT_INSET;
let chartRightInset = 0;
+ let touchFocus = false;
+ let touchTimestamp: number | null = null;
+ let pendingTouch: { pointerId: number; x: number; y: number; moved: boolean } | undefined;
const hoveredTimestamp = hoverGroup?.hoveredTimestamp ?? localHoveredTimestamp;
const setHoveredTimestamp = hoverGroup?.setHoveredTimestamp ?? setLocalHoveredTimestamp;
@@ -132,6 +135,8 @@ export function useHistoryChartState(
suppliedData !== undefined,
]);
if (selection !== previousSelection) {
+ pendingTouch = undefined;
+ touchTimestamp = null;
setHoveredPoint(null);
setHoveredTimestamp(null);
}
@@ -376,12 +381,21 @@ export function useHistoryChartState(
});
const handleFocus = () => {
+ // Touch's compatibility mouse events can focus the canvas after pointerup.
+ // That focus must not replace the tapped reading with the latest sample.
+ if (touchFocus) {
+ setHoveredTimestamp(touchTimestamp);
+ return;
+ }
setKeyboardInspecting(true);
const points = data();
setHoveredTimestamp(points.length ? points[points.length - 1].timestamp : null);
};
const handleBlur = () => {
+ touchFocus = false;
+ touchTimestamp = null;
+ pendingTouch = undefined;
setKeyboardInspecting(false);
setHoveredTimestamp(null);
};
@@ -389,12 +403,18 @@ export function useHistoryChartState(
const handleKeyDown = (event: KeyboardEvent) => {
if (event.altKey || event.ctrlKey || event.metaKey) return;
if (event.key === 'Escape') {
+ touchFocus = false;
+ touchTimestamp = null;
+ pendingTouch = undefined;
setHoveredTimestamp(null);
return;
}
const points = data();
if (!points.length || !['ArrowLeft', 'ArrowRight', 'Home', 'End'].includes(event.key)) return;
event.preventDefault();
+ touchFocus = false;
+ touchTimestamp = null;
+ pendingTouch = undefined;
setKeyboardInspecting(true);
const timestamp = hoveredTimestamp();
const index =
@@ -410,14 +430,14 @@ export function useHistoryChartState(
setHoveredTimestamp(points[next].timestamp);
};
- const handleMouseMove = (event: MouseEvent) => {
+ const inspectAtClientX = (clientX: number, snapToSample = false) => {
const canvas = refs.getCanvas();
const points = data();
if (!canvas || points.length === 0) return;
setKeyboardInspecting(false);
const rect = canvas.getBoundingClientRect();
- const x = event.clientX - rect.left;
+ const x = clientX - rect.left;
const width = rect.width;
if (x < chartLeftInset || x > width - chartRightInset) {
setHoveredTimestamp(null);
@@ -434,11 +454,58 @@ export function useHistoryChartState(
leftInset: chartLeftInset,
rightInset: chartRightInset,
});
- setHoveredTimestamp(geometry.getTimestamp(x));
+ const timestamp = geometry.getTimestamp(x);
+ const inspectedTimestamp = snapToSample
+ ? findHistoryChartClosestPoint(points, timestamp).timestamp
+ : timestamp;
+ setHoveredTimestamp(inspectedTimestamp);
+ return inspectedTimestamp;
+ };
+
+ const handleMouseMove = (event: MouseEvent) => {
+ if (!touchFocus) inspectAtClientX(event.clientX);
+ };
+
+ const handlePointerDown = (event: PointerEvent) => {
+ if (event.pointerType === 'mouse') touchFocus = false;
+ if (event.pointerType !== 'touch') return;
+ touchFocus = true;
+ touchTimestamp = null;
+ setKeyboardInspecting(false);
+ setHoveredTimestamp(null);
+ pendingTouch = event.isPrimary
+ ? { pointerId: event.pointerId, x: event.clientX, y: event.clientY, moved: false }
+ : undefined;
+ };
+
+ const handlePointerMove = (event: PointerEvent) => {
+ if (event.pointerType === 'mouse') touchFocus = false;
+ if (event.pointerType !== 'touch' || pendingTouch?.pointerId !== event.pointerId) return;
+ if (Math.hypot(event.clientX - pendingTouch.x, event.clientY - pendingTouch.y) > 10) {
+ pendingTouch.moved = true;
+ }
+ };
+
+ const handlePointerUp = (event: PointerEvent) => {
+ if (event.pointerType !== 'touch') return;
+ const touch = pendingTouch;
+ pendingTouch = undefined;
+ if (!touch || !event.isPrimary || touch.pointerId !== event.pointerId || touch.moved) return;
+ if (Math.hypot(event.clientX - touch.x, event.clientY - touch.y) > 10) return;
+ const rect = refs.getCanvas()?.getBoundingClientRect();
+ if (!rect || event.clientY < rect.top || event.clientY > rect.bottom) return;
+ touchTimestamp = inspectAtClientX(event.clientX, true) ?? null;
+ };
+
+ const handlePointerCancel = (event: PointerEvent) => {
+ if (event.pointerType !== 'touch') return;
+ pendingTouch = undefined;
+ touchTimestamp = null;
+ setHoveredTimestamp(null);
};
const handleMouseLeave = () => {
- if (keyboardInspecting()) return;
+ if (keyboardInspecting() || touchFocus) return;
setHoveredTimestamp(null);
};
@@ -454,6 +521,10 @@ export function useHistoryChartState(
keyboardInspecting,
handleMouseLeave,
handleMouseMove,
+ handlePointerDown,
+ handlePointerMove,
+ handlePointerUp,
+ handlePointerCancel,
chartHeight,
chartWidth,
hoveredPoint,
diff --git a/frontend-modern/src/features/storageBackups/__tests__/storagePoolDetailPresentation.branchcov.test.ts b/frontend-modern/src/features/storageBackups/__tests__/storagePoolDetailPresentation.branchcov.test.ts
index ce66441060..5950ed1eef 100644
--- a/frontend-modern/src/features/storageBackups/__tests__/storagePoolDetailPresentation.branchcov.test.ts
+++ b/frontend-modern/src/features/storageBackups/__tests__/storagePoolDetailPresentation.branchcov.test.ts
@@ -453,16 +453,18 @@ describe('getStoragePoolLinkedDisks branch coverage', () => {
id: 'cache-pool-1',
name: ' My Cache ',
source: unraidSource(),
- details: { type: 'unraid-cache-pool', platform: 'unraid' },
+ details: { type: 'unraid-cache-pool', platform: 'unraid', parentId: 'host-1' },
});
const disks = [
{
id: 'in',
+ parentId: 'host-1',
name: 'in',
physicalDisk: { devPath: '/dev/sda', storageRole: 'cache', storageGroup: 'My Cache' },
},
{
id: 'out',
+ parentId: 'host-1',
name: 'out',
physicalDisk: { devPath: '/dev/sdb', storageRole: 'cache', storageGroup: 'other' },
},
@@ -476,16 +478,18 @@ describe('getStoragePoolLinkedDisks branch coverage', () => {
id: 'arr-1',
name: 'Tower',
source: unraidSource(),
- details: { type: 'pool', platform: 'unraid', topology: 'array' },
+ details: { type: 'pool', platform: 'unraid', topology: 'array', parentId: 'host-1' },
});
const disks = [
{
id: 'in',
+ parentId: 'host-1',
name: 'in',
physicalDisk: { devPath: '/dev/sda', storageRole: 'data', storageGroup: 'unraid-array' },
},
{
id: 'out',
+ parentId: 'host-1',
name: 'out',
physicalDisk: { devPath: '/dev/sdb', storageRole: 'data', storageGroup: 'nope' },
},
diff --git a/frontend-modern/src/features/storageBackups/__tests__/storagePoolDetailPresentation.test.ts b/frontend-modern/src/features/storageBackups/__tests__/storagePoolDetailPresentation.test.ts
index 47f5645cfd..dc34f25a70 100644
--- a/frontend-modern/src/features/storageBackups/__tests__/storagePoolDetailPresentation.test.ts
+++ b/frontend-modern/src/features/storageBackups/__tests__/storagePoolDetailPresentation.test.ts
@@ -30,6 +30,7 @@ const buildRecord = (overrides: Partial = {}): StorageRecord => (
observedAt: Date.now(),
details: {
node: 'truenas01',
+ parentId: 'host-truenas01',
type: 'pool',
zfsPool: {
state: 'ONLINE',
@@ -66,6 +67,7 @@ const buildRecord = (overrides: Partial = {}): StorageRecord => (
const buildDisk = (): Resource =>
({
id: 'disk-1',
+ parentId: 'host-truenas01',
type: 'physical_disk',
name: 'disk-1',
displayName: 'disk-1',
@@ -249,6 +251,7 @@ describe('storagePoolDetailPresentation', () => {
},
details: {
type: 'unraid-array',
+ parentId: 'host-tower',
platform: 'unraid',
topology: 'array',
arrayState: 'STARTED',
@@ -257,6 +260,7 @@ describe('storagePoolDetailPresentation', () => {
const disks = [
{
id: 'parity',
+ parentId: 'host-tower',
type: 'physical_disk',
name: 'Parity',
displayName: 'Parity',
@@ -277,6 +281,7 @@ describe('storagePoolDetailPresentation', () => {
},
{
id: 'disk1',
+ parentId: 'host-tower',
type: 'physical_disk',
name: 'Disk 1',
displayName: 'Disk 1',
diff --git a/frontend-modern/src/features/storageBackups/__tests__/storagePoolDiskOwnership.test.ts b/frontend-modern/src/features/storageBackups/__tests__/storagePoolDiskOwnership.test.ts
new file mode 100644
index 0000000000..6ae92dafa8
--- /dev/null
+++ b/frontend-modern/src/features/storageBackups/__tests__/storagePoolDiskOwnership.test.ts
@@ -0,0 +1,170 @@
+import { describe, expect, it } from 'vitest';
+import type { State } from '@/types/api';
+import type { Resource } from '@/types/resource';
+import { buildStorageRecords } from '../storageAdapters';
+import { getStoragePoolLinkedDisks } from '../storagePoolDetailPresentation';
+
+const storage = (overrides: Partial = {}): Resource =>
+ ({
+ id: 'pool-a',
+ type: 'storage',
+ name: 'tank',
+ displayName: 'tank',
+ platformId: 'provider-a',
+ platformType: 'proxmox-pve',
+ sourceType: 'api',
+ status: 'online',
+ lastSeen: 1,
+ parentId: 'host-a',
+ parentName: 'Same display name',
+ storage: {
+ type: 'zfspool',
+ isZfs: true,
+ zfsPool: {
+ name: 'tank',
+ state: 'ONLINE',
+ readErrors: 0,
+ writeErrors: 0,
+ checksumErrors: 0,
+ devices: [{ name: 'sda', type: 'disk', state: 'ONLINE' }],
+ },
+ },
+ proxmox: { nodeName: 'pve1', instance: 'cluster-a' },
+ ...overrides,
+ }) as Resource;
+
+const disk = (id: string, overrides: Partial = {}): Resource =>
+ ({
+ id,
+ type: 'physical_disk',
+ name: id,
+ displayName: id,
+ platformId: 'provider-a',
+ platformType: 'proxmox-pve',
+ sourceType: 'api',
+ status: 'online',
+ lastSeen: 1,
+ parentId: 'host-a',
+ parentName: 'Same display name',
+ proxmox: { nodeName: 'pve1', instance: 'cluster-a' },
+ physicalDisk: { devPath: '/dev/sda', model: id, temperature: 35 },
+ ...overrides,
+ }) as Resource;
+
+const record = (resource = storage()) =>
+ buildStorageRecords({ state: {} as State, resources: [resource] })[0];
+const linked = (resource: Resource, disks: Resource[]) =>
+ getStoragePoolLinkedDisks(record(resource), disks).map((item) => item.id);
+
+describe('pool physical disk ownership', () => {
+ it('excludes other hosts and unknown owners despite identical paths and labels', () => {
+ expect(
+ linked(storage(), [
+ disk('local'),
+ disk('remote', { parentId: 'host-b' }),
+ disk('unknown', { parentId: undefined, proxmox: undefined }),
+ ]),
+ ).toEqual(['local']);
+ });
+
+ it('never lets matching provider names override mismatched canonical parents', () => {
+ expect(linked(storage(), [disk('remote', { parentId: 'host-b' })])).toEqual([]);
+ });
+
+ it('preserves direct canonical pool and legacy resource-reference children', () => {
+ const target = record();
+ target.refs = { resourceId: 'legacy-pool' };
+ expect(
+ getStoragePoolLinkedDisks(target, [
+ disk('direct', { parentId: 'pool-a', physicalDisk: { devPath: '/dev/other' } }),
+ disk('legacy', { parentId: 'legacy-pool', physicalDisk: { devPath: '/dev/legacy' } }),
+ ]).map((item) => item.id),
+ ).toEqual(['legacy', 'direct']);
+ });
+
+ it.each(['top-level', 'legacy'] as const)(
+ 'retains exact native instance/node ownership: %s',
+ (shape) => {
+ const pool = storage({ parentId: undefined, proxmox: undefined, platformData: undefined });
+ const local = disk('local', { parentId: undefined, proxmox: undefined });
+ const native = { nodeName: 'pve1', instance: 'cluster-a' };
+ if (shape === 'top-level') {
+ pool.proxmox = native;
+ local.proxmox = native;
+ } else {
+ pool.platformData = { proxmox: native };
+ local.platformData = { proxmox: native };
+ }
+ expect(record(pool).details?.instance).toBe('cluster-a');
+ expect(
+ linked(pool, [
+ local,
+ disk('other-cluster', {
+ parentId: undefined,
+ proxmox: { ...native, instance: 'cluster-b' },
+ }),
+ disk('other-node', { parentId: undefined, proxmox: { ...native, nodeName: 'pve2' } }),
+ disk('unscoped', { parentId: undefined, proxmox: { nodeName: 'pve1' } }),
+ ]),
+ ).toEqual(['local']);
+ },
+ );
+
+ it('requires both native instance and node when canonical ownership is missing', () => {
+ expect(
+ linked(storage({ parentId: undefined, proxmox: { nodeName: 'pve1' } }), [
+ disk('unscoped', { parentId: undefined, proxmox: { nodeName: 'pve1' } }),
+ ]),
+ ).toEqual([]);
+ });
+
+ it.each(['sda', '/dev/sda'])('matches whole device identifiers, not a suffix: %s', (name) => {
+ const pool = storage();
+ pool.storage!.zfsPool!.devices = [
+ { name, type: 'disk', state: 'ONLINE', readErrors: 0, writeErrors: 0, checksumErrors: 0 },
+ ];
+ expect(
+ linked(pool, [
+ disk('local'),
+ disk('suffix-collision', { physicalDisk: { devPath: '/dev/not-sda' } }),
+ disk('partition', { physicalDisk: { devPath: '/dev/sda1' } }),
+ ]),
+ ).toEqual(['local']);
+ });
+
+ it('does not confuse nvme device numbers with a shorter vdev name', () => {
+ const pool = storage();
+ pool.storage!.zfsPool!.devices = [
+ {
+ name: 'n1',
+ type: 'disk',
+ state: 'ONLINE',
+ readErrors: 0,
+ writeErrors: 0,
+ checksumErrors: 0,
+ },
+ ];
+ expect(linked(pool, [disk('nvme', { physicalDisk: { devPath: '/dev/nvme0n1' } })])).toEqual([]);
+ });
+
+ it.each([
+ ['unraid-array', 'Array', 'unraid-array'],
+ ['unraid-cache-pool', 'cache', 'cache'],
+ ])('keeps repeated UnRAID groups on their own host: %s', (type, name, group) => {
+ const pool = storage({
+ name,
+ platformType: 'unraid',
+ proxmox: undefined,
+ storage: { type, platform: 'unraid' },
+ });
+ const member = { devPath: '/dev/sda', storageGroup: group, storageRole: 'data' };
+ expect(
+ linked(pool, [
+ disk('local', { physicalDisk: member }),
+ disk('remote', { parentId: 'host-b', physicalDisk: member }),
+ disk('unknown', { parentId: undefined, proxmox: undefined, physicalDisk: member }),
+ disk('different-group', { physicalDisk: { ...member, storageGroup: 'other' } }),
+ ]),
+ ).toEqual(['local']);
+ });
+});
diff --git a/frontend-modern/src/features/storageBackups/storageAdapters.ts b/frontend-modern/src/features/storageBackups/storageAdapters.ts
index efa82a594d..19b8ac6598 100644
--- a/frontend-modern/src/features/storageBackups/storageAdapters.ts
+++ b/frontend-modern/src/features/storageBackups/storageAdapters.ts
@@ -224,6 +224,10 @@ const mapResourceStorageRecord = (resource: Resource, adapterId: string): Storag
parentId: resource.parentId,
parentName: resource.parentName,
node: proxmoxNativeNode || (platformData.node as string | undefined) || storageNodes[0],
+ instance:
+ resource.proxmox?.instance ||
+ (proxmoxPlatform?.instance as string | undefined) ||
+ (platformData.instance as string | undefined),
nodeHints,
hostLabel,
platformLabel,
diff --git a/frontend-modern/src/features/storageBackups/storagePoolDetailPresentation.ts b/frontend-modern/src/features/storageBackups/storagePoolDetailPresentation.ts
index 1fc455ff90..166571e68f 100644
--- a/frontend-modern/src/features/storageBackups/storagePoolDetailPresentation.ts
+++ b/frontend-modern/src/features/storageBackups/storagePoolDetailPresentation.ts
@@ -368,6 +368,28 @@ const diskRoleRank = (role: string): number => {
}
};
+const diskBelongsToStorageHost = (record: StorageRecord, disk: Resource): boolean => {
+ const hostId = readRecordDetailString(record, 'parentId');
+ const diskParentId = disk.parentId?.trim();
+ // Canonical ownership outranks labels, paths and source-native names.
+ // In particular, two hosts may both be called pve1 and both report /dev/sda.
+ if (hostId && diskParentId) return hostId === diskParentId;
+
+ // Older Proxmox snapshots may lack canonical parents. Only the complete
+ // native instance/node pair is a scope; a hostname alone is not one.
+ const platformData = (disk.platformData || {}) as Record;
+ const proxmox = platformData.proxmox as Resource['proxmox'] | undefined;
+ const node = readRecordDetailString(record, 'node').toLowerCase();
+ const instance = readRecordDetailString(record, 'instance');
+ const diskNode = (disk.proxmox?.nodeName || disk.proxmox?.node || proxmox?.nodeName || '')
+ .trim()
+ .toLowerCase();
+ const diskInstance = (disk.proxmox?.instance || proxmox?.instance || '').trim();
+ return Boolean(node && instance && node === diskNode && instance === diskInstance);
+};
+
+const normalizeDevicePath = (value: string): string => value.trim().replace(/^\/dev\//, '');
+
export function getStoragePoolLinkedDisks(
record: StorageRecord,
physicalDisks: Resource[],
@@ -382,9 +404,14 @@ export function getStoragePoolLinkedDisks(
.filter((disk) => {
const devPath = readDiskDevPath(disk);
const group = readDiskStorageGroup(disk).toLowerCase();
- if (directParentIds.has(disk.parentId || '')) return true;
+ if (directParentIds.has(disk.parentId?.trim() || '')) return true;
+ if (!diskBelongsToStorageHost(record, disk)) return false;
if (pool?.devices?.length && devPath) {
- return pool.devices.some((device) => devPath.endsWith(device.name));
+ return pool.devices.some(
+ (device) =>
+ Boolean(device.name.trim()) &&
+ normalizeDevicePath(devPath) === normalizeDevicePath(device.name),
+ );
}
return Boolean(unraidStorageGroup && group === unraidStorageGroup);
})
diff --git a/internal/monitoring/truenas_poller_test.go b/internal/monitoring/truenas_poller_test.go
index 9074c9cdd9..fcf3a34c68 100644
--- a/internal/monitoring/truenas_poller_test.go
+++ b/internal/monitoring/truenas_poller_test.go
@@ -2015,6 +2015,133 @@ func TestTrueNASPollerUnresponsiveRPCDoesNotFreezeOtherConnections(t *testing.T)
}
}
+// #2396's nonempty Apps path: an accepted subscription is then rejected by
+// middlewared. Keep the inventory, advance the poll ledger, observe native
+// alert disappearance and recover stats on the next ordinary poll. This is a
+// protocol/poller control, not a native appliance or incident-delivery proof.
+func TestTrueNASPollerSubscriptionRejectionKeepsPollingAndRecovers(t *testing.T) {
+ var reject atomic.Bool
+ var sessions, statsSubscriptions atomic.Int32
+ upgrader := websocket.Upgrader{}
+ server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path != "/api/current" {
+ t.Error("modern polling fell back to REST")
+ http.NotFound(w, r)
+ return
+ }
+ conn, err := upgrader.Upgrade(w, r, nil)
+ if err != nil {
+ return
+ }
+ defer conn.Close()
+ sessions.Add(1)
+ for {
+ var request struct {
+ ID int64 `json:"id"`
+ Method string `json:"method"`
+ Params []json.RawMessage `json:"params"`
+ }
+ if conn.ReadJSON(&request) != nil {
+ return
+ }
+ var result any = []any{}
+ var collection string
+ switch request.Method {
+ case "auth.login_ex":
+ result = map[string]any{"response_type": "SUCCESS"}
+ case "system.info":
+ result = map[string]any{"hostname": "reject-nas", "version": "TrueNAS-SCALE-25.04.2.6", "system_serial": "REJECT-FIXTURE"}
+ case "pool.query":
+ result = []map[string]any{{"id": 1, "name": "tank", "status": "ONLINE", "size": 1000, "allocated": 400}}
+ case "app.query":
+ result = []map[string]any{{"id": "fixture-app", "name": "fixture-app", "state": "RUNNING"}}
+ case "alert.list":
+ if !reject.Load() {
+ result = []map[string]any{{"id": "fixture-alert", "level": "WARNING", "formatted": "fixture warning", "source": "fixture"}}
+ }
+ case "core.subscribe":
+ if len(request.Params) != 1 || json.Unmarshal(request.Params[0], &collection) != nil {
+ t.Error("invalid subscription")
+ return
+ }
+ result = "fixture-sub"
+ case "core.unsubscribe":
+ result = nil
+ }
+ if conn.WriteJSON(map[string]any{"jsonrpc": "2.0", "id": request.ID, "result": result}) != nil {
+ return
+ }
+ if collection == "" {
+ continue
+ }
+ var params any
+ method := "collection_update"
+ if strings.HasPrefix(collection, "app.stats:") {
+ statsSubscriptions.Add(1)
+ if reject.Load() {
+ method = "notify_unsubscribed"
+ params = map[string]any{"collection": collection, "error": map[string]any{"error": 14, "errname": "EFAULT", "reason": "[EFAULT] Apps are not available", "trace": nil, "extra": nil}}
+ } else {
+ params = map[string]any{"collection": collection, "fields": []any{map[string]any{"app_name": "fixture-app", "cpu_usage": 12}}}
+ }
+ } else {
+ params = map[string]any{"collection": collection, "fields": map[string]any{"cpu": map[string]any{"usage": 10}}}
+ }
+ if conn.WriteJSON(map[string]any{"jsonrpc": "2.0", "method": method, "params": params}) != nil {
+ return
+ }
+ }
+ }))
+ t.Cleanup(server.Close)
+ client, err := truenas.NewClient(truenas.ClientConfig{
+ Host: server.URL, APIKey: "fixture-key", Username: "fixture-user", Timeout: 2 * time.Second,
+ InsecureSkipVerify: true, Fingerprint: fmt.Sprintf("%x", sha256.Sum256(server.Certificate().Raw)),
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(client.Close)
+ instance := config.TrueNASInstance{ID: "reject-connection", Host: server.URL, Enabled: true}
+ provider := truenas.NewLiveProviderForConnection(&truenas.APIFetcher{Client: client}, instance.ID)
+ poller := NewTrueNASPoller(nil, 0, nil)
+ poller.providersByOrg["default"] = map[string]*truenas.Provider{instance.ID: provider}
+ poller.configsByOrg["default"] = map[string]config.TrueNASInstance{instance.ID: instance}
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ var lastSuccess time.Time
+ for poll := 0; poll < 3; poll++ {
+ reject.Store(poll == 1)
+ poller.ensureConnectionRuntimeStatusLocked("default", instance.ID).nextPollAt = time.Now().Add(-time.Second)
+ started := time.Now()
+ poller.pollAll(ctx)
+ if elapsed := time.Since(started); elapsed >= time.Second {
+ t.Fatalf("rejection stalled poll %d for %s", poll, elapsed)
+ }
+ summary := poller.ConnectionSummaries("default", []config.TrueNASInstance{instance})[instance.ID]
+ if summary.Poll.LastError != nil || summary.Poll.LastSuccessAt == nil || !summary.Poll.LastSuccessAt.After(lastSuccess) || summary.Poll.LastAttemptAt == nil {
+ t.Fatalf("optional stats rejection prevented poll progress: %+v", summary)
+ }
+ lastSuccess = *summary.Poll.LastSuccessAt
+ snapshot := provider.Snapshot()
+ if snapshot == nil || snapshot.System.Hostname != "reject-nas" || len(snapshot.Pools) != 1 || len(snapshot.Apps) != 1 || snapshot.System.CPUPercent != 10 {
+ t.Fatalf("poll %d lost usable inventory/telemetry: %+v", poll, snapshot)
+ }
+ if poll == 1 {
+ if snapshot.Apps[0].Stats != nil || len(snapshot.Alerts) != 0 {
+ t.Fatal("rejected stats fabricated data or left the former native alert")
+ }
+ } else if snapshot.Apps[0].Stats == nil || snapshot.Apps[0].Stats.CPUPercent != 12 || len(snapshot.Alerts) != 1 {
+ t.Fatal("healthy stats/native alerts did not return")
+ }
+ if !hasTrueNASHostForOrg(poller, "default", "reject-nas") {
+ t.Fatal("poll lost the appliance identity")
+ }
+ }
+ if statsSubscriptions.Load() != 3 || sessions.Load() != 2 {
+ t.Fatalf("rejection replayed or churned later polls: subscriptions=%d sessions=%d", statsSubscriptions.Load(), sessions.Load())
+ }
+}
+
type trueNASMockServer struct {
server *httptest.Server
requests atomic.Int64
diff --git a/internal/truenas/client.go b/internal/truenas/client.go
index 6ad3e11046..b057e40d18 100644
--- a/internal/truenas/client.go
+++ b/internal/truenas/client.go
@@ -258,9 +258,9 @@ func (c *Client) GetSystemTelemetry(ctx context.Context) (*SystemInfo, error) {
if err != nil {
return err
}
- telemetry, err = rpc.readSystemTelemetryEvent(ctx, defaultRealtimeIntervalSeconds)
+ telemetry, err = rpc.readSystemTelemetryEvent(ctx, defaultRealtimeIntervalSeconds, subscriptionName)
if err != nil {
- return discardRPCSessionForStreamError(err)
+ return discardRPCSessionForStreamError("reporting.realtime", err)
}
return rpc.unsubscribe(ctx, subscriptionID)
})
@@ -1743,12 +1743,18 @@ func (c *Client) getAppsREST(ctx context.Context) ([]App, error) {
}
func (c *Client) parseAppsWithStats(ctx context.Context, response []map[string]any) []App {
+ apps := make([]App, 0, len(response))
+ // There is nothing to enrich when app.query returns an empty inventory.
+ // Subscribing anyway can wait on a stopped Apps service and needlessly
+ // discard an otherwise healthy, persistent monitoring session.
+ if len(response) == 0 {
+ return apps
+ }
statsByApp, err := c.GetAppStats(ctx)
if err != nil {
statsByApp = nil
}
- apps := make([]App, 0, len(response))
for _, item := range response {
activeWorkloads := readMapAny(item, "active_workloads", "activeWorkloads")
@@ -1811,9 +1817,9 @@ func (c *Client) GetAppStats(ctx context.Context) (map[string]AppStats, error) {
if err != nil {
return err
}
- stats, err = rpc.readAppStatsEvent(ctx, defaultAppStatsIntervalSeconds)
+ stats, err = rpc.readAppStatsEvent(ctx, defaultAppStatsIntervalSeconds, subscriptionName)
if err != nil {
- return discardRPCSessionForStreamError(err)
+ return discardRPCSessionForStreamError("app.stats", err)
}
return rpc.unsubscribe(ctx, subscriptionID)
})
@@ -1867,9 +1873,12 @@ func (c *Client) GetAppLogs(ctx context.Context, appName, containerID string, ta
return err
}
var reusable bool
- lines, reusable, err = rpc.readAppLogEvents(ctx, tailLines)
+ lines, reusable, err = rpc.readAppLogEvents(ctx, tailLines, subscriptionName)
+ if errors.Is(err, errRPCSubscriptionComplete) {
+ return nil
+ }
if err != nil {
- return discardRPCSessionForStreamError(err)
+ return discardRPCSessionForStreamError("app.container_log_follow", err)
}
if !reusable {
return errRPCStreamSessionConsumed
@@ -2335,6 +2344,8 @@ func appendDiskTemperature(out map[string]int, diskName string, value any) {
type trueNASRPCClient struct {
conn *websocket.Conn
nextID int64
+ openedAt time.Time
+ closeOnce sync.Once
keepaliveStop chan struct{}
keepaliveDone chan struct{}
}
@@ -2346,7 +2357,7 @@ func (c *trueNASRPCClient) subscribe(ctx context.Context, event string) (string,
}
subscriptionID = strings.TrimSpace(subscriptionID)
if subscriptionID == "" {
- return "", &discardRPCSessionError{err: fmt.Errorf("truenas rpc core.subscribe returned an empty subscription id")}
+ return "", &discardRPCSessionError{method: "core.subscribe", err: fmt.Errorf("truenas rpc core.subscribe returned an empty subscription id")}
}
return subscriptionID, nil
}
@@ -2357,7 +2368,7 @@ func (c *trueNASRPCClient) unsubscribe(ctx context.Context, subscriptionID strin
return fmt.Errorf("truenas rpc subscription id is required")
}
if err := c.call(ctx, "core.unsubscribe", []any{subscriptionID}, nil); err != nil {
- return &discardRPCSessionError{err: fmt.Errorf("unsubscribe %q: %w", subscriptionID, err)}
+ return &discardRPCSessionError{method: "core.unsubscribe", err: fmt.Errorf("unsubscribe %q: %w", subscriptionID, err)}
}
return nil
}
@@ -2648,6 +2659,21 @@ func (c *trueNASRPCClient) call(ctx context.Context, method string, params any,
return &RPCTransportError{Method: method, Phase: "read", Err: err}
}
if message.Method != "" {
+ // A rejection can race with the subscription acknowledgement.
+ // Do not drop it while waiting for the core.subscribe response.
+ if method == "core.subscribe" {
+ if args, ok := params.([]any); ok && len(args) == 1 {
+ if collection, ok := args[0].(string); ok {
+ ended, err := rpcSubscriptionTermination(message, collection, method)
+ if ended || err != nil {
+ if err == nil {
+ err = rpcSubscriptionEndedBeforeData(method)
+ }
+ return discardRPCSessionForStreamError(method, err)
+ }
+ }
+ }
+ }
continue
}
if message.ID != request.ID {
@@ -2666,7 +2692,7 @@ func (c *trueNASRPCClient) call(ctx context.Context, method string, params any,
}
}
-func (c *trueNASRPCClient) readAppStatsEvent(ctx context.Context, intervalSeconds int) (map[string]AppStats, error) {
+func (c *trueNASRPCClient) readAppStatsEvent(ctx context.Context, intervalSeconds int, collection string) (map[string]AppStats, error) {
if c == nil || c.conn == nil {
return nil, fmt.Errorf("truenas rpc connection is nil")
}
@@ -2682,6 +2708,12 @@ func (c *trueNASRPCClient) readAppStatsEvent(ctx context.Context, intervalSecond
if err := c.conn.ReadJSON(&message); err != nil {
return nil, &RPCTransportError{Method: "app.stats", Phase: "read", Err: err}
}
+ if ended, err := rpcSubscriptionTermination(message, collection, "app.stats"); ended || err != nil {
+ if err == nil {
+ err = rpcSubscriptionEndedBeforeData("app.stats")
+ }
+ return nil, err
+ }
if message.Method == "" {
if message.Error != nil {
return nil, rpcErrorFromWire("app.stats", message.Error)
@@ -2734,7 +2766,7 @@ func (c *trueNASRPCClient) readAppStatsEvent(ctx context.Context, intervalSecond
}
}
-func (c *trueNASRPCClient) readAppLogEvents(ctx context.Context, tailLines int) ([]AppLogLine, bool, error) {
+func (c *trueNASRPCClient) readAppLogEvents(ctx context.Context, tailLines int, collection string) ([]AppLogLine, bool, error) {
if c == nil || c.conn == nil {
return nil, false, fmt.Errorf("truenas rpc connection is nil")
}
@@ -2780,6 +2812,12 @@ func (c *trueNASRPCClient) readAppLogEvents(ctx context.Context, tailLines int)
}
return nil, false, &RPCTransportError{Method: "app.container_log_follow", Phase: "read", Err: err}
}
+ if ended, err := rpcSubscriptionTermination(message, collection, "app.container_log_follow"); ended || err != nil {
+ if err != nil {
+ return nil, false, err
+ }
+ return trimAppLogLines(lines, tailLines), true, errRPCSubscriptionComplete
+ }
if message.Method == "" {
if message.Error != nil {
return nil, false, rpcErrorFromWire("app.container_log_follow", message.Error)
@@ -3088,7 +3126,7 @@ func (c *trueNASRPCClient) getReportingDataWithQuery(ctx context.Context, graphs
return response, nil
}
-func (c *trueNASRPCClient) readSystemTelemetryEvent(ctx context.Context, intervalSeconds int) (*SystemInfo, error) {
+func (c *trueNASRPCClient) readSystemTelemetryEvent(ctx context.Context, intervalSeconds int, collection string) (*SystemInfo, error) {
if c == nil || c.conn == nil {
return nil, fmt.Errorf("truenas rpc connection is nil")
}
@@ -3104,6 +3142,12 @@ func (c *trueNASRPCClient) readSystemTelemetryEvent(ctx context.Context, interva
if err := c.conn.ReadJSON(&message); err != nil {
return nil, &RPCTransportError{Method: "reporting.realtime", Phase: "read", Err: err}
}
+ if ended, err := rpcSubscriptionTermination(message, collection, "reporting.realtime"); ended || err != nil {
+ if err == nil {
+ err = rpcSubscriptionEndedBeforeData("reporting.realtime")
+ }
+ return nil, err
+ }
if message.Method == "" {
if message.Error != nil {
return nil, rpcErrorFromWire("reporting.realtime", message.Error)
diff --git a/internal/truenas/subscription.go b/internal/truenas/subscription.go
new file mode 100644
index 0000000000..cd60b1864d
--- /dev/null
+++ b/internal/truenas/subscription.go
@@ -0,0 +1,50 @@
+package truenas
+
+import (
+ "encoding/json"
+ "errors"
+ "fmt"
+)
+
+// A clean terminal event completes a log tail without leaving a subscription
+// to cancel. It is not a sample for a telemetry stream.
+var errRPCSubscriptionComplete = errors.New("truenas rpc subscription completed")
+
+// rpcSubscriptionTermination recognises only the exact collection we subscribed
+// to, including its arguments. Other subscriptions (even another interval for
+// the same event source) must not terminate the current reader. middlewared's
+// notification error is not a JSON-RPC response error: it uses errno/reason
+// fields, and may contain private names and traces. Retain only its numeric
+// errno and a fixed message, never that wire text or the collection arguments.
+func rpcSubscriptionTermination(message trueNASRPCResponse, collection, method string) (bool, error) {
+ if message.Method != "notify_unsubscribed" {
+ return false, nil
+ }
+ var notification struct {
+ Collection string `json:"collection"`
+ Error json.RawMessage `json:"error"`
+ }
+ if err := json.Unmarshal(message.Params, ¬ification); err != nil {
+ return false, fmt.Errorf("invalid truenas %s termination notification", method)
+ }
+ if notification.Collection != collection {
+ return false, nil
+ }
+ if len(notification.Error) == 0 {
+ return true, fmt.Errorf("invalid truenas %s termination notification", method)
+ }
+ if string(notification.Error) == "null" {
+ return true, nil
+ }
+ var failure struct {
+ Errno int `json:"error"`
+ }
+ if err := json.Unmarshal(notification.Error, &failure); err != nil {
+ return true, fmt.Errorf("invalid truenas %s termination error", method)
+ }
+ return true, &RPCError{Method: method, Code: failure.Errno, Message: "subscription rejected"}
+}
+
+func rpcSubscriptionEndedBeforeData(method string) error {
+ return &RPCError{Method: method, Message: "subscription ended before a sample"}
+}
diff --git a/internal/truenas/transport.go b/internal/truenas/transport.go
index 16f1093e0f..bee97cd715 100644
--- a/internal/truenas/transport.go
+++ b/internal/truenas/transport.go
@@ -12,6 +12,7 @@ import (
"time"
"github.com/gorilla/websocket"
+ "github.com/rs/zerolog/log"
)
var errRPCStreamSessionConsumed = errors.New("truenas rpc stream session cannot be reused")
@@ -60,7 +61,8 @@ func (c *Client) rpcOperationContext(ctx context.Context) (context.Context, cont
}
type discardRPCSessionError struct {
- err error
+ method string
+ err error
}
func (e *discardRPCSessionError) Error() string {
@@ -69,7 +71,7 @@ func (e *discardRPCSessionError) Error() string {
func (e *discardRPCSessionError) Unwrap() error { return e.err }
-func discardRPCSessionForStreamError(err error) error {
+func discardRPCSessionForStreamError(method string, err error) error {
if err == nil {
return nil
}
@@ -77,7 +79,7 @@ func discardRPCSessionForStreamError(err error) error {
if errors.As(err, &transportErr) {
return err
}
- return &discardRPCSessionError{err: err}
+ return &discardRPCSessionError{method: method, err: err}
}
// TransportMode is the connection-local API transport selected for an
@@ -326,6 +328,7 @@ func (c *Client) openAuthenticatedRPC(ctx context.Context) (*trueNASRPCClient, s
_ = conn.Close()
return nil, "", err
}
+ rpc.openedAt = time.Now()
rpc.startKeepalive(25 * time.Second)
return rpc, authMechanism, nil
}
@@ -354,13 +357,13 @@ func (c *Client) withRPC(ctx context.Context, operation func(context.Context, *t
}
if err := operation(ctx, c.rpc); err != nil {
if errors.Is(err, errRPCStreamSessionConsumed) {
- c.closeRPCLocked()
+ c.discardRPCLocked("stream_complete", err)
c.reconnect = 0
return nil
}
var discardErr *discardRPCSessionError
if errors.As(err, &discardErr) {
- c.closeRPCLocked()
+ c.discardRPCLocked("stream_error", err)
c.recordTransportError(err)
return err
}
@@ -369,7 +372,7 @@ func (c *Client) withRPC(ctx context.Context, operation func(context.Context, *t
c.recordRPCOperationError(err)
return err
}
- c.closeRPCLocked()
+ c.discardRPCLocked("transport_error", err)
c.recordTransportError(err)
if err := c.waitReconnectBackoff(ctx); err != nil {
return err
@@ -390,9 +393,18 @@ func (c *Client) withRPC(ctx context.Context, operation func(context.Context, *t
status.LastError = ""
})
if err := operation(ctx, c.rpc); err != nil {
+ if errors.Is(err, errRPCStreamSessionConsumed) {
+ c.discardRPCLocked("stream_complete", err)
+ c.reconnect = 0
+ return nil
+ }
+ var secondDiscardErr *discardRPCSessionError
var secondTransportErr *RPCTransportError
- if errors.As(err, &secondTransportErr) {
- c.closeRPCLocked()
+ if errors.As(err, &secondDiscardErr) {
+ c.discardRPCLocked("stream_error", err)
+ c.recordTransportError(err)
+ } else if errors.As(err, &secondTransportErr) {
+ c.discardRPCLocked("transport_error", err)
c.recordTransportError(err)
} else {
c.recordRPCOperationError(err)
@@ -442,7 +454,7 @@ func (c *Client) callRPCWithRetry(ctx context.Context, method string, params any
return err
}
- c.closeRPCLocked()
+ c.discardRPCLocked("transport_error", err)
c.recordTransportError(err)
if !retryRead {
return fmt.Errorf("truenas action %s transport failed after dispatch; outcome is unknown and Pulse will not replay it: %w", method, err)
@@ -470,7 +482,7 @@ func (c *Client) callRPCWithRetry(ctx context.Context, method string, params any
if err := c.rpc.call(ctx, method, params, result); err != nil {
var secondTransportErr *RPCTransportError
if errors.As(err, &secondTransportErr) {
- c.closeRPCLocked()
+ c.discardRPCLocked("transport_error", err)
c.recordTransportError(err)
} else {
c.recordRPCOperationError(err)
@@ -502,7 +514,13 @@ func (c *Client) waitReconnectBackoff(ctx context.Context) error {
}
func (c *Client) closeRPCLocked() {
+ c.discardRPCLocked("client_closed", nil)
+}
+
+func (c *Client) discardRPCLocked(reason string, err error) {
if c.rpc != nil && c.rpc.conn != nil {
+ method, category := rpcSessionCloseDetails(err)
+ c.rpc.closeSocket(reason, method, category)
c.rpc.close()
}
c.rpc = nil
@@ -531,7 +549,7 @@ func (c *trueNASRPCClient) startKeepalive(interval time.Duration) {
if err := c.conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(5*time.Second)); err != nil {
// Unblock any in-flight reader; ordinary transport handling owns retry
// and status, including the no-replay rule for actions.
- _ = c.conn.Close()
+ c.closeSocket("keepalive_write_error", "", "transport_error")
return
}
}
@@ -544,12 +562,68 @@ func (c *trueNASRPCClient) close() {
if c.keepaliveStop != nil {
close(c.keepaliveStop)
}
- _ = c.conn.Close()
+ c.closeSocket("client_closed", "", "")
if c.keepaliveDone != nil {
<-c.keepaliveDone
}
}
+// Record the first local disposal cause, including an idle keepalive failure
+// which may precede the next poll. Never log wire errors, params, subscription
+// IDs, endpoints or credentials. These fixed classifications explain a close;
+// they do not diagnose the appliance or proxy that caused an exchange to fail.
+func (c *trueNASRPCClient) closeSocket(reason, method, category string) {
+ c.closeOnce.Do(func() {
+ event := log.Info().
+ Str("component", "truenas_rpc").
+ Str("action", "close_session").
+ Str("reason", reason)
+ if !c.openedAt.IsZero() {
+ event.Dur("session_age", time.Since(c.openedAt))
+ }
+ if method != "" {
+ event.Str("method", method)
+ }
+ if category != "" {
+ event.Str("category", category)
+ }
+ event.Msg("Closing TrueNAS RPC session")
+ _ = c.conn.Close()
+ })
+}
+
+func rpcSessionCloseDetails(err error) (method, category string) {
+ if err == nil {
+ return "", ""
+ }
+ if errors.Is(err, errRPCStreamSessionConsumed) {
+ return "app.container_log_follow", "idle_timeout"
+ }
+ var discardErr *discardRPCSessionError
+ var transportErr *RPCTransportError
+ var rpcErr *RPCError
+ if errors.As(err, &discardErr) {
+ method = discardErr.method
+ } else if errors.As(err, &transportErr) {
+ method = transportErr.Method
+ } else if errors.As(err, &rpcErr) {
+ method = rpcErr.Method
+ }
+ switch {
+ case errors.Is(err, context.Canceled):
+ category = "cancelled"
+ case errors.Is(err, context.DeadlineExceeded) || isTimeoutError(err):
+ category = "timeout"
+ case errors.As(err, &rpcErr):
+ category = "method_error"
+ case errors.As(err, &transportErr):
+ category = "transport_error"
+ default:
+ category = "protocol_error"
+ }
+ return method, category
+}
+
func (c *Client) recordTransportError(err error) {
message := c.sanitizeTransportError(err)
c.updateTransportStatus(func(status *TransportStatus) {
diff --git a/internal/truenas/transport_test.go b/internal/truenas/transport_test.go
index 00d284e872..4dc03527f5 100644
--- a/internal/truenas/transport_test.go
+++ b/internal/truenas/transport_test.go
@@ -1,6 +1,7 @@
package truenas
import (
+ "bytes"
"context"
"encoding/json"
"errors"
@@ -16,12 +17,196 @@ import (
"time"
"github.com/gorilla/websocket"
+ "github.com/rs/zerolog"
+ "github.com/rs/zerolog/log"
)
+// Subscription terminal-message regressions for #2396. The envelope is from
+// the reporter's source-derived 25.04.2.6 example, not a retained wire capture.
+func TestJSONRPCSubscriptionTerminationRejectsWithoutWaitingOrRetry(t *testing.T) {
+ for _, stream := range []string{"app.stats", "reporting.realtime", "app.container_log_follow"} {
+ for _, beforeReply := range []bool{false, true} {
+ t.Run(fmt.Sprintf("%s/before_reply=%t", stream, beforeReply), func(t *testing.T) {
+ var subscribes, unsubscribes atomic.Int32
+ fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+ switch request.Method {
+ case "auth.login_ex":
+ return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS"}}
+ case "reporting.get_data":
+ return protocolFixtureReply{result: []any{}}
+ case "core.subscribe":
+ subscribes.Add(1)
+ collection := request.Params.([]any)[0].(string)
+ return protocolFixtureReply{result: "private-sub-id", beforeReply: beforeReply, notifications: []protocolFixtureNotification{{
+ method: "notify_unsubscribed", params: map[string]any{"collection": collection, "error": map[string]any{
+ "error": 14, "errname": "EFAULT", "reason": "private-app-name fixture-key fixture-user https://private.invalid",
+ "trace": "private-trace", "extra": "private-extra",
+ }},
+ }}}
+ case "core.unsubscribe":
+ unsubscribes.Add(1)
+ return protocolFixtureReply{result: nil}
+ case "pool.query":
+ return protocolFixtureReply{result: []any{}}
+ default:
+ t.Errorf("unexpected method %s", request.Method)
+ return protocolFixtureReply{close: true}
+ }
+ }, nil)
+ client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user", Timeout: 2 * time.Second})
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ started := time.Now()
+ var err error
+ switch stream {
+ case "app.stats":
+ var stats map[string]AppStats
+ stats, err = client.GetAppStats(ctx)
+ if stats != nil {
+ t.Error("rejected stats fabricated a sample")
+ }
+ case "reporting.realtime":
+ var telemetry *SystemInfo
+ telemetry, err = client.GetSystemTelemetry(ctx)
+ if telemetry != nil {
+ t.Error("rejected telemetry fabricated a sample")
+ }
+ default:
+ var lines []AppLogLine
+ lines, err = client.GetAppLogs(ctx, "private-app-name", "private-container", 100)
+ if lines != nil {
+ t.Error("rejected logs appeared successful")
+ }
+ }
+ var rpcErr *RPCError
+ if !errors.As(err, &rpcErr) || rpcErr.Code != 14 || rpcErr.Message != "subscription rejected" {
+ t.Fatalf("termination = %v, want structured errno=14 rejection", err)
+ }
+ if elapsed := time.Since(started); elapsed >= time.Second {
+ t.Fatalf("terminal event waited for the operation budget: %s", elapsed)
+ }
+ for _, private := range []string{"private-app-name", "fixture-key", "fixture-user", "private.invalid", "private-trace", "private-extra", "private-sub-id", "private-container"} {
+ if strings.Contains(err.Error(), private) || strings.Contains(client.TransportStatus().LastError, private) {
+ t.Fatalf("termination exposed %s", private)
+ }
+ }
+ if client.rpc != nil || client.TransportStatus().Connected || subscribes.Load() != 1 || unsubscribes.Load() != 0 || fixture.sessions.Load() != 1 {
+ t.Fatal("terminal rejection retried, reused the stream or tried to cancel its ended subscription")
+ }
+ if _, err := client.GetPools(ctx); err != nil || fixture.sessions.Load() != 2 || fixture.restRequests.Load() != 0 {
+ t.Fatalf("next read did not recover on a fresh RPC session: %v", err)
+ }
+ })
+ }
+ }
+}
+
+func TestJSONRPCSubscriptionTerminationIgnoresOtherCollections(t *testing.T) {
+ var unsubscribes atomic.Int32
+ fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+ switch request.Method {
+ case "auth.login_ex":
+ return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS"}}
+ case "core.subscribe":
+ collection := request.Params.([]any)[0].(string)
+ return protocolFixtureReply{result: "stats-sub", notifications: []protocolFixtureNotification{
+ {method: "notify_unsubscribed", params: map[string]any{"collection": "app.stats:{\"interval\":99}", "error": map[string]any{"error": 14}}},
+ {method: "notify_unsubscribed", params: map[string]any{"collection": "app.stats.extra:{\"interval\":2}", "error": "malformed-unrelated-error"}},
+ {method: "notify_unsubscribed", params: map[string]any{"collection": "reporting.realtime:{\"interval\":2}", "error": nil}},
+ {method: "collection_update", params: map[string]any{"collection": collection, "fields": []any{map[string]any{"app_name": "fixture-app", "cpu_usage": 7}}}},
+ }}
+ case "core.unsubscribe":
+ unsubscribes.Add(1)
+ return protocolFixtureReply{result: nil}
+ default:
+ return protocolFixtureReply{result: []any{}}
+ }
+ }, nil)
+ client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user"})
+ for poll := 0; poll < 2; poll++ {
+ stats, err := client.GetAppStats(context.Background())
+ if err != nil || stats["fixture-app"].CPUPercent != 7 {
+ t.Fatalf("unrelated terminal event interrupted stats: %v, %+v", err, stats)
+ }
+ }
+ if fixture.sessions.Load() != 1 || unsubscribes.Load() != 2 || !client.TransportStatus().Connected {
+ t.Fatal("unrelated event prevented ordinary cleanup and session reuse")
+ }
+}
+
+func TestJSONRPCSubscriptionCleanTerminationDoesNotInventStats(t *testing.T) {
+ fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+ if request.Method == "auth.login_ex" {
+ return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS"}}
+ }
+ if request.Method == "core.subscribe" {
+ return protocolFixtureReply{result: "stats-sub", notifications: []protocolFixtureNotification{{method: "notify_unsubscribed", params: map[string]any{"collection": request.Params.([]any)[0], "error": nil}}}}
+ }
+ return protocolFixtureReply{result: []any{}}
+ }, nil)
+ client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user", Timeout: 2 * time.Second})
+ stats, err := client.GetAppStats(context.Background())
+ var rpcErr *RPCError
+ if stats != nil || !errors.As(err, &rpcErr) || rpcErr.Message != "subscription ended before a sample" || client.TransportStatus().Connected {
+ t.Fatalf("clean termination appeared sampled or left an unresolved stream: %v, %+v", err, stats)
+ }
+}
+
+func TestJSONRPCSubscriptionCleanLogEndKeepsSession(t *testing.T) {
+ var unsubscribes atomic.Int32
+ fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+ if request.Method == "auth.login_ex" {
+ return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS"}}
+ }
+ if request.Method == "core.subscribe" {
+ collection := request.Params.([]any)[0]
+ return protocolFixtureReply{result: "log-sub", notifications: []protocolFixtureNotification{
+ {method: "collection_update", params: map[string]any{"collection": collection, "fields": []any{map[string]any{"data": "fixture-line"}}}},
+ {method: "notify_unsubscribed", params: map[string]any{"collection": "app.container_log_follow:{}", "error": map[string]any{"error": 14}}},
+ {method: "notify_unsubscribed", params: map[string]any{"collection": collection, "error": nil}},
+ }}
+ }
+ if request.Method == "core.unsubscribe" {
+ unsubscribes.Add(1)
+ }
+ return protocolFixtureReply{result: []any{}}
+ }, nil)
+ client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user"})
+ lines, err := client.GetAppLogs(context.Background(), "fixture-app", "fixture-container", 100)
+ if err != nil || len(lines) != 1 || lines[0].Data != "fixture-line" {
+ t.Fatalf("clean log end lost the bounded tail: %v, %+v", err, lines)
+ }
+ if _, err := client.GetPools(context.Background()); err != nil || fixture.sessions.Load() != 1 || unsubscribes.Load() != 0 || !client.TransportStatus().Connected {
+ t.Fatalf("clean log end discarded or unsubscribed an already-ended stream: %v", err)
+ }
+}
+
+func TestJSONRPCSubscriptionMalformedTerminalDiscardsWithoutWireText(t *testing.T) {
+ for _, terminal := range []any{map[string]any{"error": "private-invalid-errno", "reason": "private-error"}, "private-error", true} {
+ t.Run(fmt.Sprintf("%T", terminal), func(t *testing.T) {
+ fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+ if request.Method == "auth.login_ex" {
+ return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS"}}
+ }
+ if request.Method == "core.subscribe" {
+ return protocolFixtureReply{result: "stats-sub", notifications: []protocolFixtureNotification{{method: "notify_unsubscribed", params: map[string]any{"collection": request.Params.([]any)[0], "error": terminal}}}}
+ }
+ return protocolFixtureReply{result: []any{}}
+ }, nil)
+ client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user", Timeout: 2 * time.Second})
+ _, err := client.GetAppStats(context.Background())
+ if err == nil || strings.Contains(err.Error(), "private-") || !strings.Contains(err.Error(), "invalid truenas app.stats termination error") || client.rpc != nil || fixture.sessions.Load() != 1 {
+ t.Fatalf("malformed termination was lost, leaked or reused: %v", err)
+ }
+ })
+ }
+}
+
type protocolFixtureReply struct {
result any
err *trueNASRPCError
notifications []protocolFixtureNotification
+ beforeReply bool
close bool
}
@@ -86,8 +271,10 @@ func newProtocolFixture(
}
response.Result = raw
}
- if err := conn.WriteJSON(response); err != nil {
- return
+ if !reply.beforeReply {
+ if err := conn.WriteJSON(response); err != nil {
+ return
+ }
}
for _, notification := range reply.notifications {
params, err := json.Marshal(notification.params)
@@ -103,6 +290,11 @@ func newProtocolFixture(
return
}
}
+ if reply.beforeReply {
+ if err := conn.WriteJSON(response); err != nil {
+ return
+ }
+ }
}
}))
t.Cleanup(fixture.server.Close)
@@ -786,8 +978,8 @@ func TestJSONRPCFetchSnapshotUsesPublishedTrueNAS26MethodsWithoutREST(t *testing
subscribeCalls := methodCalls["core.subscribe"]
unsubscribeCalls := methodCalls["core.unsubscribe"]
methodCallsMu.Unlock()
- if subscribeCalls != 2 || unsubscribeCalls != 2 {
- t.Fatalf("subscription lifecycle calls: subscribe=%d unsubscribe=%d, want 2/2", subscribeCalls, unsubscribeCalls)
+ if subscribeCalls != 1 || unsubscribeCalls != 1 {
+ t.Fatalf("subscription lifecycle calls: subscribe=%d unsubscribe=%d, want 1/1 (realtime only for empty apps)", subscribeCalls, unsubscribeCalls)
}
if fixture.sessions.Load() != 1 || fixture.restRequests.Load() != 0 {
t.Fatalf("snapshot sessions=%d REST=%d, want one modern session and no REST",
@@ -1419,3 +1611,272 @@ func TestJSONRPCSnapshotContinuesAfterTelemetryTimeout(t *testing.T) {
t.Fatalf("inventory did not recover on RPC after discarding the stream: sessions=%d rest=%d", fixture.sessions.Load(), fixture.restRequests.Load())
}
}
+
+func TestJSONRPCEmptyAppsDoNotSubscribeToStats(t *testing.T) {
+ var appQueries, statsSubscriptions atomic.Int32
+ fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+ switch request.Method {
+ case "auth.login_ex":
+ return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS"}}
+ case "app.query":
+ if appQueries.Add(1) == 1 {
+ return protocolFixtureReply{result: nil}
+ }
+ return protocolFixtureReply{result: []any{}}
+ case "core.subscribe":
+ statsSubscriptions.Add(1)
+ // Deliberately silent stats on an appliance with no apps. The final
+ // client must never issue this request, even with a cancellation-only
+ // caller context like the production poller.
+ return protocolFixtureReply{result: "unavailable-apps"}
+ case "system.info":
+ return protocolFixtureReply{result: map[string]any{"hostname": "empty-app-nas", "version": "TrueNAS-SCALE-25.04.2.6"}}
+ default:
+ t.Errorf("unexpected method %s", request.Method)
+ return protocolFixtureReply{close: true}
+ }
+ }, nil)
+ client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user", Timeout: 100 * time.Millisecond})
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ for i := 0; i < 2; i++ {
+ apps, err := client.GetApps(ctx)
+ if err != nil || apps == nil || len(apps) != 0 {
+ t.Fatalf("empty app query %d did not preserve an empty successful inventory", i)
+ }
+ }
+ if statsSubscriptions.Load() != 0 {
+ t.Fatal("empty inventory still issued a stats subscription")
+ }
+ if _, err := client.GetSystemInfo(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if fixture.sessions.Load() != 1 || fixture.restRequests.Load() != 0 || !client.TransportStatus().Connected {
+ t.Fatal("empty apps discarded the healthy authenticated session or changed transport")
+ }
+}
+
+func TestLegacyRESTEmptyAppsDoNotNegotiateStats(t *testing.T) {
+ var appQueries atomic.Int32
+ fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+ t.Errorf("empty legacy app inventory issued RPC method %s", request.Method)
+ return protocolFixtureReply{close: true}
+ }, func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path != "/api/v2.0/app" {
+ t.Errorf("unexpected legacy path %s", r.URL.Path)
+ return
+ }
+ appQueries.Add(1)
+ _ = json.NewEncoder(w).Encode([]any{})
+ })
+ client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key"})
+ // Negotiation is covered separately; isolate the already-selected legacy
+ // inventory path here, so no attempted JSON-RPC stats lookup is hidden.
+ client.mode = TransportLegacyREST
+ apps, err := client.GetApps(context.Background())
+ if err != nil || apps == nil || len(apps) != 0 || appQueries.Load() != 1 || fixture.sessions.Load() != 0 {
+ t.Fatal("empty legacy app inventory did not stay a single REST read")
+ }
+}
+
+func TestJSONRPCStreamRetryDiscardsUnusableSecondSession(t *testing.T) {
+ for _, failure := range []string{"malformed_event", "empty_subscription", "unsubscribe_error"} {
+ t.Run(failure, func(t *testing.T) {
+ var subscribes atomic.Int32
+ fixture := newProtocolFixture(t, func(session int, request trueNASRPCRequest) protocolFixtureReply {
+ switch request.Method {
+ case "auth.login_ex":
+ return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS"}}
+ case "core.subscribe":
+ subscribes.Add(1)
+ if session == 1 {
+ return protocolFixtureReply{close: true}
+ }
+ if failure == "empty_subscription" {
+ return protocolFixtureReply{result: ""}
+ }
+ var fields any = []any{map[string]any{"app_name": "fixture-app", "cpu_usage": 3}}
+ if failure == "malformed_event" {
+ fields = "invalid-fields"
+ }
+ return protocolFixtureReply{result: "private-subscription-id", notifications: []protocolFixtureNotification{{method: "collection_update", params: map[string]any{"collection": "app.stats:{\"interval\":2}", "fields": fields}}}}
+ case "core.unsubscribe":
+ return protocolFixtureReply{err: &trueNASRPCError{Code: -32001, Message: "fixture-method-error"}}
+ case "pool.query":
+ if session != 3 {
+ t.Error("next read reused an unusable second stream session")
+ }
+ return protocolFixtureReply{result: []any{}}
+ default:
+ t.Errorf("unexpected method %s", request.Method)
+ return protocolFixtureReply{close: true}
+ }
+ }, nil)
+ client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user"})
+ if _, err := client.GetAppStats(context.Background()); err == nil {
+ t.Fatal("bad second stream session appeared successful")
+ }
+ if client.TransportStatus().Connected {
+ t.Fatal("unusable second stream session remained connected")
+ }
+ if client.rpc != nil {
+ t.Fatal("unusable second stream session was retained")
+ }
+ if _, err := client.GetPools(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ if subscribes.Load() != 2 || fixture.sessions.Load() != 3 || fixture.restRequests.Load() != 0 {
+ t.Fatal("stream retry/disposal changed its retry limit or downgraded transport")
+ }
+ })
+ }
+}
+
+func TestJSONRPCStreamRetryReturnsCompletedLogTail(t *testing.T) {
+ fixture := newProtocolFixture(t, func(session int, request trueNASRPCRequest) protocolFixtureReply {
+ switch request.Method {
+ case "auth.login_ex":
+ return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS"}}
+ case "core.subscribe":
+ if session == 1 {
+ return protocolFixtureReply{close: true}
+ }
+ return protocolFixtureReply{result: "fixture-log-sub", notifications: []protocolFixtureNotification{{method: "collection_update", params: map[string]any{"collection": "app.container_log_follow:{}", "fields": []any{map[string]any{"data": "fixture-line"}}}}}}
+ case "pool.query":
+ if session != 3 {
+ t.Error("timed-out log socket was reused")
+ }
+ return protocolFixtureReply{result: []any{}}
+ default:
+ t.Errorf("unexpected method %s", request.Method)
+ return protocolFixtureReply{close: true}
+ }
+ }, nil)
+ client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user"})
+ lines, err := client.GetAppLogs(context.Background(), "fixture-app", "fixture-container", 100)
+ if err != nil || len(lines) != 1 || lines[0].Data != "fixture-line" {
+ t.Fatal("completed retry lost its log tail or returned a consumed-session sentinel")
+ }
+ if client.TransportStatus().Connected || client.rpc != nil {
+ t.Fatal("completed retry kept a timed-out log socket")
+ }
+ if _, err := client.GetPools(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ if fixture.sessions.Load() != 3 || fixture.restRequests.Load() != 0 {
+ t.Fatal("next read did not open exactly one fresh modern session")
+ }
+}
+
+func TestRPCSessionCloseDetails(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ err error
+ method, category string
+ }{
+ {"explicit", nil, "", ""},
+ {"idle tail", errRPCStreamSessionConsumed, "app.container_log_follow", "idle_timeout"},
+ {"timeout", &RPCTransportError{Method: "pool.query", Phase: "read", Err: context.DeadlineExceeded}, "pool.query", "timeout"},
+ {"cancel", &RPCTransportError{Method: "app.start", Phase: "read", Err: context.Canceled}, "app.start", "cancelled"},
+ {"transport", &RPCTransportError{Method: "system.info", Phase: "read", Err: errors.New("private-error-text")}, "system.info", "transport_error"},
+ {"stream decode", &discardRPCSessionError{method: "app.stats", err: errors.New("private-event-text")}, "app.stats", "protocol_error"},
+ {"unsubscribe refusal", &discardRPCSessionError{method: "core.unsubscribe", err: &RPCError{Method: "core.unsubscribe", Message: "private-wire-message"}}, "core.unsubscribe", "method_error"},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ method, category := rpcSessionCloseDetails(tc.err)
+ if method != tc.method || category != tc.category {
+ t.Fatalf("close classification = %s/%s, want %s/%s", method, category, tc.method, tc.category)
+ }
+ })
+ }
+}
+
+func captureRPCCloseLogs(t *testing.T) *bytes.Buffer {
+ t.Helper()
+ var output bytes.Buffer
+ previous := log.Logger
+ log.Logger = zerolog.New(&output)
+ t.Cleanup(func() { log.Logger = previous })
+ return &output
+}
+
+func rpcCloseLogRecords(t *testing.T, output *bytes.Buffer) []map[string]any {
+ t.Helper()
+ var records []map[string]any
+ decoder := json.NewDecoder(bytes.NewReader(output.Bytes()))
+ for decoder.More() {
+ var record map[string]any
+ if err := decoder.Decode(&record); err != nil {
+ t.Fatal("close log is not structured JSON")
+ }
+ records = append(records, record)
+ }
+ return records
+}
+
+func TestRPCSessionCloseLogIsBoundedAndSecretFree(t *testing.T) {
+ output := captureRPCCloseLogs(t)
+ fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+ switch request.Method {
+ case "auth.login_ex":
+ return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS"}}
+ case "core.subscribe":
+ return protocolFixtureReply{result: "private-subscription-id", notifications: []protocolFixtureNotification{{method: "collection_update", params: map[string]any{"collection": "app.stats:{\"interval\":2}", "fields": []any{}}}}}
+ case "core.unsubscribe":
+ return protocolFixtureReply{err: &trueNASRPCError{Code: -32001, Message: "private-wire-message private-api-key private-password private-subscription-id"}}
+ default:
+ return protocolFixtureReply{close: true}
+ }
+ }, nil)
+ client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "private-api-key", Password: "private-password", Username: "private-owner"})
+ if _, err := client.GetAppStats(context.Background()); err == nil {
+ t.Fatal("fixture did not fail its unsubscribe")
+ }
+ client.Close()
+ records := rpcCloseLogRecords(t, output)
+ if len(records) != 1 {
+ t.Fatalf("close records = %d, want exactly one for disposal and subsequent Close", len(records))
+ }
+ record := records[0]
+ if record["reason"] != "stream_error" || record["method"] != "core.unsubscribe" || record["category"] != "method_error" || record["component"] != "truenas_rpc" || record["action"] != "close_session" {
+ t.Fatal("close log lost its fixed local cause")
+ }
+ if _, ok := record["session_age"].(float64); !ok {
+ t.Fatal("close log lost the authenticated session age")
+ }
+ for _, private := range []string{"private-api-key", "private-password", "private-owner", "private-wire-message", "private-subscription-id", fixture.server.URL} {
+ if bytes.Contains(output.Bytes(), []byte(private)) {
+ t.Fatal("close log contains private wire/configuration data")
+ }
+ }
+ for key := range record {
+ switch key {
+ case "component", "action", "reason", "method", "category", "session_age", "level", "message":
+ default:
+ t.Fatalf("unexpected close log field %s", key)
+ }
+ }
+}
+
+func TestRPCSessionCloseLogKeepsFirstKeepaliveCause(t *testing.T) {
+ output := captureRPCCloseLogs(t)
+ fixture := newProtocolFixture(t, func(_ int, _ trueNASRPCRequest) protocolFixtureReply { return protocolFixtureReply{result: true} }, nil)
+ client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{})
+ conn, err := client.dialRPC(context.Background())
+ if err != nil {
+ t.Fatal(err)
+ }
+ rpc := &trueNASRPCClient{conn: conn, nextID: 1, openedAt: time.Now()}
+ rpc.startKeepalive(time.Millisecond)
+ _ = conn.Close() // Inject a control-write failure, without an RPC payload.
+ select {
+ case <-rpc.keepaliveDone:
+ case <-time.After(time.Second):
+ t.Fatal("failed keepalive did not stop")
+ }
+ rpc.close()
+ records := rpcCloseLogRecords(t, output)
+ if len(records) != 1 || records[0]["reason"] != "keepalive_write_error" || records[0]["category"] != "transport_error" {
+ t.Fatal("later client disposal overwrote or duplicated the first keepalive close cause")
+ }
+}
diff --git a/scripts/tests/test_pbs_docs.py b/scripts/tests/test_pbs_docs.py
index d13c26dab5..3e9f104fcb 100644
--- a/scripts/tests/test_pbs_docs.py
+++ b/scripts/tests/test_pbs_docs.py
@@ -125,6 +125,17 @@ class PBSDocsTest(unittest.TestCase):
"Do not delete backups", "Do not clear History"):
self.assertIn(phrase, text)
+ def test_history_checks_cover_batched_and_single_metric_requests(self):
+ history = DOC.read_text().split("### PBS is connected but History stays empty\n")[1].split("\n### ")[0]
+ for phrase in ("without\n `metric`", "do not wait for a separate `metric=cpu` request",
+ "`metrics: {}`", "`metrics.cpu` array", "CPU is absent",
+ "`points` array for that single metric", "A measured zero is a\n sample",
+ "Do not edit or replay", "Do not share a HAR", "enable\nDebug logging"):
+ with self.subTest(phrase=phrase):
+ self.assertIn(phrase, history)
+ self.assertNotIn("completed request with `metric=cpu`", history)
+ self.assertIn("does\n not prove there is no data under another target", history)
+
def test_every_documented_shell_recipe_parses(self):
for command in blocks():
with self.subTest(command=command):