fix: canceling jobs

This commit is contained in:
Ruben Fiszel
2023-08-27 14:40:20 +02:00
parent b492fd9884
commit 0dfdf8fa1b
5 changed files with 37 additions and 6 deletions

View File

@@ -11,6 +11,10 @@ INSERT INTO usr(workspace_id, email, username, is_admin, role) VALUES
INSERT INTO workspace_key(workspace_id, kind, key) VALUES
('test-workspace', 'cloud', 'test-key');
INSERT INTO workspace_settings (workspace_id) VALUES
('test-workspace');
insert INTO token(token, email, label, super_admin) VALUES ('SECRET_TOKEN', 'test@windmill.dev', 'test token', true);
INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES (

View File

@@ -2549,14 +2549,14 @@ async fn test_flow_lock_all(db: Pool<Postgres>) {
assert!(matches!(
m.value,
windmill_api_client::types::FlowModuleValue::RawScript(RawScript {
language: windmill_api_client::types::RawScriptLanguage::Deno | windmill_api_client::types::RawScriptLanguage::Bash,
language: windmill_api_client::types::RawScriptLanguage::Bash,
lock: Some(ref lock),
..
}) if lock == "")
|| matches!(
m.value,
windmill_api_client::types::FlowModuleValue::RawScript(RawScript{
language: windmill_api_client::types::RawScriptLanguage::Go | windmill_api_client::types::RawScriptLanguage::Python3,
language: windmill_api_client::types::RawScriptLanguage::Go | windmill_api_client::types::RawScriptLanguage::Python3 | windmill_api_client::types::RawScriptLanguage::Deno,
lock: Some(ref lock),
..
}) if lock.len() > 0)

View File

@@ -673,6 +673,26 @@ async fn cancel_all(
Ok(Json(uuids))
}
// #[derive(Serialize, Debug, FromRow)]
// struct QueueStats {
// database_length: i64,
// }
// async fn count_waiting_for_executors(
// Extension(db): Extension<DB>,
// Path(w_id): Path<String>,
// ) -> error::JsonResult<QueueStats> {
// Ok(Json(
// sqlx::query_as!(
// QueueStats,
// "SELECT coalesce(COUNT(*), 0) as \"database_length!\" FROM queue WHERE workspace_id = $1",
// w_id
// )
// .fetch_one(&db)
// .await?,
// ))
// }
#[derive(Serialize, Debug, FromRow)]
struct QueueStats {
database_length: i64,

View File

@@ -94,7 +94,13 @@ impl IntoResponse for Error {
}
_ => axum::http::StatusCode::INTERNAL_SERVER_ERROR,
};
tracing::error!(error = e.to_string());
if matches!(status, axum::http::StatusCode::NOT_FOUND) {
tracing::warn!(not_found = e.to_string());
} else {
tracing::error!(error = e.to_string());
};
axum::response::Response::builder()
.header("Content-Type", "text/plain")
.status(status)

View File

@@ -140,9 +140,9 @@ pub async fn cancel_job<'c: 'async_recursion>(
return Ok((tx, None));
}
let job_running = job_running.unwrap();
if job_running.running
&& job_running.job_kind != JobKind::Flow
&& job_running.job_kind != JobKind::FlowPreview
if ((!job_running.running && job_running.root_job.is_none())
|| (job_running.job_kind == JobKind::Flow || job_running.job_kind == JobKind::FlowPreview))
&& !force_cancel
{
sqlx::query!(
@@ -362,6 +362,7 @@ pub async fn add_completed_job<R: rsmq_async::RsmqConnection + Clone + Send>(
.await
.map_err(|e| Error::InternalErr(format!("Could not add completed job {job_id}: {e}")))?;
// tracing::error!("Added completed job {:#?}", queued_job);
tx = delete_job(tx, &queued_job.workspace_id, job_id).await?;
if !queued_job.is_flow_step
&& queued_job.schedule_path.is_some()