|
@@ -396,10 +396,6 @@ Result<ExecutionResult> WorkflowEngine::execute(const Workflow& workflow,
|
|
|
result.execution_id, workflow.id);
|
|
result.execution_id, workflow.id);
|
|
|
|
|
|
|
|
try {
|
|
try {
|
|
|
- // A Workflow Output node's value, when the workflow named one.
|
|
|
|
|
- nlohmann::json explicit_output;
|
|
|
|
|
- bool has_explicit_output = false;
|
|
|
|
|
-
|
|
|
|
|
// Build execution graph
|
|
// Build execution graph
|
|
|
std::unordered_map<std::string, std::vector<std::string>> dependencies;
|
|
std::unordered_map<std::string, std::vector<std::string>> dependencies;
|
|
|
std::unordered_map<std::string, std::vector<std::string>> dependents;
|
|
std::unordered_map<std::string, std::vector<std::string>> dependents;
|
|
@@ -679,6 +675,11 @@ Result<ExecutionResult> WorkflowEngine::execute(const Workflow& workflow,
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
NodeExecutionResult node_result;
|
|
NodeExecutionResult node_result;
|
|
|
|
|
+ // What this node's markers asked for. Set in both branches below,
|
|
|
|
|
+ // because a cached output carries the same markers a fresh one does
|
|
|
|
|
+ // - a run replayed from cache has to stop where the original did.
|
|
|
|
|
+ MarkerOutcome marker_outcome = MarkerOutcome::Continue;
|
|
|
|
|
+
|
|
|
if (use_cached) {
|
|
if (use_cached) {
|
|
|
// Use cached output instead of executing
|
|
// Use cached output instead of executing
|
|
|
node_result.node_id = node_id;
|
|
node_result.node_id = node_id;
|
|
@@ -689,7 +690,8 @@ Result<ExecutionResult> WorkflowEngine::execute(const Workflow& workflow,
|
|
|
node_result.finished_at = node_result.started_at;
|
|
node_result.finished_at = node_result.started_at;
|
|
|
node_result.from_cache = true;
|
|
node_result.from_cache = true;
|
|
|
|
|
|
|
|
- result.node_results[node_id] = node_result;
|
|
|
|
|
|
|
+ marker_outcome = applyNodeMarkers(node_result, node_id, node_id, result,
|
|
|
|
|
+ nlohmann::json::object(), {}, -1);
|
|
|
|
|
|
|
|
if (callback) {
|
|
if (callback) {
|
|
|
callback("node.completed", {
|
|
callback("node.completed", {
|
|
@@ -744,26 +746,8 @@ Result<ExecutionResult> WorkflowEngine::execute(const Workflow& workflow,
|
|
|
node_def_for_defaults);
|
|
node_def_for_defaults);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- // A node asking to run another workflow has it run here: the
|
|
|
|
|
- // node itself has no way to reach the engine, and this is the
|
|
|
|
|
- // point where the result can still become its output.
|
|
|
|
|
- if (node_result.status == NodeStatus::Completed &&
|
|
|
|
|
- node_result.output.contains("_callWorkflow")) {
|
|
|
|
|
- runSubWorkflow(node_result, result.call_depth, result.execution_id);
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- // Record what was applied. Once a node's settings can come from
|
|
|
|
|
- // elsewhere, its stored config no longer says what it ran with,
|
|
|
|
|
- // and this is the only place that closes that gap.
|
|
|
|
|
- if (!overlay.empty() && node_result.status == NodeStatus::Completed) {
|
|
|
|
|
- node_result.output["_appliedConfig"] = overlay;
|
|
|
|
|
- if (!overlay_ignored.empty()) {
|
|
|
|
|
- node_result.output["_ignoredConfigKeys"] = overlay_ignored;
|
|
|
|
|
- LOG_WARN("Node {} was given {} setting(s) it does not have",
|
|
|
|
|
- node_id, overlay_ignored.size());
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
- result.node_results[node_id] = node_result;
|
|
|
|
|
|
|
+ marker_outcome = applyNodeMarkers(node_result, node_id, node_id, result,
|
|
|
|
|
+ overlay, overlay_ignored, -1);
|
|
|
|
|
|
|
|
if (callback) {
|
|
if (callback) {
|
|
|
nlohmann::json event_data = {
|
|
nlohmann::json event_data = {
|
|
@@ -779,79 +763,7 @@ Result<ExecutionResult> WorkflowEngine::execute(const Workflow& workflow,
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- // Any node may declare the HTTP response, not just the last one to
|
|
|
|
|
- // run, so appending a node to a workflow cannot silently change
|
|
|
|
|
- // what its webhook returns.
|
|
|
|
|
- if (node_result.status == NodeStatus::Completed &&
|
|
|
|
|
- node_result.output.contains("_webhookResponse")) {
|
|
|
|
|
- if (!result.webhook_response.is_null()) {
|
|
|
|
|
- LOG_WARN("Node {} overrides a webhook response already set by an earlier node",
|
|
|
|
|
- node_id);
|
|
|
|
|
- }
|
|
|
|
|
- result.webhook_response = node_result.output["_webhookResponse"];
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- // A workflow can name its own result instead of leaving it to be
|
|
|
|
|
- // whatever the last node happened to produce. It matters most when
|
|
|
|
|
- // the workflow is called by another one, where "the last node" is
|
|
|
|
|
- // ambiguous in anything that branches.
|
|
|
|
|
- if (node_result.status == NodeStatus::Completed &&
|
|
|
|
|
- node_result.output.contains("_workflowOutput")) {
|
|
|
|
|
- explicit_output = node_result.output["_workflowOutput"];
|
|
|
|
|
- has_explicit_output = true;
|
|
|
|
|
- auto& stored_node = result.node_results[node_id];
|
|
|
|
|
- stored_node.output.erase("_workflowOutput");
|
|
|
|
|
- stored_node.output["output"] = explicit_output;
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- // A node asking to stop ends the walk without failing the run. The
|
|
|
|
|
- // distinction matters for anything that polls: a scheduled workflow
|
|
|
|
|
- // that finds nothing to do has completed successfully, and marking
|
|
|
|
|
- // it failed every five minutes buries real failures in noise.
|
|
|
|
|
- if (node_result.status == NodeStatus::Completed &&
|
|
|
|
|
- node_result.output.contains("_stop")) {
|
|
|
|
|
- const auto& stop = node_result.output["_stop"];
|
|
|
|
|
-
|
|
|
|
|
- result.stop_requested = true;
|
|
|
|
|
- result.stopped_node_id = node_id;
|
|
|
|
|
- result.stop_reason = stop.value("reason", "");
|
|
|
|
|
-
|
|
|
|
|
- // The marker is engine plumbing. What the node reports is that
|
|
|
|
|
- // it stopped and why, not the mechanism that carried it.
|
|
|
|
|
- auto& stored_node = result.node_results[node_id];
|
|
|
|
|
- stored_node.output.erase("_stop");
|
|
|
|
|
- stored_node.output["stopped"] = true;
|
|
|
|
|
- stored_node.output["reason"] = result.stop_reason;
|
|
|
|
|
-
|
|
|
|
|
- LOG_INFO("Execution {} stopped at node {}: {}", result.execution_id, node_id,
|
|
|
|
|
- result.stop_reason.empty() ? "no reason given" : result.stop_reason);
|
|
|
|
|
- break;
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- // A node asking to pause ends this pass. The execution is stored as
|
|
|
|
|
- // Waiting with everything computed so far, and a later resume picks
|
|
|
|
|
- // it up from here rather than starting again.
|
|
|
|
|
- if (node_result.status == NodeStatus::Completed &&
|
|
|
|
|
- node_result.output.contains("_pause")) {
|
|
|
|
|
- const auto& pause = node_result.output["_pause"];
|
|
|
|
|
-
|
|
|
|
|
- result.status = ExecutionStatus::Waiting;
|
|
|
|
|
- result.paused_node_id = node_id;
|
|
|
|
|
- result.pause_token = pause.value("token", "");
|
|
|
|
|
- result.pause_expires_at = pause.value("expiresAt", static_cast<int64_t>(0));
|
|
|
|
|
-
|
|
|
|
|
- // The marker is engine plumbing. What is stored is the request a
|
|
|
|
|
- // person has to answer, not the mechanism that carried it.
|
|
|
|
|
- auto& stored_node = result.node_results[node_id];
|
|
|
|
|
- stored_node.output.erase("_pause");
|
|
|
|
|
-
|
|
|
|
|
- // No callback here - the generic terminal callback below (after
|
|
|
|
|
- // storeExecution) already emits "execution.waiting" once this
|
|
|
|
|
- // pass ends, and it carries nodeId/expiresAt too. Emitting here
|
|
|
|
|
- // as well produced the same event twice with two different
|
|
|
|
|
- // payload shapes.
|
|
|
|
|
-
|
|
|
|
|
- LOG_INFO("Execution {} paused at node {}", result.execution_id, node_id);
|
|
|
|
|
|
|
+ if (marker_outcome == MarkerOutcome::Stop || marker_outcome == MarkerOutcome::Pause) {
|
|
|
break;
|
|
break;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -929,10 +841,10 @@ Result<ExecutionResult> WorkflowEngine::execute(const Workflow& workflow,
|
|
|
if (result.status == ExecutionStatus::Running) {
|
|
if (result.status == ExecutionStatus::Running) {
|
|
|
result.status = ExecutionStatus::Completed;
|
|
result.status = ExecutionStatus::Completed;
|
|
|
|
|
|
|
|
- if (has_explicit_output) {
|
|
|
|
|
|
|
+ if (result.has_explicit_output) {
|
|
|
// A Workflow Output node said what this workflow returns, which
|
|
// A Workflow Output node said what this workflow returns, which
|
|
|
// beats guessing from whichever node ran last.
|
|
// beats guessing from whichever node ran last.
|
|
|
- result.final_output = explicit_output;
|
|
|
|
|
|
|
+ result.final_output = result.explicit_output;
|
|
|
} else if (result.stop_requested) {
|
|
} else if (result.stop_requested) {
|
|
|
// The last node in the order never ran - the walk ended early on
|
|
// The last node in the order never ran - the walk ended early on
|
|
|
// purpose - so the useful final output is the node that stopped
|
|
// purpose - so the useful final output is the node that stopped
|
|
@@ -1697,6 +1609,142 @@ static std::vector<std::string> applyConfigOverlay(nlohmann::json& config,
|
|
|
// the failure would look like a hang rather than a mistake in the workflow.
|
|
// the failure would look like a hang rather than a mistake in the workflow.
|
|
|
static constexpr int kMaxCallDepth = 5;
|
|
static constexpr int kMaxCallDepth = 5;
|
|
|
|
|
|
|
|
|
|
+// See the header for why this is one function rather than a block in each walk.
|
|
|
|
|
+// iteration is -1 outside a loop body, and the item index inside one.
|
|
|
|
|
+WorkflowEngine::MarkerOutcome WorkflowEngine::applyNodeMarkers(
|
|
|
|
|
+ NodeExecutionResult& node_result,
|
|
|
|
|
+ const std::string& node_id,
|
|
|
|
|
+ const std::string& result_key,
|
|
|
|
|
+ ExecutionResult& result,
|
|
|
|
|
+ const nlohmann::json& overlay,
|
|
|
|
|
+ const std::vector<std::string>& overlay_ignored,
|
|
|
|
|
+ int iteration) {
|
|
|
|
|
+
|
|
|
|
|
+ const bool in_loop = iteration >= 0;
|
|
|
|
|
+ const std::string where = in_loop
|
|
|
|
|
+ ? (" during loop iteration " + std::to_string(iteration))
|
|
|
|
|
+ : std::string();
|
|
|
|
|
+
|
|
|
|
|
+ if (node_result.status != NodeStatus::Completed) {
|
|
|
|
|
+ // A node that failed or was skipped asked for nothing. Its result is
|
|
|
|
|
+ // still stored, so the caller can see what happened.
|
|
|
|
|
+ result.node_results[result_key] = node_result;
|
|
|
|
|
+ return MarkerOutcome::Continue;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // A node asking to run another workflow has it run here: the node itself
|
|
|
|
|
+ // has no way to reach the engine, and this is the point where the result
|
|
|
|
|
+ // can still become its output.
|
|
|
|
|
+ if (node_result.output.contains("_callWorkflow")) {
|
|
|
|
|
+ runSubWorkflow(node_result, result.call_depth, result.execution_id);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // A pause cannot work inside a loop: there is no way to resume one item of
|
|
|
|
|
+ // an iteration. Better to say so than to park a run nobody can restart.
|
|
|
|
|
+ if (in_loop && node_result.status == NodeStatus::Completed &&
|
|
|
|
|
+ node_result.output.contains("_pause")) {
|
|
|
|
|
+ node_result.status = NodeStatus::Failed;
|
|
|
|
|
+ node_result.error = "A node cannot pause inside a Loop body, because a "
|
|
|
|
|
+ "paused loop iteration cannot be resumed. Collect the "
|
|
|
|
|
+ "items first, approve once, then loop";
|
|
|
|
|
+ node_result.output.erase("_pause");
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Record what was applied. Once a node's settings can come from elsewhere,
|
|
|
|
|
+ // its stored config no longer says what it ran with, and this is the only
|
|
|
|
|
+ // place that closes that gap.
|
|
|
|
|
+ if (!overlay.empty() && node_result.status == NodeStatus::Completed) {
|
|
|
|
|
+ node_result.output["_appliedConfig"] = overlay;
|
|
|
|
|
+ if (!overlay_ignored.empty()) {
|
|
|
|
|
+ node_result.output["_ignoredConfigKeys"] = overlay_ignored;
|
|
|
|
|
+ LOG_WARN("Node {} was given {} setting(s) it does not have",
|
|
|
|
|
+ node_id, overlay_ignored.size());
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ result.node_results[result_key] = node_result;
|
|
|
|
|
+
|
|
|
|
|
+ if (node_result.status != NodeStatus::Completed) {
|
|
|
|
|
+ // Only reachable when the pause rejection above failed it.
|
|
|
|
|
+ return MarkerOutcome::Continue;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Any node may declare the HTTP response, not just the last one to run, so
|
|
|
|
|
+ // appending a node to a workflow cannot silently change what its webhook
|
|
|
|
|
+ // returns.
|
|
|
|
|
+ if (node_result.output.contains("_webhookResponse")) {
|
|
|
|
|
+ if (!result.webhook_response.is_null()) {
|
|
|
|
|
+ LOG_WARN("Node {} overrides a webhook response already set by an earlier node",
|
|
|
|
|
+ node_id);
|
|
|
|
|
+ }
|
|
|
|
|
+ result.webhook_response = node_result.output["_webhookResponse"];
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // A workflow can name its own result instead of leaving it to be whatever
|
|
|
|
|
+ // the last node happened to produce. It matters most when the workflow is
|
|
|
|
|
+ // called by another one, where "the last node" is ambiguous in anything
|
|
|
|
|
+ // that branches - and a loop body is exactly such a case, which is why this
|
|
|
|
|
+ // has to work there too.
|
|
|
|
|
+ if (node_result.output.contains("_workflowOutput")) {
|
|
|
|
|
+ result.explicit_output = node_result.output["_workflowOutput"];
|
|
|
|
|
+ result.has_explicit_output = true;
|
|
|
|
|
+ auto& stored = result.node_results[result_key];
|
|
|
|
|
+ stored.output.erase("_workflowOutput");
|
|
|
|
|
+ stored.output["output"] = result.explicit_output;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // A node asking to stop ends the walk without failing the run. The
|
|
|
|
|
+ // distinction matters for anything that polls: a scheduled workflow that
|
|
|
|
|
+ // finds nothing to do has completed successfully, and marking it failed
|
|
|
|
|
+ // every five minutes buries real failures in noise. Raised inside a loop it
|
|
|
|
|
+ // ends the whole run, not just the iteration - "stop the workflow" would be
|
|
|
|
|
+ // a strange thing to mean per-item.
|
|
|
|
|
+ if (node_result.output.contains("_stop")) {
|
|
|
|
|
+ const auto& stop = node_result.output["_stop"];
|
|
|
|
|
+
|
|
|
|
|
+ result.stop_requested = true;
|
|
|
|
|
+ result.stopped_node_id = node_id;
|
|
|
|
|
+ result.stop_reason = stop.value("reason", "");
|
|
|
|
|
+
|
|
|
|
|
+ // The marker is engine plumbing. What the node reports is that it
|
|
|
|
|
+ // stopped and why, not the mechanism that carried it.
|
|
|
|
|
+ auto& stored = result.node_results[result_key];
|
|
|
|
|
+ stored.output.erase("_stop");
|
|
|
|
|
+ stored.output["stopped"] = true;
|
|
|
|
|
+ stored.output["reason"] = result.stop_reason;
|
|
|
|
|
+
|
|
|
|
|
+ LOG_INFO("Execution {} stopped at node {}{}: {}", result.execution_id, node_id, where,
|
|
|
|
|
+ result.stop_reason.empty() ? "no reason given" : result.stop_reason);
|
|
|
|
|
+ return MarkerOutcome::Stop;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // A node asking to pause ends this pass. The execution is stored as Waiting
|
|
|
|
|
+ // with everything computed so far, and a later resume picks it up from here
|
|
|
|
|
+ // rather than starting again. Unreachable inside a loop, where the pause was
|
|
|
|
|
+ // turned into a failure above.
|
|
|
|
|
+ if (node_result.output.contains("_pause")) {
|
|
|
|
|
+ const auto& pause = node_result.output["_pause"];
|
|
|
|
|
+
|
|
|
|
|
+ result.status = ExecutionStatus::Waiting;
|
|
|
|
|
+ result.paused_node_id = node_id;
|
|
|
|
|
+ result.pause_token = pause.value("token", "");
|
|
|
|
|
+ result.pause_expires_at = pause.value("expiresAt", static_cast<int64_t>(0));
|
|
|
|
|
+
|
|
|
|
|
+ auto& stored = result.node_results[result_key];
|
|
|
|
|
+ stored.output.erase("_pause");
|
|
|
|
|
+
|
|
|
|
|
+ // No callback here - the generic terminal callback (after
|
|
|
|
|
+ // storeExecution) already emits "execution.waiting" once this pass
|
|
|
|
|
+ // ends, and it carries nodeId/expiresAt too. Emitting here as well
|
|
|
|
|
+ // produced the same event twice with two different payload shapes.
|
|
|
|
|
+
|
|
|
|
|
+ LOG_INFO("Execution {} paused at node {}", result.execution_id, node_id);
|
|
|
|
|
+ return MarkerOutcome::Pause;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return MarkerOutcome::Continue;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
bool WorkflowEngine::runSubWorkflow(NodeExecutionResult& node_result, int call_depth,
|
|
bool WorkflowEngine::runSubWorkflow(NodeExecutionResult& node_result, int call_depth,
|
|
|
const std::string& parent_execution_id) {
|
|
const std::string& parent_execution_id) {
|
|
|
const auto call = node_result.output["_callWorkflow"];
|
|
const auto call = node_result.output["_callWorkflow"];
|
|
@@ -2237,17 +2285,6 @@ bool WorkflowEngine::executeLoopBody(
|
|
|
// does not exist in this execution's stored data. Both the cache path and
|
|
// does not exist in this execution's stored data. Both the cache path and
|
|
|
// the fresh-execution path route through this single check so the
|
|
// the fresh-execution path route through this single check so the
|
|
|
// failure and its message cannot drift apart between them.
|
|
// failure and its message cannot drift apart between them.
|
|
|
- auto rejectPauseInLoopBody = [](NodeExecutionResult& body_result) {
|
|
|
|
|
- if (body_result.status == NodeStatus::Completed &&
|
|
|
|
|
- body_result.output.contains("_pause")) {
|
|
|
|
|
- body_result.status = NodeStatus::Failed;
|
|
|
|
|
- body_result.error = "A node cannot pause inside a Loop body, because a "
|
|
|
|
|
- "paused loop iteration cannot be resumed. Collect the "
|
|
|
|
|
- "items first, approve once, then loop";
|
|
|
|
|
- body_result.output.erase("_pause");
|
|
|
|
|
- }
|
|
|
|
|
- };
|
|
|
|
|
-
|
|
|
|
|
// Set when a body node asks to stop the whole run, so both the body-node
|
|
// Set when a body node asks to stop the whole run, so both the body-node
|
|
|
// loop and the per-item loop can unwind.
|
|
// loop and the per-item loop can unwind.
|
|
|
bool stopped_in_body = false;
|
|
bool stopped_in_body = false;
|
|
@@ -2460,19 +2497,15 @@ bool WorkflowEngine::executeLoopBody(
|
|
|
cached_result.finished_at = cached_result.started_at;
|
|
cached_result.finished_at = cached_result.started_at;
|
|
|
cached_result.from_cache = true;
|
|
cached_result.from_cache = true;
|
|
|
|
|
|
|
|
- rejectPauseInLoopBody(cached_result);
|
|
|
|
|
|
|
+ // A cached output carries the same markers a fresh one
|
|
|
|
|
+ // does, and a replayed run has to stop where the original
|
|
|
|
|
+ // did - so it goes through the same handling.
|
|
|
|
|
+ const MarkerOutcome cached_outcome = applyNodeMarkers(
|
|
|
|
|
+ cached_result, body_node_id,
|
|
|
|
|
+ body_node_id + "_iter_" + std::to_string(i), result,
|
|
|
|
|
+ nlohmann::json::object(), {}, static_cast<int>(i));
|
|
|
|
|
|
|
|
iteration_results[body_node_id] = cached_result;
|
|
iteration_results[body_node_id] = cached_result;
|
|
|
- result.node_results[body_node_id + "_iter_" + std::to_string(i)] = cached_result;
|
|
|
|
|
-
|
|
|
|
|
- if (cached_result.status == NodeStatus::Completed &&
|
|
|
|
|
- cached_result.output.contains("_webhookResponse")) {
|
|
|
|
|
- if (!result.webhook_response.is_null()) {
|
|
|
|
|
- LOG_WARN("Node {} in a loop body overrides a webhook response already set",
|
|
|
|
|
- body_node_id);
|
|
|
|
|
- }
|
|
|
|
|
- result.webhook_response = cached_result.output["_webhookResponse"];
|
|
|
|
|
- }
|
|
|
|
|
|
|
|
|
|
if (callback) {
|
|
if (callback) {
|
|
|
nlohmann::json cache_event_data = {
|
|
nlohmann::json cache_event_data = {
|
|
@@ -2503,6 +2536,11 @@ bool WorkflowEngine::executeLoopBody(
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ if (cached_outcome == MarkerOutcome::Stop) {
|
|
|
|
|
+ stopped_in_body = true;
|
|
|
|
|
+ break;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
continue;
|
|
continue;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -2577,19 +2615,12 @@ bool WorkflowEngine::executeLoopBody(
|
|
|
body_node_def_for_defaults);
|
|
body_node_def_for_defaults);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- if (!body_overlay.empty() && body_result.status == NodeStatus::Completed) {
|
|
|
|
|
- body_result.output["_appliedConfig"] = body_overlay;
|
|
|
|
|
- if (!body_overlay_ignored.empty()) {
|
|
|
|
|
- body_result.output["_ignoredConfigKeys"] = body_overlay_ignored;
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- if (body_result.status == NodeStatus::Completed &&
|
|
|
|
|
- body_result.output.contains("_callWorkflow")) {
|
|
|
|
|
- runSubWorkflow(body_result, result.call_depth, result.execution_id);
|
|
|
|
|
- }
|
|
|
|
|
|
|
+ // Store in main results with iteration suffix
|
|
|
|
|
+ const std::string result_key = body_node_id + "_iter_" + std::to_string(i);
|
|
|
|
|
|
|
|
- rejectPauseInLoopBody(body_result);
|
|
|
|
|
|
|
+ const MarkerOutcome body_outcome = applyNodeMarkers(
|
|
|
|
|
+ body_result, body_node_id, result_key, result,
|
|
|
|
|
+ body_overlay, body_overlay_ignored, static_cast<int>(i));
|
|
|
|
|
|
|
|
// Only a node that actually ran counts. A node on a branch that was
|
|
// Only a node that actually ran counts. A node on a branch that was
|
|
|
// not taken is skipped, and a skipped loop-back edge is exactly the
|
|
// not taken is skipped, and a skipped loop-back edge is exactly the
|
|
@@ -2601,39 +2632,9 @@ bool WorkflowEngine::executeLoopBody(
|
|
|
|
|
|
|
|
iteration_results[body_node_id] = body_result;
|
|
iteration_results[body_node_id] = body_result;
|
|
|
|
|
|
|
|
- // Store in main results with iteration suffix
|
|
|
|
|
- std::string result_key = body_node_id + "_iter_" + std::to_string(i);
|
|
|
|
|
- result.node_results[result_key] = body_result;
|
|
|
|
|
-
|
|
|
|
|
- if (body_result.status == NodeStatus::Completed &&
|
|
|
|
|
- body_result.output.contains("_webhookResponse")) {
|
|
|
|
|
- if (!result.webhook_response.is_null()) {
|
|
|
|
|
- LOG_WARN("Node {} in a loop body overrides a webhook response already set",
|
|
|
|
|
- body_node_id);
|
|
|
|
|
- }
|
|
|
|
|
- result.webhook_response = body_result.output["_webhookResponse"];
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- // A stop raised inside a loop body ends the whole run, not just the
|
|
|
|
|
- // iteration - "stop the workflow" would be a strange thing to mean
|
|
|
|
|
- // per-item. The flag travels out on the result because this walk
|
|
|
|
|
- // cannot end the outer one itself.
|
|
|
|
|
- if (body_result.status == NodeStatus::Completed &&
|
|
|
|
|
- body_result.output.contains("_stop")) {
|
|
|
|
|
- const auto& stop = body_result.output["_stop"];
|
|
|
|
|
-
|
|
|
|
|
- result.stop_requested = true;
|
|
|
|
|
- result.stopped_node_id = body_node_id;
|
|
|
|
|
- result.stop_reason = stop.value("reason", "");
|
|
|
|
|
-
|
|
|
|
|
- auto& stored_body = result.node_results[result_key];
|
|
|
|
|
- stored_body.output.erase("_stop");
|
|
|
|
|
- stored_body.output["stopped"] = true;
|
|
|
|
|
- stored_body.output["reason"] = result.stop_reason;
|
|
|
|
|
-
|
|
|
|
|
- LOG_INFO("Execution {} stopped at node {} during loop iteration {}: {}",
|
|
|
|
|
- result.execution_id, body_node_id, i,
|
|
|
|
|
- result.stop_reason.empty() ? "no reason given" : result.stop_reason);
|
|
|
|
|
|
|
+ // The flag travels out on the result because this walk cannot end
|
|
|
|
|
+ // the outer one itself.
|
|
|
|
|
+ if (body_outcome == MarkerOutcome::Stop) {
|
|
|
stopped_in_body = true;
|
|
stopped_in_body = true;
|
|
|
break;
|
|
break;
|
|
|
}
|
|
}
|