Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 44 additions & 0 deletions server/itc/serverHandlers.js
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ const serverItcHandlers = {
[hdbTerms.ITC_EVENT_TYPES.SCHEMA]: schemaHandler,
[hdbTerms.ITC_EVENT_TYPES.USER]: userHandler,
[hdbTerms.ITC_EVENT_TYPES.COMPONENT_STATUS_REQUEST]: componentStatusRequestHandler,
[hdbTerms.ITC_EVENT_TYPES.RESOURCE_OPENAPI_REQUEST]: resourceOpenApiRequestHandler,
};

/**
Expand Down Expand Up @@ -155,4 +156,47 @@ async function componentStatusRequestHandler(event) {
}
}

/**
* Handles incoming requests for the REST OpenAPI spec from the main thread.
* Generates the spec from the local resources (which are only registered on worker threads)
* and sends it back to the requesting thread.
*/
async function resourceOpenApiRequestHandler(event) {
try {
const validate = validateEvent(event);
if (validate) {
hdbLogger.error(validate);
return;
}

hdbLogger.trace(`ITC resourceOpenApiRequestHandler received request:`, event);

const { resources } = require('../../resources/Resources.ts');
// Only respond if this thread has registered resources. Job-type workers with an empty
// resources map must stay silent so that an app worker with real resources replies first.
// If no worker has resources the main thread gets a 503 after the timeout, which is a
// more honest response than silently returning an empty spec.
if (!resources || resources.size === 0) return;
const { generateJsonApi } = require('../../resources/openApi.ts');
const openapi = generateJsonApi(resources, event.message.serverHttpURL);
Comment thread
kriszyp marked this conversation as resolved.

const originatorThreadId = event.message.originator;
const responseMessage = {
type: hdbTerms.ITC_EVENT_TYPES.RESOURCE_OPENAPI_RESPONSE,
message: {
requestId: event.message.requestId,
openapi,
},
};

if (!threads.sendToThread(originatorThreadId, responseMessage)) {
hdbLogger.trace(
`Dropping resource OpenAPI response for request ${event.message.requestId}: originator thread ${originatorThreadId} is unreachable`
);
}
} catch (error) {
hdbLogger.error('Error handling resource OpenAPI request:', error);
}
}

module.exports = serverItcHandlers;
49 changes: 47 additions & 2 deletions server/operationsServer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,8 @@ type ParsedSqlObject = any;
import { generateJsonApi } from '../resources/openApi.ts';
import { Resources } from '../resources/Resources.ts';
import { ServerError } from '../utility/errors/hdbError.ts';
import { sendItcEvent } from './threads/itc.js';
import { onMessageByType } from './threads/manageThreads.js';

const DEFAULT_HEADERS_TIMEOUT = 60000;
const REQ_MAX_BODY_SIZE = env.get(terms.CONFIG_PARAMS.OPERATIONSAPI_NETWORK_MAXREQUESTBODYSIZE) ?? 1024 * 1024 * 1024; //this defaults to 1GB in bytes
Expand Down Expand Up @@ -211,12 +213,55 @@ function buildServer(isHttps: boolean, resources: Resources): FastifyInstance {
return app;
}

let nextOpenApiRequestId = 1;
let openApiResponseListenerAttached = false;
const pendingOpenApiRequests = new Map<number, (openapi: unknown) => void>();

function attachOpenApiResponseListener() {
if (openApiResponseListenerAttached) return;
onMessageByType(terms.ITC_EVENT_TYPES.RESOURCE_OPENAPI_RESPONSE, ({ message }: any) => {
const resolve = pendingOpenApiRequests.get(message.requestId);
if (resolve) {
pendingOpenApiRequests.delete(message.requestId);
resolve(message.openapi);
}
});
openApiResponseListenerAttached = true;
}

function queryWorkerForOpenApi(serverHttpURL: string): Promise<unknown> {
attachOpenApiResponseListener();
const requestId = nextOpenApiRequestId++;
return new Promise<unknown>((resolve, reject) => {
const timeoutHandle = setTimeout(() => {
pendingOpenApiRequests.delete(requestId);
reject(new ServerError('Timeout fetching OpenAPI spec from worker thread', 503));
}, 5000);
pendingOpenApiRequests.set(requestId, (openapi) => {
clearTimeout(timeoutHandle);
resolve(openapi);
});
sendItcEvent({
type: terms.ITC_EVENT_TYPES.RESOURCE_OPENAPI_REQUEST,
message: { requestId, serverHttpURL },
}).catch((err: unknown) => {
clearTimeout(timeoutHandle);
pendingOpenApiRequests.delete(requestId);
reject(err);
});
});
}

function restOpenAPIHandler(resources: Resources) {
const httpPort = env.get(terms.CONFIG_PARAMS.HTTP_PORT);
const httpSecurePort = env.get(terms.CONFIG_PARAMS.HTTP_SECUREPORT);
return (req: FastifyRequest & { hdb_user?: { role?: { permission?: { super_user: boolean } } } }) => {
return async (req: FastifyRequest & { hdb_user?: { role?: { permission?: { super_user: boolean } } } }) => {
if (req.hdb_user?.role?.permission?.super_user) {
return generateJsonApi(resources, calculateRestHttpURL(httpPort, httpSecurePort, req));
const serverHttpURL = calculateRestHttpURL(httpPort, httpSecurePort, req);
if (resources.size > 0) {
return generateJsonApi(resources, serverHttpURL);
}
return queryWorkerForOpenApi(serverHttpURL);
} else {
harperLogger.warn(
`{"ip":"${req.socket.remoteAddress}", "error":"attempt to access /api/openapi/rest without being super_user"`
Expand Down
2 changes: 2 additions & 0 deletions server/threads/manageThreads.js
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,8 @@ listenersByType.set(hdbTerms.ITC_EVENT_TYPES.CHILD_STARTED, null);
listenersByType.set(hdbTerms.ITC_EVENT_TYPES.SCHEMA, null);
listenersByType.set(hdbTerms.ITC_EVENT_TYPES.USER, null);
listenersByType.set(hdbTerms.ITC_EVENT_TYPES.COMPONENT_STATUS_REQUEST, null);
listenersByType.set(hdbTerms.ITC_EVENT_TYPES.RESOURCE_OPENAPI_REQUEST, null);
listenersByType.set(hdbTerms.ITC_EVENT_TYPES.RESOURCE_OPENAPI_RESPONSE, null);

function startWorker(path, options = {}) {
// Take a percentage of total memory to determine the max memory for each thread. The percentage is based
Expand Down
101 changes: 101 additions & 0 deletions unitTests/server/itc/serverHandlers.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ const harperBridge = require('#src/dataLayer/harperBridge/harperBridge').default
// Note: rewire is used to access private functions (schemaHandler, userHandler, componentStatusRequestHandler)
// for testing validation logic, not for replacing dependencies with mocks
const server_itc_handlers = rewire('#js/server/itc/serverHandlers');
const { resetResources } = require('#src/resources/Resources');

describe('Test hdbChildIpcHandler module', () => {
const TEST_ERR = 'The roof is on fire';
Expand Down Expand Up @@ -221,4 +222,104 @@ describe('Test hdbChildIpcHandler module', () => {
sendToThreadStub.restore();
});
});

describe('Test resourceOpenApiRequestHandler function', () => {
let resource_openapi_handler;
let resources_instance;

before(() => {
resource_openapi_handler = server_itc_handlers.__get__('resourceOpenApiRequestHandler');
resources_instance = resetResources();
});

afterEach(() => {
resources_instance.clear();
});

// Tests validation: invalid events should be rejected and logged
it('logs error on invalid event (missing type)', async () => {
const test_event = {
message: { originator: 1, requestId: 42, serverHttpURL: 'http://localhost' },
};
await resource_openapi_handler(test_event);
expect(log_error_stub).to.have.been.called;
});

// Tests validation: invalid events should be rejected and logged
it('logs error on invalid event (missing message)', async () => {
const test_event = {
type: 'resource_openapi_request',
};
await resource_openapi_handler(test_event);
expect(log_error_stub).to.have.been.called;
});

// Tests validation: invalid events should be rejected and logged
it('logs error on invalid event (missing originator)', async () => {
const test_event = {
type: 'resource_openapi_request',
message: { requestId: 42, serverHttpURL: 'http://localhost' },
};
await resource_openapi_handler(test_event);
expect(log_error_stub).to.have.been.called;
});

// Tests guard: a thread with no registered resources must stay silent so that an app
// worker (which has real resources) responds first and the caller gets the correct spec.
it('does not respond when this thread has no registered resources', async () => {
sandbox.resetHistory();
const sendToThreadStub = sandbox.stub(global.threads, 'sendToThread').returns(true);

const test_event = {
type: 'resource_openapi_request',
message: { originator: 5, requestId: 99, serverHttpURL: 'http://localhost:9925' },
};
await resource_openapi_handler(test_event);

expect(sendToThreadStub).to.not.have.been.called;
expect(log_error_stub).to.not.have.been.called;
sendToThreadStub.restore();
});

it('sends OpenAPI response directly when originator is reachable', async () => {
sandbox.resetHistory();
// Resources.set wraps the argument as entry.Resource; passing {isError:true} makes
// generateJsonApi skip the entry so the spec is minimal but the send path is exercised.
resources_instance.set('test', { isError: true });
const sendToThreadStub = sandbox.stub(global.threads, 'sendToThread').returns(true);

const test_event = {
type: 'resource_openapi_request',
message: { originator: 5, requestId: 99, serverHttpURL: 'http://localhost:9925' },
};
await resource_openapi_handler(test_event);

expect(sendToThreadStub).to.have.been.calledOnce;
expect(sendToThreadStub.firstCall.args[0]).to.equal(5);
const responseMessage = sendToThreadStub.firstCall.args[1];
expect(responseMessage.type).to.equal('resource_openapi_response');
expect(responseMessage.message.requestId).to.equal(99);
expect(responseMessage.message.openapi).to.be.an('object');
expect(log_error_stub).to.not.have.been.called;
sendToThreadStub.restore();
});

it('drops response silently when originator is unreachable', async () => {
sandbox.resetHistory();
resources_instance.set('test', { isError: true });
const sendToThreadStub = sandbox.stub(global.threads, 'sendToThread').returns(false);

const test_event = {
type: 'resource_openapi_request',
message: { originator: 99, requestId: 7, serverHttpURL: 'http://localhost:9925' },
};
await resource_openapi_handler(test_event);

expect(sendToThreadStub).to.have.been.calledOnce;
expect(log_error_stub).to.not.have.been.called;
const traceCalls = log_trace_stub.getCalls().map((call) => String(call.args[0]));
expect(traceCalls.some((msg) => msg.includes('Dropping resource OpenAPI response'))).to.be.true;
sendToThreadStub.restore();
});
});
});
2 changes: 2 additions & 0 deletions utility/hdbTerms.ts
Original file line number Diff line number Diff line change
Expand Up @@ -802,6 +802,8 @@ export const ITC_EVENT_TYPES = {
START_JOB: 'start_job',
COMPONENT_STATUS_REQUEST: 'component_status_request',
COMPONENT_STATUS_RESPONSE: 'component_status_response',
RESOURCE_OPENAPI_REQUEST: 'resource_openapi_request',
RESOURCE_OPENAPI_RESPONSE: 'resource_openapi_response',
} as const;

/** Supported thread types */
Expand Down
Loading