From b50260fd1a682e920f6b6b1e1ba7419082de07fb Mon Sep 17 00:00:00 2001 From: Tiffany Trinh Date: Wed, 21 Oct 2026 16:53:26 -0400 Subject: [PATCH] feat(apps): run local backend function calls in parallel with per-call log prefixes Each call already gets a fresh process, so calls no longer queue and run in parallel, with no local limit: production's concurrency limits vary by org, and dev:verify runs against them. Calls that start together share one runtime context query, so a burst sends Datadog one query instead of one per call. Each line a call writes is prefixed with its function name and call number. A failed $.Actions call now rejects as it does in a deployed app: with a bare string holding the platform's error detail, the deployed inputs or no-arguments message, or the local allowlist message, not an Error. An uncaught failure in the function's code reports as it does in a deployed app: the dev server's error is "Execution failed: " followed by the thrown Error's stack or the thrown value, and the dev transport throws that message instead of the status and raw response body. Refs: APPS-2961 Co-Authored-By: Claude Opus 5.5 --- packages/plugins/apps/README.md | 9 +- .../client/execute-backend-function.test.ts | 35 +++ .../client/transports/dev-server-transport.ts | 25 +- .../plugins/apps/src/vite/dev-server.test.ts | 252 ++++++++++++++++++ packages/plugins/apps/src/vite/dev-server.ts | 51 +++- .../apps/src/vite/local-execution-child.ts | 11 +- ...ocal-execution-process.integration.test.ts | 196 +++++++++++--- .../apps/src/vite/local-execution-process.ts | 100 ++++--- .../apps/src/vite/local-execution-protocol.ts | 4 +- .../apps/src/vite/local-execution-shared.ts | 45 +++- .../apps/src/vite/local-execution.test.ts | 24 +- .../plugins/apps/src/vite/local-execution.ts | 5 +- .../childActions.backend.ts | 18 ++ 13 files changed, 675 insertions(+), 100 deletions(-) diff --git a/packages/plugins/apps/README.md b/packages/plugins/apps/README.md index 4042e8be8..2e442ebe8 100644 --- a/packages/plugins/apps/README.md +++ b/packages/plugins/apps/README.md @@ -61,17 +61,18 @@ To call one by hand, send that content type, for example `curl -X POST -H 'Conte ## Local execution -On Vite 6.0.4 or later, `/__dd/executeAction` runs each backend function call the way production does: in a fresh Node process of its own. Calls run one at a time. +On Vite 6.0.4 or later, `/__dd/executeAction` runs each backend function call the way production does: in a fresh Node process of its own, in parallel with other calls. - **Fresh state**: module-level variables and globals start over on every call. +- **Concurrency**: calls run in parallel, with no local limit, so a large burst starts that many processes. Production's concurrency limits vary by org, and `dev:verify` runs against them. Calls that start together share one Datadog query for their runtime context (`$.Source`). - **Timeout**: a call that runs past its time budget is killed, even inside a synchronous loop. - **Sandbox**: Node's [permission model](https://nodejs.org/api/permissions.html) lets a call read the project, its `node_modules` folders and the packages linked from them, Vite and Vite's [`server.fs.allow`](https://vite.dev/config/server-options#server-fs-allow) folders, and write only to its own temp dir (`os.tmpdir()`), which is also its working directory, and `/tmp`. Modules loaded through Vite, though not `fs` reads or a native `require`, refuse files matching Vite's [`server.fs.deny`](https://vite.dev/config/server-options#server-fs-deny), such as `.env`, and, unless `server.fs.strict` is `false`, files outside those folders. It refuses subprocesses, worker threads and native addons, allows the network, and caps the heap at 256 MB. A known gap, from Node's [documented limitations](https://nodejs.org/api/permissions.html#limitations-and-known-issues): "Symbolic links will be followed even to locations outside of the set of paths that access has been granted to." Here, a symlink in a writable folder (the call's temp dir or `/tmp`) lets a call read or write wherever it points, project files included. Local execution is a parity aid, not a security boundary. - **Environment**: a call sees no variables from the dev server's environment, keys included, except Node's TLS and time zone settings (`NODE_EXTRA_CA_CERTS`, `SSL_CERT_FILE`, `SSL_CERT_DIR`, `TZ`), and no Node options from `NODE_OPTIONS`. `$.Actions` calls go back to the dev server, which holds the credentials and checks each call's connection against the function's allowed connections. -- **Logs**: a call's `stdout` and `stderr` reach the dev server's terminal line by line, as each line ends. -- **Debugging**: when the dev server runs under a debugger (`--inspect`), each call's process listens on a random local port and prints its `Debugger listening on ws://…` URL. Attach to that URL to debug the call; calls don't time out while the dev server is under a debugger. +- **Logs**: a call's `stdout` and `stderr` reach the dev server's terminal line by line, each prefixed with its call, for example `[getGreeting#3]`. +- **Debugging**: when the dev server runs under a debugger (`--inspect`), each call's process listens on a random local port and prints its `Debugger listening on ws://…` URL with its prefix. Attach to that URL to debug the call; calls don't time out while the dev server is under a debugger. - **Older Node**: on a Node without the permission model, or without `--allow-inspector` while debugging, a call's process runs under in-process guards instead. They refuse writes outside the temp dirs, subprocesses and worker threads, from the module's top-level code on. -Vite 5 has no module runner transport, so there each call runs inside the dev server's process. Vite 6.0.0 to 6.0.3, whose dev server answers the module runner in an older shape, do the same with a warning at startup, as do a Yarn Plug'n'Play project, whose packages load through hooks a call's process doesn't get, and any Vite 6+ project where `vite/module-runner` can't be resolved. +Vite 5 has no module runner transport, so there each call runs inside the dev server's process, one at a time. Vite 6.0.0 to 6.0.3, whose dev server answers the module runner in an older shape, do the same with a warning at startup, as do a Yarn Plug'n'Play project, whose packages load through hooks a call's process doesn't get, and any Vite 6+ project where `vite/module-runner` can't be resolved. ## Package output diff --git a/packages/plugins/apps/src/backend/client/execute-backend-function.test.ts b/packages/plugins/apps/src/backend/client/execute-backend-function.test.ts index be1504dde..fcb9083df 100644 --- a/packages/plugins/apps/src/backend/client/execute-backend-function.test.ts +++ b/packages/plugins/apps/src/backend/client/execute-backend-function.test.ts @@ -73,6 +73,41 @@ describe('executeBackendFunction', () => { ); }); + it("should throw with the dev server's error, as a deployed app does", async () => { + global.fetch = jest.fn().mockResolvedValue({ + ok: false, + status: 500, + text: async () => + JSON.stringify({ success: false, error: 'Execution failed: plain error x' }), + }); + + const error = await executeBackendFunction('testFunction', []).catch((e: unknown) => e); + + expect(error).toBeInstanceOf(BackendFunctionError); + expect(error).toMatchObject({ + message: 'Execution failed: plain error x', + functionName: 'testFunction', + statusCode: 500, + }); + }); + + it.each([ + ['a non-string error', JSON.stringify({ success: false, error: 42 })], + ['an empty error', JSON.stringify({ success: false, error: '' })], + ['a success body', JSON.stringify({ success: true, error: 'unused' })], + ['JSON that is not an object', 'null'], + ])('should keep the status and raw body for %s', async (_label, body) => { + global.fetch = jest.fn().mockResolvedValue({ + ok: false, + status: 500, + text: async () => body, + }); + + await expect(executeBackendFunction('testFunction', [])).rejects.toThrow( + `Backend function "testFunction" failed with status 500: ${body}`, + ); + }); + it('should throw BackendFunctionError when response contains error field', async () => { const mockResponse = { error: 'Function execution failed', diff --git a/packages/plugins/apps/src/backend/client/transports/dev-server-transport.ts b/packages/plugins/apps/src/backend/client/transports/dev-server-transport.ts index 3ed187233..4f2ac956e 100644 --- a/packages/plugins/apps/src/backend/client/transports/dev-server-transport.ts +++ b/packages/plugins/apps/src/backend/client/transports/dev-server-transport.ts @@ -8,6 +8,26 @@ import { BackendFunctionError } from '../types'; const ENDPOINT = '/__dd/executeAction'; +function parseErrorMessage(body: string): string | undefined { + try { + const parsed: unknown = JSON.parse(body); + if ( + typeof parsed === 'object' && + parsed !== null && + 'success' in parsed && + parsed.success === false && + 'error' in parsed && + typeof parsed.error === 'string' && + parsed.error !== '' + ) { + return parsed.error; + } + } catch { + return undefined; + } + return undefined; +} + export const devServerTransport: BackendFunctionTransport = async ( functionName: string, args: unknown[], @@ -39,7 +59,10 @@ export const devServerTransport: BackendFunctionTransport = async ( let errorMessage = `Backend function "${functionName}" failed with status ${response.status}`; try { const errorBody = await response.text(); - if (errorBody) { + const parsedMessage = parseErrorMessage(errorBody); + if (parsedMessage !== undefined) { + errorMessage = parsedMessage; + } else if (errorBody) { errorMessage += `: ${errorBody}`; } } catch { diff --git a/packages/plugins/apps/src/vite/dev-server.test.ts b/packages/plugins/apps/src/vite/dev-server.test.ts index 9e1d332fa..a6dd886f9 100644 --- a/packages/plugins/apps/src/vite/dev-server.test.ts +++ b/packages/plugins/apps/src/vite/dev-server.test.ts @@ -13,6 +13,7 @@ import { formatRef, getRetryDelay, } from '@dd/apps-plugin/vite/dev-server'; +import type { LocalExecutionRequest } from '@dd/apps-plugin/vite/local-execution-process'; import { DEFAULT_TIMEOUT_MS, withTimeout } from '@dd/apps-plugin/vite/local-execution'; import type { AuthOptionsWithDefaults, RequestOpts } from '@dd/core/types'; import { @@ -1204,6 +1205,67 @@ describe('Dev Server Middleware', () => { ); expect(mockLoadModule).not.toHaveBeenCalled(); }); + + test('Should send one runtime context query for calls that need it at the same time', async () => { + nock.cleanAll(); + const hydration = mockRuntimeContextHydration(1); + const callCount = 3; + const allStarted = createDeferred(); + let started = 0; + const executeLocally = jest.fn(async (request: LocalExecutionRequest) => { + started += 1; + if (started === callCount) { + allStarted.resolve(); + } + await withTimeout(allStarted.promise, 5000, 'Waiting for every call to start'); + const data = await request.getRuntimeContext(); + return { data }; + }); + const childMiddleware = createTestMiddleware({ executeLocally }); + const responses = Array.from({ length: callCount }, () => { + const req = createMockRequest('/__dd/executeAction', { + functionName: encodeQueryName(mockFunctions[0]), + args: [], + }); + const res = createMockResponse(); + childMiddleware(req, res, jest.fn()); + return res; + }); + const allDone = responses.map((res) => res.done); + await Promise.all(allDone); + + for (const res of responses) { + expect(res.statusCode).toBe(200); + expect(JSON.parse(res.getBody()).result).toEqual({ + data: previewRuntimeContext, + }); + } + expect(hydration.isDone()).toBe(true); + }); + + test('Should query the runtime context again for a call that starts after the last query ended', async () => { + nock.cleanAll(); + const hydration = mockRuntimeContextHydration(2); + const executeLocally = jest.fn(async (request: LocalExecutionRequest) => { + const data = await request.getRuntimeContext(); + return { data }; + }); + const childMiddleware = createTestMiddleware({ executeLocally }); + const callAndWait = async () => { + const req = createMockRequest('/__dd/executeAction', { + functionName: encodeQueryName(mockFunctions[0]), + args: [], + }); + const res = createMockResponse(); + childMiddleware(req, res, jest.fn()); + await res.done; + return res.statusCode; + }; + + expect(await callAndWait()).toBe(200); + expect(await callAndWait()).toBe(200); + expect(hydration.isDone()).toBe(true); + }); }); test('Should hydrate context, run the function directly in-process, and return its result without bundling', async () => { @@ -1588,6 +1650,196 @@ describe('Dev Server Middleware', () => { ); }); + describe('What a failed $.Actions call rejects with, as in a deployed app', () => { + const catchingPostMessage = (call: Record) => async () => { + try { + await testDollarActions().slack.chat.postMessage(call); + return 'no error'; + } catch (error) { + return { type: typeof error, value: error }; + } + }; + const runAndReadData = async () => { + const req = createMockRequest('/__dd/executeAction', { + functionName: encodeQueryName(mockFunctions[0]), + args: [], + }); + const res = createMockResponse(); + middleware(req, res, jest.fn()); + await res.done; + return JSON.parse(res.getBody()).result.data; + }; + const mockActionErrors = (errors: Array<{ detail?: string; title?: string }>) => + nock(DD_API_ORIGIN) + .post('/api/v2/app-builder/queries/preview-async') + .reply(200, { data: { id: 'receipt-action-failure' } }) + .get( + '/api/v2/app-builder/queries/execution-long-polling/receipt-action-failure', + ) + .reply(200, { errors }); + + test("Should reject with the platform's bare error detail as a string", async () => { + mockLoadModuleReturning( + mockFunctions[0], + catchingPostMessage({ inputs: { text: 'hi' } }), + ); + mockActionErrors([ + { detail: 'rpc error: code = NotFound desc = connection not found' }, + ]); + + await expect(runAndReadData()).resolves.toEqual({ + type: 'string', + value: 'rpc error: code = NotFound desc = connection not found', + }); + }); + + test("Should join several of the platform's errors, using a title when there's no detail", async () => { + mockLoadModuleReturning( + mockFunctions[0], + catchingPostMessage({ inputs: { text: 'hi' } }), + ); + mockActionErrors([{ detail: 'first' }, { title: 'second' }]); + + await expect(runAndReadData()).resolves.toEqual({ + type: 'string', + value: 'first; second', + }); + }); + + test('Should reject a call with no inputs with the deployed wording, as a string', async () => { + mockLoadModuleReturning(mockFunctions[0], catchingPostMessage({})); + + await expect(runAndReadData()).resolves.toEqual({ + type: 'string', + value: 'First argument to action $.Actions.slack.chat.postMessage must have an inputs field that is an object', + }); + }); + + test('Should report an uncaught failure as "Execution failed: " and the bare detail', async () => { + mockLoadModuleReturning(mockFunctions[0], () => + testDollarActions().slack.chat.postMessage({ inputs: { text: 'hi' } }), + ); + mockActionErrors([ + { detail: 'a connection is required for com.datadoghq.slack.chat' }, + ]); + const req = createMockRequest('/__dd/executeAction', { + functionName: encodeQueryName(mockFunctions[0]), + args: [], + }); + const res = createMockResponse(); + middleware(req, res, jest.fn()); + await res.done; + + expect(res.statusCode).toBe(500); + expect(JSON.parse(res.getBody())).toEqual({ + success: false, + error: 'Execution failed: a connection is required for com.datadoghq.slack.chat', + }); + }); + + test('Should reject a call with no arguments with the deployed wording, as a string', async () => { + mockLoadModuleReturning(mockFunctions[0], async () => { + try { + await testDollarActions().slack.chat.postMessage(); + return 'no error'; + } catch (error) { + return { type: typeof error, value: error }; + } + }); + + await expect(runAndReadData()).resolves.toEqual({ + type: 'string', + value: 'No arguments provided to action $.Actions.slack.chat.postMessage', + }); + }); + + test('Should reject with a fallback message when no platform error has a detail or a title', async () => { + mockLoadModuleReturning( + mockFunctions[0], + catchingPostMessage({ inputs: { text: 'hi' } }), + ); + mockActionErrors([{}, { detail: '' }]); + + await expect(runAndReadData()).resolves.toEqual({ + type: 'string', + value: 'Query execution failed', + }); + }); + + test("Should reject a connection outside the function's allowlist as a string", async () => { + mockLoadModuleReturning( + mockFunctions[0], + catchingPostMessage({ inputs: { text: 'hi' }, connectionId: 'not-listed' }), + ); + + await expect(runAndReadData()).resolves.toEqual({ + type: 'string', + value: `Action $.Actions.slack.chat.postMessage used connection "not-listed", which is not in this function's allowed connections: []`, + }); + }); + }); + + describe('What an uncaught failure reports, as in a deployed app', () => { + const runAndReadError = async (fn: () => unknown) => { + mockLoadModuleReturning(mockFunctions[0], fn); + const req = createMockRequest('/__dd/executeAction', { + functionName: encodeQueryName(mockFunctions[0]), + args: [], + }); + const res = createMockResponse(); + middleware(req, res, jest.fn()); + await res.done; + return { statusCode: res.statusCode, error: JSON.parse(res.getBody()).error }; + }; + + test('Should prefix a thrown string with "Execution failed: "', async () => { + await expect( + runAndReadError(() => { + const failure = 'plain string failure'; + throw failure; + }), + ).resolves.toEqual({ + statusCode: 500, + error: 'Execution failed: plain string failure', + }); + }); + + test("Should report a thrown Error's stack after the prefix", async () => { + const { statusCode, error } = await runAndReadError(() => { + throw new TypeError('plain error x'); + }); + expect(statusCode).toBe(500); + expect(error).toMatch(/^Execution failed: TypeError: plain error x\n {4}at /); + }); + + test('Should report any other thrown value as a string', async () => { + await expect( + runAndReadError(() => { + const failure = { a: 1 }; + throw failure; + }), + ).resolves.toEqual({ statusCode: 500, error: 'Execution failed: [object Object]' }); + }); + + test("Should leave the dev server's own failures unprefixed", async () => { + mockLoadModule.mockImplementation( + moduleResolverFor(mockFunctions[0], { notTheFunction: () => 'unused' }), + ); + const req = createMockRequest('/__dd/executeAction', { + functionName: encodeQueryName(mockFunctions[0]), + args: [], + }); + const res = createMockResponse(); + middleware(req, res, jest.fn()); + await res.done; + + expect(res.statusCode).toBe(500); + expect(JSON.parse(res.getBody()).error).toBe( + `"${mockFunctions[0].name}" is not a function exported from ${mockFunctions[0].absolutePath}`, + ); + }); + }); + // The priming loadModule call (see handleExecuteAction) is the only place the entry's // top-level code runs (Vite caches the module for executeScriptLocally's reuse), so it // must carry the same $-scoping guarantee executeScriptLocally's own load would provide. diff --git a/packages/plugins/apps/src/vite/dev-server.ts b/packages/plugins/apps/src/vite/dev-server.ts index d8f6f9354..3d05dd6ad 100644 --- a/packages/plugins/apps/src/vite/dev-server.ts +++ b/packages/plugins/apps/src/vite/dev-server.ts @@ -25,6 +25,7 @@ import { getBaseBackendBuildConfig } from './build-config'; import type { HostCheck } from './dev-server-request-guard'; import { checkDevServerRequest } from './dev-server-request-guard'; import type { LocalExecutor } from './local-execution-process'; +import { FunctionThrewError, throwActionFailure } from './local-execution-shared'; import type { ExecuteAction, GetRuntimeContext, LoadModule } from './local-execution'; import { DEFAULT_TIMEOUT_MS, executeColdActionLocally } from './local-execution'; import { getMaxRetryDelayMs } from './retry-delay'; @@ -290,6 +291,17 @@ async function executeScriptViaDatadog( return outputs; } +/** Callers that arrive while a run is in flight share its result instead of starting their own. */ +function shareInFlight(): (run: () => Promise) => Promise { + let pending: Promise | undefined; + return (run) => { + pending ??= run().finally(() => { + pending = undefined; + }); + return pending; + }; +} + /** * Resolves the serializable `$` context exposed by an authenticated preview execution. The * platform response is deliberately kept opaque here; local-execution validates the identity @@ -315,9 +327,9 @@ function makeGetRuntimeContextRemotely( }); try { // The request signal cancels active fetches and our own retry delays. Racing the whole - // operation also releases the local-execution queue if doRequest is already sleeping - // inside async-retry, whose backoff timer does not observe AbortSignal. Promise.race - // keeps observing submitQuery's eventual settlement after the deadline wins. + // operation also fails the call on time, freeing the queue on Vite 5, if doRequest is + // already sleeping inside async-retry, whose backoff timer does not observe AbortSignal. + // Promise.race keeps observing submitQuery's eventual settlement after the deadline wins. const outputs = await Promise.race([ submitQuery( { @@ -391,11 +403,21 @@ function makeExecuteActionRemotely( } catch (error: unknown) { const message = error instanceof Error ? error.message : String(error); log.error(`$.Actions call to "${fqn}" failed: ${message}`); + if (error instanceof QueryExecutionError) { + throwActionFailure(error.detail); + } throw error; } }; } +/** A query that ran and failed; `detail` is what a deployed `$.Actions` call rejects with. */ +class QueryExecutionError extends Error { + constructor(readonly detail: string) { + super(`Query execution failed: ${detail}`); + } +} + interface PollResult { data?: { attributes?: { done?: boolean; outputs?: unknown } }; errors?: Array<{ detail?: string; title?: string }>; @@ -453,8 +475,11 @@ async function pollQueryExecution( // Check for error responses. if (result.errors?.length) { - const details = result.errors.map((e) => e.detail || e.title).join('; '); - throw new Error(`Query execution failed: ${details}`); + const details = result.errors + .map((e) => e.detail || e.title) + .filter(Boolean) + .join('; '); + throw new QueryExecutionError(details || 'Query execution failed'); } const attrs = result.data?.attributes; @@ -511,7 +536,14 @@ function guardAuthenticated( /** Shared catch-block shape for every handler below: an `HttpError` carries its own status code, anything else is a 500. `label` is omitted for handlers with no `log` in scope. */ function handleHttpError(res: ServerResponse, error: unknown, log?: Logger, label?: string): void { const statusCode = error instanceof HttpError ? error.statusCode : 500; - const message = error instanceof Error ? error.message : 'Internal server error'; + const message = + error instanceof FunctionThrewError + ? `Execution failed: ${error.detail}` + : typeof error === 'string' + ? error + : error instanceof Error + ? error.message + : 'Internal server error'; if (log && label) { log.debug(`Error handling ${label}: ${message}`); } @@ -602,6 +634,7 @@ async function handleExecuteAction( projectRoot: string, log: Logger, executeLocally: LocalExecutor | undefined, + shareRuntimeContext: (run: GetRuntimeContext) => Promise, ): Promise { try { const { func, args } = await parseAndLookupFunction(req, functionsByName); @@ -617,12 +650,13 @@ async function handleExecuteAction( longPolling, log, ); - const getRuntimeContext = makeGetRuntimeContextRemotely( + const getRuntimeContextRemotely = makeGetRuntimeContextRemotely( auth, doAuthenticatedRequest, longPolling, log, ); + const getRuntimeContext = () => shareRuntimeContext(getRuntimeContextRemotely); const result = executeLocally ? await executeLocally({ func, @@ -759,6 +793,8 @@ export function createDevServerMiddleware( longPolling, log, ); + // One query for a burst of parallel calls, so they don't each count against production's limit. + const shareRuntimeContext = shareInFlight(); const runLocally: RouteHandler = (req, res, functionsByName) => guardAuthenticated(res, doAuthenticatedRequest, (authedRequest) => handleExecuteAction( @@ -773,6 +809,7 @@ export function createDevServerMiddleware( projectRoot, log, executeLocally, + shareRuntimeContext, ), ); const routeHandlers: Record = { diff --git a/packages/plugins/apps/src/vite/local-execution-child.ts b/packages/plugins/apps/src/vite/local-execution-child.ts index c419728f3..0832a2965 100644 --- a/packages/plugins/apps/src/vite/local-execution-child.ts +++ b/packages/plugins/apps/src/vite/local-execution-child.ts @@ -15,6 +15,8 @@ import type { ActionCallArgs } from './local-execution-shared'; import { assertJsonRoundTrippable, assertNoSelfDeclaredScope, + callBackendFunction, + FunctionThrewError, isIndexableRecord, makeActionsProxy, returnValueSubject, @@ -34,7 +36,7 @@ type ModuleRunnerModule = { ESModulesEvaluator: new () => unknown; }; -type Settle = { resolve(value: unknown): void; reject(error: Error): void }; +type Settle = { resolve(value: unknown): void; reject(reason: unknown): void }; const send = process.send?.bind(process); let nextRequestId = 0; @@ -103,8 +105,8 @@ function onParentMessage(message: ParentMessage): void { if (message.ok) { settle?.resolve(message.value); } else { - const error = new Error(message.message); - settle?.reject(error); + const failure = message.thrownString ? message.message : new Error(message.message); + settle?.reject(failure); } } } @@ -188,7 +190,7 @@ async function load(config: ChildStartConfig): Promise { Reflect.set(globalThis, '$', $); await registerSdkAdapters(runner, config, $); await sendMessage({ type: 'loaded' }); - const returned = await fn(...config.args); + const returned = await callBackendFunction(() => fn(...config.args)); const subject = returnValueSubject(config.functionName); return assertJsonRoundTrippable(returned, subject); } @@ -211,6 +213,7 @@ async function run(config: ChildStartConfig): Promise { outcome = { type: 'error', message: error instanceof Error ? error.message : String(error), + ...(error instanceof FunctionThrewError && { detail: error.detail }), }; } const stdoutFlushed = flush(process.stdout, writeStdout, config.flushTimeoutMs); diff --git a/packages/plugins/apps/src/vite/local-execution-process.integration.test.ts b/packages/plugins/apps/src/vite/local-execution-process.integration.test.ts index 58f0e7ac5..4690e8ad6 100644 --- a/packages/plugins/apps/src/vite/local-execution-process.integration.test.ts +++ b/packages/plugins/apps/src/vite/local-execution-process.integration.test.ts @@ -53,19 +53,14 @@ const fixtureFunction = ( allowedConnectionIds, }); -type TimedCall = { calls: number; pid: number; startedAt: number; endedAt: number }; +type PidResult = { pid: number; result: unknown }; -const isTimedCall = (value: unknown): value is TimedCall => +const isPidResult = (value: unknown): value is PidResult => typeof value === 'object' && value !== null && - 'calls' in value && 'pid' in value && - 'startedAt' in value && - 'endedAt' in value && - typeof value.calls === 'number' && - typeof value.pid === 'number' && - typeof value.startedAt === 'number' && - typeof value.endedAt === 'number'; + 'result' in value && + typeof value.pid === 'number'; const isProcessAlive = (pid: number) => { try { @@ -194,37 +189,54 @@ describe('Local execution in a child process', () => { ); test( - 'Should run concurrent calls one at a time, each in its own fresh process', + 'Should run concurrent calls in parallel, each in its own fresh process', async () => { const execute = makeExecutor(); + const reportPid = fixtureFunction('childActions', 'callHttpReportingPid'); + const callCount = 4; + // Each call's $.Actions request is held until all of them have arrived, which only + // happens if every call is running at once. + const held: Array<() => void> = []; + const executeAction: ExecuteAction = () => + new Promise((resolve) => { + held.push(() => resolve({ released: true })); + if (held.length === callCount) { + held.forEach((release) => release()); + } + }); + const results = await Promise.all( - [1, 2, 3].map(() => execute(makeRequest(countCalls, [300]))), + Array.from({ length: callCount }, () => + execute(makeRequest(reportPid, [], { executeAction })), + ), ); - const data = results.map((r) => r.data).filter(isTimedCall); - expect(data).toHaveLength(3); - expect(data.map((d) => d.calls)).toEqual([1, 1, 1]); - expect(new Set(data.map((d) => d.pid)).size).toBe(3); - const byStart = [...data].sort((a, b) => a.startedAt - b.startedAt); - const overlapping = byStart.filter( - (d, i) => i > 0 && d.startedAt < byStart[i - 1].endedAt, + const data = results.map((r) => r.data).filter(isPidResult); + expect(data.map((d) => d.result)).toEqual( + Array(callCount).fill({ released: true }), ); - expect(overlapping).toEqual([]); + const pids = data.map((d) => d.pid); + expect(new Set(pids).size).toBe(callCount); + expect(pids).not.toContain(process.pid); }, CALL_TIMEOUT_MS, ); test( - 'Should run each call in its own temp dir, so a .env in the project is not picked up from the working directory', + 'Should run each call, concurrent ones included, in its own temp dir, so a .env in the project is not picked up from the working directory', async () => { const execute = makeExecutor(); - const result = await execute( - makeRequest(fixtureFunction('childState', 'readWorkingDirectory'), []), - ); - expect(result.data).toEqual({ - cwd: expect.stringContaining('dd-apps-exec-'), - cwdIsTmpdir: true, - }); + const readWorkingDirectory = fixtureFunction('childState', 'readWorkingDirectory'); + const calls = [1, 2].map(() => execute(makeRequest(readWorkingDirectory, []))); + const results = await Promise.all(calls); + for (const result of results) { + expect(result.data).toEqual({ + cwd: expect.stringContaining('dd-apps-exec-'), + cwdIsTmpdir: true, + }); + } + const workingDirectories = results.map((result) => JSON.stringify(result.data)); + expect(new Set(workingDirectories).size).toBe(2); }, CALL_TIMEOUT_MS, ); @@ -1012,6 +1024,126 @@ describe('Local execution in a child process', () => { CALL_TIMEOUT_MS, ); + describe('What a failed call rejects with, as in a deployed app', () => { + const catchHttpFailure = fixtureFunction('childActions', 'catchHttpFailure', [ + 'conn-a', + ]); + const execute = (request: LocalExecutionRequest) => makeExecutor()(request); + + test( + "Should reject with the dev server's string as that string", + async () => { + const executeAction: ExecuteAction = () => { + const detail = 'rpc error: code = NotFound desc = connection not found'; + return Promise.reject(detail); + }; + const result = await execute( + makeRequest(catchHttpFailure, ['conn-a'], { executeAction }), + ); + expect(result.data).toEqual({ + type: 'string', + value: 'rpc error: code = NotFound desc = connection not found', + }); + }, + CALL_TIMEOUT_MS, + ); + + test( + "Should reject a connection outside the function's allowlist as a string", + async () => { + const result = await execute(makeRequest(catchHttpFailure, ['conn-b'])); + expect(result.data).toEqual({ + type: 'string', + value: `Action $.Actions.http.request used connection "conn-b", which is not in this function's allowed connections: [conn-a]`, + }); + }, + CALL_TIMEOUT_MS, + ); + + test( + 'Should reject a call with no inputs with the deployed wording, as a string', + async () => { + const result = await execute(makeRequest(catchHttpFailure, ['conn-a', false])); + expect(result.data).toEqual({ + type: 'string', + value: 'First argument to action $.Actions.http.request must have an inputs field that is an object', + }); + }, + CALL_TIMEOUT_MS, + ); + + test( + 'Should still reject with an Error when the dev server failed with one', + async () => { + const executeAction: ExecuteAction = () => Promise.reject(new Error('boom')); + const result = await execute( + makeRequest(catchHttpFailure, ['conn-a'], { executeAction }), + ); + expect(result.data).toEqual({ type: 'Error', value: 'boom' }); + }, + CALL_TIMEOUT_MS, + ); + }); + + describe('What an uncaught failure reports, as in a deployed app', () => { + const execute = (request: LocalExecutionRequest) => makeExecutor()(request); + const throwValue = fixtureFunction('childActions', 'throwValue'); + + test( + "Should report a thrown Error's stack as its detail", + async () => { + await expect(execute(makeRequest(throwValue, ['error']))).rejects.toMatchObject( + { + message: 'plain error x', + detail: expect.stringMatching(/^Error: plain error x\n {4}at /), + }, + ); + }, + CALL_TIMEOUT_MS, + ); + + test( + 'Should report a thrown string as its detail', + async () => { + await expect( + execute(makeRequest(throwValue, ['string'])), + ).rejects.toMatchObject({ + message: 'plain string failure', + detail: 'plain string failure', + }); + }, + CALL_TIMEOUT_MS, + ); + + test( + "Should report an uncaught $.Actions failure's bare detail", + async () => { + const executeAction: ExecuteAction = () => { + const detail = 'a connection is required for com.datadoghq.http.request'; + return Promise.reject(detail); + }; + const callHttp = fixtureFunction('childActions', 'callHttp'); + await expect( + execute(makeRequest(callHttp, [], { executeAction })), + ).rejects.toMatchObject({ + detail: 'a connection is required for com.datadoghq.http.request', + }); + }, + CALL_TIMEOUT_MS, + ); + + test( + "Should give the dev server's own failures no detail", + async () => { + const missing = fixtureFunction('childActions', 'notExported'); + const error = await execute(makeRequest(missing, [])).catch((e: unknown) => e); + expect(error).toBeInstanceOf(Error); + expect(error).not.toHaveProperty('detail'); + }, + CALL_TIMEOUT_MS, + ); + }); + test( 'Should refuse a forged IPC action request outside the allowed connections', async () => { @@ -1024,6 +1156,7 @@ describe('Local execution in a child process', () => { id: 999_999, ok: false, message: `Action forged used connection "conn-evil", which is not in this function's allowed connections: [conn-a]`, + thrownString: true, }); expect(calls).toEqual([]); }, @@ -1042,6 +1175,7 @@ describe('Local execution in a child process', () => { id: 999_999, ok: false, message: expect.stringContaining('Malformed $.Actions request'), + thrownString: false, }); expect(calls).toEqual([]); }, @@ -1136,7 +1270,7 @@ describe('Local execution in a child process', () => { describe('Logs', () => { test( - 'Should stream each line as it is written, not after the call ends', + 'Should stream each line with its call prefix as it is written, not after the call ends', async () => { const lines: Array<{ stream: OutputStream; line: string; at: number }> = []; const execute = makeExecutor({ @@ -1151,9 +1285,9 @@ describe('Local execution in a child process', () => { expect(result.data).toBe('logged'); expect(lines.map(({ stream, line }) => ({ stream, line }))).toEqual([ - { stream: 'stdout', line: 'first line' }, - { stream: 'stderr', line: 'second line' }, - { stream: 'stdout', line: 'no trailing newline' }, + { stream: 'stdout', line: '[logLines#2] first line' }, + { stream: 'stderr', line: '[logLines#2] second line' }, + { stream: 'stdout', line: '[logLines#2] no trailing newline' }, ]); expect(endedAt - lines[0].at).toBeGreaterThanOrEqual(delayMs * 0.75); }, diff --git a/packages/plugins/apps/src/vite/local-execution-process.ts b/packages/plugins/apps/src/vite/local-execution-process.ts index 17f2051d4..d188c211a 100644 --- a/packages/plugins/apps/src/vite/local-execution-process.ts +++ b/packages/plugins/apps/src/vite/local-execution-process.ts @@ -30,12 +30,15 @@ import { sharedTmpPaths, withRealPath, } from './local-execution-sandbox'; -import { isIndexableRecord, validateActionCall } from './local-execution-shared'; +import { + FunctionThrewError, + isIndexableRecord, + validateActionCall, +} from './local-execution-shared'; import type { ExecuteAction, GetRuntimeContext } from './local-execution'; import { assertValidRuntimeContext, deriveActionTimeouts, - enqueue, localExecutionResolutionContext, withTimeout, } from './local-execution'; @@ -264,6 +267,7 @@ interface ChildRun { executeAction: ExecuteAction; timeoutMs: number; longPolling: Required; + prefix: string; } export function createChildProcessExecutor( @@ -287,6 +291,7 @@ export function createChildProcessExecutor( const moduleRunnerUrl = pathToFileURL(moduleRunnerPath).href; // Node checks read access on each linked part of the entry's path while it starts. const [realChildEntryPath = childEntryPath] = withRealPath(childEntryPath).slice(-1); + let callCount = 0; const ownChildren: ChildTmpDirs = new Map(); let closed = false; @@ -299,7 +304,7 @@ export function createChildProcessExecutor( }; const runChild = async (run: ChildRun): Promise => { - const { func, config, allowedConnectionIds, executeAction, timeoutMs } = run; + const { func, config, allowedConnectionIds, executeAction, timeoutMs, prefix } = run; // A call still starting when its server closed must not fork on the closed server. refuseIfClosed(func); const { actionCallTimeoutMs, totalExecutionTimeoutMs } = deriveActionTimeouts( @@ -348,7 +353,7 @@ export function createChildProcessExecutor( let ranOutOfMemory = false; const writeChildLine = (stream: OutputStream) => (line: string) => { ranOutOfMemory ||= stream === 'stderr' && V8_OUT_OF_MEMORY_LINE.test(line); - writeOutput(stream, line); + writeOutput(stream, `${prefix} ${line}`); }; if (child.stdout && child.stderr) { const writeStdoutLine = writeChildLine('stdout'); @@ -466,8 +471,15 @@ export function createChildProcessExecutor( }, ) => { const replyWithError = (error: unknown) => { + const thrownString = typeof error === 'string'; const errorMessage = error instanceof Error ? error.message : String(error); - reply({ type: 'action-result', id, ok: false, message: errorMessage }); + reply({ + type: 'action-result', + id, + ok: false, + message: errorMessage, + thrownString, + }); }; let validated: { inputs: Record; @@ -535,6 +547,7 @@ export function createChildProcessExecutor( ok: false, message: 'Malformed $.Actions request: it needs a string actionId, an inputs object and an optional string connectionId.', + thrownString: false, }); } } else if (message.type === 'loaded' && !loaded) { @@ -551,7 +564,11 @@ export function createChildProcessExecutor( } else if (message.type === 'error') { const reason = typeof message.message === 'string' ? message.message : 'Unknown error'; - fail(reason); + if (typeof message.detail === 'string') { + finish(new FunctionThrewError(reason, message.detail)); + } else { + fail(reason); + } } }); // A process that fails to start may never emit 'close'. @@ -581,43 +598,42 @@ export function createChildProcessExecutor( }); }; - // One call at a time, like the in-process path. - const execute: LocalExecutor = (request) => - enqueue(async () => { - const { func, args, executeAction, getRuntimeContext, getAllowedConnectionIds } = - request; - refuseIfClosed(func); - const displayName = `${func.relativePath}/${func.name}`; - const runtimeContext = await getRuntimeContext(); - assertValidRuntimeContext(runtimeContext); - const connectionIdsResolving = getAllowedConnectionIds(func.absolutePath); - const allowedConnectionIds = await withTimeout( - connectionIdsResolving, - request.timeoutMs, - `Resolving allowed connections for "${displayName}"`, - ); - const actionCatalogInstalled = isActionCatalogInstalled(projectRoot); - const appsBackendInstalled = isDatadogAppsBackendInstalled(projectRoot); - return runChild({ - func, - allowedConnectionIds, - executeAction, - timeoutMs: request.timeoutMs, - longPolling: request.longPolling, - config: { - moduleRunnerUrl, - projectRoot, - entryUrl: func.absolutePath + LOCAL_EXECUTION_LOAD_SUFFIX, - entryPath: func.absolutePath, - functionName: func.name, - args, - runtimeContext, - actionCatalogInstalled, - appsBackendInstalled, - flushTimeoutMs, - }, - }); + const execute: LocalExecutor = async (request) => { + const { func, args, executeAction, getRuntimeContext, getAllowedConnectionIds } = request; + refuseIfClosed(func); + const displayName = `${func.relativePath}/${func.name}`; + const runtimeContext = await getRuntimeContext(); + assertValidRuntimeContext(runtimeContext); + const connectionIdsResolving = getAllowedConnectionIds(func.absolutePath); + const allowedConnectionIds = await withTimeout( + connectionIdsResolving, + request.timeoutMs, + `Resolving allowed connections for "${displayName}"`, + ); + const actionCatalogInstalled = isActionCatalogInstalled(projectRoot); + const appsBackendInstalled = isDatadogAppsBackendInstalled(projectRoot); + callCount += 1; + return runChild({ + func, + allowedConnectionIds, + executeAction, + timeoutMs: request.timeoutMs, + longPolling: request.longPolling, + prefix: `[${func.name}#${callCount}]`, + config: { + moduleRunnerUrl, + projectRoot, + entryUrl: func.absolutePath + LOCAL_EXECUTION_LOAD_SUFFIX, + entryPath: func.absolutePath, + functionName: func.name, + args, + runtimeContext, + actionCatalogInstalled, + appsBackendInstalled, + flushTimeoutMs, + }, }); + }; const close = () => { closed = true; killChildren(ownChildren); diff --git a/packages/plugins/apps/src/vite/local-execution-protocol.ts b/packages/plugins/apps/src/vite/local-execution-protocol.ts index c9d780e58..aabf6690a 100644 --- a/packages/plugins/apps/src/vite/local-execution-protocol.ts +++ b/packages/plugins/apps/src/vite/local-execution-protocol.ts @@ -23,7 +23,7 @@ export type ParentMessage = | { type: 'start'; config: ChildStartConfig } | { type: 'invoke-result'; id: number; response: unknown } | { type: 'action-result'; id: number; ok: true; value: unknown } - | { type: 'action-result'; id: number; ok: false; message: string }; + | { type: 'action-result'; id: number; ok: false; message: string; thrownString: boolean }; export type ChildMessage = | { type: 'invoke'; id: number; payload: unknown } @@ -37,7 +37,7 @@ export type ChildMessage = } | { type: 'loaded' } | { type: 'result'; data: unknown } - | { type: 'error'; message: string }; + | { type: 'error'; message: string; detail?: string }; // The child trusts the dev server, so this only tells the message kinds apart. export function isParentMessage(message: unknown): message is ParentMessage { diff --git a/packages/plugins/apps/src/vite/local-execution-shared.ts b/packages/plugins/apps/src/vite/local-execution-shared.ts index 84a0a09af..b815f1432 100644 --- a/packages/plugins/apps/src/vite/local-execution-shared.ts +++ b/packages/plugins/apps/src/vite/local-execution-shared.ts @@ -23,6 +23,43 @@ export function isIndexableRecord(value: unknown): value is Record unknown): Promise { + try { + return await call(); + } catch (thrown) { + const { message, detail } = describeThrown(thrown); + throw new FunctionThrewError(message, detail); + } +} + /** Checks this one call's own connectionId field against the function's allowlist — not full parity with the cloud path's server-side restriction, since a nested meta-action's own inputs can carry a separate scope this never inspects (see validateActionCall's own check for that). */ export function assertConnectionIdAllowed( connectionId: string | undefined, @@ -30,7 +67,7 @@ export function assertConnectionIdAllowed( actionDescription: string, ): void { if (connectionId !== undefined && !allowedConnectionIds.includes(connectionId)) { - throw new Error( + throwActionFailure( `Action ${actionDescription} used connection "${connectionId}", which is not in this function's allowed connections: [${allowedConnectionIds.join(', ')}]`, ); } @@ -56,7 +93,9 @@ export function validateActionCallShape( ): { inputs: Record; connectionId: string | undefined } { const { inputs, connectionId } = call; if (typeof inputs !== 'object' || !inputs) { - throw new Error(`Action ${actionDescription} must have an inputs field`); + throwActionFailure( + `First argument to action ${actionDescription} must have an inputs field that is an object`, + ); } assertNoSelfDeclaredScope(inputs, actionDescription); return { inputs, connectionId }; @@ -103,7 +142,7 @@ export function makeActionsProxy(dispatch: ActionDispatch, pathParts: string[] = async apply(_target, _thisArg, args: unknown[]) { const actionPath = pathParts.join('.'); if (args.length === 0) { - throw new Error(`No arguments provided to action $.Actions.${actionPath}`); + throwActionFailure(`No arguments provided to action $.Actions.${actionPath}`); } const call: Partial = isIndexableRecord(args[0]) ? args[0] : {}; const fqn = `com.datadoghq.${actionPath}`; diff --git a/packages/plugins/apps/src/vite/local-execution.test.ts b/packages/plugins/apps/src/vite/local-execution.test.ts index b838eb8a9..4e0846265 100644 --- a/packages/plugins/apps/src/vite/local-execution.test.ts +++ b/packages/plugins/apps/src/vite/local-execution.test.ts @@ -435,7 +435,11 @@ describe('local-execution — executeScriptLocally', () => { }), mockLogger, ), - ).rejects.toThrow(/not in this function's allowed connections/); + ).rejects.toMatchObject({ + detail: expect.stringMatching( + /^Action .+ not in this function's allowed connections: \[.*\]$/, + ), + }); expect(executeAction).not.toHaveBeenCalled(); }); @@ -469,7 +473,11 @@ describe('local-execution — executeScriptLocally', () => { }), mockLogger, ), - ).rejects.toThrow(/must have an inputs field/); + ).rejects.toMatchObject({ + detail: expect.stringMatching( + /^First argument to action .+ must have an inputs field that is an object$/, + ), + }); }); // A call with no connectionId skips assertConnectionIdAllowed, so a scope nested in its inputs @@ -1533,7 +1541,11 @@ describe('local-execution — executeScriptLocally', () => { loadModule, mockLogger, ), - ).rejects.toThrow(/not in this function's allowed connections/); + ).rejects.toMatchObject({ + detail: expect.stringMatching( + /^Action .+ not in this function's allowed connections: \[.*\]$/, + ), + }); expect(executeAction).not.toHaveBeenCalled(); }); @@ -1576,7 +1588,11 @@ describe('local-execution — executeScriptLocally', () => { loadModule, mockLogger, ), - ).rejects.toThrow(/must have an inputs field/); + ).rejects.toMatchObject({ + detail: expect.stringMatching( + /^First argument to action .+ must have an inputs field that is an object$/, + ), + }); expect(executeAction).not.toHaveBeenCalled(); }); diff --git a/packages/plugins/apps/src/vite/local-execution.ts b/packages/plugins/apps/src/vite/local-execution.ts index aacbb3160..e19924849 100644 --- a/packages/plugins/apps/src/vite/local-execution.ts +++ b/packages/plugins/apps/src/vite/local-execution.ts @@ -20,6 +20,7 @@ import type { ActionCallArgs } from './local-execution-shared'; import { assertJsonRoundTrippable, assertNoSelfDeclaredScope, + callBackendFunction, isIndexableRecord, makeActionsProxy, returnValueSubject, @@ -246,7 +247,7 @@ async function invokeAction( /** Serializes local executions — a customer function deleting `globalThis.$` mid-flight would otherwise break `$` access for any other execution concurrently in progress (see `ensureDollarAccessorInstalled`). */ let queueTail: Promise = Promise.resolve(); -export function enqueue(run: () => Promise): Promise { +function enqueue(run: () => Promise): Promise { const result = queueTail.then(run); queueTail = result.then( () => undefined, @@ -697,7 +698,7 @@ async function runScriptLocally( // assertJsonSerializable runs inside runBlocked's callback, not after, since its toJSON()/getter calls must run while access is still blocked. const data = await runBlocked( async () => { - const result = await fn(...args); + const result = await callBackendFunction(() => fn(...args)); return assertJsonSerializable(result, func); }, { signal: blockedScopeController.signal }, diff --git a/packages/tests/src/_jest/fixtures/apps_backend_project/childActions.backend.ts b/packages/tests/src/_jest/fixtures/apps_backend_project/childActions.backend.ts index 8cfd1dc1c..01edbe553 100644 --- a/packages/tests/src/_jest/fixtures/apps_backend_project/childActions.backend.ts +++ b/packages/tests/src/_jest/fixtures/apps_backend_project/childActions.backend.ts @@ -9,6 +9,24 @@ export async function callHttp(connectionId?: string, delayMs = 0) { return $.Actions.http.request({ inputs: { url: 'https://example.com' }, connectionId }); } +export async function catchHttpFailure(connectionId?: string, withInputs = true) { + try { + await $.Actions.http.request(withInputs ? { inputs: {}, connectionId } : { connectionId }); + return 'no error'; + } catch (error) { + return error instanceof Error ? { type: 'Error', value: error.message } : { type: typeof error, value: error }; + } +} + +export async function callHttpReportingPid() { + return { pid: process.pid, result: await $.Actions.http.request({ inputs: {} }) }; +} + +export async function throwValue(kind: 'error' | 'string') { + const failure = kind === 'error' ? new Error('plain error x') : 'plain string failure'; + throw failure; +} + export async function callHttpWithMap() { return $.Actions.http.request({ inputs: { headers: new Map([['a', 'b']]) } }); }