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

fix: worker detects deleted jobs and skips them

Browse Source 
When a job is deleted from the database (e.g., by unregister),
the worker now checks if the job still exists before processing.
If the job no longer exists, it skips it instead of getting stuck.

This fixes the issue where unregistering a file would leave the
worker stuck trying to process a non-existent job.
This commit is contained in:
Accusys
2026-07-06 10:28:11 +08:00
parent 799ede5a0e
commit 221aa4c4cc
1 changed files with 48 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
src/worker/job_worker.rs
View File

@@ -278,6 +278,13 @@ impl JobWorker {
} }
async fn process_job(&self, job: crate::core::db::MonitorJob) -> Result<()> { async fn process_job(&self, job: crate::core::db::MonitorJob) -> Result<()> {
// Check if job still exists in database (may have been deleted by unregister)
let current_job = self.db.get_monitor_job_by_uuid(&job.uuid).await?;
if current_job.is_none() {
info!("Job {} no longer exists in database (possibly unregistered), skipping", job.uuid);
return Ok(());
}
info!("Processing job: {} ({})", job.uuid, job.id); info!("Processing job: {} ({})", job.uuid, job.id);
// Determine which processors to run based on job.processors field // Determine which processors to run based on job.processors field
@@ -1833,39 +1840,53 @@ impl JobWorker {
// 🚀 P4 Trigger: TKG Build (Face + ASRX) → then Rule2 ingestion // 🚀 P4 Trigger: TKG Build (Face + ASRX) → then Rule2 ingestion
if has_face && has_asrx { if has_face && has_asrx {
info!("📝 Prerequisites met for TKG Build. Starting graph construction..."); // Guard: only spawn TKG if nodes don't exist yet
let db_clone = self.db.clone(); let tkg_nodes_table = schema::table_name("tkg_nodes");
let redis_clone = self.redis.clone(); let tkg_exists: bool = sqlx::query_scalar::<_, i32>(&format!(
let uuid_clone = uuid.to_string(); "SELECT 1 FROM {tkg_nodes_table} WHERE file_uuid = $1 LIMIT 1"
let output_dir_clone = crate::core::config::OUTPUT_DIR.clone(); ))
tokio::spawn(async move { .bind(uuid)
match crate::core::processor::tkg::build_tkg(&db_clone, &uuid_clone, &output_dir_clone, Some(redis_clone.clone())).await { .fetch_optional(self.db.pool())
Ok(r) => { .await?
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; .unwrap_or(0) > 0;
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); if tkg_exists {
pp.update_stage("tkg_nodes", 1.0, "completed", Some(format!("{} nodes", total_nodes))); info!("✅ TKG already built for {}, skipping", uuid);
pp.update_stage("tkg_edges", 1.0, "completed", Some(format!("{} edges", total_edges))); } else {
publish_pipeline_progress(redis_clone.as_ref(), &uuid_clone, &pp).await; 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);
// Trigger Rule 2 ingestion after TKG complete let mut pp = PipelineProgress::new(&uuid_clone);
if total_edges > 0 { pp.update_stage("tkg_nodes", 1.0, "completed", Some(format!("{} nodes", total_nodes)));
match crate::core::chunk::rule2_ingest::ingest_rule2(db_clone.pool(), &uuid_clone, None, None).await { pp.update_stage("tkg_edges", 1.0, "completed", Some(format!("{} edges", total_edges)));
Ok(rule2_count) => { publish_pipeline_progress(redis_clone.as_ref(), &uuid_clone, &pp).await;
info!("✅ Rule 2 ingestion completed for {}: {} relationship chunks", uuid_clone, rule2_count);
let mut pp = PipelineProgress::new(&uuid_clone); // Trigger Rule 2 ingestion after TKG complete
pp.update_stage("rule2_ingestion", 1.0, "completed", Some(format!("{} chunks", rule2_count))); if total_edges > 0 {
publish_pipeline_progress(redis_clone.as_ref(), &uuid_clone, &pp).await; 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!("❌ 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 { if !Self::ingestion_complete(self.db.pool(), uuid, job_processors).await {