diff --git a/.github/actions/prepare-test-runtime/action.yml b/.github/actions/prepare-test-runtime/action.yml index 098f8b037..821fe45d5 100644 --- a/.github/actions/prepare-test-runtime/action.yml +++ b/.github/actions/prepare-test-runtime/action.yml @@ -37,27 +37,15 @@ runs: - name: Compile test files run: pnpm run clean && pnpm exec tsc shell: bash - - name: Install arf binary for Interactive tests + # mise.toml and mise.lock pin the arf version and its checksums. + - name: Install arf for Interactive tests + if: runner.os != 'Windows' + uses: jdx/mise-action@6612f0ef07ed2a736cd1e00a94a065522dd41195 # v5.2.5 + - name: Expose arf for Interactive tests if: runner.os != 'Windows' shell: bash - env: - RUNNER_OS_NAME: ${{ runner.os }} run: | - ARF_ARCH="$(uname -m)" - if [ "${ARF_ARCH}" = "arm64" ]; then ARF_ARCH="aarch64"; fi - if [ "${RUNNER_OS_NAME}" = "macOS" ]; then - ARF_TARGET="${ARF_ARCH}-apple-darwin" - else - ARF_TARGET="${ARF_ARCH}-unknown-linux-gnu" - fi - ARF_ARCHIVE="arf-console-${ARF_TARGET}.tar.xz" - ARF_RELEASE="https://github.com/eitsupi/arf/releases/download/v0.5.3" - curl -fsSL "${ARF_RELEASE}/${ARF_ARCHIVE}" -o "${RUNNER_TEMP}/${ARF_ARCHIVE}" - curl -fsSL "${ARF_RELEASE}/sha256.sum" -o "${RUNNER_TEMP}/arf.sha256" - (cd "${RUNNER_TEMP}" && awk -v name="${ARF_ARCHIVE}" '$2 == name || $2 == "*" name' arf.sha256 > arf-selected.sha256 && test -s arf-selected.sha256 && shasum -a 256 -c arf-selected.sha256) - mkdir -p "${RUNNER_TEMP}/arf" - tar -xJf "${RUNNER_TEMP}/${ARF_ARCHIVE}" -C "${RUNNER_TEMP}/arf" - ARF_EXECUTABLE=$(find "${RUNNER_TEMP}/arf" -type f -name arf | head -n 1) - test -n "${ARF_EXECUTABLE}" - echo "$(dirname "${ARF_EXECUTABLE}")" >> "$GITHUB_PATH" + ARF_EXECUTABLE="$(mise which arf)" + test -x "${ARF_EXECUTABLE}" + "${ARF_EXECUTABLE}" --version echo "ARF_PATH=${ARF_EXECUTABLE}" >> "$GITHUB_ENV" diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index c0091396b..642d31879 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -86,7 +86,7 @@ Interactive session. Run the source/bootstrap checks with The [architecture and backend contract](src/interactive/README.md) live beside the implementation. User setup and behavior belong in the [R Interactive wiki page](https://github.com/REditorSupport/vscode-R/wiki/R-Interactive). -Use Linux or macOS for Interactive runtime tests, with R and an installed arf 0.5.3 executable. Bundled sess is pure R and needs no compiler when its Imports are already installed; missing Imports are installed from the configured repository. Windows runs the remaining extension checks; persistent Interactive supervision remains limited to Linux/macOS. From the repository root, after installing pnpm dependencies, prepare a test library: +Use Linux or macOS for Interactive runtime tests, with R and arf 0.7.1 or later. `mise install` installs the arf version pinned in `mise.toml`, which CI also uses. Bundled sess is pure R and needs no compiler when its Imports are already installed; missing Imports are installed from the configured repository. Windows runs the remaining extension checks; persistent Interactive supervision remains limited to Linux/macOS. From the repository root, after installing pnpm dependencies, prepare a test library: ```sh export R_LIBS=/path/to/test-library diff --git a/mise.lock b/mise.lock new file mode 100644 index 000000000..b2eb53bd7 --- /dev/null +++ b/mise.lock @@ -0,0 +1,25 @@ +# @generated - this file is auto-generated by `mise lock` https://mise.jdx.dev/dev-tools/mise-lock.html + +[[tools."github:eitsupi/arf"]] +version = "0.7.1" +backend = "github:eitsupi/arf" + +[tools."github:eitsupi/arf"."platforms.linux-arm64"] +checksum = "sha256:ac305016f81568a8382ce60e4840ec1e08d10c19784accaf3d20fc60c4be35a2" +url = "https://github.com/eitsupi/arf/releases/download/v0.7.1/arf-console-aarch64-unknown-linux-gnu.tar.xz" +url_api = "https://api.github.com/repos/eitsupi/arf/releases/assets/747777115" + +[tools."github:eitsupi/arf"."platforms.linux-x64"] +checksum = "sha256:05e9f4df17b6def75a6b8356e732c64d70e173ff4dee41d844d50c63b136f211" +url = "https://github.com/eitsupi/arf/releases/download/v0.7.1/arf-console-x86_64-unknown-linux-gnu.tar.xz" +url_api = "https://api.github.com/repos/eitsupi/arf/releases/assets/747777208" + +[tools."github:eitsupi/arf"."platforms.macos-arm64"] +checksum = "sha256:c790f1459355c6774589f8a7f058be203519c105f8605c9380d5b94fca6e935a" +url = "https://github.com/eitsupi/arf/releases/download/v0.7.1/arf-console-aarch64-apple-darwin.tar.xz" +url_api = "https://api.github.com/repos/eitsupi/arf/releases/assets/747777120" + +[tools."github:eitsupi/arf"."platforms.macos-x64"] +checksum = "sha256:a2b9cd4b00775674a1d2d2af56e4684268a5042afa6f12b0d02dd9407c498f44" +url = "https://github.com/eitsupi/arf/releases/download/v0.7.1/arf-console-x86_64-apple-darwin.tar.xz" +url_api = "https://api.github.com/repos/eitsupi/arf/releases/assets/747777182" diff --git a/mise.toml b/mise.toml new file mode 100644 index 000000000..40270fe1d --- /dev/null +++ b/mise.toml @@ -0,0 +1,8 @@ +# Tools for the Interactive runtime tests. mise.lock pins each platform's +# download and checksum; run `mise lock` after changing a version. +[tools] +# arf is not installed on Windows, where Interactive runtime tests do not run. +"github:eitsupi/arf" = { version = "0.7.1", os = ["linux", "macos"] } + +[settings] +lockfile = true diff --git a/sess/R/interactive.R b/sess/R/interactive.R index 78beb36c9..ccd178c1b 100644 --- a/sess/R/interactive.R +++ b/sess/R/interactive.R @@ -53,6 +53,26 @@ invisible(NULL) } +# Under arf, tell the agent which arf execution runs this submission, so that +# it can interrupt that execution by ID. The event carries arf's execution ID. +# Detect the routines on each call: on Unix arf registers them only after +# startup, so an early negative result must not be cached. +.interactive_arf_started <- function(id) { + # arity-lint skip routine-registration: arf registers these routines in the embedding host, not in a package that sess could bind + if ( + is.loaded("arf_api_version", PACKAGE = "(embedding)", type = ".Call") && + identical(.Call("arf_api_version", PACKAGE = "(embedding)"), 1L) + ) { + .Call( + "arf_publish_event", + "vscode-r.interactive.started", + list(id = id), + PACKAGE = "(embedding)" + ) + } + invisible(NULL) +} + .interactive_write <- function(connection, text) { remaining <- processx::conn_write(connection, charToRaw(enc2utf8(text))) while (is.raw(remaining) && length(remaining)) { @@ -333,6 +353,7 @@ interactive_execute <- function(id, code, source = NULL) { # The agent can interrupt as soon as started arrives. Install the condition # handlers before publishing it or doing any interruptible setup work. .interactive_event("started") + .interactive_arf_started(id) .interactive_plot_context() trace_state$depth <- sys.nframe() filename <- if (is.null(source$uri)) "" else source$uri diff --git a/src/interactive/README.md b/src/interactive/README.md index 290745047..fc8e7fe88 100644 --- a/src/interactive/README.md +++ b/src/interactive/README.md @@ -39,9 +39,10 @@ The shipped `sess` backend uses arf with pure R sess helpers: [SessBackend](backends/sessBackend.ts) owns readiness, metadata, capabilities, and ordered output. [SessBridge](backends/sessBridge.ts) owns authenticated sess JSON-RPC, inspection, rich events, and requests to the editor. [Arf](backends/arf.ts) -owns launch/adoption and dispatch; [process.ts](backends/process.ts) handles -signals, ownership, and confirmed process exit. Shared discovery/HTTP helpers -remain in [arf.ts](arf.ts). Plain R Interactive creation is unavailable; ordinary +owns launch/adoption, dispatch, interruption, and Stop; [process.ts](backends/process.ts) +handles ownership, signals, and confirmed process exit. Shared discovery/HTTP +helpers and the version check remain in [arf.ts](arf.ts), and the long-lived +connection in [arfAttach.ts](arfAttach.ts). Plain R Interactive creation is unavailable; ordinary R terminals are unaffected, and existing agents can still be reconnected. Managed sessions dispatch the private `sess:::interactive_execute` entry point @@ -55,6 +56,17 @@ R sinks, frontend styling, and forked workers cannot rewrite those frames. A server reply timeout after execution started does not cancel R; its events still establish completion. Other transport failures remain ambiguous without retries. +The backend attaches to arf before submitting code. Before running a submission, +sess publishes an arf event carrying its ID, so the event's arf execution ID +identifies that submission; a managed `user_input` acknowledgement reports the +same ID. Interruption is an arf IPC `interrupt` for that execution ID, held until +the ID is known and dropped once sess reports completion. It is never a signal: +headless arf ends the whole session on SIGINT when R is idle, so a signal that +races with completion would end the session. sess reports the outcome; arf's +acceptance does not. Stop sends arf `shutdown` and waits for the process to exit, +killing it after a deadline. An interactive arf rejects `shutdown` as an unknown +method; only that answer makes Stop fall back to SIGTERM and then SIGKILL. + Adopted sessions use visible arf evaluation and return captured console output on completion, while rich events use sess RPC. The backend waits for both completion and the captured response. There is no subscription for arbitrary terminal console @@ -64,7 +76,8 @@ sess hooks without changing frontend console callbacks or killing adopted R. Both providers report notebook stdin/debugger support as unavailable: arf IPC `user_input` submits code, rather than answering nested R prompts. Use ordinary R terminals for `readline`, `scan`, and debugger interaction. arf IPC is -experimental; the integration is tested against arf 0.5.3. Persistent supervision +experimental; the integration requires arf 0.7.1 or later and is tested with +the version pinned in `mise.toml`. Persistent supervision currently remains limited to Linux/macOS. [SessGraphics](backends/sessGraphics.ts) owns JGD negotiation, its font-metrics worker, plot attribution/coalescing, SVG production, and resize transport. It emits display payloads with stable identities. The agent stores assets and journals references. Static graphics remains available without JGD; another backend can supply SVG/PNG directly. diff --git a/src/interactive/arf.ts b/src/interactive/arf.ts index e4ba799ff..38af2921a 100644 --- a/src/interactive/arf.ts +++ b/src/interactive/arf.ts @@ -13,6 +13,87 @@ export interface ArfSession { session_type?: string; } +// TODO: Require arf 0.7.2 once it is released. It reports an R error that +// follows a caught interrupt as an error rather than as interrupted, and bounds +// console reads that cannot be answered. +export const MIN_ARF_VERSION = '0.7.1'; + +/** arf JSON-RPC error codes that the Interactive client acts on. */ +export const ARF_METHOD_NOT_FOUND = -32601; +export const ARF_TIMEOUT = -32007; + +/** A JSON-RPC error response from arf, keeping its code and data. */ +export class ArfRpcError extends Error { + constructor( + readonly code: number, + readonly rpcMessage: string, + readonly data: unknown, + raw: unknown, + ) { + super(JSON.stringify(raw)); + this.name = 'ArfRpcError'; + } +} + +/** The client stopped waiting for arf; the request may still be running. */ +export class ArfTransportTimeoutError extends Error { + constructor() { + super('arf request timed out; evaluation may still be running'); + this.name = 'ArfTransportTimeoutError'; + } +} + +export function rpcError(error: unknown): ArfRpcError { + const value = + error && typeof error === 'object' && !Array.isArray(error) + ? (error as Record) + : {}; + return new ArfRpcError( + typeof value.code === 'number' ? value.code : NaN, + typeof value.message === 'string' ? value.message : '', + value.data, + error, + ); +} + +/** Compares release versions; prereleases count as older than their release. */ +export function arfVersionAtLeast(version: unknown, minimum: string): boolean { + const parse = (text: string) => { + const match = /^v?(\d+)\.(\d+)\.(\d+)(-[0-9A-Za-z.-]+)?(?:\+[0-9A-Za-z.-]+)?$/.exec(text); + return match + ? { core: [Number(match[1]), Number(match[2]), Number(match[3])], pre: !!match[4] } + : undefined; + }; + const actual = typeof version === 'string' ? parse(version) : undefined; + const required = parse(minimum); + if (!actual || !required) { + return false; + } + for (let i = 0; i < 3; i++) { + if (actual.core[i] !== required.core[i]) { + return actual.core[i] > required.core[i]; + } + } + return !actual.pre || required.pre; +} + +/** Rejects sessions older than MIN_ARF_VERSION, or without the deadline used by inspection. */ +export function checkArfSession(session: Record): void { + if (!arfVersionAtLeast(session.arf_version, MIN_ARF_VERSION)) { + const actual = + typeof session.arf_version === 'string' ? session.arf_version : 'an unknown version'; + throw new Error( + `R Interactive requires arf ${MIN_ARF_VERSION} or later, but the session runs ${actual}. Update arf and start a new session.`, + ); + } + const capabilities = Array.isArray(session.capabilities) ? session.capabilities : []; + if (!capabilities.includes('call_execution_timeout')) { + throw new Error( + 'This arf session does not support execution deadlines for handler calls. Update arf and start a new session.', + ); + } +} + /** arf speaks HTTP JSON-RPC, not sess's JSON Lines framing. */ export function arfRequest( endpoint: string, @@ -52,7 +133,7 @@ export function arfRequest( try { const result = object(JSON.parse(Buffer.concat(chunks).toString('utf8'))); if (result.error) { - reject(new Error(JSON.stringify(result.error))); + reject(rpcError(result.error)); } else { resolve(result.result); } @@ -63,11 +144,7 @@ export function arfRequest( }, ); if (timeout > 0) { - request.setTimeout(timeout, () => - request.destroy( - new Error('arf request timed out; evaluation may still be running'), - ), - ); + request.setTimeout(timeout, () => request.destroy(new ArfTransportTimeoutError())); } request.on('error', reject); request.end(body); diff --git a/src/interactive/arfAttach.ts b/src/interactive/arfAttach.ts new file mode 100644 index 000000000..e080eee3e --- /dev/null +++ b/src/interactive/arfAttach.ts @@ -0,0 +1,219 @@ +import * as net from 'net'; +import { JsonLines } from './framing'; +import { object } from './protocol'; +import { ArfTransportTimeoutError, rpcError } from './arf'; + +/** One event notification from an attached arf connection. */ +export interface ArfEvent { + seq: number; + topic: string; + data: unknown; + /** arf's execution ID, or null when the event belongs to no execution. */ + execution: number | null; +} + +export interface ArfAttachResult { + connectionId: number; + apiVersion: number; + capabilities: string[]; + sessionId: string; + session: Record; +} + +export interface ArfAttachOptions { + subscribe?: string[]; + client?: { name: string; version?: string }; + /** Bounds connecting and the attach response, not later requests. */ + timeout?: number; + /** Registered before attaching, so no event after the response is missed. */ + onEvent?: (event: ArfEvent) => void; + onClose?: (reason: Error) => void; +} + +interface Pending { + resolve(value: unknown): void; + reject(error: Error): void; + timer?: NodeJS.Timeout; +} + +/** + * A long-lived arf IPC connection (newline-delimited JSON-RPC). + * + * Responses are matched by ID because they arrive in completion order. A + * request whose wait ends locally (timeout or abort) may still run in R; its + * late response is ignored and its ID is never reused. The connection closes + * once, rejecting every pending request with the same reason, and is never + * reopened: callers attach again and reconcile their own state. + */ +export class ArfAttachConnection { + private pending = new Map(); + private nextId = 1; + private closeReason?: Error; + + private constructor( + private socket: net.Socket, + private onClose?: (reason: Error) => void, + ) {} + + static open( + endpoint: string, + options: ArfAttachOptions = {}, + ): Promise<{ connection: ArfAttachConnection; result: ArfAttachResult }> { + return new Promise((resolve, reject) => { + const socket = net.connect(endpoint); + const connection = new ArfAttachConnection(socket, options.onClose); + const attachId = connection.nextId++; + let attached = false; + const timer = setTimeout( + () => connection.close(new ArfTransportTimeoutError()), + options.timeout ?? 30000, + ); + connection.pending.set(attachId, { + resolve: (value) => { + clearTimeout(timer); + try { + const result = object(value); + attached = true; + resolve({ + connection, + result: { + connectionId: Number(result.connection_id), + apiVersion: Number(result.api_version), + capabilities: Array.isArray(result.capabilities) + ? result.capabilities.map(String) + : [], + sessionId: + typeof result.session_id === 'string' ? result.session_id : '', + session: object(result.session), + }, + }); + } catch (error) { + connection.close(error as Error); + reject(error); + } + }, + reject: (error) => { + clearTimeout(timer); + // arf closes the connection after refusing attach. + connection.close(error); + reject(error); + }, + }); + const parser = new JsonLines((message) => { + // No event is sent before the attach response. + if (message.method === 'event' && attached) { + options.onEvent?.(event(message.params)); + } else if (message.id !== undefined) { + connection.response(message); + } + }); + socket.on('data', (chunk: Buffer) => { + try { + parser.push(chunk); + } catch (error) { + connection.close(error as Error); + } + }); + socket.on('error', (error) => connection.close(error)); + socket.on('close', () => connection.close(new Error('arf closed the IPC connection'))); + socket.write( + JSON.stringify({ + jsonrpc: '2.0', + id: attachId, + method: 'attach', + params: { + protocol: 1, + subscribe: options.subscribe ?? [], + client: options.client ?? { name: 'vscode-R' }, + }, + }) + '\n', + ); + }); + } + + get closed(): Error | undefined { + return this.closeReason; + } + + request( + method: string, + params: Record = {}, + options: { timeout?: number; signal?: AbortSignal } = {}, + ): Promise { + if (this.closeReason) { + return Promise.reject(this.closeReason); + } + const id = this.nextId++; + return new Promise((resolve, reject) => { + const settle = (): void => { + clearTimeout(entry.timer); + options.signal?.removeEventListener('abort', abort); + this.pending.delete(id); + }; + const abort = (): void => { + settle(); + reject(new Error('arf request was aborted; it may still be running')); + }; + const entry: Pending = { + resolve: (value) => { + settle(); + resolve(value); + }, + reject: (error) => { + settle(); + reject(error); + }, + }; + if (options.timeout && options.timeout > 0) { + entry.timer = setTimeout(() => { + settle(); + reject(new ArfTransportTimeoutError()); + }, options.timeout); + } + if (options.signal?.aborted) { + abort(); + return; + } + options.signal?.addEventListener('abort', abort, { once: true }); + this.pending.set(id, entry); + this.socket.write(JSON.stringify({ jsonrpc: '2.0', id, method, params }) + '\n'); + }); + } + + close(reason: Error = new Error('arf IPC connection was closed')): void { + if (this.closeReason) { + return; + } + this.closeReason = reason; + for (const entry of this.pending.values()) { + entry.reject(reason); + } + this.pending.clear(); + this.socket.destroy(); + this.onClose?.(reason); + } + + private response(message: Record): void { + const entry = typeof message.id === 'number' ? this.pending.get(message.id) : undefined; + if (!entry) { + // A request that timed out or was aborted locally. + return; + } + this.pending.delete(Number(message.id)); + if (message.error !== undefined) { + entry.reject(rpcError(message.error)); + } else { + entry.resolve(message.result); + } + } +} + +function event(params: unknown): ArfEvent { + const value = object(params); + return { + seq: Number(value.seq), + topic: String(value.topic), + data: value.data, + execution: Number.isSafeInteger(value.execution) ? Number(value.execution) : null, + }; +} diff --git a/src/interactive/backends/arf.ts b/src/interactive/backends/arf.ts index 064dfa4e1..79b04f395 100644 --- a/src/interactive/backends/arf.ts +++ b/src/interactive/backends/arf.ts @@ -1,15 +1,37 @@ -import { arfRequest } from '../arf'; +import { + ARF_METHOD_NOT_FOUND, + ARF_TIMEOUT, + ArfRpcError, + arfRequest, + checkArfSession, +} from '../arf'; +import { ArfAttachConnection, ArfEvent } from '../arfAttach'; import { BackendEvent } from '../backend'; import { object, Submission } from '../protocol'; import { RuntimeProcess } from './process'; import { rString } from './rCode'; import { ArfOutput } from './arfOutput'; +/** Topic that sess publishes before running submitted code under arf. */ +const STARTED_TOPIC = 'vscode-r.interactive.started'; + +/** The submission sent to arf, and arf's execution ID once known. */ +interface Execution { + submission: string; + id?: number; + interruptRequested: boolean; +} + export class Arf { readonly process: RuntimeProcess; ready = false; private abort = new AbortController(); private started = false; + private attach?: ArfAttachConnection; + private sessionId?: string; + private execution?: Execution; + private failure?: Error; + private stopping?: Promise; constructor( private command: string, private directory: string, @@ -100,9 +122,9 @@ export class Arf { if (!this.endpoint) { throw new Error('Missing arf endpoint'); } - const metadata = object( - await arfRequest(this.endpoint, 'session', {}, 30000, this.abort.signal), - ); + // Subscribe before submitting anything so that no started event is missed. + const metadata = await this.connect(); + checkArfSession(metadata); this.emit({ type: 'provider', data: { @@ -120,19 +142,169 @@ export class Arf { this.abort.signal, ), ); - if (result.error) { - throw new Error( - typeof result.error === 'string' ? result.error : JSON.stringify(result.error), - ); + // A missing or unknown status is not a successful bootstrap. + if (result.status !== 'ok') { + const detail = + typeof result.error === 'string' && result.error + ? result.error + : typeof result.stderr === 'string' && result.stderr.trim() + ? result.stderr.trim() + : `status ${JSON.stringify(result.status ?? null)}`; + throw new Error(`Interactive bootstrap failed in arf: ${detail}`); } } + private async connect(): Promise> { + const { connection, result } = await ArfAttachConnection.open(this.endpoint!, { + subscribe: [STARTED_TOPIC], + client: { name: 'vscode-R Interactive' }, + onEvent: (event) => this.arfEvent(event), + }); + if (this.sessionId !== undefined && result.sessionId !== this.sessionId) { + connection.close(); + throw new Error('The arf endpoint now belongs to a different session'); + } + this.sessionId = result.sessionId; + this.attach = connection; + return result.session; + } dispose(): void { this.abort.abort(); + this.attach?.close(); this.process.dispose(); } + /** Called once sess reports that a submission finished. */ + finished(submission: string): void { + if (this.execution?.submission === submission) { + this.execution = undefined; + } + } + async interrupt(): Promise { + const execution = this.execution; + if (!execution) { + // TODO: Interrupt slow inspection through execution deadlines of + // handler calls; sess runs it where arf does not deliver interrupts. + throw new Error('No submitted code is running'); + } + execution.interruptRequested = true; + if (execution.id !== undefined) { + await this.sendInterrupt(execution); + } else if (this.attach?.closed) { + throw new Error( + 'Cannot interrupt: the connection to arf was lost before the execution started', + ); + } + // Otherwise the request is sent once arf reports the execution ID. + } + private async sendInterrupt(execution: Execution): Promise { + if (this.execution !== execution || execution.id === undefined) { + return; + } + // Acceptance only registers the request; sess reports the outcome. A + // rejection means the execution already ended, which races with Stop. + await arfRequest(this.endpoint!, 'interrupt', { execution_id: execution.id }, 5000); + } + /** Records arf's execution ID for a submission, from its reply or its started event. */ + private register(submission: unknown, id: unknown): void { + const execution = this.execution; + if (!execution || execution.submission !== submission) { + return; + } + if (!Number.isSafeInteger(id) || (id as number) < 0) { + this.fail(new Error('arf reported an invalid execution ID')); + return; + } + if (execution.id === undefined) { + execution.id = id as number; + if (execution.interruptRequested) { + void this.sendInterrupt(execution).catch((error: Error) => + this.emit({ type: 'error', message: error.message }), + ); + } + } else if (execution.id !== id) { + this.fail(new Error('arf reported conflicting execution IDs for one submission')); + } + } + private arfEvent(event: ArfEvent): void { + if (event.topic !== STARTED_TOPIC) { + return; + } + const data = event.data; + const submission = + data && typeof data === 'object' ? (data as Record).id : undefined; + this.register(submission, event.execution); + } + private fail(error: Error): void { + this.failure ??= error; + this.emit({ type: 'unavailable', message: error.message }); + } + stop(force = false): Promise { + return (this.stopping ??= this.stopRuntime(force).finally(() => { + this.stopping = undefined; + })); + } + private async stopRuntime(force: boolean): Promise { + if (!this.process.alive()) { + this.process.markExited(); + return; + } + if (force) { + await this.kill(undefined, 0); + return; + } + if (!this.endpoint) { + if (this.process.adopted) { + throw new Error('The arf endpoint is unknown'); + } + await this.kill('SIGTERM', 3000); + return; + } + try { + await arfRequest(this.endpoint, 'shutdown', {}, 5000); + } catch (error) { + if ( + this.process.adopted && + error instanceof ArfRpcError && + error.code === ARF_METHOD_NOT_FOUND + ) { + // An interactive arf rejects shutdown; stop it as a terminal would. + await this.kill('SIGTERM', 3000); + return; + } + if (!this.process.adopted) { + // The managed process is ours even if its IPC no longer answers. + await this.kill('SIGTERM', 3000); + return; + } + throw error; + } + // Acceptance only starts shutdown, and running R code may delay it. + await this.kill(undefined, 5000); + } + /** Sends `first`, waits up to `grace` ms, then kills; resolves after a confirmed exit. */ + private async kill(first: NodeJS.Signals | undefined, grace: number): Promise { + if (first) { + this.process.send(first); + } + if (await this.process.waitForExit(grace)) { + return; + } + this.process.send('SIGKILL'); + if (!(await this.process.waitForExit(2000))) { + throw new Error('R has not exited after Stop'); + } + } async dispatch(submission: Submission): Promise { + if (this.failure) { + throw this.failure; + } + if (!this.attach || this.attach.closed) { + // The previous execution has finished, so attaching again loses no + // state; the session must still be the same arf process. + await this.connect(); + } const code = `sess:::interactive_execute(${rString(submission.id)}, ${rString(submission.code)}, jsonlite::fromJSON(${rString(JSON.stringify(submission.source ?? null))}, simplifyVector=FALSE))`; this.started = false; + this.execution = { submission: submission.id, interruptRequested: false }; // Managed sessions use send to avoid arf's unbounded evaluation capture. // R's ordered events establish the outcome even if arf's reply times out. let result: Record; @@ -149,11 +321,20 @@ export class Arf { } catch (error) { // A server timeout does not cancel R. Keep accepting its event stream. // Other transport failures remain ambiguous and must block dispatch. - if (!this.process.adopted && this.started && /timed out|timeout/i.test(String(error))) { + if ( + !this.process.adopted && + this.started && + error instanceof ArfRpcError && + error.code === ARF_TIMEOUT + ) { return; } + if (!this.started) { + this.finished(submission.id); + } throw error; } + this.register(submission.id, result.execution_id); if (this.process.adopted) { for (const channel of ['stdout', 'stderr']) { if (typeof result[channel] === 'string' && result[channel]) { diff --git a/src/interactive/backends/process.ts b/src/interactive/backends/process.ts index 8263df2c3..0c8643d5f 100644 --- a/src/interactive/backends/process.ts +++ b/src/interactive/backends/process.ts @@ -9,7 +9,6 @@ export class RuntimeProcess { private exited = false; private closed = false; private monitor?: NodeJS.Timeout; - private stopping?: Promise; constructor( readonly adopted: boolean, private emit: (event: BackendEvent) => void, @@ -53,7 +52,7 @@ export class RuntimeProcess { this.monitor.unref(); } } - private alive(): boolean { + alive(): boolean { if (this.exited || !this.pid) { return false; } @@ -77,7 +76,15 @@ export class RuntimeProcess { this.emit({ type: 'exit', code, signal }); } } - private signal(signal: NodeJS.Signals): void { + /** Records an exit observed without a process event, such as an adopted PID that vanished. */ + markExited(): void { + this.exit(); + } + /** + * Sends a signal to the R process. The interrupt and stop paths use arf IPC + * instead: headless arf ends the whole session on SIGINT. + */ + send(signal: NodeJS.Signals): void { if (!this.alive()) { this.exit(); return; @@ -92,46 +99,27 @@ export class RuntimeProcess { } } } - interrupt(): void { - this.signal('SIGINT'); - } - stop(force = false): Promise { - return (this.stopping ??= this.stopProcess(force)); - } - private async stopProcess(force: boolean): Promise { - if (!this.pid) { - return; - } - this.signal(force ? 'SIGKILL' : 'SIGINT'); + /** Resolves true once the process has exited, or false after `timeout` ms. */ + async waitForExit(timeout: number): Promise { const start = Date.now(); - let terminated = force, - killed = force; while (this.alive()) { if (this.closed) { - return; - } - const elapsed = Date.now() - start; - if (!terminated && elapsed >= 100) { - this.signal('SIGTERM'); - terminated = true; - } - if (!killed && elapsed >= 3000) { - this.signal('SIGKILL'); - killed = true; + return true; } - if (elapsed > 5000) { - throw new Error('R has not exited after Stop'); + if (Date.now() - start >= timeout) { + return false; } await new Promise((resolve) => setTimeout(resolve, 25)); } this.exit(); + return true; } dispose(): void { if (this.closed) { return; } if (!this.adopted) { - this.signal('SIGKILL'); + this.send('SIGKILL'); } this.closed = true; clearInterval(this.monitor); diff --git a/src/interactive/backends/sessBackend.ts b/src/interactive/backends/sessBackend.ts index 39736be14..01e15bb5e 100644 --- a/src/interactive/backends/sessBackend.ts +++ b/src/interactive/backends/sessBackend.ts @@ -208,13 +208,12 @@ export class SessBackend implements SessionBackend { return Promise.resolve(); } interrupt(): Promise { - this.frontend.process.interrupt(); - return Promise.resolve(); + return this.frontend.interrupt(); } async stop(options?: { force?: boolean }): Promise { // Stop may arrive after the control endpoint opens but before R is spawned. await this.startup; - await this.frontend.process.stop(options?.force); + await this.frontend.stop(options?.force); } resizePlot(request: ResizePlotRequest): Promise { this.graphics.resize(request); @@ -394,6 +393,7 @@ export class SessBackend implements SessionBackend { return; } this.graphics.flush(executionId); + this.frontend.finished(executionId); this.current = undefined; this.completion = undefined; this.emit({ diff --git a/src/test/node/arfAttach.test.ts b/src/test/node/arfAttach.test.ts new file mode 100644 index 000000000..bb0bca127 --- /dev/null +++ b/src/test/node/arfAttach.test.ts @@ -0,0 +1,250 @@ +import * as assert from 'assert'; +import * as fs from 'fs'; +import * as net from 'net'; +import * as os from 'os'; +import * as path from 'path'; +import { + ArfRpcError, + ArfTransportTimeoutError, + arfVersionAtLeast, + checkArfSession, +} from '../../interactive/arf'; +import { ArfAttachConnection, ArfEvent } from '../../interactive/arfAttach'; + +/** A fake arf server that hands each parsed line to `handle`. */ +function fakeArf( + handle: (message: Record, socket: net.Socket) => void, +): Promise<{ endpoint: string; server: net.Server; dispose(): void }> { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'r-arf-attach-')); + // arf listens on a named pipe on Windows and a Unix socket elsewhere. + const endpoint = + process.platform === 'win32' + ? `\\\\.\\pipe\\${path.basename(directory)}` + : path.join(directory, 'arf.sock'); + const sockets = new Set(); + const server = net.createServer((socket) => { + sockets.add(socket); + let buffer = ''; + socket.on('data', (chunk) => { + buffer += chunk.toString(); + let newline: number; + while ((newline = buffer.indexOf('\n')) >= 0) { + const line = buffer.slice(0, newline); + buffer = buffer.slice(newline + 1); + handle(JSON.parse(line) as Record, socket); + } + }); + socket.on('error', () => undefined); + }); + return new Promise((resolve) => + server.listen(endpoint, () => + resolve({ + endpoint, + server, + dispose() { + sockets.forEach((socket) => socket.destroy()); + server.close(); + fs.rmSync(directory, { recursive: true, force: true }); + }, + }), + ), + ); +} + +const line = (message: unknown): string => JSON.stringify(message) + '\n'; +const attachResult = { + connection_id: 3, + api_version: 1, + capabilities: ['events', 'call_execution_timeout'], + session_id: 'session-a', + session: { arf_version: '0.7.1', capabilities: ['call_execution_timeout'] }, +}; + +suite('arf attach connection', () => { + test('delivers an event that arrives in the same chunk as the attach response', async () => { + const arf = await fakeArf((message, socket) => { + if (message.method === 'attach') { + socket.write( + line({ jsonrpc: '2.0', id: message.id, result: attachResult }) + + line({ + jsonrpc: '2.0', + method: 'event', + params: { + seq: 9, + topic: 'demo.started', + data: { id: 'a' }, + execution: 7, + }, + }), + ); + } + }); + try { + const events: ArfEvent[] = []; + const { connection, result } = await ArfAttachConnection.open(arf.endpoint, { + subscribe: ['demo.*'], + onEvent: (event) => events.push(event), + }); + assert.strictEqual(result.sessionId, 'session-a'); + assert.strictEqual(result.session.arf_version, '0.7.1'); + await new Promise((resolve) => setImmediate(resolve)); + assert.deepStrictEqual(events, [ + { seq: 9, topic: 'demo.started', data: { id: 'a' }, execution: 7 }, + ]); + connection.close(); + } finally { + arf.dispose(); + } + }); + + test('matches responses by ID and keeps RPC error codes and data', async () => { + const held: Record[] = []; + const arf = await fakeArf((message, socket) => { + if (message.method === 'attach') { + socket.write(line({ jsonrpc: '2.0', id: message.id, result: attachResult })); + } else { + held.push(message); + if (held.length === 2) { + // Answer in reverse order, the second one with an error. + socket.write( + line({ + jsonrpc: '2.0', + id: held[1].id, + error: { + code: -32007, + message: 'Request timed out', + data: { execution_id: 4 }, + }, + }) + line({ jsonrpc: '2.0', id: held[0].id, result: { first: true } }), + ); + } + } + }); + try { + const { connection } = await ArfAttachConnection.open(arf.endpoint); + const first = connection.request('session'); + const second = connection.request('evaluate', { code: '1' }).then( + () => assert.fail('expected an RPC error'), + (error: unknown) => error, + ); + assert.deepStrictEqual(await first, { first: true }); + const error = (await second) as ArfRpcError; + assert.ok(error instanceof ArfRpcError); + assert.strictEqual(error.code, -32007); + assert.strictEqual(error.rpcMessage, 'Request timed out'); + assert.deepStrictEqual(error.data, { execution_id: 4 }); + connection.close(); + } finally { + arf.dispose(); + } + }); + + test('ignores a late response after a local timeout and never reuses its ID', async () => { + const ids: unknown[] = []; + let late: (() => void) | undefined; + const arf = await fakeArf((message, socket) => { + if (message.method === 'attach') { + socket.write(line({ jsonrpc: '2.0', id: message.id, result: attachResult })); + return; + } + ids.push(message.id); + if (message.method === 'slow') { + late = () => socket.write(line({ jsonrpc: '2.0', id: message.id, result: 'late' })); + } else { + late?.(); + socket.write(line({ jsonrpc: '2.0', id: message.id, result: 'fresh' })); + } + }); + try { + const { connection } = await ArfAttachConnection.open(arf.endpoint); + await assert.rejects( + connection.request('slow', {}, { timeout: 20 }), + ArfTransportTimeoutError, + ); + assert.strictEqual(await connection.request('fast'), 'fresh'); + assert.notStrictEqual(ids[0], ids[1]); + assert.strictEqual(connection.closed, undefined); + connection.close(); + } finally { + arf.dispose(); + } + }); + + test('closes once, rejecting pending and later requests with the same reason', async () => { + const arf = await fakeArf((message, socket) => { + if (message.method === 'attach') { + socket.write(line({ jsonrpc: '2.0', id: message.id, result: attachResult })); + } else { + socket.destroy(); + } + }); + try { + const closes: Error[] = []; + const { connection } = await ArfAttachConnection.open(arf.endpoint, { + onClose: (reason) => closes.push(reason), + }); + const pending = connection.request('evaluate').then( + () => assert.fail('expected the request to fail'), + (error: Error) => error, + ); + const reason = await pending; + assert.strictEqual(closes.length, 1); + assert.strictEqual(closes[0], reason); + assert.strictEqual(connection.closed, reason); + await assert.rejects(connection.request('session'), (error) => error === reason); + connection.close(); + assert.strictEqual(closes.length, 1); + } finally { + arf.dispose(); + } + }); + + test('rejects a refused attach with its RPC error', async () => { + const arf = await fakeArf((message, socket) => { + socket.end( + line({ + jsonrpc: '2.0', + id: message.id, + error: { code: -32602, message: 'unsupported protocol', data: null }, + }), + ); + }); + try { + await assert.rejects( + ArfAttachConnection.open(arf.endpoint), + (error) => error instanceof ArfRpcError && error.code === -32602, + ); + } finally { + arf.dispose(); + } + }); +}); + +suite('arf version requirement', () => { + test('compares release versions and treats prereleases as older', () => { + assert.ok(arfVersionAtLeast('0.7.1', '0.7.1')); + assert.ok(arfVersionAtLeast('0.7.2', '0.7.1')); + assert.ok(arfVersionAtLeast('0.10.0', '0.7.1')); + assert.ok(arfVersionAtLeast('1.0.0', '0.7.1')); + assert.ok(arfVersionAtLeast('0.7.2-beta.1', '0.7.1')); + assert.ok(!arfVersionAtLeast('0.7.1-beta.1', '0.7.1')); + assert.ok(!arfVersionAtLeast('0.7.0', '0.7.1')); + assert.ok(!arfVersionAtLeast('0.5.3', '0.7.1')); + assert.ok(!arfVersionAtLeast(undefined, '0.7.1')); + assert.ok(!arfVersionAtLeast('0.7', '0.7.1')); + assert.ok(!arfVersionAtLeast('latest', '0.7.1')); + }); + + test('rejects old sessions and sessions without handler call deadlines', () => { + checkArfSession({ arf_version: '0.7.1', capabilities: ['call_execution_timeout'] }); + assert.throws( + () => checkArfSession({ arf_version: '0.5.3', capabilities: [] }), + /requires arf 0\.7\.1 or later, but the session runs 0\.5\.3/, + ); + assert.throws(() => checkArfSession({}), /an unknown version/); + assert.throws( + () => checkArfSession({ arf_version: '0.7.1', capabilities: [] }), + /execution deadlines/, + ); + }); +}); diff --git a/src/test/node/interactiveBackend.test.ts b/src/test/node/interactiveBackend.test.ts index 61fd303cb..e894ea725 100644 --- a/src/test/node/interactiveBackend.test.ts +++ b/src/test/node/interactiveBackend.test.ts @@ -4,6 +4,7 @@ import * as os from 'os'; import * as path from 'path'; import { randomUUID } from 'crypto'; import { createServer } from 'http'; +import * as net from 'net'; import { SessionAgent } from '../../interactive/agent'; import { AgentClient } from '../../interactive/client'; import { AgentConfig, ExecutionRecord, SessionEvent, Submission } from '../../interactive/protocol'; @@ -449,11 +450,41 @@ suite('Interactive backend configuration', () => { return; } } - response.end(JSON.stringify({ jsonrpc: '2.0', id: 1, result: {} })); + response.end(JSON.stringify({ jsonrpc: '2.0', id: 1, result: { status: 'ok' } })); }); }); + // arf serves attach (newline-delimited JSON) and HTTP on one endpoint. + const listener = net.createServer((socket) => { + socket.once('data', (chunk: Buffer) => { + if (chunk[0] === '{'.charCodeAt(0)) { + const attach = JSON.parse(chunk.toString()) as { id: number }; + socket.write( + JSON.stringify({ + jsonrpc: '2.0', + id: attach.id, + result: { + connection_id: 1, + api_version: 1, + capabilities: ['events', 'call_execution_timeout'], + session_id: 'fake', + session: { + arf_version: '0.7.1', + capabilities: ['call_execution_timeout'], + }, + }, + }) + '\n', + ); + } else { + socket.pause(); + socket.unshift(chunk); + server.emit('connection', socket); + socket.resume(); + } + }); + socket.on('error', () => undefined); + }); const endpoint = path.join(root, 'arf.sock'); - await new Promise((resolve) => server.listen(endpoint, resolve)); + await new Promise((resolve) => listener.listen(endpoint, resolve)); const adapter = new Arf( 'unused', root, @@ -474,7 +505,7 @@ suite('Interactive backend configuration', () => { } finally { adapter.dispose(); server.closeAllConnections(); - await new Promise((resolve) => server.close(() => resolve())); + await new Promise((resolve) => listener.close(() => resolve())); } }); }); diff --git a/src/test/node/interactiveRuntime.test.ts b/src/test/node/interactiveRuntime.test.ts index a7543c33d..b948c0bf8 100644 --- a/src/test/node/interactiveRuntime.test.ts +++ b/src/test/node/interactiveRuntime.test.ts @@ -408,6 +408,17 @@ sess::display("display-${cell}", "text/plain"); cat("stderr-${cell}\\n", file=st assert.match(text(next), /17/); }); + test('an interrupt between executions does not end the session', async () => { + // A signal that reaches idle headless arf ends the session, so an + // interrupt that races with completion must not be sent as a signal. + const first = await submit('kept_between <- 23'); + assert.strictEqual((await finished(first)).state, 'success'); + await assert.rejects(client.request('interrupt'), /No submitted code is running/); + const next = await submit('kept_between'); + assert.strictEqual((await finished(next)).state, 'success'); + assert.match(text(next), /23/); + }); + test('interrupts a slow inspection and continues serving R requests', async () => { for (const afterTimeout of [false, true]) { const marker = `inspection-started-${String(afterTimeout)}`;