Skip to content
This repository was archived by the owner on Oct 19, 2026. It is now read-only.

Commit ca49c16

Browse files
AidChengAidCheng
authored
[buck2]ADD: BuildComplete Message Handler Compatible with new DB (#1997)
* [buck2]ADD: BuildComplete Message Handler Compatible with new DB Signed-off-by: AidCheng <cn.aiden.cheng@gmail.com> * [orion]FIX: fixes fatal data model api bugs Signed-off-by: AidCheng <cn.aiden.cheng@gmail.com> * [buck2]UPDATE:fmt Signed-off-by: AidCheng <cn.aiden.cheng@gmail.com> * [buck2]FIX: change Err enum detection to is_err() Signed-off-by: AidCheng <cn.aiden.cheng@gmail.com> --------- Signed-off-by: AidCheng <cn.aiden.cheng@gmail.com>
1 parent 448b9d2 commit ca49c16

5 files changed

Lines changed: 313 additions & 7 deletions

File tree

‎api-model/src/buck2/ws.rs‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,14 @@ pub enum WSMessage {
5959
exit_code: Option<i32>,
6060
message: String,
6161
},
62+
63+
TaskBuildCompleteV2 {
64+
build_id: String,
65+
success: bool,
66+
exit_code: Option<i32>,
67+
message: String,
68+
},
69+
6270
/// Batch of target build status updates for real-time build progress tracking.
6371
TargetBuildStatusBatch {
6472
events: Vec<WSTargetBuildStatusEvent>,

‎orion-server/src/api.rs‎

Lines changed: 166 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ use crate::{
4848
auto_retry::AutoRetryJudger,
4949
log::log_service::LogService,
5050
model::{
51-
build_events::BuildEventDTO,
51+
build_events::{BuildEvent, BuildEventDTO},
5252
build_targets::BuildTarget,
5353
builds,
5454
orion_tasks::{OrionTask, OrionTaskDTO},
@@ -143,6 +143,7 @@ pub fn routers() -> Router<AppState> {
143143
get(get_orion_client_status_by_id),
144144
)
145145
.route("/retry-build", post(build_retry_handler))
146+
.route("/v2/task-handler", get(task_handler_v2))
146147
.route("/v2/health", get(health_check_handler))
147148
.route("/v2/task-retry/{id}", post(task_retry_handler))
148149
.route("/v2/task/{cl}", get(task_get_handler))
@@ -507,7 +508,6 @@ pub async fn target_logs_handler(
507508
}
508509
}
509510

510-
#[allow(dead_code)]
511511
#[utoipa::path(
512512
post,
513513
path = "/v2/task",
@@ -778,7 +778,6 @@ async fn handle_immediate_task_dispatch_v2(
778778
return activate_worker(&build_info, &state.scheduler).await;
779779
}
780780

781-
#[allow(dead_code)]
782781
async fn activate_worker(build_info: &BuildInfo, scheduler: &TaskScheduler) -> OrionBuildResult {
783782
// Create WS Message
784783
let msg = WSMessage::TaskBuild {
@@ -1182,6 +1181,170 @@ async fn process_message(
11821181
build_info.auto_retry_judger.judge_by_output(&output);
11831182
}
11841183
}
1184+
WSMessage::TaskBuildCompleteV2 {
1185+
build_id,
1186+
success,
1187+
exit_code,
1188+
message,
1189+
} => {
1190+
// Handle build completion
1191+
tracing::info!(
1192+
"Build {build_id} completed by worker {current_worker_id} with exit code: {exit_code:?}"
1193+
);
1194+
1195+
// Get build information
1196+
let (
1197+
mut auto_retry_judger,
1198+
mut retry_count,
1199+
repo,
1200+
changes,
1201+
cl_link,
1202+
task_id,
1203+
_target_id,
1204+
_target_path,
1205+
) = if let Some(build_info) = state.scheduler.active_builds.get(&build_id) {
1206+
(
1207+
build_info.auto_retry_judger.clone(),
1208+
build_info.event_payload.retry_count,
1209+
build_info.event_payload.repo.clone(),
1210+
build_info.changes.clone(),
1211+
build_info.event_payload.cl_link.clone(),
1212+
build_info.event_payload.task_id,
1213+
build_info.target_id,
1214+
build_info.target_path.clone(),
1215+
)
1216+
} else {
1217+
tracing::error!("Not found build {build_id}");
1218+
return ControlFlow::Continue(());
1219+
};
1220+
1221+
// Judge auto retry by exit code
1222+
auto_retry_judger.judge_by_exit_code(exit_code.unwrap_or(0));
1223+
1224+
let can_auto_retry = auto_retry_judger.get_can_auto_retry();
1225+
1226+
if can_auto_retry && retry_count < RETRY_COUNT_MAX {
1227+
tracing::info!(
1228+
"Build {build_id} will retry, current retry count: {retry_count}"
1229+
);
1230+
1231+
// Add retry count
1232+
retry_count += 1;
1233+
1234+
// Update build information
1235+
if let Some(mut build_info) =
1236+
state.scheduler.active_builds.get_mut(&build_id)
1237+
{
1238+
build_info.event_payload.retry_count = retry_count;
1239+
// New AutoRetryJudger
1240+
build_info.auto_retry_judger = AutoRetryJudger::new();
1241+
}
1242+
1243+
// Update database
1244+
let _ = builds::Entity::update_many()
1245+
.set(builds::ActiveModel {
1246+
retry_count: Set(retry_count),
1247+
..Default::default()
1248+
})
1249+
.filter(builds::Column::Id.eq(build_id.parse::<uuid::Uuid>().unwrap()))
1250+
.exec(&state.conn)
1251+
.await;
1252+
1253+
// Send task to this worker
1254+
let msg = WSMessage::TaskBuild {
1255+
build_id: build_id.clone(),
1256+
repo: repo.clone(),
1257+
cl_link,
1258+
changes,
1259+
};
1260+
if let Some(worker) = state.scheduler.workers.get_mut(current_worker_id)
1261+
&& worker.sender.send(msg).is_ok()
1262+
{
1263+
tracing::info!(
1264+
"Retry build: {}, worker: {}",
1265+
build_id,
1266+
current_worker_id
1267+
);
1268+
return ControlFlow::Continue(());
1269+
}
1270+
}
1271+
1272+
// Send final log event
1273+
let log_event = LogEvent {
1274+
task_id: task_id.to_string(),
1275+
repo_name: LogService::last_segment(&repo).to_string(),
1276+
build_id: build_id.to_string(),
1277+
line: String::from(""),
1278+
is_end: true,
1279+
};
1280+
state.log_service.publish(log_event);
1281+
1282+
// Remove from active
1283+
state.scheduler.active_builds.remove(&build_id);
1284+
1285+
// TODO: update to jupiter's build_event's model
1286+
if BuildEvent::update_build_complete_result(
1287+
&build_id,
1288+
exit_code,
1289+
success,
1290+
&message,
1291+
&state.conn,
1292+
)
1293+
.await
1294+
.is_err()
1295+
{
1296+
tracing::error!(
1297+
"Failed to update build complete result for build: {}",
1298+
build_id
1299+
);
1300+
}
1301+
1302+
// Update target state
1303+
let target_state = match (success, exit_code) {
1304+
(true, Some(0)) => TargetState::Completed,
1305+
(_, None) => TargetState::Interrupted,
1306+
_ => TargetState::Failed,
1307+
};
1308+
let mut _error_summary = None;
1309+
if matches!(target_state, TargetState::Failed) {
1310+
let repo_segment = LogService::last_segment(&repo);
1311+
if let Ok(log_content) = state
1312+
.log_service
1313+
.read_full_log(
1314+
&task_id.to_string(),
1315+
&repo_segment,
1316+
&build_id.to_string(),
1317+
)
1318+
.await
1319+
{
1320+
_error_summary =
1321+
find_caused_by_next_line_in_content(&log_content).await;
1322+
}
1323+
}
1324+
1325+
if let Err(e) =
1326+
BuildTarget::update_build_targets(target_state, &build_id, &state.conn)
1327+
.await
1328+
{
1329+
tracing::error!(
1330+
"unable to update build targets for build {}: {}",
1331+
build_id,
1332+
e,
1333+
)
1334+
}
1335+
1336+
// Mark the worker as idle or error depending on whether the task succeeds.
1337+
if let Some(mut worker) = state.scheduler.workers.get_mut(current_worker_id) {
1338+
worker.status = if success {
1339+
WorkerStatus::Idle
1340+
} else {
1341+
WorkerStatus::Error(message)
1342+
};
1343+
}
1344+
1345+
// Notify scheduler to process queued tasks
1346+
state.scheduler.notify_task_available();
1347+
}
11851348
WSMessage::TaskBuildComplete {
11861349
build_id,
11871350
success,

‎orion-server/src/model/build_events.rs‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
use chrono::Utc;
2+
use sea_orm::{ActiveValue::Set, ColumnTrait, ConnectionTrait, DbErr, EntityTrait, QueryFilter};
23
use serde::Serialize;
34
use utoipa::ToSchema;
5+
use uuid::Uuid;
46

57
#[derive(ToSchema, Serialize)]
68
pub struct BuildEventDTO {
@@ -28,3 +30,25 @@ impl From<&callisto::build_events::Model> for BuildEventDTO {
2830
}
2931
}
3032
}
33+
34+
pub struct BuildEvent;
35+
36+
impl BuildEvent {
37+
pub async fn update_build_complete_result(
38+
build_id: &str,
39+
exit_code: Option<i32>,
40+
_success: bool,
41+
_message: &str,
42+
db_connection: &impl ConnectionTrait,
43+
) -> Result<(), DbErr> {
44+
callisto::build_events::Entity::update_many()
45+
.filter(callisto::build_events::Column::Id.eq(build_id.parse::<Uuid>().unwrap()))
46+
.set(callisto::build_events::ActiveModel {
47+
exit_code: Set(exit_code),
48+
..Default::default()
49+
})
50+
.exec(db_connection)
51+
.await?;
52+
Ok(())
53+
}
54+
}

‎orion-server/src/model/build_targets.rs‎

Lines changed: 87 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,14 @@
1-
use sea_orm::{ActiveModelTrait, ConnectionTrait, DbErr, IntoActiveModel};
1+
use sea_orm::{
2+
ActiveModelTrait, ColumnTrait, ConnectionTrait, DbErr, EntityTrait, IntoActiveModel,
3+
QueryFilter,
4+
};
25
use uuid::Uuid;
36

4-
#[allow(dead_code)]
7+
use crate::model::targets::TargetState;
58
/// A collection of utility methods for the `build_targets` database table.
69
pub struct BuildTarget;
710

811
impl BuildTarget {
9-
#[allow(dead_code)]
1012
pub fn create_default_target(id: Uuid, task_id: Uuid) -> callisto::build_targets::Model {
1113
let default_path = "//";
1214
callisto::build_targets::Model {
@@ -17,7 +19,6 @@ impl BuildTarget {
1719
}
1820
}
1921

20-
#[allow(dead_code)]
2122
pub async fn insert_default_target(
2223
id: Uuid,
2324
task_id: Uuid,
@@ -28,4 +29,86 @@ impl BuildTarget {
2829
target.into_active_model().insert(db).await?;
2930
Ok(path.to_string())
3031
}
32+
33+
#[allow(dead_code)]
34+
pub async fn find_build_targets(
35+
build_id: Uuid,
36+
task_id: Uuid,
37+
db: &impl ConnectionTrait,
38+
) -> Result<Vec<BuildTargetDTO>, DbErr> {
39+
// Get all targets of corresponding build
40+
let all_targets = callisto::build_targets::Entity::find()
41+
.filter(callisto::build_targets::Column::TaskId.eq(task_id))
42+
.all(db)
43+
.await?;
44+
45+
let mut result = Vec::with_capacity(all_targets.len());
46+
47+
// Search corresponding target state under current build
48+
for target in all_targets {
49+
let status = callisto::target_state_histories::Entity::find()
50+
.filter(callisto::target_state_histories::Column::BuildTargetId.eq(target.id))
51+
.filter(callisto::target_state_histories::Column::BuildEventId.eq(build_id))
52+
.one(db)
53+
.await?;
54+
55+
let state = match status {
56+
Some(s) => TargetState::from(s.target_state),
57+
None => TargetState::Pending,
58+
};
59+
60+
result.push(BuildTargetDTO {
61+
id: target.id,
62+
path: target.path,
63+
state,
64+
});
65+
}
66+
67+
Ok(result)
68+
}
69+
70+
pub async fn update_build_targets(
71+
target_state: TargetState,
72+
build_id: &str,
73+
db: &impl ConnectionTrait,
74+
) -> Result<(), DbErr> {
75+
let build_id = Uuid::parse_str(build_id)
76+
.map_err(|e| DbErr::Custom(format!("Invalid build_id format: {}", e)))?;
77+
let task_id = callisto::build_events::Entity::find_by_id(build_id)
78+
.one(db)
79+
.await?
80+
.ok_or_else(|| {
81+
DbErr::RecordNotFound(format!("Build event with id {} not found", build_id))
82+
})?
83+
.task_id;
84+
85+
let all_targets = callisto::build_targets::Entity::find()
86+
.filter(callisto::build_targets::Column::TaskId.eq(task_id))
87+
.all(db)
88+
.await?;
89+
90+
for target in all_targets {
91+
let status = callisto::target_state_histories::Entity::find()
92+
.filter(callisto::target_state_histories::Column::BuildTargetId.eq(target.id))
93+
.filter(callisto::target_state_histories::Column::BuildEventId.eq(build_id))
94+
.one(db)
95+
.await?;
96+
97+
if let Some(s) = status {
98+
let mut active: callisto::target_state_histories::ActiveModel =
99+
s.into_active_model();
100+
active.target_state = sea_orm::Set(target_state.to_string());
101+
active.update(db).await?;
102+
}
103+
}
104+
105+
Ok(())
106+
}
107+
}
108+
109+
#[allow(dead_code)]
110+
pub struct BuildTargetDTO {
111+
id: Uuid,
112+
path: String,
113+
state: TargetState,
31114
}

0 commit comments

Comments
 (0)