mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-23 20:25:56 +00:00
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.
This commit is contained in:
courtmanr@gmail.com
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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`
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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;
|
||||
});
|
||||
@@ -0,0 +1,11 @@
|
||||
<!doctype html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="UTF-8" />
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1" />
|
||||
</head>
|
||||
<body>
|
||||
<div id="root"></div>
|
||||
<script type="module" src="/browser-tests/update-progress-modal.tsx"></script>
|
||||
</body>
|
||||
</html>
|
||||
@@ -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<string, unknown>;
|
||||
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 ? (
|
||||
<main class="p-6">
|
||||
<p data-testid="reloaded">Reloaded after update (boot {boots})</p>
|
||||
</main>
|
||||
) : (
|
||||
<main class="p-6">
|
||||
<h1 class="text-lg font-semibold">Update progress harness</h1>
|
||||
<UpdateProgressModal
|
||||
isOpen={true}
|
||||
onClose={() => undefined}
|
||||
onViewHistory={() => undefined}
|
||||
connected={() => true}
|
||||
reconnecting={() => false}
|
||||
/>
|
||||
</main>
|
||||
),
|
||||
document.getElementById('root')!,
|
||||
);
|
||||
@@ -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=<playwright 1.56.1 node_modules> PULSE_BROWSER_ARTIFACTS=<proof dir> PULSE_CHROMIUM_PATH=<cached chrome-headless-shell 151.0.7922.34> 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",
|
||||
|
||||
@@ -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) {
|
||||
</Show>
|
||||
</CalloutCard>
|
||||
</Show>
|
||||
<Show when={!isRestarting()}>
|
||||
<Show when={!isRestarting() && !progressStalled()}>
|
||||
<CalloutCard
|
||||
tone="warning"
|
||||
scale="compact"
|
||||
@@ -480,6 +718,32 @@ export function UpdateProgressModal(props: UpdateProgressModalProps) {
|
||||
}
|
||||
/>
|
||||
</Show>
|
||||
<Show when={progressStalled()}>
|
||||
<CalloutCard
|
||||
tone="warning"
|
||||
scale="compact"
|
||||
padding="md"
|
||||
class="mt-6"
|
||||
icon={<AlertTriangleIcon class="h-5 w-5" aria-hidden="true" />}
|
||||
title="No progress reported for a while"
|
||||
description={
|
||||
<span class="text-sm">
|
||||
The update may have finished already. Reload the page to check. Pulse keeps
|
||||
updating on the server either way.
|
||||
</span>
|
||||
}
|
||||
>
|
||||
<Button
|
||||
onClick={() => window.location.reload()}
|
||||
variant="primary"
|
||||
size="sm"
|
||||
class="mt-2"
|
||||
type="button"
|
||||
>
|
||||
Reload to check
|
||||
</Button>
|
||||
</CalloutCard>
|
||||
</Show>
|
||||
</Show>
|
||||
</div>
|
||||
|
||||
|
||||
@@ -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<UpdateStatus>) {
|
||||
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(() => (
|
||||
<UpdateProgressModal
|
||||
isOpen={true}
|
||||
onClose={() => undefined}
|
||||
onViewHistory={() => undefined}
|
||||
connected={() => true}
|
||||
reconnecting={() => false}
|
||||
/>
|
||||
));
|
||||
|
||||
describe('UpdateProgressModal progress feed', () => {
|
||||
const originalEventSource = globalThis.EventSource;
|
||||
const originalLocation = window.location;
|
||||
let reloadMock: ReturnType<typeof vi.fn>;
|
||||
|
||||
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<UpdateStatus | Error> = [
|
||||
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);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -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');
|
||||
});
|
||||
|
||||
@@ -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;
|
||||
};
|
||||
|
||||
+11
-5
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
+75
-36
@@ -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) {
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user