diff --git a/packages/adt-server-client/scripts/generate.ts b/packages/adt-server-client/scripts/generate.ts index 485f0ff69..5b0f9138a 100644 --- a/packages/adt-server-client/scripts/generate.ts +++ b/packages/adt-server-client/scripts/generate.ts @@ -227,11 +227,11 @@ export async function renderOpenApiClient(document: unknown): Promise { .map((operation) => { const name = typeName(operation.id); return operation.hasInput - ? `${key(operation.id)}: (params: ${name}Params) => request<${name}Response>(operationDefinitions.${operation.id}, params),` - : `${key(operation.id)}: () => request<${name}Response>(operationDefinitions.${operation.id}),`; + ? `${key(operation.id)}: (params: ${name}Params, requestOptions?: AdtServerRequestOptions) => request<${name}Response>(operationDefinitions.${operation.id}, params, requestOptions),` + : `${key(operation.id)}: (requestOptions?: AdtServerRequestOptions) => request<${name}Response>(operationDefinitions.${operation.id}, undefined, requestOptions),`; }) .join('\n'); - const source = `/**\n * GENERATED FROM @abapify/adt-server's OpenAPI document.\n * Run \`bun run --filter @abapify/adt-server-client generate\`; do not edit manually.\n */\n\nexport interface AdtServerClientOptions {\n baseUrl: string;\n fetch?: typeof globalThis.fetch;\n headers?: Record;\n}\n\nexport class AdtServerHttpError extends Error {\n constructor(\n readonly status: number,\n readonly body: unknown,\n ) {\n super(\`ADT Server request failed (\${status})\`);\n }\n}\n\ntype OperationDefinition = {\n method: 'GET' | 'POST' | 'PUT' | 'PATCH' | 'DELETE';\n path: string;\n pathParameters: readonly string[];\n queryParameters: readonly string[];\n hasBody: boolean;\n};\n\nconst operationDefinitions = {\n${definitionSource}\n} as const satisfies Record;\n\n${typeSource}\n\nfunction inputValue(input: object | undefined, name: string): unknown {\n return (input as Record | undefined)?.[name];\n}\n\nexport function createAdtServerClient(options: AdtServerClientOptions) {\n const fetcher = options.fetch ?? globalThis.fetch;\n const request = async (\n definition: OperationDefinition,\n input?: object,\n ): Promise => {\n const path = definition.path.replace(/\\{([A-Za-z_$][A-Za-z0-9_$]*)\\}/gu, (_match, name: string) => {\n const value = inputValue(input, name);\n if (value === undefined || value === null) {\n throw new Error(\`Missing required path parameter: \${name}\`);\n }\n return encodeURIComponent(String(value)); // NOSONAR\n });\n const url = new URL(path, options.baseUrl);\n for (const name of definition.queryParameters) {\n const value = inputValue(input, name);\n if (value !== undefined && value !== null) url.searchParams.set(name, String(value)); // NOSONAR\n }\n const body = inputValue(input, 'body');\n const response = await fetcher(url, {\n method: definition.method,\n headers: {\n accept: 'application/json',\n ...(definition.hasBody ? { 'content-type': 'application/json' } : {}),\n ...options.headers,\n },\n ...(definition.hasBody ? { body: JSON.stringify(body) } : {}),\n });\n const responseBody = await response.json().catch(() => undefined);\n if (!response.ok) {\n throw new AdtServerHttpError(response.status, responseBody);\n }\n return responseBody as T;\n };\n\n return {\n${methodSource}\n };\n}\n\nexport type AdtServerClient = ReturnType;\n`; + const source = `/**\n * GENERATED FROM @abapify/adt-server's OpenAPI document.\n * Run \`bun run --filter @abapify/adt-server-client generate\`; do not edit manually.\n */\n\nexport interface AdtServerClientOptions {\n baseUrl: string;\n fetch?: typeof globalThis.fetch;\n headers?: Record;\n}\n\nexport interface AdtServerRequestOptions {\n signal?: AbortSignal;\n}\n\nexport class AdtServerHttpError extends Error {\n constructor(\n readonly status: number,\n readonly body: unknown,\n ) {\n super(\`ADT Server request failed (\${status})\`);\n }\n}\n\ntype OperationDefinition = {\n method: 'GET' | 'POST' | 'PUT' | 'PATCH' | 'DELETE';\n path: string;\n pathParameters: readonly string[];\n queryParameters: readonly string[];\n hasBody: boolean;\n};\n\nconst operationDefinitions = {\n${definitionSource}\n} as const satisfies Record;\n\n${typeSource}\n\nfunction inputValue(input: object | undefined, name: string): unknown {\n return (input as Record | undefined)?.[name];\n}\n\nexport function createAdtServerClient(options: AdtServerClientOptions) {\n const fetcher = options.fetch ?? globalThis.fetch;\n const request = async (\n definition: OperationDefinition,\n input?: object,\n requestOptions?: AdtServerRequestOptions,\n ): Promise => {\n const path = definition.path.replace(/\\{([A-Za-z_$][A-Za-z0-9_$]*)\\}/gu, (_match, name: string) => {\n const value = inputValue(input, name);\n if (value === undefined || value === null) {\n throw new Error(\`Missing required path parameter: \${name}\`);\n }\n return encodeURIComponent(String(value)); // NOSONAR\n });\n const url = new URL(path, options.baseUrl);\n for (const name of definition.queryParameters) {\n const value = inputValue(input, name);\n if (value !== undefined && value !== null) url.searchParams.set(name, String(value)); // NOSONAR\n }\n const body = inputValue(input, 'body');\n const response = await fetcher(url, {\n method: definition.method,\n headers: {\n accept: 'application/json',\n ...(definition.hasBody ? { 'content-type': 'application/json' } : {}),\n ...options.headers,\n },\n ...(definition.hasBody ? { body: JSON.stringify(body) } : {}),\n signal: requestOptions?.signal,\n });\n const responseBody = await response.json().catch(() => undefined);\n if (!response.ok) {\n throw new AdtServerHttpError(response.status, responseBody);\n }\n return responseBody as T;\n };\n\n return {\n${methodSource}\n };\n}\n\nexport type AdtServerClient = ReturnType;\n`; return await format(source, { ...(await resolveConfig(generatedFile)), filepath: generatedFile, diff --git a/packages/adt-server-client/src/generated.ts b/packages/adt-server-client/src/generated.ts index 68b176f6b..9de13dff2 100644 --- a/packages/adt-server-client/src/generated.ts +++ b/packages/adt-server-client/src/generated.ts @@ -9,6 +9,10 @@ export interface AdtServerClientOptions { headers?: Record; } +export interface AdtServerRequestOptions { + signal?: AbortSignal; +} + export class AdtServerHttpError extends Error { constructor( readonly status: number, @@ -34,6 +38,13 @@ const operationDefinitions = { queryParameters: [], hasBody: true, }, + getBadi: { + method: 'GET', + path: '/v1/destinations/{destination}/badi/{name}', + pathParameters: ['destination', 'name'], + queryParameters: ['implementations'], + hasBody: false, + }, getObjectMetadata: { method: 'GET', path: '/v1/destinations/{destination}/objects/{type}/{name}', @@ -195,6 +206,37 @@ export type BuildTransportSourceManifestResponse = { diagnostic?: { code: string; message: string }; }>; }; +export type GetBadiParams = { + destination: string; + name: string; + implementations?: boolean; +}; +export type GetBadiResponse = { + kind: 'definition' | 'implementation' | 'enhancement'; + name: string; + type: string; + description?: string; + language?: string; + version?: string; + packageName?: string; + packageUri?: string; + responsible?: string; + masterLanguage?: string; + masterSystem?: string; + implementations?: Array<{ + kind: 'implementation'; + name: string; + type: string; + description?: string; + language?: string; + version?: string; + packageName?: string; + packageUri?: string; + responsible?: string; + masterLanguage?: string; + masterSystem?: string; + }>; +}; export type GetObjectMetadataParams = { destination: string; type: string; @@ -489,6 +531,7 @@ export function createAdtServerClient(options: AdtServerClientOptions) { const request = async ( definition: OperationDefinition, input?: object, + requestOptions?: AdtServerRequestOptions, ): Promise => { const path = definition.path.replace( /\{([A-Za-z_$][A-Za-z0-9_$]*)\}/gu, @@ -515,6 +558,7 @@ export function createAdtServerClient(options: AdtServerClientOptions) { ...options.headers, }, ...(definition.hasBody ? { body: JSON.stringify(body) } : {}), + signal: requestOptions?.signal, }); const responseBody = await response.json().catch(() => undefined); if (!response.ok) { @@ -526,74 +570,141 @@ export function createAdtServerClient(options: AdtServerClientOptions) { return { buildTransportSourceManifest: ( params: BuildTransportSourceManifestParams, + requestOptions?: AdtServerRequestOptions, ) => request( operationDefinitions.buildTransportSourceManifest, params, + requestOptions, + ), + getBadi: ( + params: GetBadiParams, + requestOptions?: AdtServerRequestOptions, + ) => + request( + operationDefinitions.getBadi, + params, + requestOptions, ), - getObjectMetadata: (params: GetObjectMetadataParams) => + getObjectMetadata: ( + params: GetObjectMetadataParams, + requestOptions?: AdtServerRequestOptions, + ) => request( operationDefinitions.getObjectMetadata, params, + requestOptions, ), - getObjectSourceHistory: (params: GetObjectSourceHistoryParams) => + getObjectSourceHistory: ( + params: GetObjectSourceHistoryParams, + requestOptions?: AdtServerRequestOptions, + ) => request( operationDefinitions.getObjectSourceHistory, params, + requestOptions, ), - getPackageTree: (params: GetPackageTreeParams) => + getPackageTree: ( + params: GetPackageTreeParams, + requestOptions?: AdtServerRequestOptions, + ) => request( operationDefinitions.getPackageTree, params, + requestOptions, ), - getTransportDetail: (params: GetTransportDetailParams) => + getTransportDetail: ( + params: GetTransportDetailParams, + requestOptions?: AdtServerRequestOptions, + ) => request( operationDefinitions.getTransportDetail, params, + requestOptions, + ), + listDestinations: (requestOptions?: AdtServerRequestOptions) => + request( + operationDefinitions.listDestinations, + undefined, + requestOptions, ), - listDestinations: () => - request(operationDefinitions.listDestinations), - listPackageObjects: (params: ListPackageObjectsParams) => + listPackageObjects: ( + params: ListPackageObjectsParams, + requestOptions?: AdtServerRequestOptions, + ) => request( operationDefinitions.listPackageObjects, params, + requestOptions, ), - listTransportObjects: (params: ListTransportObjectsParams) => + listTransportObjects: ( + params: ListTransportObjectsParams, + requestOptions?: AdtServerRequestOptions, + ) => request( operationDefinitions.listTransportObjects, params, + requestOptions, ), - listTransports: (params: ListTransportsParams) => + listTransports: ( + params: ListTransportsParams, + requestOptions?: AdtServerRequestOptions, + ) => request( operationDefinitions.listTransports, params, + requestOptions, ), - readAtcFindingDocumentation: (params: ReadAtcFindingDocumentationParams) => + readAtcFindingDocumentation: ( + params: ReadAtcFindingDocumentationParams, + requestOptions?: AdtServerRequestOptions, + ) => request( operationDefinitions.readAtcFindingDocumentation, params, + requestOptions, ), - readObjectSource: (params: ReadObjectSourceParams) => + readObjectSource: ( + params: ReadObjectSourceParams, + requestOptions?: AdtServerRequestOptions, + ) => request( operationDefinitions.readObjectSource, params, + requestOptions, ), - readSourceVersion: (params: ReadSourceVersionParams) => + readSourceVersion: ( + params: ReadSourceVersionParams, + requestOptions?: AdtServerRequestOptions, + ) => request( operationDefinitions.readSourceVersion, params, + requestOptions, ), - runAtc: (params: RunAtcParams) => - request(operationDefinitions.runAtc, params), - searchObjects: (params: SearchObjectsParams) => + runAtc: (params: RunAtcParams, requestOptions?: AdtServerRequestOptions) => + request( + operationDefinitions.runAtc, + params, + requestOptions, + ), + searchObjects: ( + params: SearchObjectsParams, + requestOptions?: AdtServerRequestOptions, + ) => request( operationDefinitions.searchObjects, params, + requestOptions, ), - searchPackages: (params: SearchPackagesParams) => + searchPackages: ( + params: SearchPackagesParams, + requestOptions?: AdtServerRequestOptions, + ) => request( operationDefinitions.searchPackages, params, + requestOptions, ), }; } diff --git a/packages/adt-server-client/tests/client.test.ts b/packages/adt-server-client/tests/client.test.ts index f212a02ab..08ae77a00 100644 --- a/packages/adt-server-client/tests/client.test.ts +++ b/packages/adt-server-client/tests/client.test.ts @@ -71,3 +71,56 @@ test('serializes generated path, query and JSON body inputs', async () => { JSON.stringify({ sourceCapability: 'opaque-capability', maxBytes: 64 }), ); }); + +test('passes isolated optional abort signals to parameterized and parameterless calls', async () => { + const requests: RequestInit[] = []; + const parameterizedController = new AbortController(); + const parameterlessController = new AbortController(); + const client = createAdtServerClient({ + baseUrl: 'http://adt-server.test', + fetch: async (_input, init) => { + requests.push(init ?? {}); + return new Response(JSON.stringify({ data: [], truncated: false }), { + status: 200, + headers: { 'content-type': 'application/json' }, + }); + }, + }); + + await client.getPackageTree( + { destination: 'dev', root: 'ZROOT' }, + { signal: parameterizedController.signal }, + ); + await client.listDestinations({ signal: parameterlessController.signal }); + + assert.strictEqual(requests[0]?.signal, parameterizedController.signal); + assert.strictEqual(requests[1]?.signal, parameterlessController.signal); + assert.notStrictEqual(requests[0]?.signal, requests[1]?.signal); +}); + +test('aborts a generated request with its caller-provided signal', async () => { + const controller = new AbortController(); + let observedSignal: AbortSignal | undefined; + const client = createAdtServerClient({ + baseUrl: 'http://adt-server.test', + fetch: async (_input, init) => { + const signal = init?.signal; + if (!signal) { + throw new Error('Generated client did not forward a request signal'); + } + observedSignal = signal; + return await new Promise((_resolve, reject) => { + signal.addEventListener('abort', () => { + reject(signal.reason); + }); + }); + }, + }); + + const request = client.listDestinations({ signal: controller.signal }); + controller.abort(new Error('caller cancelled')); + + await assert.rejects(request, /caller cancelled/u); + assert.strictEqual(observedSignal, controller.signal); + assert.strictEqual(observedSignal?.aborted, true); +}); diff --git a/packages/adt-server/src/request-handler.ts b/packages/adt-server/src/request-handler.ts index 0c6638478..036082a99 100644 --- a/packages/adt-server/src/request-handler.ts +++ b/packages/adt-server/src/request-handler.ts @@ -1,6 +1,7 @@ import http from 'node:http'; import { AdtResponseTooLargeError, + runWithAdtAbortSignal, SourceVersionTooLargeError, } from '@abapify/adt-client'; import { z } from 'zod'; @@ -1000,7 +1001,40 @@ async function dispatchRoute( !(await ensureRestAuthorized(ctx, request, response)) ) return; - await route.handler(ctx, request, response, match); + if (!route.requiresRest) { + await route.handler(ctx, request, response, match); + return; + } + await runWithRequestAbortSignal(request, response, () => + route.handler(ctx, request, response, match), + ); +} + +async function runWithRequestAbortSignal( + request: http.IncomingMessage, + response: http.ServerResponse, + operation: () => Promise, +): Promise { + const abortController = new AbortController(); + const abort = () => { + if (!abortController.signal.aborted) abortController.abort(); + }; + const abortOnPrematureResponseClose = () => { + if (!response.writableEnded) abort(); + }; + + request.once('aborted', abort); + response.once('close', abortOnPrematureResponseClose); + if (request.aborted || (response.destroyed && !response.writableEnded)) { + abort(); + } + + try { + return await runWithAdtAbortSignal(abortController.signal, operation); + } finally { + request.removeListener('aborted', abort); + response.removeListener('close', abortOnPrematureResponseClose); + } } function handleServerError( diff --git a/packages/adt-server/tests/server.test.ts b/packages/adt-server/tests/server.test.ts index b36891528..0c90d242c 100644 --- a/packages/adt-server/tests/server.test.ts +++ b/packages/adt-server/tests/server.test.ts @@ -1,4 +1,5 @@ import assert from 'node:assert/strict'; +import http from 'node:http'; import test from 'node:test'; import { Client } from '@modelcontextprotocol/sdk/client/index.js'; import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js'; @@ -8,7 +9,10 @@ import { createMcpInvocationVerifier, } from '../../adt-mcp/src/index.ts'; import { generateKeyPair, SignJWT } from 'jose'; -import { SourceVersionTooLargeError } from '@abapify/adt-client'; +import { + createAdtAdapter, + SourceVersionTooLargeError, +} from '@abapify/adt-client'; import { createRestBearerAuthorizer } from '../src/rest-auth.js'; import { resolveMcpFrozenSource, startAdtServer } from '../src/server.js'; import { createRestSourceCapabilityService } from '../src/source-capabilities.js'; @@ -342,6 +346,121 @@ test('forwards validated transport search criteria only after REST authenticatio } }); +test('keeps downstream ADT work active after a REST response completes normally', async () => { + let observedSignal: AbortSignal | undefined; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async (_input, init) => { + observedSignal = init?.signal ?? undefined; + return new Response('{}', { + status: 200, + headers: { 'content-type': 'application/json' }, + }); + }) as typeof globalThis.fetch; + const adapter = createAdtAdapter({ + baseUrl: 'https://sap.example.test', + username: 'test', + password: 'test', + }); + const server = await startTestAdtServer({ + operations: { + ...operations, + async searchPackages() { + await adapter.request({ method: 'GET', url: '/packages' }); + return { data: [], truncated: false }; + }, + }, + host: '127.0.0.1', + port: 0, + restAuthorizer: { + async authorize() { + return true; + }, + }, + }); + + try { + const response = await originalFetch( + `${server.url}/v1/destinations/dev/packages?q=z`, + ); + assert.strictEqual(response.status, 200); + await new Promise((resolve) => setImmediate(resolve)); + assert.ok(observedSignal); + assert.strictEqual(observedSignal.aborted, false); + } finally { + await server.close(); + globalThis.fetch = originalFetch; + } +}); + +test( + 'aborts downstream ADT work when a REST client disconnects', + { timeout: 2_000 }, + async () => { + let observedSignal: AbortSignal | undefined; + let markOperationStarted: (() => void) | undefined; + let markOperationAborted: (() => void) | undefined; + const operationStarted = new Promise((resolve) => { + markOperationStarted = resolve; + }); + const operationAborted = new Promise((resolve) => { + markOperationAborted = resolve; + }); + const originalFetch = globalThis.fetch; + globalThis.fetch = ((_input, init) => { + observedSignal = init?.signal ?? undefined; + markOperationStarted?.(); + return new Promise((_resolve, reject) => { + observedSignal?.addEventListener( + 'abort', + () => { + markOperationAborted?.(); + reject(observedSignal?.reason); + }, + { once: true }, + ); + }); + }) as typeof globalThis.fetch; + const adapter = createAdtAdapter({ + baseUrl: 'https://sap.example.test', + username: 'test', + password: 'test', + }); + const server = await startTestAdtServer({ + operations: { + ...operations, + async searchPackages() { + await adapter.request({ method: 'GET', url: '/packages' }); + return { data: [], truncated: false }; + }, + }, + host: '127.0.0.1', + port: 0, + restAuthorizer: { + async authorize() { + return true; + }, + }, + }); + + try { + const request = http.request( + `${server.url}/v1/destinations/dev/packages?q=z`, + ); + request.on('error', () => undefined); + request.end(); + await operationStarted; + request.destroy(); + await operationAborted; + + assert.ok(observedSignal); + assert.strictEqual(observedSignal.aborted, true); + } finally { + await server.close(); + globalThis.fetch = originalFetch; + } + }, +); + test('serves canonical transport detail and aggregated objects without SAP URI fields', async () => { const calls: Array<{ operation: string;