diff --git a/crates/openplotva-app/src/dialog_jobs/queue.rs b/crates/openplotva-app/src/dialog_jobs/queue.rs index 1638aec9..ab7e21db 100644 --- a/crates/openplotva-app/src/dialog_jobs/queue.rs +++ b/crates/openplotva-app/src/dialog_jobs/queue.rs @@ -101,9 +101,7 @@ impl DialogJobWorkerQueue for InMemoryTaskQueue { ); Ok(item.map(|item| { let schedule = self - .records() - .into_iter() - .find(|record| record.id == item.id) + .record(item.id) .map(|record| (record.source_update_ids, record.latest_update_id)) .unwrap_or_default(); dialog_work_item_from_taskman(item, schedule.0, schedule.1) diff --git a/crates/openplotva-app/src/dialog_jobs/tests.rs b/crates/openplotva-app/src/dialog_jobs/tests.rs index 8656ea4f..a69e1e3b 100644 --- a/crates/openplotva-app/src/dialog_jobs/tests.rs +++ b/crates/openplotva-app/src/dialog_jobs/tests.rs @@ -79,6 +79,42 @@ fn activity_pulse( ) } +#[tokio::test] +async fn dialog_dequeue_preserves_claim_and_causation_with_terminal_history() +-> Result<(), Box> { + let queue = InMemoryTaskQueue::new(); + let now = OffsetDateTime::now_utc(); + let old_id = queue.assign( + DIALOG_AIFARM_QUEUE_NAME, + new_dialog_job_at(dialog_params("old"), now), + ); + queue.complete(old_id, now)?; + let expected = queue.assign_with_schedule( + DIALOG_AIFARM_QUEUE_NAME, + new_dialog_job_at(dialog_params("current"), now), + openplotva_taskman::TaskQueueSchedule { + source_update_ids: vec![71, 72], + latest_update_id: Some(72), + ..Default::default() + }, + ); + let item = queue + .dequeue_dialog_job(DIALOG_AIFARM_QUEUE_NAME) + .await? + .expect("current dialog"); + assert_eq!(item.id, expected); + assert_eq!(item.source_update_ids, vec![71, 72]); + assert_eq!(item.latest_update_id, Some(72)); + let record = queue.record(expected).expect("claimed record"); + assert_eq!(record.status, JobStatus::Processing); + assert_eq!(record.started_at, Some(item.claim_started_at)); + assert_eq!( + queue.record(old_id).expect("terminal history").status, + JobStatus::Completed + ); + Ok(()) +} + #[tokio::test] async fn dialog_worker_runs_provider_sends_answer_and_completes_job() -> Result<(), Box> { diff --git a/crates/openplotva-app/src/runtime_taskman.rs b/crates/openplotva-app/src/runtime_taskman.rs index 3a82be21..9820a6be 100644 --- a/crates/openplotva-app/src/runtime_taskman.rs +++ b/crates/openplotva-app/src/runtime_taskman.rs @@ -27,7 +27,7 @@ pub(crate) struct RuntimeTaskmanInspectorHandle { impl RuntimeTaskmanInspectorHandle { pub(crate) fn is_configured(&self) -> bool { - self.records().is_some() + self.shared_queue().is_ok() } pub(crate) fn set_shared_queue( diff --git a/crates/openplotva-app/src/task_queue.rs b/crates/openplotva-app/src/task_queue.rs index b7707cd5..2cf99b5a 100644 --- a/crates/openplotva-app/src/task_queue.rs +++ b/crates/openplotva-app/src/task_queue.rs @@ -514,7 +514,7 @@ impl SharedTaskQueueRuntime { journal.flush_dirty().await?; } let report = SharedTaskQueueRestoreReport { - restored: queue.records().len().max(restored_before_requeue), + restored: restored_before_requeue, requeued, }; Ok((