diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index af091c6477..a8a2e2ebed 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -133,6 +133,43 @@ precede marker production. Legacy reports without the marker retain the existing default policy; matched agent and server support is required. This proves ingestion behaviour, not reporter installation or release delivery. +### TrueNAS RPC operation budgets and poll isolation + +The configured client timeout (30 seconds by default) bounds each serialized +RPC operation, including its session-lock wait, any inline authentication, +method exchange or subscription, and the permitted read retry/backoff. They +share one budget; initial transport negotiation before a method is separately +bounded by the same timeout. A shorter caller deadline remains effective. A cancelled +waiter neither dispatches a request nor alters the current owner's socket or +transport status. RPC/stream readers receive the bounded context, and a +timed-out socket is discarded before a subsequent operation authenticates a +fresh session. Modern appliances never downgrade to REST on timeout, and an +action timeout after dispatch retains its unknown outcome without replay. +Keepalive still belongs to the session, not any completed operation's context. + +Optional live-telemetry timeouts do not prevent otherwise usable inventory; +unobserved metrics remain unavailable. A required-method timeout returns a +truthful failure, preserves the prior successful observation/cached identity, +and lets the shared poll cycle continue to other connections. Socket deadline +errors are classified as timeout, not credential or generic connection failures. +The next normal poll can recover; completion-based failure backoff is unchanged. +This is an operation budget, not a new total-snapshot deadline or parallel poller. +The app-log idle window still yields a bounded tail; an earlier caller/operation +deadline is an error rather than a successful empty log response. + +`TestJSONRPCConfiguredTimeout*`, +`TestJSONRPCWaitingDeadlineDoesNotInterruptSessionOwner`, +`TestJSONRPCPreCancelledCallDoesNotDispatch` and +`TestJSONRPCSnapshotContinuesAfterTelemetryTimeout` in +`internal/truenas/transport_test.go` cover silent handshake/authentication, +read/retry budgets, all three stream readers, queued cancellation, caller +deadlines, recovery and action no-replay. +`TestTrueNASPollerUnresponsiveRPCDoesNotFreezeOtherConnections` and +`TestClassifyTrueNASError` in `internal/monitoring/truenas_poller_test.go` +cover actual TLS/RPC clients through two successful/failed/recovered poll cycles, +cached identities, truthful health and unchanged backoff. These are controlled +runtime proofs, not native SCALE acceptance or a diagnosis/resolution of #2382. + ### TrueNAS persistent-session liveness and successful poll cadence Authenticated JSON-RPC WebSocket sessions send transport-only PING controls diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index 987a1e1836..6431645e11 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -6245,3 +6245,19 @@ state. The Recovery settings item and its capability gate are untouched. `internal/api/onboarding_handlers.go` changed only the human-readable pairing readiness messages so they name Settings > Pulse Mobile. This opens no new storage or recovery path. + +### Live physical-disk row continuity + +Physical Disks uses the canonical keyed windowed renderer, which retains row +owners across resource snapshots. Disk presentation must therefore stay +reactive: health and its evidence, temperature, endurance, capacity, device, +host and placement labels reflect the latest snapshot rather than the mount-time +sample. Removed readings revert to their existing unknown/placeholders without +retaining old warning text or colour. Metric summary identities, highlight +matching and disclosure control IDs track the current canonical disk target. +Unchanged resource IDs keep their mounted row, focus and expanded detail; +attention filtering and the displayed health must agree through fault/recovery. +This changes no collector, health thresholds, history policy or server identity. +`frontend-modern/src/components/Storage/__tests__/DiskList.test.tsx` exercises +snapshot replacement, in-place reconciliation, missing evidence, target changes +and fault/recovery using the real keyed renderer. diff --git a/frontend-modern/browser-tests/disk-live-refresh.cjs b/frontend-modern/browser-tests/disk-live-refresh.cjs new file mode 100644 index 0000000000..11f7d22373 --- /dev/null +++ b/frontend-modern/browser-tests/disk-live-refresh.cjs @@ -0,0 +1,177 @@ +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const path = require('node:path'); +const { createHash } = require('node:crypto'); +const { chromium, webkit } = require('playwright'); + +(async () => { + const root = '/workspace/frontend-modern'; + process.chdir(root); + const artifacts = path.join(root, 'node_modules/disk-live-refresh-proof'); + fs.mkdirSync(artifacts, { recursive: true }); + const { createServer } = await import(path.join(root, 'node_modules/vite/dist/node/index.js')); + const server = await createServer({ + root, + configFile: path.join(root, 'vite.config.ts'), + cacheDir: fs.mkdtempSync(path.join(artifacts, 'vite-cache-')), + server: { host: '127.0.0.1', port: 5221, strictPort: true }, + }); + const observations = []; + let browser; + try { + await server.listen(); + for (const scenario of [ + { engine: chromium, name: 'chromium', width: 1365, height: 900, dark: false }, + { engine: webkit, name: 'webkit', width: 390, height: 844, dark: true }, + ]) { + browser = await scenario.engine.launch( + scenario.name === 'chromium' + ? { headless: true, channel: 'chromium', args: ['--no-sandbox'] } + : { headless: true }, + ); + const page = await browser.newPage({ + viewport: { width: scenario.width, height: scenario.height }, + isMobile: scenario.name === 'webkit', + hasTouch: scenario.name === 'webkit', + }); + const errors = [], + requests = []; + page.on('pageerror', (error) => errors.push(error.message)); + await page.route('**/*', async (route) => { + const url = new URL(route.request().url()); + if (url.origin !== 'http://127.0.0.1:5221') return route.abort(); + if (!url.pathname.startsWith('/api/')) return route.continue(); + requests.push(url.pathname); + if (url.pathname === '/api/license/runtime-capabilities') + return route.fulfill({ + json: { + capabilities: [], + limits: [], + max_history_days: 7, + hosted_mode: false, + runtime: { build: 'community', label: 'Pulse Community runtime' }, + blocked_capabilities: [], + }, + }); + return route.fulfill({ json: { data: [], enabled: false } }); + }); + await page.goto('http://127.0.0.1:5221/browser-tests/disk-live-refresh.html'); + await page.evaluate( + (dark) => document.documentElement.classList.toggle('dark', dark), + scenario.dark, + ); + const fixture = page.getByTestId('disk-live-refresh-fixture'); + const row = fixture.locator('[data-row-id="disk-one"]'); + await row.getByText('Healthy', { exact: true }).waitFor(); + await row.getByText('Archive SSD', { exact: true }).click(); + const disclosure = row.getByRole('button', { name: 'Collapse Archive SSD', exact: true }); + await disclosure.focus(); + await row.evaluate((element) => { + element.dataset.proofOwner = 'retained'; + }); + const initialControls = await disclosure.getAttribute('aria-controls'); + const snapshot = (phase) => + page.locator(`button[data-update="${phase}"]`).evaluate((button) => button.click()); + const checkContinuity = async () => { + assert.equal(await row.getAttribute('data-proof-owner'), 'retained'); + assert.equal( + await row.locator('button').evaluate((button) => document.activeElement === button), + true, + ); + assert.equal(await row.locator('button').getAttribute('aria-expanded'), 'true'); + const controls = await row.locator('button').getAttribute('aria-controls'); + assert.equal(await page.locator(`[id="${controls}"]`).count(), 1); + assert.equal(await fixture.locator('[data-inline-detail-for]').count(), 1); + const dimensions = await page.evaluate(() => ({ + scroll: document.documentElement.scrollWidth, + width: innerWidth, + })); + assert.ok(dimensions.scroll <= dimensions.width + 1, JSON.stringify(dimensions)); + return dimensions; + }; + const screenshot = (state) => + fixture.screenshot({ path: path.join(artifacts, `${scenario.name}-${state}.png`) }); + + await snapshot('fault'); + await row.getByText('Replace Now', { exact: true }).waitFor(); + for (const text of ['Archive SSD (fault)', '4%', '63°C']) + assert.equal(await row.getByText(text, { exact: true }).count(), 1); + assert.equal(await row.getByText('Healthy', { exact: true }).count(), 0); + assert.equal(await row.getAttribute('data-summary-series-id'), 'agent-archive:sdz'); + assert.notEqual(await row.locator('button').getAttribute('aria-controls'), initialControls); + await fixture.getByText('2', { exact: true }).waitFor(); + const faultDimensions = await checkContinuity(); + await screenshot('fault'); + + await snapshot('missing'); + await row.getByText('Unknown', { exact: true }).waitFor(); + for (const text of ['Replace Now', 'SMART failed.', '4%', '63°C']) + assert.equal(await row.getByText(text, { exact: true }).count(), 0); + for (const column of ['temp', 'life', 'size', 'role', 'parent']) + assert.equal( + (await row.locator(`td[data-storage-column="${column}"]`).innerText()).trim(), + '—', + ); + await fixture + .getByText('Temperature is temporarily unavailable: No current reading', { exact: true }) + .waitFor(); + const missingDimensions = await checkContinuity(); + await screenshot('missing'); + + await snapshot('healthy'); + await row.getByText('Healthy', { exact: true }).waitFor(); + assert.equal(await row.getByText('96%', { exact: true }).count(), 1); + assert.equal(await row.getByText('41°C', { exact: true }).count(), 1); + assert.equal(await row.getAttribute('data-summary-series-id'), 'agent-archive:sda'); + assert.equal(await row.locator('button').getAttribute('aria-controls'), initialControls); + const recoveredDimensions = await checkContinuity(); + await screenshot('recovered'); + + await snapshot('fault'); + await row.getByText('Replace Now', { exact: true }).waitFor(); + await page.locator('[data-filter="attention"]').evaluate((button) => button.click()); + assert.equal(await row.getByText('Replace Now', { exact: true }).count(), 1); + await snapshot('healthy'); + await fixture.getByText('No disks need attention', { exact: true }).waitFor(); + await page.locator('[data-filter="all"]').evaluate((button) => button.click()); + await row.getByText('Healthy', { exact: true }).waitFor(); + assert.deepEqual(errors, []); + observations.push({ + browser: scenario.name, + version: browser.version(), + viewport: { width: scenario.width, height: scenario.height }, + dark: scenario.dark, + faultDimensions, + missingDimensions, + recoveredDimensions, + errors, + requests, + }); + await browser.close(); + browser = undefined; + } + const source = 'src/components/Storage/DiskList.tsx'; + const result = { + result: 'passed', + playwright: require('playwright/package.json').version, + content_sha256: { + [`frontend-modern/${source}`]: createHash('sha256') + .update(fs.readFileSync(path.join(root, source))) + .digest('hex'), + }, + observations, + }; + fs.writeFileSync(path.join(artifacts, 'result.json'), JSON.stringify(result, null, 2) + '\n'); + console.log(JSON.stringify(result)); + } finally { + if (browser) await browser.close(); + await server.close(); + for (const name of fs.readdirSync(artifacts)) { + const file = path.join(artifacts, name); + if (fs.statSync(file).isFile()) fs.chmodSync(file, 0o644); + } + } +})().catch((error) => { + console.error(error); + process.exitCode = 1; +}); diff --git a/frontend-modern/browser-tests/disk-live-refresh.html b/frontend-modern/browser-tests/disk-live-refresh.html new file mode 100644 index 0000000000..e3fc664d80 --- /dev/null +++ b/frontend-modern/browser-tests/disk-live-refresh.html @@ -0,0 +1,12 @@ + + + + + + Physical disk live refresh verification + + +
+ + + diff --git a/frontend-modern/browser-tests/disk-live-refresh.tsx b/frontend-modern/browser-tests/disk-live-refresh.tsx new file mode 100644 index 0000000000..e7a7dc87dc --- /dev/null +++ b/frontend-modern/browser-tests/disk-live-refresh.tsx @@ -0,0 +1,116 @@ +// Real DiskList/DiskDetail and keyed renderer, with synthetic collector snapshots. +// This is browser behaviour evidence, not an installed SMART or appliance result. +import { createSignal } from 'solid-js'; +import { render } from 'solid-js/web'; +import { DiskList } from '../src/components/Storage/DiskList'; +import type { StorageHealthFilter } from '../src/features/storageBackups/models'; +import type { Resource } from '../src/types/resource'; +import '../src/index.css'; + +const disk = (phase: 'healthy' | 'fault' | 'missing'): Resource => ({ + id: 'disk-one', + type: 'physical_disk', + name: 'Archive disk', + platformType: 'proxmox-pbs', + sourceType: 'agent', + status: 'online', + lastSeen: Date.now(), + metricsTarget: { + resourceType: 'disk', + resourceId: phase === 'healthy' ? 'agent-archive:sda' : 'agent-archive:sdz', + }, + identity: { hostname: 'archive-host' }, + canonicalIdentity: { hostname: 'archive-host' }, + physicalDisk: { + model: phase === 'fault' ? 'Archive SSD (fault)' : 'Archive SSD', + devPath: phase === 'healthy' ? '/dev/sda' : '/dev/sdz', + serial: 'SYNTHETIC-ONE', + diskType: phase === 'missing' ? undefined : 'ssd', + health: phase === 'healthy' ? 'PASSED' : phase === 'fault' ? 'FAILED' : 'UNKNOWN', + wearout: phase === 'healthy' ? 96 : phase === 'fault' ? 4 : -1, + temperature: phase === 'healthy' ? 41 : phase === 'fault' ? 63 : 0, + sizeBytes: phase === 'missing' ? 0 : 2_000_000_000_000, + storageRole: phase === 'missing' ? undefined : 'cache_pool', + storageGroup: phase === 'missing' ? undefined : 'Archive Pool', + risk: + phase === 'fault' + ? { + level: 'critical', + reasons: [{ code: 'smart-failed', severity: 'critical', summary: 'SMART failed.' }], + } + : undefined, + smart: + phase === 'missing' + ? undefined + : { powerOnHours: 100, pendingSectors: phase === 'fault' ? 2 : 0 }, + collection: + phase === 'missing' + ? { + temperature: { state: 'unavailable', source: 'fixture', reason: 'No current reading' }, + } + : undefined, + }, +}); + +const Fixture = () => { + const [disks, setDisks] = createSignal([disk('healthy')]); + const [selectedDiskId, setSelectedDiskId] = createSignal(null); + const [healthFilter, setHealthFilter] = createSignal('all'); + return ( +
+

Physical disk live refresh verification

+

+ Synthetic snapshots in the production PBS disk table and detail. +

+
+ + + + + +
+
+ +
+
+ ); +}; +render(() => , document.getElementById('root')!); diff --git a/frontend-modern/browser-verification.json b/frontend-modern/browser-verification.json index 45fb502a24..bc8f3ba8bc 100644 --- a/frontend-modern/browser-verification.json +++ b/frontend-modern/browser-verification.json @@ -1,49 +1,53 @@ { "version": 1, - "base_sha": "1a0cd7c9d5aa02de754535a293dc0cacd4ef6fa0", - "verified_at": "2026-10-01T18:27:14.912695Z", + "base_sha": "33d6d729c1dc7c0913a970fd4c463a9f9575b7b1", + "verified_at": "2026-10-02T10:16:57.632722Z", "result": "passed", "changed_paths": [ - "frontend-modern/src/features/docs/docMarkdown.ts" + "frontend-modern/src/components/Storage/DiskList.tsx" ], "content_sha256": { - "frontend-modern/src/features/docs/docMarkdown.ts": "572a3b4bc5bc1114adef5aa64bb0ea833c0dbe5865bbd57d3dc767fd1965aa55" + "frontend-modern/src/components/Storage/DiskList.tsx": "df63db162779ff0d2064475a9db1fd67ac748703f398a88d108ecdd154dae9d2" }, "routes": [ - "/docs/PULSE_PRO", - "/browser-tests/docs-fragment-navigation.html?scenario=plans" + "/browser-tests/disk-live-refresh.html" ], "viewports": [ { - "width": 1280, + "width": 1365, "height": 900 }, { "width": 390, - "height": 900 + "height": 844 } ], "states": [ - "Production Docs page/router/styles with the current shipped eight-column plan comparison, not a replacement table fixture", - "Every sanitized thead TH has trusted scope=col and all eight native columnheader roles; document role/scope/class remain stripped", - "Desktop Chromium light and phone-emulated WebKit dark; table-local horizontal scrolling and outer-page containment; four final full-page/header screenshots inspected", - "This establishes browser accessibility roles, not screen-reader speech or published availability" + "Production DiskList, DiskDetail, keyed renderer and styles with synthetic PBS collector snapshots, not installed appliance evidence", + "Healthy to failed disk updates health, warning text, endurance, temperature, model and canonical target without remounting the row or its expanded detail", + "Missing measurements remove obsolete values and critical styling; unavailable detail evidence remains visible; recovery restores current values", + "Disclosure keyboard focus and aria-controls survive same-ID updates; changed metric targets update summary/detail linkage", + "Attention filter includes the failed disk, removes the recovered disk and returns current healthy values in All", + "Desktop Chromium light and phone-emulated WebKit dark have no outer-page overflow or page errors; six fault/missing/recovered screenshots inspected; existing dense phone truncation is unchanged" ], "interactions": [ - "Navigate to the real plans document and inspect table DOM and accessibility snapshot", - "On phone, scroll the table horizontally without moving the outer page; bring its header into view" + "Expand the real disk row and focus its disclosure, apply fault/missing/recovery snapshots without navigating or reloading", + "Check current values, stable owner, focus, live disclosure target and selected detail after each snapshot", + "Toggle Attention/All across fault and recovery" ], - "command": "pulse-worker-browser frontend-modern/browser-tests/docs-table-headers.cjs", + "command": "pulse-worker-browser frontend-modern/browser-tests/disk-live-refresh.cjs", "browser_versions": { "playwright": "1.56.1", "chromium": "141.0.7390.37", "webkit": "26.0" }, "artifacts": [ - "/var/lib/pulse-maintainer/worker-outputs/web-product-yqww6hbj/browser/docs/chromium-after.png", - "/var/lib/pulse-maintainer/worker-outputs/web-product-yqww6hbj/browser/docs/chromium-headers.png", - "/var/lib/pulse-maintainer/worker-outputs/web-product-yqww6hbj/browser/docs/webkit-after.png", - "/var/lib/pulse-maintainer/worker-outputs/web-product-yqww6hbj/browser/docs/webkit-headers.png", - "/var/lib/pulse-maintainer/worker-outputs/web-product-yqww6hbj/browser/docs/after.json" + "/var/lib/pulse-maintainer/worker-outputs/web-product-zuzev1gs/browser/chromium-fault.png", + "/var/lib/pulse-maintainer/worker-outputs/web-product-zuzev1gs/browser/chromium-missing.png", + "/var/lib/pulse-maintainer/worker-outputs/web-product-zuzev1gs/browser/chromium-recovered.png", + "/var/lib/pulse-maintainer/worker-outputs/web-product-zuzev1gs/browser/webkit-fault.png", + "/var/lib/pulse-maintainer/worker-outputs/web-product-zuzev1gs/browser/webkit-missing.png", + "/var/lib/pulse-maintainer/worker-outputs/web-product-zuzev1gs/browser/webkit-recovered.png", + "/var/lib/pulse-maintainer/worker-outputs/web-product-zuzev1gs/browser/result.json" ] } diff --git a/frontend-modern/src/components/Storage/DiskList.tsx b/frontend-modern/src/components/Storage/DiskList.tsx index 280ebce9cf..5715390316 100644 --- a/frontend-modern/src/components/Storage/DiskList.tsx +++ b/frontend-modern/src/components/Storage/DiskList.tsx @@ -320,15 +320,20 @@ export const DiskList: Component = (props) => { {(disk) => { - const data = model.getDiskData(disk); - const status = getPhysicalDiskHealthStatus(data); - const hostLabel = getPhysicalDiskHostLabel(data, disk); - const healthSummary = getPhysicalDiskHealthSummary(status); + // Windowed rows retain their owner by resource ID across live + // snapshots. Derive the presentation reactively, not just when + // the row first mounts, so health and identity cannot go stale. + const data = createMemo(() => model.getDiskData(disk)); + const status = createMemo(() => getPhysicalDiskHealthStatus(data())); + const hostLabel = createMemo(() => getPhysicalDiskHostLabel(data(), disk)); + const healthSummary = createMemo(() => getPhysicalDiskHealthSummary(status())); const isSelected = () => model.selectedDisk()?.id === disk.id; - const summarySeriesId = resolvePhysicalDiskMetricResourceId(disk); + const summarySeriesId = createMemo(() => resolvePhysicalDiskMetricResourceId(disk)); const isSummaryHighlighted = () => - props.highlightedSummarySeriesId === summarySeriesId; - const detailControlsId = buildSummaryDisclosureControlsId(summarySeriesId); + props.highlightedSummarySeriesId === summarySeriesId(); + const detailControlsId = createMemo(() => + buildSummaryDisclosureControlsId(summarySeriesId()), + ); const interactiveRowHandlers = createSummaryInteractiveRowPreviewHandlers({ onPreview: () => props.onHoverChange?.(disk.id), onPreviewClear: () => props.onHoverChange?.(null), @@ -338,7 +343,7 @@ export const DiskList: Component = (props) => { <> = (props) => {
model.toggleSelectedDisk(disk)} onPreviewClear={() => props.onHoverChange?.(null)} /> - {data.model || 'Unknown Disk'} + {data().model || 'Unknown Disk'}
@@ -379,11 +384,11 @@ export const DiskList: Component = (props) => { data-storage-column="device" > —} > - - {data.devPath} + + {data().devPath} @@ -393,11 +398,11 @@ export const DiskList: Component = (props) => { data-storage-column="host" > —} > - - {hostLabel} + + {hostLabel()} @@ -407,14 +412,14 @@ export const DiskList: Component = (props) => { data-storage-column="role" > —} > - {getPhysicalDiskRoleLabel(data)} + {getPhysicalDiskRoleLabel(data())} @@ -424,14 +429,14 @@ export const DiskList: Component = (props) => { data-storage-column="parent" > —} > - {getPhysicalDiskParentLabel(data)} + {getPhysicalDiskParentLabel(data())} @@ -441,12 +446,15 @@ export const DiskList: Component = (props) => { data-storage-column="health" >
- - {status.label} + + {status().label} - - - {healthSummary} + + + {healthSummary()}
@@ -457,13 +465,13 @@ export const DiskList: Component = (props) => { data-storage-column="life" > —} > - {getPhysicalDiskLifeLabel(data)} + {getPhysicalDiskLifeLabel(data())} @@ -473,17 +481,17 @@ export const DiskList: Component = (props) => { data-storage-column="temp" > 0} + when={data().temperature > 0} fallback={—} > - {formatTemperature(data.temperature)} + {formatTemperature(data().temperature)} @@ -493,7 +501,7 @@ export const DiskList: Component = (props) => { data-storage-column="size" > 0} + when={data().size > 0} fallback={ = (props) => { } > - {formatBytes(data.size)} + {formatBytes(data().size)} - + diff --git a/frontend-modern/src/components/Storage/__tests__/DiskList.test.tsx b/frontend-modern/src/components/Storage/__tests__/DiskList.test.tsx index 0956cb9176..c8b0ad9432 100644 --- a/frontend-modern/src/components/Storage/__tests__/DiskList.test.tsx +++ b/frontend-modern/src/components/Storage/__tests__/DiskList.test.tsx @@ -1,5 +1,6 @@ -import { cleanup, fireEvent, render, screen } from '@solidjs/testing-library'; +import { cleanup, fireEvent, render, screen, waitFor, within } from '@solidjs/testing-library'; import { createSignal } from 'solid-js'; +import { createStore, reconcile } from 'solid-js/store'; import { afterEach, describe, expect, it, vi } from 'vitest'; import type { Resource } from '@/types/resource'; import { DiskList } from '@/components/Storage/DiskList'; @@ -83,6 +84,159 @@ describe('DiskList', () => { cleanup(); }); + it('refreshes keyed disk rows without losing the expanded detail or keyboard focus', async () => { + const initial = buildDisk('sda', 'tower', { diskType: 'ssd', wearout: 96 }); + const [disks, setDisks] = createSignal([initial]); + const [selectedDiskId, setSelectedDiskId] = createSignal('sda'); + const view = render(() => ( + + )); + const row = view.container.querySelector('[data-row-id="sda"]')!; + const detail = screen.getByTestId('disk-detail'); + const disclosure = within(row as HTMLElement).getByRole('button'); + disclosure.focus(); + const initialControls = disclosure.getAttribute('aria-controls'); + + setDisks([ + { + ...buildDisk('sda', 'archive', { + model: 'Replacement SSD', + devPath: '/dev/sdz', + diskType: 'ssd', + wearout: 4, + sizeBytes: 4_000_000_000_000, + temperature: 63, + storageRole: 'cache_pool', + storageGroup: 'Archive Pool', + health: 'FAILED', + risk: { + level: 'critical', + reasons: [{ code: 'smart-failed', severity: 'critical', summary: 'SMART failed.' }], + }, + }), + metricsTarget: { resourceType: 'disk', resourceId: 'disk:archive:sdz' }, + }, + ]); + + await waitFor(() => + expect(within(row as HTMLElement).getByText('Replace Now')).toBeInTheDocument(), + ); + for (const text of [ + 'Replacement SSD', + '/dev/sdz', + 'archive', + 'Cache Pool', + 'Archive Pool', + 'SMART failed.', + '4%', + '63°C', + ]) { + expect(within(row as HTMLElement).getByText(text)).toBeInTheDocument(); + } + expect(within(row as HTMLElement).getByText('3.64 TB')).toBeInTheDocument(); + expect(screen.queryByText('Healthy')).not.toBeInTheDocument(); + expect(screen.queryByText('96%')).not.toBeInTheDocument(); + expect(view.container.querySelector('[data-row-id="sda"]')).toBe(row); + expect(screen.getByTestId('disk-detail')).toBe(detail); + expect(document.activeElement).toBe(disclosure); + expect(disclosure.getAttribute('aria-label')).toContain('Replacement SSD'); + expect(row).toHaveAttribute('data-summary-series-id', 'disk:archive:sdz'); + expect(disclosure.getAttribute('aria-controls')).not.toBe(initialControls); + expect( + view.container.querySelector('[data-inline-detail-for="disk:archive:sdz"]'), + ).not.toBeNull(); + expect(document.getElementById(disclosure.getAttribute('aria-controls')!)).not.toBeNull(); + }); + + it('removes obsolete readings and fault styling when a keyed snapshot stops reporting them', async () => { + const [disks, setDisks] = createSignal([ + buildDisk('sda', 'tower', { + health: 'FAILED', + diskType: 'ssd', + wearout: 4, + temperature: 63, + risk: { + level: 'critical', + reasons: [{ code: 'smart-failed', severity: 'critical', summary: 'SMART failed.' }], + }, + }), + ]); + const view = render(() => ( + {}} + /> + )); + const row = view.container.querySelector('[data-row-id="sda"]')!; + expect(within(row as HTMLElement).getByText('Replace Now')).toHaveClass('text-red-700'); + setDisks([ + buildDisk('sda', 'tower', { + health: 'UNKNOWN', + diskType: '', + wearout: -1, + temperature: 0, + sizeBytes: 0, + storageRole: '', + storageGroup: '', + model: '', + devPath: '', + }), + ]); + await waitFor(() => + expect(within(row as HTMLElement).getByText('Unknown')).toBeInTheDocument(), + ); + for (const text of ['Replace Now', 'SMART failed.', '4%', '63°C', 'Parity', 'Tower Array']) { + expect(within(row as HTMLElement).queryByText(text)).not.toBeInTheDocument(); + } + expect(within(row as HTMLElement).getByText('Unknown')).not.toHaveClass('text-red-700'); + expect(within(row as HTMLElement).getByText('sda')).toBeInTheDocument(); + for (const column of ['temp', 'life', 'size', 'device', 'role', 'parent']) { + expect(row.querySelector(`td[data-storage-column="${column}"]`)).toHaveTextContent('—'); + } + expect(view.container.querySelector('[data-row-id="sda"]')).toBe(row); + }); + + it('keeps the attention filter and row health consistent during live updates and recovery', async () => { + const [disks, setDisks] = createStore({ items: [buildDisk('sda', 'tower')] }); + const [healthFilter, setHealthFilter] = createSignal<'all' | 'attention'>('all'); + const view = render(() => ( + {}} + /> + )); + const row = view.container.querySelector('[data-row-id="sda"]'); + setDisks( + 'items', + reconcile([buildDisk('sda', 'tower', { smart: { pendingSectors: 2 }, temperature: 52 })]), + ); + await waitFor(() => expect(screen.getByText('Needs Attention')).toBeInTheDocument()); + expect(view.container.querySelector('[data-row-id="sda"]')).toBe(row); + setHealthFilter('attention'); + expect(screen.getByText('Needs Attention')).toBeInTheDocument(); + setDisks('items', reconcile([buildDisk('sda', 'tower')])); + await waitFor(() => expect(screen.getByText('No disks need attention')).toBeInTheDocument()); + setHealthFilter('all'); + await waitFor(() => expect(screen.getByText('Healthy')).toBeInTheDocument()); + expect(screen.queryByText('52°C')).not.toBeInTheDocument(); + }); + it('renders physical disks in a single-line operational grid', () => { renderDiskList({ disks: [ diff --git a/internal/monitoring/truenas_poller.go b/internal/monitoring/truenas_poller.go index 1f18061c19..51bf22ffc9 100644 --- a/internal/monitoring/truenas_poller.go +++ b/internal/monitoring/truenas_poller.go @@ -1540,8 +1540,8 @@ func classifyTrueNASError(err error, connectionID string) *internalerrors.Monito } } // Transport-level errors: timeout takes precedence over generic connection failures. - var urlErr *url.Error - if (errors.As(err, &urlErr) && urlErr.Timeout()) || errors.Is(err, context.DeadlineExceeded) { + var netErr net.Error + if (errors.As(err, &netErr) && netErr.Timeout()) || errors.Is(err, context.DeadlineExceeded) { errType = internalerrors.ErrorTypeTimeout } else { var netOpErr *net.OpError diff --git a/internal/monitoring/truenas_poller_test.go b/internal/monitoring/truenas_poller_test.go index bd99a72e19..9074c9cdd9 100644 --- a/internal/monitoring/truenas_poller_test.go +++ b/internal/monitoring/truenas_poller_test.go @@ -17,6 +17,7 @@ import ( "testing" "time" + "github.com/gorilla/websocket" "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/internal/truenas" @@ -1815,6 +1816,14 @@ func TestClassifyTrueNASError(t *testing.T) { expectedType: "timeout", expectedRetry: true, }, + { + name: "WebSocket socket deadline classifies as timeout", + err: &truenas.RPCTransportError{Method: "system.info", Phase: "read", Err: &net.OpError{ + Op: "read", Net: "tcp", Err: os.ErrDeadlineExceeded, + }}, + expectedType: "timeout", + expectedRetry: true, + }, { name: "net.OpError classifies as connection", err: &net.OpError{Op: "dial", Net: "tcp", Addr: nil, Err: fmt.Errorf("connection refused")}, @@ -1866,6 +1875,146 @@ func TestClassifyTrueNASError(t *testing.T) { } } +func TestTrueNASPollerUnresponsiveRPCDoesNotFreezeOtherConnections(t *testing.T) { + var stalled atomic.Bool + release := make(chan struct{}) + newServer := func(hostname string, mayStall bool) *httptest.Server { + upgrader := websocket.Upgrader{} + return 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() + for { + var request struct { + ID int64 `json:"id"` + Method string `json:"method"` + Params []json.RawMessage `json:"params"` + } + if err := conn.ReadJSON(&request); err != nil { + return + } + var result any = []any{} + switch request.Method { + case "auth.login_ex": + result = map[string]any{"response_type": "SUCCESS"} + case "system.info": + if mayStall && stalled.Load() { + <-release + } + result = map[string]any{"hostname": hostname, "version": "TrueNAS-SCALE-25.10.7", "system_serial": hostname} + case "core.subscribe": + result = "fixture-realtime" + case "core.unsubscribe": + result = nil + } + if err := conn.WriteJSON(map[string]any{"jsonrpc": "2.0", "id": request.ID, "result": result}); err != nil { + return + } + if request.Method == "core.subscribe" { + var event string + if len(request.Params) != 1 || json.Unmarshal(request.Params[0], &event) != nil { + t.Error("subscription did not supply one event") + return + } + var fields any = map[string]any{"cpu": map[string]any{"usage": 12}} + if strings.HasPrefix(event, "app.stats:") { + fields = []any{} + } + if err := conn.WriteJSON(map[string]any{ + "jsonrpc": "2.0", "method": "collection_update", + "params": map[string]any{"collection": event, "fields": fields}, + }); err != nil { + return + } + } + } + })) + } + brokenServer := newServer("timeout-nas", true) + healthyServer := newServer("healthy-nas", false) + t.Cleanup(brokenServer.Close) + t.Cleanup(healthyServer.Close) + poller := NewTrueNASPoller(nil, 0, nil) + instances := []config.TrueNASInstance{ + {ID: "timeout-connection", Host: brokenServer.URL, Enabled: true}, + {ID: "healthy-connection", Host: healthyServer.URL, Enabled: true}, + } + poller.providersByOrg["default"] = make(map[string]*truenas.Provider) + poller.configsByOrg["default"] = make(map[string]config.TrueNASInstance) + for i, instance := range instances { + server := []*httptest.Server{brokenServer, healthyServer}[i] + client, err := truenas.NewClient(truenas.ClientConfig{ + Host: server.URL, APIKey: "fixture-key", Username: "fixture-user", Timeout: 200 * time.Millisecond, + InsecureSkipVerify: true, Fingerprint: fmt.Sprintf("%x", sha256.Sum256(server.Certificate().Raw)), + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(client.Close) + poller.providersByOrg["default"][instance.ID] = truenas.NewLiveProviderForConnection(&truenas.APIFetcher{Client: client}, instance.ID) + poller.configsByOrg["default"][instance.ID] = instance + } + t.Cleanup(func() { close(release) }) + due := func() { + for _, instance := range instances { + poller.ensureConnectionRuntimeStatusLocked("default", instance.ID).nextPollAt = time.Now().Add(-time.Second) + } + } + poller.pollAll(context.Background()) + before := poller.ConnectionSummaries("default", instances) + for _, instance := range instances { + if before[instance.ID].Poll.LastSuccessAt == nil { + t.Fatal("initial successful poll was not established") + } + } + stalled.Store(true) + due() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan struct{}) + go func() { poller.pollAll(ctx); close(done) }() + select { + case <-done: + case <-time.After(time.Second): + t.Error("one unresponsive RPC froze the shared poll cycle") + cancel() // Safely terminate the pre-repair adverse control. + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("poll cycle did not stop after cancellation") + } + } + after := poller.ConnectionSummaries("default", instances) + broken, healthy := after[instances[0].ID], after[instances[1].ID] + if broken.Poll.LastError == nil || broken.Poll.LastError.Category != "timeout" || broken.Poll.ConsecutiveFailures != 1 || !broken.Poll.LastSuccessAt.Equal(*before[instances[0].ID].Poll.LastSuccessAt) { + t.Fatalf("failed connection lost truthful timeout or previous success: %+v", broken) + } + if healthy.Poll.LastError != nil || !healthy.Poll.LastSuccessAt.After(*before[instances[1].ID].Poll.LastSuccessAt) { + t.Fatalf("healthy connection did not continue polling: %+v", healthy) + } + if !hasTrueNASHostForOrg(poller, "default", "timeout-nas") || !hasTrueNASHostForOrg(poller, "default", "healthy-nas") { + t.Fatal("timeout discarded a cached host identity") + } + status := poller.statusByOrg["default"][instances[0].ID] + if !status.nextPollAt.Equal(status.lastAttemptAt.Add(defaultTrueNASPollInterval)) { + t.Fatal("timeout changed completion-based failure backoff") + } + stalled.Store(false) + due() + poller.pollAll(context.Background()) + recovered := poller.ConnectionSummaries("default", instances)[instances[0].ID] + if recovered.Poll.LastError != nil || recovered.Poll.ConsecutiveFailures != 0 || !recovered.Poll.LastSuccessAt.After(*broken.Poll.LastSuccessAt) || recovered.Transport == nil || !recovered.Transport.Connected || recovered.Transport.Mode != truenas.TransportJSONRPC { + t.Fatalf("next poll did not recover the original connection over RPC: %+v", recovered) + } +} + type trueNASMockServer struct { server *httptest.Server requests atomic.Int64 diff --git a/internal/truenas/client.go b/internal/truenas/client.go index a6ff48f5a6..6ad3e11046 100644 --- a/internal/truenas/client.go +++ b/internal/truenas/client.go @@ -66,7 +66,7 @@ type Client struct { baseURL string rpcURL string - rpcMu sync.Mutex + rpcMu rpcSessionMutex rpc *trueNASRPCClient mode TransportMode closed bool @@ -251,7 +251,7 @@ func (c *Client) GetSystemTelemetry(ctx context.Context) (*SystemInfo, error) { var telemetry *SystemInfo var temperatures map[string]float64 - err = c.withRPC(ctx, func(rpc *trueNASRPCClient) error { + err = c.withRPC(ctx, func(ctx context.Context, rpc *trueNASRPCClient) error { temperatures, _ = rpc.getSystemTemperatures(ctx) subscriptionName := fmt.Sprintf("reporting.realtime:{\"interval\":%d}", defaultRealtimeIntervalSeconds) subscriptionID, err := rpc.subscribe(ctx, subscriptionName) @@ -506,7 +506,7 @@ func (c *Client) GetSystemMetricHistory(ctx context.Context, duration time.Durat } var history *SystemMetricHistory - err = c.withRPC(ctx, func(rpc *trueNASRPCClient) error { + err = c.withRPC(ctx, func(ctx context.Context, rpc *trueNASRPCClient) error { var err error history, err = rpc.getSystemMetricHistory(ctx, duration) return err @@ -1332,7 +1332,7 @@ func (c *Client) GetDiskTemperatureHistory(ctx context.Context, identifiers []st } var history map[string][]TimeSeriesPoint - err := c.withRPC(ctx, func(rpc *trueNASRPCClient) error { + err := c.withRPC(ctx, func(ctx context.Context, rpc *trueNASRPCClient) error { var err error history, err = rpc.getDiskTemperatureHistory(ctx, identifiers, duration) return err @@ -1434,7 +1434,7 @@ func (c *Client) getDiskTemperaturesFromReporting(ctx context.Context, identifie } var temperatures map[string]int - err := c.withRPC(ctx, func(rpc *trueNASRPCClient) error { + err := c.withRPC(ctx, func(ctx context.Context, rpc *trueNASRPCClient) error { var err error temperatures, err = rpc.getDiskTemperatures(ctx, identifiers) return err @@ -1449,7 +1449,7 @@ func (c *Client) getDiskTemperatureAggregates(ctx context.Context, identifiers [ } var aggregates map[string]DiskTemperatureAggregate - err := c.withRPC(ctx, func(rpc *trueNASRPCClient) error { + err := c.withRPC(ctx, func(ctx context.Context, rpc *trueNASRPCClient) error { var err error aggregates, err = rpc.getDiskTemperatureAggregates(ctx, identifiers, windowDays) return err @@ -1805,7 +1805,7 @@ func (c *Client) parseAppsWithStats(ctx context.Context, response []map[string]a // failure. func (c *Client) GetAppStats(ctx context.Context) (map[string]AppStats, error) { var stats map[string]AppStats - err := c.withRPC(ctx, func(rpc *trueNASRPCClient) error { + err := c.withRPC(ctx, func(ctx context.Context, rpc *trueNASRPCClient) error { subscriptionName := fmt.Sprintf("app.stats:{\"interval\":%d}", defaultAppStatsIntervalSeconds) subscriptionID, err := rpc.subscribe(ctx, subscriptionName) if err != nil { @@ -1861,7 +1861,7 @@ func (c *Client) GetAppLogs(ctx context.Context, appName, containerID string, ta } subscriptionName := fmt.Sprintf("app.container_log_follow:%s", string(subscriptionJSON)) var lines []AppLogLine - err = c.withRPC(ctx, func(rpc *trueNASRPCClient) error { + err = c.withRPC(ctx, func(ctx context.Context, rpc *trueNASRPCClient) error { subscriptionID, err := rpc.subscribe(ctx, subscriptionName) if err != nil { return err @@ -2615,6 +2615,11 @@ func (c *trueNASRPCClient) call(ctx context.Context, method string, params any, if c == nil || c.conn == nil { return fmt.Errorf("truenas rpc connection is nil") } + // A cancelled waiter must not dispatch an action or poison a healthy + // session by setting an already-expired socket deadline. + if err := ctx.Err(); err != nil { + return err + } stopContext := c.armContext(ctx) defer stopContext() @@ -2759,6 +2764,15 @@ func (c *trueNASRPCClient) readAppLogEvents(ctx context.Context, tailLines int) var message trueNASRPCResponse if err := c.conn.ReadJSON(&message); err != nil { if isTimeoutError(err) { + // The normal idle window completes a bounded tail. A caller or + // operation budget expiring first is a failure, not evidence that + // the log stream successfully returned no data. + if ctxErr := ctx.Err(); ctxErr != nil { + return nil, false, &RPCTransportError{Method: "app.container_log_follow", Phase: "read", Err: ctxErr} + } + if ctxDeadline, ok := ctx.Deadline(); ok && !time.Now().Before(ctxDeadline) { + return nil, false, &RPCTransportError{Method: "app.container_log_follow", Phase: "read", Err: context.DeadlineExceeded} + } // Gorilla WebSocket documents a timed-out read as terminal for // the connection. Preserve the collected log data, then make // the caller discard this stream session instead of reusing it. diff --git a/internal/truenas/transport.go b/internal/truenas/transport.go index 68ef382ce5..16f1093e0f 100644 --- a/internal/truenas/transport.go +++ b/internal/truenas/transport.go @@ -8,6 +8,7 @@ import ( "net/http" "strconv" "strings" + "sync" "time" "github.com/gorilla/websocket" @@ -15,6 +16,49 @@ import ( var errRPCStreamSessionConsumed = errors.New("truenas rpc stream session cannot be reused") +// rpcSessionMutex serializes the sole WebSocket reader/writer, while allowing +// a caller to abandon its wait without interrupting the current session owner. +// Its zero value is usable, including by Client.Close and protocol fixtures. +type rpcSessionMutex struct { + once sync.Once + token chan struct{} +} + +func (m *rpcSessionMutex) LockContext(ctx context.Context) error { + m.once.Do(func() { m.token = make(chan struct{}, 1) }) + if err := ctx.Err(); err != nil { + return err + } + select { + case <-ctx.Done(): + return ctx.Err() + case m.token <- struct{}{}: + if err := ctx.Err(); err != nil { + m.Unlock() + return err + } + return nil + } +} + +func (m *rpcSessionMutex) Lock() { _ = m.LockContext(context.Background()) } +func (m *rpcSessionMutex) Unlock() { <-m.token } + +// rpcOperationContext applies the same configured timeout as HTTP requests to +// a serialized RPC/negotiation operation: lock wait, inline authentication, exchange or +// subscription, and its one permitted read retry share a single budget. A +// shorter caller deadline is never extended. Keepalive has its own lifetime. +func (c *Client) rpcOperationContext(ctx context.Context) (context.Context, context.CancelFunc) { + if ctx == nil { + ctx = context.Background() + } + timeout := c.config.Timeout + if timeout <= 0 { + timeout = defaultHTTPTimeout + } + return context.WithTimeout(ctx, timeout) +} + type discardRPCSessionError struct { err error } @@ -170,7 +214,11 @@ func (c *Client) ensureTransport(ctx context.Context) (TransportMode, error) { if c == nil { return TransportUnknown, fmt.Errorf("truenas client is nil") } - c.rpcMu.Lock() + ctx, cancel := c.rpcOperationContext(ctx) + defer cancel() + if err := c.rpcMu.LockContext(ctx); err != nil { + return TransportUnknown, err + } defer c.rpcMu.Unlock() return c.ensureTransportLocked(ctx) } @@ -286,11 +334,15 @@ func (c *Client) callRPC(ctx context.Context, method string, params any, result return c.callRPCWithRetry(ctx, method, params, result, true) } -func (c *Client) withRPC(ctx context.Context, operation func(*trueNASRPCClient) error) error { +func (c *Client) withRPC(ctx context.Context, operation func(context.Context, *trueNASRPCClient) error) error { if c == nil { return fmt.Errorf("truenas client is nil") } - c.rpcMu.Lock() + ctx, cancel := c.rpcOperationContext(ctx) + defer cancel() + if err := c.rpcMu.LockContext(ctx); err != nil { + return err + } defer c.rpcMu.Unlock() mode, err := c.ensureTransportLocked(ctx) @@ -300,7 +352,7 @@ func (c *Client) withRPC(ctx context.Context, operation func(*trueNASRPCClient) if mode != TransportJSONRPC || c.rpc == nil { return fmt.Errorf("truenas JSON-RPC operation is unavailable over negotiated transport %s", mode) } - if err := operation(c.rpc); err != nil { + if err := operation(ctx, c.rpc); err != nil { if errors.Is(err, errRPCStreamSessionConsumed) { c.closeRPCLocked() c.reconnect = 0 @@ -337,7 +389,7 @@ func (c *Client) withRPC(ctx context.Context, operation func(*trueNASRPCClient) status.LastConnectedAt = &now status.LastError = "" }) - if err := operation(c.rpc); err != nil { + if err := operation(ctx, c.rpc); err != nil { var secondTransportErr *RPCTransportError if errors.As(err, &secondTransportErr) { c.closeRPCLocked() @@ -360,7 +412,11 @@ func (c *Client) callRPCWithRetry(ctx context.Context, method string, params any if c == nil { return fmt.Errorf("truenas client is nil") } - c.rpcMu.Lock() + ctx, cancel := c.rpcOperationContext(ctx) + defer cancel() + if err := c.rpcMu.LockContext(ctx); err != nil { + return err + } defer c.rpcMu.Unlock() mode, err := c.ensureTransportLocked(ctx) diff --git a/internal/truenas/transport_test.go b/internal/truenas/transport_test.go index 34d16e2d25..00d284e872 100644 --- a/internal/truenas/transport_test.go +++ b/internal/truenas/transport_test.go @@ -1095,3 +1095,327 @@ func TestRPCSessionKeepaliveConcurrentCallsAndShutdown(t *testing.T) { t.Fatal("keepalive did not exit after socket failure") } } + +// The production poll loop has a cancellation context, not a per-request +// deadline. A silent peer must still be bounded by ClientConfig.Timeout. These +// fixtures never infer an appliance's reason for failing to reply. +func awaitConfiguredRPCBudget(t *testing.T, operation func(context.Context) error) error { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- operation(ctx) }() + select { + case err := <-done: + return err + case <-time.After(time.Second): + t.Error("RPC operation ignored the configured timeout") + cancel() // Also lets the pre-repair adverse control terminate safely. + select { + case err := <-done: + return err + case <-time.After(5 * time.Second): + t.Fatal("cancelled RPC operation did not stop") + return nil + } + } +} + +func assertRPCTimeout(t *testing.T, err error) { + t.Helper() + if err == nil || (!errors.Is(err, context.DeadlineExceeded) && !isTimeoutError(err)) { + t.Errorf("expected a timeout, got %v", err) + } +} + +func TestJSONRPCConfiguredTimeoutBoundsHandshake(t *testing.T) { + release := make(chan struct{}) + var restRequests atomic.Int32 + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/current" { + restRequests.Add(1) + } + <-release // TCP/TLS succeeds, but no WebSocket handshake response follows. + })) + t.Cleanup(server.Close) + client := protocolFixtureClient(t, server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user", Timeout: 200 * time.Millisecond}) + t.Cleanup(func() { close(release) }) + err := awaitConfiguredRPCBudget(t, client.TestConnection) + assertRPCTimeout(t, err) + var handshake *RPCHandshakeError + if !errors.As(err, &handshake) || client.TransportStatus().Connected || restRequests.Load() != 0 { + t.Fatalf("handshake timeout lost its type/status or downgraded to REST: %v", err) + } +} + +func TestJSONRPCConfiguredTimeoutBoundsAuthentication(t *testing.T) { + release := make(chan struct{}) + var logins atomic.Int32 + fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply { + if request.Method == "auth.login_ex" { + logins.Add(1) + <-release + } + return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS"}} + }, nil) + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user", Timeout: 200 * time.Millisecond}) + t.Cleanup(func() { close(release) }) + err := awaitConfiguredRPCBudget(t, client.TestConnection) + assertRPCTimeout(t, err) + if fixture.sessions.Load() != 1 || logins.Load() != 1 || fixture.restRequests.Load() != 0 || client.TransportStatus().Connected { + t.Fatalf("silent authentication retried, downgraded or appeared connected: %+v", client.TransportStatus()) + } +} + +func TestJSONRPCConfiguredTimeoutBoundsReadAndRecovers(t *testing.T) { + release := make(chan struct{}) + var reads 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 == "system.info" && reads.Add(1) == 1 { + <-release + } + return protocolFixtureReply{result: map[string]any{"hostname": "timeout-fixture", "version": "TrueNAS-SCALE-25.10.7"}} + }, nil) + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user", Timeout: 200 * time.Millisecond}) + t.Cleanup(func() { close(release) }) + err := awaitConfiguredRPCBudget(t, client.TestConnection) + assertRPCTimeout(t, err) + if fixture.sessions.Load() != 1 || reads.Load() != 1 || client.TransportStatus().Connected { + t.Fatal("expired operation gained a retry budget or retained its timed-out session") + } + if err := client.TestConnection(context.Background()); err != nil { + t.Fatalf("subsequent operation did not recover: %v", err) + } + if fixture.sessions.Load() != 2 || reads.Load() != 2 || fixture.restRequests.Load() != 0 || !client.TransportStatus().Connected { + t.Fatalf("subsequent operation did not negotiate one fresh RPC session: %+v", client.TransportStatus()) + } +} + +func TestJSONRPCConfiguredTimeoutIncludesReadRetryBackoff(t *testing.T) { + release := make(chan struct{}) + var reads 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 reads.Add(1) == 1 { + // The 100 ms reconnect backoff cannot fit in the remaining part of + // a 200 ms operation. It must not create a new deadline on retry. + time.Sleep(150 * time.Millisecond) + return protocolFixtureReply{close: true} + } + <-release + return protocolFixtureReply{result: nil} + }, nil) + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user", Timeout: 200 * time.Millisecond}) + t.Cleanup(func() { close(release) }) + err := awaitConfiguredRPCBudget(t, client.TestConnection) + assertRPCTimeout(t, err) + if fixture.sessions.Load() != 1 || reads.Load() != 1 || fixture.restRequests.Load() != 0 { + t.Fatal("read retry restarted an exhausted operation budget") + } +} + +func TestJSONRPCConfiguredTimeoutDoesNotExtendCallerDeadline(t *testing.T) { + release := make(chan struct{}) + fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply { + if request.Method == "auth.login_ex" { + return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS"}} + } + <-release + return protocolFixtureReply{result: nil} + }, nil) + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user", Timeout: 5 * time.Second}) + t.Cleanup(func() { close(release) }) + err := awaitConfiguredRPCBudget(t, func(parent context.Context) error { + ctx, cancel := context.WithTimeout(parent, 100*time.Millisecond) + defer cancel() + return client.TestConnection(ctx) + }) + assertRPCTimeout(t, err) + if fixture.sessions.Load() != 1 || fixture.restRequests.Load() != 0 { + t.Fatal("caller deadline gained a retry or REST downgrade") + } +} + +func TestJSONRPCConfiguredTimeoutDoesNotReplayAction(t *testing.T) { + release := make(chan struct{}) + var actions 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 == "app.start" { + actions.Add(1) + <-release + } + return protocolFixtureReply{result: nil} + }, nil) + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user", Timeout: 200 * time.Millisecond}) + t.Cleanup(func() { close(release) }) + err := awaitConfiguredRPCBudget(t, func(ctx context.Context) error { return client.StartApp(ctx, "fixture-app") }) + assertRPCTimeout(t, err) + if !strings.Contains(err.Error(), "outcome is unknown") || actions.Load() != 1 || fixture.sessions.Load() != 1 || fixture.restRequests.Load() != 0 { + t.Fatalf("timed-out action was replayed or lost its unknown outcome: %v", err) + } +} + +func TestJSONRPCConfiguredTimeoutBoundsSubscriptions(t *testing.T) { + for _, stream := range []string{"telemetry", "app-stats", "app-logs"} { + t.Run(stream, func(t *testing.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: "fixture-subscription"} // No stream event ever arrives. + default: + return protocolFixtureReply{result: []any{}} + } + }, nil) + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user", Timeout: 200 * time.Millisecond}) + err := awaitConfiguredRPCBudget(t, func(ctx context.Context) error { + switch stream { + case "telemetry": + _, err := client.GetSystemTelemetry(ctx) + return err + case "app-stats": + _, err := client.GetAppStats(ctx) + return err + default: + _, err := client.GetAppLogs(ctx, "fixture-app", "fixture-container", 100) + return err + } + }) + assertRPCTimeout(t, err) + if fixture.sessions.Load() != 1 || fixture.restRequests.Load() != 0 || client.TransportStatus().Connected { + t.Fatal("expired stream was retried, downgraded or left reusable") + } + }) + } +} + +func TestJSONRPCWaitingDeadlineDoesNotInterruptSessionOwner(t *testing.T) { + release := make(chan struct{}) + started := make(chan struct{}) + var reads 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 reads.Add(1) == 1 { + close(started) + <-release + } + return protocolFixtureReply{result: map[string]any{"hostname": "session-owner"}} + }, nil) + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user", Timeout: 5 * time.Second}) + var released atomic.Bool + unblock := func() { + if released.CompareAndSwap(false, true) { + close(release) + } + } + t.Cleanup(unblock) + owner := make(chan error, 1) + go func() { owner <- client.TestConnection(context.Background()) }() + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("session owner did not dispatch") + } + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + waiter := make(chan error, 1) + go func() { waiter <- client.TestConnection(ctx) }() + select { + case err := <-waiter: + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("waiter error = %v, want caller deadline", err) + } + case <-time.After(time.Second): + t.Error("caller deadline could not abandon the session lock wait") + unblock() // Do not strand the predecessor's uncancellable mutex waiter. + <-waiter + } + if reads.Load() != 1 { + t.Error("cancelled waiter dispatched a request") + } + unblock() + if err := <-owner; err != nil { + t.Fatalf("waiter interrupted the session owner: %v", err) + } + if err := client.TestConnection(context.Background()); err != nil || fixture.sessions.Load() != 1 { + t.Fatalf("cancelled waiter poisoned the shared session: %v, sessions=%d", err, fixture.sessions.Load()) + } +} + +func TestJSONRPCPreCancelledCallDoesNotDispatch(t *testing.T) { + var actions 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 == "app.start" { + actions.Add(1) + } + return protocolFixtureReply{result: map[string]any{"hostname": "cancel-fixture"}} + }, nil) + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user"}) + if err := client.TestConnection(context.Background()); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if err := client.StartApp(ctx, "fixture-app"); !errors.Is(err, context.Canceled) { + t.Fatalf("pre-cancelled action error = %v", err) + } + if err := client.TestConnection(context.Background()); err != nil || fixture.sessions.Load() != 1 || actions.Load() != 0 { + t.Fatalf("pre-cancelled action dispatched or poisoned the session: %v", err) + } +} + +func TestJSONRPCSnapshotContinuesAfterTelemetryTimeout(t *testing.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 "system.info": + return protocolFixtureReply{result: map[string]any{"hostname": "inventory-timeout", "version": "TrueNAS-SCALE-25.10.7", "system_serial": "FIXTURE-1", "physmem": 1024}} + case "core.subscribe": + params, _ := request.Params.([]any) + event, _ := params[0].(string) + if strings.HasPrefix(event, "app.stats:") { + // Only live system telemetry is silent in this fixture. App + // inventory's separate best-effort subscription still replies. + return protocolFixtureReply{result: "empty-app-stats", notifications: []protocolFixtureNotification{{ + method: "collection_update", params: map[string]any{"collection": event, "fields": []any{}}, + }}} + } + return protocolFixtureReply{result: "silent-realtime"} + case "pool.query": + return protocolFixtureReply{result: []map[string]any{{"id": 1, "name": "tank", "status": "ONLINE"}}} + default: + return protocolFixtureReply{result: []any{}} + } + }, nil) + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user", Timeout: 200 * time.Millisecond}) + var snapshot *FixtureSnapshot + err := awaitConfiguredRPCBudget(t, func(ctx context.Context) error { + var err error + snapshot, err = client.FetchSnapshot(ctx) + return err + }) + if err != nil || snapshot == nil || snapshot.System.Hostname != "inventory-timeout" || len(snapshot.Pools) != 1 { + t.Fatalf("optional telemetry timeout prevented usable inventory: %v, %+v", err, snapshot) + } + if snapshot.System.Telemetry == nil || snapshot.System.Telemetry.ErrorCategory != "timeout" || snapshot.System.Telemetry.CPU || snapshot.System.Telemetry.Memory { + t.Fatalf("silent telemetry appeared fresh or lost its timeout: %+v", snapshot.System.Telemetry) + } + if fixture.sessions.Load() != 2 || fixture.restRequests.Load() != 0 { + t.Fatalf("inventory did not recover on RPC after discarding the stream: sessions=%d rest=%d", fixture.sessions.Load(), fixture.restRequests.Load()) + } +}