From d46552a529b449c3c05ff9f644b3dbdf2c4495db Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Tue, 4 Aug 2026 11:39:17 +0100 Subject: [PATCH 1/4] engine: feed globals option through to the runtime --- .changeset/pink-bobcats-study.md | 5 ++++ packages/engine-multi/src/api/execute.ts | 1 + packages/engine-multi/src/types.ts | 2 ++ .../engine-multi/src/worker/thread/run.ts | 3 +++ .../engine-multi/test/api/execute.test.ts | 26 +++++++++++++++++++ 5 files changed, 37 insertions(+) create mode 100644 .changeset/pink-bobcats-study.md diff --git a/.changeset/pink-bobcats-study.md b/.changeset/pink-bobcats-study.md new file mode 100644 index 000000000..4584133b7 --- /dev/null +++ b/.changeset/pink-bobcats-study.md @@ -0,0 +1,5 @@ +--- +'@openfn/engine-multi': patch +--- + +Allow global scope context to be fed through to individual runs diff --git a/packages/engine-multi/src/api/execute.ts b/packages/engine-multi/src/api/execute.ts index 57f5fa561..3f7f1b440 100644 --- a/packages/engine-multi/src/api/execute.ts +++ b/packages/engine-multi/src/api/execute.ts @@ -40,6 +40,7 @@ const execute = async (context: ExecutionContext) => { const whitelist = options.whitelist?.map((w) => w.toString()); const runOptions = { + globals: options.globals, statePropsToRemove: options.statePropsToRemove, whitelist, jobLogLevel: options.jobLogLevel, diff --git a/packages/engine-multi/src/types.ts b/packages/engine-multi/src/types.ts index 5320857e8..f141571e6 100644 --- a/packages/engine-multi/src/types.ts +++ b/packages/engine-multi/src/types.ts @@ -54,6 +54,8 @@ export type ExecuteOptions = { runTimeoutMs?: number; sanitize?: SanitizePolicies; jobLogLevel?: string; + // inject globals into the environment + globals?: any; }; export type ExecutionContextOptions = ExecuteOptions & EngineOptions; diff --git a/packages/engine-multi/src/worker/thread/run.ts b/packages/engine-multi/src/worker/thread/run.ts index 8d6103192..df6f7c3e1 100644 --- a/packages/engine-multi/src/worker/thread/run.ts +++ b/packages/engine-multi/src/worker/thread/run.ts @@ -21,6 +21,7 @@ export type RunOptions = { profile?: boolean; profilePollInterval?: number; stateLimitMb?: number; + globals?: any; }; const eventMap = { @@ -41,6 +42,7 @@ register({ profile, profilePollInterval, stateLimitMb, + globals, } = runOptions; const { logger, jobLogger, adaptorLogger } = createLoggers( plan.id!, @@ -74,6 +76,7 @@ register({ profilePollInterval, statePropsToRemove, stateLimitMb, + globals, callbacks: { // TODO: this won't actually work across the worker boundary // For now I am preloading credentials diff --git a/packages/engine-multi/test/api/execute.test.ts b/packages/engine-multi/test/api/execute.test.ts index 8ec2a9ea6..a8d50310b 100644 --- a/packages/engine-multi/test/api/execute.test.ts +++ b/packages/engine-multi/test/api/execute.test.ts @@ -403,3 +403,29 @@ test.serial('should forward the jobLogLevel option', async (t) => { t.truthy(passedOptions); t.deepEqual(passedOptions.jobLogLevel, 'none'); }); + +test.serial('should forward the globals option', async (t) => { + let passedOptions: any; + + const state = { + id: 'x', + plan, + } as WorkflowState; + + const context = createContext({ + state, + options: { + ...options, + globals: { x: 22 }, + }, + }); + // @ts-ignore + context.callWorker = (_command, args) => { + passedOptions = args[2]; + }; + + await execute(context); + + t.truthy(passedOptions); + t.deepEqual(passedOptions.globals, { x: 22 }); +}); From 84c12ea3410cbdb3be162fe09c088a54a6489224 Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Tue, 4 Aug 2026 12:28:29 +0100 Subject: [PATCH 2/4] set meta in worker --- .changeset/late-games-deny.md | 5 +++ .../src/util/convert-lightning-plan.ts | 8 ++++ .../test/util/convert-lightning-plan.test.ts | 38 ++++++++++++++++--- 3 files changed, 45 insertions(+), 6 deletions(-) create mode 100644 .changeset/late-games-deny.md diff --git a/.changeset/late-games-deny.md b/.changeset/late-games-deny.md new file mode 100644 index 000000000..d19c2f664 --- /dev/null +++ b/.changeset/late-games-deny.md @@ -0,0 +1,5 @@ +--- +'@openfn/ws-worker': minor +--- + +Pass a meta object to each run diff --git a/packages/ws-worker/src/util/convert-lightning-plan.ts b/packages/ws-worker/src/util/convert-lightning-plan.ts index f9fe2180b..2b00a2473 100644 --- a/packages/ws-worker/src/util/convert-lightning-plan.ts +++ b/packages/ws-worker/src/util/convert-lightning-plan.ts @@ -133,6 +133,14 @@ export default ( // But some need to get passed down into the engine's options const engineOpts: WorkerRunOptions = {}; + + engineOpts.globals = { + meta: { + runId: run.id, + startTime: Date.now(), + }, + }; + if (run.options) { if ('run_timeout_ms' in run.options) { engineOpts.runTimeoutMs = run.options.run_timeout_ms; diff --git a/packages/ws-worker/test/util/convert-lightning-plan.test.ts b/packages/ws-worker/test/util/convert-lightning-plan.test.ts index 224855125..2989480ab 100644 --- a/packages/ws-worker/test/util/convert-lightning-plan.test.ts +++ b/packages/ws-worker/test/util/convert-lightning-plan.test.ts @@ -55,6 +55,15 @@ const createJob = (props = {}) => ({ ...props, }); +// Pulls the generated globals out of the options object and checks them, +// returning the remainder so callers can deepEqual the rest as before +const checkGlobals = (t: any, options: any, runId: string) => { + const { globals, ...rest } = options; + t.is(globals.meta.runId, runId); + t.is(typeof globals.meta.startTime, 'number'); + return rest; +}; + test('convert a single job', (t) => { const run: Partial = { id: 'w', @@ -115,7 +124,7 @@ test('convert a single job with options', (t) => { steps: [createJob()], }, }); - t.deepEqual(options, { + t.deepEqual(checkGlobals(t, options, 'w'), { runTimeoutMs: 10, memoryLimitMb: 500, payloadLimitMb: 20, @@ -136,7 +145,7 @@ test('convert a single job with state_limit_mb', (t) => { }; const { options } = convertPlan(run as LightningPlan); - t.deepEqual(options, { + t.deepEqual(checkGlobals(t, options, 'w'), { memoryLimitMb: 500, stateLimitMb: 50, }); @@ -165,7 +174,7 @@ test('convert a single job with log_payload_limit_mb', (t) => { steps: [createJob()], }, }); - t.deepEqual(options, { + t.deepEqual(checkGlobals(t, options, 'w'), { runTimeoutMs: 10, memoryLimitMb: 500, payloadLimitMb: 20, @@ -192,7 +201,7 @@ test('convert a single job with data', (t) => { steps: [createJob({ state: { data: { x: 22 } } })], }, }); - t.deepEqual(options, {}); + t.deepEqual(checkGlobals(t, options, 'w'), {}); }); test('Accept a partial run object', (t) => { @@ -208,7 +217,7 @@ test('Accept a partial run object', (t) => { steps: [], }, }); - t.deepEqual(options, {}); + t.deepEqual(checkGlobals(t, options, 'w'), {}); }); test('handle dataclip_id as input', (t) => { @@ -241,7 +250,7 @@ test('handle output_dataclip as options', (t) => { }, }; const { options } = convertPlan(run as LightningPlan); - t.deepEqual(options, { + t.deepEqual(checkGlobals(t, options, 'w'), { outputDataclips: false, }); }); @@ -747,3 +756,20 @@ test("ignore globals when it isn't a string", (t) => { const { plan } = convertPlan(run as LightningPlan); t.deepEqual(plan.workflow.globals, undefined); }); + +test('sets runId and startTime on the engine options globals meta', (t) => { + const run: Partial = { + id: 'some-run-id', + jobs: [createNode()], + triggers: [], + edges: [], + }; + + const before = Date.now(); + const { options } = convertPlan(run as LightningPlan); + const after = Date.now(); + + t.is(options.globals!.meta.runId, 'some-run-id'); + t.true(options.globals!.meta.startTime >= before); + t.true(options.globals!.meta.startTime <= after); +}); From 6724dd87e25ba7c2ec06b6b300bc3f5e5f1465c5 Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Tue, 4 Aug 2026 14:31:03 +0100 Subject: [PATCH 3/4] fix --- packages/engine-multi/src/engine.ts | 1 + packages/runtime/src/runtime.ts | 1 - packages/ws-worker/test/worker.test.ts | 2 +- 3 files changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/engine-multi/src/engine.ts b/packages/engine-multi/src/engine.ts index 38467b97c..0c58da00e 100644 --- a/packages/engine-multi/src/engine.ts +++ b/packages/engine-multi/src/engine.ts @@ -178,6 +178,7 @@ const createEngine = async ( payloadLimitMb: opts.payloadLimitMb ?? defaultPayloadLimit, logPayloadLimitMb: opts.logPayloadLimitMb ?? defaultLogPayloadLimit, jobLogLevel: opts.jobLogLevel, + globals: opts.globals, profile: defaultProfile, profilePollInterval: defaultProfilePollInterval, }, diff --git a/packages/runtime/src/runtime.ts b/packages/runtime/src/runtime.ts index f0072a404..587eaa183 100644 --- a/packages/runtime/src/runtime.ts +++ b/packages/runtime/src/runtime.ts @@ -87,7 +87,6 @@ const run = ( } const logger = opts.logger || defaultLogger; - if (typeof xplan === 'string') { xplan = loadPlanFromString( xplan, diff --git a/packages/ws-worker/test/worker.test.ts b/packages/ws-worker/test/worker.test.ts index 6c845a2cf..86b20cf02 100644 --- a/packages/ws-worker/test/worker.test.ts +++ b/packages/ws-worker/test/worker.test.ts @@ -43,7 +43,7 @@ const execute = async (plan: ExecutionPlan, input = {}, options = {}) => [RUN_START]: async () => true, [STEP_START]: async () => true, [RUN_LOG]: async (_evt) => { - //console.log(evt.source, evt.message) + console.log(evt.source, evt.message); return true; }, [STEP_COMPLETE]: async () => true, From 339989b9d2b1da80f8f032bba950753c74fbc176 Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Tue, 4 Aug 2026 15:04:50 +0100 Subject: [PATCH 4/4] types --- packages/ws-worker/test/worker.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/ws-worker/test/worker.test.ts b/packages/ws-worker/test/worker.test.ts index 86b20cf02..3cdddbab1 100644 --- a/packages/ws-worker/test/worker.test.ts +++ b/packages/ws-worker/test/worker.test.ts @@ -42,7 +42,7 @@ const execute = async (plan: ExecutionPlan, input = {}, options = {}) => const channel = mockChannel({ [RUN_START]: async () => true, [STEP_START]: async () => true, - [RUN_LOG]: async (_evt) => { + [RUN_LOG]: async (evt) => { console.log(evt.source, evt.message); return true; },