From 119ac49d10d796db7daec2e58942884d9a965105 Mon Sep 17 00:00:00 2001 From: "courtmanr@gmail.com" Date: Wed, 30 Sep 2026 19:33:16 +0100 Subject: [PATCH] Keep the update progress modal moving when the update stream goes quiet Updating to v6.4.5-rc.5 and then v6.4.5 left the progress modal on "Downloading update... 10%" even though the update finished in the background. The status stream could go silent behind a proxy (Tailscale Serve, nginx buffering) and the modal only advanced on stream events. Backend: the update SSE endpoint now sends the current status on connect, writes events through one ordered writer with an explicit flush, sends a 15s heartbeat and sets X-Accel-Buffering: no. Frontend: a silence watchdog falls back to status polling, the restart phase is entered only on real restart evidence (restarting status, the server going away, or a version change) rather than a single failed poll, and the page never auto-reloads on unconfirmed completion without a pre-update version baseline to compare against. --- .../v6/internal/subsystems/agent-lifecycle.md | 8 + .../v6/internal/subsystems/api-contracts.md | 13 + .../subsystems/deployment-installability.md | 35 ++ .../internal/subsystems/storage-recovery.md | 7 + .../browser-tests/update-progress-modal.cjs | 348 ++++++++++++ .../browser-tests/update-progress-modal.html | 11 + .../browser-tests/update-progress-modal.tsx | 101 ++++ frontend-modern/browser-verification.json | 128 +++-- .../src/components/UpdateProgressModal.tsx | 480 ++++++++++++---- .../__tests__/UpdateProgressModal.test.tsx | 514 ++++++++++++++++++ .../__tests__/updateReadinessModel.test.ts | 49 +- .../src/components/updateReadinessModel.ts | 90 ++- internal/api/updates.go | 16 +- internal/api/updates_test.go | 162 ++++++ internal/updates/manager.go | 8 +- internal/updates/sse.go | 111 ++-- internal/updates/sse_test.go | 93 ++++ 17 files changed, 1985 insertions(+), 189 deletions(-) create mode 100644 frontend-modern/browser-tests/update-progress-modal.cjs create mode 100644 frontend-modern/browser-tests/update-progress-modal.html create mode 100644 frontend-modern/browser-tests/update-progress-modal.tsx create mode 100644 frontend-modern/src/components/__tests__/UpdateProgressModal.test.tsx diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 7e651555d2..69e9c7318f 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -3452,6 +3452,14 @@ Agent` secondary handoff against the live setup wizard instead of relying ## Current State +### Update progress stream delivery + +Server update progress delivery is owned by internal/updates and its API +adapter. Immediate status on connect, ordered flushed events, and the modal's +polling fallback change how progress is observed only. They grant no agent +update, installation or command-execution authority, and a completed server +update is not evidence that any agent was upgraded. + ### Hosted runtimes always mint agent install tokens `POST /api/agent-install-command` mints and embeds an agent install token diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index be2175712d..e1030c0536 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -4579,6 +4579,19 @@ auto-register mutation boundary. ## Current State +### Update progress stream delivery + +`GET /api/updates/stream` answers `text/event-stream` with +`Cache-Control: no-cache, no-transform`, `X-Accel-Buffering: no`, and no +content encoding. The first flushed frames are a `: connected` comment and a +`data:` event carrying the current `UpdateStatus`, so a connecting client never +waits for the next stage change. Later statuses are `data:` events in emission +order, each flushed on write, with `: heartbeat` comments between them. The +payload shape is the same `UpdateStatus` returned by `/api/updates/status`, +which stays the polling fallback for clients whose stream goes quiet. The +stream handler tests in `internal/api/updates_test.go` cover this transport +contract. + ### Hosted install command payloads carry a minted token `POST /api/agent-install-command` returns a non-empty `token` (and, for diff --git a/docs/release-control/v6/internal/subsystems/deployment-installability.md b/docs/release-control/v6/internal/subsystems/deployment-installability.md index fe54ff248f..5d864a6c72 100644 --- a/docs/release-control/v6/internal/subsystems/deployment-installability.md +++ b/docs/release-control/v6/internal/subsystems/deployment-installability.md @@ -2527,6 +2527,41 @@ artifact-selection behaviour. ## Current State +### Update progress stream delivery + +`GET /api/updates/stream` is the in-app updater's progress feed and must never +look like a stalled update. The handler marks the response `text/event-stream` +with `Cache-Control: no-cache, no-transform` and `X-Accel-Buffering: no` so +reverse proxies and compressing intermediaries do not hold events back, and the +gzip middleware keeps excluding event streams. `SSEBroadcaster.AddClient` in +`internal/updates/sse.go` writes the connection preamble and the current cached +status and flushes them before returning, and every later broadcast reaches a +client through that client's single ordered writer under the client lock, so +stages arrive in emission order and no two goroutines write one +ResponseWriter. Write failures remove the client off the writer goroutine so +removal never waits on the broadcaster lock while holding a client lock. +Comment heartbeats run every 15 seconds. `UpdateProgressModal` must not trust +an open-but-silent EventSource. It also polls `/api/updates/status` after +`UPDATE_STREAM_SILENCE_FALLBACK_MS` without a stream message, ignores polled +stages that would move progress backwards or are update-check chatter, keeps +reloading only once `/api/version` reports a different version, and after +`UPDATE_PROGRESS_STALL_TIMEOUT_MS` without movement shows a reload-to-check +state instead of an endless spinner. A failed status poll is never restart +evidence on its own. The modal enters its restart phase only when the backend +reports `restarting` or `completed`, or, unconfirmed, when the stream has +closed and consecutive polls fail after the update reached `applying` or +`restoring`. An unconfirmed restart keeps polling so a later progress or +`error` status still wins, and the same-version reload fallback in +`resolvePostUpdateReload` applies only after a confirmed restart. When the +modal opens before the running version is known it fetches `/api/version` +immediately and also adopts the update store's version once it loads, but +only while no restart signal has been seen. With no baseline it never reloads +on unconfirmed completion, and after confirmed completion it reloads only +once a probe has seen the old process go away, or after the bounded fallback. `internal/updates/sse_test.go`, the +stream handler tests in `internal/api/updates_test.go`, and +`frontend-modern/src/components/__tests__/UpdateProgressModal.test.tsx` are +the proof surface. + ### Provider MSP upgrades move an install to a new release bundle `setup.sh` resolves the image pins in `.env` to digests once, so `upgrade.sh` diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index 1d0f8017ec..fe2083f3b8 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -2640,6 +2640,13 @@ vdev layout is reported` in ## Current State +### Update progress stream delivery + +Update progress delivery changes observation only. Streaming or polling a +`backing-up`, `restoring` or `completed` stage does not create a backup, +perform rollback or establish recovery success. Update history and retained +backups remain the recovery authority, independent of `/api/updates/stream`. + ### Hosted install tokens persist like any other workspace token Hosted client workspaces now mint agent install tokens where they previously diff --git a/frontend-modern/browser-tests/update-progress-modal.cjs b/frontend-modern/browser-tests/update-progress-modal.cjs new file mode 100644 index 0000000000..3be13f22d7 --- /dev/null +++ b/frontend-modern/browser-tests/update-progress-modal.cjs @@ -0,0 +1,348 @@ +// Offline browser proof: the update progress modal keeps moving when the +// update stream goes quiet, and reloads only on real restart evidence. +// +// node frontend-modern/browser-tests/update-progress-modal.cjs +// +// Needs Playwright 1.56.1 resolvable (NODE_PATH works) with its Chromium. +// PULSE_BROWSER_ARTIFACTS picks the screenshot directory and +// PULSE_CHROMIUM_PATH an alternative Chromium executable. Every API answer is +// synthetic and off-origin traffic is aborted; no Pulse server is contacted. +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const path = require('node:path'); +const { chromium } = require('playwright'); + +const PORT = 5208; +const ORIGIN = `http://127.0.0.1:${PORT}`; +const HARNESS = `${ORIGIN}/browser-tests/update-progress-modal.html`; +const OLD = 'v6.4.4'; +const NEW = 'v6.4.5'; + +const status = (stage, progress, message) => ({ + status: stage, + progress, + message, + updatedAt: new Date().toISOString(), +}); +const MESSAGES = { + downloading: 'Downloading update...', + verifying: 'Verifying download...', + extracting: 'Extracting update...', + 'backing-up': 'Backing up current installation...', + applying: 'Applying update...', + restarting: 'Restarting Pulse...', + completed: 'Update completed', + idle: '', +}; +const s = (stage, progress) => status(stage, progress, MESSAGES[stage]); + +const ABORT = Symbol('abort'); +const UNAVAILABLE = Symbol('unavailable'); + +async function openHarness(browser, viewport, server) { + const context = await browser.newContext({ viewport }); + const page = await context.newPage(); + const record = { + pageErrors: [], + documentLoads: 0, + statusRequests: [], + versionRequests: [], + offOrigin: 0, + }; + page.on('pageerror', (error) => record.pageErrors.push(error.message)); + await page.route('**/*', (route) => { + const request = route.request(); + const url = new URL(request.url()); + if (url.origin !== ORIGIN) { + record.offOrigin += 1; + return route.abort(); + } + if (url.pathname === '/browser-tests/update-progress-modal.html') { + record.documentLoads += 1; + return route.continue(); + } + const answer = (value, log) => { + log.push(value === ABORT ? 'unreachable' : value === UNAVAILABLE ? '503' : value); + if (value === ABORT) return route.abort('connectionrefused'); + if (value === UNAVAILABLE) return route.fulfill({ status: 503, body: 'unavailable' }); + return route.fulfill({ json: value }); + }; + if (url.pathname === '/api/updates/status') { + return answer(server.status(), record.statusRequests); + } + if (url.pathname === '/api/version') { + const version = server.version(); + return answer( + typeof version === 'string' ? { version, build: 'release' } : version, + record.versionRequests, + ); + } + if (url.pathname.startsWith('/api/')) return route.abort(); + return route.continue(); + }); + await page.goto(HARNESS, { waitUntil: 'domcontentloaded', timeout: 120_000 }); + await page.waitForFunction(() => window.__updateStream && window.__updateStream.open()); + await page.getByRole('dialog').waitFor(); + return { context, page, record }; +} + +const emit = (page, value) => page.evaluate((v) => window.__updateStream.emit(v), value); +const seen = (page) => page.evaluate(() => JSON.parse(JSON.stringify(window.__seen))); +const progressText = (page, value) => + page.getByRole('dialog').getByText(`${value}%`, { exact: true }); + +async function layoutCheck(page) { + const dimensions = await page.evaluate(() => { + const panel = document.querySelector('[role="dialog"]'); + const rect = panel ? panel.getBoundingClientRect() : null; + return { + scroll: document.documentElement.scrollWidth, + inner: innerWidth, + panelLeft: rect ? Math.round(rect.left) : null, + panelRight: rect ? Math.round(rect.right) : null, + }; + }); + assert.ok(dimensions.scroll <= dimensions.inner + 1, `horizontal overflow ${JSON.stringify(dimensions)}`); + if (dimensions.panelLeft !== null) { + assert.ok(dimensions.panelLeft >= 0 && dimensions.panelRight <= dimensions.inner + 1, + `dialog outside viewport ${JSON.stringify(dimensions)}`); + } + return dimensions; +} + +async function shot(page, artifacts, name, width) { + await page.waitForTimeout(350); // let the dialog entrance settle + const file = path.join(artifacts, `${name}-${width}.png`); + await page.screenshot({ path: file }); + return file; +} + +const scenarios = { + // (a) One downloading event, then a held-open silent stream. + async 'stalled-stream-fallback'(browser, viewport, artifacts) { + const polls = [s('downloading', 35), s('downloading', 70), s('extracting', 80), s('applying', 90)]; + let pollIndex = 0; + const server = { + status: () => polls[Math.min(pollIndex++, polls.length - 1)], + version: () => OLD, + }; + const { context, page, record } = await openHarness(browser, viewport, server); + await emit(page, s('downloading', 10)); + const silentSince = Date.now(); + await progressText(page, 10).waitFor(); + await page.getByText('Downloading update...').waitFor(); + const screenshots = [await shot(page, artifacts, 'a-stream-10pct', viewport.width)]; + await progressText(page, 35).waitFor({ timeout: 15_000 }); + const firstPollAfterMs = Date.now() - silentSince; + await progressText(page, 90).waitFor({ timeout: 20_000 }); + await page.getByText('Applying update...').waitFor(); + const observed = await seen(page); + assert.ok(firstPollAfterMs >= 5_000, `polling began before the silence window (${firstPollAfterMs}ms)`); + assert.deepEqual(observed.progress, [10, 35, 70, 80, 90]); + assert.equal(observed.restarting, false); + assert.equal(await page.evaluate(() => window.__updateStream.open()), true, 'stream stays open alongside polling'); + screenshots.push(await shot(page, artifacts, 'a-polled-90pct', viewport.width)); + const dimensions = await layoutCheck(page); + assert.deepEqual(record.pageErrors, []); + await context.close(); + return { firstPollAfterMs, progress: observed.progress, statusPolls: record.statusRequests.length, dimensions, screenshots }; + }, + + // (b) One failed poll mid-download must not look like a restart. + async 'transient-poll-failure'(browser, viewport, artifacts) { + const polls = [s('downloading', 40), ABORT, s('downloading', 60), s('downloading', 80)]; + let pollIndex = 0; + const server = { + status: () => polls[Math.min(pollIndex++, polls.length - 1)], + version: () => OLD, + }; + const { context, page, record } = await openHarness(browser, viewport, server); + await emit(page, s('downloading', 20)); + await progressText(page, 20).waitFor(); + await page.evaluate(() => window.__updateStream.fail()); // stream closed, polling only + await progressText(page, 80).waitFor({ timeout: 20_000 }); + await page.waitForTimeout(3_000); // one more poll cycle at the same stage + const observed = await seen(page); + assert.ok(record.statusRequests.includes('unreachable'), 'a poll failed'); + assert.equal(observed.restarting, false, 'restart phase entered on a single failed poll'); + assert.deepEqual(observed.progress, [20, 40, 60, 80]); + await page.getByText('Please do not close this window or refresh the page during the update.').waitFor(); + assert.equal(record.documentLoads, 1); + const screenshots = [await shot(page, artifacts, 'b-after-failed-poll-80pct', viewport.width)]; + const dimensions = await layoutCheck(page); + assert.deepEqual(record.pageErrors, []); + await context.close(); + return { statusRequests: record.statusRequests.map((r) => (typeof r === 'string' ? r : `${r.status} ${r.progress}`)), progress: observed.progress, dimensions, screenshots }; + }, + + // (c) restarting status, then the server goes away, then the new version. + async 'restart-evidence-reload'(browser, viewport, artifacts) { + let restartServed = false; + let versionAfterRestart = 0; + const polls = [s('applying', 90), s('restarting', 100)]; + let pollIndex = 0; + const server = { + status: () => { + if (restartServed) return ABORT; + const next = polls[Math.min(pollIndex++, polls.length - 1)]; + if (next.status === 'restarting') restartServed = true; + return next; + }, + version: () => { + if (!restartServed) return OLD; + versionAfterRestart += 1; + return versionAfterRestart <= 2 ? ABORT : NEW; + }, + }; + const { context, page, record } = await openHarness(browser, viewport, server); + await emit(page, s('downloading', 10)); + await progressText(page, 10).waitFor(); + await page.getByText('Pulse is restarting...').waitFor({ timeout: 20_000 }); + const screenshots = [await shot(page, artifacts, 'c-restarting', viewport.width)]; + const restartDimensions = await layoutCheck(page); + await page.getByTestId('reloaded').waitFor({ timeout: 30_000 }); + await page.waitForTimeout(4_000); // no second reload + assert.equal(record.documentLoads, 2, 'reloaded exactly once'); + const afterRestart = record.versionRequests.slice(record.versionRequests.findIndex((v) => v === 'unreachable')); + assert.deepEqual(afterRestart.slice(0, 3), ['unreachable', 'unreachable', { version: NEW, build: 'release' }]); + screenshots.push(await shot(page, artifacts, 'c-reloaded', viewport.width)); + assert.deepEqual(record.pageErrors, []); + await context.close(); + return { + documentLoads: record.documentLoads, + versionRequests: record.versionRequests.map((v) => (typeof v === 'string' ? v : v.version)), + dimensions: restartDimensions, + screenshots, + }; + }, + + // (d) No pre-update version; requests fail late and a new version answers, + // but the backend never confirmed completion. + async 'no-baseline-no-autoreload'(browser, viewport, artifacts) { + let failuresStarted = false; + const polls = [s('applying', 95), ABORT, ABORT, ABORT, ABORT, s('idle', 0)]; + let pollIndex = 0; + const server = { + status: () => { + const next = polls[Math.min(pollIndex++, polls.length - 1)]; + if (next === ABORT) failuresStarted = true; + return next; + }, + version: () => (failuresStarted ? NEW : UNAVAILABLE), + }; + const { context, page, record } = await openHarness(browser, viewport, server); + await emit(page, s('applying', 90)); + await progressText(page, 90).waitFor(); + await page.evaluate(() => window.__updateStream.fail()); + await page.getByText('Pulse is restarting...').waitFor({ timeout: 20_000 }); + // Past the idle answers and several healthy new-version probes. + const deadline = Date.now() + 25_000; + while (Date.now() < deadline && !(record.statusRequests.filter((r) => r.status === 'idle').length >= 2 + && record.versionRequests.filter((v) => v.version === NEW).length >= 3)) { + await page.waitForTimeout(500); + } + await page.waitForTimeout(2_000); + const idleAnswers = record.statusRequests.filter((r) => r.status === 'idle').length; + const newVersionAnswers = record.versionRequests.filter((v) => v.version === NEW).length; + assert.ok(idleAnswers >= 2, `idle answers ${idleAnswers}`); + assert.ok(newVersionAnswers >= 3, `new-version answers ${newVersionAnswers}`); + assert.ok(record.versionRequests.includes('503'), 'baseline probe was unavailable'); + assert.equal(record.documentLoads, 1, 'auto-reloaded without a baseline or confirmation'); + assert.equal(await page.getByTestId('reloaded').count(), 0); + const observed = await seen(page); + assert.equal(observed.completed, false, 'claimed completion without confirmation'); + const screenshots = [await shot(page, artifacts, 'd-unconfirmed-waiting', viewport.width)]; + const dimensions = await layoutCheck(page); + assert.deepEqual(record.pageErrors, []); + await context.close(); + return { documentLoads: record.documentLoads, idleAnswers, newVersionAnswers, dimensions, screenshots }; + }, + + // (e) The stream delivers every stage through completion. + async 'stream-happy-path'(browser, viewport, artifacts) { + let completedEmitted = false; + let versionAfterComplete = 0; + const server = { + status: () => s('applying', 95), + version: () => { + if (!completedEmitted) return OLD; + versionAfterComplete += 1; + if (versionAfterComplete === 1) return OLD; // old process still answering + if (versionAfterComplete === 2) return ABORT; // it exits + return NEW; + }, + }; + const { context, page, record } = await openHarness(browser, viewport, server); + const stages = [ + s('downloading', 10), s('downloading', 45), s('downloading', 80), s('verifying', 85), + s('extracting', 88), s('backing-up', 90), s('applying', 95), + ]; + const screenshots = []; + for (const stage of stages) { + await emit(page, stage); + await progressText(page, stage.progress).waitFor(); + if (stage.status === 'applying') screenshots.push(await shot(page, artifacts, 'e-stream-applying', viewport.width)); + await page.waitForTimeout(300); + } + const midDimensions = await layoutCheck(page); + completedEmitted = true; + await emit(page, s('completed', 100)); + await page.getByText('Pulse is restarting...').waitFor({ timeout: 10_000 }); + screenshots.push(await shot(page, artifacts, 'e-restarting', viewport.width)); + await layoutCheck(page); + await page.getByTestId('reloaded').waitFor({ timeout: 30_000 }); + await page.waitForTimeout(3_000); + assert.equal(record.documentLoads, 2, 'reloaded exactly once'); + assert.equal(record.statusRequests.length, 0, 'fallback polling ran while the stream was live'); + screenshots.push(await shot(page, artifacts, 'e-reloaded', viewport.width)); + assert.deepEqual(record.pageErrors, []); + await context.close(); + return { + documentLoads: record.documentLoads, + statusPolls: record.statusRequests.length, + versionRequests: record.versionRequests.map((v) => (typeof v === 'string' ? v : v.version)), + dimensions: midDimensions, + screenshots, + }; + }, +}; + +(async () => { + const root = path.resolve(__dirname, '..'); + const artifacts = process.env.PULSE_BROWSER_ARTIFACTS || path.join(root, 'browser-tests', 'update-progress-modal-proof'); + fs.mkdirSync(artifacts, { recursive: true }); + process.chdir(root); + 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'), + server: { host: '127.0.0.1', port: PORT, strictPort: true }, + logLevel: 'warn', + }); + let browser; + try { + await server.listen(); + browser = await chromium.launch({ + headless: true, + args: ['--no-sandbox'], + // Optional: a locally cached Chromium when Playwright's own build is absent. + ...(process.env.PULSE_CHROMIUM_PATH ? { executablePath: process.env.PULSE_CHROMIUM_PATH } : {}), + }); + const results = []; + for (const viewport of [{ width: 1365, height: 900 }, { width: 390, height: 844 }]) { + for (const [name, run] of Object.entries(scenarios)) { + const started = Date.now(); + const detail = await run(browser, viewport, artifacts); + results.push({ viewport: `${viewport.width}x${viewport.height}`, scenario: name, result: 'passed', ms: Date.now() - started, ...detail }); + console.error(`passed ${name} @ ${viewport.width}x${viewport.height}`); + } + } + console.log(JSON.stringify({ result: 'passed', browser: browser.version(), playwright: require('playwright/package.json').version, results }, null, 2)); + } finally { + if (browser) await browser.close(); + await server.close(); + } +})().catch((error) => { + console.error(error); + process.exitCode = 1; +}); diff --git a/frontend-modern/browser-tests/update-progress-modal.html b/frontend-modern/browser-tests/update-progress-modal.html new file mode 100644 index 0000000000..e49ffd5654 --- /dev/null +++ b/frontend-modern/browser-tests/update-progress-modal.html @@ -0,0 +1,11 @@ + + + + + + + +
+ + + diff --git a/frontend-modern/browser-tests/update-progress-modal.tsx b/frontend-modern/browser-tests/update-progress-modal.tsx new file mode 100644 index 0000000000..1f09a23324 --- /dev/null +++ b/frontend-modern/browser-tests/update-progress-modal.tsx @@ -0,0 +1,101 @@ +// Mount the production UpdateProgressModal against a scripted update stream. +// HTTP (/api/updates/status, /api/version) is scripted by the Playwright +// runner; the SSE transport is replaced here so the runner can hold the +// stream open and silent, which a fulfilled HTTP response cannot do. +import { render } from 'solid-js/web'; +import { UpdateProgressModal } from '../src/components/UpdateProgressModal'; +import '../src/index.css'; + +type Listener = ((event: MessageEvent | Event) => void) | null; + +class ScriptedEventSource { + static instances: ScriptedEventSource[] = []; + readonly url: string; + readyState = 0; + onopen: Listener = null; + onmessage: Listener = null; + onerror: Listener = null; + + constructor(url: string) { + this.url = url; + ScriptedEventSource.instances.push(this); + setTimeout(() => { + if (this.readyState === 2) return; + this.readyState = 1; + this.onopen?.(new Event('open')); + }, 0); + } + + close() { + this.readyState = 2; + } + + addEventListener() {} + removeEventListener() {} +} + +const live = () => + ScriptedEventSource.instances.filter((instance) => instance.readyState !== 2).at(-1); + +const w = window as unknown as Record; +w.EventSource = ScriptedEventSource; +w.__updateStream = { + emit(status: unknown) { + const stream = live(); + if (!stream) return false; + stream.onmessage?.(new MessageEvent('message', { data: JSON.stringify(status) })); + return true; + }, + fail() { + const stream = live(); + if (!stream) return false; + stream.onerror?.(new Event('error')); + return true; + }, + opened: () => ScriptedEventSource.instances.length, + open: () => Boolean(live()), +}; + +// Everything the modal ever showed, so a transient restart claim that was +// later corrected still fails the run. +const seen = { restarting: false, completed: false, progress: [] as number[] }; +w.__seen = seen; +new MutationObserver(() => { + const text = document.body.innerText; + if (text.includes('Pulse is restarting')) seen.restarting = true; + if (text.includes('Update Completed Successfully')) seen.completed = true; + const match = /Progress\s+(\d+)%/.exec(text); + if (match) { + const value = Number(match[1]); + if (seen.progress.at(-1) !== value) seen.progress.push(value); + } +}).observe(document.body, { subtree: true, childList: true, characterData: true }); + +let boots = 1; +try { + boots = Number(sessionStorage.getItem('update-progress-boots') || '0') + 1; + sessionStorage.setItem('update-progress-boots', String(boots)); +} catch { + // Storage unavailable: treat as a first boot. +} + +render( + () => + boots > 1 ? ( +
+

Reloaded after update (boot {boots})

+
+ ) : ( +
+

Update progress harness

+ undefined} + onViewHistory={() => undefined} + connected={() => true} + reconnecting={() => false} + /> +
+ ), + document.getElementById('root')!, +); diff --git a/frontend-modern/browser-verification.json b/frontend-modern/browser-verification.json index 46f3f9af25..719decc0b4 100644 --- a/frontend-modern/browser-verification.json +++ b/frontend-modern/browser-verification.json @@ -1,16 +1,18 @@ { "version": 1, - "base_sha": "21a35494aecb335755de773dd099e1364e65b3e0", - "verified_at": "2026-09-30T20:12:09Z", + "base_sha": "f3c92a242ebf9b7062dfc9f6ff711a56639f536c", + "verified_at": "2026-09-30T22:39:35Z", "result": "passed", "changed_paths": [ - "frontend-modern/src/components/Workloads/GuestDrawerHistory.tsx" + "frontend-modern/src/components/UpdateProgressModal.tsx", + "frontend-modern/src/components/updateReadinessModel.ts" ], "content_sha256": { - "frontend-modern/src/components/Workloads/GuestDrawerHistory.tsx": "b2b31441b19c2e50283ceedf319cd09a1f94791750cd424c052717a4957fdc67" + "frontend-modern/src/components/UpdateProgressModal.tsx": "d51df7a98a379e313f306b68529c8db0687ff91176af23869411c4b46feb2599", + "frontend-modern/src/components/updateReadinessModel.ts": "b25bf82441fc6572b95891af48a1ed2a0e8ea1f4e583c171ea9352a1e67d76ce" }, "routes": [ - "/browser-tests/pbs-identity-boundary.html (production PBS table and real resource History renderer; synthetic API data)" + "/browser-tests/update-progress-modal.html (production UpdateProgressModal with a scripted update stream and scripted /api/updates/status and /api/version)" ], "viewports": [ { @@ -23,38 +25,102 @@ } ], "states": [ - "corroborated host with seven observed metric paths", - "uncached range replacement pending with no old paths", - "new range loaded before the old range response", - "withdrawn host link with replacement target pending", - "replacement target returns 503 with no former-host history", - "service-only response retains CPU/memory, not host disk/network/I/O", - "locked 14-day range issues no history read" + "stream delivers one downloading 10% event then stays open and silent; fallback polling advances the modal through 35, 70, 80 and 90% (applying)", + "stream closed during download with a single unreachable status poll between two successful ones; download progress continues and the restart phase is never entered", + "restarting status from polling, then /api/version unreachable twice, then the new version reported; Pulse is restarting shown, then exactly one page reload", + "pre-update version unavailable (503), late polls failing and a new version plus idle answering without a restarting or completed status; modal stays on Pulse is restarting with no reload and no completion claim", + "stream delivers downloading, verifying, extracting, backing-up, applying and completed; old version answers once, becomes unreachable, then the new version; exactly one reload and no fallback polls" ], "interactions": [ - "desktop keyboard Enter / phone row click to open PBS details", - "select 6h while its GET is held; select 1h, then fulfil the obsolete 6h request", - "withdraw third host link while service GET is held, then return 503", - "select 6h after failure and inspect the service-only response", - "select locked 14d; check all ten screenshots, retained History tab, zero horizontal overflow and no page errors" + "emit scripted SSE statuses and hold the stream silent past the 6s silence window; assert the first fallback poll lands after it and the stream stays open", + "fail the stream and abort one status poll mid-download; assert no restart text ever rendered (MutationObserver) and the do-not-close warning remains", + "count document loads of the harness to assert a single reload after real restart evidence, and none without a baseline or backend confirmation", + "inspect screenshots at desktop and phone widths; assert no page errors, no horizontal overflow and the dialog inside the viewport" ], - "command": "pulse-worker-browser frontend-modern/browser-tests/pbs-history-source-isolation.cjs from assigned workspace root; Vite 6.4.3, Chromium 141.0.7390.37, Playwright 1.56.1", + "command": "NODE_PATH= PULSE_BROWSER_ARTIFACTS= PULSE_CHROMIUM_PATH= node frontend-modern/browser-tests/update-progress-modal.cjs from the worktree root; Vite dev server on 127.0.0.1:5208, Playwright 1.56.1, Chromium headless shell 151.0.7922.34", "artifacts": [ - "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/receipts/history-browser-final.log", - "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/result.json", - "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/host-loaded-1365.png", - "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/host-loaded-390.png", - "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/range-loading-1365.png", - "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/range-loading-390.png", - "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/service-loaded-1365.png", - "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/service-loaded-390.png", - "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/target-failed-1365.png", - "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/target-failed-390.png", - "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/target-loading-1365.png", - "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/target-loading-390.png" + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/harness-rebased-3.log", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/a-polled-90pct-1365.png", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/a-polled-90pct-390.png", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/a-stream-10pct-1365.png", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/a-stream-10pct-390.png", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/b-after-failed-poll-80pct-1365.png", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/b-after-failed-poll-80pct-390.png", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/c-reloaded-1365.png", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/c-reloaded-390.png", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/c-restarting-1365.png", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/c-restarting-390.png", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/d-unconfirmed-waiting-1365.png", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/d-unconfirmed-waiting-390.png", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/e-reloaded-1365.png", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/e-reloaded-390.png", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/e-restarting-1365.png", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/e-restarting-390.png", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/e-stream-applying-1365.png", + "/private/tmp/claude-501/-Users-rcourtman-Development-pulse-repos-pulse/6fb259b1-7b46-484b-80c9-554dd254295e/scratchpad/update-progress-modal-proof-rebased/e-stream-applying-390.png" + ], + "notes": [ + "Mock-backed UI proof of the production modal; all five scenarios passed at 1365x900 and 390x844 and the screenshots were inspected. The same harness run against the pre-fix UpdateProgressModal and updateReadinessModel (fa1d9eb36) fails scenario (a), staying on 10% (pre-fix/run.log). This does not exercise the backend SSE endpoint, a real proxy, or an installed update. Re-run after rebasing onto 8d5969fb3 (lockfile-only brace-expansion/DOMPurify floors); both full runs passed all five scenarios at both viewports with unchanged component content hashes.", + "Re-run after rebasing onto f3c92a242; all five scenarios passed at both viewports with unchanged component content hashes." ], - "notes": "Ten final screenshots inspected. The superseded GET is observed ERR_ABORTED at both widths; the unit regression also resolves mocks that ignore cancellation. Mock-backed proof establishes source/range isolation, not installed PBS collection, VirtualBox telemetry, all three #1723 drawers or release availability. Earlier unconfigured-CSS red/attempt-one results are retained separately.", "prior_verifications": [ + { + "version": 1, + "base_sha": "21a35494aecb335755de773dd099e1364e65b3e0", + "verified_at": "2026-09-30T20:12:09Z", + "result": "passed", + "changed_paths": [ + "frontend-modern/src/components/Workloads/GuestDrawerHistory.tsx" + ], + "content_sha256": { + "frontend-modern/src/components/Workloads/GuestDrawerHistory.tsx": "b2b31441b19c2e50283ceedf319cd09a1f94791750cd424c052717a4957fdc67" + }, + "routes": [ + "/browser-tests/pbs-identity-boundary.html (production PBS table and real resource History renderer; synthetic API data)" + ], + "viewports": [ + { + "width": 1365, + "height": 900 + }, + { + "width": 390, + "height": 844 + } + ], + "states": [ + "corroborated host with seven observed metric paths", + "uncached range replacement pending with no old paths", + "new range loaded before the old range response", + "withdrawn host link with replacement target pending", + "replacement target returns 503 with no former-host history", + "service-only response retains CPU/memory, not host disk/network/I/O", + "locked 14-day range issues no history read" + ], + "interactions": [ + "desktop keyboard Enter / phone row click to open PBS details", + "select 6h while its GET is held; select 1h, then fulfil the obsolete 6h request", + "withdraw third host link while service GET is held, then return 503", + "select 6h after failure and inspect the service-only response", + "select locked 14d; check all ten screenshots, retained History tab, zero horizontal overflow and no page errors" + ], + "command": "pulse-worker-browser frontend-modern/browser-tests/pbs-history-source-isolation.cjs from assigned workspace root; Vite 6.4.3, Chromium 141.0.7390.37, Playwright 1.56.1", + "artifacts": [ + "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/receipts/history-browser-final.log", + "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/result.json", + "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/host-loaded-1365.png", + "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/host-loaded-390.png", + "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/range-loading-1365.png", + "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/range-loading-390.png", + "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/service-loaded-1365.png", + "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/service-loaded-390.png", + "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/target-failed-1365.png", + "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/target-failed-390.png", + "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/target-loading-1365.png", + "/var/lib/pulse-maintainer/worker-outputs/web-product-j4h_mbv3/history-browser-final/target-loading-390.png" + ], + "notes": "Ten final screenshots inspected. The superseded GET is observed ERR_ABORTED at both widths; the unit regression also resolves mocks that ignore cancellation. Mock-backed proof establishes source/range isolation, not installed PBS collection, VirtualBox telemetry, all three #1723 drawers or release availability. Earlier unconfigured-CSS red/attempt-one results are retained separately." + }, { "version": 1, "base_sha": "6ade01cbd4d02cd25b691f9ff5919206cceae764", diff --git a/frontend-modern/src/components/UpdateProgressModal.tsx b/frontend-modern/src/components/UpdateProgressModal.tsx index 546a8c2db6..f1c9b33096 100644 --- a/frontend-modern/src/components/UpdateProgressModal.tsx +++ b/frontend-modern/src/components/UpdateProgressModal.tsx @@ -1,4 +1,4 @@ -import { createSignal, Show, onCleanup, createEffect } from 'solid-js'; +import { createSignal, Show, onCleanup, createEffect, untrack } from 'solid-js'; import { UpdatesAPI, type UpdateStatus } from '@/api/updates'; import AlertTriangleIcon from 'lucide-solid/icons/alert-triangle'; import CheckCircleIcon from 'lucide-solid/icons/check-circle'; @@ -11,7 +11,16 @@ import { ProgressBar } from '@/components/shared/ProgressBar'; import { apiFetch } from '@/utils/apiClient'; import { logger } from '@/utils/logger'; import { updateStore } from '@/stores/updates'; -import { resolvePostUpdateReload } from '@/components/updateReadinessModel'; +import { + RESTART_SUSPECT_POLL_FAILURES, + UPDATE_PROGRESS_STALL_TIMEOUT_MS, + UPDATE_STATUS_POLL_INTERVAL_MS, + UPDATE_STREAM_SILENCE_FALLBACK_MS, + isLateUpdateStage, + isUpdateInProgressStage, + resolvePostUpdateReload, + shouldApplyUpdateStatus, +} from '@/components/updateReadinessModel'; import XIcon from 'lucide-solid/icons/x'; interface UpdateProgressModalProps { @@ -29,14 +38,37 @@ export function UpdateProgressModal(props: UpdateProgressModalProps) { const [isRestarting, setIsRestarting] = createSignal(false); const [wsDisconnected, setWsDisconnected] = createSignal(false); const [healthCheckAttempts, setHealthCheckAttempts] = createSignal(0); + const [progressStalled, setProgressStalled] = createSignal(false); let pollInterval: number | undefined; let healthCheckTimer: number | undefined; + let streamSilenceTimer: number | undefined; + let progressStallTimer: number | undefined; let eventSource: EventSource | undefined; // The version that started this update. The backend keeps serving (and // answering health checks) for a grace period after reporting 'completed', // so "different version than this" is the only trustworthy restart signal. let preUpdateVersion: string | null = null; let sameVersionHealthyAttempts = 0; + // Set while a terminal status is being resolved (version probe in flight), + // so SSE and polling reporting the same terminal status act on it once. + let resolvingTerminalStatus = false; + // Bumped on every open/close so late async results from a previous open + // cannot drive the current one. + let session = 0; + // True once the backend itself reported 'restarting' or 'completed'. Only + // then may the modal stop listening for status and, as a last resort, + // reload on an unchanged version. A restart inferred from failing + // requests stays unconfirmed: the feeds keep running so a late progress + // or error event can still correct it. + let restartConfirmed = false; + let consecutivePollFailures = 0; + // Identifies the live health-check loop. Clearing the timer alone cannot + // stop a probe already in flight from re-arming itself. + let healthCheckLoop = 0; + // Set when a version probe fails after confirmed completion: the old + // process was seen going away, so the next healthy answer is the new one. + let restartObserved = false; + let baselineFetchInFlight = false; const resetModalState = () => { setStatus(null); @@ -45,8 +77,62 @@ export function UpdateProgressModal(props: UpdateProgressModalProps) { setIsRestarting(false); setWsDisconnected(false); setHealthCheckAttempts(0); + setProgressStalled(false); preUpdateVersion = updateStore.versionInfo()?.version ?? null; sameVersionHealthyAttempts = 0; + resolvingTerminalStatus = false; + restartConfirmed = false; + consecutivePollFailures = 0; + restartObserved = false; + baselineFetchInFlight = false; + }; + + // The pre-update version is only trustworthy while the old process is + // certainly the one answering: before any restart signal, and before a + // terminal or fresh-process status. + const canAdoptBaseline = () => { + if (preUpdateVersion || restartConfirmed || isRestarting() || resolvingTerminalStatus) { + return false; + } + const current = status()?.status; + return current !== 'restarting' && current !== 'completed' && current !== 'idle'; + }; + + const adoptBaseline = (version: string, source: string) => { + if (!version || !canAdoptBaseline()) { + return; + } + preUpdateVersion = version; + logger.info('Captured pre-update version for restart detection', { version, source }); + }; + + // The global watcher can open the modal before the update store has loaded + // the running version. Without that baseline an old-process answer looks + // like the new one, so fetch it directly while the old process is serving. + const fetchBaseline = () => { + if (baselineFetchInFlight || !canAdoptBaseline()) { + return; + } + baselineFetchInFlight = true; + const fetchSession = session; + void (async () => { + try { + const response = await apiFetch('/api/version', { cache: 'no-store' }); + if (fetchSession !== session || !response.ok) { + return; + } + const info = (await response.json()) as { version?: unknown }; + if (fetchSession === session && typeof info.version === 'string') { + adoptBaseline(info.version, 'version-probe'); + } + } catch (error) { + logger.warn('Could not read the pre-update version, will retry', error); + } finally { + if (fetchSession === session) { + baselineFetchInFlight = false; + } + } + })(); }; // Probe the backend and reload only once it reports a different version @@ -57,6 +143,9 @@ export function UpdateProgressModal(props: UpdateProgressModalProps) { const response = await apiFetch('/api/version', { cache: 'no-store' }); if (!response.ok) { sameVersionHealthyAttempts = 0; + if (restartConfirmed) { + restartObserved = true; + } return false; } const info = (await response.json()) as { version?: unknown }; @@ -65,6 +154,8 @@ export function UpdateProgressModal(props: UpdateProgressModalProps) { preUpdateVersion, reportedVersion, sameVersionHealthyAttempts, + completionConfirmed: restartConfirmed, + restartObserved, }); if (decision === 'reload') { logger.info('Backend ready after update, reloading...', { @@ -80,12 +171,16 @@ export function UpdateProgressModal(props: UpdateProgressModalProps) { // Connection refused here usually means the restart is actually // happening now; the pre-restart healthy answers no longer count. sameVersionHealthyAttempts = 0; + if (restartConfirmed) { + restartObserved = true; + } logger.warn('Version probe failed while waiting for restart, will retry', error); return false; } }; const clearHealthCheckTimer = () => { + healthCheckLoop += 1; if (healthCheckTimer !== undefined) { clearTimeout(healthCheckTimer); healthCheckTimer = undefined; @@ -99,7 +194,22 @@ export function UpdateProgressModal(props: UpdateProgressModalProps) { } }; + const clearStreamSilenceTimer = () => { + if (streamSilenceTimer !== undefined) { + clearTimeout(streamSilenceTimer); + streamSilenceTimer = undefined; + } + }; + + const clearProgressStallTimer = () => { + if (progressStallTimer !== undefined) { + clearTimeout(progressStallTimer); + progressStallTimer = undefined; + } + }; + const closeSSE = () => { + clearStreamSilenceTimer(); if (!eventSource) { return; } @@ -108,6 +218,172 @@ export function UpdateProgressModal(props: UpdateProgressModalProps) { logger.info('SSE connection closed'); }; + // Stop the status feeds (stream and poll); restart detection and the stall + // timer are managed separately. + const stopStatusFeeds = () => { + closeSSE(); + clearPollInterval(); + }; + + const stopEverything = () => { + stopStatusFeeds(); + clearHealthCheckTimer(); + clearProgressStallTimer(); + }; + + // Any real movement (new stage, new progress, entering the restart) pushes + // the "may have finished" fallback back out. Repeated identical answers do + // not, so a stream or poll that keeps echoing one stage still times out. + const armProgressStallTimer = () => { + clearProgressStallTimer(); + setProgressStalled(false); + progressStallTimer = window.setTimeout(() => { + progressStallTimer = undefined; + if (!isComplete()) { + logger.warn('Update progress stalled, offering a manual reload'); + setProgressStalled(true); + } + }, UPDATE_PROGRESS_STALL_TIMEOUT_MS); + }; + + // A quiet stream is not a closed stream. If nothing arrives for a while, + // poll as well rather than trusting the open EventSource. + const armStreamSilenceWatchdog = () => { + clearStreamSilenceTimer(); + streamSilenceTimer = window.setTimeout(() => { + streamSilenceTimer = undefined; + if (!eventSource || isComplete() || restartConfirmed) { + return; + } + logger.warn('Update progress stream went quiet, polling status as well'); + startPolling(); + }, UPDATE_STREAM_SILENCE_FALLBACK_MS); + }; + + // The backend reported the restart itself: stop listening for status and + // wait for the new version to serve. + const enterConfirmedRestartPhase = () => { + restartConfirmed = true; + stopStatusFeeds(); + armProgressStallTimer(); + // A suspected restart already has a health-check loop running; it reads + // the confirmation on its next probe. + if (!isRestarting()) { + setIsRestarting(true); + startHealthCheckPolling(); + } + }; + + // Requests are failing late in the update, which usually means the old + // process just exited. Probe for the new version, but keep the status + // feeds running so a status from a still-running old process wins. + const enterSuspectedRestartPhase = () => { + if (isRestarting()) { + return; + } + logger.warn('Status requests failing late in the update, probing for the restarted backend'); + setIsRestarting(true); + startHealthCheckPolling(); + }; + + const leaveSuspectedRestartPhase = () => { + if (!isRestarting() || restartConfirmed) { + return; + } + setIsRestarting(false); + clearHealthCheckTimer(); + setHealthCheckAttempts(0); + }; + + const finishWithResult = (failed: boolean) => { + stopStatusFeeds(); + clearHealthCheckTimer(); + clearProgressStallTimer(); + setProgressStalled(false); + // A failure reported while a restart was only suspected ends that too. + setIsRestarting(false); + setIsComplete(true); + if (failed) { + setHasError(true); + } + }; + + // Single entry point for statuses from the SSE stream and from polling. + const handleStatus = (next: UpdateStatus) => { + if (isComplete() || restartConfirmed || resolvingTerminalStatus) { + return; + } + const previous = status(); + if (!shouldApplyUpdateStatus(previous, next)) { + return; + } + + if (next.status === 'idle') { + handleIdleStatus(previous); + return; + } + + setStatus(next); + if (!previous || previous.status !== next.status || previous.progress !== next.progress) { + armProgressStallTimer(); + } + // Still no baseline: retry while the old process is known to be serving. + fetchBaseline(); + + if (next.status === 'restarting') { + enterConfirmedRestartPhase(); + return; + } + + if (next.status === 'error' || (next.status === 'completed' && next.error)) { + finishWithResult(true); + return; + } + + if (next.status === 'completed') { + // Reported by the old process just before it exits. Reload once the + // new version is serving. + restartConfirmed = true; + resolvingTerminalStatus = true; + stopStatusFeeds(); + const probeSession = session; + void attemptReadyReload().then((reloaded) => { + if (reloaded || probeSession !== session) { + return; + } + resolvingTerminalStatus = false; + // Backend not on the new version yet — restart in progress. + enterConfirmedRestartPhase(); + }); + return; + } + + // Any other live status means the old process is still answering. + leaveSuspectedRestartPhase(); + }; + + // 'idle' is what a freshly started process reports, so it is the restart + // signal when the stream was held open and never delivered 'completed'. + // It is also what an update check reports, so it only ends the update + // once the version has actually moved; otherwise the feeds keep running. + const handleIdleStatus = (previous: UpdateStatus | null) => { + const updateWasRunning = previous !== null && isUpdateInProgressStage(previous.status); + resolvingTerminalStatus = true; + if (!updateWasRunning) { + stopStatusFeeds(); + } + const probeSession = session; + void attemptReadyReload().then((reloaded) => { + if (reloaded || probeSession !== session) { + return; + } + resolvingTerminalStatus = false; + if (!updateWasRunning) { + finishWithResult(false); + } + }); + }; + const setupSSE = () => { // Close existing connection if any closeSSE(); @@ -115,48 +391,18 @@ export function UpdateProgressModal(props: UpdateProgressModalProps) { try { // Create EventSource connection to SSE endpoint eventSource = new EventSource('/api/updates/stream'); + // Armed before the connection opens: a request that never connects is + // as silent as one that connects and then stops delivering. + armStreamSilenceWatchdog(); eventSource.onopen = () => { logger.info('SSE connection established'); }; eventSource.onmessage = (event) => { + armStreamSilenceWatchdog(); try { - const updateStatus = JSON.parse(event.data) as UpdateStatus; - setStatus(updateStatus); - - // Check if restarting - if (updateStatus.status === 'restarting') { - setIsRestarting(true); - closeSSE(); - startHealthCheckPolling(); - return; - } - - // Check if complete or error - if ( - updateStatus.status === 'completed' || - updateStatus.status === 'idle' || - updateStatus.status === 'error' - ) { - if (updateStatus.status === 'completed' && !updateStatus.error) { - closeSSE(); - void attemptReadyReload().then((reloaded) => { - if (!reloaded) { - // Backend not on the new version yet — restart in progress. - setIsRestarting(true); - startHealthCheckPolling(); - } - }); - return; - } - - setIsComplete(true); - if (updateStatus.status === 'error' || updateStatus.error) { - setHasError(true); - } - closeSSE(); - } + handleStatus(JSON.parse(event.data) as UpdateStatus); } catch (error) { logger.error('Failed to parse SSE update status', error); } @@ -178,73 +424,43 @@ export function UpdateProgressModal(props: UpdateProgressModalProps) { const startPolling = () => { // Don't start polling if already polling - if (pollInterval !== undefined) { + if (pollInterval !== undefined || isComplete() || restartConfirmed) { return; } - logger.info('Starting status polling (SSE not available)'); - pollStatus(); - pollInterval = setInterval(pollStatus, 2000) as unknown as number; + logger.info('Starting update status polling'); + void pollStatus(); + pollInterval = setInterval(pollStatus, UPDATE_STATUS_POLL_INTERVAL_MS) as unknown as number; }; const pollStatus = async () => { + const pollSession = session; try { const currentStatus = await UpdatesAPI.getUpdateStatus(); - setStatus(currentStatus); - - // Check if restarting - if (currentStatus.status === 'restarting') { - setIsRestarting(true); - clearPollInterval(); - // Start health check polling - startHealthCheckPolling(); + if (pollSession !== session) { return; } - - // Check if complete or error - if ( - currentStatus.status === 'completed' || - currentStatus.status === 'idle' || - currentStatus.status === 'error' - ) { - // If completed successfully, reload once the new version is serving - if (currentStatus.status === 'completed' && !currentStatus.error) { - clearPollInterval(); - if (await attemptReadyReload()) { - return; - } - // Backend not on the new version yet — restart in progress. - setIsRestarting(true); - startHealthCheckPolling(); - return; - } - - setIsComplete(true); - if (currentStatus.status === 'error' || currentStatus.error) { - setHasError(true); - } - clearPollInterval(); - } + consecutivePollFailures = 0; + handleStatus(currentStatus); } catch (error) { - logger.error('Failed to poll update status', error); - // If we get errors during update, assume we're restarting - const currentStatus = status(); - const shouldAssumeRestart = - !isRestarting() && - (!currentStatus || (currentStatus.status !== 'idle' && currentStatus.status !== 'error')); - - if (shouldAssumeRestart) { - if (!currentStatus) { - setStatus({ - status: 'restarting', - progress: 95, - message: 'Restarting service...', - updatedAt: new Date().toISOString(), - }); - } - setIsRestarting(true); - clearPollInterval(); - startHealthCheckPolling(); + if (pollSession !== session || isComplete() || restartConfirmed || resolvingTerminalStatus) { + return; + } + consecutivePollFailures += 1; + logger.warn('Failed to poll update status, will retry', { + consecutiveFailures: consecutivePollFailures, + error, + }); + // One failed poll proves nothing: a transient network error or a 500 + // during a long download must not end the session. Treat failures as + // restart evidence only when the stream has also closed, several polls + // in a row failed, and the update had reached the stage that restarts. + if ( + !eventSource && + consecutivePollFailures >= RESTART_SUSPECT_POLL_FAILURES && + isLateUpdateStage(status()?.status) + ) { + enterSuspectedRestartPhase(); } } }; @@ -252,16 +468,25 @@ export function UpdateProgressModal(props: UpdateProgressModalProps) { const startHealthCheckPolling = () => { clearHealthCheckTimer(); setHealthCheckAttempts(0); + // clearHealthCheckTimer (also run on close and on leaving a suspected + // restart) moves the loop id on, retiring this loop even mid-probe. + const loop = healthCheckLoop; const checkHealth = async () => { + healthCheckTimer = undefined; + if (loop !== healthCheckLoop) { + return; + } if (await attemptReadyReload()) { return; } + if (loop !== healthCheckLoop) { + return; + } const attempt = Math.min(healthCheckAttempts(), 3); const nextDelay = Math.min(2000 * Math.pow(2, attempt), 15000); setHealthCheckAttempts((current) => current + 1); - clearHealthCheckTimer(); healthCheckTimer = window.setTimeout(checkHealth, nextDelay); }; @@ -295,24 +520,37 @@ export function UpdateProgressModal(props: UpdateProgressModalProps) { } }); - // Start/stop SSE or polling based on modal visibility + // Start/stop SSE or polling based on modal visibility. Only isOpen is + // tracked: the setup reads other signals (versionInfo, isComplete), and a + // change in any of them must not tear down and restart a live session. createEffect(() => { - if (props.isOpen) { - resetModalState(); - // Try SSE first, will fall back to polling if it fails - setupSSE(); - } else { - // Stop everything when modal closes - closeSSE(); - clearPollInterval(); - clearHealthCheckTimer(); + const open = props.isOpen; + untrack(() => { + session += 1; + stopEverything(); + if (open) { + resetModalState(); + fetchBaseline(); + armProgressStallTimer(); + // Try SSE first; a stream error or prolonged silence adds polling. + setupSSE(); + } + }); + }); + + // Take the baseline from the store as soon as it loads, if the modal opened + // first. canAdoptBaseline refuses it once a restart may be under way. + createEffect(() => { + const version = updateStore.versionInfo()?.version; + if (!props.isOpen || !version) { + return; } + untrack(() => adoptBaseline(version, 'update-store')); }); onCleanup(() => { - closeSSE(); - clearPollInterval(); - clearHealthCheckTimer(); + session += 1; + stopEverything(); }); const getStageIcon = () => { @@ -466,7 +704,7 @@ export function UpdateProgressModal(props: UpdateProgressModalProps) { - + + + + diff --git a/frontend-modern/src/components/__tests__/UpdateProgressModal.test.tsx b/frontend-modern/src/components/__tests__/UpdateProgressModal.test.tsx new file mode 100644 index 0000000000..9eadac6ea7 --- /dev/null +++ b/frontend-modern/src/components/__tests__/UpdateProgressModal.test.tsx @@ -0,0 +1,514 @@ +import { cleanup, fireEvent, render, screen } from '@solidjs/testing-library'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { UpdateStatus } from '@/api/updates'; +import { + MAX_SAME_VERSION_HEALTHY_ATTEMPTS, + UPDATE_PROGRESS_STALL_TIMEOUT_MS, + UPDATE_STATUS_POLL_INTERVAL_MS, + UPDATE_STREAM_SILENCE_FALLBACK_MS, +} from '@/components/updateReadinessModel'; + +const getUpdateStatusMock = vi.hoisted(() => vi.fn()); +const apiFetchMock = vi.hoisted(() => vi.fn()); + +vi.mock('@/api/updates', () => ({ + UpdatesAPI: { getUpdateStatus: getUpdateStatusMock }, +})); + +vi.mock('@/utils/apiClient', () => ({ + apiFetch: apiFetchMock, +})); + +vi.mock('@/stores/updates', async () => { + const { createSignal } = await import('solid-js'); + const [versionInfo, setVersionInfo] = createSignal<{ version: string } | null>({ + version: '6.4.5-rc.5', + }); + return { updateStore: { versionInfo }, __setStoreVersionInfo: setVersionInfo }; +}); + +vi.mock('@/utils/logger', () => ({ + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }, +})); + +import { UpdateProgressModal } from '@/components/UpdateProgressModal'; +import * as updatesStoreModule from '@/stores/updates'; + +const setStoreVersionInfo = ( + updatesStoreModule as unknown as { + __setStoreVersionInfo: (value: { version: string } | null) => void; + } +).__setStoreVersionInfo; + +class MockEventSource { + static instances: MockEventSource[] = []; + + readonly url: string; + onopen: ((event: Event) => void) | null = null; + onmessage: ((event: MessageEvent) => void) | null = null; + onerror: ((event: Event) => void) | null = null; + readyState = 0; + closed = false; + + constructor(url: string) { + this.url = url; + MockEventSource.instances.push(this); + } + + close() { + this.closed = true; + this.readyState = 2; + } + + emitOpen() { + this.readyState = 1; + this.onopen?.(new Event('open')); + } + + emitError() { + this.onerror?.(new Event('error')); + } + + emitStatus(status: Partial) { + this.onmessage?.({ + data: JSON.stringify({ message: '', updatedAt: '', ...status }), + } as MessageEvent); + } +} + +const statusOf = (status: string, progress: number, message = ''): UpdateStatus => ({ + status, + progress, + message, + updatedAt: new Date().toISOString(), +}); + +const versionResponse = (version: string) => ({ + ok: true, + json: async () => ({ version }), +}); + +const renderModal = () => + render(() => ( + undefined} + onViewHistory={() => undefined} + connected={() => true} + reconnecting={() => false} + /> + )); + +describe('UpdateProgressModal progress feed', () => { + const originalEventSource = globalThis.EventSource; + const originalLocation = window.location; + let reloadMock: ReturnType; + + beforeEach(() => { + vi.useFakeTimers(); + MockEventSource.instances = []; + (globalThis as unknown as { EventSource: typeof EventSource }).EventSource = + MockEventSource as unknown as typeof EventSource; + reloadMock = vi.fn(); + // jsdom marks location.reload as non-configurable, so replace the whole object + Object.defineProperty(window, 'location', { + value: { ...originalLocation, reload: reloadMock }, + writable: true, + configurable: true, + }); + getUpdateStatusMock.mockReset(); + apiFetchMock.mockReset(); + setStoreVersionInfo({ version: '6.4.5-rc.5' }); + }); + + afterEach(() => { + cleanup(); + vi.useRealTimers(); + (globalThis as unknown as { EventSource: typeof EventSource }).EventSource = + originalEventSource; + Object.defineProperty(window, 'location', { + value: originalLocation, + writable: true, + configurable: true, + }); + }); + + it('polls when the stream opens and then goes silent, and reloads once the new version serves', async () => { + // The stream delivers the first stage and then nothing, without erroring: + // the "stuck at Downloading 10%" report. + const polled = [ + statusOf('applying', 80, 'Applying update...'), + statusOf('restarting', 95, 'Restarting service...'), + ]; + getUpdateStatusMock.mockImplementation(async () => polled.shift() ?? polled[0]); + apiFetchMock.mockResolvedValue(versionResponse('6.4.5')); + + renderModal(); + const stream = MockEventSource.instances[0]; + expect(stream.url).toBe('/api/updates/stream'); + stream.emitOpen(); + stream.emitStatus({ status: 'downloading', progress: 10, message: 'Downloading update...' }); + + expect(await screen.findByText('Downloading update...')).toBeInTheDocument(); + expect(screen.getByText('10%')).toBeInTheDocument(); + + await vi.advanceTimersByTimeAsync(UPDATE_STREAM_SILENCE_FALLBACK_MS - 1); + expect(getUpdateStatusMock).not.toHaveBeenCalled(); + + // The EventSource is still open and never errored; silence alone starts polling. + await vi.advanceTimersByTimeAsync(1); + expect(stream.closed).toBe(false); + expect(getUpdateStatusMock).toHaveBeenCalledTimes(1); + expect(await screen.findByText('Applying update...')).toBeInTheDocument(); + expect(screen.getByText('80%')).toBeInTheDocument(); + + await vi.advanceTimersByTimeAsync(UPDATE_STATUS_POLL_INTERVAL_MS); + expect(screen.getByText('Pulse is restarting...')).toBeInTheDocument(); + expect(stream.closed).toBe(true); + + // The first health check is scheduled immediately (fake timers treat a + // zero delay scheduled mid-tick as 1ms). + await vi.advanceTimersByTimeAsync(1); + expect(apiFetchMock).toHaveBeenCalledWith('/api/version', { cache: 'no-store' }); + expect(reloadMock).toHaveBeenCalledTimes(1); + }); + + it('keeps probing through the restart and reloads only when a different version is served', async () => { + const versions = [ + Promise.resolve(versionResponse('6.4.5-rc.5')), + Promise.reject(new Error('connection refused')), + Promise.resolve(versionResponse('6.4.5')), + ]; + // Avoid unhandled-rejection noise for the pre-built rejected promise. + versions[1].catch(() => undefined); + apiFetchMock.mockImplementation(() => versions.shift() ?? versions[versions.length - 1]); + + renderModal(); + const stream = MockEventSource.instances[0]; + stream.emitOpen(); + stream.emitStatus({ + status: 'completed', + progress: 100, + message: 'Update completed, restarting...', + }); + + // First probe still answers with the old process's version: wait. + await vi.advanceTimersByTimeAsync(0); + expect(reloadMock).not.toHaveBeenCalled(); + expect(await screen.findByText('Pulse is restarting...')).toBeInTheDocument(); + + // Health check during the restart window fails, then the new version answers. + await vi.advanceTimersByTimeAsync(0); + expect(reloadMock).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(2000); + expect(reloadMock).toHaveBeenCalledTimes(1); + expect(getUpdateStatusMock).not.toHaveBeenCalled(); + }); + + it('offers a reload instead of an endless spinner when progress stops moving', async () => { + getUpdateStatusMock.mockResolvedValue(statusOf('downloading', 10, 'Downloading update...')); + + renderModal(); + const stream = MockEventSource.instances[0]; + stream.emitOpen(); + stream.emitStatus({ status: 'downloading', progress: 10, message: 'Downloading update...' }); + expect( + screen.getByText('Please do not close this window or refresh the page during the update.'), + ).toBeInTheDocument(); + + await vi.advanceTimersByTimeAsync(UPDATE_PROGRESS_STALL_TIMEOUT_MS - 1); + expect(screen.queryByText('No progress reported for a while')).not.toBeInTheDocument(); + // Polling kept echoing the same stage; that is not progress. + expect(getUpdateStatusMock.mock.calls.length).toBeGreaterThan(1); + + await vi.advanceTimersByTimeAsync(1); + expect(screen.getByText('No progress reported for a while')).toBeInTheDocument(); + expect( + screen.queryByText('Please do not close this window or refresh the page during the update.'), + ).not.toBeInTheDocument(); + + fireEvent.click(screen.getByRole('button', { name: 'Reload to check' })); + expect(reloadMock).toHaveBeenCalledTimes(1); + }); + + it('ignores stale polled stages that would move progress backwards', async () => { + getUpdateStatusMock.mockResolvedValue(statusOf('downloading', 10, 'Downloading update...')); + + renderModal(); + const stream = MockEventSource.instances[0]; + stream.emitOpen(); + stream.emitStatus({ status: 'extracting', progress: 40, message: 'Extracting update...' }); + + await vi.advanceTimersByTimeAsync(UPDATE_STREAM_SILENCE_FALLBACK_MS); + expect(getUpdateStatusMock).toHaveBeenCalled(); + expect(screen.getByText('Extracting update...')).toBeInTheDocument(); + expect(screen.queryByText('Downloading update...')).not.toBeInTheDocument(); + }); + it('rides out a transient poll failure during a download without reloading', async () => { + const polled: Array = [ + new Error('Request failed with status 500'), + statusOf('downloading', 10, 'Downloading update...'), + statusOf('extracting', 40, 'Extracting update...'), + ]; + getUpdateStatusMock.mockImplementation(async () => { + const next = polled.shift() ?? statusOf('extracting', 40, 'Extracting update...'); + if (next instanceof Error) { + throw next; + } + return next; + }); + apiFetchMock.mockResolvedValue(versionResponse('6.4.5-rc.5')); + + renderModal(); + const stream = MockEventSource.instances[0]; + stream.emitOpen(); + stream.emitStatus({ status: 'downloading', progress: 10, message: 'Downloading update...' }); + + // Silence starts polling; the first poll fails with a 500. + await vi.advanceTimersByTimeAsync(UPDATE_STREAM_SILENCE_FALLBACK_MS); + expect(getUpdateStatusMock).toHaveBeenCalledTimes(1); + expect(stream.closed).toBe(false); + expect(screen.queryByText('Pulse is restarting...')).not.toBeInTheDocument(); + expect(screen.getByText('Downloading update...')).toBeInTheDocument(); + + // Polling carries on and progress keeps moving. + await vi.advanceTimersByTimeAsync(UPDATE_STATUS_POLL_INTERVAL_MS * 2); + expect(screen.getByText('Extracting update...')).toBeInTheDocument(); + expect(screen.getByText('40%')).toBeInTheDocument(); + + // The stream is still live and later stages from it are still honoured. + stream.emitStatus({ status: 'backing-up', progress: 60, message: 'Creating backup...' }); + expect(screen.getByText('Creating backup...')).toBeInTheDocument(); + expect(stream.closed).toBe(false); + expect(apiFetchMock).not.toHaveBeenCalled(); + expect(reloadMock).not.toHaveBeenCalled(); + }); + + it('keeps polling through repeated failures early in the update, even with the stream closed', async () => { + let failing = true; + getUpdateStatusMock.mockImplementation(async () => { + if (failing) { + throw new Error('network error'); + } + return statusOf('verifying', 25, 'Verifying signature...'); + }); + apiFetchMock.mockResolvedValue(versionResponse('6.4.5-rc.5')); + + renderModal(); + const stream = MockEventSource.instances[0]; + stream.emitOpen(); + stream.emitStatus({ status: 'downloading', progress: 10, message: 'Downloading update...' }); + stream.emitError(); + expect(stream.closed).toBe(true); + + await vi.advanceTimersByTimeAsync(UPDATE_STATUS_POLL_INTERVAL_MS * 6); + expect(getUpdateStatusMock.mock.calls.length).toBeGreaterThanOrEqual(6); + expect(screen.queryByText('Pulse is restarting...')).not.toBeInTheDocument(); + expect(apiFetchMock).not.toHaveBeenCalled(); + + failing = false; + await vi.advanceTimersByTimeAsync(UPDATE_STATUS_POLL_INTERVAL_MS); + expect(screen.getByText('Verifying signature...')).toBeInTheDocument(); + expect(reloadMock).not.toHaveBeenCalled(); + }); + + it('never reloads on an unchanged version when the restart was only inferred', async () => { + getUpdateStatusMock.mockRejectedValue(new Error('connection refused')); + let serving = '6.4.5-rc.5'; + apiFetchMock.mockImplementation(async () => versionResponse(serving)); + + renderModal(); + const stream = MockEventSource.instances[0]; + stream.emitOpen(); + stream.emitStatus({ status: 'applying', progress: 80, message: 'Applying update...' }); + stream.emitError(); + + // Two failed polls late in the update, with the stream gone: probe for + // the restarted backend, but the restart is not confirmed. + await vi.advanceTimersByTimeAsync(UPDATE_STATUS_POLL_INTERVAL_MS); + expect(screen.getByText('Pulse is restarting...')).toBeInTheDocument(); + + // Far more same-version answers than the confirmed-restart fallback allows. + await vi.advanceTimersByTimeAsync(15000 * (MAX_SAME_VERSION_HEALTHY_ATTEMPTS + 4)); + expect(apiFetchMock.mock.calls.length).toBeGreaterThan(MAX_SAME_VERSION_HEALTHY_ATTEMPTS + 1); + expect(reloadMock).not.toHaveBeenCalled(); + // The user gets the manual check instead. + expect(screen.getByText('No progress reported for a while')).toBeInTheDocument(); + + // Real evidence (the new version serving) still reloads. + serving = '6.4.5'; + await vi.advanceTimersByTimeAsync(15000); + expect(reloadMock).toHaveBeenCalledTimes(1); + }); + + it('lets a late failure correct an inferred restart instead of reloading', async () => { + let pollFails = true; + getUpdateStatusMock.mockImplementation(async () => { + if (pollFails) { + throw new Error('connection refused'); + } + return { ...statusOf('error', 80, 'Failed to apply update'), error: 'disk full' }; + }); + apiFetchMock.mockResolvedValue(versionResponse('6.4.5-rc.5')); + + renderModal(); + const stream = MockEventSource.instances[0]; + stream.emitOpen(); + stream.emitStatus({ status: 'applying', progress: 80, message: 'Applying update...' }); + stream.emitError(); + + await vi.advanceTimersByTimeAsync(UPDATE_STATUS_POLL_INTERVAL_MS); + expect(screen.getByText('Pulse is restarting...')).toBeInTheDocument(); + + pollFails = false; + await vi.advanceTimersByTimeAsync(UPDATE_STATUS_POLL_INTERVAL_MS); + expect(screen.getByText('Update Failed')).toBeInTheDocument(); + expect(screen.getByText('disk full')).toBeInTheDocument(); + + await vi.advanceTimersByTimeAsync(60000); + expect(reloadMock).not.toHaveBeenCalled(); + }); + + it('reloads after a confirmed restart only once the new version serves', async () => { + getUpdateStatusMock.mockResolvedValue(statusOf('downloading', 10, 'Downloading update...')); + let serving = '6.4.5-rc.5'; + apiFetchMock.mockImplementation(async () => versionResponse(serving)); + + renderModal(); + const stream = MockEventSource.instances[0]; + stream.emitOpen(); + stream.emitStatus({ status: 'restarting', progress: 95, message: 'Restarting service...' }); + expect(stream.closed).toBe(true); + + await vi.advanceTimersByTimeAsync(2000); + expect(apiFetchMock).toHaveBeenCalled(); + expect(reloadMock).not.toHaveBeenCalled(); + // A stale status from a poll cannot drag a confirmed restart back. + expect(getUpdateStatusMock).not.toHaveBeenCalled(); + + serving = '6.4.5'; + await vi.advanceTimersByTimeAsync(15000); + expect(reloadMock).toHaveBeenCalledTimes(1); + }); + + describe('when the modal opens before the running version is known', () => { + beforeEach(() => { + setStoreVersionInfo(null); + }); + + it('fetches the pre-update version on open and uses it for restart detection', async () => { + let serving = '6.4.5-rc.5'; + apiFetchMock.mockImplementation(async () => versionResponse(serving)); + + renderModal(); + expect(apiFetchMock).toHaveBeenCalledWith('/api/version', { cache: 'no-store' }); + await vi.advanceTimersByTimeAsync(0); + + const stream = MockEventSource.instances[0]; + stream.emitOpen(); + stream.emitStatus({ status: 'completed', progress: 100, message: 'Update completed' }); + + // The old process still answers with the captured version: wait. + await vi.advanceTimersByTimeAsync(4000); + expect(reloadMock).not.toHaveBeenCalled(); + + serving = '6.4.5'; + await vi.advanceTimersByTimeAsync(15000); + expect(reloadMock).toHaveBeenCalledTimes(1); + }); + + it('takes the baseline from the store once it loads', async () => { + getUpdateStatusMock.mockRejectedValue(new Error('connection refused')); + let serving = '6.4.5-rc.5'; + let baselineCall = true; + apiFetchMock.mockImplementation(async () => { + if (baselineCall) { + baselineCall = false; + throw new Error('network blip'); + } + return versionResponse(serving); + }); + + renderModal(); + const stream = MockEventSource.instances[0]; + stream.emitOpen(); + stream.emitStatus({ status: 'downloading', progress: 10, message: 'Downloading update...' }); + await vi.advanceTimersByTimeAsync(0); + + setStoreVersionInfo({ version: '6.4.5-rc.5' }); + stream.emitStatus({ status: 'applying', progress: 80, message: 'Applying update...' }); + stream.emitError(); + + // Suspected restart; old-version answers must not reload. + await vi.advanceTimersByTimeAsync(60000); + expect(screen.getByText('Pulse is restarting...')).toBeInTheDocument(); + expect(reloadMock).not.toHaveBeenCalled(); + + serving = '6.4.5'; + await vi.advanceTimersByTimeAsync(15000); + expect(reloadMock).toHaveBeenCalledTimes(1); + }); + + it('never reloads without a baseline while completion is unconfirmed', async () => { + getUpdateStatusMock.mockRejectedValue(new Error('connection refused')); + let baselineCall = true; + apiFetchMock.mockImplementation(async () => { + if (baselineCall) { + baselineCall = false; + throw new Error('network blip'); + } + // The old process, still healthy. + return versionResponse('6.4.5-rc.5'); + }); + + renderModal(); + const stream = MockEventSource.instances[0]; + stream.emitOpen(); + stream.emitStatus({ status: 'applying', progress: 80, message: 'Applying update...' }); + stream.emitError(); + + await vi.advanceTimersByTimeAsync(UPDATE_STATUS_POLL_INTERVAL_MS); + expect(screen.getByText('Pulse is restarting...')).toBeInTheDocument(); + + await vi.advanceTimersByTimeAsync(15000 * (MAX_SAME_VERSION_HEALTHY_ATTEMPTS + 4)); + expect(apiFetchMock.mock.calls.length).toBeGreaterThan(MAX_SAME_VERSION_HEALTHY_ATTEMPTS + 1); + expect(reloadMock).not.toHaveBeenCalled(); + expect(screen.getByText('No progress reported for a while')).toBeInTheDocument(); + }); + + it('after confirmed completion without a baseline, reloads only once the restart was seen', async () => { + const answers: Array<'blip' | 'down' | string> = [ + 'blip', // baseline fetch on open fails + '6.4.5-rc.5', // old process still answering after 'completed' + '6.4.5-rc.5', + 'down', // old process exits + '6.4.5', // new process + ]; + apiFetchMock.mockImplementation(async () => { + const next = answers.length > 1 ? answers.shift()! : answers[0]; + if (next === 'blip' || next === 'down') { + throw new Error(next); + } + return versionResponse(next); + }); + + renderModal(); + const stream = MockEventSource.instances[0]; + stream.emitOpen(); + await vi.advanceTimersByTimeAsync(0); + stream.emitStatus({ status: 'completed', progress: 100, message: 'Update completed' }); + + // Two healthy answers from the old process right after 'completed', then + // the probe at 2s finds it gone. + await vi.advanceTimersByTimeAsync(2000); + expect(apiFetchMock).toHaveBeenCalledTimes(4); + expect(reloadMock).not.toHaveBeenCalled(); + expect(screen.getByText('Pulse is restarting...')).toBeInTheDocument(); + + // The next healthy answer after the restart was seen reloads. + await vi.advanceTimersByTimeAsync(4000); + expect(apiFetchMock).toHaveBeenCalledTimes(5); + expect(reloadMock).toHaveBeenCalledTimes(1); + }); + }); +}); diff --git a/frontend-modern/src/components/__tests__/updateReadinessModel.test.ts b/frontend-modern/src/components/__tests__/updateReadinessModel.test.ts index 7bf8b00af3..c17b824305 100644 --- a/frontend-modern/src/components/__tests__/updateReadinessModel.test.ts +++ b/frontend-modern/src/components/__tests__/updateReadinessModel.test.ts @@ -58,12 +58,55 @@ describe('resolvePostUpdateReload', () => { ).toBe('wait'); }); - it('reloads on the first healthy response when no pre-update version is known', () => { + it('never reloads without a pre-update version while completion is unconfirmed', () => { + // An old process answering healthy is indistinguishable from the new one. + for (const completionConfirmed of [undefined, false]) { + expect( + resolvePostUpdateReload({ + preUpdateVersion: null, + reportedVersion: '6.1.0-rc.4', + sameVersionHealthyAttempts: MAX_SAME_VERSION_HEALTHY_ATTEMPTS + 10, + completionConfirmed, + restartObserved: true, + }), + ).toBe('wait'); + } + }); + + it('without a pre-update version, reloads after confirmed completion once the restart was seen', () => { + const base = { + preUpdateVersion: null, + reportedVersion: '6.1.0-rc.4', + completionConfirmed: true, + }; + // The about-to-exit process still answers right after 'completed'. + expect(resolvePostUpdateReload({ ...base, sameVersionHealthyAttempts: 0 })).toBe('wait'); + expect( + resolvePostUpdateReload({ ...base, sameVersionHealthyAttempts: 0, restartObserved: true }), + ).toBe('reload'); expect( resolvePostUpdateReload({ - preUpdateVersion: null, - reportedVersion: '6.1.0-rc.4', + ...base, + sameVersionHealthyAttempts: MAX_SAME_VERSION_HEALTHY_ATTEMPTS, + }), + ).toBe('reload'); + }); + + it('never falls back to a same-version reload while completion is unconfirmed', () => { + expect( + resolvePostUpdateReload({ + preUpdateVersion: '6.4.5-rc.5', + reportedVersion: '6.4.5-rc.5', + sameVersionHealthyAttempts: MAX_SAME_VERSION_HEALTHY_ATTEMPTS + 10, + completionConfirmed: false, + }), + ).toBe('wait'); + expect( + resolvePostUpdateReload({ + preUpdateVersion: '6.4.5-rc.5', + reportedVersion: '6.4.5', sameVersionHealthyAttempts: 0, + completionConfirmed: false, }), ).toBe('reload'); }); diff --git a/frontend-modern/src/components/updateReadinessModel.ts b/frontend-modern/src/components/updateReadinessModel.ts index 0b76b69e80..6827c8aeed 100644 --- a/frontend-modern/src/components/updateReadinessModel.ts +++ b/frontend-modern/src/components/updateReadinessModel.ts @@ -21,19 +21,101 @@ export const resolvePostUpdateReload = (input: { preUpdateVersion: string | null; reportedVersion: string; sameVersionHealthyAttempts: number; + // Whether the backend itself reported 'restarting' or 'completed'. Without + // that, an unchanged version may simply mean the update is still running, + // so the same-version fallback must never fire. + completionConfirmed?: boolean; + // Whether a probe failed (connection refused or a non-OK answer) after + // completion was confirmed, i.e. the old process was seen going away. + restartObserved?: boolean; }): PostUpdateReloadDecision => { - // Without a known pre-update version there is nothing to compare against; - // a healthy response is the best remaining signal. + // Without a known pre-update version a healthy answer cannot tell the old + // process from the new one. Never reload before the backend confirms + // completion. After that, reload once the old process was seen going away + // and something healthy answers again, or after the bounded fallback. if (!input.preUpdateVersion) { - return 'reload'; + if (input.completionConfirmed !== true) { + return 'wait'; + } + if (input.restartObserved === true) { + return 'reload'; + } + return input.sameVersionHealthyAttempts >= MAX_SAME_VERSION_HEALTHY_ATTEMPTS + ? 'reload' + : 'wait'; } if (input.reportedVersion && input.reportedVersion !== input.preUpdateVersion) { return 'reload'; } // Healthy but still the pre-update version: the about-to-exit process is - // answering, or this deployment never restarts. Wait, but not forever. + // answering, or this deployment never restarts. Wait, but not forever, + // and only once completion is confirmed. Unconfirmed, the modal's stall + // state offers a manual reload instead. + if (input.completionConfirmed === false) { + return 'wait'; + } if (input.sameVersionHealthyAttempts >= MAX_SAME_VERSION_HEALTHY_ATTEMPTS) { return 'reload'; } return 'wait'; }; + +// The progress stream can go quiet without ever erroring: a reverse proxy or +// compressing intermediary can hold events back, and a held connection never +// reports the old process exiting. After this much SSE silence during an +// in-progress update the modal also polls /api/updates/status, while leaving +// the stream open in case it recovers. +export const UPDATE_STREAM_SILENCE_FALLBACK_MS = 6000; + +// Cadence of the /api/updates/status fallback poll. +export const UPDATE_STATUS_POLL_INTERVAL_MS = 2500; + +// With no new stage, progress, or restart signal for this long, stop implying +// the update is still running and offer a reload so the user can check. +export const UPDATE_PROGRESS_STALL_TIMEOUT_MS = 120_000; + +const UPDATE_IN_PROGRESS_STAGES = new Set([ + 'downloading', + 'verifying', + 'extracting', + 'backing-up', + 'applying', + 'restoring', +]); + +// Consecutive failed status polls, with the stream closed and the update at +// a stage that restarts the service, before the modal starts probing for the +// restarted backend. +export const RESTART_SUSPECT_POLL_FAILURES = 2; + +// Stages after which the service restarts on success. +const UPDATE_LATE_STAGES = new Set(['applying', 'restoring', 'restarting', 'completed']); + +export const isLateUpdateStage = (status: string | undefined): boolean => + status !== undefined && UPDATE_LATE_STAGES.has(status); + +// Update checks share the status channel with the apply lifecycle. +const UPDATE_CHECK_STATUSES = new Set(['checking', 'available']); + +export const isUpdateInProgressStage = (status: string | undefined): boolean => + status !== undefined && UPDATE_IN_PROGRESS_STAGES.has(status); + +// SSE and polling both feed the modal, and a rate-limited or slow poll can +// answer with a stage the stream already moved past. Once an apply is under +// way, ignore update-check chatter and in-progress stages that would move +// the progress backwards. +export const shouldApplyUpdateStatus = ( + current: { status: string; progress: number } | null, + next: { status: string; progress: number }, +): boolean => { + if (!current || !isUpdateInProgressStage(current.status)) { + return true; + } + if (UPDATE_CHECK_STATUSES.has(next.status)) { + return false; + } + if (isUpdateInProgressStage(next.status) && next.progress < current.progress) { + return false; + } + return true; +}; diff --git a/internal/api/updates.go b/internal/api/updates.go index 185d586c78..0d77b35193 100644 --- a/internal/api/updates.go +++ b/internal/api/updates.go @@ -481,10 +481,15 @@ func (h *UpdateHandlers) HandleUpdateStream(w http.ResponseWriter, r *http.Reque return } - // Set SSE headers + // Set SSE headers. no-transform and X-Accel-Buffering keep reverse + // proxies (nginx, Cloudflare, compressing ingresses) from buffering or + // re-encoding the stream: a buffered stream looks exactly like an update + // that stopped reporting progress. The gzip middleware already skips + // text/event-stream, so this is the only encoding the stream ever has. w.Header().Set("Content-Type", "text/event-stream") - w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Cache-Control", "no-cache, no-transform") w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") w.Header().Set("Access-Control-Allow-Origin", "*") // Generate client ID @@ -503,9 +508,10 @@ func (h *UpdateHandlers) HandleUpdateStream(w http.ResponseWriter, r *http.Reque Str("client_ip", clientIP). Msg("Update progress SSE stream started") - // Send initial connection message - fmt.Fprintf(w, ": connected\n\n") - client.Flusher.Flush() + // AddSSEClient has already written the connection preamble and the + // current status and flushed them. All further writes go through the + // broadcaster under the client's lock; writing here as well would race + // the broadcaster on a ResponseWriter that is not safe for concurrent use. // Wait for client disconnect or context cancellation select { diff --git a/internal/api/updates_test.go b/internal/api/updates_test.go index fe5ab7a706..08c8a6e64e 100644 --- a/internal/api/updates_test.go +++ b/internal/api/updates_test.go @@ -1,12 +1,15 @@ package api import ( + "bytes" "context" "encoding/json" "errors" + "fmt" "net/http" "net/http/httptest" "strings" + "sync" "testing" "time" @@ -645,6 +648,165 @@ func TestHandleUpdateStream(t *testing.T) { } } +// streamFlushRecorder is a goroutine-safe ResponseWriter that records what +// had actually been flushed to the wire at each Flush, so tests can tell a +// flushed SSE frame from one sitting in a buffer. +type streamFlushRecorder struct { + mu sync.Mutex + header http.Header + code int + written bytes.Buffer + flushed string + flushCh chan struct{} +} + +func newStreamFlushRecorder() *streamFlushRecorder { + return &streamFlushRecorder{header: make(http.Header), flushCh: make(chan struct{}, 64)} +} + +func (r *streamFlushRecorder) Header() http.Header { return r.header } + +func (r *streamFlushRecorder) WriteHeader(code int) { + r.mu.Lock() + defer r.mu.Unlock() + if r.code == 0 { + r.code = code + } +} + +func (r *streamFlushRecorder) Write(p []byte) (int, error) { + r.mu.Lock() + defer r.mu.Unlock() + if r.code == 0 { + r.code = http.StatusOK + } + return r.written.Write(p) +} + +func (r *streamFlushRecorder) Flush() { + r.mu.Lock() + r.flushed = r.written.String() + r.mu.Unlock() + select { + case r.flushCh <- struct{}{}: + default: + } +} + +func (r *streamFlushRecorder) flushedBody() string { + r.mu.Lock() + defer r.mu.Unlock() + return r.flushed +} + +// waitForFlushed blocks until the flushed body contains every needle. +func (r *streamFlushRecorder) waitForFlushed(t *testing.T, needles ...string) string { + t.Helper() + deadline := time.After(2 * time.Second) + for { + body := r.flushedBody() + missing := false + for _, needle := range needles { + if !strings.Contains(body, needle) { + missing = true + break + } + } + if !missing { + return body + } + select { + case <-r.flushCh: + case <-time.After(20 * time.Millisecond): + case <-deadline: + t.Fatalf("timed out waiting for flushed %q, flushed so far:\n%s", needles, body) + } + } +} + +// The update modal stalled at "Downloading 10%" when the progress stream +// went quiet. The stream must advertise itself as unbufferable, put the +// current status on the wire the moment a client connects, and flush every +// later stage in order. +func TestHandleUpdateStream_HeadersImmediateStatusAndFlushedStages(t *testing.T) { + broadcaster := updates.NewSSEBroadcaster() + defer broadcaster.Close() + + broadcaster.Broadcast(updates.UpdateStatus{Status: "downloading", Progress: 10, Message: "Downloading update..."}) + + h := NewUpdateHandlers(&MockUpdateManager{ + AddSSEClientFunc: broadcaster.AddClient, + RemoveSSEClientFunc: broadcaster.RemoveClient, + }, nil) + + rec := newStreamFlushRecorder() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + req := httptest.NewRequest(http.MethodGet, "/api/updates/stream", nil).WithContext(ctx) + + done := make(chan struct{}) + go func() { + h.HandleUpdateStream(rec, req) + close(done) + }() + + first := rec.waitForFlushed(t, ": connected", `"status":"downloading"`, `"progress":10`) + if !strings.Contains(first, "data: {") || !strings.HasSuffix(first, "\n\n") { + t.Fatalf("initial status must be a complete, flushed SSE data frame, got:\n%s", first) + } + + if got := rec.Header().Get("Content-Type"); got != "text/event-stream" { + t.Fatalf("Content-Type = %q, want text/event-stream", got) + } + if got := rec.Header().Get("Cache-Control"); !strings.Contains(got, "no-cache") || !strings.Contains(got, "no-transform") { + t.Fatalf("Cache-Control = %q, want no-cache and no-transform", got) + } + if got := rec.Header().Get("X-Accel-Buffering"); got != "no" { + t.Fatalf("X-Accel-Buffering = %q, want no", got) + } + if got := rec.Header().Get("Content-Encoding"); got != "" { + t.Fatalf("Content-Encoding = %q, the stream must never be compressed", got) + } + + stages := []updates.UpdateStatus{ + {Status: "verifying", Progress: 25, Message: "Verifying signature..."}, + {Status: "verifying", Progress: 30, Message: "Verifying download..."}, + {Status: "extracting", Progress: 40, Message: "Extracting update..."}, + {Status: "verifying", Progress: 50, Message: "Validating new binary..."}, + {Status: "backing-up", Progress: 60, Message: "Creating backup..."}, + {Status: "applying", Progress: 80, Message: "Applying update..."}, + {Status: "restarting", Progress: 95, Message: "Restarting service..."}, + {Status: "completed", Progress: 100, Message: "Update completed, restarting..."}, + } + for _, stage := range stages { + broadcaster.Broadcast(stage) + } + + body := rec.waitForFlushed(t, `"status":"completed"`) + last := strings.Index(body, `"progress":10,`) + for _, stage := range stages { + marker := fmt.Sprintf(`"progress":%d,`, stage.Progress) + idx := strings.Index(body, marker) + if idx < 0 { + t.Fatalf("stage %s %d never reached the wire:\n%s", stage.Status, stage.Progress, body) + } + if idx < last { + t.Fatalf("stage %s %d was delivered out of order:\n%s", stage.Status, stage.Progress, body) + } + last = idx + } + + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("HandleUpdateStream didn't return after client disconnect") + } + if n := broadcaster.GetClientCount(); n != 0 { + t.Fatalf("expected disconnected client to be removed, %d remain", n) + } +} + func TestHandleUpdateStream_StreamingNotSupported(t *testing.T) { mockManager := &MockUpdateManager{ AddSSEClientFunc: func(w http.ResponseWriter, clientID string) *updates.SSEClient { diff --git a/internal/updates/manager.go b/internal/updates/manager.go index 68f8904697..b579af250f 100644 --- a/internal/updates/manager.go +++ b/internal/updates/manager.go @@ -298,7 +298,7 @@ func NewManager(cfg *config.Config) *Manager { // Clean up old temp directories and stale update backups from previous runs. go m.cleanupOldUpdateArtifacts() - // Start heartbeat for SSE connections (every 30 seconds) + // Start heartbeat for SSE connections (see sseHeartbeatInterval) m.heartbeatWg.Add(1) go m.sseHeartbeatLoop() @@ -2280,11 +2280,15 @@ func (m *Manager) updateStatus(status string, progress int, message string, err } } +// sseHeartbeatInterval keeps idle update streams under common reverse-proxy +// read timeouts (nginx defaults to 60s) during long download stages. +const sseHeartbeatInterval = 15 * time.Second + // sseHeartbeatLoop sends periodic heartbeats to SSE clients func (m *Manager) sseHeartbeatLoop() { defer m.heartbeatWg.Done() - ticker := time.NewTicker(30 * time.Second) + ticker := time.NewTicker(sseHeartbeatInterval) defer ticker.Stop() for { diff --git a/internal/updates/sse.go b/internal/updates/sse.go index bd7a20bbe7..700790efb8 100644 --- a/internal/updates/sse.go +++ b/internal/updates/sse.go @@ -11,6 +11,10 @@ import ( "github.com/rs/zerolog/log" ) +// sseClientQueueSize bounds how many undelivered statuses a single slow +// client may hold before new ones are dropped for that client. +const sseClientQueueSize = 64 + // SSEClient represents a Server-Sent Events client type SSEClient struct { ID string @@ -19,6 +23,9 @@ type SSEClient struct { Done chan bool LastActive time.Time mu sync.Mutex // protects writes to Writer and Flusher + // queue delivers broadcasts to this client in order through a single + // writer goroutine. Nil for clients not registered via AddClient. + queue chan UpdateStatus } // SSEBroadcaster manages Server-Sent Events connections for update progress @@ -77,6 +84,7 @@ func (b *SSEBroadcaster) AddClient(w http.ResponseWriter, clientID string) *SSEC Flusher: flusher, Done: make(chan bool, 1), LastActive: time.Now(), + queue: make(chan UpdateStatus, sseClientQueueSize), } b.mu.Lock() @@ -93,18 +101,59 @@ func (b *SSEBroadcaster) AddClient(w http.ResponseWriter, clientID string) *SSEC Int("total_clients", clientCount). Msg("SSE client connected") - // Send the current cached status immediately to the new client + // Write the connection preamble and the current status synchronously, + // before the ordered writer starts, so a (re)connecting client always + // sees where the update is right now instead of waiting for the next + // stage change. The cache is read after registration so a broadcast + // racing this connect is either in the snapshot or queued behind it. b.statusMu.RLock() cachedStatus := b.cachedStatus b.statusMu.RUnlock() - go func() { - b.sendToClient(client, cachedStatus) - }() + if data, err := json.Marshal(cachedStatus); err == nil { + b.writeToClient(client, ": connected\n\n"+formatSSEData(data)) + } else { + log.Error().Err(err).Str("client_id", clientID).Msg("Failed to marshal initial status for SSE") + b.writeToClient(client, ": connected\n\n") + } + + go b.clientWriteLoop(client) return client } +// clientWriteLoop drains a client's queue so broadcasts reach it in the +// order they were emitted. A goroutine per message (the previous design) +// let a later stage overtake an earlier one on the wire. +func (b *SSEBroadcaster) clientWriteLoop(client *SSEClient) { + for { + select { + case <-client.Done: + return + case status := <-client.queue: + b.sendToClient(client, status) + } + } +} + +// enqueueForClient hands a status to the client's ordered writer without +// blocking the broadcast loop on a slow connection. +func (b *SSEBroadcaster) enqueueForClient(client *SSEClient, status UpdateStatus) { + if client.queue == nil { + go b.sendToClient(client, status) + return + } + select { + case <-client.Done: + case client.queue <- status: + default: + log.Warn(). + Str("client_id", client.ID). + Str("status", status.Status). + Msg("SSE client queue full, dropping update status for this client") + } +} + // RemoveClient unregisters an SSE client func (b *SSEBroadcaster) RemoveClient(clientID string) { b.mu.Lock() @@ -176,9 +225,10 @@ func (b *SSEBroadcaster) broadcastLoop() { } b.mu.RUnlock() - // Send to all clients in parallel + // Each client has its own ordered writer, so a slow client + // neither blocks the others nor receives stages out of order. for _, client := range clients { - go b.sendToClient(client, status) + b.enqueueForClient(client, status) } log.Debug(). @@ -206,10 +256,19 @@ func (b *SSEBroadcaster) sendToClient(client *SSEClient, status UpdateStatus) { return } - // Write SSE message format - // Format: data: {json}\n\n - message := fmt.Sprintf("data: %s\n\n", string(data)) + b.writeToClient(client, formatSSEData(data)) +} +// formatSSEData frames a JSON payload as a single SSE data event. +func formatSSEData(data []byte) string { + return fmt.Sprintf("data: %s\n\n", string(data)) +} + +// writeToClient writes a raw SSE frame and flushes it immediately. Every +// write to a client goes through here under the client lock: the +// ResponseWriter is not safe for concurrent use, and an unflushed frame is +// indistinguishable from a silent stream on the browser side. +func (b *SSEBroadcaster) writeToClient(client *SSEClient, frame string) { // Acquire client lock to prevent concurrent writes client.mu.Lock() defer client.mu.Unlock() @@ -227,18 +286,19 @@ func (b *SSEBroadcaster) sendToClient(client *SSEClient, status UpdateStatus) { Str("client_id", client.ID). Interface("panic", r). Msg("Recovered from panic while sending to SSE client") - b.RemoveClient(client.ID) + go b.RemoveClient(client.ID) } }() - // Write to client - _, err = fmt.Fprint(client.Writer, message) - if err != nil { + if _, err := fmt.Fprint(client.Writer, frame); err != nil { log.Debug(). Err(err). Str("client_id", client.ID). Msg("Failed to write to SSE client, removing") - b.RemoveClient(client.ID) + // Removal takes the broadcaster lock; do it off this goroutine so a + // concurrent cleanup pass holding that lock while waiting on this + // client's lock cannot deadlock. + go b.RemoveClient(client.ID) return } @@ -332,28 +392,7 @@ func (b *SSEBroadcaster) sendHeartbeatToClient(c *SSEClient) { default: } - c.mu.Lock() - defer c.mu.Unlock() - - select { - case <-c.Done: - return - default: - } - - defer func() { - if r := recover(); r != nil { - b.RemoveClient(c.ID) - } - }() - - _, err := fmt.Fprint(c.Writer, ": heartbeat\n\n") - if err != nil { - b.RemoveClient(c.ID) - return - } - c.Flusher.Flush() - c.LastActive = time.Now() + b.writeToClient(c, ": heartbeat\n\n") } func closeDone(ch chan bool) { diff --git a/internal/updates/sse_test.go b/internal/updates/sse_test.go index 9bf3ffe11e..15da113887 100644 --- a/internal/updates/sse_test.go +++ b/internal/updates/sse_test.go @@ -1,8 +1,11 @@ package updates import ( + "net/http" "net/http/httptest" + "strconv" "strings" + "sync" "testing" "time" ) @@ -239,3 +242,93 @@ func TestSSEBroadcaster_CloseIsIdempotentAndPostCloseSafe(t *testing.T) { t.Fatal("expected no connected clients after close") } } + +// A (re)connecting update modal must see the current stage immediately, not +// after the next stage change, which during a long download can be minutes. +func TestSSEBroadcaster_AddClientWritesCurrentStatusBeforeReturning(t *testing.T) { + broadcaster := NewSSEBroadcaster() + defer broadcaster.Close() + + broadcaster.Broadcast(UpdateStatus{Status: "extracting", Progress: 40, Message: "Extracting update..."}) + + w := &mockFlushWriter{ResponseRecorder: httptest.NewRecorder()} + if client := broadcaster.AddClient(w, "client-now"); client == nil { + t.Fatal("AddClient returned nil") + } + + body := w.Body.String() + if !strings.HasPrefix(body, ": connected\n\n") { + t.Fatalf("expected connection preamble first, got %q", body) + } + if !strings.Contains(body, `data: {"status":"extracting","progress":40`) { + t.Fatalf("expected current status on connect, got %q", body) + } + if w.flushed < 1 { + t.Fatal("initial status must be flushed, not left buffered") + } +} + +// Broadcasts used to be sent from one goroutine per message, so stages +// emitted in quick succession could reach a client out of order. +func TestSSEBroadcaster_DeliversStagesInOrder(t *testing.T) { + broadcaster := NewSSEBroadcaster() + defer broadcaster.Close() + + w := &syncFlushWriter{} + if client := broadcaster.AddClient(w, "client-order"); client == nil { + t.Fatal("AddClient returned nil") + } + + const stages = 40 + for i := 1; i <= stages; i++ { + broadcaster.Broadcast(UpdateStatus{Status: "verifying", Progress: i}) + } + + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) && !strings.Contains(w.String(), `"progress":40,`) { + time.Sleep(5 * time.Millisecond) + } + body := w.String() + last := -1 + for i := 1; i <= stages; i++ { + idx := strings.Index(body, `"progress":`+strconv.Itoa(i)+`,`) + if idx < 0 { + t.Fatalf("stage %d never delivered:\n%s", i, body) + } + if idx < last { + t.Fatalf("stage %d delivered out of order:\n%s", i, body) + } + last = idx + } +} + +type syncFlushWriter struct { + mu sync.Mutex + header http.Header + buf strings.Builder +} + +func (w *syncFlushWriter) Header() http.Header { + w.mu.Lock() + defer w.mu.Unlock() + if w.header == nil { + w.header = make(http.Header) + } + return w.header +} + +func (w *syncFlushWriter) Write(p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + return w.buf.Write(p) +} + +func (w *syncFlushWriter) WriteHeader(int) {} + +func (w *syncFlushWriter) Flush() {} + +func (w *syncFlushWriter) String() string { + w.mu.Lock() + defer w.mu.Unlock() + return w.buf.String() +}