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

fix(backend): correct early return with stream + prevent delta miss (#7872)

Browse Source
This commit is contained in:
hugocasa
2026-02-10 18:20:14 +01:00
committed by GitHub
parent ea0b18e9e8
commit 5a203fcdcb
5 changed files with 75 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-454a611a5a162b2ace137c139bd5383bc7fe142c515dac3edd47991839485e51.json
Generated
+20
View File
@@ -0,0 +1,20 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM job_result_stream_v2\n WHERE job_id NOT IN (SELECT id FROM v2_job_queue)\n AND job_id NOT IN (\n SELECT id FROM v2_job_completed\n WHERE completed_at > NOW() - INTERVAL '60 seconds'\n )\n RETURNING job_id",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "job_id",
"type_info": "Uuid"
}
],
"parameters": {
"Left": []
},
"nullable": [
false
]
},
"hash": "454a611a5a162b2ace137c139bd5383bc7fe142c515dac3edd47991839485e51"
}
backend/.sqlx/query-a3e75f0309be42aca0fd74834f34b3f18dbb388bd8b9bc88b99aebedae9c3fec.json
Generated
-20
View File
@@ -1,20 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM job_result_stream_v2 WHERE job_id NOT IN (SELECT id FROM v2_job_queue) RETURNING job_id",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "job_id",
"type_info": "Uuid"
}
],
"parameters": {
"Left": []
},
"nullable": [
false
]
},
"hash": "a3e75f0309be42aca0fd74834f34b3f18dbb388bd8b9bc88b99aebedae9c3fec"
}
backend/src/monitor.rs
+7 -1
View File
@@ -3092,7 +3092,13 @@ RETURNING job_id"
async fn cleanup_job_result_stream_orphaned_jobs(db: &DB) -> error::Result<()> {
let result = sqlx::query!(
"DELETE FROM job_result_stream_v2 WHERE job_id NOT IN (SELECT id FROM v2_job_queue) RETURNING job_id",
"DELETE FROM job_result_stream_v2
WHERE job_id NOT IN (SELECT id FROM v2_job_queue)
AND job_id NOT IN (
SELECT id FROM v2_job_completed
WHERE completed_at > NOW() - INTERVAL '60 seconds'
)
RETURNING job_id",
)
.fetch_all(db)
.await?;
backend/windmill-api/src/jobs.rs
+46 -16
View File
@@ -114,7 +114,10 @@ use windmill_common::{
get_latest_flow_version_info_for_path, get_script_info_for_hash, utils::empty_as_none,
ScriptHashInfo, BASE_URL,
};
use windmill_queue::{job_is_complete, push, PushArgs, PushArgsOwned, PushIsolationLevel};
use windmill_queue::{
get_result_and_success_by_id_from_flow, job_is_complete, push, PushArgs, PushArgsOwned,
PushIsolationLevel,
};
pub fn workspaced_service() -> Router {
let cors = CorsLayer::new()
@@ -3471,8 +3474,11 @@ pub async fn run_workflow_as_code(
let JobExtended { inner: job, raw_code, raw_lock, .. } = job;
let (_debouncing_settings, concurrency_settings) =
windmill_common::runnable_settings::prefetch_cached_from_handle(job.runnable_settings_handle, &db)
.await?;
windmill_common::runnable_settings::prefetch_cached_from_handle(
job.runnable_settings_handle,
&db,
)
.await?;
let (job_payload, tag, _delete_after_use, timeout, on_behalf_of) = match job.job_kind {
JobKind::Preview => (
@@ -4232,7 +4238,7 @@ pub async fn stream_job(
};
let poll_delay_ms = run_query.poll_delay_ms;
let uuid = match runnable_id {
let (uuid, early_return) = match runnable_id {
RunnableId::ScriptId(ScriptId::ScriptPath(script_path))
| RunnableId::HubScript(script_path) => {
let (uuid, _, _) = push_script_job_by_path_into_queue(
@@ -4247,10 +4253,10 @@ pub async fn stream_job(
None,
)
.await?;
uuid
(uuid, None)
}
RunnableId::ScriptId(ScriptId::ScriptHash(script_hash)) => {
run_job_by_hash_inner(
let (uuid, _) = run_job_by_hash_inner(
authed.clone(),
db.clone(),
user_db,
@@ -4260,11 +4266,11 @@ pub async fn stream_job(
args,
None,
)
.await?
.0
.await?;
(uuid, None)
}
RunnableId::FlowId(FlowId::FlowPath(flow_path)) => {
push_flow_job_by_path_into_queue(
let (uuid, early_return, _) = push_flow_job_by_path_into_queue(
authed.clone(),
db.clone(),
None,
@@ -4275,11 +4281,11 @@ pub async fn stream_job(
args,
None,
)
.await?
.0
.await?;
(uuid, early_return)
}
RunnableId::FlowId(FlowId::FlowVersion(version)) => {
run_flow_by_version_inner(
let (uuid, early_return) = run_flow_by_version_inner(
authed.clone(),
db.clone(),
user_db,
@@ -4289,8 +4295,8 @@ pub async fn stream_job(
args,
None,
)
.await?
.0
.await?;
(uuid, early_return)
}
};
@@ -4321,6 +4327,7 @@ pub async fn stream_job(
None,
tx,
poll_delay_ms,
early_return,
);
let body = axum::body::Body::from_stream(stream.map(Result::<_, std::convert::Infallible>::Ok));
@@ -4643,12 +4650,19 @@ fn register_potential_assets_on_inline_execution(
let columns = asset.columns.as_ref().map(|cols| {
cols.iter()
.map(|(col_name, col_access_type)| {
(col_name.clone(), windmill_common::assets::asset_access_type_from_parser(*col_access_type))
(
col_name.clone(),
windmill_common::assets::asset_access_type_from_parser(
*col_access_type,
),
)
})
.collect()
});
register_runtime_asset(InsertRuntimeAssetParams {
access_type: asset.access_type.map(windmill_common::assets::asset_access_type_from_parser),
access_type: asset
.access_type
.map(windmill_common::assets::asset_access_type_from_parser),
asset_kind: windmill_common::assets::asset_kind_from_parser(asset.kind),
asset_path: asset.path,
columns,
@@ -5825,6 +5839,7 @@ async fn get_job_update(
no_logs,
is_flow,
None,
None,
)
.await?,
))
@@ -5865,6 +5880,7 @@ async fn get_job_update_sse(
is_flow,
tx,
poll_delay_ms,
None,
);
let stream = tokio_stream::wrappers::ReceiverStream::new(rx).map(|x| {
@@ -5902,6 +5918,7 @@ pub fn start_job_update_sse_stream(
is_flow: Option<bool>,
tx: tokio::sync::mpsc::Sender<JobUpdateSSEStream>,
poll_delay_ms: Option<u64>,
early_return: Option<String>,
) -> () {
tokio::spawn(async move {
let mut log_offset = initial_log_offset;
@@ -5929,6 +5946,7 @@ pub fn start_job_update_sse_stream(
no_logs,
is_flow,
flow_stream_job_id,
early_return.as_deref(),
)
.await
{
@@ -6046,6 +6064,7 @@ pub fn start_job_update_sse_stream(
no_logs,
is_flow,
flow_stream_job_id,
early_return.as_deref(),
)
.await
{
@@ -6176,6 +6195,7 @@ async fn get_job_update_data(
no_logs: Option<bool>,
is_flow: Option<bool>,
flow_stream_job_id: Option<Uuid>,
early_return: Option<&str>,
) -> error::Result<JobUpdate> {
let tags = if log_view {
log_job_view(
@@ -6339,6 +6359,16 @@ async fn get_job_update_data(
let flow_stream_job_id = flow_stream_job_id.or(new_flow_stream_job_id);
let result = if let Some(early_return) = early_return {
match get_result_and_success_by_id_from_flow(db, w_id, job_id, early_return, None).await
{
Ok((early_result, _)) => Some(early_result),
Err(_) => result,
}
} else {
result
};
let flow_stream_delta =
get_flow_stream_delta(db, flow_stream_job_id, stream_offset).await?;
backend/windmill-api/src/triggers/http/handler.rs
+2 -1
View File
@@ -505,7 +505,7 @@ async fn route_job(
match trigger.request_type {
RequestType::SyncSse => {
// Trigger the job (always async when streaming)
let (uuid, _, _, _) = trigger_runnable_inner(
let (uuid, _, early_return, _) = trigger_runnable_inner(
&db,
None,
Some(user_db.clone()),
@@ -553,6 +553,7 @@ async fn route_job(
None,
tx,
None,
early_return,
);
let body = axum::body::Body::from_stream(