diff --git a/packages/plugins/apps/README.md b/packages/plugins/apps/README.md index 164da2849..89e080a99 100644 --- a/packages/plugins/apps/README.md +++ b/packages/plugins/apps/README.md @@ -97,14 +97,15 @@ When Datadog rejects a bundle as longer than the `jsFunctionWithActions` `script ## Local execution -`/__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. +`/__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. A call that starts while another call's runtime context (`$`) query is in flight shares that query instead of sending its own. - **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`, so a project inside `/tmp` is writable too. A refused read or write throws `ERR_ACCESS_DENIED`, not `ENOENT`, so code that tolerates only a missing file fails locally. 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. On a Node without [`module.registerHooks`](https://nodejs.org/api/module.html#moduleregisterhooksoptions), added in Node 22.15, modules get no `import.meta.resolve`, which Vite resolves on a worker thread there, and the dev server warns once. 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`), plus `SystemRoot` on Windows, and no Node options from `NODE_OPTIONS`. Vite still inlines `import.meta.env` and `define` values into the code it serves. `$.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. Local execution doesn't support Yarn Plug'n'Play, whose packages load through hooks a call's process doesn't get: each call fails with a hint to set `nodeLinker: node-modules` or use `dev:verify`. 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 88309ca6f..d36ad9845 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 @@ -69,6 +69,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 31aee417e..7ea5b5948 100644 --- a/packages/plugins/apps/src/vite/dev-server.test.ts +++ b/packages/plugins/apps/src/vite/dev-server.test.ts @@ -18,7 +18,8 @@ import type { LocalExecutor, } from '@dd/apps-plugin/vite/local-execution-process'; import { createChildProcessExecutor } from '@dd/apps-plugin/vite/local-execution-process'; -import { DEFAULT_TIMEOUT_MS } from '@dd/apps-plugin/vite/local-execution'; +import { FunctionThrewError } from '@dd/apps-plugin/vite/local-execution-shared'; +import { DEFAULT_TIMEOUT_MS, withTimeout } from '@dd/apps-plugin/vite/local-execution'; import { rmSync } from '@dd/core/helpers/fs'; import type { AuthOptionsWithDefaults, RequestOpts } from '@dd/core/types'; import { @@ -229,6 +230,14 @@ function postMessageExecutor(connectionId: string | undefined): LocalExecutor { }; } +function createDeferred() { + let resolve: () => void = () => {}; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + const previewRuntimeContext = { Source: { initiator: { id: 'preview-initiator', orgId: 'preview-org' }, @@ -1548,6 +1557,67 @@ describe('Dev Server Middleware', () => { ); }); + 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 reject a malformed hydrated identity before starting a call's process without exposing the response body", async () => { nock.cleanAll(); mockRuntimeContextHydration(1, { @@ -1607,9 +1677,13 @@ describe('Dev Server Middleware', () => { nock.cleanAll(); const executeLocally = createTestExecutor(); let hydrationSignal: AbortSignal | undefined; + const recoveringRuntimeContextRequest = makeImmediateRuntimeContextRequest(); const nonAbortableRuntimeContextRequest = ( options: Omit, ) => { + if (hydrationSignal) { + return recoveringRuntimeContextRequest(options) as Promise; + } hydrationSignal = options.signal; // Models doRequest sleeping in async-retry backoff: the signal is aborted, // but the request promise does not settle until the retry timer wakes up. @@ -1618,7 +1692,9 @@ describe('Dev Server Middleware', () => { const hangingHydrationMiddleware = createDevServerMiddleware( mockViteBuild, () => mockFunctions, - async () => [], + async () => { + throw new Error('reached the connection lookup'); + }, mockAuth, nonAbortableRuntimeContextRequest, mockLongPolling, @@ -1648,31 +1724,15 @@ describe('Dev Server Middleware', () => { expect(hydrationSignal?.aborted).toBe(true); expect(forkMock).not.toHaveBeenCalled(); - const recoveringRuntimeContextRequest = makeImmediateRuntimeContextRequest(); - // Reaching the connection lookup shows the stalled call no longer holds the queue. - const recoveringMiddleware = createDevServerMiddleware( - mockViteBuild, - () => mockFunctions, - async () => { - throw new Error('reached the connection lookup'); - }, - mockAuth, - recoveringRuntimeContextRequest, - mockLongPolling, - '/project', - mockLog, - 'development', - localOnlyHostCheck, - defaultServerConditions, - executeLocally, - ); + // The same dev server reaching the connection lookup shows the timed-out query no + // longer holds the shared runtime context. const recoveringReq = createMockRequest('/__dd/executeAction', { functionName: encodeQueryName(mockFunctions[0]), args: [], }); const recoveringRes = createMockResponse(); - recoveringMiddleware(recoveringReq, recoveringRes, jest.fn()); + hangingHydrationMiddleware(recoveringReq, recoveringRes, jest.fn()); const recoveringDone = recoveringRes.done; await jest.advanceTimersByTimeAsync(0); await recoveringDone; @@ -1894,6 +1954,109 @@ describe('Dev Server Middleware', () => { ); }); + describe('What a failed $.Actions call rejects with, as in a deployed app', () => { + const catchingPostMessage: LocalExecutor = async ({ executeAction }) => { + try { + await executeAction( + 'com.datadoghq.slack.chat.postMessage', + { text: 'hi' }, + undefined, + ); + return { data: 'no error' }; + } catch (error) { + return { data: { type: typeof error, value: error } }; + } + }; + const runAndReadData = async () => { + const actionMiddleware = createTestMiddleware({ + executeLocally: catchingPostMessage, + }); + const req = createMockRequest('/__dd/executeAction', { + functionName: encodeQueryName(mockFunctions[0]), + args: [], + }); + const res = createMockResponse(); + actionMiddleware(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 () => { + 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 () => { + mockActionErrors([{ detail: 'first' }, { title: 'second' }]); + + await expect(runAndReadData()).resolves.toEqual({ + type: 'string', + value: 'first; second', + }); + }); + + test('Should reject with a fallback message when no platform error has a detail or a title', async () => { + mockActionErrors([{}, { detail: '' }]); + + await expect(runAndReadData()).resolves.toEqual({ + type: 'string', + value: 'Query execution failed', + }); + expect(mockLogFn).toHaveBeenCalledWith( + '$.Actions call to "com.datadoghq.slack.chat.postMessage" failed: Query execution failed', + 'error', + ); + }); + }); + + describe('What an uncaught failure reports, as in a deployed app', () => { + const runAndReadError = async (failure: unknown) => { + const failingMiddleware = createTestMiddleware({ + executeLocally: jest.fn().mockRejectedValue(failure), + }); + const req = createMockRequest('/__dd/executeAction', { + functionName: encodeQueryName(mockFunctions[0]), + args: [], + }); + const res = createMockResponse(); + failingMiddleware(req, res, jest.fn()); + await res.done; + return { statusCode: res.statusCode, error: JSON.parse(res.getBody()).error }; + }; + + test('Should prefix what the function threw with "Execution failed: "', async () => { + const thrown = new FunctionThrewError('plain error x', 'TypeError: plain error x'); + + await expect(runAndReadError(thrown)).resolves.toEqual({ + statusCode: 500, + error: 'Execution failed: TypeError: plain error x', + }); + }); + + test("Should leave the dev server's own failures unprefixed", async () => { + await expect( + runAndReadError(new Error('"getGreeting" is not a function')), + ).resolves.toEqual({ + statusCode: 500, + error: '"getGreeting" is not a function', + }); + }); + }); + // getAllowedConnectionIds only resolves and transforms modules, so it rejects a banned // import before any of the entry's code runs. test("Should never start a call's process when getAllowedConnectionIds rejects", async () => { diff --git a/packages/plugins/apps/src/vite/dev-server.ts b/packages/plugins/apps/src/vite/dev-server.ts index a3a1dab38..c4930928e 100644 --- a/packages/plugins/apps/src/vite/dev-server.ts +++ b/packages/plugins/apps/src/vite/dev-server.ts @@ -30,6 +30,7 @@ import { getBaseBackendBuildConfig, withProductionSettings } from './build-confi 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 } from './local-execution'; import { DEFAULT_TIMEOUT_MS } from './local-execution'; import { getMaxRetryDelayMs } from './retry-delay'; @@ -328,6 +329,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 @@ -353,9 +365,8 @@ 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 if doRequest is already sleeping inside + // async-retry, whose backoff timer does not observe AbortSignal. const outputs = await Promise.race([ submitQuery( { @@ -429,11 +440,23 @@ 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 { + readonly detail: string; + constructor(details: string) { + super(details ? `Query execution failed: ${details}` : 'Query execution failed'); + this.detail = details || this.message; + } +} + interface PollResult { data?: { attributes?: { done?: boolean; outputs?: unknown } }; errors?: Array<{ detail?: string; title?: string }>; @@ -491,8 +514,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); } const attrs = result.data?.attributes; @@ -554,7 +580,12 @@ 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}` + : error instanceof Error + ? error.message + : 'Internal server error'; if (log && label) { log.debug(`Error handling ${label}: ${message}`); } @@ -671,6 +702,7 @@ async function handleExecuteAction( getAllowedConnectionIds: (entryId: string) => Promise, log: Logger, executeLocally: LocalExecutor | undefined, + shareRuntimeContext: (run: GetRuntimeContext) => Promise, ): Promise { try { if (!executeLocally) { @@ -687,12 +719,13 @@ async function handleExecuteAction( longPolling, log, ); - const getRuntimeContext = makeGetRuntimeContextRemotely( + const getRuntimeContextRemotely = makeGetRuntimeContextRemotely( auth, doAuthenticatedRequest, longPolling, log, ); + const getRuntimeContext = () => shareRuntimeContext(getRuntimeContextRemotely); const result = await executeLocally({ func, args, @@ -828,6 +861,8 @@ export function createDevServerMiddleware( log, oversizedFunctionsWarned, ); + // 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( @@ -840,6 +875,7 @@ export function createDevServerMiddleware( getAllowedConnectionIds, 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 9cefe4baf..1115824aa 100644 --- a/packages/plugins/apps/src/vite/local-execution-child.ts +++ b/packages/plugins/apps/src/vite/local-execution-child.ts @@ -16,7 +16,9 @@ import type { ActionCallArgs } from './local-execution-shared'; import { assertJsonRoundTrippable, assertNoSelfDeclaredScope, + callBackendFunction, describeError, + FunctionThrewError, isIndexableRecord, makeActionsProxy, returnValueSubject, @@ -38,7 +40,7 @@ type ModuleRunnerModule = { createNodeImportMeta: CreateImportMeta; }; -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; @@ -107,8 +109,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); } } } @@ -193,7 +195,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); } @@ -216,6 +218,7 @@ async function run(config: ChildStartConfig): Promise { outcome = { type: 'error', message: describeError(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 3c19b16db..418763b4b 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 @@ -69,19 +69,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 { @@ -222,37 +217,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, ); @@ -1314,6 +1326,126 @@ export class ModuleRunner { 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 () => { @@ -1326,6 +1458,7 @@ export class ModuleRunner { 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([]); }, @@ -1344,6 +1477,7 @@ export class ModuleRunner { id: 999_999, ok: false, message: expect.stringContaining('Malformed $.Actions request'), + thrownString: false, }); expect(calls).toEqual([]); }, @@ -1610,7 +1744,7 @@ export class ModuleRunner { 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({ @@ -1625,13 +1759,48 @@ export class ModuleRunner { 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); }, CALL_TIMEOUT_MS, ); + + test( + 'Should give each of several parallel calls its own prefix on every line', + async () => { + const lines: Array<{ stream: OutputStream; line: string }> = []; + const execute = makeExecutor({ + writeOutput: (stream, line) => lines.push({ stream, line }), + }); + const logLines = fixtureFunction('childState', 'logLines'); + const results = await Promise.all( + [1, 2].map(() => execute(makeRequest(logLines, [300]))), + ); + + expect(results.map((result) => result.data)).toEqual(['logged', 'logged']); + const linesByPrefix = new Map< + string, + Array<{ stream: OutputStream; line: string }> + >(); + for (const { stream, line } of lines) { + const [prefix, ...rest] = line.split(' '); + const own = linesByPrefix.get(prefix) ?? []; + own.push({ stream, line: rest.join(' ') }); + linesByPrefix.set(prefix, own); + } + expect([...linesByPrefix.keys()].sort()).toEqual(['[logLines#1]', '[logLines#2]']); + for (const own of linesByPrefix.values()) { + expect(own).toEqual([ + { stream: 'stdout', line: 'first line' }, + { stream: 'stderr', line: 'second line' }, + { stream: 'stdout', line: 'no trailing newline' }, + ]); + } + }, + CALL_TIMEOUT_MS, + ); }); }); diff --git a/packages/plugins/apps/src/vite/local-execution-process.ts b/packages/plugins/apps/src/vite/local-execution-process.ts index f2e54704e..408dd4802 100644 --- a/packages/plugins/apps/src/vite/local-execution-process.ts +++ b/packages/plugins/apps/src/vite/local-execution-process.ts @@ -30,12 +30,16 @@ import { sharedTmpPaths, withRealPath, } from './local-execution-sandbox'; -import { describeError, isIndexableRecord, validateActionCall } from './local-execution-shared'; +import { + describeError, + FunctionThrewError, + isIndexableRecord, + validateActionCall, +} from './local-execution-shared'; import type { ExecuteAction, GetRuntimeContext } from './local-execution'; import { assertValidRuntimeContext, deriveActionTimeouts, - enqueue, localExecutionResolutionContext, withTimeout, } from './local-execution'; @@ -251,6 +255,7 @@ interface ChildRun { executeAction: ExecuteAction; timeoutMs: number; longPolling: Required; + prefix: string; } export function createChildProcessExecutor( @@ -291,6 +296,7 @@ export function createChildProcessExecutor( const [realChildEntryPath = childEntryPath] = withRealPath(childEntryPath).slice(-1); const [realModuleRunnerPath = moduleRunnerPath] = withRealPath(moduleRunnerPath).slice(-1); const moduleRunnerUrl = pathToFileURL(realModuleRunnerPath).href; + let callCount = 0; const ownChildren: ChildTmpDirs = new Map(); let closed = false; @@ -304,7 +310,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( @@ -358,7 +364,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'); @@ -483,10 +489,17 @@ export function createChildProcessExecutor( }, ) => { const replyWithError = (error: unknown) => { + const thrownString = typeof error === 'string'; const errorMessage = describeError(error); const failCall = () => fail(errorMessage); reply( - { type: 'action-result', id, ok: false, message: errorMessage }, + { + type: 'action-result', + id, + ok: false, + message: errorMessage, + thrownString, + }, failCall, ); }; @@ -556,6 +569,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) { @@ -572,7 +586,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'. @@ -602,41 +620,41 @@ export function createChildProcessExecutor( }); }; - 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, - 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, + 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 4ba0e3e6d..4689b3283 100644 --- a/packages/plugins/apps/src/vite/local-execution-protocol.ts +++ b/packages/plugins/apps/src/vite/local-execution-protocol.ts @@ -22,7 +22,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 } @@ -36,7 +36,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.test.ts b/packages/plugins/apps/src/vite/local-execution-shared.test.ts index e60f8c2d0..eb0e9e3b3 100644 --- a/packages/plugins/apps/src/vite/local-execution-shared.test.ts +++ b/packages/plugins/apps/src/vite/local-execution-shared.test.ts @@ -6,23 +6,24 @@ import type { ActionCallArgs } from '@dd/apps-plugin/vite/local-execution-shared import { validateActionCallShape } from '@dd/apps-plugin/vite/local-execution-shared'; const actionDescription = '$.Actions.http.request'; +const missingInputs = `First argument to action ${actionDescription} must have an inputs field that is an object`; describe('local-execution-shared — validateActionCallShape', () => { const rejected: Array<{ description: string; call: Partial; error: string }> = [ { description: 'reject a call with no inputs', call: {}, - error: `Action ${actionDescription} must have an inputs field`, + error: missingInputs, }, { description: 'reject null inputs', call: { inputs: null as unknown as Record }, - error: `Action ${actionDescription} must have an inputs field`, + error: missingInputs, }, { description: 'reject non-object inputs', call: { inputs: 'url' as unknown as Record }, - error: `Action ${actionDescription} must have an inputs field`, + error: missingInputs, }, { description: 'reject inputs that declare their own allowedConnectionIds', diff --git a/packages/plugins/apps/src/vite/local-execution-shared.ts b/packages/plugins/apps/src/vite/local-execution-shared.ts index 9ba160929..5f6b192a7 100644 --- a/packages/plugins/apps/src/vite/local-execution-shared.ts +++ b/packages/plugins/apps/src/vite/local-execution-shared.ts @@ -32,6 +32,44 @@ 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 only this call's own connectionId, so it's weaker than the cloud's server-side restriction: a meta-action's inputs can carry their own scope. */ export function assertConnectionIdAllowed( connectionId: string | undefined, @@ -39,7 +77,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(', ')}]`, ); } @@ -65,7 +103,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 }; @@ -112,7 +152,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.ts b/packages/plugins/apps/src/vite/local-execution.ts index 467b12199..b54426217 100644 --- a/packages/plugins/apps/src/vite/local-execution.ts +++ b/packages/plugins/apps/src/vite/local-execution.ts @@ -89,7 +89,7 @@ export function deriveActionTimeouts(longPolling: LongPollingConfig): { longPolling.maxRetries * longPolling.timeoutMs + getTotalRetryDelayBudgetMs(longPolling); return { // Bounds a single $.Actions call, exempt from the hang-detection timer — doRequest has - // no deadline of its own, so an unsettled call would wedge the whole serialized queue. + // no deadline of its own, so an unsettled call would otherwise never end. actionCallTimeoutMs: worstCaseMs * 2, // Absolute wall-clock ceiling, independent of the pause-and-extend mechanism above — // that can't tell a genuinely slow $.Actions call from a fire-and-forgot one masking a @@ -108,18 +108,6 @@ export type ExecuteAction = ( /** Resolves a fresh serializable `$` snapshot from an authenticated preview execution. */ export type GetRuntimeContext = () => Promise; -/** Runs local executions one at a time, in arrival order. */ -let queueTail: Promise = Promise.resolve(); - -export function enqueue(run: () => Promise): Promise { - const result = queueTail.then(run); - queueTail = result.then( - () => undefined, - () => undefined, - ); - return result; -} - /** Bounds a promise that could otherwise hang forever, such as a `$.Actions` call with no deadline of its own, without cancelling it. `label` is the full subject of the timeout message. */ export function withTimeout(promise: Promise, timeoutMs: number, label: string): Promise { return new Promise((resolve, reject) => { 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 67bea56db..b22c760c6 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 @@ -14,6 +14,24 @@ export async function callHttpWithoutAwaiting() { return 'returned'; } +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']]) } }); }