Skip to content

Runtime: wait for notify events to send #1390

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
josephjclark merged 2 commits into from
Apr 24, 2026
Merged

Runtime: wait for notify events to send #1390

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
944ed19
runtime: ensure notify calls are awaited
josephjclark Apr 24, 2026
8d2954d
engine: return result of publish
josephjclark Apr 24, 2026
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
5 changes: 5 additions & 0 deletions .changeset/hungry-bees-call.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
'@openfn/runtime': patch
---

make sure that notify are properly awaited
5 changes: 5 additions & 0 deletions .changeset/strict-windows-think.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
'@openfn/engine-multi': patch
---

Make publish() async
2 changes: 1 addition & 1 deletion packages/engine-multi/src/worker/thread/run.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ register({
notify: (name: NotifyEvents, payload: any) => {
// @ts-ignore
const mappedPayload = eventMap[name]?.(payload) ?? payload;
publish(`worker:${name}`, {
return publish(`worker:${name}`, {
workflowId: plan.id,
...mappedPayload,
});
Expand Down
10 changes: 5 additions & 5 deletions packages/runtime/src/execute/step.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -147,7 +147,7 @@ const executeStep = async (
const jobName = job.name || job.id;

// The notify events only apply to jobs - not steps - so names don't need to be changed here
notify(NOTIFY_INIT_START, { jobId });
await notify(NOTIFY_INIT_START, { jobId });

// lazy load config and state
const configuration = await loadCredentials(
Expand All @@ -166,7 +166,7 @@ const executeStep = async (
plan.workflow?.credentials
);

notify(NOTIFY_INIT_COMPLETE, {
await notify(NOTIFY_INIT_COMPLETE, {
jobId,
duration: Date.now() - duration,
});
Expand All @@ -182,7 +182,7 @@ const executeStep = async (
const startTime = Date.now();
try {
// TODO include the upstream job?
notify(NOTIFY_JOB_START, { jobId });
await notify(NOTIFY_JOB_START, { jobId });
result = await executeExpression(
ctx,
job.expression!,
Expand Down Expand Up @@ -212,7 +212,7 @@ const executeStep = async (
report(state, jobId, error);

next = calculateNext(step, result, logger);
notify(NOTIFY_JOB_ERROR, {
await notify(NOTIFY_JOB_ERROR, {
duration: Date.now() - startTime,
error,
state,
Expand Down Expand Up @@ -267,7 +267,7 @@ const executeStep = async (
}

next = calculateNext(step, result, logger);
notify(NOTIFY_JOB_COMPLETE, {
await notify(NOTIFY_JOB_COMPLETE, {
duration: Date.now() - duration,
state: result,
jobId,
Expand Down
Loading