diff --git a/integration-tests/worker/test/exit-reasons.test.ts b/integration-tests/worker/test/exit-reasons.test.ts index 8088ed244..598fa272e 100644 --- a/integration-tests/worker/test/exit-reasons.test.ts +++ b/integration-tests/worker/test/exit-reasons.test.ts @@ -242,5 +242,5 @@ test('crash: process.exit() triggered by postgres', async (t) => { t.is(reason, 'crash'); t.is(error_type, 'ExitError'); - t.regex(error_message, /Process exited with code: 1/i); + t.regex(error_message, /Worker thread exited with code: 1/i); }); diff --git a/packages/cli/src/checkout/command.ts b/packages/cli/src/checkout/command.ts index 8fab9d994..1400e5850 100644 --- a/packages/cli/src/checkout/command.ts +++ b/packages/cli/src/checkout/command.ts @@ -4,14 +4,14 @@ import { ensure, build } from '../util/command-builders'; import * as o from '../options'; export type CheckoutOptions = Required< - Pick + Pick > & Pick; -const options = [o.projectName, o.projectPath, o.log]; +const options = [o.projectId, o.projectPath, o.log]; const checkoutCommand: yargs.CommandModule = { - command: 'checkout ', + command: 'checkout ', describe: 'Switch to a different openfn project in the same workspace', handler: ensure('checkout', options), builder: (yargs) => build(options, yargs), diff --git a/packages/cli/src/checkout/handler.ts b/packages/cli/src/checkout/handler.ts index 746801354..2f071e825 100644 --- a/packages/cli/src/checkout/handler.ts +++ b/packages/cli/src/checkout/handler.ts @@ -19,20 +19,20 @@ const checkoutHandler = async (options: CheckoutOptions, logger: Logger) => { // get the project let switchProject; - if (/\.(yaml|json)$/.test(options.projectName)) { + if (/\.(yaml|json)$/.test(options.projectId)) { // TODO: should we allow checkout into an arbitrary folder? - const filePath = path.join(commandPath, options.projectName); + const filePath = path.join(commandPath, options.projectId); logger.debug('Loading project from path ', filePath); switchProject = await Project.from('path', filePath, { config, }); } else { - switchProject = workspace.get(options.projectName); + switchProject = workspace.get(options.projectId); } if (!switchProject) { logger.error( - `Project with id/name ${options.projectName} not found in the workspace` + `Project with id/name ${options.projectId} not found in the workspace` ); return; } diff --git a/packages/cli/src/merge/command.ts b/packages/cli/src/merge/command.ts index a07447e1d..4e7872262 100644 --- a/packages/cli/src/merge/command.ts +++ b/packages/cli/src/merge/command.ts @@ -7,17 +7,16 @@ export type MergeOptions = Required< Pick< Opts, | 'command' - | 'projectName' + | 'projectId' | 'projectPath' | 'removeUnmapped' | 'workflowMappings' - | 'log' - | 'force' > ->; +> & + Pick; const options = [ - o.projectName, + o.projectId, o.projectPath, o.removeUnmapped, o.workflowMappings, @@ -28,7 +27,7 @@ const options = [ ]; const mergeCommand: yargs.CommandModule = { - command: 'merge [project-name]', + command: 'merge ', describe: 'Merges the specified project into the checked out project', handler: ensure('merge', options), builder: (yargs) => build(options, yargs), diff --git a/packages/cli/src/merge/handler.ts b/packages/cli/src/merge/handler.ts index 80c8b8d06..cc0591c59 100644 --- a/packages/cli/src/merge/handler.ts +++ b/packages/cli/src/merge/handler.ts @@ -23,29 +23,29 @@ const mergeHandler = async (options: MergeOptions, logger: Logger) => { // Lookup the source project - the thing we are getting changes from let sourceProject; - if (/\.(yaml|json)$/.test(options.projectName)) { - const filePath = path.join(commandPath, options.projectName); + if (/\.(yaml|json)$/.test(options.projectId)) { + const filePath = path.join(commandPath, options.projectId); logger.debug('Loading source project from path ', filePath); sourceProject = await Project.from('path', filePath); } else { - sourceProject = workspace.get(options.projectName); + sourceProject = workspace.get(options.projectId); } if (!sourceProject) { - logger.error(`Project "${options.projectName}" not found in the workspace`); + logger.error(`Project "${options.projectId}" not found in the workspace`); return; } - if (targetProject.name === sourceProject.name) { + if (targetProject.id === sourceProject.id) { logger.error('Merging into the same project not allowed'); return; } - if (!targetProject.name) { - logger.error('The checked out project has no name/id'); + if (!targetProject.id) { + logger.error('The checked out project has no id'); return; } - const finalPath = workspace.getProjectPath(targetProject.name); + const finalPath = workspace.getProjectPath(targetProject.id); if (!finalPath) { logger.error('Path to checked out project not found.'); return; @@ -64,13 +64,13 @@ const mergeHandler = async (options: MergeOptions, logger: Logger) => { { command: 'checkout', projectPath: commandPath, - projectName: final.name || '', + projectId: final.id, log: options.log, }, logger ); logger.success( - `Project ${sourceProject.name} has been merged into Project ${targetProject.name} successfully` + `Project ${sourceProject.id} has been merged into Project ${targetProject.id} successfully` ); }; diff --git a/packages/cli/src/options.ts b/packages/cli/src/options.ts index b467406c2..514006042 100644 --- a/packages/cli/src/options.ts +++ b/packages/cli/src/options.ts @@ -336,19 +336,7 @@ const getBaseDir = (opts: { path?: string }) => { export const projectId: CLIOption = { name: 'project-id', yargs: { - hidden: true, - }, - ensure: (opts) => { - const projectId = opts.projectId; - //check that this is a uuid - return projectId; - }, -}; - -export const projectName: CLIOption = { - name: 'project-name', - yargs: { - description: 'The name of an openfn project', + description: 'The id or UUID of an openfn project', string: true, }, ensure: (opts) => { diff --git a/packages/cli/src/projects/handler.ts b/packages/cli/src/projects/handler.ts index af76e4bc9..31b8fa048 100644 --- a/packages/cli/src/projects/handler.ts +++ b/packages/cli/src/projects/handler.ts @@ -13,19 +13,17 @@ const projectsHandler = async (options: ProjectsOptions, logger: Logger) => { logger.success(`Available openfn projects\n\n${workspace .list() - .map((p) => describeProject(p, p.name === workspace.activeProjectId)) + .map((p) => describeProject(p, p.id === workspace.activeProjectId)) .join('\n\n')} `); }; function describeProject(project: Project, active = false) { // @ts-ignore - const pId = project.openfn?.uuid; - return `${project.name} ${active ? '(active)' : ''}\n ${ - pId || '' - }\n workflows:\n${project.workflows - .map((w) => ' - ' + w.name) - .join('\n')}`; + const uuid = project.openfn?.uuid; + return `${project.id} ${active ? '(active)' : ''}\n ${ + uuid || '' + }\n workflows:\n${project.workflows.map((w) => ' - ' + w.id).join('\n')}`; } export default projectsHandler; diff --git a/packages/cli/src/version/command.ts b/packages/cli/src/version/command.ts index 1141f2a76..73686f18c 100644 --- a/packages/cli/src/version/command.ts +++ b/packages/cli/src/version/command.ts @@ -6,22 +6,11 @@ import * as o from '../options'; export type VersionOptions = Required< Pick< Opts, - | 'command' - | 'workflow' - | 'projectName' - | 'projectPath' - | 'workflowMappings' - | 'json' + 'command' | 'workflow' | 'projectPath' | 'workflowMappings' | 'json' > >; -const options = [ - o.workflow, - o.projectName, - o.projectPath, - o.workflowMappings, - o.json, -]; +const options = [o.workflow, o.projectPath, o.workflowMappings, o.json]; const workflowVersionCommand: yargs.CommandModule = { command: 'project version [workflow]', diff --git a/packages/cli/test/checkout/handler.test.ts b/packages/cli/test/checkout/handler.test.ts index ecfde24af..3bf9d03c3 100644 --- a/packages/cli/test/checkout/handler.test.ts +++ b/packages/cli/test/checkout/handler.test.ts @@ -9,17 +9,21 @@ test.beforeEach(() => { mock({ '/ws/workflows': {}, '/ws/openfn.yaml': jsonToYaml({ - name: 'some-project-name', - workflowRoot: 'workflows', - formats: { - openfn: 'yaml', - project: 'yaml', - workflow: 'yaml', + project: { + id: 'my-project', + }, + workspace: { + workflowRoot: 'workflows', + formats: { + openfn: 'yaml', + project: 'yaml', + workflow: 'yaml', + }, }, }), '/ws/.projects/staging@app.openfn.org.yaml': jsonToYaml({ - id: 'some-id', - name: 'some-project-name', + id: '>', + name: 'My Staging', workflows: [ { name: 'simple-workflow', @@ -80,8 +84,8 @@ test.beforeEach(() => { ], }), '/ws/.projects/project@app.openfn.org.yaml': jsonToYaml({ - id: 'main-id', - name: 'main-project-id', + id: '', + name: 'My Project', workflows: [ { name: 'simple-workflow-main', @@ -149,12 +153,12 @@ const logger = createMockLogger('', { level: 'debug' }); test.serial('get active project', (t) => { const workspace = new Workspace('/ws'); t.is(workspace.valid, true); - t.is(workspace.activeProjectId, 'some-project-name'); + t.is(workspace.activeProjectId, 'my-project'); }); test.serial('checkout: invalid project id', (t) => { checkoutHandler( - { command: 'checkout', projectName: 'not-known', projectPath: '/ws' }, + { command: 'checkout', projectId: 'not-known', projectPath: '/ws' }, logger ); const { message } = logger._parse(logger._last); @@ -162,22 +166,20 @@ test.serial('checkout: invalid project id', (t) => { }); test.serial('checkout: to a different valid project', async (t) => { - // before checkout. some-project-name is active and expanded + // before checkout. my-project is active and expanded const bcheckout = new Workspace('/ws'); - t.is(bcheckout.projectMeta.name, 'some-project-name'); - t.is(bcheckout.getActiveProject()?.name, 'some-project-name'); + t.is(bcheckout.activeProject.id, 'my-project'); await checkoutHandler( - { command: 'checkout', projectName: 'main-project-id', projectPath: '/ws' }, + { command: 'checkout', projectId: 'my-project', projectPath: '/ws' }, logger ); const { message } = logger._parse(logger._last); t.is(message, 'Expanded project to /ws'); - // after checkout. main-project-id is active and expanded + // after checkout. my-project is active and expanded const acheckout = new Workspace('/ws'); - t.is(acheckout.projectMeta.name, 'main-project-id'); - t.is(acheckout.getActiveProject()?.name, 'main-project-id'); + t.is(acheckout.activeProject.id, 'my-project'); // check if files where well expanded t.deepEqual( @@ -187,15 +189,14 @@ test.serial('checkout: to a different valid project', async (t) => { }); test.serial('checkout: same id as active', async (t) => { - // before checkout. some-project-name is active and expanded + // before checkout. my-project is active and expanded const bcheckout = new Workspace('/ws'); - t.is(bcheckout.projectMeta.name, 'some-project-name'); - t.is(bcheckout.getActiveProject()?.name, 'some-project-name'); + t.is(bcheckout.activeProject.id, 'my-project'); await checkoutHandler( { command: 'checkout', - projectName: 'some-project-name', + projectId: 'my-project', projectPath: '/ws', }, logger @@ -203,48 +204,45 @@ test.serial('checkout: same id as active', async (t) => { const { message } = logger._parse(logger._last); t.is(message, 'Expanded project to /ws'); - // after checkout. main-project-id is active and expanded + // after checkout. my-project is active and expanded const acheckout = new Workspace('/ws'); - t.is(acheckout.projectMeta.name, 'some-project-name'); - t.is(acheckout.getActiveProject()?.name, 'some-project-name'); + t.is(acheckout.activeProject.id, 'my-project'); // check if files where well expanded t.deepEqual( fs.readdirSync('/ws/workflows').sort(), - ['simple-workflow', 'another-workflow'].sort() + ['simple-workflow-main', 'another-workflow-main'].sort() ); }); test.serial('checkout: switching to and back between projects', async (t) => { - // before checkout. some-project-name is active and expanded + // before checkout. my-project is active and expanded const bcheckout = new Workspace('/ws'); - t.is(bcheckout.projectMeta.name, 'some-project-name'); - t.is(bcheckout.getActiveProject()?.name, 'some-project-name'); + t.is(bcheckout.activeProject.id, 'my-project'); - // 1. switch from some-project-name to main-project-id + // 1. switch from my-project to my-staging await checkoutHandler( - { command: 'checkout', projectName: 'main-project-id', projectPath: '/ws' }, + { command: 'checkout', projectId: 'my-staging', projectPath: '/ws' }, logger ); const { message } = logger._parse(logger._last); t.is(message, 'Expanded project to /ws'); - // after checkout. main-project-id is active and expanded + // after checkout. my-staging is active and expanded const acheckout = new Workspace('/ws'); - t.is(acheckout.projectMeta.name, 'main-project-id'); - t.is(acheckout.getActiveProject()?.name, 'main-project-id'); + t.is(acheckout.activeProject.id, 'my-staging'); // check if files where well expanded t.deepEqual( fs.readdirSync('/ws/workflows').sort(), - ['simple-workflow-main', 'another-workflow-main'].sort() + ['simple-workflow', 'another-workflow'].sort() ); - // 2. switch back from main-project-id to some-project-name + // 2. switch back from my-project to my-project await checkoutHandler( { command: 'checkout', - projectName: 'some-project-name', + projectId: 'my-project', projectPath: '/ws', }, logger @@ -252,14 +250,13 @@ test.serial('checkout: switching to and back between projects', async (t) => { const { message: lastMsg } = logger._parse(logger._last); t.is(lastMsg, 'Expanded project to /ws'); - // after checkout. main-project-id is active and expanded + // after checkout. my-project is active and expanded const fcheckout = new Workspace('/ws'); - t.is(fcheckout.projectMeta.name, 'some-project-name'); - t.is(fcheckout.getActiveProject()?.name, 'some-project-name'); + t.is(fcheckout.activeProject.id, 'my-project'); // check if files where well expanded t.deepEqual( fs.readdirSync('/ws/workflows').sort(), - ['simple-workflow', 'another-workflow'].sort() + ['simple-workflow-main', 'another-workflow-main'].sort() ); }); diff --git a/packages/cli/test/merge/handler.test.ts b/packages/cli/test/merge/handler.test.ts index fe9a183b6..9f5e85e2b 100644 --- a/packages/cli/test/merge/handler.test.ts +++ b/packages/cli/test/merge/handler.test.ts @@ -8,21 +8,28 @@ test.beforeEach(() => { mock({ '/ws/workflows': {}, '/ws/openfn.yaml': jsonToYaml({ - name: 'main-project-id', - workflowRoot: 'workflows', - formats: { - openfn: 'yaml', - project: 'yaml', - workflow: 'yaml', + project: { + id: 'my-project', + name: 'My Project', + }, + workspace: { + dirs: { + workflows: 'workflows', + }, + formats: { + openfn: 'yaml', + project: 'yaml', + workflow: 'yaml', + }, }, }), '/ws/.projects/staging@app.openfn.org.yaml': jsonToYaml({ - id: 'sandbox-uuid', - name: 'sandbox-project-id', + id: '', + name: 'My Sandbox', workflows: [ { - name: 'first-workflow', - id: 'first-workflow-id', + name: 'Workflow 1', + id: 'workflow-1', jobs: [ { id: 'job-x', @@ -62,12 +69,12 @@ test.beforeEach(() => { ], }), '/ws/.projects/project@app.openfn.org.yaml': jsonToYaml({ - id: 'main-project-uuid', - name: 'main-project-id', + id: '', + name: 'My Project', workflows: [ { - name: 'first-workflow', - id: 'first-workflow-id', + name: 'Workflow 1', + id: 'workflow-1', jobs: [{ id: 'job-a', name: 'Job A' }], triggers: [ { @@ -98,7 +105,7 @@ test('merging into the same project', async (t) => { { command: 'merge', projectPath: '/ws', - projectName: 'main-project-id', + projectId: 'my-project', removeUnmapped: false, workflowMappings: {}, }, @@ -112,40 +119,39 @@ test('merging into the same project', async (t) => { test('merging a different project into checked-out', async (t) => { // state of main projects workflow before sandbox is merged in - const bworkspace = new Workspace('/ws'); - t.is(bworkspace.projectMeta.name, 'main-project-id'); - t.is(bworkspace.getActiveProject()?.name, 'main-project-id'); - const bprojects = bworkspace.list(); - t.is(bprojects[0].workflows[0].steps.length, 2); - t.is(bprojects[0].workflows[0].steps[1].name, 'Job A'); + const beforeWs = new Workspace('/ws'); + t.is(beforeWs.activeProject.id, 'my-project'); + const beforeProjects = beforeWs.list(); + t.is(beforeProjects[0].workflows[0].steps.length, 2); + t.is(beforeProjects[0].workflows[0].steps[1].name, 'Job A'); // do merging await mergeHandler( { command: 'merge', projectPath: '/ws', - projectName: 'sandbox-project-id', + projectId: 'my-sandbox', removeUnmapped: false, workflowMappings: {}, }, logger ); - // state of main projects workflow before sandbox is merged in - const workspace = new Workspace('/ws'); - t.is(workspace.projectMeta.name, 'main-project-id'); - t.is(workspace.getActiveProject()?.name, 'main-project-id'); - const projects = workspace.list(); - t.is(projects[0].workflows[0].steps.length, 3); - t.is(projects[0].workflows[0].steps[1].name, 'Job X'); - t.is(projects[0].workflows[0].steps[1].openfn?.uuid, 'job-a'); // id got retained - t.is(projects[0].workflows[0].steps[2].name, 'Job Y'); - t.is(projects[0].workflows[0].steps[2].openfn?.uuid, 'job-y'); // id not retained - new nod + // state of main projects workflow AFTER sandbox is merged in + const afterWorkspace = new Workspace('/ws'); + t.is(afterWorkspace.activeProject.id, 'my-project'); + const afterProjects = afterWorkspace.list(); + const wf = afterProjects[0].workflows[0]; + t.is(wf.steps.length, 3); + t.is(wf.steps[1].name, 'Job X'); + t.is(wf.steps[1].openfn?.uuid, 'job-a'); // id got retained + t.is(wf.steps[2].name, 'Job Y'); + t.is(wf.steps[2].openfn?.uuid, 'job-y'); // id not retained - new nod const { message, level } = logger._parse(logger._last); t.is(level, 'success'); t.is( message, - 'Project sandbox-project-id has been merged into Project main-project-id successfully' + 'Project my-sandbox has been merged into Project my-project successfully' ); }); diff --git a/packages/cli/test/projects/handler.test.ts b/packages/cli/test/projects/handler.test.ts index 6a26a71cb..086b475e1 100644 --- a/packages/cli/test/projects/handler.test.ts +++ b/packages/cli/test/projects/handler.test.ts @@ -7,17 +7,21 @@ import { jsonToYaml } from '@openfn/project'; mock({ 'no-ws/': { 'some.yaml': 'name: smth' }, '/ws/openfn.yaml': jsonToYaml({ - name: 'some-project-name', - workflowRoot: 'workflows', - formats: { - openfn: 'yaml', - project: 'yaml', - workflow: 'yaml', + project: { + id: 'my-project', + }, + workspace: { + workflowRoot: 'workflows', + formats: { + openfn: 'yaml', + project: 'yaml', + workflow: 'yaml', + }, }, }), '/ws/.projects/staging@app.openfn.org.yaml': jsonToYaml({ - id: 'some-id', - name: 'some-project-name', + id: '', + name: 'my-project', workflows: [ { name: 'simple-workflow', @@ -93,15 +97,15 @@ test('openfn projects: not a workspace', (t) => { t.is(message, 'Command was run in an invalid openfn workspace'); }); -test('openfn projects: valid workspace', (t) => { +test.only('openfn projects: valid workspace', (t) => { projectsHandler({ command: 'projects', projectPath: '/ws' }, logger); const { message, level } = logger._parse(logger._last); t.is('success', level); t.is( `Available openfn projects -some-project-name (active) - some-id +my-project (active) + workflows: - simple-workflow - another-workflow diff --git a/packages/engine-multi/CHANGELOG.md b/packages/engine-multi/CHANGELOG.md index 4ee9895e1..ddb04f467 100644 --- a/packages/engine-multi/CHANGELOG.md +++ b/packages/engine-multi/CHANGELOG.md @@ -1,5 +1,13 @@ # engine-multi +## 1.8.2 + +### Patch Changes + +- 8632629: Remove log line +- 0e5b7d4: Fix an issue where an empty stderr buffer from a crashed run could fail to return, triggering Lost +- 7e12e4b: Wait for child processes to exit properly when shutting down (shouldn't affect prod environments) + ## 1.8.1 ### Patch Changes diff --git a/packages/engine-multi/package.json b/packages/engine-multi/package.json index 36d12d07d..cec240182 100644 --- a/packages/engine-multi/package.json +++ b/packages/engine-multi/package.json @@ -1,6 +1,6 @@ { "name": "@openfn/engine-multi", - "version": "1.8.1", + "version": "1.8.2", "description": "Multi-process runtime engine", "main": "dist/index.js", "type": "module", diff --git a/packages/engine-multi/src/engine.ts b/packages/engine-multi/src/engine.ts index 144e345d7..4f2aae9f4 100644 --- a/packages/engine-multi/src/engine.ts +++ b/packages/engine-multi/src/engine.ts @@ -229,7 +229,7 @@ const createEngine = async ( // How does this work if deferred? }; - const destroy = (instant?: boolean) => closeWorkers(instant); + const destroy = async (instant?: boolean) => closeWorkers(instant); return Object.assign(engine, { options, diff --git a/packages/engine-multi/src/test/worker-functions.ts b/packages/engine-multi/src/test/worker-functions.ts index f562edbcb..15fd27bf6 100644 --- a/packages/engine-multi/src/test/worker-functions.ts +++ b/packages/engine-multi/src/test/worker-functions.ts @@ -56,6 +56,11 @@ const tasks = { throw new Error('test_error'); }, + weirdExit: async () => { + // https://www.youtube.com/watch?v=Z2cXRtblqjQ + process.exit(72); + }, + // Experiments with freezing the global scope // We may do this in the actual worker freeze: async () => { diff --git a/packages/engine-multi/src/types.ts b/packages/engine-multi/src/types.ts index 92044b052..f0a75eed8 100644 --- a/packages/engine-multi/src/types.ts +++ b/packages/engine-multi/src/types.ts @@ -57,7 +57,7 @@ export type ExecutionContextOptions = ExecuteOptions & EngineOptions; export interface EngineAPI extends EventEmitter { callWorker: CallWorker; - closeWorkers: (instant?: boolean) => void; + closeWorkers: (instant?: boolean) => Promise; } export interface RuntimeEngine { @@ -74,7 +74,7 @@ export interface RuntimeEngine { options?: Partial ): Pick; - destroy(): void; + destroy(instant?: boolean): Promise; on: (evt: string, fn: (...args: any[]) => void) => void; } diff --git a/packages/engine-multi/src/worker/pool.ts b/packages/engine-multi/src/worker/pool.ts index 891085e5b..a3257f8e4 100644 --- a/packages/engine-multi/src/worker/pool.ts +++ b/packages/engine-multi/src/worker/pool.ts @@ -67,7 +67,6 @@ export const returnToPool = ( // creates a new pool of workers which use the same script function createPool(script: string, options: PoolOptions = {}, logger: Logger) { const capacity = options.capacity || options.maxWorkers || 5; - logger.debug(`pool: Creating new child process pool | capacity: ${capacity}`); let destroyed = false; @@ -144,7 +143,7 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { // TODO what should we do if a process in the pool dies, perhaps due to OOM? const onExit = async (code: number) => { if (code !== HANDLED_EXIT_CODE) { - logger.debug('pool: Worker exited unexpectedly'); + logger.debug(`pool: Worker exited unexpectedly with code ${code}`); clearTimeout(timeout); // Read the stderr stream from the worked to see if this looks like an OOM error @@ -154,22 +153,22 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { }); try { - for await (const line of rl) { - logger.debug(line); - if (line.match(/JavaScript heap out of memory/)) { - reject(new OOMError()); - - killWorker(worker); - // restore a placeholder to the queue - finish(false); - return; + if (worker.stderr && worker.stderr?.readableLength > 0) { + for await (const line of rl) { + if (line.match(/JavaScript heap out of memory/)) { + killWorker(worker); + // restore a placeholder to the queue + finish(false); + reject(new OOMError()); + return; + } } } } catch (e) { // do nothing } - reject(new ExitError(code)); finish(worker); + reject(new ExitError(code)); } }; @@ -224,8 +223,6 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { worker.on('exit', onExit); worker.on('message', (evt: any) => { - // TODO I think here we may have to decode the payload - // forward the message out of the pool opts.on?.(evt); @@ -265,18 +262,53 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { } }; - const destroy = (immediate = false) => { + const waitForWorkerExit = ( + worker: ChildProcess, + forceKillTimeout = 5000 + ): Promise => { + return new Promise((resolve) => { + if (!worker || worker.killed || !worker.connected) { + resolve(); + return; + } + + const timeout = setTimeout(() => { + logger.debug('pool: force killing worker', worker.pid); + worker.kill('SIGKILL'); + resolve(); + }, forceKillTimeout); + + worker.once('exit', () => { + clearTimeout(timeout); + resolve(); + }); + + worker.kill(); + }); + }; + + const destroy = async (immediate = false): Promise => { destroyed = true; + const killPromises: Promise[] = []; + // Drain the pool while (pool.length) { - killWorker(pool.pop()!); + const worker = pool.pop(); + if (worker) { + killPromises.push(waitForWorkerExit(worker)); + delete allWorkers[worker.pid!]; + } } if (immediate) { - Object.values(allWorkers).forEach(killWorker); + Object.values(allWorkers).forEach((worker) => { + killPromises.push(waitForWorkerExit(worker, 1000)); + delete allWorkers[worker.pid!]; + }); } - // TODO set a timeout and force any outstanding workers to die + + await Promise.all(killPromises); }; const api = { diff --git a/packages/engine-multi/src/worker/thread/runtime.ts b/packages/engine-multi/src/worker/thread/runtime.ts index 0b3eae7ea..4e5e862b7 100644 --- a/packages/engine-multi/src/worker/thread/runtime.ts +++ b/packages/engine-multi/src/worker/thread/runtime.ts @@ -80,6 +80,16 @@ const run = (task: string, args: any[], options: Options = {}) => { }); }; +process.on('exit', (code) => { + publish(ENGINE_REJECT_TASK, { + error: { + name: 'ExitError', + message: `Worker thread exited with code: ${code}`, + severity: 'crash', + }, + }); +}); + parentPort!.on('message', async (evt) => { if (evt.type === ENGINE_RUN_TASK) { const args = evt.args || []; diff --git a/packages/engine-multi/test/api.test.ts b/packages/engine-multi/test/api.test.ts index b56573864..40759f5ed 100644 --- a/packages/engine-multi/test/api.test.ts +++ b/packages/engine-multi/test/api.test.ts @@ -12,9 +12,9 @@ import loadVersions from '../src/util/load-versions'; const logger = createMockLogger(undefined, { level: 'debug' }); let api: RuntimeEngine; -test.afterEach(() => { +test.afterEach(async () => { logger._reset(); - api?.destroy(); + await api?.destroy(); }); test.serial('create a default engine api without throwing', async (t) => { diff --git a/packages/engine-multi/test/errors.test.ts b/packages/engine-multi/test/errors.test.ts index bd8020135..a206b37d2 100644 --- a/packages/engine-multi/test/errors.test.ts +++ b/packages/engine-multi/test/errors.test.ts @@ -20,7 +20,7 @@ test.before(async () => { }, maxWorkers: 1, memoryLimitMb: 200, - proxyStdout: true, + // proxyStdout: true, }; // This uses the real runtime and real worker @@ -252,7 +252,7 @@ test.serial('emit a crash error on process.exit()', (t) => { engine.execute(plan, {}).on(WORKFLOW_ERROR, (evt) => { t.is(evt.type, 'ExitError'); t.is(evt.severity, 'crash'); - t.is(evt.message, 'Process exited with code: 42'); + t.is(evt.message, 'Worker thread exited with code: 42'); done(); }); }); diff --git a/packages/engine-multi/test/integration.test.ts b/packages/engine-multi/test/integration.test.ts index f03a2f693..bff1b982d 100644 --- a/packages/engine-multi/test/integration.test.ts +++ b/packages/engine-multi/test/integration.test.ts @@ -11,9 +11,9 @@ let api: RuntimeEngine; const emptyState = {}; -test.afterEach(() => { +test.afterEach(async () => { logger._reset(); - api.destroy(); + await api.destroy(); }); // this tests the full API with the actual runtime diff --git a/packages/engine-multi/test/worker/pool.test.ts b/packages/engine-multi/test/worker/pool.test.ts index aa74d6a6f..190dacbbe 100644 --- a/packages/engine-multi/test/worker/pool.test.ts +++ b/packages/engine-multi/test/worker/pool.test.ts @@ -185,9 +185,19 @@ test('throw if memory limit is exceeded', async (t) => { } }); +test('handle weird exit', async (t) => { + const pool = createPool(workerPath, {}, logger); + + try { + await pool.exec('weirdExit', []); + } catch (e: any) { + t.is(e.message, 'Worker thread exited with code: 72'); + } +}); + test('destroy should handle un-initialised workers', async (t) => { const pool = createPool(workerPath, { capacity: 10 }, logger); - pool.destroy(); + await pool.destroy(); t.is(pool._pool.length, 0); }); @@ -201,7 +211,7 @@ test('destroy should close all child processes', async (t) => { const workers = Object.values(pool._allWorkers); // now destroy it - pool.destroy(); + await pool.destroy(); // check that every child is disconnected t.true(workers.every((child) => child.killed)); @@ -210,27 +220,26 @@ test('destroy should close all child processes', async (t) => { t.is(pool._pool.length, 0); }); -test('destroy gracefully', (t) => { - return new Promise((done) => { - const pool = createPool(workerPath, {}, logger); - const workers = Object.values(pool._allWorkers); +test('destroy gracefully', async (t) => { + const pool = createPool(workerPath, {}, logger); + const workers = Object.values(pool._allWorkers); - t.is(pool._pool.length, 5); + t.is(pool._pool.length, 5); - pool.exec('wait', [100]).then((result) => { - t.is(result, 1); - setTimeout(() => { - t.true(workers.every((child) => child.killed)); - t.is(pool._pool.length, 0); + const taskPromise = pool.exec('wait', [100]); - done(); - }, 1); - }); + await pool.destroy(); + + t.is(pool._pool.length, 0); - pool.destroy(); + // The task should complete gracefully + const result = await taskPromise; + t.is(result, 1); - t.is(pool._pool.length, 0); - }); + // Give a moment for cleanup to finish + await new Promise((resolve) => setTimeout(resolve, 10)); + + t.true(workers.every((child) => child.killed)); }); // TODO should the worker throw on sigterm? diff --git a/packages/lightning-mock/CHANGELOG.md b/packages/lightning-mock/CHANGELOG.md index 5df4b84c9..8548c8591 100644 --- a/packages/lightning-mock/CHANGELOG.md +++ b/packages/lightning-mock/CHANGELOG.md @@ -1,5 +1,14 @@ # @openfn/lightning-mock +## 2.3.5 + +### Patch Changes + +- Updated dependencies [8632629] +- Updated dependencies [0e5b7d4] +- Updated dependencies [7e12e4b] + - @openfn/engine-multi@1.8.2 + ## 2.3.4 ### Patch Changes diff --git a/packages/lightning-mock/package.json b/packages/lightning-mock/package.json index 7eaf0b41b..9c618d918 100644 --- a/packages/lightning-mock/package.json +++ b/packages/lightning-mock/package.json @@ -1,6 +1,6 @@ { "name": "@openfn/lightning-mock", - "version": "2.3.4", + "version": "2.3.5", "private": true, "description": "A mock Lightning server", "main": "dist/index.js", diff --git a/packages/project/CHANGELOG.md b/packages/project/CHANGELOG.md index ac71a051e..8d6615fcf 100644 --- a/packages/project/CHANGELOG.md +++ b/packages/project/CHANGELOG.md @@ -1,5 +1,11 @@ # @openfn/project +## 0.7.1 + +### Patch Changes + +- 8632629: Internal refactor for clarity + ## 0.7.0 ### Minor Changes diff --git a/packages/project/package.json b/packages/project/package.json index a62c8ee67..830bea041 100644 --- a/packages/project/package.json +++ b/packages/project/package.json @@ -1,6 +1,6 @@ { "name": "@openfn/project", - "version": "0.7.0", + "version": "0.7.1", "description": "Read, serialize, replicate and sync OpenFn projects", "scripts": { "test": "pnpm ava", @@ -37,6 +37,7 @@ "@openfn/lexicon": "workspace:^", "@openfn/logger": "workspace:*", "glob": "^11.0.2", + "human-id": "^4.1.1", "lodash": "^4.17.21", "lodash-es": "^4.17.21", "ohm-js": "^17.2.1", diff --git a/packages/project/src/Project.ts b/packages/project/src/Project.ts index f96de8280..2242f2a54 100644 --- a/packages/project/src/Project.ts +++ b/packages/project/src/Project.ts @@ -1,3 +1,4 @@ +import { humanId } from 'human-id'; import Workflow from './Workflow'; import * as serializers from './serialize'; import fromAppState, { FromAppStateConfig } from './parse/from-app-state'; @@ -19,35 +20,6 @@ type MergeOptions = { const maybeCreateWorkflow = (wf: any) => wf instanceof Workflow ? wf : new Workflow(wf); -// TODO -------------- -// I think this needs renaming to config -// and it's part of the workspace technically -// I need to support custom props -// When serializing, for now, we always write defaults -// -------------- -// repo-wide options -type RepoOptions = { - /**default workflow root when serializing to fs (relative to openfn.yaml) */ - // TODO deprecate this - workflowRoot?: string; - - formats: { - openfn: FileFormats; - workflow: FileFormats; - project: FileFormats; - }; -}; - -// A local collection of openfn projects? -// class Repo { - -// projects: {} -// } - -// TODO maybe use an npm for this, or create util - -// TODO this need to be controlled by the workspace - // A single openfn project // could be an app project or a checked out fs export class Project { @@ -55,11 +27,15 @@ export class Project { // And how are we tracking this? // version; - /** project name */ + /** Human readable project name. This corresponds to the label in Lightning */ name?: string; + + /** Project id. Must be url safe. May be derived from the name. NOT a UUID */ + id: string; + description?: string; - // array of version shas + // array of version hashes history: string[] = []; workflows: Workflow[]; @@ -79,12 +55,6 @@ export class Project { config: WorkspaceConfig; - // load a project from a state file (project.json) - // or from a path (the file system) - // TODO presumably we can detect a state file? Not a big deal? - - // collections for the project - // TODO to be well typed collections: any; static from( @@ -129,10 +99,16 @@ export class Project { // stuff that's external to the actual project and managed by the repo // TODO maybe the constructor is (data, Workspace) - constructor(data: l.Project, repoConfig: RepoOptions = {}) { - this.setConfig(repoConfig); + constructor(data: l.Project, config: RepoOptions = {}) { + this.setConfig(config); + + this.id = + data.id ?? data.name + ? slugify(data.name) + : humanId({ separator: '-', capitalize: false }); this.name = data.name; + this.description = data.description; this.openfn = data.openfn; this.options = data.options; @@ -154,19 +130,12 @@ export class Project { throw new Error(`Cannot serialize ${type}`); } - // would like a better name for this - // stamp? id? sha? - // this builds a version string for the current state - getVersionHash() {} - - // what else might we need? - - // get workflow by name or id - // this is fuzzy, but is that wrong? + // get workflow by name, id or uuid getWorkflow(idOrName: string) { return ( this.workflows.find((wf) => wf.id == idOrName) || - this.workflows.find((wf) => wf.name === idOrName) + this.workflows.find((wf) => wf.name === idOrName) || + this.workflows.find((wf) => wf.openfn?.uuid === idOrName) ); } diff --git a/packages/project/src/Workspace.ts b/packages/project/src/Workspace.ts index 0b196bf0c..027cacb5d 100644 --- a/packages/project/src/Workspace.ts +++ b/packages/project/src/Workspace.ts @@ -16,7 +16,7 @@ const PROJECT_EXTENSIONS = ['.yaml', '.yml']; export class Workspace { config?: WorkspaceConfig; - projectMeta: ProjectMeta; + activeProject: ProjectMeta; private projects: Project[] = []; private projectPaths = new Map(); @@ -26,7 +26,6 @@ export class Workspace { let context; try { const { type, content } = findWorkspaceFile(workspacePath); - console.log(content); context = loadWorkspaceFile(content, type); this.isValid = true; } catch (e) { @@ -36,7 +35,7 @@ export class Workspace { } this.config = buildConfig(context.workspace); - this.projectMeta = context.project; + this.activeProject = context.project; const projectsPath = path.join(workspacePath, this.config.dirs.projects); @@ -55,7 +54,7 @@ export class Workspace { const stateFilePath = path.join(projectsPath, file); const data = fs.readFileSync(stateFilePath, 'utf-8'); const project = fromAppState(data, { format: 'yaml' }); - this.projectPaths.set(project.name, stateFilePath); + this.projectPaths.set(project.id, stateFilePath); return project; }) .filter((s) => s); @@ -73,9 +72,12 @@ export class Workspace { return this.projects; } - // TODO clear up name/id confusion + /** Get a project by its id or UUID */ get(id: string) { - return this.projects.find((p) => p.name === id); + return ( + this.projects.find((p) => p.id === id) ?? + this.projects.find((p) => p.openfn?.uuid === id) + ); } getProjectPath(id: string) { @@ -83,8 +85,10 @@ export class Workspace { } getActiveProject() { - // TODO should use id, not name - return this.projects.find((p) => p.name === this.projectMeta?.name); + return ( + this.projects.find((p) => p.id === this.activeProject?.id) ?? + this.projects.find((p) => p.openfn?.uuid === this.activeProject?.id) + ); } // TODO this needs to return default values @@ -94,8 +98,7 @@ export class Workspace { } get activeProjectId() { - // TODO should return activeProject.id - return this.projectMeta?.name; + return this.activeProject?.id; } get valid() { diff --git a/packages/project/src/parse/from-fs.ts b/packages/project/src/parse/from-fs.ts index e408197ea..2cfedc534 100644 --- a/packages/project/src/parse/from-fs.ts +++ b/packages/project/src/parse/from-fs.ts @@ -53,6 +53,7 @@ export const parseProject = async (options: FromFsConfig = {}) => { } const proj = { + name: state?.name, openfn: context.project, config: config, workflows: [], @@ -81,7 +82,7 @@ export const parseProject = async (options: FromFsConfig = {}) => { const wfState = (state && state.getWorkflow(wf.id)) ?? {}; wf.openfn = { uuid: wfState.openfn?.uuid ?? null, - // TODO do we need to transfer more stuff? + // TODO do we need to transfer more stuff? Options maybe? }; //console.log('Loading workflow at ', filePath); // TODO logger.debug diff --git a/packages/project/src/util/config.ts b/packages/project/src/util/config.ts index df80ac59b..cc7a26efa 100644 --- a/packages/project/src/util/config.ts +++ b/packages/project/src/util/config.ts @@ -81,6 +81,7 @@ export const buildConfig = (config: WorkspaceConfig = {}) => ({ export const extractConfig = (source: Project) => { const project = { ...(source.openfn || {}), + id: source.id, }; const workspace = { ...source.config, @@ -146,13 +147,10 @@ export const loadWorkspaceFile = ( }; export const findWorkspaceFile = (dir: string = '.') => { - console.log({ dir }); let content, type; try { type = 'yaml'; - console.log(path.resolve(path.join(dir, 'openfn.yaml'))); content = readFileSync(path.resolve(path.join(dir, 'openfn.yaml')), 'utf8'); - console.log({ content }); } catch (e) { // Not found - try and parse as JSON try { diff --git a/packages/project/test/serialize/to-fs.test.ts b/packages/project/test/serialize/to-fs.test.ts index a455338aa..d864cb2bd 100644 --- a/packages/project/test/serialize/to-fs.test.ts +++ b/packages/project/test/serialize/to-fs.test.ts @@ -187,6 +187,7 @@ test('extractWorkflow: single simple workflow with custom root', (t) => { test('toFs: extract a project with 1 workflow and 1 step', (t) => { const project = new Project( { + name: 'My Project', workflows: [ { id: 'my-workflow', @@ -219,7 +220,9 @@ test('toFs: extract a project with 1 workflow and 1 step', (t) => { formats: { openfn: 'json', project: 'yaml', workflow: 'json' }, dirs: { projects: '.projects', workflows: 'workflows' }, }, - project: {}, + project: { + id: 'my-project', + }, }); const workflow = JSON.parse(files['workflows/my-workflow/my-workflow.json']); diff --git a/packages/project/test/util/config.test.ts b/packages/project/test/util/config.test.ts index b463d52d0..8bc520264 100644 --- a/packages/project/test/util/config.test.ts +++ b/packages/project/test/util/config.test.ts @@ -167,6 +167,7 @@ test('generate openfn.yaml', (t) => { result.content, `project: uuid: 1234 + id: my-project workspace: formats: openfn: yaml diff --git a/packages/project/test/workspace.test.ts b/packages/project/test/workspace.test.ts index e1bdc2747..856e67fbe 100644 --- a/packages/project/test/workspace.test.ts +++ b/packages/project/test/workspace.test.ts @@ -5,27 +5,28 @@ import test from 'ava'; // TODO need a test on the legacy and new yaml formats here mock({ '/ws/openfn.yaml': jsonToYaml({ - name: 'some-project-name', project: { + id: 'project-1', uuid: '1234', - name: 'some-project-name', }, - formats: { - openfn: 'yaml', - project: 'yaml', - workflow: 'yaml', - custom: true, // Note tha this will be excluded + workspace: { + formats: { + openfn: 'yaml', + project: 'yaml', + workflow: 'yaml', + custom: true, // Note that this will be excluded + }, + // deliberately exclude dirs + custom: true, }, - // deliberately exclude dirs - custom: true, }), '/ws/.projects/staging@app.openfn.org.yaml': jsonToYaml({ - id: 'some-id', - name: 'some-project-name', + name: 'Project 1', + id: '', workflows: [ { - name: 'simple-workflow', - id: 'wf-id', + name: 'simple-workflow', // TODO clean up + id: 'wf-id', // TODO clean up jobs: [ { name: 'Transform data to FHIR standard', @@ -97,18 +98,18 @@ test('workspace-list: list projects in the workspace', (t) => { const ws = new Workspace('/ws'); t.is(ws.list().length, 1); t.deepEqual( - ws.list().map((w) => w.name), - ['some-project-name'] + ws.list().map((w) => w.id), + ['project-1'] ); }); test('workspace-get: get projects in the workspace', (t) => { const ws = new Workspace('/ws'); - const found = ws.get('some-project-name'); + const found = ws.get('project-1'); t.truthy(found); t.is(found?.workflows.length, 2); t.deepEqual( - found?.workflows.map((w) => w.name), + found?.workflows.map((w) => w.id), ['simple-workflow', 'another-workflow'] ); }); @@ -131,8 +132,8 @@ test('load config', (t) => { test('load project meta', (t) => { const ws = new Workspace('/ws'); - t.deepEqual(ws.projectMeta, { + t.deepEqual(ws.activeProject, { uuid: '1234', - name: 'some-project-name', + id: 'project-1', }); }); diff --git a/packages/ws-worker/CHANGELOG.md b/packages/ws-worker/CHANGELOG.md index df21cdc2e..a9a83669e 100644 --- a/packages/ws-worker/CHANGELOG.md +++ b/packages/ws-worker/CHANGELOG.md @@ -1,5 +1,14 @@ # ws-worker +## 1.19.2 + +### Patch Changes + +- Updated dependencies [8632629] +- Updated dependencies [0e5b7d4] +- Updated dependencies [7e12e4b] + - @openfn/engine-multi@1.8.2 + ## 1.19.1 ### Patch Changes diff --git a/packages/ws-worker/README.md b/packages/ws-worker/README.md index a417cb856..198f435dc 100644 --- a/packages/ws-worker/README.md +++ b/packages/ws-worker/README.md @@ -54,7 +54,7 @@ This will wrap a real runtime engine into the server. It will rebuild when the W ### Disabling auto-fetch -When working in dev it is convinient to disable the workloop. To switch it off, run: +When working in dev it is convenient to disable the workloop. To switch it off, run: ``` pnpm start --no-loop diff --git a/packages/ws-worker/package.json b/packages/ws-worker/package.json index 6edd97f7a..d66917a02 100644 --- a/packages/ws-worker/package.json +++ b/packages/ws-worker/package.json @@ -1,6 +1,6 @@ { "name": "@openfn/ws-worker", - "version": "1.19.1", + "version": "1.19.2", "description": "A Websocket Worker to connect Lightning to a Runtime Engine", "main": "dist/index.js", "type": "module", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index e3783ede4..b2a21b30c 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -628,6 +628,9 @@ importers: glob: specifier: ^11.0.2 version: 11.0.3 + human-id: + specifier: ^4.1.1 + version: 4.1.1 lodash: specifier: ^4.17.21 version: 4.17.21 @@ -3733,6 +3736,7 @@ packages: keygrip@1.1.0: resolution: {integrity: sha512-iYSchDJ+liQ8iwbSI2QqsQOvqv58eJCEanyJPJi+Khyu8smkcKSFUCbPwzFcL7YVtZ6eONjqRX/38caJ7QjRAQ==} engines: {node: '>= 0.6'} + deprecated: Package no longer supported. Contact Support at https://www.npmjs.com/support for more info. koa-bodyparser@4.4.1: resolution: {integrity: sha512-kBH3IYPMb+iAXnrxIhXnW+gXV8OTzCu8VPDqvcDHW9SQrbkHmqPQtiZwrltNmSq6/lpipHnT7k7PsjlVD7kK0w==}