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

fix: handle better same_worker flow monitor

Browse Source
This commit is contained in:
Ruben Fiszel
2024-09-11 01:29:40 +02:00
parent 90f7ba47f2
commit 458dcfe1e8
4 changed files with 12 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e.json → backend/.sqlx/query-0ad36c1598ff4ece0c325eaeb9a9177a87e1accd192402e21db5ae09c3498ab0.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO queue\n (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, \n script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job, tag, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl, priority)\n VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30) RETURNING id",
"query": "INSERT INTO queue\n (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, \n script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job, tag, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl, priority, last_ping)\n VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, NULL) RETURNING id",
"describe": {
"columns": [
{
@@ -94,5 +94,5 @@
false
]
},
"hash": "620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e"
"hash": "0ad36c1598ff4ece0c325eaeb9a9177a87e1accd192402e21db5ae09c3498ab0"
}
backend/src/monitor.rs
+7 -5
View File
@@ -1379,11 +1379,13 @@ async fn handle_zombie_flows(
for flow in flows {
let status = flow.parse_flow_status();
if status.is_some_and(|s| {
s.modules
.get(0)
.is_some_and(|x| matches!(x, FlowStatusModule::WaitingForPriorSteps { .. }))
}) {
if !flow.same_worker
&& status.is_some_and(|s| {
s.modules
.get(0)
.is_some_and(|x| matches!(x, FlowStatusModule::WaitingForPriorSteps { .. }))
})
{
let error_message = format!(
"Zombie flow detected: {} in workspace {}. It hasn't started yet, restarting it.",
flow.id, flow.workspace_id
backend/windmill-queue/src/jobs.rs
+2 -2
View File
@@ -3810,8 +3810,8 @@ pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>(
script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, \
flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, \
visible_to_owner, root_job, tag, concurrent_limit, concurrency_time_window_s, timeout, \
flow_step_id, cache_ttl, priority)
VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30) \
flow_step_id, cache_ttl, priority, last_ping)
VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, NULL) \
RETURNING id",
workspace_id,
job_id,
backend/windmill-worker/src/worker.rs
+1 -1
View File
@@ -1456,7 +1456,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
if let Ok(same_worker_job) = same_worker_rx.try_recv() {
tracing::debug!("received {} from same worker channel", same_worker_job.job_id);
let r = sqlx::query_as::<_, QueuedJob>("SELECT * FROM queue WHERE id = $1")
let r = sqlx::query_as::<_, QueuedJob>("UPDATE queue SET last_ping = now() WHERE id = $1 RETURNING *")
.bind(same_worker_job.job_id)
.fetch_optional(db)
.await