From b5a9111d07646e8bfbb49edbfe8952b457d3f3ec Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 11 Sep 2024 23:52:02 +0200 Subject: [PATCH] fix: improve runFlowAsync and run_flow_async default behavior + time formatting of scheduled for --- backend/windmill-common/src/variables.rs | 4 ++-- python-client/wmill/wmill/client.py | 18 ++++++++++++++---- typescript-client/client.ts | 16 +++++++++------- 3 files changed, 25 insertions(+), 13 deletions(-) diff --git a/backend/windmill-common/src/variables.rs b/backend/windmill-common/src/variables.rs index 1081c76809..26cbca7f08 100644 --- a/backend/windmill-common/src/variables.rs +++ b/backend/windmill-common/src/variables.rs @@ -6,7 +6,7 @@ * LICENSE-AGPL for a copy of the license. */ -use chrono::Utc; +use chrono::{SecondsFormat, Utc}; use magic_crypt::{MagicCrypt256, MagicCryptError, MagicCryptTrait}; use serde::{Deserialize, Serialize}; @@ -263,7 +263,7 @@ pub async fn get_reserved_variables( ContextualVariable { name: WM_SCHEDULED_FOR.to_string(), value: scheduled_for - .map(|ts| ts.to_string()) + .map(|ts| ts.to_rfc3339_opts(SecondsFormat::Secs, true)) .unwrap_or_else(|| "".to_string()), description: "date-time in UTC (e.g: 2014-11-28T12:45:59.324310806Z) of when the job was scheduled".to_string(), is_custom: false, diff --git a/python-client/wmill/wmill/client.py b/python-client/wmill/wmill/client.py index d794980d64..f68ee13dcc 100644 --- a/python-client/wmill/wmill/client.py +++ b/python-client/wmill/wmill/client.py @@ -108,14 +108,19 @@ class Windmill: path: str, args: dict = None, scheduled_in_secs: int = None, + # can only be set to false if this the job will be fully await and not concurrent with any other job + # as otherwise the child flow and its own child will store their state in the parent job which will + # lead to incorrectness and failures + do_not_track_in_parent: bool = True, ) -> str: """Create a flow job and return its job id.""" args = args or {} params = {"scheduled_in_secs": scheduled_in_secs} if scheduled_in_secs else {} - if os.environ.get("WM_JOB_ID"): - params["parent_job"] = os.environ.get("WM_JOB_ID") - if os.environ.get("WM_ROOT_FLOW_JOB_ID"): - params["root_job"] = os.environ.get("WM_ROOT_FLOW_JOB_ID") + if not do_not_track_in_parent: + if os.environ.get("WM_JOB_ID"): + params["parent_job"] = os.environ.get("WM_JOB_ID") + if os.environ.get("WM_ROOT_FLOW_JOB_ID"): + params["root_job"] = os.environ.get("WM_ROOT_FLOW_JOB_ID") if path: endpoint = f"/w/{self.workspace}/jobs/run/f/{path}" else: @@ -655,11 +660,16 @@ def run_flow_async( path: str, args: Dict[str, Any] = None, scheduled_in_secs: int = None, + # can only be set to false if this the job will be fully await and not concurrent with any other job + # as otherwise the child flow and its own child will store their state in the parent job which will + # lead to incorrectness and failures + do_not_track_in_parent: bool = True, ) -> str: return _client.run_flow_async( path=path, args=args, scheduled_in_secs=scheduled_in_secs, + do_not_track_in_parent=do_not_track_in_parent, ) diff --git a/typescript-client/client.ts b/typescript-client/client.ts index 0c7c672f78..1ebda79548 100644 --- a/typescript-client/client.ts +++ b/typescript-client/client.ts @@ -273,7 +273,10 @@ export async function runFlowAsync( path: string | null, args: Record | null, scheduledInSeconds: number | null = null, - flowOutlivesParent: boolean = true + // can only be set to false if this the job will be fully await and not concurrent with any other job + // as otherwise the child flow and its own child will store their state in the parent job which will + // lead to incorrectness and failures + doNotTrackInParent: boolean = true ): Promise { // Create a script job and return its job id. @@ -284,16 +287,15 @@ export async function runFlowAsync( params["scheduled_in_secs"] = scheduledInSeconds; } - if (!flowOutlivesParent) { + if (!doNotTrackInParent) { let parentJobId = getEnv("WM_JOB_ID"); if (parentJobId !== undefined) { params["parent_job"] = parentJobId; } - } - - let rootJobId = getEnv("WM_ROOT_FLOW_JOB_ID"); - if (rootJobId != undefined && rootJobId != "") { - params["root_job"] = rootJobId; + let rootJobId = getEnv("WM_ROOT_FLOW_JOB_ID"); + if (rootJobId != undefined && rootJobId != "") { + params["root_job"] = rootJobId; + } } let endpoint: string;