starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-26 08:01:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(backend): increase dynamic js timeout + improve client passing

Browse Source
This commit is contained in:
Ruben Fiszel
2023-03-23 10:31:27 +01:00
parent a2f3df3d22
commit 63bd0cfb28
7 changed files with 161 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.lock
Generated
+1
View File
@@ -4996,6 +4996,7 @@ dependencies = [
"git-version",
"itertools",
"lazy_static",
"once_cell",
"prometheus",
"rand 0.8.5",
"regex",
backend/Cargo.toml
+1
View File
@@ -153,3 +153,4 @@ async-stripe = { version = "0.14", features = [
"checkout",
] }
async_zip = { version = "0.0.11", features = ["full"] }
once_cell = "1.17.1"
backend/windmill-worker/Cargo.toml
+1
View File
@@ -48,3 +48,4 @@ deno_core.workspace = true
const_format.workspace = true
git-version.workspace = true
dyn-iter.workspace = true
once_cell.workspace = true
backend/windmill-worker/src/js_eval.rs
+76 -68
View File
@@ -6,21 +6,20 @@
* LICENSE-AGPL for a copy of the license.
*/
use std::collections::HashMap;
use std::{collections::HashMap, cell::RefCell, rc::Rc};
use deno_core::{op, serde_v8, v8, v8::IsolateHandle, Extension, JsRuntime, RuntimeOptions};
use deno_core::{op, serde_v8, v8, v8::IsolateHandle, Extension, JsRuntime, RuntimeOptions, OpState};
use itertools::Itertools;
use lazy_static::lazy_static;
use regex::Regex;
use serde_json::Value;
use tokio::{sync::oneshot, time::timeout};
use tokio::{sync::{oneshot}, time::timeout};
use uuid::Uuid;
use windmill_common::{error::Error, flow_status::JobResult};
pub struct EvalCreds {
pub workspace: String,
pub token: String,
}
use crate::AuthedClient;
#[derive(Debug, Clone)]
pub struct IdContext {
@@ -29,22 +28,23 @@ pub struct IdContext {
pub previous_id: String,
}
pub struct OptAuthedClient(Option<AuthedClient>);
pub async fn eval_timeout(
expr: String,
env: Vec<(String, serde_json::Value)>,
creds: Option<EvalCreds>,
authed_client: Option<&AuthedClient>,
by_id: Option<IdContext>,
base_internal_url: &str,
) -> anyhow::Result<serde_json::Value> {
let expr2 = expr.clone();
let (sender, mut receiver) = oneshot::channel::<IsolateHandle>();
let base_internal_url: String = base_internal_url.to_string();
let has_client = authed_client.is_some();
let authed_client = authed_client.cloned();
timeout(
std::time::Duration::from_millis(3000),
std::time::Duration::from_millis(10000),
tokio::task::spawn_blocking(move || {
let mut ops = vec![];
if creds.is_some() {
if authed_client.is_some() {
ops.extend([
// An op for summing an array of numbers
// The op-layer automatically deserializes inputs
@@ -54,7 +54,7 @@ pub async fn eval_timeout(
])
}
if by_id.is_some() {
if by_id.is_some() && authed_client.is_some() {
ops.push(op_get_result::decl());
ops.push(op_get_id::decl());
}
@@ -68,6 +68,11 @@ pub async fn eval_timeout(
};
let mut js_runtime = JsRuntime::new(options);
{
let op_state = js_runtime.op_state();
let mut op_state = op_state.borrow_mut();
op_state.put(OptAuthedClient(authed_client.clone()));
}
sender
.send(js_runtime.v8_isolate().thread_safe_handle())
@@ -90,9 +95,8 @@ pub async fn eval_timeout(
&mut js_runtime,
&expr,
env,
creds,
by_id,
&base_internal_url,
has_client,
))?;
Ok(r) as anyhow::Result<Value>
@@ -104,7 +108,7 @@ pub async fn eval_timeout(
isolate.terminate_execution();
};
Error::ExecutionErr(format!(
"The expression of evaluation `{expr2}` took too long to execute (>3000ms)"
"The expression of evaluation `{expr2}` took too long to execute (>10000ms)"
))
})??
}
@@ -150,9 +154,8 @@ async fn eval(
context: &mut JsRuntime,
expr: &str,
env: Vec<(String, serde_json::Value)>,
creds: Option<EvalCreds>,
by_id: Option<IdContext>,
base_internal_url: &str,
has_client: bool,
) -> anyhow::Result<serde_json::Value> {
let exprs = expr
.trim()
@@ -169,7 +172,7 @@ async fn eval(
exprs.last().unwrap()
)
};
let (api_code, by_id_code) = if let Some(EvalCreds { workspace, token }) = creds {
let (api_code, by_id_code) = if has_client {
let by_id_code = if let Some(by_id) = by_id {
format!(
r#"
@@ -186,12 +189,12 @@ async function result_by_id(node_id) {{
}}
}} else {{
let flow_job_id = "{}";
return await Deno.core.opAsync("op_get_id", [workspace, flow_job_id, token, base_url, node_id]);
return await Deno.core.opAsync("op_get_id", [ flow_job_id, node_id]);
}}
}}
async function get_result(id) {{
return await Deno.core.opAsync("op_get_result", [workspace, id, token, base_url]);
return await Deno.core.opAsync("op_get_result", [id]);
}}
const results = new Proxy({{}}, {{
get: function(target, name, receiver) {{
@@ -222,17 +225,13 @@ const results = new Proxy({{}}, {{
let api_code = format!(
r#"
let workspace = "{workspace}";
let base_url = "{}";
let token = "{token}";
async function variable(path) {{
return await Deno.core.opAsync("op_variable", [workspace, path, token, base_url]);
return await Deno.core.opAsync("op_variable", [path]);
}}
async function resource(path) {{
return await Deno.core.opAsync("op_resource", [workspace, path, token, base_url]);
return await Deno.core.opAsync("op_resource", [path]);
}}
"#,
base_internal_url,
);
(api_code, by_id_code)
} else {
@@ -281,59 +280,68 @@ async function resource(path) {{
// TODO: Can we a) share the api configuration here somehow or b) just implement this natively in deno, via the deno client?
#[op]
async fn op_variable(args: Vec<String>) -> Result<String, anyhow::Error> {
let workspace = &args[0];
let path = &args[1];
let token = &args[2];
let base_url = &args[3];
let client = windmill_api_client::create_client(base_url, token.clone());
let result = client.get_variable(workspace, path, None).await?;
Ok(result.into_inner().value.unwrap_or_else(|| "".to_owned()))
async fn op_variable(op_state: Rc<RefCell<OpState>>, args: Vec<String>) -> Result<String, anyhow::Error> {
let path = &args[0];
let client = op_state.borrow().borrow::<OptAuthedClient>().0.clone();
if let Some(client) = client {
let result = client.get_client().get_variable(&client.workspace, path, None).await?;
Ok(result.into_inner().value.unwrap_or_else(|| "".to_owned()))
} else {
anyhow::bail!("No client found in op state");
}
}
#[op]
async fn op_get_result(args: Vec<String>) -> Result<serde_json::Value, anyhow::Error> {
let workspace = &args[0];
let id = &args[1];
let token = &args[2];
let base_url = &args[3];
let client = windmill_api_client::create_client(base_url, token.clone());
let result = client
.get_completed_job_result(workspace, &id.parse()?)
async fn op_get_result(op_state: Rc<RefCell<OpState>>, args: Vec<String>) -> Result<serde_json::Value, anyhow::Error> {
let id = &args[0];
let client = op_state.borrow().borrow::<OptAuthedClient>().0.clone();
if let Some(client) = client {
let result = client
.get_client()
.get_completed_job_result(&client.workspace, &id.parse()?)
.await?
.clone();
Ok(serde_json::json!(result))
} else {
anyhow::bail!("No client found in op state");
}
}
#[op]
async fn op_get_id(args: Vec<String>) -> Result<Option<serde_json::Value>, anyhow::Error> {
let workspace = &args[0];
let flow_job_id = &args[1];
let token = &args[2];
let base_url = &args[3];
let node_id = &args[4];
async fn op_get_id(op_state: Rc<RefCell<OpState>>, args: Vec<String>) -> Result<Option<serde_json::Value>, anyhow::Error> {
let flow_job_id = &args[0];
let node_id = &args[1];
let client = windmill_api_client::create_client(base_url, token.clone());
let result = client
.result_by_id(workspace, flow_job_id, node_id)
let client = op_state.borrow().borrow::<OptAuthedClient>().0.clone();
if let Some(client) = client {
let result = client
.get_client()
.result_by_id(&client.workspace, flow_job_id, node_id)
.await
.map_or(None, |e| Some(e.into_inner()));
Ok(result)
Ok(result)
} else {
anyhow::bail!("No client found in op state");
}
}
#[op]
async fn op_resource(args: Vec<String>) -> Result<serde_json::Value, anyhow::Error> {
let workspace = &args[0];
let path = &args[1];
let token = &args[2];
let base_url = &args[3];
let client = windmill_api_client::create_client(base_url, token.clone());
let result = client.get_resource(workspace, path).await?;
Ok(result
.into_inner()
.value
.unwrap_or_else(|| serde_json::json!({})))
async fn op_resource(op_state: Rc<RefCell<OpState>>, args: Vec<String>) -> Result<serde_json::Value, anyhow::Error> {
let path = &args[0];
let client = op_state.borrow().borrow::<OptAuthedClient>().0.clone();
if let Some(client) = client {
let result = client.get_client().get_resource(&client.workspace, path).await?;
Ok(result
.into_inner()
.value
.unwrap_or_else(|| serde_json::json!({})))
} else {
anyhow::bail!("No client found in op state");
}
}
#[cfg(test)]
@@ -353,7 +361,7 @@ mod tests {
let code = "value.test + params.test";
let mut runtime = JsRuntime::new(RuntimeOptions::default());
let res = eval(&mut runtime, code, env, None, None, String::new().as_str()).await?;
let res = eval(&mut runtime, code, env, None, false).await?;
assert_eq!(res, json!(4));
Ok(())
}
@@ -366,7 +374,7 @@ mod tests {
multiline template`";
let mut runtime = JsRuntime::new(RuntimeOptions::default());
let res = eval(&mut runtime, code, env, None, None, String::new().as_str()).await?;
let res = eval(&mut runtime, code, env, None, false).await?;
assert_eq!(res, json!("my 5\nmultiline template"));
Ok(())
}
@@ -379,7 +387,7 @@ multiline template`";
];
let code = r#"params.test"#;
let res = eval_timeout(code.to_string(), env, None, None, String::new().as_str()).await?;
let res = eval_timeout(code.to_string(), env, None, None).await?;
assert_eq!(res, json!(2));
Ok(())
}
backend/windmill-worker/src/worker.rs
+33 -16
View File
@@ -9,8 +9,10 @@
use const_format::concatcp;
use itertools::Itertools;
use lazy_static::lazy_static;
use once_cell::sync::OnceCell;
use regex::Regex;
use sqlx::{Pool, Postgres, Transaction};
use windmill_api_client::Client;
use std::{
borrow::Borrow, collections::HashMap, io, os::unix::process::ExitStatusExt, panic,
process::Stdio, time::Duration,
@@ -34,7 +36,7 @@ use tokio::{
io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader},
process::{Child, Command},
sync::{
mpsc::{self, Sender}, watch, broadcast,
mpsc::{self, Sender}, watch, broadcast
},
time::{interval, sleep, Instant, MissedTickBehavior},
};
@@ -389,9 +391,9 @@ lazy_static::lazy_static! {
.ok()
.and_then(|x| x.parse::<u16>().ok())
.unwrap_or(DEFAULT_TIMEOUT as u16);
static ref TIMEOUT_DURATION: Duration = Duration::from_secs(*TIMEOUT as u64);
static ref ZOMBIE_JOB_TIMEOUT: String = (*TIMEOUT as u32 * 5).to_string();
static ref SESSION_TOKEN_EXPIRY: i32 = (*TIMEOUT as i32) * 2;
@@ -400,6 +402,21 @@ lazy_static::lazy_static! {
//only matter if CLOUD_HOSTED
const MAX_RESULT_SIZE: usize = 1024 * 1024 * 2; // 2MB
#[derive(Clone)]
pub struct AuthedClient {
pub base_internal_url: String,
pub workspace: String,
pub token: String,
pub client: OnceCell<Client>
}
impl AuthedClient {
pub fn get_client(&self) -> &Client {
return self.client.get_or_init(|| {
windmill_api_client::create_client(&self.base_internal_url, self.token.clone())
});
}
}
#[tracing::instrument(level = "trace")]
@@ -669,13 +686,13 @@ pub async fn run_worker(
)
.await.expect("could not create job token");
tx.commit().await.expect("could not commit job token");
let job_client = windmill_api_client::create_client(base_internal_url, token.clone());
let authed_client = AuthedClient { base_internal_url: base_internal_url.to_string(), token: token.clone(), workspace: job.workspace_id.to_string(), client: OnceCell::new() };
let is_flow = job.job_kind == JobKind::Flow || job.job_kind == JobKind::FlowPreview || job.job_kind == JobKind::FlowDependencies;
if let Some(err) = handle_queued_job(
job.clone(),
db,
&job_client,
&authed_client,
token,
&worker_name,
&worker_dir,
@@ -689,7 +706,7 @@ pub async fn run_worker(
{
handle_job_error(
db,
&job_client,
&authed_client,
job,
err,
Some(metrics),
@@ -733,7 +750,7 @@ pub async fn run_worker(
async fn handle_job_error(
db: &Pool<Postgres>,
client: &windmill_api_client::Client,
client: &AuthedClient,
job: QueuedJob,
err: Error,
metrics: Option<Metrics>,
@@ -832,7 +849,7 @@ fn extract_error_value(log_lines: &str) -> serde_json::Value {
async fn handle_queued_job(
job: QueuedJob,
db: &sqlx::Pool<sqlx::Postgres>,
client: &windmill_api_client::Client,
client: &AuthedClient,
token: String,
worker_name: &str,
worker_dir: &str,
@@ -1012,14 +1029,14 @@ async fn write_file(dir: &str, path: &str, content: &str) -> error::Result<File>
#[async_recursion]
async fn transform_json_value(
name: &str,
client: &windmill_api_client::Client,
client: &AuthedClient,
workspace: &str,
v: Value,
) -> error::Result<Value> {
match v {
Value::String(y) if y.starts_with("$var:") => {
let path = y.strip_prefix("$var:").unwrap();
let v = client
let v = client.get_client()
.get_variable(workspace, path, Some(true))
.await
.map_err(|_| Error::NotFound(format!("Variable {path} not found for `{name}`")))
@@ -1035,7 +1052,7 @@ async fn transform_json_value(
"Argument `{name}` is an invalid resource path: {path}",
)));
}
let v = client
let v = client.get_client()
.get_resource_value(workspace, path)
.await
.map_err(|_| Error::NotFound(format!("Resource {path} not found for `{name}`")))?
@@ -1059,7 +1076,7 @@ async fn transform_json_value(
async fn handle_code_execution_job(
job: &QueuedJob,
db: &sqlx::Pool<sqlx::Postgres>,
client: &windmill_api_client::Client,
client: &AuthedClient,
token: String,
job_dir: &str,
worker_dir: &str,
@@ -1205,7 +1222,7 @@ async fn handle_go_job(
logs: &mut String,
job: &QueuedJob,
db: &sqlx::Pool<sqlx::Postgres>,
client: &windmill_api_client::Client,
client: &AuthedClient,
token: String,
inner_content: &str,
job_dir: &str,
@@ -1505,7 +1522,7 @@ async fn handle_deno_job(
logs: &mut String,
job: &QueuedJob,
db: &sqlx::Pool<sqlx::Postgres>,
client: &windmill_api_client::Client,
client: &AuthedClient,
token: String,
job_dir: &str,
inner_content: &String,
@@ -1648,7 +1665,7 @@ run().catch(async (e) => {{
#[tracing::instrument(level = "trace", skip_all)]
async fn create_args_and_out_file(
client: &windmill_api_client::Client,
client: &AuthedClient,
job: &QueuedJob,
job_dir: &str,
) -> Result<(), Error> {
@@ -1676,7 +1693,7 @@ async fn handle_python_job(
job: &QueuedJob,
logs: &mut String,
db: &sqlx::Pool<sqlx::Postgres>,
client: &windmill_api_client::Client,
client: &AuthedClient,
token: String,
inner_content: &String,
shared_mount: &str,
@@ -2756,7 +2773,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str) {
.await
.expect("could not create job token");
tx.commit().await.expect("could not commit job token");
let client = windmill_api_client::create_client(base_internal_url, token.clone());
let client = AuthedClient { base_internal_url: base_internal_url.to_string(), token: token.clone(), workspace: job.workspace_id.to_string(), client: OnceCell::new() };
let _ = handle_job_error(
db,
backend/windmill-worker/src/worker_flow.rs
+32 -66
View File
@@ -11,8 +11,8 @@ use std::iter;
use std::time::Duration;
use crate::jobs::{add_completed_job, add_completed_job_error, schedule_again_if_scheduled};
use crate::js_eval::{eval_timeout, EvalCreds, IdContext};
use crate::{worker, KEEP_JOB_DIR};
use crate::js_eval::{eval_timeout, IdContext};
use crate::{worker, KEEP_JOB_DIR, AuthedClient};
use anyhow::Context;
use async_recursion::async_recursion;
use dyn_iter::DynIter;
@@ -40,7 +40,7 @@ use windmill_queue::{
// #[instrument(level = "trace", skip_all)]
pub async fn update_flow_status_after_job_completion(
db: &DB,
client: &windmill_api_client::Client,
client: &AuthedClient,
flow: uuid::Uuid,
job_id_for_status: &Uuid,
w_id: &str,
@@ -120,7 +120,7 @@ pub async fn update_flow_status_after_job_completion(
let stop_early = success
&& if let Some(expr) = r.stop_early_expr.clone() {
compute_bool_from_expr(expr, &r.args, result.clone(), None, None, base_internal_url)
compute_bool_from_expr(expr, &r.args, result.clone(), None, Some(client))
.await?
} else {
false
@@ -598,8 +598,7 @@ async fn compute_bool_from_expr(
flow_args: &Option<serde_json::Value>,
result: serde_json::Value,
by_id: Option<IdContext>,
creds: Option<EvalCreds>,
base_internal_url: &str,
client: Option<&AuthedClient>,
) -> error::Result<bool> {
let flow_input = flow_args.clone().unwrap_or_else(|| json!({}));
match eval_timeout(
@@ -610,9 +609,8 @@ async fn compute_bool_from_expr(
("previous_result".to_string(), result),
]
.into(),
creds,
client,
by_id,
base_internal_url,
)
.await?
{
@@ -687,12 +685,10 @@ async fn transform_input(
flow_args: &Option<serde_json::Value>,
last_result: serde_json::Value,
input_transforms: &HashMap<String, InputTransform>,
workspace: &str,
token: &str,
resumes: &[Value],
approvers: Vec<String>,
by_id: &IdContext,
base_internal_url: &str,
client: &AuthedClient,
) -> windmill_common::error::Result<Map<String, serde_json::Value>> {
let mut mapped = serde_json::Map::new();
@@ -734,9 +730,8 @@ async fn transform_input(
let v = eval_timeout(
expr.to_string(),
context,
Some(EvalCreds { workspace: workspace.to_string(), token: token.to_string() }),
Some(by_id.clone()),
base_internal_url,
Some(client),
Some(by_id.clone())
)
.await
.map_err(|e| {
@@ -757,7 +752,7 @@ async fn transform_input(
pub async fn handle_flow(
flow_job: &QueuedJob,
db: &sqlx::Pool<sqlx::Postgres>,
client: &windmill_api_client::Client,
client: &AuthedClient,
last_result: serde_json::Value,
same_worker_tx: Sender<Uuid>,
worker_dir: &str,
@@ -797,7 +792,7 @@ async fn push_next_flow_job(
mut status: FlowStatus,
flow: FlowValue,
db: &sqlx::Pool<sqlx::Postgres>,
client: &windmill_api_client::Client,
client: &AuthedClient,
mut last_result: serde_json::Value,
same_worker_tx: Sender<Uuid>,
worker_dir: &str,
@@ -870,8 +865,7 @@ async fn push_next_flow_job(
]
.into(),
None,
None,
base_internal_url,
None
)
.await
.map_err(|e| {
@@ -1114,15 +1108,15 @@ async fn push_next_flow_job(
_ => (),
}
let mut transform_context: Option<TransformContext> = None;
let mut transform_context: Option<IdContext> = None;
let args: windmill_common::error::Result<_> = match &module.value {
FlowModuleValue::Script { input_transforms, .. }
| FlowModuleValue::RawScript { input_transforms, .. }
| FlowModuleValue::Flow { input_transforms, .. } => {
let ctx = get_transform_context(db, &flow_job, previous_id.clone(), &status).await?;
let ctx = get_transform_context(&flow_job, previous_id.clone(), &status).await?;
transform_context = Some(ctx);
let (token, by_id) = transform_context.as_ref().unwrap();
let by_id = transform_context.as_ref().unwrap();
transform_input(
&flow_job.args,
last_result.clone(),
@@ -1131,12 +1125,10 @@ async fn push_next_flow_job(
} else {
&module.input_transforms
},
&flow_job.workspace_id,
&token,
resume_messages.as_slice(),
approvers,
by_id,
base_internal_url,
client
)
.await
}
@@ -1169,14 +1161,13 @@ async fn push_next_flow_job(
flow_job,
&flow,
transform_context,
db,
tx,
&module,
&status,
&status_module,
last_result.clone(),
previous_id,
base_internal_url,
client
)
.await?;
tx.commit().await?;
@@ -1507,20 +1498,18 @@ async fn script_path_to_payload<'c>(
Ok(job_payload)
}
type TransformContext = (String, IdContext);
async fn compute_next_flow_transform<'c>(
flow_job: &QueuedJob,
flow: &FlowValue,
transform_context: Option<TransformContext>,
db: &DB,
by_id: Option<IdContext>,
mut tx: sqlx::Transaction<'c, sqlx::Postgres>,
module: &FlowModule,
status: &FlowStatus,
status_module: &FlowStatusModule,
last_result: serde_json::Value,
previous_id: String,
base_internal_url: &str,
client: &AuthedClient
) -> error::Result<(sqlx::Transaction<'c, sqlx::Postgres>, NextFlowTransform)> {
match &module.value {
FlowModuleValue::Identity => Ok((
@@ -1573,10 +1562,10 @@ async fn compute_next_flow_transform<'c>(
FlowStatusModule::WaitingForPriorSteps { .. }
| FlowStatusModule::WaitingForEvents { .. }
| FlowStatusModule::WaitingForExecutor { .. } => {
let (token, by_id) = if let Some(x) = transform_context {
let by_id = if let Some(x) = by_id {
x
} else {
get_transform_context(db, &flow_job, previous_id, &status).await?
get_transform_context(&flow_job, previous_id, &status).await?
};
let flow_input = flow_job.args.clone().unwrap_or_else(|| json!({}));
/* Iterator is an InputTransform, evaluate it into an array. */
@@ -1589,10 +1578,8 @@ async fn compute_next_flow_transform<'c>(
("previous_result".to_string(), last_result.clone()),
]
},
token,
flow_job.workspace_id.clone(),
Some(by_id),
base_internal_url,
Some(client),
Some(by_id)
)
.await?
.into_array()
@@ -1709,19 +1696,15 @@ async fn compute_next_flow_transform<'c>(
| FlowStatusModule::WaitingForEvents { .. }
| FlowStatusModule::WaitingForExecutor { .. } => {
let mut branch_chosen = BranchChosen::Default;
let (token, idcontext) =
get_transform_context(db, &flow_job, previous_id, &status).await?;
let idcontext =
get_transform_context(&flow_job, previous_id, &status).await?;
for (i, b) in branches.iter().enumerate() {
let pred = compute_bool_from_expr(
b.expr.to_string(),
&flow_job.args,
last_result.clone(),
Some(idcontext.clone()),
Some(EvalCreds {
workspace: flow_job.workspace_id.clone(),
token: token.to_string(),
}),
base_internal_url,
Some(client),
)
.await?;
@@ -1886,43 +1869,27 @@ async fn compute_next_flow_transform<'c>(
}
async fn get_transform_context(
db: &DB,
flow_job: &QueuedJob,
previous_id: String,
status: &FlowStatus,
) -> error::Result<TransformContext> {
let tx = db.begin().await?;
let (tx, new_token) = crate::create_token_for_owner(
tx,
&flow_job.workspace_id,
&flow_job.permissioned_as,
"transform-input",
10,
&flow_job.email,
)
.await?;
//we need to commit asap otherwise the token won't be valid for auth to check outside of this transaction
//which will happen with client http calls
tx.commit().await?;
) -> error::Result<IdContext> {
let steps_results: HashMap<String, JobResult> = status
.modules
.iter()
.filter_map(|x| x.job_result().map(|y| (x.id(), y)))
.collect();
Ok((
new_token,
Ok(
IdContext { flow_job: flow_job.id, steps_results, previous_id },
))
)
}
async fn evaluate_with<F>(
transform: InputTransform,
vars: F,
token: String,
workspace: String,
client: Option<&AuthedClient>,
by_id: Option<IdContext>,
base_internal_url: &str,
) -> anyhow::Result<serde_json::Value>
where
F: FnOnce() -> Vec<(String, serde_json::Value)>,
@@ -1933,9 +1900,8 @@ where
eval_timeout(
expr,
vars(),
Some(EvalCreds { workspace, token }),
client,
by_id,
base_internal_url,
)
.await
}
frontend/src/lib/components/DisplayResult.svelte
+17 -14
View File
@@ -6,6 +6,7 @@
import { Button, Drawer, DrawerContent } from './common'
import autosize from 'svelte-autosize'
import { ClipboardCopy } from 'lucide-svelte'
import Portal from 'svelte-portal'
export let result: any
export let requireHtmlApproval = false
@@ -91,20 +92,22 @@
let jsonViewer: Drawer
</script>
<Drawer bind:this={jsonViewer} size="900px">
<DrawerContent title="Expanded Result" on:close={jsonViewer.closeDrawer}>
<svelte:fragment slot="actions">
<Button
on:click={() => copyToClipboard(JSON.stringify(result, null, 4))}
color="light"
size="xs"
>
<div class="flex gap-2 items-center">Copy to clipboard <ClipboardCopy /> </div>
</Button>
</svelte:fragment>
<Highlight language={json} code={JSON.stringify(result, null, 4).replace(/\\n/g, '\n')} />
</DrawerContent>
</Drawer>
<Portal>
<Drawer bind:this={jsonViewer} size="900px">
<DrawerContent title="Expanded Result" on:close={jsonViewer.closeDrawer}>
<svelte:fragment slot="actions">
<Button
on:click={() => copyToClipboard(JSON.stringify(result, null, 4))}
color="light"
size="xs"
>
<div class="flex gap-2 items-center">Copy to clipboard <ClipboardCopy /> </div>
</Button>
</svelte:fragment>
<Highlight language={json} code={JSON.stringify(result, null, 4).replace(/\\n/g, '\n')} />
</DrawerContent>
</Drawer>
</Portal>
<div class="inline-highlight">
{#if result != undefined}