diff --git a/CHANGELOG.md b/CHANGELOG.md index 56f165740..a7adb50ee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,8 @@ not per individual commit. - A spent launch address in the tab falls back to the kept session and says so, and a first load that failed, of an opened or a pasted address, can be retried with a reload. - The workflow canvas API contract says where the session token may be kept (`sessionStorage` only) and requires the served page to send a `Content-Security-Policy` header with `frame-ancestors 'none'`. - The workflow canvas uses the Konductor docs theme, with light and dark modes, and system colours in forced-colors mode. +- The Execution tab runs the workflow and shows live node states. +- A launch code trade that gets no answer shows "Still waiting for the server" with a way to stop, and ends with an error after 120 seconds. ### Changed diff --git a/workflow-canvas/README.md b/workflow-canvas/README.md index 7e104e521..73c7e0f7e 100644 --- a/workflow-canvas/README.md +++ b/workflow-canvas/README.md @@ -18,11 +18,12 @@ stub that serves it and the tests around both are in this folder. IO; `api/` is the typed client of the API; `launch/` and `workflow/` open the session and load the workflow; `designer/` and `canvas/` draw it; `execution/` runs it and shows the run; `execution/core/` does no IO and reads a run - once a second. A read that fails with no answer or a server error is tried + once a second. A read that fails with no answer, a server error, 408 or 429 is tried again, up to three in a row; any other failure ends the following. A start that finds a run already active follows that run. Every run call takes a - signal, and every call ends on its own after 15 seconds, so a server that - never answers cannot hold the tab. A failure counts only for the run it belongs to. + signal, and every call ends on its own, after 15 + seconds, or 120 seconds for the launch code trade (see **The launch code + trade** below), so a server that never answers cannot hold the tab. A failure counts only for the run it belongs to. - `api/`: the HTTP API contract (`openapi.yaml`) and its test. - `stub-api/`: a TypeScript stub server for the contract, for local development and tests. @@ -284,6 +285,20 @@ The rest wraps the core, so the rules can be tested without a browser: `frame-ancestors 'none'` added (see **The page** in `api/openapi.yaml`). The server's token lasts until it stops, with no idle timeout, so a kept token does not change how long a session lives. +- The launch code trade (`createSession`) is the one call with no short limit: + the code works once, so giving up after the server has taken it would leave + the page with no way in. It is bounded in two ways. After 5 seconds with no + answer, `launch/LaunchWait.tsx` shows "Still waiting for the server" above + the tabs, with **Stop waiting**, which aborts the trade of the first load + (`WorkflowSource.cancel`). After 120 seconds (twice the validity of the + code) the trade ends on its own. Stopping, the limit, and a dropped + connection all end in the same error state, which says that the server may + have used the code and to get a new launch code from the terminal that + printed the launch address. A reload does not retry the trade, since the + code is removed from the address. A late answer after the + notice appeared still opens the workflow. A pasted code is not cancellable: + the workflow or error already on the page stays, and the 120 seconds end the + trade. - `designer/` shows the result: loading, error, empty (no file yet, or a file with errors, with their messages), or the canvas. - `canvas/` draws the graph with [React Flow](https://reactflow.dev) @@ -306,6 +321,38 @@ intent, because the API needs one (1 to 2048 characters) to start a run. Run is off and the goal box is read-only from the click until the run ends. Read-only, not greyed out, so the box keeps focus and shows the goal of that run. Enter in the box starts a run, like the button. +- If the server answers 409 `already_running`, as it does when a run from + another tab is going, the tab follows the run named in that answer instead + of starting one. The line under the button says so, and the goal typed is not + used for that run. The API does not give the goal of a run, so while the tab + follows a run that was already going the Goal box is empty, says "Not known" + and is labelled so. When that run ends, the status line still says it was + already in progress. +- Resume works after a reload, though the launch code works once: the page + resumes from the session it kept (see above). If the server was started + again, the kept token is refused and the page asks for a fresh launch + address; the run goes on unseen. The tab keeps the id of the latest run in + the storage of the browser tab (`sessionStorage`; the id is no secret) and + drops it once the run has ended or turns out to be gone. When the page + opens again with a session it reads that run, trying again like any read. If the run is still running, the + tab follows it as if the person had joined it, without a click on Run. A run + that has ended, one that the server will not give (for instance after it + started again) or a browser that blocks storage leaves the tab ready for a + new run. If the readings fail for another reason, such as no answer, the tab + does not know whether the run is going, and shows the alert for a run that + may still be going. Pressing Run while the page still looks for the run + stops the look and starts a run; a run that is going answers 409 and is + joined. +- Pasting another launch address into the open tab lets go of the run: the + tab stops following it, shows no run status and forgets its id. The run goes + on at its server. The Goal box and Run belong to the workflow that is open. +- A read of the run that fails is tried again after a second; the tab gives up + after three failures in a row. Only a missing answer, a 5xx, 408 or 429 is tried again: + any 4xx or a reply the page cannot read shows the alert at once. A call that + gets no answer in 15 seconds counts as a failure, except the one that trades + the launch code, which has a limit of 120 seconds because the code works once. A start that + gets no answer, or is accepted in a reply the page cannot read, says that the + run may not have started and that Run joins it if it did. - Each node shows a badge with a word and a colour: **Queued** (with a dashed outline), **Running**, **Done** and **Failed**. The word carries the state; only Queued and Not run also differ in shape. The intent is Done once the run @@ -325,8 +372,10 @@ intent, because the API needs one (1 to 2048 characters) to start a run. alert. The goal is one line: line breaks in the outcome of the intent become spaces, and Enter that ends an input method composition starts no run. - A workflow can be saved while a run goes on, so the run may have other stages - than the graph. Node states attach by stage name, so a warning above the - graph names the stages that are not drawn and those that are not in the run. + than the graph. The graph is read once, when the page loads, while the run + works from the saved file. Node states attach by stage name, so a warning + above the graph names the stages that are not drawn and those that are not in + the run. - A run goes on while the Designer tab is open, and the tab shows it again when you come back, with the goal box as it was: what you typed, or the goal of the run in flight. Closing the page stops the reading, not the run. @@ -343,8 +392,12 @@ The code follows the ports-and-adapters split of the graph: fails if the core imports anything but its own modules, the graph core and the wire types, or reads a clock, randomness, the environment or the network. - The adapters are `createHttpApi` (`startRun`, `getRun`), which `loadWorkflow` - ties to the session token, and `timerScheduler`. The token is held only by - those run calls. + ties to the session token, `timerScheduler`, and `sessionRunMemory`, which + keeps the id of the latest run for `resumeRun`. The token is held only by + those run calls. `createHttpApi` ends a call after a timeout of 15 seconds, and the launch + code call after 120 seconds because it works once, and + `loadWorkflow` turns the 409 `already_running`, which carries the `run_id`, + into the core's `RunAlreadyActive` so the core can follow that run. - `useExecution` keeps the state in `App`, and `Execution` draws it. The Designer and Execution tabs share `WorkflowGate` (loading, error and empty states) and `WorkflowGraphCard` (the card with the graph). @@ -430,9 +483,11 @@ cannot be run, and in both theme modes. The stub runs one run at a time and never fails a stage. So the one test that starts a real run on it (`execution.spec.ts`, the first run test) waits for the -end of the run, and the tests of a failed stage or a refused run answer the -run calls themselves (`e2e/support/run-routes.ts`). A test that starts another -run on the stub must also wait for its end, or the next test gets a 409. +end of the run, and the tests of a failed stage, a refused run, a joined run +and a dropped read answer the run calls themselves (`e2e/support/run-routes.ts`). +A test that starts another run on the stub must also wait for its end, or the +next test gets a 409. For the same reason, if that one test fails and Playwright +retries it, the retry gets a 409 and follows the run still going. Shared page objects and helpers live in `e2e/support/`. New tests should extend them rather than repeat selectors: import `test` and `expect` from diff --git a/workflow-canvas/e2e/support/run-routes.ts b/workflow-canvas/e2e/support/run-routes.ts index ee1896c1b..347754611 100644 --- a/workflow-canvas/e2e/support/run-routes.ts +++ b/workflow-canvas/e2e/support/run-routes.ts @@ -81,3 +81,40 @@ export async function refuseRunsWith( route.fulfill({ status, json: body }), ); } + +/** + * Makes the page's call to start a run answer 409 `already_running` with the + * id of a run, and its reads answer with `reply`, as a server does while a + * run from an earlier page load is still going. + */ +export async function joinActiveRunWith( + page: Page, + reply: ReturnType, +): Promise { + await page.route('**/api/run', (route) => + route.fulfill({ + status: 409, + json: { + error: 'A run is already active.', + code: 'already_running', + run_id: RUN_ID, + }, + }), + ); + await page.route('**/api/runs/*', (route) => route.fulfill({ json: reply })); +} + +/** + * Makes the first read of a run fail as if the connection dropped, then + * answers every read with `reply`. + */ +export async function dropFirstReadThenAnswerWith( + page: Page, + reply: ReturnType, +): Promise { + let reads = 0; + await page.route('**/api/runs/*', (route) => { + reads += 1; + return reads === 1 ? route.abort() : route.fulfill({ json: reply }); + }); +} diff --git a/workflow-canvas/e2e/tests/execution.spec.ts b/workflow-canvas/e2e/tests/execution.spec.ts index 2e829d29a..eb9be9f89 100644 --- a/workflow-canvas/e2e/tests/execution.spec.ts +++ b/workflow-canvas/e2e/tests/execution.spec.ts @@ -3,6 +3,8 @@ import { expect, test } from '../support/fixtures'; import { answerRunsInTurn, answerRunsWith, + dropFirstReadThenAnswerWith, + joinActiveRunWith, refuseRunsWith, runReply, } from '../support/run-routes'; @@ -61,6 +63,33 @@ test('runs the workflow on the stub: each stage goes from queued to running to d await expect(canvas.runError).toHaveCount(0); }); +test('resumes the run that is going when the page is reloaded, though the launch code works once', async ({ + canvas, +}) => { + const sessionCalls: string[] = []; + await canvas.runButton.click(); + await expect(canvas.badge('node-stage-draft')).toHaveText('Running', { + timeout: STEP_TIMEOUT_MS, + }); + canvas.page.on('request', (request) => { + if (request.url().endsWith('/api/session')) + sessionCalls.push(request.url()); + }); + + await canvas.page.reload(); + await canvas.title.waitFor(); + await canvas.selectTab('Execution'); + + await expect(canvas.runStatus).toContainText( + 'Following the run that was already in progress.', + ); + await expect(canvas.runButton).toBeDisabled(); + await expect(canvas.badge('node-end')).toHaveText('Passed', { + timeout: STEP_TIMEOUT_MS * 2, + }); + expect(sessionCalls).toEqual([]); +}); + test('announces the status in a polite live region and names the state of each node', async ({ canvas, }) => { @@ -139,6 +168,42 @@ test('shows the error of the API in one alert when the run does not start', asyn await expect(canvas.runButton).toBeEnabled(); }); +test('follows the run that is already active when the server answers 409 with its id', async ({ + canvas, +}) => { + await joinActiveRunWith( + canvas.page, + runReply('running', ['running', 'pending']), + ); + + await canvas.runButton.click(); + + await canvas.expectBadges(['Done', 'Running', 'Queued', 'Queued']); + await expect(canvas.runStatus).toContainText( + 'Following the run that was already in progress. Running draft.', + ); + await expect(canvas.runError).toHaveCount(0); + await expect(canvas.runButton).toBeDisabled(); +}); + +test('keeps following the run when one read of it fails', async ({ + canvas, +}) => { + await answerRunsWith( + canvas.page, + runReply('running', ['running', 'pending']), + ); + await dropFirstReadThenAnswerWith( + canvas.page, + runReply('passed', ['passed', 'passed']), + ); + + await canvas.runButton.click(); + + await canvas.expectBadges(['Done', 'Done', 'Done', 'Passed']); + await expect(canvas.runError).toHaveCount(0); +}); + test('says the run may still be going when it can no longer be followed', async ({ canvas, }) => { @@ -254,3 +319,85 @@ test('starts the run when Enter is pressed in the goal box, and keeps focus ther await expect(canvas.runStatus).toContainText('Running draft.'); await expect(canvas.goal).toBeFocused(); }); + +test('shows the run that is still going when the page is opened again', async ({ + canvas, +}) => { + await answerRunsWith( + canvas.page, + runReply('running', ['running', 'pending']), + ); + await canvas.runButton.click(); + await expect(canvas.runStatus).toContainText('Running draft.'); + + await canvas.page.reload(); + await canvas.title.waitFor(); + await canvas.selectTab('Execution'); + + await canvas.expectBadges(['Done', 'Running', 'Queued', 'Queued']); + await expect(canvas.runStatus).toContainText( + 'Following the run that was already in progress. Running draft.', + ); + await expect(canvas.runButton).toBeDisabled(); + await expect(canvas.goal).toHaveValue(''); + await expect(canvas.goal).toHaveAttribute('placeholder', 'Not known'); +}); + +test('shows Ready when the page is opened again after the run has ended', async ({ + canvas, +}) => { + await answerRunsWith(canvas.page, runReply('passed', ['passed', 'passed'])); + await canvas.runButton.click(); + await canvas.expectBadges(['Done', 'Done', 'Done', 'Passed']); + + await canvas.page.reload(); + await canvas.title.waitFor(); + await canvas.selectTab('Execution'); + + await expect(canvas.runStatus).toContainText('Ready. Press Run to start.'); + await canvas.expectBadges([null, null, null, null]); + await expect(canvas.runButton).toBeEnabled(); +}); + +test('has no Execution tab after a reload when the server no longer knows the session', async ({ + canvas, +}) => { + await canvas.expectWorkflowDrawn(); + await canvas.page.route('**/api/workflow', (route) => + route.fulfill({ + status: 401, + json: { + error: 'The session token is missing or wrong.', + code: 'unauthorized', + }, + }), + ); + + await canvas.page.reload(); + await canvas.title.waitFor(); + + await expect( + canvas.page.getByText(/server no longer knows this session/), + ).toBeVisible(); + await canvas.selectTab('Execution'); + await expect(canvas.runButton).toHaveCount(0); +}); + +test('drops the session when the run start is answered 401, so a reload does not reuse it', async ({ + canvas, +}) => { + await refuseRunsWith(canvas.page, 401, { + error: 'The session token is missing or wrong.', + code: 'unauthorized', + }); + + await canvas.runButton.click(); + + await expect(canvas.runError).toContainText( + 'The server no longer knows this session', + ); + await expect(canvas.runError).toContainText('fresh launch address'); + expect(await canvas.page.evaluate(() => window.sessionStorage.length)).toBe( + 0, + ); +}); diff --git a/workflow-canvas/e2e/tests/launch-wait.spec.ts b/workflow-canvas/e2e/tests/launch-wait.spec.ts new file mode 100644 index 000000000..46cf13081 --- /dev/null +++ b/workflow-canvas/e2e/tests/launch-wait.spec.ts @@ -0,0 +1,64 @@ +// SPDX-License-Identifier: Apache-2.0 +import { expect, test, type Page } from '@playwright/test'; +import { STUB_LAUNCH_CODE } from '../../playwright.shared'; +import { sharedSession } from '../support/session'; +import { WorkflowCanvasPage } from '../support/workflow-canvas-page'; + +// The server takes the launch code and never answers. +async function openAgainstSilentServer(page: Page) { + await page.clock.install(); + await page.route('**/api/session', () => new Promise(() => {})); + await page.goto(`/#${STUB_LAUNCH_CODE}`); +} + +test('says it is still waiting, and stopping ends in an error that names the next step', async ({ + page, +}) => { + await openAgainstSilentServer(page); + await expect(page.getByText('Loading the workflow')).toBeVisible(); + await expect(page.getByTestId('launch-wait')).toHaveCount(0); + + await page.clock.fastForward(5_000); + await expect(page.getByTestId('launch-wait')).toContainText( + 'Still waiting for the server', + ); + await page.getByRole('button', { name: 'Stop waiting' }).click(); + + await expect(page.getByText('The workflow did not load')).toBeVisible(); + await expect(page.getByText(/may have used the launch code/)).toBeVisible(); + await expect(page.getByTestId('launch-wait')).toHaveCount(0); +}); + +test('ends in the same error at the hard limit', async ({ page }) => { + await openAgainstSilentServer(page); + + await page.clock.fastForward(121_000); + + await expect(page.getByText('The workflow did not load')).toBeVisible(); + await expect(page.getByText(/may have used it/)).toBeVisible(); +}); + +test('opens the workflow when the answer comes after the notice', async ({ + page, +}) => { + await page.clock.install(); + let answer: () => void = () => {}; + await page.route( + '**/api/session', + (route) => + new Promise((resolve) => { + answer = () => { + void route.fulfill({ json: sharedSession() }).then(resolve); + }; + }), + ); + await page.goto(`/#${STUB_LAUNCH_CODE}`); + await page.clock.fastForward(5_000); + await expect(page.getByTestId('launch-wait')).toBeVisible(); + + answer(); + + const canvas = new WorkflowCanvasPage(page); + await canvas.expectWorkflowDrawn(); + await expect(page.getByTestId('launch-wait')).toHaveCount(0); +}); diff --git a/workflow-canvas/src/App.tsx b/workflow-canvas/src/App.tsx index 91c4a4a29..3a25017f3 100644 --- a/workflow-canvas/src/App.tsx +++ b/workflow-canvas/src/App.tsx @@ -9,7 +9,7 @@ import './chrome/chrome.css'; import { TOP_BAR_ID, TopBar } from './chrome/TopBar'; import { Designer } from './designer/Designer'; import { useSettled } from './designer/use-settled'; -import type { Scheduler } from './execution/core'; +import type { RunMemory, Scheduler } from './execution/core'; import { Execution } from './execution/Execution'; import { RunAnnouncements } from './execution/RunAnnouncements'; import { timerScheduler } from './execution/timer-scheduler'; @@ -29,11 +29,14 @@ export function App({ load, notice, scheduler = timerScheduler, + runMemory, }: { load: Promise; /** Shown above the tabs, such as the refusal of a pasted launch address. */ notice?: ReactNode; scheduler?: Scheduler; + /** Keeps the latest run across a reload; a page without it starts afresh. */ + runMemory?: RunMemory; }) { // Read here, not in the tab, so switching tabs does not load again. const result = useSettled(load); @@ -42,6 +45,7 @@ export function App({ const execution = useExecution( result?.ok ? result.workflow.runs : undefined, scheduler, + runMemory, ); const [goal, setGoal] = useGoal(result); const [mode, toggleMode] = useThemeMode(); diff --git a/workflow-canvas/src/api/api.ts b/workflow-canvas/src/api/api.ts index c113d59d5..99b34c249 100644 --- a/workflow-canvas/src/api/api.ts +++ b/workflow-canvas/src/api/api.ts @@ -6,8 +6,11 @@ import type { RunView, Session, WorkflowDocument } from './types'; * this port; `createHttpApi` is the adapter that talks to the server. */ export interface WorkflowApi { - /** Trades the launch code for a session. */ - createSession(code: string): Promise; + /** + * Trades the launch code for a session. The code works once, so the call + * has a far longer limit than the others; aborting the signal ends it. + */ + createSession(code: string, signal?: AbortSignal): Promise; /** Loads the root workflow of the session. */ getWorkflow(token: string): Promise; /** diff --git a/workflow-canvas/src/api/http-api-launch.test.ts b/workflow-canvas/src/api/http-api-launch.test.ts new file mode 100644 index 000000000..bd6253d4b --- /dev/null +++ b/workflow-canvas/src/api/http-api-launch.test.ts @@ -0,0 +1,135 @@ +// SPDX-License-Identifier: Apache-2.0 +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { ApiError } from './api'; +import { createHttpApi, LAUNCH_CODE_TIMEOUT_MS } from './http-api'; + +const SESSION_BODY = { + token: 't-1', + root: 'hello', + harness: 'kiro-cli', + api_version: 1, +}; + +function hangingFetch() { + let aborted = false; + let answer: (response: Response) => void = () => {}; + const fetchFn = vi.fn( + (_path: string, init?: RequestInit) => + new Promise((resolve, reject) => { + answer = resolve; + init?.signal?.addEventListener('abort', () => { + aborted = true; + reject(new Error('aborted')); + }); + }), + ); + return { + fetchFn: fetchFn as unknown as typeof fetch, + isAborted: () => aborted, + answer: (body: unknown) => answer(new Response(JSON.stringify(body))), + }; +} + +describe('the launch code trade', () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it('is not cut short at the ordinary timeout, and a late answer works', async () => { + vi.useFakeTimers(); + const server = hangingFetch(); + const api = createHttpApi(server.fetchFn, 5000); + + const session = api.createSession('code-1'); + await vi.advanceTimersByTimeAsync(60_000); + expect(server.isAborted()).toBe(false); + server.answer(SESSION_BODY); + + await expect(session).resolves.toMatchObject({ token: 't-1' }); + expect(vi.getTimerCount()).toBe(0); + }); + + it('ends at its own hard limit with a message that says the code may be spent', async () => { + vi.useFakeTimers(); + const server = hangingFetch(); + const api = createHttpApi(server.fetchFn); + + const session = api.createSession('code-1'); + const outcome = expect(session).rejects.toMatchObject({ + status: 0, + message: expect.stringContaining('may have used it'), + }); + await vi.advanceTimersByTimeAsync(LAUNCH_CODE_TIMEOUT_MS - 1); + expect(server.isAborted()).toBe(false); + await vi.advanceTimersByTimeAsync(1); + + await outcome; + expect(server.isAborted()).toBe(true); + expect(vi.getTimerCount()).toBe(0); + }); + + it('ends at once when the signal aborts, and says the person stopped', async () => { + vi.useFakeTimers(); + const server = hangingFetch(); + const api = createHttpApi(server.fetchFn); + const stop = new AbortController(); + + const session = api.createSession('code-1', stop.signal); + const outcome = expect(session).rejects.toMatchObject({ + status: 0, + message: expect.stringContaining('You stopped waiting'), + }); + stop.abort(); + + await outcome; + expect(server.isAborted()).toBe(true); + expect(vi.getTimerCount()).toBe(0); + }); + + it('says the code may be spent when the connection drops', async () => { + const fetchFn = vi.fn(() => Promise.reject(new TypeError('network'))); + const api = createHttpApi(fetchFn as unknown as typeof fetch); + + await expect(api.createSession('code-1')).rejects.toMatchObject({ + status: 0, + message: expect.stringContaining('may have used it'), + }); + }); + + it('keeps the server sentence when the server refuses the code', async () => { + const fetchFn = vi.fn(() => + Promise.resolve( + new Response( + JSON.stringify({ error: 'The launch code is not valid.' }), + { + status: 401, + }, + ), + ), + ); + const api = createHttpApi(fetchFn as unknown as typeof fetch); + + const failure = await api.createSession('code-1').catch((e: unknown) => e); + + expect(failure).toBeInstanceOf(ApiError); + expect(failure).toMatchObject({ + status: 401, + message: 'The launch code is not valid.', + }); + }); + + it('leaves the ordinary calls at their own timeout', async () => { + vi.useFakeTimers(); + const server = hangingFetch(); + const api = createHttpApi(server.fetchFn, 5000); + + const read = api.getRun('t', 'r'); + const outcome = expect(read).rejects.toMatchObject({ + status: 0, + message: 'The server did not answer. Check that it is still running.', + }); + await vi.advanceTimersByTimeAsync(5000); + + await outcome; + }); +}); diff --git a/workflow-canvas/src/api/http-api.test.ts b/workflow-canvas/src/api/http-api.test.ts index b69a84cf4..891506d10 100644 --- a/workflow-canvas/src/api/http-api.test.ts +++ b/workflow-canvas/src/api/http-api.test.ts @@ -634,6 +634,31 @@ describe('createHttpApi, when a call does not go through', () => { await outcome; }); + it('does not cut the launch code short at the ordinary timeout', async () => { + vi.useFakeTimers(); + let answer: (response: Response) => void = () => {}; + let aborted = false; + const fetchFn = vi.fn( + (_path: string, init?: RequestInit) => + new Promise((resolve, reject) => { + answer = resolve; + init?.signal?.addEventListener('abort', () => { + aborted = true; + reject(new Error('aborted')); + }); + }), + ); + const api = createHttpApi(fetchFn as unknown as typeof fetch, 5000); + + const session = api.createSession('code-1'); + await vi.advanceTimersByTimeAsync(60_000); + expect(aborted).toBe(false); + answer(reply(SESSION_BODY)); + + await expect(session).resolves.toMatchObject({ token: 't-1' }); + expect(vi.getTimerCount()).toBe(0); + }); + it('stops its timer when the call ends', async () => { vi.useFakeTimers(); const { api } = apiFor(reply(RUN_BODY)); diff --git a/workflow-canvas/src/api/http-api.ts b/workflow-canvas/src/api/http-api.ts index 01c4630d8..21fc2c158 100644 --- a/workflow-canvas/src/api/http-api.ts +++ b/workflow-canvas/src/api/http-api.ts @@ -22,6 +22,19 @@ const RUN_ACCEPTED_UNREADABLE = /** How long a call may take before the page gives up on it. */ export const REQUEST_TIMEOUT_MS = 15_000; +/** + * How long the launch code trade may take. The code works once, so this is far + * longer than an ordinary call: a healthy server answers in milliseconds, and + * the code is valid for 60 seconds, so twice that ends only a hung server. + */ +export const LAUNCH_CODE_TIMEOUT_MS = 120_000; + +const LAUNCH_UNANSWERED = + 'The server did not answer the launch code, and may have used it. Get a new launch code from the terminal that printed the launch address.'; + +const LAUNCH_CANCELLED = + 'You stopped waiting for the server, which may have used the launch code. Get a new launch code from the terminal that printed the launch address.'; + // A body that is not JSON is `undefined`. Any other failure to read it, such // as a connection that dropped or a timeout, is thrown. async function bodyOf(response: Response): Promise { @@ -39,19 +52,20 @@ async function bodyOf(response: Response): Promise { // aborts. The body is read inside the call, so a hung body is covered, and the // timer and the listener never outlive the call. async function withTimeout( - timeoutMs: number, + timeoutMs: number | undefined, outer: AbortSignal | undefined, call: (signal: AbortSignal) => Promise, ): Promise { const controller = new AbortController(); const abort = () => controller.abort(); - const timer = setTimeout(abort, timeoutMs); + const timer = + timeoutMs === undefined ? undefined : setTimeout(abort, timeoutMs); if (outer?.aborted) abort(); outer?.addEventListener('abort', abort, { once: true }); try { return await call(controller.signal); } finally { - clearTimeout(timer); + if (timer !== undefined) clearTimeout(timer); outer?.removeEventListener('abort', abort); } } @@ -68,7 +82,7 @@ async function exchange( async function send( fetchFn: Fetch, - timeoutMs: number, + timeoutMs: number | undefined, path: string, init: RequestInit, outer?: AbortSignal, @@ -112,17 +126,32 @@ function bearer(token: string): Record { export function createHttpApi( fetchFn: Fetch = fetch, timeoutMs = REQUEST_TIMEOUT_MS, + launchTimeoutMs = LAUNCH_CODE_TIMEOUT_MS, ): WorkflowApi { const call = (path: string, init: RequestInit, signal?: AbortSignal) => send(fetchFn, timeoutMs, path, init, signal); return { - async createSession(code) { - const body = await call('/api/session', { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ code }), - }); - return parseSession(body); + async createSession(code, signal) { + try { + const body = await send( + fetchFn, + launchTimeoutMs, + '/api/session', + { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ code }), + }, + signal, + ); + return parseSession(body); + } catch (error) { + if (!(error instanceof ApiError) || error.status !== 0) throw error; + throw new ApiError( + signal?.aborted ? LAUNCH_CANCELLED : LAUNCH_UNANSWERED, + 0, + ); + } }, async getWorkflow(token) { const body = await call('/api/workflow', { diff --git a/workflow-canvas/src/execution/Execution.test.tsx b/workflow-canvas/src/execution/Execution.test.tsx index 900dbc5d6..e0997d535 100644 --- a/workflow-canvas/src/execution/Execution.test.tsx +++ b/workflow-canvas/src/execution/Execution.test.tsx @@ -16,6 +16,7 @@ import { HELLO_DOCUMENT, helloRun, loaded } from '../test-support/fixtures'; import { IDLE, MAX_READ_FAILURES, + RunAlreadyActive, RunReadRefused, RunStartUnsure, type RunPort, @@ -317,6 +318,73 @@ describe('a run', () => { }); }); +describe('a run that is already active', () => { + function serverBusyWith(...readings: RunView[]): RunPort { + const runs = serverShowing(...readings); + runs.start = vi + .fn() + .mockRejectedValue( + new RunAlreadyActive('A run is already active.', RUN_ID), + ); + return runs; + } + + it('is followed, with no alert, and says so', async () => { + const runs = serverBusyWith(helloRun(1), helloRun(2)); + const scheduler = await openExecution(runs); + + await press(runButton()); + + expect(runs.get).toHaveBeenCalledWith(RUN_ID, expect.any(AbortSignal)); + expect(screen.queryByTestId('run-error')).toBeNull(); + expect(statusLine().textContent).toBe( + 'Following the run that was already in progress. Running review. 1 of 2 stages done.', + ); + expect(badges()).toEqual(['Done', 'Done', 'Running', 'Queued']); + expect(isOff(runButton())).toBe(true); + + await tick(scheduler); + + expect(badges()).toEqual(['Done', 'Done', 'Done', 'Passed']); + expect(isOff(runButton())).toBe(false); + }); + + it('keeps saying it was already going once it has ended', async () => { + const runs = serverBusyWith(helloRun(1), helloRun(2)); + const scheduler = await openExecution(runs); + await press(runButton()); + + await tick(scheduler); + + expect(statusLine().textContent).toBe( + 'The run that was already in progress passed. 2 of 2 stages done. Exit code 0.', + ); + }); + + it('does not show a goal of its own, which the API does not give', async () => { + const runs = serverBusyWith(helloRun(1), helloRun(2)); + const scheduler = await openExecution(runs); + const goalBox = () => screen.getByLabelText('Goal') as HTMLInputElement; + fireEvent.change(goalBox(), { target: { value: 'Fix the typo' } }); + + await press(runButton()); + + expect(goalBox().value).toBe(''); + expect(goalBox().placeholder).toBe('Not known'); + expect(goalBox().readOnly).toBe(true); + expect( + screen.getByText( + 'This run was already going, so its goal is not known here.', + ), + ).toBeTruthy(); + + await tick(scheduler); + + expect(goalBox().value).toBe('Fix the typo'); + expect(goalBox().readOnly).toBe(false); + }); +}); + describe('the goal', () => { const goalBox = () => screen.getByLabelText('Goal') as HTMLInputElement; const openTab = (name: string) => @@ -494,6 +562,30 @@ describe('errors from the API', () => { expect(isOff(runButton())).toBe(false); }); + it('says the run may have started when the server did not answer the start', async () => { + const runs: RunPort = { + start: vi + .fn() + .mockRejectedValue(new RunStartUnsure('The server did not answer.')), + get: vi.fn(), + }; + await openExecution(runs); + + await press(runButton()); + + const alert = screen.getByTestId('run-error'); + expect( + within(alert).getByText('The run may not have started'), + ).toBeTruthy(); + expect(within(alert).getByText('The server did not answer.')).toBeTruthy(); + expect( + within(alert).getByText( + 'The run may be going on the server. Press Run to join it.', + ), + ).toBeTruthy(); + expect(isOff(runButton())).toBe(false); + }); + it('shows one alert, and clears the states, when a reading fails', async () => { const runs: RunPort = { start: vi.fn().mockResolvedValue(RUN_ID), @@ -522,6 +614,27 @@ describe('errors from the API', () => { expect(isOff(runButton())).toBe(false); }); + it('keeps following the run when one reading fails and the next works', async () => { + const runs: RunPort = { + start: vi.fn().mockResolvedValue(RUN_ID), + get: vi + .fn() + .mockResolvedValueOnce(helloRun(0)) + .mockRejectedValueOnce(new Error('The server did not answer.')) + .mockResolvedValue(helloRun(1)), + }; + const scheduler = await openExecution(runs); + await press(runButton()); + + await tick(scheduler); + expect(screen.queryByTestId('run-error')).toBeNull(); + expect(isOff(runButton())).toBe(true); + await tick(scheduler); + + expect(badges()).toEqual(['Done', 'Done', 'Running', 'Queued']); + expect(screen.queryByTestId('run-error')).toBeNull(); + }); + it('puts the alert under the Run button', async () => { const runs: RunPort = { start: vi.fn().mockRejectedValue(new Error('A run is already active.')), diff --git a/workflow-canvas/src/execution/Execution.tsx b/workflow-canvas/src/execution/Execution.tsx index e5381a5c0..4818ef4d0 100644 --- a/workflow-canvas/src/execution/Execution.tsx +++ b/workflow-canvas/src/execution/Execution.tsx @@ -18,6 +18,7 @@ import { GOAL_MAX_LENGTH, goalProblem, isInFlight, + isJoinedRun, nodeStatuses, runOf, stageDrift, @@ -91,6 +92,11 @@ function StageDriftNotice({ ); } +// A run that was already going has a goal that the API does not give, and the +// box must not show another goal in its place. +const NO_GOAL_OF_JOINED_RUN = + 'This run was already going, so its goal is not known here.'; + function RunPanel({ goal, onGoalChange, @@ -101,6 +107,7 @@ function RunPanel({ execution: ExecutionController; }) { const inFlight = isInFlight(execution.state); + const joined = isJoinedRun(execution.state); const problem = goalProblem(goal); const canRun = !inFlight && problem === undefined; return ( @@ -108,12 +115,17 @@ function RunPanel({ onGoalChange(detail.value)} onKeyDown={({ detail }) => { diff --git a/workflow-canvas/src/execution/core/execution-state.test.ts b/workflow-canvas/src/execution/core/execution-state.test.ts index 4aa1e693a..a26229b29 100644 --- a/workflow-canvas/src/execution/core/execution-state.test.ts +++ b/workflow-canvas/src/execution/core/execution-state.test.ts @@ -4,6 +4,7 @@ import { helloRun } from '../../test-support/fixtures'; import { IDLE, isInFlight, + isJoinedRun, reduce, runOf, type ExecutionEvent, @@ -175,6 +176,17 @@ describe('reduce', () => { expect(reduce(joined, { type: 'snapshot', run: helloRun(2) })).toEqual({ phase: 'finished', run: helloRun(2), + joined: true, + }); + }); + + it('says a run that it started itself was not joined', () => { + const own = after(...STARTED); + + expect(isJoinedRun(own)).toBe(false); + expect(reduce(own, { type: 'snapshot', run: helloRun(2) })).toEqual({ + phase: 'finished', + run: helloRun(2), }); }); diff --git a/workflow-canvas/src/execution/core/execution-state.ts b/workflow-canvas/src/execution/core/execution-state.ts index c605740fe..ffa6c818e 100644 --- a/workflow-canvas/src/execution/core/execution-state.ts +++ b/workflow-canvas/src/execution/core/execution-state.ts @@ -12,7 +12,12 @@ export type ExecutionState = /** Set when the tab joined a run that was already active. */ joined?: true; } - | { phase: 'finished'; run: RunView } + | { + phase: 'finished'; + run: RunView; + /** Set when the tab joined a run that was already active. */ + joined?: true; + } | { phase: 'error'; message: string; @@ -45,6 +50,11 @@ export function isInFlight(state: ExecutionState): boolean { return state.phase === 'starting' || state.phase === 'following'; } +/** Whether the run in flight is one that was already going when joined. */ +export function isJoinedRun(state: ExecutionState): boolean { + return state.phase === 'following' && state.joined === true; +} + /** The run that the tab shows, if it has heard of one. */ export function runOf(state: ExecutionState): RunView | undefined { return state.phase === 'following' || state.phase === 'finished' @@ -69,7 +79,11 @@ function startedOf( function snapshotOf(state: ExecutionState, run: RunView): ExecutionState { if (state.phase !== 'following' || state.runId !== run.runId) return state; - if (run.status !== 'running') return { phase: 'finished', run }; + if (run.status !== 'running') { + return state.joined + ? { phase: 'finished', run, joined: true } + : { phase: 'finished', run }; + } return state.joined ? { phase: 'following', runId: run.runId, run, joined: true } : { phase: 'following', runId: run.runId, run }; diff --git a/workflow-canvas/src/execution/core/follow-run.ts b/workflow-canvas/src/execution/core/follow-run.ts index 1e2c37315..fb8ad04fb 100644 --- a/workflow-canvas/src/execution/core/follow-run.ts +++ b/workflow-canvas/src/execution/core/follow-run.ts @@ -22,15 +22,10 @@ function messageOf(error: unknown): string { return error instanceof Error ? error.message : UNKNOWN_FAILURE; } -/** - * What `startAndFollow` needs: the ports, the goal and where events go. The - * caller runs one loop at a time: a failure with no run id is taken for the - * failure of the start in progress. - */ -export interface FollowRequest { +/** What following a run needs: the ports and where events go. */ +interface Watch { runs: RunPort; scheduler: Scheduler; - goal: string; /** Must not throw: a throw would be taken for a failure of the run. */ emit: (event: ExecutionEvent) => void; /** Aborting stops the loop quietly: no further event is emitted. */ @@ -48,6 +43,15 @@ function failure( : failed; } +/** + * What `startAndFollow` needs: the ports, the goal and where events go. The + * caller runs one loop at a time: a failure with no run id is taken for the + * failure of the start in progress. + */ +export interface FollowRequest extends Watch { + goal: string; +} + // Starts a run, or names the run that is already active in its place. async function begin({ runs, @@ -67,9 +71,8 @@ async function begin({ // One reading of the run. A failed read is tried again after a poll interval // until `MAX_READ_FAILURES` in a row have failed; then the last error is thrown. // A refused read is thrown at once, as trying again cannot change the answer. -// Resolves to `undefined` when aborted. async function read( - { runs, scheduler, signal }: FollowRequest, + { runs, scheduler, signal }: Watch, runId: string, ): Promise { for (let failures = 1; ; failures += 1) { @@ -89,11 +92,16 @@ async function read( } } -// Reads the run every poll interval until it is no longer running. -async function follow(request: FollowRequest, runId: string): Promise { +// Reads the run every poll interval until it is no longer running. `first` is +// a reading that the caller already has. +async function follow( + request: Watch, + runId: string, + first?: RunView, +): Promise { const { scheduler, emit, signal } = request; - for (;;) { - const run = await read(request, runId); + for (let known = first; ; known = undefined) { + const run = known ?? (await read(request, runId)); if (signal.aborted || run === undefined) return; if (run.runId !== runId) { emit({ type: 'failed', message: OTHER_RUN, runId }); @@ -130,3 +138,53 @@ export async function startAndFollow(request: FollowRequest): Promise { if (!signal.aborted) emit(failure(error, runId)); } } + +/** What `resumeRun` needs: what `startAndFollow` does, and the run to join. */ +export interface ResumeRequest extends Watch { + runId: string; +} + +// The look at the remembered run could not finish, so nobody knows whether it +// is going. Say so under the run, which the alert names. +function reportUnknownRun( + { runId, emit }: ResumeRequest, + error: unknown, +): void { + emit({ type: 'start' }); + emit({ type: 'started', runId, joined: true }); + emit({ type: 'failed', message: messageOf(error), runId }); +} + +/** + * Joins the run `runId` if it is still running, as a page that was reloaded + * does, and follows it like `startAndFollow`. The first reading is tried + * again like any other. A run that has ended, another run, or one that the + * server will not give (it started again) says nothing: there is nothing to + * resume. When the readings fail for any other reason, whether the run is + * going is not known, and the failure is reported like a lost run. It never + * rejects. + */ +export async function resumeRun(request: ResumeRequest): Promise { + const { runId, emit, signal } = request; + let run: RunView | undefined; + try { + run = await read(request, runId); + } catch (error) { + if (!signal.aborted && !(error instanceof RunReadRefused)) { + reportUnknownRun(request, error); + } + return; + } + if (signal.aborted || run?.runId !== runId || run.status !== 'running') { + return; + } + emit({ type: 'start' }); + emit({ type: 'started', runId, joined: true }); + try { + await follow(request, runId, run); + } catch (error) { + if (!signal.aborted) { + emit({ type: 'failed', message: messageOf(error), runId }); + } + } +} diff --git a/workflow-canvas/src/execution/core/index.ts b/workflow-canvas/src/execution/core/index.ts index fd6f1a35b..6267c40fa 100644 --- a/workflow-canvas/src/execution/core/index.ts +++ b/workflow-canvas/src/execution/core/index.ts @@ -8,6 +8,7 @@ export { IDLE, isInFlight, + isJoinedRun, reduce, runOf, type ExecutionEvent, @@ -16,15 +17,18 @@ export { export { MAX_READ_FAILURES, POLL_INTERVAL_MS, + resumeRun, startAndFollow, type FollowRequest, } from './follow-run'; export { GOAL_MAX_LENGTH, goalProblem } from './goal'; export { nodeStatuses } from './node-statuses'; export { + NO_RUN_MEMORY, RunAlreadyActive, RunReadRefused, RunStartUnsure, + type RunMemory, type RunPort, type Scheduler, } from './ports'; diff --git a/workflow-canvas/src/execution/core/ports.ts b/workflow-canvas/src/execution/core/ports.ts index 3f21beb9f..f3cdb4db3 100644 --- a/workflow-canvas/src/execution/core/ports.ts +++ b/workflow-canvas/src/execution/core/ports.ts @@ -59,3 +59,23 @@ export interface Scheduler { /** Resolves after `ms` milliseconds, or at once when the signal aborts. */ wait(ms: number, signal: AbortSignal): Promise; } + +/** + * What survives a reload of the page: the id of the latest run, so that the + * page can find a run that is still going. The adapter is + * `execution/run-memory.ts`. + */ +export interface RunMemory { + /** The id of the latest run that was remembered, if any. */ + recall(): string | undefined; + remember(runId: string): void; + /** Drops the run that was remembered, once it has ended or is gone. */ + forget(): void; +} + +/** A memory that keeps nothing, for a page that does not survive a reload. */ +export const NO_RUN_MEMORY: RunMemory = { + recall: () => undefined, + remember: () => undefined, + forget: () => undefined, +}; diff --git a/workflow-canvas/src/execution/core/resume-run.test.ts b/workflow-canvas/src/execution/core/resume-run.test.ts new file mode 100644 index 000000000..501331a25 --- /dev/null +++ b/workflow-canvas/src/execution/core/resume-run.test.ts @@ -0,0 +1,166 @@ +// SPDX-License-Identifier: Apache-2.0 +import { describe, expect, it, vi } from 'vitest'; +import type { RunView } from '../../api/types'; +import { FakeScheduler } from '../../test-support/fake-scheduler'; +import { helloRun } from '../../test-support/fixtures'; +import type { ExecutionEvent } from './execution-state'; +import { POLL_INTERVAL_MS, resumeRun } from './follow-run'; +import { RunReadRefused, type RunPort } from './ports'; + +const RUN_ID = helloRun(0).runId; + +// Lets the loop run until it waits for the scheduler or ends. +async function settle(): Promise { + for (let turn = 0; turn < 20; turn += 1) await Promise.resolve(); +} + +function setup(readings: (RunView | Error)[]) { + const queue = [...readings]; + const runs: RunPort = { + start: vi.fn(), + get: vi.fn(async () => { + const next = queue.shift(); + if (next === undefined) throw new Error('No reading left.'); + if (next instanceof Error) throw next; + return next; + }), + }; + const scheduler = new FakeScheduler(); + const events: ExecutionEvent[] = []; + const abort = new AbortController(); + const done = resumeRun({ + runs, + scheduler, + runId: RUN_ID, + emit: (event) => events.push(event), + signal: abort.signal, + }); + return { runs, scheduler, events, abort, done }; +} + +describe('resumeRun', () => { + it('joins a run that is still running, with what it read, and goes on reading it', async () => { + const { runs, scheduler, events, done } = setup([helloRun(1), helloRun(2)]); + + await settle(); + expect(events).toEqual([ + { type: 'start' }, + { type: 'started', runId: RUN_ID, joined: true }, + { type: 'snapshot', run: helloRun(1) }, + ]); + expect(scheduler.waits).toEqual([POLL_INTERVAL_MS]); + + scheduler.tick(); + await done; + + expect(runs.get).toHaveBeenCalledTimes(2); + expect(events.at(-1)).toEqual({ type: 'snapshot', run: helloRun(2) }); + }); + + it('says nothing about a run that has ended', async () => { + const { events, done } = setup([helloRun(2)]); + + await done; + + expect(events).toEqual([]); + }); + + it('says nothing when the server refuses to give the run, as after it started again', async () => { + const { runs, events, done } = setup([new RunReadRefused('Gone.')]); + + await done; + + expect(events).toEqual([]); + expect(runs.get).toHaveBeenCalledTimes(1); + }); + + it('reads again after a reading that fails once, and joins the run', async () => { + const { runs, scheduler, events, done } = setup([ + new Error('The server did not answer.'), + helloRun(1), + helloRun(2), + ]); + + await settle(); + expect(events).toEqual([]); + scheduler.tick(); + await settle(); + expect(events.slice(0, 2)).toEqual([ + { type: 'start' }, + { type: 'started', runId: RUN_ID, joined: true }, + ]); + scheduler.tick(); + await done; + + expect(runs.get).toHaveBeenCalledTimes(3); + }); + + it('reports a run that may be going when every reading fails', async () => { + const down = () => new Error('The server did not answer.'); + const { runs, scheduler, events, done } = setup([down(), down(), down()]); + + for (let turn = 0; turn < 3; turn += 1) { + await settle(); + scheduler.tick(); + } + await done; + + expect(runs.get).toHaveBeenCalledTimes(3); + expect(events).toEqual([ + { type: 'start' }, + { type: 'started', runId: RUN_ID, joined: true }, + { type: 'failed', message: 'The server did not answer.', runId: RUN_ID }, + ]); + }); + + it('says nothing when it is stopped while the readings fail', async () => { + const { events, abort, done } = setup([ + new Error('The server did not answer.'), + ]); + + abort.abort(); + await done; + + expect(events).toEqual([]); + }); + + it('says nothing when the server answers with another run', async () => { + const other = { ...helloRun(1), runId: 'another-run' }; + const { events, done } = setup([other]); + + await done; + + expect(events).toEqual([]); + }); + + it('says nothing once it is aborted', async () => { + const { events, abort, done } = setup([helloRun(1)]); + abort.abort(); + + await done; + + expect(events).toEqual([]); + }); + + it('ends the join with a failure when the run can no longer be read', async () => { + const { scheduler, events, done } = setup([ + helloRun(1), + new Error('Down.'), + new Error('Down.'), + new Error('Down.'), + ]); + + await settle(); + for (let turn = 0; turn < 4; turn += 1) { + scheduler.tick(); + await settle(); + } + await done; + + expect(events.at(-1)).toEqual({ + type: 'failed', + message: 'Down.', + runId: RUN_ID, + }); + }); +}); diff --git a/workflow-canvas/src/execution/core/status-text.test.ts b/workflow-canvas/src/execution/core/status-text.test.ts index 8d7691433..f708191dc 100644 --- a/workflow-canvas/src/execution/core/status-text.test.ts +++ b/workflow-canvas/src/execution/core/status-text.test.ts @@ -26,6 +26,25 @@ describe('statusText for a run that was already active', () => { }); }); +describe('statusText for a joined run that has ended', () => { + it('says it was already in progress', () => { + expect( + statusText({ phase: 'finished', run: helloRun(2), joined: true }), + ).toBe( + 'The run that was already in progress passed. 2 of 2 stages done. Exit code 0.', + ); + expect( + statusText({ + phase: 'finished', + run: helloRun(0, { blockedAt: 1 }), + joined: true, + }), + ).toBe( + 'The run that was already in progress halted at review. 1 of 2 stages done. Exit code 1.', + ); + }); +}); + describe('statusText and statusKind', () => { it.each<[string, ExecutionState, string, string]>([ ['idle', IDLE, 'Ready. Press Run to start.', 'idle'], diff --git a/workflow-canvas/src/execution/core/status-text.ts b/workflow-canvas/src/execution/core/status-text.ts index fa45ac3c5..c90ce5753 100644 --- a/workflow-canvas/src/execution/core/status-text.ts +++ b/workflow-canvas/src/execution/core/status-text.ts @@ -21,13 +21,14 @@ function followingText(run: RunView | undefined): string { return `${lead} ${stagesDone(run)}`; } -function finishedText(run: RunView): string { +function finishedText(run: RunView, joined: boolean | undefined): string { const exit = run.exitCode === null ? '' : ` Exit code ${run.exitCode}.`; const blocked = run.stages.find((stage) => stage.status === 'blocked'); + const subject = joined ? 'The run that was already in progress' : 'Run'; const lead = run.status === 'passed' - ? 'Run passed.' - : `Run halted${blocked ? ` at ${blocked.name}` : ''}.`; + ? `${subject} passed.` + : `${subject} halted${blocked ? ` at ${blocked.name}` : ''}.`; return `${lead} ${stagesDone(run)}${exit}`; } @@ -43,7 +44,7 @@ export function statusText(state: ExecutionState): string { ? `${JOINED} ${followingText(state.run)}` : followingText(state.run); case 'finished': - return finishedText(state.run); + return finishedText(state.run, state.joined); case 'error': return 'See the error below.'; } diff --git a/workflow-canvas/src/execution/run-memory.test.ts b/workflow-canvas/src/execution/run-memory.test.ts new file mode 100644 index 000000000..46c1692d7 --- /dev/null +++ b/workflow-canvas/src/execution/run-memory.test.ts @@ -0,0 +1,76 @@ +// SPDX-License-Identifier: Apache-2.0 +import { describe, expect, it } from 'vitest'; +import { sessionRunMemory } from './run-memory'; + +function memoryStorage(initial: Record = {}) { + const data = new Map(Object.entries(initial)); + return { + data, + getItem: (key: string) => data.get(key) ?? null, + setItem: (key: string, value: string) => void data.set(key, value), + removeItem: (key: string) => void data.delete(key), + }; +} + +const denied = () => { + throw new Error('Access is denied.'); +}; +// A browser that blocks storage throws on reading the property already. +const blockedAtAccess = { + get sessionStorage(): never { + return denied(); + }, +}; +const blockedAtUse = { + sessionStorage: { getItem: denied, setItem: denied, removeItem: denied }, +}; + +describe('sessionRunMemory', () => { + it('recalls nothing before a run is remembered', () => { + expect( + sessionRunMemory({ sessionStorage: memoryStorage() }).recall(), + ).toBeUndefined(); + }); + + it('recalls the run that was remembered, also through another memory of the same storage', () => { + const env = { sessionStorage: memoryStorage() }; + sessionRunMemory(env).remember('run-1'); + + expect(sessionRunMemory(env).recall()).toBe('run-1'); + }); + + it('keeps only the latest run', () => { + const memory = sessionRunMemory({ sessionStorage: memoryStorage() }); + memory.remember('run-1'); + memory.remember('run-2'); + + expect(memory.recall()).toBe('run-2'); + }); + + it('recalls nothing once the run is forgotten', () => { + const memory = sessionRunMemory({ sessionStorage: memoryStorage() }); + memory.remember('run-1'); + + memory.forget(); + + expect(memory.recall()).toBeUndefined(); + }); + + it('treats an empty value as nothing', () => { + const storage = memoryStorage(); + storage.setItem('konductor-workflow-canvas-run', ''); + + expect( + sessionRunMemory({ sessionStorage: storage }).recall(), + ).toBeUndefined(); + }); + + it('works without storage, or with storage that the browser blocks', () => { + for (const env of [{}, blockedAtAccess, blockedAtUse]) { + const memory = sessionRunMemory(env); + expect(() => memory.remember('run-1')).not.toThrow(); + expect(() => memory.forget()).not.toThrow(); + expect(memory.recall()).toBeUndefined(); + } + }); +}); diff --git a/workflow-canvas/src/execution/run-memory.ts b/workflow-canvas/src/execution/run-memory.ts new file mode 100644 index 000000000..9d00f5323 --- /dev/null +++ b/workflow-canvas/src/execution/run-memory.ts @@ -0,0 +1,41 @@ +// SPDX-License-Identifier: Apache-2.0 +import type { RunMemory } from './core'; + +const RUN_KEY = 'konductor-workflow-canvas-run'; + +/** The part of `window` that the memory of a run needs. */ +export interface RunMemoryEnvironment { + sessionStorage?: Pick; +} + +/** + * Remembers the id of the latest run in the storage of the browser tab, so + * that a reload of the page can find it again. The id is no secret; the + * session is kept apart, by `launch/session-memory.ts`. A browser that blocks + * storage leaves it with no memory. + */ +export function sessionRunMemory(env: RunMemoryEnvironment): RunMemory { + return { + recall() { + try { + return env.sessionStorage?.getItem(RUN_KEY) || undefined; + } catch { + return undefined; + } + }, + remember(runId) { + try { + env.sessionStorage?.setItem(RUN_KEY, runId); + } catch { + // Without storage there is nothing to remember. + } + }, + forget() { + try { + env.sessionStorage?.removeItem(RUN_KEY); + } catch { + // Without storage there is nothing to forget. + } + }, + }; +} diff --git a/workflow-canvas/src/execution/use-execution.test.ts b/workflow-canvas/src/execution/use-execution.test.ts new file mode 100644 index 000000000..90c6655cf --- /dev/null +++ b/workflow-canvas/src/execution/use-execution.test.ts @@ -0,0 +1,425 @@ +// SPDX-License-Identifier: Apache-2.0 +import { act, renderHook, waitFor } from '@testing-library/react'; +import { describe, expect, it, vi } from 'vitest'; +import { FakeScheduler } from '../test-support/fake-scheduler'; +import { helloRun } from '../test-support/fixtures'; +import { + RunReadRefused, + type RunMemory, + type RunPort, + type Scheduler, +} from './core'; +import { useExecution } from './use-execution'; + +const RUN_ID = helloRun(0).runId; + +function runsThatEnd(): RunPort { + return { + start: vi.fn().mockResolvedValue(RUN_ID), + get: vi.fn().mockResolvedValue(helloRun(2)), + }; +} + +const emptyMemory = (runId?: string): RunMemory => ({ + recall: () => runId, + remember: vi.fn(), + forget: vi.fn(), +}); + +describe('useExecution', () => { + it('starts one run when start is called twice before anything renders', async () => { + const runs = runsThatEnd(); + const { result } = renderHook(() => + useExecution(runs, new FakeScheduler()), + ); + + await act(async () => { + result.current.start('Ship it'); + result.current.start('Ship it'); + }); + + expect(runs.start).toHaveBeenCalledOnce(); + }); + + it('starts again once the run has ended', async () => { + const runs = runsThatEnd(); + const { result } = renderHook(() => + useExecution(runs, new FakeScheduler()), + ); + await act(async () => result.current.start('Ship it')); + + await act(async () => result.current.start('Ship it')); + + expect(runs.start).toHaveBeenCalledTimes(2); + }); + + it('does nothing without a connection to a server', () => { + const { result } = renderHook(() => + useExecution(undefined, new FakeScheduler()), + ); + + act(() => result.current.start('Ship it')); + + expect(result.current.state).toEqual({ phase: 'idle' }); + }); +}); + +describe('useExecution after a reload', () => { + const memoryOf = (runId?: string): RunMemory => ({ + recall: vi.fn(() => runId), + remember: vi.fn(), + forget: vi.fn(), + }); + + it('joins the run that the memory names when that run is still going', async () => { + const runs: RunPort = { + start: vi.fn(), + get: vi.fn().mockResolvedValue(helloRun(1)), + }; + + const { result } = renderHook(() => + useExecution(runs, new FakeScheduler(), memoryOf(RUN_ID)), + ); + + await waitFor(() => + expect(result.current.state).toMatchObject({ + phase: 'following', + runId: RUN_ID, + joined: true, + }), + ); + expect(runs.start).not.toHaveBeenCalled(); + }); + + it('stays ready when the run of the memory has ended', async () => { + const runs: RunPort = { + start: vi.fn(), + get: vi.fn().mockResolvedValue(helloRun(2)), + }; + + const { result } = renderHook(() => + useExecution(runs, new FakeScheduler(), memoryOf(RUN_ID)), + ); + await act(async () => {}); + + expect(runs.get).toHaveBeenCalledWith(RUN_ID, expect.any(AbortSignal)); + expect(result.current.state).toEqual({ phase: 'idle' }); + }); + + it('reads nothing when the memory is empty', async () => { + const runs = runsThatEnd(); + + renderHook(() => useExecution(runs, new FakeScheduler(), memoryOf())); + await act(async () => {}); + + expect(runs.get).not.toHaveBeenCalled(); + }); + + it('remembers the id of the run that it starts and of the run that it joins', async () => { + const memory = memoryOf(); + const { result } = renderHook(() => + useExecution(runsThatEnd(), new FakeScheduler(), memory), + ); + + await act(async () => result.current.start('Ship it')); + + expect(memory.remember).toHaveBeenCalledWith(RUN_ID); + }); +}); + +describe('useExecution when the remembered run cannot be looked up', () => { + const memoryOf = (runId: string): RunMemory => ({ + recall: () => runId, + remember: vi.fn(), + forget: vi.fn(), + }); + + it('stays at rest when the server will not give the run', async () => { + const runs: RunPort = { + start: vi.fn(), + get: vi.fn().mockRejectedValue(new RunReadRefused('Gone.')), + }; + const { result } = renderHook(() => + useExecution(runs, new FakeScheduler(), memoryOf(RUN_ID)), + ); + await act(async () => {}); + + expect(result.current.state).toEqual({ phase: 'idle' }); + }); + + it('says the run may be going when the server does not answer', async () => { + const runs: RunPort = { + start: vi.fn(), + get: vi.fn().mockRejectedValue(new Error('The server did not answer.')), + }; + const scheduler = new FakeScheduler(); + const { result } = renderHook(() => + useExecution(runs, scheduler, memoryOf(RUN_ID)), + ); + + for (let turn = 0; turn < 3; turn += 1) { + await act(async () => scheduler.tick()); + } + + expect(result.current.state).toEqual({ + phase: 'error', + message: 'The server did not answer.', + started: true, + runId: RUN_ID, + }); + }); +}); + +describe('useExecution while a run of the memory is being looked for', () => { + const memoryOf = (runId?: string): RunMemory => ({ + recall: vi.fn(() => runId), + remember: vi.fn(), + forget: vi.fn(), + }); + + /** A server whose first read of the remembered run never answers. */ + function slowToAnswer() { + const runs: RunPort = { + start: vi.fn().mockResolvedValue(RUN_ID), + get: vi.fn( + (_id: string, signal: AbortSignal) => + new Promise((_resolve, reject) => + signal.addEventListener('abort', () => + reject(new Error('aborted')), + ), + ), + ), + }; + return runs; + } + + it('starts the run when Run is pressed, instead of ignoring the click', async () => { + const runs = slowToAnswer(); + const { result } = renderHook(() => + useExecution(runs, new FakeScheduler(), memoryOf(RUN_ID)), + ); + await act(async () => {}); + expect(runs.get).toHaveBeenCalledOnce(); + + await act(async () => result.current.start('Ship it')); + + expect(runs.start).toHaveBeenCalledWith('Ship it', expect.any(AbortSignal)); + expect(result.current.state.phase).not.toBe('idle'); + }); + + it('stops looking for the earlier run when Run is pressed', async () => { + const runs = slowToAnswer(); + const { result } = renderHook(() => + useExecution(runs, new FakeScheduler(), memoryOf(RUN_ID)), + ); + await act(async () => {}); + const lookup = (runs.get as ReturnType).mock + .calls[0]![1] as AbortSignal; + + await act(async () => result.current.start('Ship it')); + + expect(lookup.aborted).toBe(true); + }); + + it('ignores a second click once the run is under way', async () => { + const runs = slowToAnswer(); + const { result } = renderHook(() => + useExecution(runs, new FakeScheduler(), memoryOf(RUN_ID)), + ); + await act(async () => {}); + + await act(async () => { + result.current.start('Ship it'); + result.current.start('Ship it'); + }); + + expect(runs.start).toHaveBeenCalledOnce(); + }); + + it('ignores Run once the earlier run has been found and is being followed', async () => { + const runs: RunPort = { + start: vi.fn(), + get: vi.fn().mockResolvedValue(helloRun(1)), + }; + const scheduler = new FakeScheduler(); + const memory = memoryOf(RUN_ID); + const { result } = renderHook(() => useExecution(runs, scheduler, memory)); + await waitFor(() => expect(result.current.state.phase).toBe('following')); + + await act(async () => result.current.start('Ship it')); + + expect(runs.start).not.toHaveBeenCalled(); + }); +}); + +describe('useExecution and the memory', () => { + it('remembers no run when the run does not start', async () => { + const memory: RunMemory = { + recall: () => undefined, + remember: vi.fn(), + forget: vi.fn(), + }; + const runs: RunPort = { + start: vi.fn().mockRejectedValue(new Error('kiro-cli is not installed.')), + get: vi.fn(), + }; + const { result } = renderHook(() => + useExecution(runs, new FakeScheduler(), memory), + ); + + await act(async () => result.current.start('Ship it')); + + expect(memory.remember).not.toHaveBeenCalled(); + }); + + it('forgets the run once it has ended', async () => { + const memory = { ...emptyMemory(), forget: vi.fn() }; + const runs = runsThatEnd(); + const { result } = renderHook(() => + useExecution(runs, new FakeScheduler(), memory), + ); + + await act(async () => result.current.start('Ship it')); + + expect(memory.forget).toHaveBeenCalledOnce(); + }); + + it('keeps the run in the memory while it goes on', async () => { + const memory = { ...emptyMemory(), forget: vi.fn() }; + const runs: RunPort = { + start: vi.fn().mockResolvedValue(RUN_ID), + get: vi.fn().mockResolvedValue(helloRun(1)), + }; + const { result } = renderHook(() => + useExecution(runs, new FakeScheduler(), memory), + ); + + await act(async () => result.current.start('Ship it')); + + expect(memory.forget).not.toHaveBeenCalled(); + }); + + it.each([ + ['has ended', () => Promise.resolve(helloRun(2))], + ['is gone', () => Promise.reject(new RunReadRefused('Gone.'))], + ])( + 'forgets a remembered run that %s when the page is opened again', + async (_, get) => { + const memory = { ...emptyMemory(RUN_ID), forget: vi.fn() }; + const runs: RunPort = { start: vi.fn(), get: vi.fn(get) }; + + renderHook(() => useExecution(runs, new FakeScheduler(), memory)); + await act(async () => {}); + + expect(memory.forget).toHaveBeenCalledOnce(); + }, + ); + + it('keeps a remembered run that could not be looked up', async () => { + const memory = { ...emptyMemory(RUN_ID), forget: vi.fn() }; + const runs: RunPort = { + start: vi.fn(), + get: vi.fn().mockRejectedValue(new Error('The server did not answer.')), + }; + const scheduler = new FakeScheduler(); + + renderHook(() => useExecution(runs, scheduler, memory)); + for (let turn = 0; turn < 3; turn += 1) { + await act(async () => scheduler.tick()); + } + + expect(memory.forget).not.toHaveBeenCalled(); + }); + + it('keeps a remembered run when the look for it is stopped', async () => { + const memory = { ...emptyMemory(RUN_ID), forget: vi.fn() }; + const runs: RunPort = { + start: vi.fn(), + get: vi.fn( + (_id: string, signal: AbortSignal) => + new Promise((_resolve, reject) => + signal.addEventListener('abort', () => + reject(new Error('stopped')), + ), + ), + ), + }; + + const { unmount } = renderHook(() => + useExecution(runs, new FakeScheduler(), memory), + ); + await act(async () => {}); + unmount(); + await act(async () => {}); + + expect(memory.forget).not.toHaveBeenCalled(); + }); +}); + +describe('useExecution when the page renders again', () => { + it('goes on following a run that it joined', async () => { + const memory = emptyMemory(RUN_ID); + const scheduler = new FakeScheduler(); + const joined: RunPort = { + start: vi.fn(), + get: vi + .fn() + .mockResolvedValueOnce(helloRun(1)) + .mockResolvedValue(helloRun(2)), + }; + const { result, rerender } = renderHook( + ({ tick }) => useExecution(joined, tick, memory), + { initialProps: { tick: scheduler as Scheduler } }, + ); + await waitFor(() => expect(result.current.state.phase).toBe('following')); + + rerender({ tick: scheduler }); + await act(async () => scheduler.tick()); + + expect(joined.get).toHaveBeenCalledTimes(2); + expect(result.current.state.phase).toBe('finished'); + }); +}); + +describe('useExecution when another workflow opens', () => { + async function joinThenOpenOther(memory: RunMemory) { + const scheduler = new FakeScheduler(); + const joined: RunPort = { + start: vi.fn(), + get: vi.fn().mockResolvedValue(helloRun(1)), + }; + const other: RunPort = { + start: vi.fn(), + get: vi.fn().mockResolvedValue(helloRun(2)), + }; + const hook = renderHook( + ({ runs }) => useExecution(runs, scheduler, memory), + { initialProps: { runs: joined } }, + ); + await waitFor(() => + expect(hook.result.current.state.phase).toBe('following'), + ); + hook.rerender({ runs: other }); + return { ...hook, scheduler, joined }; + } + + it('lets go of a run that it joined and shows no run', async () => { + const { result, scheduler, joined } = await joinThenOpenOther( + emptyMemory(RUN_ID), + ); + const reads = vi.mocked(joined.get).mock.calls.length; + + await act(async () => scheduler.tick()); + + expect(result.current.state.phase).toBe('idle'); + expect(vi.mocked(joined.get).mock.calls).toHaveLength(reads); + }); + + it('forgets the run, so a reload does not look for it on another server', async () => { + const memory = emptyMemory(RUN_ID); + + await joinThenOpenOther(memory); + + expect(memory.forget).toHaveBeenCalledOnce(); + }); +}); diff --git a/workflow-canvas/src/execution/use-execution.ts b/workflow-canvas/src/execution/use-execution.ts index cb18384e1..d95a219aa 100644 --- a/workflow-canvas/src/execution/use-execution.ts +++ b/workflow-canvas/src/execution/use-execution.ts @@ -2,14 +2,23 @@ import { useCallback, useEffect, useReducer, useRef } from 'react'; import { IDLE, + NO_RUN_MEMORY, reduce, + resumeRun, startAndFollow, type ExecutionEvent, type ExecutionState, + type RunMemory, type RunPort, type Scheduler, } from './core'; +// A loop that follows a run. `resuming` holds until it has found a run to join. +interface Loop { + controller: AbortController; + resuming: boolean; +} + // `reset` is the hook's own event: another workflow was opened, so the run of // the one that was left is no longer the run of this page. type Action = ExecutionEvent | { type: 'reset' }; @@ -28,46 +37,92 @@ export interface ExecutionController { /** * Drives one run at a time through `runs`, waiting with `scheduler`, and * stops following it when the component unmounts. The state lives with the - * caller, so it survives a switch of tab. When `runs` is replaced by those of - * another workflow, the run in hand is let go: it stays on its server, but - * this page no longer follows it or shows its status. + * caller, so it survives a switch of tab. `memory` keeps the id of the latest + * run, so that a page which is opened again joins a run that is still going. + * When `runs` is replaced by those of another workflow, the run in hand is let + * go: it stays on its server, but this page no longer follows it or shows its + * status, and `memory` forgets it. `runs` is the same object for as long as the + * session lasts. */ export function useExecution( runs: RunPort | undefined, scheduler: Scheduler, + memory: RunMemory = NO_RUN_MEMORY, ): ExecutionController { const [state, dispatch] = useReducer(reduceAction, IDLE); - const active = useRef(undefined); + const active = useRef(undefined); const opened = useRef(runs); - useEffect(() => () => active.current?.abort(), []); + const emit = useCallback( + (event: ExecutionEvent) => { + if (event.type === 'started') memory.remember(event.runId); + if (event.type === 'snapshot' && event.run.status !== 'running') { + memory.forget(); + } + if (event.type === 'start' && active.current) { + active.current.resuming = false; + } + dispatch(event); + }, + [memory], + ); + + // Claims the one place for a loop, runs it, and gives the place back when it + // ends. Returns the loop, whose controller stops it. + const track = useCallback( + (resuming: boolean, run: (signal: AbortSignal) => Promise): Loop => { + const loop: Loop = { controller: new AbortController(), resuming }; + active.current = loop; + void run(loop.controller.signal).finally(() => { + if (active.current === loop) active.current = undefined; + }); + return loop; + }, + [], + ); + + useEffect(() => () => active.current?.controller.abort(), []); useEffect(() => { const left = opened.current; opened.current = runs; if (left === undefined || runs === undefined || left === runs) return; - active.current?.abort(); + active.current?.controller.abort(); active.current = undefined; + memory.forget(); dispatch({ type: 'reset' }); - }, [runs]); + }, [runs, memory]); + + useEffect(() => { + const runId = runs ? memory.recall() : undefined; + if (!runs || runId === undefined || active.current) return; + const loop: Loop = track(true, async (signal) => { + await resumeRun({ runs, scheduler, runId, emit, signal }); + // Nothing was joined: the run has ended, or the server will not give it. + if (loop.resuming && !signal.aborted) memory.forget(); + }); + return () => { + // A look that has not found a run is stopped; a run that was joined goes + // on, like one that a click started. + if (!loop.resuming) return; + loop.controller.abort(); + if (active.current === loop) active.current = undefined; + }; + }, [runs, scheduler, memory, emit, track]); const start = useCallback( (goal: string) => { - if (!runs || active.current) return; - const controller = new AbortController(); - active.current = controller; + if (!runs) return; + // A page that is only looking for a run to join gives way to a click on + // Run: a run that is still going answers 409 and is joined anyway. + if (active.current?.resuming) active.current.controller.abort(); + else if (active.current) return; dispatch({ type: 'start' }); - void startAndFollow({ - runs, - scheduler, - goal, - emit: dispatch, - signal: controller.signal, - }).finally(() => { - if (active.current === controller) active.current = undefined; - }); + track(false, (signal) => + startAndFollow({ runs, scheduler, goal, emit, signal }), + ); }, - [runs, scheduler], + [runs, scheduler, emit, track], ); return { state, start }; diff --git a/workflow-canvas/src/launch/LaunchWait.test.tsx b/workflow-canvas/src/launch/LaunchWait.test.tsx new file mode 100644 index 000000000..8bf0aec1a --- /dev/null +++ b/workflow-canvas/src/launch/LaunchWait.test.tsx @@ -0,0 +1,126 @@ +// SPDX-License-Identifier: Apache-2.0 +import { act, fireEvent, render, screen } from '@testing-library/react'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { App } from '../App'; +import type { WorkflowApi } from '../api/api'; +import { createHttpApi } from '../api/http-api'; +import { HELLO_DOCUMENT, HELLO_SESSION } from '../test-support/fixtures'; +import { freshMemory } from '../test-support/memory-storage'; +import { useWorkflowSource } from '../workflow/use-workflow-source'; +import { createWorkflowSource } from '../workflow/workflow-source'; +import { LaunchWait, STILL_WAITING_MS } from './LaunchWait'; + +// The page as main.tsx wires it, over a server whose launch code trade a test +// controls. +function pageWith(createSession: WorkflowApi['createSession']) { + const api: WorkflowApi = { + createSession, + getWorkflow: vi.fn().mockResolvedValue(HELLO_DOCUMENT), + startRun: vi.fn(), + getRun: vi.fn(), + }; + const source = createWorkflowSource(api, 'code-1', freshMemory()); + function Root() { + const load = useWorkflowSource(source); + return ( + } /> + ); + } + render(); +} + +function hangingSession() { + let finish: (value: typeof HELLO_SESSION) => void = () => {}; + const createSession: WorkflowApi['createSession'] = (_code, signal) => + new Promise((resolve, reject) => { + finish = resolve; + signal?.addEventListener('abort', () => + reject(new Error('The trade was stopped.')), + ); + }); + return { createSession, finish: (value = HELLO_SESSION) => finish(value) }; +} + +describe('LaunchWait', () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it('says nothing while the server answers in time', async () => { + vi.useFakeTimers(); + pageWith(() => Promise.resolve(HELLO_SESSION)); + + await act(() => vi.advanceTimersByTimeAsync(STILL_WAITING_MS + 1000)); + + expect(screen.queryByTestId('launch-wait')).toBeNull(); + expect(screen.getByTestId('node-stage-draft')).toBeTruthy(); + }); + + it('says it is still waiting after the threshold, not before', async () => { + vi.useFakeTimers(); + const server = hangingSession(); + pageWith(server.createSession); + + await act(() => vi.advanceTimersByTimeAsync(STILL_WAITING_MS - 1)); + expect(screen.queryByTestId('launch-wait')).toBeNull(); + await act(() => vi.advanceTimersByTimeAsync(1)); + + expect(screen.getByTestId('launch-wait').textContent).toContain( + 'Still waiting for the server', + ); + }); + + it('draws the workflow and drops the notice when a late answer comes', async () => { + vi.useFakeTimers(); + const server = hangingSession(); + pageWith(server.createSession); + await act(() => vi.advanceTimersByTimeAsync(STILL_WAITING_MS)); + + await act(async () => server.finish()); + + expect(screen.queryByTestId('launch-wait')).toBeNull(); + expect(screen.getByTestId('node-stage-draft')).toBeTruthy(); + }); + + it('aborts the trade on Stop waiting and shows the error', async () => { + vi.useFakeTimers(); + const server = hangingSession(); + pageWith(server.createSession); + await act(() => vi.advanceTimersByTimeAsync(STILL_WAITING_MS)); + + await act(async () => { + fireEvent.click(screen.getByRole('button', { name: 'Stop waiting' })); + }); + + expect(screen.queryByTestId('launch-wait')).toBeNull(); + expect(screen.getByText('The workflow did not load')).toBeTruthy(); + expect(screen.getByText(/The trade was stopped/)).toBeTruthy(); + }); + + it('ends in the explicit error at the hard limit of the real client', async () => { + vi.useFakeTimers(); + const fetchFn = vi.fn( + (_path: string, init?: RequestInit) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener('abort', () => + reject(new Error('aborted')), + ); + }), + ); + const api = createHttpApi(fetchFn as unknown as typeof fetch); + const source = createWorkflowSource(api, 'code-1', freshMemory()); + function Root() { + const load = useWorkflowSource(source); + return ( + } /> + ); + } + render(); + + await act(() => vi.advanceTimersByTimeAsync(120_000)); + + expect(screen.queryByTestId('launch-wait')).toBeNull(); + expect(screen.getByText('The workflow did not load')).toBeTruthy(); + expect(screen.getByText(/may have used it/)).toBeTruthy(); + }); +}); diff --git a/workflow-canvas/src/launch/LaunchWait.tsx b/workflow-canvas/src/launch/LaunchWait.tsx new file mode 100644 index 000000000..0119b52e4 --- /dev/null +++ b/workflow-canvas/src/launch/LaunchWait.tsx @@ -0,0 +1,42 @@ +// SPDX-License-Identifier: Apache-2.0 +import Alert from '@cloudscape-design/components/alert'; +import Button from '@cloudscape-design/components/button'; +import { useEffect, useState } from 'react'; +import { useSettled } from '../designer/use-settled'; +import type { LoadResult } from '../workflow/load-workflow'; +import type { WorkflowSource } from '../workflow/workflow-source'; + +/** How long the first load may take before the page says it is still waiting. */ +export const STILL_WAITING_MS = 5_000; + +/** + * Tells that the server is slow to answer the first load, and lets the person + * stop waiting. Shows nothing while the load is quick or once it has settled. + */ +export function LaunchWait({ + load, + source, +}: { + load: Promise; + source: Pick; +}) { + const settled = useSettled(load) !== undefined; + const [slow, setSlow] = useState(false); + useEffect(() => { + const timer = setTimeout(() => setSlow(true), STILL_WAITING_MS); + return () => clearTimeout(timer); + }, []); + if (settled || !slow) return null; + return ( +
+ Stop waiting} + > + The server has not answered yet. It may be busy, or stuck. If you stop + waiting, the launch code may already have been used. + +
+ ); +} diff --git a/workflow-canvas/src/main.tsx b/workflow-canvas/src/main.tsx index 19ab66c34..eefff50e5 100644 --- a/workflow-canvas/src/main.tsx +++ b/workflow-canvas/src/main.tsx @@ -4,8 +4,10 @@ import './theme/fonts'; import { StrictMode } from 'react'; import { createRoot } from 'react-dom/client'; import { App } from './App'; +import { sessionRunMemory } from './execution/run-memory'; import { createHttpApi } from './api/http-api'; import { LaunchNotice } from './launch/LaunchNotice'; +import { LaunchWait } from './launch/LaunchWait'; import { takeLaunchCode, watchLaunchCode } from './launch/launch-code'; import { sessionStorageMemory } from './launch/session-memory'; import { installTheme } from './theme/install'; @@ -24,11 +26,20 @@ const source = createWorkflowSource( ); watchLaunchCode(window, (code) => void source.offer(code)); +const runMemory = sessionRunMemory(window); + function Root() { + const load = useWorkflowSource(source); return ( } + load={load} + notice={ + <> + + + + } + runMemory={runMemory} /> ); } diff --git a/workflow-canvas/src/workflow/load-workflow.test.ts b/workflow-canvas/src/workflow/load-workflow.test.ts index 525870fbd..7778199ac 100644 --- a/workflow-canvas/src/workflow/load-workflow.test.ts +++ b/workflow-canvas/src/workflow/load-workflow.test.ts @@ -27,7 +27,7 @@ describe('loadWorkflow', () => { const result = await loadWorkflow(api, 'code-1', freshMemory()); - expect(api.createSession).toHaveBeenCalledWith('code-1'); + expect(api.createSession).toHaveBeenCalledWith('code-1', undefined); expect(api.getWorkflow).toHaveBeenCalledWith('token-1'); expect(result).toEqual({ ok: true, @@ -304,7 +304,7 @@ describe('loadWorkflow with a launch code', () => { await loadWorkflow(api, 'code-1', memory); - expect(api.createSession).toHaveBeenCalledWith('code-1'); + expect(api.createSession).toHaveBeenCalledWith('code-1', undefined); expect(api.getWorkflow).toHaveBeenCalledOnce(); expect(api.getWorkflow).toHaveBeenCalledWith('token-1'); }); diff --git a/workflow-canvas/src/workflow/load-workflow.ts b/workflow-canvas/src/workflow/load-workflow.ts index dc8ae9e2d..be75c77b4 100644 --- a/workflow-canvas/src/workflow/load-workflow.ts +++ b/workflow-canvas/src/workflow/load-workflow.ts @@ -174,10 +174,14 @@ async function openWithCode( launchCode: string, memory: SessionMemory, pasted: boolean, + signal?: AbortSignal, ): Promise { const startKept = memory.recall(); try { - const { token, harness, apiVersion } = await api.createSession(launchCode); + const { token, harness, apiVersion } = await api.createSession( + launchCode, + signal, + ); if (apiVersion !== SUPPORTED_API_VERSION) return failure(UNSUPPORTED_API); const session = { token, harness }; keepEarly(memory, session, startKept, pasted && startKept !== undefined); @@ -247,15 +251,17 @@ export async function loadWorkflow( * of the code replaces the kept one at once, as no workflow is on the page * yet, and a code that fails (as a spent one does) falls back to the sessions * kept before it. The error of the code is shown only if those fail too. + * Aborting `signal` ends the trade of the code. */ export async function loadOnOpen( api: WorkflowApi, launchCode: string | undefined, memory: SessionMemory, + signal?: AbortSignal, ): Promise { if (launchCode === undefined) return loadWorkflow(api, undefined, memory); const before = resumable(memory); - const result = await openWithCode(api, launchCode, memory, false); + const result = await openWithCode(api, launchCode, memory, false, signal); return result.ok ? result : resumeAny(api, before, memory, result.message); } diff --git a/workflow-canvas/src/workflow/use-workflow-source.test.tsx b/workflow-canvas/src/workflow/use-workflow-source.test.tsx index d4cb95ac5..7b6cee3a9 100644 --- a/workflow-canvas/src/workflow/use-workflow-source.test.tsx +++ b/workflow-canvas/src/workflow/use-workflow-source.test.tsx @@ -29,6 +29,9 @@ describe('the page, under strict mode', () => { ); expect(await screen.findByTestId('node-stage-draft')).toBeTruthy(); - expect(api.createSession).toHaveBeenCalledExactlyOnceWith('code-1'); + expect(api.createSession).toHaveBeenCalledExactlyOnceWith( + 'code-1', + expect.any(AbortSignal), + ); }); }); diff --git a/workflow-canvas/src/workflow/workflow-source.test.ts b/workflow-canvas/src/workflow/workflow-source.test.ts index 36977585b..d19e19798 100644 --- a/workflow-canvas/src/workflow/workflow-source.test.ts +++ b/workflow-canvas/src/workflow/workflow-source.test.ts @@ -39,7 +39,10 @@ describe('createWorkflowSource', () => { await source.current(); await source.current(); - expect(createSession).toHaveBeenCalledExactlyOnceWith('first'); + expect(createSession).toHaveBeenCalledExactlyOnceWith( + 'first', + expect.any(AbortSignal), + ); }); it('changes to the workflow of a new code once its session is open', async () => { @@ -55,7 +58,7 @@ describe('createWorkflowSource', () => { await source.offer('second'); - expect(createSession).toHaveBeenLastCalledWith('second'); + expect(createSession).toHaveBeenLastCalledWith('second', undefined); expect(source.current()).not.toBe(before); expect(await source.current()).toMatchObject({ ok: true }); expect(listener).toHaveBeenCalledOnce(); @@ -666,3 +669,42 @@ describe('createWorkflowSource session memory', () => { expect(memory.recallPending()).toEqual(pending); }); }); + +describe('createWorkflowSource cancel', () => { + it('aborts the trade of the first code and shows its error', async () => { + const createSession = vi.fn( + (_code: string, signal?: AbortSignal) => + new Promise((_resolve, reject) => { + signal?.addEventListener('abort', () => + reject(new ApiError('You stopped waiting.', 0)), + ); + }), + ); + const source = createWorkflowSource( + apiWith(createSession), + 'first', + freshMemory(), + ); + + source.cancel(); + + expect(await source.current()).toEqual({ + ok: false, + message: 'You stopped waiting.', + }); + }); + + it('does nothing once the first load has settled', async () => { + const source = createWorkflowSource( + apiWith(vi.fn().mockResolvedValue(HELLO_SESSION)), + 'first', + freshMemory(), + ); + const first = await source.current(); + + source.cancel(); + + expect(first.ok).toBe(true); + expect(await source.current()).toBe(first); + }); +}); diff --git a/workflow-canvas/src/workflow/workflow-source.ts b/workflow-canvas/src/workflow/workflow-source.ts index 48cb755ec..6de685d6a 100644 --- a/workflow-canvas/src/workflow/workflow-source.ts +++ b/workflow-canvas/src/workflow/workflow-source.ts @@ -20,6 +20,8 @@ export interface WorkflowSource { subscribeNotice(listener: () => void): () => void; /** Drops the notice. */ dismissNotice(): void; + /** Stops waiting for the launch code of the first load; a no-op after it. */ + cancel(): void; } /** @@ -33,7 +35,8 @@ export function createWorkflowSource( firstCode: string | undefined, memory: SessionMemory, ): WorkflowSource { - let current = loadOnOpen(api, firstCode, memory); + const firstLoad = new AbortController(); + let current = loadOnOpen(api, firstCode, memory, firstLoad.signal); let lastOffer = 0; let shownOffer = 0; let shownWorkflowOffer = 0; @@ -86,6 +89,7 @@ export function createWorkflowSource( return () => noticeListeners.delete(listener); }, dismissNotice: () => setNotice(undefined), + cancel: () => firstLoad.abort(), async offer(code) { const offer = ++lastOffer; await offerLoaded(offer, await loadWorkflow(api, code, memory));