From 73e65203126d0e2bd12c09c1f6e8eb64abc70a46 Mon Sep 17 00:00:00 2001 From: Lucas Abel <22837557+uael@users.noreply.github.com> Date: Tue, 17 Dec 2024 20:44:51 +0100 Subject: [PATCH] cache: improve error reporting (#4940) --- ...68ef270567ebd70266f07139f3e3e34940699.json | 22 + ...3ab6a104bbcfb827a4a9e4e03e38732375d55.json | 22 + ...0fd7f2969aab0c422a00bbdade6a7236aa75d.json | 72 +++ backend/windmill-api/src/apps.rs | 4 +- backend/windmill-common/src/cache.rs | 444 ++++++++++-------- backend/windmill-common/src/error.rs | 22 +- 6 files changed, 376 insertions(+), 210 deletions(-) create mode 100644 backend/.sqlx/query-7a9a711c7cb05ed8a61116586fb68ef270567ebd70266f07139f3e3e34940699.json create mode 100644 backend/.sqlx/query-c7de95ef4934752b62d6d01bf663ab6a104bbcfb827a4a9e4e03e38732375d55.json create mode 100644 backend/.sqlx/query-d061e7ca73987036928e17245360fd7f2969aab0c422a00bbdade6a7236aa75d.json diff --git a/backend/.sqlx/query-7a9a711c7cb05ed8a61116586fb68ef270567ebd70266f07139f3e3e34940699.json b/backend/.sqlx/query-7a9a711c7cb05ed8a61116586fb68ef270567ebd70266f07139f3e3e34940699.json new file mode 100644 index 0000000000..4a849599fe --- /dev/null +++ b/backend/.sqlx/query-7a9a711c7cb05ed8a61116586fb68ef270567ebd70266f07139f3e3e34940699.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT value AS \"value!: Json>\"\n FROM flow_version WHERE id = $1 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "value!: Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + false + ] + }, + "hash": "7a9a711c7cb05ed8a61116586fb68ef270567ebd70266f07139f3e3e34940699" +} diff --git a/backend/.sqlx/query-c7de95ef4934752b62d6d01bf663ab6a104bbcfb827a4a9e4e03e38732375d55.json b/backend/.sqlx/query-c7de95ef4934752b62d6d01bf663ab6a104bbcfb827a4a9e4e03e38732375d55.json new file mode 100644 index 0000000000..c044973cc0 --- /dev/null +++ b/backend/.sqlx/query-c7de95ef4934752b62d6d01bf663ab6a104bbcfb827a4a9e4e03e38732375d55.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT value AS \"value!: Json>\"\n FROM flow_version_lite WHERE id = $1 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "value!: Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + true + ] + }, + "hash": "c7de95ef4934752b62d6d01bf663ab6a104bbcfb827a4a9e4e03e38732375d55" +} diff --git a/backend/.sqlx/query-d061e7ca73987036928e17245360fd7f2969aab0c422a00bbdade6a7236aa75d.json b/backend/.sqlx/query-d061e7ca73987036928e17245360fd7f2969aab0c422a00bbdade6a7236aa75d.json new file mode 100644 index 0000000000..d6d6e77aa4 --- /dev/null +++ b/backend/.sqlx/query-d061e7ca73987036928e17245360fd7f2969aab0c422a00bbdade6a7236aa75d.json @@ -0,0 +1,72 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT lock AS \"lock: String\", content AS \"code!: String\",\n language AS \"language: Option\", envs AS \"envs: Vec\", codebase AS \"codebase: String\" FROM script WHERE hash = $1 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "lock: String", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "code!: String", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "language: Option", + "type_info": { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell", + "mssql", + "php", + "bunnative", + "rust", + "ansible", + "csharp" + ] + } + } + } + }, + { + "ordinal": 3, + "name": "envs: Vec", + "type_info": "VarcharArray" + }, + { + "ordinal": 4, + "name": "codebase: String", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + true, + false, + false, + true, + true + ] + }, + "hash": "d061e7ca73987036928e17245360fd7f2969aab0c422a00bbdade6a7236aa75d" +} diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index f9228a6693..814fd58cb0 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -1324,7 +1324,7 @@ async fn execute_component( let cache = cache::anon!({ u64 => Arc } in "policy" <= 1000); arc_policy = policy_fut .map_ok(sqlx::types::Json) // cache as json. - .cached(cache, &(id as u64), |sqlx::types::Json(x)| Arc::new(x)) + .cached(cache, id as u64, |sqlx::types::Json(x)| Arc::new(x)) .await?; &*arc_policy } else { @@ -1353,7 +1353,7 @@ async fn execute_component( .fetch_one(&db) .map_err(Into::::into) .map_ok(sqlx::types::Json) // cache as json. - .cached(cache, &(*id as u64), |sqlx::types::Json(x)| Arc::new(x)) + .cached(cache, *id as u64, |sqlx::types::Json(x)| Arc::new(x)) .await? } _ => unreachable!(), diff --git a/backend/windmill-common/src/cache.rs b/backend/windmill-common/src/cache.rs index 9f3ec024ae..e222d1202c 100644 --- a/backend/windmill-common/src/cache.rs +++ b/backend/windmill-common/src/cache.rs @@ -7,11 +7,10 @@ use crate::scripts::ScriptLang; use std::future::Future; use std::hash::Hash; +use std::panic::Location; use std::path::{Path, PathBuf}; use std::sync::Arc; -use futures::future::TryFutureExt; -use quick_cache::Equivalent; use serde::{Deserialize, Serialize}; use sqlx::types::{Json, JsonRawValue as RawValue}; use sqlx::PgExecutor; @@ -48,19 +47,18 @@ impl FsBackedCache { } /// Gets or inserts an item in the cache with key `key`. - pub async fn get_or_insert_async<'a, T: fs::Bundle, Q, F>( + pub async fn get_or_insert_async<'a, T: fs::Bundle, F>( &'a self, - key: &Q, + key: Key, map: impl Fn(T) -> Val, with: F, ) -> error::Result where - Q: Hash + Equivalent + ToOwned, + Key: Clone, F: Future>, { self.cache - .get_or_insert_async(key, async { - let key = key.to_owned(); + .get_or_insert_async(&key, async { fs::import_or_insert_with(self.path(&key), with) .await .map(map) @@ -138,22 +136,22 @@ pub mod future { /// #[allow(unused)] /// async { /// let result = std::future::ready(Ok(Json(42))) - /// .cached(cache::anon!({ u64 => Json } in "test" <= 1), &42, |x| x) + /// .cached(cache::anon!({ u64 => Json } in "test" <= 1), 42, |x| x) /// .await; /// /// assert_eq!(result.unwrap(), Json(42)); /// }; /// ``` - fn cached( + fn cached( self, cache: &FsBackedCache, - key: &Q, + key: Key, map: impl Fn(T) -> Val, ) -> impl Future> where - Q: Hash + Equivalent + ToOwned, + Key: Clone, { - cache.get_or_insert_async(key, map, self) + cache.get_or_insert_async(key.to_owned(), map, self) } } @@ -221,6 +219,17 @@ pub struct ScriptMetadata { pub codebase: Option, } +fn unwrap_or_error( + at: &'static Location, + entity: &'static str, + key: Key, +) -> impl FnOnce(Option) -> error::Result { + move |optional| { + optional + .ok_or_else(|| error::Error::InternalErrAt(at, format!("{key:?}: {entity} not found"))) + } +} + const _: () = { impl fs::Bundle for FlowData { type Item = &'static str; @@ -302,110 +311,124 @@ pub mod flow { /// If not present, import from the file-system cache or fetch it from the database and write /// it to the file system and cache. /// This should be preferred over fetching the database directly. - pub async fn fetch_script( - e: impl PgExecutor<'_>, + #[track_caller] + pub fn fetch_script<'c>( + e: impl PgExecutor<'c>, node: FlowNodeId, - ) -> error::Result> { - fetch_node(e, node).await.and_then(|data| match data { - RawData::Script(data) => Ok(data), - RawData::Flow(_) => Err(error::Error::InternalErr(format!( - "Flow node ({:x}) isn't a script node.", - node.0 - ))), - }) + ) -> impl Future>> { + let fetch_node = fetch_node(e, node); + async move { + fetch_node.await.and_then(|data| match data { + RawData::Script(data) => Ok(data), + RawData::Flow(_) => Err(error::Error::InternalErr(format!( + "Flow node ({:x}) isn't a script node.", + node.0 + ))), + }) + } } /// Fetch the flow node flow value referenced by `node` from the cache. /// If not present, import from the file-system cache or fetch it from the database and write /// it to the file system and cache. /// This should be preferred over fetching the database directly. - pub async fn fetch_flow( - e: impl PgExecutor<'_>, + #[track_caller] + pub fn fetch_flow<'c>( + e: impl PgExecutor<'c>, node: FlowNodeId, - ) -> error::Result> { - fetch_node(e, node).await.and_then(|data| match data { - RawData::Flow(data) => Ok(data), - RawData::Script(_) => Err(error::Error::InternalErr(format!( - "Flow node ({:x}) isn't a flow node.", - node.0 - ))), - }) + ) -> impl Future>> { + let fetch_node = fetch_node(e, node); + async move { + fetch_node.await.and_then(|data| match data { + RawData::Flow(data) => Ok(data), + RawData::Script(_) => Err(error::Error::InternalErr(format!( + "Flow node ({:x}) isn't a flow node.", + node.0 + ))), + }) + } } /// Fetch the flow node referenced by `node` from the cache. /// If not present, import from the file-system cache or fetch it from the database and write /// it to the file system and cache. /// This should be preferred over fetching the database directly. - pub(super) async fn fetch_node( - e: impl PgExecutor<'_>, + #[track_caller] + pub(super) fn fetch_node<'c>( + e: impl PgExecutor<'c>, node: FlowNodeId, - ) -> error::Result { + ) -> impl Future> { + let loc = Location::caller(); // If not present, `get_or_insert_async` will lock the key until the future completes, // so only one thread will be able to fetch the data from the database and write it to // the file system and cache, hence no race on the file system. - NODES - .get_or_insert_async( - &node, - |(script, flow)| match flow { - Some(flow) => RawData::Flow(Arc::new(flow)), - _ => RawData::Script(Arc::new(script)), - }, - async { - sqlx::query!( - "SELECT \ - lock AS \"lock: String\", \ - code AS \"code: String\", \ - flow AS \"flow: Json>\" \ - FROM flow_node WHERE id = $1 LIMIT 1", - node.0, + NODES.get_or_insert_async( + node, + |(script, flow)| match flow { + Some(flow) => RawData::Flow(Arc::new(flow)), + _ => RawData::Script(Arc::new(script)), + }, + async move { + sqlx::query!( + "SELECT \ + lock AS \"lock: String\", \ + code AS \"code: String\", \ + flow AS \"flow: Json>\" \ + FROM flow_node WHERE id = $1 LIMIT 1", + node.0, + ) + .fetch_optional(e) + .await + .map_err(Into::into) + .and_then(unwrap_or_error(&loc, "Flow node", node)) + .map(|r| { + ( + ScriptData::from_raw(r.lock, r.code), + r.flow.map(|Json(raw_flow)| FlowData::from_raw(raw_flow)), ) - .fetch_one(e) - .await - .map_err(Into::into) - .map(|r| { - ( - ScriptData::from_raw(r.lock, r.code), - r.flow.map(|Json(raw_flow)| FlowData::from_raw(raw_flow)), - ) - }) - }, - ) - .await + }) + }, + ) } - pub async fn fetch_version(e: impl PgExecutor<'_>, id: i64) -> error::Result> { - FLOWS - .get_or_insert_async(&id, Arc::new, async { - sqlx::query_scalar!( - "SELECT value AS \"value!: Json>\" - FROM flow_version WHERE id = $1 LIMIT 1", - id, - ) - .fetch_one(e) - .await - .map_err(Into::into) - .map(|Json(raw_flow)| FlowData::from_raw(raw_flow)) - }) - .await - } - - pub async fn fetch_version_lite( - e: impl PgExecutor<'_>, + #[track_caller] + pub fn fetch_version<'c>( + e: impl PgExecutor<'c>, id: i64, - ) -> error::Result> { - FLOWS_LITE - .get_or_insert_async(&id, Arc::new, async { - sqlx::query_scalar!( - "SELECT value AS \"value!: Json>\" - FROM flow_version_lite WHERE id = $1 LIMIT 1", - id, - ) - .fetch_one(e) - .await - .map_err(Into::into) - .map(|Json(raw_flow)| FlowData::from_raw(raw_flow)) - }) + ) -> impl Future>> { + let loc = Location::caller(); + FLOWS.get_or_insert_async(id, Arc::new, async move { + sqlx::query_scalar!( + "SELECT value AS \"value!: Json>\" + FROM flow_version WHERE id = $1 LIMIT 1", + id, + ) + .fetch_optional(e) .await + .map_err(Into::into) + .and_then(unwrap_or_error(&loc, "Flow version", id)) + .map(|Json(raw_flow)| FlowData::from_raw(raw_flow)) + }) + } + + #[track_caller] + pub fn fetch_version_lite<'c>( + e: impl PgExecutor<'c>, + id: i64, + ) -> impl Future>> { + let loc = Location::caller(); + FLOWS_LITE.get_or_insert_async(id, Arc::new, async move { + sqlx::query_scalar!( + "SELECT value AS \"value!: Json>\" + FROM flow_version_lite WHERE id = $1 LIMIT 1", + id, + ) + .fetch_optional(e) + .await + .map_err(Into::into) + .and_then(unwrap_or_error(&loc, "Flow version \"lite\"", id)) + .map(|Json(raw_flow)| FlowData::from_raw(raw_flow)) + }) } } @@ -422,44 +445,41 @@ pub mod script { /// If not present, import from the file-system cache or fetch it from the database and write /// it to the file system and cache. /// This should be preferred over fetching the database directly. - pub async fn fetch( - e: impl PgExecutor<'_>, + #[track_caller] + pub fn fetch<'c>( + e: impl PgExecutor<'c>, hash: ScriptHash, - ) -> error::Result<(Arc, Arc)> { + ) -> impl Future, Arc)>> { // If not present, `get_or_insert_async` will lock the key until the future completes, // so only one thread will be able to fetch the data from the database and write it to // the file system and cache, hence no race on the file system. - CACHE - .get_or_insert_async( - &hash, - |(data, metadata)| (Arc::new(data), Arc::new(metadata)), - async { - sqlx::query!( - "SELECT \ - lock AS \"lock: String\", \ - content AS \"code!: String\", - language AS \"language: Option\", \ - envs AS \"envs: Vec\", \ - codebase AS \"codebase: String\" \ - FROM script WHERE hash = $1 LIMIT 1", - hash.0 + let loc = Location::caller(); + CACHE.get_or_insert_async( + hash, + |(data, metadata)| (Arc::new(data), Arc::new(metadata)), + async move { + sqlx::query!( + "SELECT \ + lock AS \"lock: String\", \ + content AS \"code!: String\", + language AS \"language: Option\", \ + envs AS \"envs: Vec\", \ + codebase AS \"codebase: String\" \ + FROM script WHERE hash = $1 LIMIT 1", + hash.0 + ) + .fetch_optional(e) + .await + .map_err(Into::into) + .and_then(unwrap_or_error(&loc, "Script", hash)) + .map(|r| { + ( + ScriptData::from_raw(r.lock, Some(r.code)), + ScriptMetadata { language: r.language, envs: r.envs, codebase: r.codebase }, ) - .fetch_one(e) - .await - .map_err(Into::into) - .map(|r| { - ( - ScriptData::from_raw(r.lock, Some(r.code)), - ScriptMetadata { - language: r.language, - envs: r.envs, - codebase: r.codebase, - }, - ) - }) - }, - ) - .await + }) + }, + ) } /// Invalidate the script cache for the given `hash`. @@ -480,25 +500,26 @@ pub mod app { /// If not present, import from the file-system cache or fetch it from the database and write /// it to the file system and cache. /// This should be preferred over fetching the database directly. - pub async fn fetch_script( - e: impl PgExecutor<'_>, + #[track_caller] + pub fn fetch_script<'c>( + e: impl PgExecutor<'c>, id: AppScriptId, - ) -> error::Result> { + ) -> impl Future>> { // If not present, `get_or_insert_async` will lock the key until the future completes, // so only one thread will be able to fetch the data from the database and write it to // the file system and cache, hence no race on the file system. - CACHE - .get_or_insert_async(&id, Arc::new, async { - sqlx::query!( - "SELECT lock, code FROM app_script WHERE id = $1 LIMIT 1", - id.0, - ) - .fetch_one(e) - .await - .map_err(Into::into) - .map(|r| ScriptData::from_raw(r.lock, Some(r.code))) - }) + let loc = Location::caller(); + CACHE.get_or_insert_async(id, Arc::new, async move { + sqlx::query!( + "SELECT lock, code FROM app_script WHERE id = $1 LIMIT 1", + id.0, + ) + .fetch_optional(e) .await + .map_err(Into::into) + .and_then(unwrap_or_error(&loc, "Application script", id)) + .map(|r| ScriptData::from_raw(r.lock, Some(r.code))) + }) } } @@ -513,108 +534,123 @@ pub mod job { static ref PREVIEWS: Cache = Cache::new(50); } - pub async fn fetch_preview_flow( - e: impl PgExecutor<'_>, - job: &Uuid, + #[track_caller] + pub fn fetch_preview_flow<'a, 'c>( + e: impl PgExecutor<'c> + 'a, + job: &'a Uuid, // original raw values from `queue` or `completed_job` tables: // kept for backward compatibility. raw_flow: Option>>, - ) -> error::Result> { - fetch_preview(e, job, None, None, raw_flow) - .await - .and_then(|data| match data { + ) -> impl Future>> + 'a { + let fetch_preview = fetch_preview(e, job, None, None, raw_flow); + async move { + fetch_preview.await.and_then(|data| match data { RawData::Flow(data) => Ok(data), RawData::Script(_) => Err(error::Error::InternalErr(format!( "Job ({job}) isn't a flow job." ))), }) + } } - pub async fn fetch_preview_script( - e: impl PgExecutor<'_>, - job: &Uuid, + #[track_caller] + pub fn fetch_preview_script<'a, 'c>( + e: impl PgExecutor<'c> + 'a, + job: &'a Uuid, // original raw values from `queue` or `completed_job` tables: // kept for backward compatibility. raw_lock: Option, raw_code: Option, - ) -> error::Result> { - fetch_preview(e, job, raw_lock, raw_code, None) - .await - .and_then(|data| match data { + ) -> impl Future>> + 'a { + let fetch_preview = fetch_preview(e, job, raw_lock, raw_code, None); + async move { + fetch_preview.await.and_then(|data| match data { RawData::Script(data) => Ok(data), RawData::Flow(_) => Err(error::Error::InternalErr(format!( "Job ({job}) isn't a script job." ))), }) + } } - pub async fn fetch_preview( - e: impl PgExecutor<'_>, - job: &Uuid, + #[track_caller] + pub fn fetch_preview<'a, 'c>( + e: impl PgExecutor<'c> + 'a, + job: &'a Uuid, // original raw values from `queue` or `completed_job` tables: // kept for backward compatibility. raw_lock: Option, raw_code: Option, raw_flow: Option>>, - ) -> error::Result { - PREVIEWS - .get_or_insert_async(job, async { - match (raw_lock, raw_code, raw_flow) { - (None, None, None) => sqlx::query!( - "SELECT raw_code, raw_lock, raw_flow AS \"raw_flow: Json>\" \ - FROM job WHERE id = $1 LIMIT 1", - job - ) - .fetch_one(e) - .map_err(Into::into) - .await - .map(|r| (r.raw_lock, r.raw_code, r.raw_flow)), - (lock, code, flow) => Ok((lock, code, flow)), - } - .map(|(lock, code, flow)| match flow { - Some(Json(flow)) => RawData::Flow(Arc::new(FlowData::from_raw(flow))), - _ => RawData::Script(Arc::new(ScriptData::from_raw(lock, code))), - }) + ) -> impl Future> + 'a { + let loc = Location::caller(); + PREVIEWS.get_or_insert_async(job, async move { + match (raw_lock, raw_code, raw_flow) { + (None, None, None) => sqlx::query!( + "SELECT raw_code, raw_lock, raw_flow AS \"raw_flow: Json>\" \ + FROM job WHERE id = $1 LIMIT 1", + job + ) + .fetch_optional(e) + .await + .map_err(Into::into) + .and_then(unwrap_or_error(&loc, "Preview", job)) + .map(|r| (r.raw_lock, r.raw_code, r.raw_flow)), + (lock, code, flow) => Ok((lock, code, flow)), + } + .map(|(lock, code, flow)| match flow { + Some(Json(flow)) => RawData::Flow(Arc::new(FlowData::from_raw(flow))), + _ => RawData::Script(Arc::new(ScriptData::from_raw(lock, code))), }) - .await + }) } - pub async fn fetch_script( - e: impl PgExecutor<'_>, + #[track_caller] + pub fn fetch_script<'c>( + e: impl PgExecutor<'c>, kind: JobKind, hash: Option, - ) -> error::Result> { + ) -> impl Future>> { use JobKind::*; - match (kind, hash.map(|ScriptHash(id)| id)) { - (FlowScript, Some(id)) => flow::fetch_script(e, FlowNodeId(id)).await, - (Script | Dependencies, Some(hash)) => script::fetch(e, ScriptHash(hash)) - .await - .map(|(raw_script, _metadata)| raw_script), - (AppScript, Some(id)) => app::fetch_script(e, AppScriptId(id)).await, - _ => Err(error::Error::InternalErr(format!( - "Isn't a script job: {:?}", - kind - ))), + let loc = Location::caller(); + async move { + match (kind, hash.map(|ScriptHash(id)| id)) { + (FlowScript, Some(id)) => flow::fetch_script(e, FlowNodeId(id)).await, + (Script | Dependencies, Some(hash)) => script::fetch(e, ScriptHash(hash)) + .await + .map(|(raw_script, _metadata)| raw_script), + (AppScript, Some(id)) => app::fetch_script(e, AppScriptId(id)).await, + _ => Err(error::Error::InternalErr(format!( + "Isn't a script job: {:?}", + kind + ))), + } + .map_err(error::relocate_internal(loc)) } } - pub async fn fetch_flow( - e: impl PgExecutor<'_> + Copy, + #[track_caller] + pub fn fetch_flow<'c>( + e: impl PgExecutor<'c> + Copy, kind: JobKind, hash: Option, - ) -> error::Result> { + ) -> impl Future>> { use JobKind::*; - match (kind, hash.map(|ScriptHash(id)| id)) { - (FlowDependencies, Some(id)) => flow::fetch_version(e, id).await, - (FlowNode, Some(id)) => flow::fetch_flow(e, FlowNodeId(id)).await, - (Flow, Some(id)) => match flow::fetch_version_lite(e, id).await { - Ok(raw_flow) => Ok(raw_flow), - Err(_) => flow::fetch_version(e, id).await, - }, - _ => Err(error::Error::InternalErr(format!( - "Isn't a flow job {:?}", - kind - ))), + let loc = Location::caller(); + async move { + match (kind, hash.map(|ScriptHash(id)| id)) { + (FlowDependencies, Some(id)) => flow::fetch_version(e, id).await, + (FlowNode, Some(id)) => flow::fetch_flow(e, FlowNodeId(id)).await, + (Flow, Some(id)) => match flow::fetch_version_lite(e, id).await { + Ok(raw_flow) => Ok(raw_flow), + Err(_) => flow::fetch_version(e, id).await, + }, + _ => Err(error::Error::InternalErr(format!( + "Isn't a flow job {:?}", + kind + ))), + } + .map_err(error::relocate_internal(loc)) } } } diff --git a/backend/windmill-common/src/error.rs b/backend/windmill-common/src/error.rs index 7b959ebed1..0080568f5e 100644 --- a/backend/windmill-common/src/error.rs +++ b/backend/windmill-common/src/error.rs @@ -6,6 +6,8 @@ * LICENSE-AGPL for a copy of the license. */ +use std::panic::Location; + use axum::body::Body; use axum::response::Response; use axum::{response::IntoResponse, response::Json}; @@ -48,6 +50,8 @@ pub enum Error { QuotaExceeded(String), #[error("Internal: {0}")] InternalErr(String), + #[error("Internal: {0}: {1}")] + InternalErrAt(&'static Location<'static>, String), #[error("Hexadecimal decoding error: {0}")] HexErr(#[from] hex::FromHexError), #[error("Migrating database: {0}")] @@ -77,6 +81,17 @@ impl Error { pub fn dbg(&self) -> String { format!("{:?}", self) } + + pub fn relocate_internal(self, loc: &'static Location<'static>) -> Self { + match self { + Self::InternalErr(s) | Self::InternalErrAt(_, s) => Self::InternalErrAt(loc, s), + _ => self, + } + } +} + +pub fn relocate_internal(loc: &'static Location<'static>) -> impl FnOnce(Error) -> Error { + move |e| e.relocate_internal(loc) } pub fn to_anyhow(e: T) -> anyhow::Error { @@ -92,10 +107,9 @@ impl IntoResponse for Error { Self::NotFound(_) => axum::http::StatusCode::NOT_FOUND, Self::NotAuthorized(_) => axum::http::StatusCode::UNAUTHORIZED, Self::RequireAdmin(_) => axum::http::StatusCode::FORBIDDEN, - Self::SqlErr(_) - | Self::BadRequest(_) - | Self::AiError(_) - | Self::QuotaExceeded(_) => axum::http::StatusCode::BAD_REQUEST, + Self::SqlErr(_) | Self::BadRequest(_) | Self::AiError(_) | Self::QuotaExceeded(_) => { + axum::http::StatusCode::BAD_REQUEST + } _ => axum::http::StatusCode::INTERNAL_SERVER_ERROR, };