warren/momentry_core
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases 2 Wiki Activity

fix: remove TKG guard to prevent deadlock

Browse Source 
The TKG guard was checking if nodes exist before spawning TKG build.
This caused a deadlock when:
1. TKG spawns (async)
2. ingestion_complete checks → false (nodes not yet created)
3. Returns Ok(false), job stays running
4. Next poll: guard checks nodes → no nodes yet (TKG still running)
5. Skips TKG → deadlock!

Since build_tkg uses ON CONFLICT (idempotent), it's safe to call
multiple times. Removed the guard to fix the deadlock.
This commit is contained in:
Accusys
2026-07-06 11:28:40 +08:00
parent 221aa4c4cc
commit 552f539bdf
1 changed files with 28 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/worker/job_worker.rs
+28 -41
View File
@@ -1839,54 +1839,41 @@ impl JobWorker {
}
// 🚀 P4 Trigger: TKG Build (Face + ASRX) → then Rule2 ingestion
// Note: build_tkg uses ON CONFLICT, so it's safe to call multiple times
if has_face && has_asrx {
// Guard: only spawn TKG if nodes don't exist yet
let tkg_nodes_table = schema::table_name("tkg_nodes");
let tkg_exists: bool = sqlx::query_scalar::<_, i32>(&format!(
"SELECT 1 FROM {tkg_nodes_table} WHERE file_uuid = $1 LIMIT 1"
))
.bind(uuid)
.fetch_optional(self.db.pool())
.await?
.unwrap_or(0) > 0;
info!("📝 Prerequisites met for TKG Build. Starting graph construction...");
let db_clone = self.db.clone();
let redis_clone = self.redis.clone();
let uuid_clone = uuid.to_string();
let output_dir_clone = crate::core::config::OUTPUT_DIR.clone();
tokio::spawn(async move {
match crate::core::processor::tkg::build_tkg(&db_clone, &uuid_clone, &output_dir_clone, Some(redis_clone.clone())).await {
Ok(r) => {
let total_nodes = r.face_track_nodes + r.gaze_track_nodes + r.lip_track_nodes + r.text_region_nodes + r.appearance_trace_nodes + r.accessory_nodes + r.object_nodes + r.hand_nodes + r.speaker_nodes;
let total_edges = r.co_occurrence_edges + r.speaker_face_edges + r.face_face_edges + r.mutual_gaze_edges + r.lip_sync_edges + r.has_appearance_edges + r.wears_edges + r.hand_object_edges;
info!("✅ TKG build completed for {}: {} nodes, {} edges", uuid_clone, total_nodes, total_edges);
if tkg_exists {
info!("✅ TKG already built for {}, skipping", uuid);
} else {
info!("📝 Prerequisites met for TKG Build. Starting graph construction...");
let db_clone = self.db.clone();
let redis_clone = self.redis.clone();
let uuid_clone = uuid.to_string();
let output_dir_clone = crate::core::config::OUTPUT_DIR.clone();
tokio::spawn(async move {
match crate::core::processor::tkg::build_tkg(&db_clone, &uuid_clone, &output_dir_clone, Some(redis_clone.clone())).await {
Ok(r) => {
let total_nodes = r.face_track_nodes + r.gaze_track_nodes + r.lip_track_nodes + r.text_region_nodes + r.appearance_trace_nodes + r.accessory_nodes + r.object_nodes + r.hand_nodes + r.speaker_nodes;
let total_edges = r.co_occurrence_edges + r.speaker_face_edges + r.face_face_edges + r.mutual_gaze_edges + r.lip_sync_edges + r.has_appearance_edges + r.wears_edges + r.hand_object_edges;
info!("✅ TKG build completed for {}: {} nodes, {} edges", uuid_clone, total_nodes, total_edges);
let mut pp = PipelineProgress::new(&uuid_clone);
pp.update_stage("tkg_nodes", 1.0, "completed", Some(format!("{} nodes", total_nodes)));
pp.update_stage("tkg_edges", 1.0, "completed", Some(format!("{} edges", total_edges)));
publish_pipeline_progress(redis_clone.as_ref(), &uuid_clone, &pp).await;
let mut pp = PipelineProgress::new(&uuid_clone);
pp.update_stage("tkg_nodes", 1.0, "completed", Some(format!("{} nodes", total_nodes)));
pp.update_stage("tkg_edges", 1.0, "completed", Some(format!("{} edges", total_edges)));
publish_pipeline_progress(redis_clone.as_ref(), &uuid_clone, &pp).await;
// Trigger Rule 2 ingestion after TKG complete
if total_edges > 0 {
match crate::core::chunk::rule2_ingest::ingest_rule2(db_clone.pool(), &uuid_clone, None, None).await {
Ok(rule2_count) => {
info!("✅ Rule 2 ingestion completed for {}: {} relationship chunks", uuid_clone, rule2_count);
let mut pp = PipelineProgress::new(&uuid_clone);
pp.update_stage("rule2_ingestion", 1.0, "completed", Some(format!("{} chunks", rule2_count)));
publish_pipeline_progress(redis_clone.as_ref(), &uuid_clone, &pp).await;
}
Err(e) => error!("❌ Rule 2 ingestion failed for {}: {}", uuid_clone, e),
// Trigger Rule 2 ingestion after TKG complete
if total_edges > 0 {
match crate::core::chunk::rule2_ingest::ingest_rule2(db_clone.pool(), &uuid_clone, None, None).await {
Ok(rule2_count) => {
info!("✅ Rule 2 ingestion completed for {}: {} relationship chunks", uuid_clone, rule2_count);
let mut pp = PipelineProgress::new(&uuid_clone);
pp.update_stage("rule2_ingestion", 1.0, "completed", Some(format!("{} chunks", rule2_count)));
publish_pipeline_progress(redis_clone.as_ref(), &uuid_clone, &pp).await;
}
Err(e) => error!("❌ Rule 2 ingestion failed for {}: {}", uuid_clone, e),
}
}
Err(e) => error!("❌ TKG build failed for {}: {}", uuid_clone, e),
}
});
}
Err(e) => error!("❌ TKG build failed for {}: {}", uuid_clone, e),
}
});
}
if !Self::ingestion_complete(self.db.pool(), uuid, job_processors).await {