diff --git a/crates/openplotva-app/src/admin.rs b/crates/openplotva-app/src/admin.rs index e39bfc61..f894a69f 100644 --- a/crates/openplotva-app/src/admin.rs +++ b/crates/openplotva-app/src/admin.rs @@ -966,7 +966,7 @@ where match command.name { ADMIN_QUEUE_STATUS_COMMAND => { - let text = admin_queue_status_text(&runtime); + let text = admin_queue_status_text(&runtime).await; send_admin_queue_result( runtime.effects, message, @@ -1835,10 +1835,10 @@ fn admin_queue_disabled_text() -> &'static str { "❌ Персистентная система очередей отключена" } -fn admin_queue_status_text(runtime: &AdminQueueRuntime<'_, Effects>) -> String { +async fn admin_queue_status_text(runtime: &AdminQueueRuntime<'_, Effects>) -> String { let queue_names = runtime.config.queue_names(); - let pending_by_queue = queue_counts_by_status(runtime.taskman, "pending"); - let processing_by_queue = queue_counts_by_status(runtime.taskman, "processing"); + let pending_by_queue = queue_counts_by_status(runtime.taskman, "pending").await; + let processing_by_queue = queue_counts_by_status(runtime.taskman, "processing").await; let eta_by_queue = queue_eta_seconds(runtime.taskman, &queue_names); let mut lines = Vec::with_capacity(queue_names.len() + 5); lines.push("📊 Статус очередей:\n".to_owned()); @@ -1867,8 +1867,8 @@ fn admin_queue_status_text(runtime: &AdminQueueRuntime<'_, Effects>) -> async fn admin_queue_stats_text(runtime: &AdminQueueRuntime<'_, Effects>) -> String { let queue_names = runtime.config.queue_names(); - let pending_by_queue = queue_counts_by_status(runtime.taskman, "pending"); - let processing_by_queue = queue_counts_by_status(runtime.taskman, "processing"); + let pending_by_queue = queue_counts_by_status(runtime.taskman, "pending").await; + let processing_by_queue = queue_counts_by_status(runtime.taskman, "processing").await; let (total_pending, total_processing) = queue_stats_totals(&queue_names, &pending_by_queue, &processing_by_queue); let mut lines = Vec::new(); @@ -1883,11 +1883,14 @@ async fn admin_queue_stats_text(runtime: &AdminQueueRuntime<'_, Effects &processing_by_queue, )); lines.push("\n🕒 Старейшие:".to_owned()); - lines.extend(queue_stats_oldest_lines( - runtime.taskman, - &queue_names, - runtime.config.default_processing_timeout, - )); + lines.extend( + queue_stats_oldest_lines( + runtime.taskman, + &queue_names, + runtime.config.default_processing_timeout, + ) + .await, + ); lines.push("\n📮 Очередь отправки:".to_owned()); lines.extend(queue_stats_dispatcher_lines( runtime @@ -1915,18 +1918,20 @@ async fn admin_queue_stats_text(runtime: &AdminQueueRuntime<'_, Effects lines.join("\n") } -fn queue_counts_by_status( +async fn queue_counts_by_status( taskman: Option<&dyn RuntimeTaskmanInspector>, status: &str, ) -> BTreeMap { let Some(taskman) = taskman else { return BTreeMap::new(); }; - let result = taskman.list_jobs(RuntimeTaskmanJobsFilter { - status: vec![status.to_owned()], - limit: 1, - ..RuntimeTaskmanJobsFilter::default() - }); + let result = taskman + .list_jobs(RuntimeTaskmanJobsFilter { + status: vec![status.to_owned()], + limit: 1, + ..RuntimeTaskmanJobsFilter::default() + }) + .await; match result { Ok(result) => queue_count_map(&result.summary.by_queue), Err(error) => { @@ -2032,44 +2037,46 @@ fn queue_stats_count_lines( .collect() } -fn queue_stats_oldest_lines( +async fn queue_stats_oldest_lines( taskman: Option<&dyn RuntimeTaskmanInspector>, queue_names: &[String], timeout: Duration, ) -> Vec { - queue_names - .iter() - .map(|queue_name| { - let pending_age = queue_oldest_age(taskman, queue_name, "pending", "created_at"); - let processing_age = queue_oldest_age(taskman, queue_name, "processing", "started_at"); - let processing_flag = processing_age - .map(|age| queue_stats_processing_flag(age, timeout)) - .unwrap_or_default(); - format!( - "• {queue_name}: pending {}, processing {}{}", - format_queue_optional_age(pending_age), - format_queue_optional_age(processing_age), - processing_flag - ) - }) - .collect() + let mut lines = Vec::with_capacity(queue_names.len()); + for queue_name in queue_names { + let pending_age = queue_oldest_age(taskman, queue_name, "pending", "created_at").await; + let processing_age = + queue_oldest_age(taskman, queue_name, "processing", "started_at").await; + let processing_flag = processing_age + .map(|age| queue_stats_processing_flag(age, timeout)) + .unwrap_or_default(); + lines.push(format!( + "• {queue_name}: pending {}, processing {}{}", + format_queue_optional_age(pending_age), + format_queue_optional_age(processing_age), + processing_flag + )); + } + lines } -fn queue_oldest_age( +async fn queue_oldest_age( taskman: Option<&dyn RuntimeTaskmanInspector>, queue_name: &str, status: &str, time_field: &str, ) -> Option { let taskman = taskman?; - let result = taskman.list_jobs(RuntimeTaskmanJobsFilter { - status: vec![status.to_owned()], - queue: vec![queue_name.to_owned()], - sort_by: time_field.to_owned(), - sort_dir: "asc".to_owned(), - limit: 1, - ..RuntimeTaskmanJobsFilter::default() - }); + let result = taskman + .list_jobs(RuntimeTaskmanJobsFilter { + status: vec![status.to_owned()], + queue: vec![queue_name.to_owned()], + sort_by: time_field.to_owned(), + sort_dir: "asc".to_owned(), + limit: 1, + ..RuntimeTaskmanJobsFilter::default() + }) + .await; let result = match result { Ok(result) => result, Err(error) => { @@ -2559,7 +2566,7 @@ fn format_go_fraction(value: u64, unit: u64, suffix: &str) -> String { mod tests { use super::*; use openplotva_server::{ - RuntimeGeminiCachePurgerFuture, RuntimeTaskmanDiagnosticsData, + RuntimeGeminiCachePurgerFuture, RuntimeTaskmanDiagnosticsData, RuntimeTaskmanJobListFuture, RuntimeTaskmanJobListResultData, RuntimeTaskmanJobSummaryData, RuntimeTaskmanQueueDiagnosticsData, RuntimeUpdatesInspectorFuture, }; @@ -4320,10 +4327,10 @@ mod tests { } impl RuntimeTaskmanInspector for TaskmanInspectorStub { - fn list_jobs( - &self, + fn list_jobs<'a>( + &'a self, filter: RuntimeTaskmanJobsFilter, - ) -> Result { + ) -> RuntimeTaskmanJobListFuture<'a> { let status = filter.status.first().map(String::as_str).unwrap_or(""); let by_queue = match status { "pending" => self.pending.clone(), @@ -4351,7 +4358,7 @@ mod tests { } else { Vec::new() }; - Ok(RuntimeTaskmanJobListResultData { + Box::pin(std::future::ready(Ok(RuntimeTaskmanJobListResultData { total: items.len() as i32, offset: 0, limit: filter.limit, @@ -4360,7 +4367,7 @@ mod tests { by_queue: json!(by_queue), }, items, - }) + }))) } fn job<'a>(&'a self, _id: i64) -> openplotva_server::RuntimeTaskmanJobFuture<'a> { diff --git a/crates/openplotva-app/src/dialog_jobs/queue.rs b/crates/openplotva-app/src/dialog_jobs/queue.rs index ab7e21db..d0e326b3 100644 --- a/crates/openplotva-app/src/dialog_jobs/queue.rs +++ b/crates/openplotva-app/src/dialog_jobs/queue.rs @@ -4,8 +4,8 @@ use std::fmt; use openplotva_taskman::{ - InMemoryTaskQueue, JobType, Priority, StatelessJobItem, TaskQueueError, TaskQueueJobEvent, - TaskQueueWorkItem, + AGENT_CHECKPOINT_EVENT_STAGE, InMemoryTaskQueue, JobType, Priority, StatelessJobItem, + TaskQueueError, TaskQueueJobEvent, TaskQueueWorkItem, }; use time::OffsetDateTime; @@ -146,7 +146,7 @@ impl DialogJobWorkerQueue for InMemoryTaskQueue { at: OffsetDateTime, ) -> DialogJobWorkerFuture<'a, (), Self::Error> { Box::pin(async move { - if event.stage == "agent_checkpoint" { + if event.stage == AGENT_CHECKPOINT_EVENT_STAGE { self.upsert_job_event(job_id, event, at) } else { self.append_job_event(job_id, event, at) diff --git a/crates/openplotva-app/src/dialog_jobs/tests.rs b/crates/openplotva-app/src/dialog_jobs/tests.rs index 8627b1d9..58f9ea0f 100644 --- a/crates/openplotva-app/src/dialog_jobs/tests.rs +++ b/crates/openplotva-app/src/dialog_jobs/tests.rs @@ -13,9 +13,10 @@ use openplotva_llm::{ retry::{FailureReason, ProviderError}, }; use openplotva_taskman::{ - DEFAULT_LLM_JOB_MAX_ATTEMPTS, DIALOG_AIFARM_QUEUE_NAME, DialogJobParams, HIGH_PRIORITY, - InMemoryTaskQueue, JobStatus, LLM_JOB_RETRY_EXHAUSTED_STAGE, LLM_JOB_RETRY_STAGE, Priority, - TEXT_QUEUE_NAME, TaskQueueError, TaskQueueJobEvent, new_dialog_job_at, + AGENT_CHECKPOINT_EVENT_STAGE, DEFAULT_LLM_JOB_MAX_ATTEMPTS, DIALOG_AIFARM_QUEUE_NAME, + DialogJobParams, HIGH_PRIORITY, InMemoryTaskQueue, JobStatus, LLM_JOB_RETRY_EXHAUSTED_STAGE, + LLM_JOB_RETRY_STAGE, Priority, TEXT_QUEUE_NAME, TaskQueueError, TaskQueueJobEvent, + new_dialog_job_at, }; use openplotva_telegram::DispatcherQueue; use time::{Duration as TimeDuration, OffsetDateTime}; @@ -7280,6 +7281,20 @@ async fn agent_retry_restores_context_and_completed_tool_results() -> Result<(), ) .await; assert!(first.retry_requeued, "{first:?}"); + let checkpoints = |queue: &InMemoryTaskQueue| { + queue + .record(id) + .expect("record") + .events + .iter() + .filter(|event| event.stage == AGENT_CHECKPOINT_EVENT_STAGE) + .count() + }; + assert_eq!( + checkpoints(&queue), + 1, + "a requeued turn keeps one checkpoint" + ); let second = process_dialog_job_once_in_queue_with_materializer_history_and_retry_at( &queue, &provider, @@ -7297,14 +7312,10 @@ async fn agent_retry_restores_context_and_completed_tool_results() -> Result<(), message, openplotva_dialog::SessionMessage::ToolResult { .. } ))); - let record = queue.record(id).expect("record"); assert_eq!( - record - .events - .iter() - .filter(|event| event.stage == "agent_checkpoint") - .count(), - 1 + checkpoints(&queue), + 0, + "a finished turn drops its checkpoint" ); Ok(()) } diff --git a/crates/openplotva-app/src/dialog_turn/session.rs b/crates/openplotva-app/src/dialog_turn/session.rs index 2b78b094..172d21d3 100644 --- a/crates/openplotva-app/src/dialog_turn/session.rs +++ b/crates/openplotva-app/src/dialog_turn/session.rs @@ -33,7 +33,7 @@ use openplotva_dialog::{ }, }; use openplotva_llm::ChatStepProvider; -use openplotva_taskman::TaskQueueJobEvent; +use openplotva_taskman::{AGENT_CHECKPOINT_EVENT_STAGE, TaskQueueJobEvent}; use serde_json::Value; use time::{Duration as TimeDuration, OffsetDateTime, format_description::well_known::Rfc3339}; @@ -466,7 +466,7 @@ where .item_events .iter() .rev() - .find(|event| event.stage == "agent_checkpoint") + .find(|event| event.stage == AGENT_CHECKPOINT_EVENT_STAGE) .and_then(|event| event.data.get("checkpoint")) .map(|value| serde_json::from_str::<(DialogInput, SessionState)>(value)) .transpose(); @@ -603,7 +603,7 @@ where let duplicate_guard_history = &duplicate_guard_history; let run_id = &run_id; openplotva_agent::run((state, report), |(state, report)| async move { - if let Err(error) = persist_agent_event(queue, effects, ctx.item_id, "agent_checkpoint", BTreeMap::from([("checkpoint".into(), serde_json::to_string(&(base_input, &state)).expect("serializable agent state"))]), ctx.now).await { + if let Err(error) = persist_agent_event(queue, effects, ctx.item_id, AGENT_CHECKPOINT_EVENT_STAGE, BTreeMap::from([("checkpoint".into(), serde_json::to_string(&(base_input, &state)).expect("serializable agent state"))]), ctx.now).await { return std::ops::ControlFlow::Break(agent_persistence_failed(error)); } let SessionState { mut requested_tool_call_offset, mut tool_context, mut gift_used, mut budget, mut agent, mut sent, mut side_effect_tickets, mut recorded_tool_calls, mut regenerations, mut anti_loop, mut repeated_final_repair, mut requires_novel_final, mut draws_scheduled, mut songs_scheduled, mut reacted_message_ids, mut tool_result_cache, mut media_reference_aliases, mut successful_web_search, mut web_source_urls, mut links, mut search_citation_repairs, mut repeated_tool_call } = state; diff --git a/crates/openplotva-app/src/lib.rs b/crates/openplotva-app/src/lib.rs index fc317479..181befde 100644 --- a/crates/openplotva-app/src/lib.rs +++ b/crates/openplotva-app/src/lib.rs @@ -1521,7 +1521,7 @@ async fn admin_taskman_jobs( RawQuery(raw_query): RawQuery, Extension(routes): Extension, ) -> Response { - admin_taskman_jobs_response(&routes, method, &headers, raw_query.as_deref()) + admin_taskman_jobs_response(&routes, method, &headers, raw_query.as_deref()).await } async fn admin_taskman_jobs_clear( @@ -1530,7 +1530,7 @@ async fn admin_taskman_jobs_clear( RawQuery(raw_query): RawQuery, Extension(routes): Extension, ) -> Response { - admin_taskman_jobs_clear_response(&routes, method, &headers, raw_query.as_deref()) + admin_taskman_jobs_clear_response(&routes, method, &headers, raw_query.as_deref()).await } async fn admin_taskman_job( @@ -1557,7 +1557,7 @@ async fn admin_taskman_job_restart( RawQuery(raw_query): RawQuery, Extension(routes): Extension, ) -> Response { - admin_taskman_job_restart_response(&routes, method, &headers, raw_query.as_deref()) + admin_taskman_job_restart_response(&routes, method, &headers, raw_query.as_deref()).await } async fn admin_redis_list( @@ -5355,7 +5355,7 @@ fn admin_shield_test_json( serde_json::Value::Object(value) } -fn admin_taskman_jobs_response( +async fn admin_taskman_jobs_response( routes: &StaticWebRoutes, method: Method, headers: &HeaderMap, @@ -5377,7 +5377,9 @@ fn admin_taskman_jobs_response( Ok(filter) => filter, Err(response) => return *response, }; - match openplotva_server::RuntimeTaskmanInspector::list_jobs(&routes.taskman_inspector, filter) { + match openplotva_server::RuntimeTaskmanInspector::list_jobs(&routes.taskman_inspector, filter) + .await + { Ok(result) => { admin_json_no_cache_response(StatusCode::OK, admin_taskman_list_json(&result)) } @@ -5388,7 +5390,7 @@ fn admin_taskman_jobs_response( } } -fn admin_taskman_jobs_clear_response( +async fn admin_taskman_jobs_clear_response( routes: &StaticWebRoutes, method: Method, headers: &HeaderMap, @@ -5412,7 +5414,7 @@ fn admin_taskman_jobs_clear_response( }; filter.offset = 0; filter.limit = 0; - match routes.taskman_inspector.delete_jobs(filter) { + match routes.taskman_inspector.delete_jobs(filter).await { Ok(result) => admin_json_response( StatusCode::OK, serde_json::json!({ @@ -5495,7 +5497,7 @@ fn admin_taskman_job_cancel_response( } } -fn admin_taskman_job_restart_response( +async fn admin_taskman_job_restart_response( routes: &StaticWebRoutes, method: Method, headers: &HeaderMap, @@ -5517,7 +5519,7 @@ fn admin_taskman_job_restart_response( Ok(job_id) => job_id, Err(response) => return *response, }; - match routes.taskman_inspector.restart_job(job_id) { + match routes.taskman_inspector.restart_job(job_id).await { Ok(new_job_id) => admin_json_response( StatusCode::OK, serde_json::json!({ "ok": true, "new_job_id": new_job_id }), @@ -10325,7 +10327,7 @@ fn build_shared_task_queue_maintenance_group( let report = task_queue::run_shared_task_queue_terminal_cleanup_worker_until( cleanup_runtime, cleanup_interval, - completed_retention, + task_queue::FINISHED_JOB_MEMORY_RETENTION, wait_for_runtime_stop(cleanup_stop), ) .await; @@ -10336,9 +10338,9 @@ fn build_shared_task_queue_maintenance_group( readiness.push(ReadinessCheck::ok( "shared_task_queue_terminal_cleanup", format!( - "terminal cleanup every {}s, retention {}d", + "finished-job eviction from memory every {}s, kept {}m", cleanup_interval.as_secs(), - completed_retention.whole_days() + task_queue::FINISHED_JOB_MEMORY_RETENTION.whole_minutes() ), )); @@ -10360,8 +10362,9 @@ fn build_shared_task_queue_maintenance_group( readiness.push(ReadinessCheck::ok( "shared_task_queue_postgres_purge", format!( - "Postgres purge every {}s, retention {}d", + "Postgres purge every {}s, finished jobs kept {}d, deleted jobs {}d", cleanup_interval.as_secs(), + completed_retention.whole_days() * 2, completed_retention.whole_days() ), )); @@ -11005,6 +11008,7 @@ async fn start_runtime_workers( let llm_event_cleanup_worker = tokio::spawn(async move { let report = runtime_llm::run_llm_request_event_cleanup_worker_until( llm_event_cleanup_pool, + runtime_llm::LLM_REQUEST_EVENTS_CLEANUP_FIRST_PASS_DELAY, llm_event_cleanup_interval, llm_request_events_retention_days, runtime_llm::LLM_REQUEST_EVENTS_CLEANUP_BATCH_SIZE, @@ -11017,8 +11021,9 @@ async fn start_runtime_workers( readiness_checks.push(ReadinessCheck::ok( "llm_request_events_cleanup", format!( - "LLM analytics rollup and raw request events cleanup every {}s, retention {}d", + "LLM analytics rollup and raw request events cleanup every {}s after a {}s start delay, retention {}d", llm_event_cleanup_interval.as_secs(), + runtime_llm::LLM_REQUEST_EVENTS_CLEANUP_FIRST_PASS_DELAY.as_secs(), llm_request_events_retention_days ), )); @@ -16526,7 +16531,8 @@ mod tests { Method::GET, &headers, Some("status=pending&queue=image-vip"), - ); + ) + .await; assert_eq!(response.status(), StatusCode::OK); assert_eq!( response.headers().get(header::CACHE_CONTROL), @@ -16569,7 +16575,8 @@ mod tests { Method::PUT, &headers, Some(&format!("job_id={diagnostic_id}")), - ); + ) + .await; assert_eq!(response.status(), StatusCode::OK); let body = to_bytes(response.into_body(), usize::MAX).await?; let value: serde_json::Value = serde_json::from_slice(&body)?; @@ -16580,7 +16587,8 @@ mod tests { Method::DELETE, &headers, Some("status=cancelled"), - ); + ) + .await; assert_eq!(response.status(), StatusCode::OK); let body = to_bytes(response.into_body(), usize::MAX).await?; let value: serde_json::Value = serde_json::from_slice(&body)?; @@ -16754,7 +16762,7 @@ mod tests { let value = signed.split(';').next().expect("cookie pair"); headers.insert(header::COOKIE, value.parse()?); } - let response = admin_taskman_jobs_response(&routes, Method::GET, &headers, None); + let response = admin_taskman_jobs_response(&routes, Method::GET, &headers, None).await; assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); let body = to_bytes(response.into_body(), usize::MAX).await?; assert_eq!(&body[..], b"{\"error\":\"task manager not configured\"}\n"); @@ -16762,7 +16770,8 @@ mod tests { let response = admin_taskman_job_response(&routes, Method::GET, &headers, None).await; assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); let response = - admin_taskman_jobs_response(&routes, Method::POST, &headers, Some("status=pending")); + admin_taskman_jobs_response(&routes, Method::POST, &headers, Some("status=pending")) + .await; assert_eq!(response.status(), StatusCode::METHOD_NOT_ALLOWED); Ok(()) } diff --git a/crates/openplotva-app/src/memory_runtime.rs b/crates/openplotva-app/src/memory_runtime.rs index 82d87480..61679218 100644 --- a/crates/openplotva-app/src/memory_runtime.rs +++ b/crates/openplotva-app/src/memory_runtime.rs @@ -1424,7 +1424,7 @@ where .ensure_hourly_runs( now, cfg.retention, - memory_enqueue_lookback(now, startup, cfg.retention), + memory_enqueue_lookback(now, cfg.retention), cfg.enqueue_policy, ) .await @@ -1485,7 +1485,7 @@ where .ensure_hourly_runs( now, cfg.retention, - memory_enqueue_lookback(now, startup, cfg.retention), + memory_enqueue_lookback(now, cfg.retention), cfg.enqueue_policy, ) .await @@ -2528,15 +2528,13 @@ pub const MEMORY_RECENT_LOOKBACK: Duration = Duration::from_secs(2 * 60 * 60); /// outage or a full queue left without a run. const MEMORY_FULL_SCAN_HOUR: u8 = 3; -/// History span one enqueue tick scans: the whole retention window on startup -/// and once a day, otherwise the recent hours. +/// History span one enqueue tick scans: the whole retention window once a day, +/// otherwise the recent hours. Startup scans recent hours too, so a deploy does +/// not refill the queue while the process catches up; the daily scan covers +/// hours missed while it was down. #[must_use] -pub fn memory_enqueue_lookback( - now: OffsetDateTime, - startup: bool, - retention: Duration, -) -> Duration { - if startup || now.to_offset(time::UtcOffset::UTC).hour() == MEMORY_FULL_SCAN_HOUR { +pub fn memory_enqueue_lookback(now: OffsetDateTime, retention: Duration) -> Duration { + if now.to_offset(time::UtcOffset::UTC).hour() == MEMORY_FULL_SCAN_HOUR { retention } else { MEMORY_RECENT_LOOKBACK.min(retention) @@ -6650,8 +6648,12 @@ mod tests { .collect(); assert_eq!( lookbacks, - vec![MEMORY_RECENT_LOOKBACK, cfg.retention, cfg.retention], - "hourly ticks scan recent hours; the daily tick and startup scan the retention window" + vec![ + MEMORY_RECENT_LOOKBACK, + cfg.retention, + MEMORY_RECENT_LOOKBACK + ], + "hourly ticks and startup scan recent hours; the daily tick scans the retention window" ); } diff --git a/crates/openplotva-app/src/runtime_llm.rs b/crates/openplotva-app/src/runtime_llm.rs index 6e2b707b..6e08f11b 100644 --- a/crates/openplotva-app/src/runtime_llm.rs +++ b/crates/openplotva-app/src/runtime_llm.rs @@ -31,6 +31,9 @@ const LLM_EVENT_WRITER_INSERT_TIMEOUT: Duration = Duration::from_secs(15); const LLM_EVENT_WRITER_FULL_WARNING_INTERVAL_MS: u64 = 60_000; pub const LLM_REQUEST_EVENTS_CLEANUP_BATCH_SIZE: i64 = 10_000; pub const LLM_REQUEST_EVENTS_CLEANUP_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60); +/// The first archive pass waits until startup catch-up has drained, so its +/// rollup queries do not compete with the backlog for disk and connections. +pub const LLM_REQUEST_EVENTS_CLEANUP_FIRST_PASS_DELAY: Duration = Duration::from_secs(30 * 60); const SQL_INSERT_LLM_REQUEST_EVENTS_PREFIX: &str = r#"INSERT INTO llm_request_events ( created_at, provider, @@ -278,42 +281,6 @@ const SQL_DELETE_OLD_MEMORY_RUNS: &str = r#" DELETE FROM memory_runs WHERE status IN ('completed','skipped','failed') AND range_start_at < $1::timestamptz"#; -const SQL_ROLLUP_CHAT_HISTORY_INTERESTS: &str = r#" -WITH grouped AS ( - SELECT - date_trunc($1, e.range_end_at)::timestamptz AS bucket_start, - e.chat_id, - e.thread_id, - topic.topic, - count(*)::int AS episode_count, - COALESCE(sum(e.message_count), 0)::int AS message_count - FROM memory_episodes e - CROSS JOIN LATERAL unnest(e.topics) AS topic(topic) - WHERE e.range_end_at < $2::timestamptz - AND btrim(topic.topic) <> '' - GROUP BY 1, e.chat_id, e.thread_id, topic.topic -), -prepared AS ( - SELECT - 'chat_history'::text AS source, - 'interest'::text AS kind, - $1::text AS granularity, - bucket_start, - bucket_start + CASE WHEN $1 = 'day' THEN interval '1 day' ELSE interval '1 hour' END AS bucket_end, - jsonb_build_object('chat_id', chat_id, 'thread_id', thread_id, 'topic', topic) AS dimensions, - jsonb_build_object('episode_count', episode_count, 'message_count', message_count) AS metrics - FROM grouped -) -INSERT INTO telemetry_rollups ( - source, kind, granularity, bucket_start, bucket_end, dimensions_hash, dimensions, metrics -) -SELECT source, kind, granularity, bucket_start, bucket_end, md5(dimensions::text), dimensions, metrics -FROM prepared -ON CONFLICT (source, kind, granularity, bucket_start, dimensions_hash) DO UPDATE SET - bucket_end = EXCLUDED.bucket_end, - dimensions = EXCLUDED.dimensions, - metrics = EXCLUDED.metrics, - updated_at = CURRENT_TIMESTAMP"#; #[derive(Clone, Debug)] pub struct RuntimeLlmTraceBuffer { @@ -943,11 +910,6 @@ async fn archive_old_analytics( .bind(cutoff) .execute(&mut *tx) .await?; - sqlx::query(SQL_ROLLUP_CHAT_HISTORY_INTERESTS) - .bind(granularity) - .bind(cutoff) - .execute(&mut *tx) - .await?; } sqlx::query(SQL_DELETE_OLD_MEMORY_RUNS) .bind(cutoff) @@ -972,6 +934,7 @@ async fn delete_old_llm_request_events_batch( pub async fn run_llm_request_event_cleanup_worker_until( pool: PgPool, + first_pass_delay: Duration, interval: Duration, retention_days: i32, batch_size: i64, @@ -989,6 +952,11 @@ where } tokio::pin!(stop); + tokio::select! { + biased; + () = &mut stop => return report, + () = tokio::time::sleep(first_pass_delay) => {} + } 'passes: loop { let archive_result = tokio::select! { biased; @@ -2286,6 +2254,7 @@ mod tests { let report = run_llm_request_event_cleanup_worker_until( pool.clone(), + Duration::ZERO, Duration::from_secs(3600), 14, 2, @@ -2324,6 +2293,7 @@ mod tests { let rerun = run_llm_request_event_cleanup_worker_until( pool.clone(), + Duration::ZERO, Duration::from_secs(3600), 14, 2, @@ -2356,6 +2326,7 @@ mod tests { .expect("make archive storage unavailable"); let failed = run_llm_request_event_cleanup_worker_until( pool.clone(), + Duration::ZERO, Duration::from_secs(3600), 14, 2, @@ -2380,6 +2351,7 @@ mod tests { .expect("restore archive storage"); let recovered = run_llm_request_event_cleanup_worker_until( pool.clone(), + Duration::ZERO, Duration::from_secs(3600), 14, 2, @@ -2400,6 +2372,7 @@ mod tests { Duration::from_millis(100), run_llm_request_event_cleanup_worker_until( pool, + Duration::ZERO, Duration::from_secs(3600), 14, 10_000, diff --git a/crates/openplotva-app/src/runtime_llm_analytics.rs b/crates/openplotva-app/src/runtime_llm_analytics.rs index b672d767..2bf1409a 100644 --- a/crates/openplotva-app/src/runtime_llm_analytics.rs +++ b/crates/openplotva-app/src/runtime_llm_analytics.rs @@ -685,17 +685,19 @@ fn first_nonzero_i32(values: [i32; N]) -> i32 { async fn load_runtime_job_details( taskman: &dyn RuntimeTaskmanInspector, ) -> Result, String> { - let result = taskman.list_jobs(RuntimeTaskmanJobsFilter { - queue: vec![ - IMAGE_REGULAR_QUEUE_NAME.to_owned(), - IMAGE_VIP_QUEUE_NAME.to_owned(), - MUSIC_VIP_QUEUE_NAME.to_owned(), - ], - sort_by: "id".to_owned(), - sort_dir: "asc".to_owned(), - limit: 1000, - ..RuntimeTaskmanJobsFilter::default() - })?; + let result = taskman + .list_jobs(RuntimeTaskmanJobsFilter { + queue: vec![ + IMAGE_REGULAR_QUEUE_NAME.to_owned(), + IMAGE_VIP_QUEUE_NAME.to_owned(), + MUSIC_VIP_QUEUE_NAME.to_owned(), + ], + sort_by: "id".to_owned(), + sort_dir: "asc".to_owned(), + limit: 1000, + ..RuntimeTaskmanJobsFilter::default() + }) + .await?; let mut jobs = Vec::with_capacity(result.items.len()); for item in result.items { let job_type = item.job_type; diff --git a/crates/openplotva-app/src/runtime_taskman.rs b/crates/openplotva-app/src/runtime_taskman.rs index 676c9082..4bdfcb87 100644 --- a/crates/openplotva-app/src/runtime_taskman.rs +++ b/crates/openplotva-app/src/runtime_taskman.rs @@ -7,13 +7,15 @@ use std::{ use openplotva_server::{ RuntimeTaskmanDiagnosticsData, RuntimeTaskmanInspector, RuntimeTaskmanJobData, RuntimeTaskmanJobDetailsData, RuntimeTaskmanJobFuture, RuntimeTaskmanJobListEntryData, - RuntimeTaskmanJobListResultData, RuntimeTaskmanJobSummaryData, RuntimeTaskmanJobsFilter, - RuntimeTaskmanQueueDiagnosticsData, + RuntimeTaskmanJobListFuture, RuntimeTaskmanJobListResultData, RuntimeTaskmanJobSummaryData, + RuntimeTaskmanJobsFilter, RuntimeTaskmanQueueDiagnosticsData, +}; +use openplotva_storage::{ + PostgresTaskQueueStore, TaskQueueJobFilter, TaskQueueSortColumn, TaskQueueTimeColumn, }; -use openplotva_storage::PostgresTaskQueueStore; use openplotva_taskman::{ - CONTROL_QUEUE_NAME, InMemoryTaskQueue, JobPayload, JobStatus, JobType, TaskQueueRecord, - fallback_queue_time_estimate, + CONTROL_QUEUE_NAME, InMemoryTaskQueue, JobPayload, JobStatus, JobType, TaskQueueError, + TaskQueueRecord, fallback_queue_time_estimate, }; use serde_json::json; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; @@ -79,18 +81,59 @@ impl RuntimeTaskmanInspectorHandle { .map_err(|error| error.to_string()) } - pub(crate) fn restart_job(&self, id: i64) -> Result { - self.shared_queue()? - .restart(id, OffsetDateTime::now_utc()) - .map_err(|error| error.to_string()) + /// Recreate a job as pending. Finished jobs that already left memory are + /// restarted from their persisted record. + pub(crate) async fn restart_job(&self, id: i64) -> Result { + let queue = self.shared_queue()?; + let now = OffsetDateTime::now_utc(); + if let Some(record) = queue.record(id) { + return Ok(queue.restart_from(&record, now)); + } + let not_found = || TaskQueueError::JobNotFound(id).to_string(); + let store = self.sql_store_handle().ok_or_else(not_found)?; + let record = store + .load_task_queue_record(id) + .await + .map_err(|error| error.to_string())? + .ok_or_else(not_found)?; + Ok(queue.restart_from(&record, now)) } - pub(crate) fn delete_jobs( + /// Delete matching jobs held in memory. Virtual-dialog cleanup uses this + /// directly because session deletion purges the persisted rows itself. + pub(crate) fn delete_in_memory_jobs( &self, filter: RuntimeTaskmanJobsFilter, ) -> Result { - let queue = self.shared_queue()?; let filter = NormalizedTaskmanFilter::try_from(filter)?; + self.delete_matching_in_memory(&filter) + .map(|(result, _)| result) + } + + /// Delete matching jobs from memory and their persisted finished rows. + pub(crate) async fn delete_jobs( + &self, + filter: RuntimeTaskmanJobsFilter, + ) -> Result { + let filter = NormalizedTaskmanFilter::try_from(filter)?; + let (mut result, deleted_ids) = self.delete_matching_in_memory(&filter)?; + if let Some(store) = self.sql_store_handle() { + let persisted = store + .delete_finished_task_queue_jobs(&filter.storage_filter(), &deleted_ids) + .await + .map_err(|error| error.to_string())?; + let persisted = i32::try_from(persisted).unwrap_or(i32::MAX); + result.matched = result.matched.saturating_add(persisted); + result.deleted = result.deleted.saturating_add(persisted); + } + Ok(result) + } + + fn delete_matching_in_memory( + &self, + filter: &NormalizedTaskmanFilter, + ) -> Result<(RuntimeTaskmanDeleteResult, Vec), String> { + let queue = self.shared_queue()?; let records = queue.inspect_records(|records| { records .iter() @@ -99,6 +142,7 @@ impl RuntimeTaskmanInspectorHandle { .collect::>() }); let mut result = RuntimeTaskmanDeleteResult::default(); + let mut deleted_ids = Vec::with_capacity(records.len()); for (id, active) in records { result.matched += 1; if active { @@ -106,8 +150,35 @@ impl RuntimeTaskmanInspectorHandle { } queue.delete(id).map_err(|error| error.to_string())?; result.deleted += 1; + deleted_ids.push(id); } - Ok(result) + Ok((result, deleted_ids)) + } + + fn list_in_memory_jobs( + &self, + filter: &NormalizedTaskmanFilter, + ) -> Option { + self.inspect_records(|mut records| { + records.retain(|record| filter.matches(record)); + let total = records.len() as i32; + let summary = taskman_summary(&records); + sort_taskman_records(&mut records, &filter.sort_by, &filter.sort_dir); + let offset = filter.offset.min(total.max(0)) as usize; + let items = records + .into_iter() + .skip(offset) + .take(filter.limit as usize) + .map(taskman_list_entry_from_record) + .collect(); + RuntimeTaskmanJobListResultData { + total, + offset: filter.offset, + limit: filter.limit, + summary, + items, + } + }) } fn shared_queue(&self) -> Result, String> { @@ -128,34 +199,21 @@ pub(crate) struct RuntimeTaskmanDeleteResult { } impl RuntimeTaskmanInspector for RuntimeTaskmanInspectorHandle { - fn list_jobs( - &self, + /// Active jobs always live in memory; listings that can include finished + /// jobs read Postgres, which keeps them after the queue drops them. + fn list_jobs<'a>( + &'a self, filter: RuntimeTaskmanJobsFilter, - ) -> Result { - Ok(self - .inspect_records(|mut records| { - let filter = NormalizedTaskmanFilter::try_from(filter)?; - records.retain(|record| filter.matches(record)); - let total = records.len() as i32; - let summary = taskman_summary(&records); - sort_taskman_records(&mut records, &filter.sort_by, &filter.sort_dir); - let offset = filter.offset.min(total.max(0)) as usize; - let items = records - .into_iter() - .skip(offset) - .take(filter.limit as usize) - .map(taskman_list_entry_from_record) - .collect(); - Ok::<_, String>(RuntimeTaskmanJobListResultData { - total, - offset: filter.offset, - limit: filter.limit, - summary, - items, - }) - }) - .transpose()? - .unwrap_or_else(empty_taskman_job_list)) + ) -> RuntimeTaskmanJobListFuture<'a> { + Box::pin(async move { + let filter = NormalizedTaskmanFilter::try_from(filter)?; + match self.sql_store_handle() { + Some(store) if !filter.active_only() => list_persisted_jobs(&store, &filter).await, + _ => Ok(self + .list_in_memory_jobs(&filter) + .unwrap_or_else(empty_taskman_job_list)), + } + }) } fn job<'a>(&'a self, id: i64) -> RuntimeTaskmanJobFuture<'a> { @@ -319,7 +377,72 @@ impl TryFrom for NormalizedTaskmanFilter { } } +async fn list_persisted_jobs( + store: &PostgresTaskQueueStore, + filter: &NormalizedTaskmanFilter, +) -> Result { + let page = store + .list_task_queue_jobs( + &filter.storage_filter(), + filter.sort_column(), + filter.sort_dir != "asc", + i64::from(filter.offset), + i64::from(filter.limit), + ) + .await + .map_err(|error| error.to_string())?; + Ok(RuntimeTaskmanJobListResultData { + total: i32::try_from(page.total).unwrap_or(i32::MAX), + offset: filter.offset, + limit: filter.limit, + summary: RuntimeTaskmanJobSummaryData { + by_status: json!(page.by_status), + by_queue: json!(page.by_queue), + }, + items: page + .records + .iter() + .map(taskman_list_entry_from_record) + .collect(), + }) +} + impl NormalizedTaskmanFilter { + fn active_only(&self) -> bool { + !self.status.is_empty() + && self + .status + .iter() + .all(|status| matches!(status.as_str(), "pending" | "processing")) + } + + fn storage_filter(&self) -> TaskQueueJobFilter { + TaskQueueJobFilter { + statuses: self.status.iter().cloned().collect(), + queues: self.queue.iter().cloned().collect(), + user_id: self.user_id, + chat_id: self.chat_id, + time_column: match self.time_field.as_str() { + "started_at" => TaskQueueTimeColumn::StartedAt, + "completed_at" => TaskQueueTimeColumn::CompletedAt, + _ => TaskQueueTimeColumn::CreatedAt, + }, + from: self.from, + to: self.to, + search: self.q.clone(), + } + } + + fn sort_column(&self) -> TaskQueueSortColumn { + match self.sort_by.as_str() { + "id" => TaskQueueSortColumn::Id, + "priority" => TaskQueueSortColumn::Priority, + "started_at" => TaskQueueSortColumn::StartedAt, + "completed_at" => TaskQueueSortColumn::CompletedAt, + _ => TaskQueueSortColumn::CreatedAt, + } + } + fn matches(&self, record: &TaskQueueRecord) -> bool { if !self.status.is_empty() && !self.status.contains(record.status.as_str()) { return false; @@ -870,6 +993,7 @@ mod tests { queue: vec![CONTROL_QUEUE_NAME.to_owned()], ..RuntimeTaskmanJobsFilter::default() }) + .await .expect("list jobs"); assert_eq!(result.total, 1); @@ -990,6 +1114,7 @@ mod tests { sort_dir: "asc".to_owned(), ..RuntimeTaskmanJobsFilter::default() }) + .await .expect("list jobs"); assert_eq!(result.total, 2); assert_eq!(result.summary.by_queue[TEXT_QUEUE_NAME], 1); @@ -1012,6 +1137,7 @@ mod tests { limit: 1, ..RuntimeTaskmanJobsFilter::default() }) + .await .expect("second page"); assert_eq!(page.total, 2); assert_eq!(page.summary.by_status["pending"], 2); @@ -1026,6 +1152,7 @@ mod tests { status: vec!["pending".to_owned()], ..RuntimeTaskmanJobsFilter::default() }) + .await .expect("filtered image"); assert_eq!(filtered.total, 1); assert_eq!(filtered.summary.by_queue[IMAGE_REGULAR_QUEUE_NAME], 1); @@ -1036,6 +1163,7 @@ mod tests { status: vec!["not_a_status".to_owned()], ..RuntimeTaskmanJobsFilter::default() }) + .await .is_err() ); @@ -1134,6 +1262,7 @@ mod tests { sort_dir: "asc".to_owned(), ..RuntimeTaskmanJobsFilter::default() }) + .await .expect("list jobs"); assert_eq!( result.items.iter().map(|item| item.id).collect::>(), @@ -1141,7 +1270,10 @@ mod tests { ); assert_eq!(shared_id, 2); - let restarted = inspector.restart_job(shared_id).expect("restart shared"); + let restarted = inspector + .restart_job(shared_id) + .await + .expect("restart shared"); assert_eq!(restarted, 3); assert!(inspector.job(restarted).await.expect("lookup").is_some()); } diff --git a/crates/openplotva-app/src/runtime_virtual_dialog.rs b/crates/openplotva-app/src/runtime_virtual_dialog.rs index aee8a34d..bb1aa81f 100644 --- a/crates/openplotva-app/src/runtime_virtual_dialog.rs +++ b/crates/openplotva-app/src/runtime_virtual_dialog.rs @@ -794,7 +794,7 @@ fn cleanup_runtime_virtual_live_artifacts( llm_runs: Option<&crate::runtime_llm_runs::RuntimeLlmRunBuffer>, ) -> LiveCleanupReport { let taskman_deleted = taskman - .delete_jobs(RuntimeTaskmanJobsFilter { + .delete_in_memory_jobs(RuntimeTaskmanJobsFilter { chat_id: Some(record.chat_id), user_id: Some(record.user_id), limit: 1_000, diff --git a/crates/openplotva-app/src/task_queue.rs b/crates/openplotva-app/src/task_queue.rs index b8e9aeea..ba5272dc 100644 --- a/crates/openplotva-app/src/task_queue.rs +++ b/crates/openplotva-app/src/task_queue.rs @@ -11,8 +11,7 @@ use std::{ use openplotva_config::PersistentQueueConfig; use openplotva_taskman::{ - InMemoryTaskQueue, TaskQueueIdAllocator, TaskQueueSnapshot, TaskQueueWalRecord, - TaskQueueWalSink, + InMemoryTaskQueue, TaskQueueIdAllocator, TaskQueueWalRecord, TaskQueueWalSink, }; use openplotva_telegram::{ DeleteMessageRequest, TelegramOutboundMethod, build_delete_message_method, @@ -26,6 +25,11 @@ pub const SHARED_TASK_QUEUE_STUCK_SCAN_INTERVAL: Duration = Duration::from_secs( pub const SHARED_TASK_QUEUE_STUCK_DURATION: Duration = Duration::from_secs(4 * 60 * 60); pub const TASK_QUEUE_DB_SYNC_INTERVAL: Duration = Duration::from_secs(1); pub const TASK_QUEUE_DB_SYNC_MUTATION_THRESHOLD: usize = 1_000; +/// Finished jobs stay in memory this long for late message updates and update +/// replay dedupe; Postgres keeps them until the retention purge. +pub const FINISHED_JOB_MEMORY_RETENTION: TimeDuration = TimeDuration::hours(1); +/// Newest finished jobs kept in memory per queue for queue-time estimates. +pub const FINISHED_JOBS_KEPT_PER_QUEUE: usize = 32; pub type TaskQueueStoreFuture<'a, T> = Pin + Send + 'a>>; pub type TaskQueueDurabilityFuture<'a> = @@ -40,10 +44,6 @@ pub trait TaskQueueDurabilityBarrier: Send + Sync { pub trait SharedTaskQueueDurableStore: Clone + Send + Sync + 'static { type Error: fmt::Display + Send + Sync + 'static; - fn load_snapshot<'a>( - &'a self, - ) -> TaskQueueStoreFuture<'a, Result>; - fn apply_wal_batch<'a>( &'a self, batch: Vec, @@ -53,12 +53,6 @@ pub trait SharedTaskQueueDurableStore: Clone + Send + Sync + 'static { impl SharedTaskQueueDurableStore for openplotva_storage::PostgresTaskQueueStore { type Error = openplotva_storage::StorageError; - fn load_snapshot<'a>( - &'a self, - ) -> TaskQueueStoreFuture<'a, Result> { - Box::pin(async move { self.load_task_queue_snapshot().await }) - } - fn apply_wal_batch<'a>( &'a self, batch: Vec, @@ -380,8 +374,8 @@ pub struct SharedTaskQueueRecoveryWorkerReport { pub struct SharedTaskQueueTerminalCleanupWorkerReport { /// Number of cleanup ticks observed. pub ticks: usize, - /// Number of terminal jobs deleted. - pub deleted: usize, + /// Number of finished jobs dropped from memory. + pub evicted: usize, } /// Periodic Postgres hard-purge worker report. @@ -495,7 +489,12 @@ impl SharedTaskQueueRuntime { // so issued ids never regress below a value handed out before this restart. let (next_job_id, next_message_id) = store.reserve_id_high_water().await?; ids.seed_high_water(next_job_id, next_message_id); - let snapshot = store.load_task_queue_snapshot().await?; + let snapshot = store + .load_task_queue_snapshot( + OffsetDateTime::now_utc() - FINISHED_JOB_MEMORY_RETENTION, + FINISHED_JOBS_KEPT_PER_QUEUE as i64, + ) + .await?; let restored_before_requeue = snapshot.records.len(); let journal = BufferedTaskQueueJournal::new(store); let journal_sink: Arc = Arc::new(journal.clone()); @@ -554,8 +553,9 @@ impl SharedTaskQueueRuntime { self.queue.fail_stuck_processing(now, stuck_duration) } - pub fn prune_terminal_before(&self, cutoff: OffsetDateTime) -> Vec { - self.queue.prune_terminal_before(cutoff) + pub fn evict_finished_before(&self, cutoff: OffsetDateTime) -> usize { + self.queue + .evict_terminal_before(cutoff, FINISHED_JOBS_KEPT_PER_QUEUE) } #[must_use] @@ -722,13 +722,12 @@ pub async fn run_shared_task_queue_terminal_cleanup_worker_until( () = &mut stop => break, _ = ticker.tick() => { report.ticks += 1; - let cutoff = OffsetDateTime::now_utc() - retention; - let deleted = runtime.prune_terminal_before(cutoff); - if deleted.is_empty() { + let evicted = runtime.evict_finished_before(OffsetDateTime::now_utc() - retention); + if evicted == 0 { continue; } - report.deleted += deleted.len(); - tracing::warn!(deleted = deleted.len(), "deleted old shared taskman terminal jobs"); + report.evicted += evicted; + tracing::debug!(evicted, "dropped finished shared taskman jobs from memory"); } } } @@ -736,9 +735,11 @@ pub async fn run_shared_task_queue_terminal_cleanup_worker_until( report } -/// Hard-delete soft-deleted jobs and old history from Postgres periodically. The -/// in-memory prune only soft-deletes rows (via delete WAL); without this purge the -/// `taskman_jobs` and `taskman_job_history` tables grow without bound. +/// Hard-delete finished and soft-deleted jobs and old history from Postgres +/// periodically. The in-memory eviction leaves rows untouched; without this purge +/// the `taskman_jobs` and `taskman_job_history` tables grow without bound. +/// Finished jobs stay for twice the retention, the window operator listings and +/// the analytics dashboard had when the in-memory queue held them for one. pub async fn run_shared_task_queue_db_purge_worker_until( store: openplotva_storage::PostgresTaskQueueStore, interval: Duration, @@ -755,8 +756,8 @@ pub async fn run_shared_task_queue_db_purge_worker_until( () = &mut stop => break, _ = ticker.tick() => { report.ticks += 1; - let cutoff = OffsetDateTime::now_utc() - retention; - match store.purge_task_queue_terminal(cutoff).await { + let now = OffsetDateTime::now_utc(); + match store.purge_task_queue_terminal(now - retention.saturating_mul(2), now - retention).await { Ok(0) => {} Ok(purged) => { report.purged += purged as usize; @@ -895,8 +896,7 @@ mod tests { use super::*; use openplotva_taskman::{ DialogJobParams, MESSAGE_STATUS_COMPLETED, MESSAGE_STATUS_PLACEHOLDER, MESSAGE_TYPE_RESULT, - TEXT_QUEUE_NAME, TaskQueueJobMessageParams, TaskQueueRecord, empty_task_queue_snapshot, - new_dialog_job_at, + TEXT_QUEUE_NAME, TaskQueueJobMessageParams, TaskQueueRecord, new_dialog_job_at, }; use std::{ collections::BTreeMap, @@ -1151,12 +1151,6 @@ mod tests { impl SharedTaskQueueDurableStore for TaskQueueDurableStoreStub { type Error = String; - fn load_snapshot<'a>( - &'a self, - ) -> TaskQueueStoreFuture<'a, Result> { - Box::pin(async { Ok(empty_task_queue_snapshot()) }) - } - fn apply_wal_batch<'a>( &'a self, batch: Vec, diff --git a/crates/openplotva-server/src/lib.rs b/crates/openplotva-server/src/lib.rs index 6b8ce88c..2052183c 100644 --- a/crates/openplotva-server/src/lib.rs +++ b/crates/openplotva-server/src/lib.rs @@ -58,9 +58,9 @@ pub use runtime_graphql::{ RuntimeSafetyChecksFilter, RuntimeSqlReadRequest, RuntimeSqlReadResult, RuntimeSqlReader, RuntimeSqlReaderFuture, RuntimeSubscriptionData, RuntimeTaskmanDiagnosticsData, RuntimeTaskmanInspector, RuntimeTaskmanJobData, RuntimeTaskmanJobDetailsData, - RuntimeTaskmanJobFuture, RuntimeTaskmanJobListEntryData, RuntimeTaskmanJobListResultData, - RuntimeTaskmanJobMessageData, RuntimeTaskmanJobSummaryData, RuntimeTaskmanJobsFilter, - RuntimeTaskmanQueueDiagnosticsData, RuntimeTelegramDeliveryFuture, + RuntimeTaskmanJobFuture, RuntimeTaskmanJobListEntryData, RuntimeTaskmanJobListFuture, + RuntimeTaskmanJobListResultData, RuntimeTaskmanJobMessageData, RuntimeTaskmanJobSummaryData, + RuntimeTaskmanJobsFilter, RuntimeTaskmanQueueDiagnosticsData, RuntimeTelegramDeliveryFuture, RuntimeTelegramDeliveryInspector, RuntimeTelegramDeliveryListFilter, RuntimeTelegramOutboxAttemptData, RuntimeTelegramOutboxItemData, RuntimeTelegramOutboxMutationResultData, RuntimeTelegramOutboxRetryRequest, @@ -941,7 +941,7 @@ mod tests { RuntimeSqlReadResult, RuntimeSqlReader, RuntimeSqlReaderFuture, RuntimeSubscriptionData, RuntimeTaskmanDiagnosticsData, RuntimeTaskmanInspector, RuntimeTaskmanJobData, RuntimeTaskmanJobDetailsData, RuntimeTaskmanJobFuture, RuntimeTaskmanJobListEntryData, - RuntimeTaskmanJobListResultData, RuntimeTaskmanJobMessageData, + RuntimeTaskmanJobListFuture, RuntimeTaskmanJobListResultData, RuntimeTaskmanJobMessageData, RuntimeTaskmanJobSummaryData, RuntimeTaskmanJobsFilter, RuntimeTaskmanQueueDiagnosticsData, RuntimeTelegramDeliveryFuture, RuntimeTelegramDeliveryInspector, RuntimeTelegramDeliveryListFilter, RuntimeTelegramOutboxAttemptData, @@ -1826,17 +1826,17 @@ mod tests { struct MemoryTaskmanInspector; impl RuntimeTaskmanInspector for MemoryTaskmanInspector { - fn list_jobs( - &self, + fn list_jobs<'a>( + &'a self, filter: RuntimeTaskmanJobsFilter, - ) -> Result { + ) -> RuntimeTaskmanJobListFuture<'a> { assert_eq!(filter.q, "pay"); assert_eq!(filter.status, vec!["pending"]); assert_eq!(filter.queue, vec!["control"]); assert_eq!(filter.user_id, Some(7)); assert_eq!(filter.chat_id, Some(9)); assert_eq!(filter.limit, 500); - Ok(RuntimeTaskmanJobListResultData { + Box::pin(std::future::ready(Ok(RuntimeTaskmanJobListResultData { total: 1, offset: 0, limit: 500, @@ -1868,7 +1868,7 @@ mod tests { actual_processing_time: None, preview: Some("vip_invoice".to_owned()), }], - }) + }))) } fn job<'a>(&'a self, id: i64) -> RuntimeTaskmanJobFuture<'a> { diff --git a/crates/openplotva-server/src/runtime_graphql.rs b/crates/openplotva-server/src/runtime_graphql.rs index 68ca05ce..644b1db3 100644 --- a/crates/openplotva-server/src/runtime_graphql.rs +++ b/crates/openplotva-server/src/runtime_graphql.rs @@ -65,6 +65,11 @@ pub type RuntimeVirtualDialogFuture<'a, T> = pub type RuntimeTaskmanJobFuture<'a> = Pin, String>> + Send + 'a>>; +/// Boxed future returned by the taskman job listing, which reads persisted +/// storage for finished jobs. +pub type RuntimeTaskmanJobListFuture<'a> = + Pin> + Send + 'a>>; + /// Runtime API Redis diagnostics boundary. pub trait RuntimeRedisInspector: Send + Sync { fn prefix_groups<'a>( @@ -316,10 +321,8 @@ pub trait RuntimeGeminiCachePurger: Send + Sync { /// Runtime API taskman diagnostics boundary. pub trait RuntimeTaskmanInspector: Send + Sync { - fn list_jobs( - &self, - filter: RuntimeTaskmanJobsFilter, - ) -> Result; + fn list_jobs<'a>(&'a self, filter: RuntimeTaskmanJobsFilter) + -> RuntimeTaskmanJobListFuture<'a>; /// Inspect one taskman job by ID, falling back to persisted storage when /// the in-memory queue no longer holds the row. @@ -2171,6 +2174,7 @@ impl RuntimeQuery { let filter = taskman_jobs_filter_from_input(filter)?; let result = inspector .list_jobs(filter) + .await .map_err(async_graphql::Error::new)?; Ok(result) } diff --git a/crates/openplotva-storage/Cargo.toml b/crates/openplotva-storage/Cargo.toml index b215533f..6f0d919d 100644 --- a/crates/openplotva-storage/Cargo.toml +++ b/crates/openplotva-storage/Cargo.toml @@ -24,7 +24,7 @@ sha2.workspace = true sqlx = { workspace = true, features = ["macros", "migrate", "postgres", "runtime-tokio", "time"] } thiserror.workspace = true time.workspace = true -tokio = { workspace = true, features = ["sync"] } +tokio = { workspace = true, features = ["sync", "time"] } tracing.workspace = true [dev-dependencies] diff --git a/crates/openplotva-storage/src/lib.rs b/crates/openplotva-storage/src/lib.rs index f230966a..e36f7124 100644 --- a/crates/openplotva-storage/src/lib.rs +++ b/crates/openplotva-storage/src/lib.rs @@ -43,10 +43,15 @@ pub mod gradius_ads; pub mod llm_routing; pub mod maintenance; pub mod maintenance_notifications; +mod task_queue_listing; pub mod telegram_delivery; pub mod telegram_outbox; pub mod telegram_projection; +pub use task_queue_listing::{ + TASK_QUEUE_ACTIVE_STATUSES, TaskQueueJobFilter, TaskQueueJobPage, TaskQueueSortColumn, + TaskQueueTimeColumn, +}; pub use telegram_delivery::{ ClaimedTelegramUpdate, MATERIALIZED_UPDATE_BINDS_PER_ROW, MaterializationReport, MaterializedUpdateDisposition, MaterializedUpdateInput, PostgresTelegramDeliveryStore, @@ -7104,7 +7109,15 @@ impl PostgresTaskQueueStore { &self.pool } - pub async fn load_task_queue_snapshot(&self) -> Result { + /// Load the jobs the in-memory queue holds: every active job, finished jobs + /// that ended at or after `finished_since`, and the newest + /// `finished_per_queue` finished jobs of each queue. Older finished jobs stay + /// in Postgres only, until the retention purge. + pub async fn load_task_queue_snapshot( + &self, + finished_since: OffsetDateTime, + finished_per_queue: i64, + ) -> Result { // A connection/query failure here is transient and stays fatal (the caller // aborts startup and a restart retries the load). But a single undeserializable // row (schema drift, partial write) must NOT crash the boot, or one bad row @@ -7115,9 +7128,31 @@ impl PostgresTaskQueueStore { SELECT id, record FROM taskman_jobs WHERE deleted_at IS NULL + AND ( + status = ANY($1) + OR COALESCE(completed_at, updated_at, created_at) >= $2 + OR id IN ( + SELECT id + FROM ( + SELECT + id, + row_number() OVER ( + PARTITION BY queue_name + ORDER BY COALESCE(completed_at, updated_at, created_at) DESC, id DESC + ) AS position + FROM taskman_jobs + WHERE deleted_at IS NULL + AND status <> ALL($1) + ) AS finished + WHERE position <= $3 + ) + ) ORDER BY id ASC "#, ) + .bind(TASK_QUEUE_ACTIVE_STATUSES.map(str::to_owned).to_vec()) + .bind(finished_since) + .bind(finished_per_queue) .fetch_all(&self.pool) .await?; let mut records = Vec::with_capacity(rows.len()); @@ -7224,22 +7259,38 @@ impl PostgresTaskQueueStore { Ok((next_job_id.max(1), next_message_id.max(1))) } - /// Hard-delete soft-deleted jobs and history older than the cutoff. Returns the - /// number of rows reclaimed. Soft-deleted rows and history are retention-only; - /// without this the tables grow without bound. + /// Hard-delete finished jobs that ended before `finished_before`, plus + /// soft-deleted jobs and history older than `deleted_before`. Returns the + /// number of rows reclaimed. The in-memory queue drops finished jobs without + /// touching Postgres, so this purge is what bounds the tables. pub async fn purge_task_queue_terminal( &self, - older_than: OffsetDateTime, + finished_before: OffsetDateTime, + deleted_before: OffsetDateTime, ) -> Result { - let jobs = sqlx::query( + let finished = sqlx::query( + r#" + DELETE FROM taskman_jobs + WHERE deleted_at IS NULL + AND status <> ALL($1) + AND COALESCE(completed_at, updated_at, created_at) < $2 + "#, + ) + .bind(TASK_QUEUE_ACTIVE_STATUSES.map(str::to_owned).to_vec()) + .bind(finished_before) + .execute(&self.pool) + .await? + .rows_affected(); + let deleted = sqlx::query( "DELETE FROM taskman_jobs WHERE deleted_at IS NOT NULL AND deleted_at < $1", ) - .bind(older_than) + .bind(deleted_before) .execute(&self.pool) .await? .rows_affected(); + let jobs = finished + deleted; let history = sqlx::query("DELETE FROM taskman_job_history WHERE at < $1") - .bind(older_than) + .bind(deleted_before) .execute(&self.pool) .await? .rows_affected(); @@ -12543,7 +12594,9 @@ mod tests { store.apply_task_queue_wal_batch(vec![upsert]).await?; // The record round-trips through the JSONB column and the generated columns. - let snapshot = store.load_task_queue_snapshot().await?; + let snapshot = store + .load_task_queue_snapshot(time::OffsetDateTime::now_utc(), 0) + .await?; let loaded = snapshot .records .iter() @@ -12582,7 +12635,9 @@ mod tests { record: None, }; store.apply_task_queue_wal_batch(vec![delete]).await?; - let after_delete = store.load_task_queue_snapshot().await?; + let after_delete = store + .load_task_queue_snapshot(time::OffsetDateTime::now_utc(), 0) + .await?; assert!( after_delete.records.iter().all(|r| r.id != start_id), "a soft-deleted job must not load" @@ -12590,8 +12645,9 @@ mod tests { // Purge hard-deletes the soft-deleted row and its history, yet the durable // high-water must NOT regress: a restart can never reissue the id (#1). + let purge_cutoff = time::OffsetDateTime::now_utc() + time::Duration::days(1); let purged = store - .purge_task_queue_terminal(time::OffsetDateTime::now_utc() + time::Duration::days(1)) + .purge_task_queue_terminal(purge_cutoff, purge_cutoff) .await?; assert!( purged >= 1, @@ -12605,6 +12661,189 @@ mod tests { Ok(()) } + #[tokio::test] + async fn live_taskman_startup_load_listing_and_purge_follow_finish_time_when_postgres_dsn_is_set() + -> Result<(), Box> { + let Ok(dsn) = env::var("OPENPLOTVA_TEST_POSTGRES_DSN") else { + return Ok(()); + }; + let pool = PgPoolOptions::new() + .max_connections(2) + .connect(&dsn) + .await?; + super::run_migrations_on(&pool).await?; + let store = super::PostgresTaskQueueStore::new(pool.clone()); + let (start_id, _) = store.reserve_id_high_water().await?; + let queue_name = format!("live-retention-{start_id}"); + let now = time::OffsetDateTime::now_utc(); + let record = |offset: i64, + status: openplotva_taskman::JobStatus, + finished: Option, + title: &str| { + let mut job = openplotva_taskman::new_control_job_at( + openplotva_taskman::ControlJobParams { + chat_id: -200, + message_id: 1, + user_id: 9, + user_full_name: "retention test".to_owned(), + thread_id: None, + data: openplotva_taskman::ControlJobData { + text: title.to_owned(), + ..openplotva_taskman::ControlJobData::default() + }, + }, + now - time::Duration::days(3), + ); + job.title = title.to_owned(); + openplotva_taskman::TaskQueueRecord { + id: start_id + offset, + queue_name: queue_name.clone(), + status, + job, + available_at: None, + debounce_key: None, + lane_key: None, + source_update_ids: Vec::new(), + latest_update_id: None, + pending_dialog_inputs: Vec::new(), + worker_id: None, + started_at: None, + execution_started_at: None, + completed_at: finished, + error: None, + progress_message_id: None, + result_message_id: None, + messages: Vec::new(), + events: vec![openplotva_taskman::TaskQueueJobEvent { + stage: "trace".to_owned(), + ..openplotva_taskman::TaskQueueJobEvent::default() + }], + agent_state: None, + } + }; + use openplotva_taskman::JobStatus::{Completed, Failed, Pending}; + let rows = [ + record(0, Pending, None, "waiting job"), + record(1, Completed, Some(now - time::Duration::days(2)), "ancient"), + record(2, Failed, Some(now - time::Duration::hours(5)), "older"), + record( + 3, + Completed, + Some(now - time::Duration::hours(3)), + "newest old", + ), + record( + 4, + Completed, + Some(now - time::Duration::minutes(5)), + "fresh", + ), + ]; + store + .apply_task_queue_wal_batch( + rows.iter() + .map(|row| openplotva_taskman::TaskQueueWalRecord { + format: openplotva_taskman::TASK_QUEUE_WAL_FORMAT.to_owned(), + op: openplotva_taskman::TASK_QUEUE_WAL_UPSERT_JOB.to_owned(), + job_id: row.id, + record: Some(row.clone()), + }) + .collect(), + ) + .await?; + + let loaded_ids = |snapshot: openplotva_taskman::TaskQueueSnapshot| { + snapshot + .records + .into_iter() + .filter(|loaded| loaded.queue_name == queue_name) + .map(|loaded| loaded.id) + .collect::>() + }; + let snapshot = store + .load_task_queue_snapshot(now - time::Duration::hours(1), 1) + .await?; + assert_eq!( + loaded_ids(snapshot), + vec![start_id, start_id + 4], + "startup loads active jobs plus recently finished ones" + ); + let snapshot = store + .load_task_queue_snapshot(now - time::Duration::minutes(1), 2) + .await?; + assert_eq!( + loaded_ids(snapshot), + vec![start_id, start_id + 3, start_id + 4], + "the newest finished jobs of each queue are loaded regardless of age" + ); + + let filter = super::TaskQueueJobFilter { + queues: vec![queue_name.clone()], + ..super::TaskQueueJobFilter::default() + }; + let page = store + .list_task_queue_jobs(&filter, super::TaskQueueSortColumn::CompletedAt, true, 1, 2) + .await?; + assert_eq!(page.total, 5); + assert_eq!(page.by_status.get("completed"), Some(&3)); + assert_eq!(page.by_queue.get(&queue_name), Some(&5)); + assert_eq!( + page.records.iter().map(|row| row.id).collect::>(), + vec![start_id + 3, start_id + 2] + ); + assert!(page.records.iter().all(|row| row.events.is_empty())); + let searched = store + .list_task_queue_jobs( + &super::TaskQueueJobFilter { + search: "newest".to_owned(), + ..filter.clone() + }, + super::TaskQueueSortColumn::Id, + false, + 0, + 10, + ) + .await?; + assert_eq!( + searched + .records + .iter() + .map(|row| row.id) + .collect::>(), + vec![start_id + 3] + ); + + let deleted = store + .delete_finished_task_queue_jobs( + &super::TaskQueueJobFilter { + statuses: vec!["failed".to_owned(), "pending".to_owned()], + ..filter.clone() + }, + &[], + ) + .await?; + assert_eq!(deleted, 1, "only the finished match is deleted in Postgres"); + + store + .purge_task_queue_terminal(now - time::Duration::days(1), now) + .await?; + let remaining: Vec = + sqlx::query_scalar("SELECT id FROM taskman_jobs WHERE queue_name = $1 ORDER BY id") + .bind(&queue_name) + .fetch_all(&pool) + .await?; + assert_eq!( + remaining, + vec![start_id, start_id + 2, start_id + 3, start_id + 4], + "purge drops jobs finished before the cutoff and keeps the rest" + ); + sqlx::query("DELETE FROM taskman_jobs WHERE queue_name = $1") + .bind(&queue_name) + .execute(&pool) + .await?; + Ok(()) + } + #[test] fn redis_connection_info_matches_go_cache_defaults() -> Result<(), redis::RedisError> { let config = RedisConfig { diff --git a/crates/openplotva-storage/src/task_queue_listing.rs b/crates/openplotva-storage/src/task_queue_listing.rs new file mode 100644 index 00000000..cb90165e --- /dev/null +++ b/crates/openplotva-storage/src/task_queue_listing.rs @@ -0,0 +1,219 @@ +//! Filtered, paged reads of persisted taskman jobs for operator surfaces. +//! +//! The in-memory queue only holds active jobs and a short tail of finished ones, +//! so listings read `taskman_jobs`, which keeps finished jobs until retention. + +use std::collections::BTreeMap; + +use sqlx::{Postgres, QueryBuilder, Row}; +use time::OffsetDateTime; + +use crate::{PostgresTaskQueueStore, StorageError, task_queue_record_from_row}; +use openplotva_taskman::TaskQueueRecord; + +/// Statuses whose jobs the in-memory queue always holds. +pub const TASK_QUEUE_ACTIVE_STATUSES: [&str; 3] = ["pending", "processing", "waiting_delivery"]; + +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub enum TaskQueueTimeColumn { + #[default] + CreatedAt, + StartedAt, + CompletedAt, +} + +impl TaskQueueTimeColumn { + const fn column(self) -> &'static str { + match self { + Self::CreatedAt => "created_at", + Self::StartedAt => "started_at", + Self::CompletedAt => "completed_at", + } + } +} + +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub enum TaskQueueSortColumn { + Id, + Priority, + #[default] + CreatedAt, + StartedAt, + CompletedAt, +} + +impl TaskQueueSortColumn { + const fn column(self) -> &'static str { + match self { + Self::Id => "id", + Self::Priority => "priority", + Self::CreatedAt => "created_at", + Self::StartedAt => "started_at", + Self::CompletedAt => "completed_at", + } + } +} + +/// Filter over persisted jobs. Empty lists and `None` mean "no constraint". +#[derive(Clone, Debug, Default)] +pub struct TaskQueueJobFilter { + pub statuses: Vec, + pub queues: Vec, + pub user_id: Option, + pub chat_id: Option, + pub time_column: TaskQueueTimeColumn, + pub from: Option, + pub to: Option, + /// Lowercase substring matched against queue, type, status, title, user name, + /// and the payload text fields that job previews are built from. + pub search: String, +} + +#[derive(Clone, Debug, Default)] +pub struct TaskQueueJobPage { + pub total: i64, + pub by_status: BTreeMap, + pub by_queue: BTreeMap, + /// Matching records without their event log. + pub records: Vec, +} + +const SEARCH_HAYSTACK: &str = "lower(concat_ws(' ', queue_name, job_type, status, \ + record #>> '{job,title}', \ + record #>> '{job,data,telegram_data,user_full_name}', \ + record #>> '{job,data,control_data,text}', \ + record #>> '{job,data,asr_data,file_unique_id}', \ + record #>> '{job,data,image_data,prompt}', \ + record #>> '{job,data,image_data,original_text}', \ + record #>> '{job,data,music_data,topic}', \ + record #>> '{job,data,music_data,lyrics}', \ + record #>> '{job,data,dialog_data,message_text}', \ + record #>> '{job,data,dialog_data,original_text}', \ + record #>> '{job,data,agent_data,goal}', \ + record #>> '{job,data,agent_data,profile_id}'))"; + +fn push_filter(builder: &mut QueryBuilder, filter: &TaskQueueJobFilter) { + builder.push(" WHERE deleted_at IS NULL"); + if !filter.statuses.is_empty() { + builder + .push(" AND status = ANY(") + .push_bind(filter.statuses.clone()) + .push(")"); + } + if !filter.queues.is_empty() { + builder + .push(" AND queue_name = ANY(") + .push_bind(filter.queues.clone()) + .push(")"); + } + if let Some(user_id) = filter.user_id { + builder + .push(" AND COALESCE(user_id, 0) = ") + .push_bind(user_id); + } + if let Some(chat_id) = filter.chat_id { + builder + .push(" AND COALESCE(chat_id, 0) = ") + .push_bind(chat_id); + } + let time_column = filter.time_column.column(); + if let Some(from) = filter.from { + builder + .push(format_args!(" AND {time_column} >= ")) + .push_bind(from); + } + if let Some(to) = filter.to { + builder + .push(format_args!(" AND {time_column} <= ")) + .push_bind(to); + } + if !filter.search.is_empty() { + builder + .push(format_args!(" AND strpos({SEARCH_HAYSTACK}, ")) + .push_bind(filter.search.clone()) + .push(") > 0"); + } +} + +impl PostgresTaskQueueStore { + /// Page through persisted jobs with per-status and per-queue counts of all + /// matches. Rows with an equal sort value are ordered by id in the same + /// direction; missing timestamps sort last. + pub async fn list_task_queue_jobs( + &self, + filter: &TaskQueueJobFilter, + sort: TaskQueueSortColumn, + descending: bool, + offset: i64, + limit: i64, + ) -> Result { + let mut counts = QueryBuilder::::new( + "SELECT status, queue_name, count(*) AS jobs FROM taskman_jobs", + ); + push_filter(&mut counts, filter); + counts.push(" GROUP BY status, queue_name"); + let mut page = TaskQueueJobPage::default(); + for row in counts.build().fetch_all(&self.pool).await? { + let status: String = row.try_get("status")?; + let queue_name: String = row.try_get("queue_name")?; + let jobs: i64 = row.try_get("jobs")?; + page.total += jobs; + *page.by_status.entry(status).or_default() += jobs; + *page.by_queue.entry(queue_name).or_default() += jobs; + } + if page.total == 0 || offset >= page.total { + return Ok(page); + } + + let direction = if descending { "DESC" } else { "ASC" }; + let mut items = QueryBuilder::::new( + "SELECT id, record - 'events' AS record FROM taskman_jobs", + ); + push_filter(&mut items, filter); + items + .push(format_args!( + " ORDER BY {} {direction} NULLS LAST, id {direction} OFFSET ", + sort.column() + )) + .push_bind(offset.max(0)) + .push(" LIMIT ") + .push_bind(limit.max(0)); + for row in items.build().fetch_all(&self.pool).await? { + let id: i64 = row.try_get("id")?; + match task_queue_record_from_row(row) { + Ok(record) => page.records.push(record), + Err(error) => { + tracing::warn!(job_id = id, %error, "skipping undeserializable taskman job row"); + } + } + } + Ok(page) + } + + /// Soft-delete persisted finished jobs that match `filter`, except the ids the + /// caller already deleted through the in-memory queue. Active jobs are never + /// touched here because the in-memory queue owns them. + pub async fn delete_finished_task_queue_jobs( + &self, + filter: &TaskQueueJobFilter, + skip_ids: &[i64], + ) -> Result { + let mut builder = QueryBuilder::::new( + "WITH deleted AS (UPDATE taskman_jobs SET deleted_at = now(), updated_at = now()", + ); + push_filter(&mut builder, filter); + builder + .push(" AND status <> ALL(") + .push_bind(TASK_QUEUE_ACTIVE_STATUSES.map(str::to_owned).to_vec()) + .push(") AND id <> ALL(") + .push_bind(skip_ids.to_vec()) + .push( + ") RETURNING id, record) \ + INSERT INTO taskman_job_history (job_id, op, record) \ + SELECT id, ", + ) + .push_bind(openplotva_taskman::TASK_QUEUE_WAL_DELETE_JOB) + .push(", record FROM deleted"); + Ok(builder.build().execute(&self.pool).await?.rows_affected()) + } +} diff --git a/crates/openplotva-storage/src/telegram_outbox.rs b/crates/openplotva-storage/src/telegram_outbox.rs index 774a5099..d5df5ed1 100644 --- a/crates/openplotva-storage/src/telegram_outbox.rs +++ b/crates/openplotva-storage/src/telegram_outbox.rs @@ -1,6 +1,6 @@ //! Durable Postgres outbox for Telegram operations. -use std::collections::HashMap; +use std::{collections::HashMap, time::Duration}; use serde_json::Value; use sha2::{Digest, Sha256}; @@ -287,6 +287,16 @@ pub fn deterministic_telegram_operation_id( } /// Postgres-backed Telegram outbox store. +/// Waits between enqueue attempts that could not get a Postgres connection. A +/// pool timeout fails before any statement runs, so a retry cannot enqueue the +/// batch twice; it keeps a finished answer from being lost to a short connection +/// shortage. +const ENQUEUE_POOL_TIMEOUT_RETRY_DELAYS: [Duration; 3] = [ + Duration::from_millis(500), + Duration::from_secs(1), + Duration::from_secs(2), +]; + #[derive(Clone, Debug)] pub struct PostgresTelegramOutboxStore { pool: PgPool, @@ -308,7 +318,25 @@ impl PostgresTelegramOutboxStore { &self, batch: &TelegramOutboxBatchInput, ) -> Result { - enqueue_telegram_outbox_batch(&self.pool, batch).await + let mut retry_delays = ENQUEUE_POOL_TIMEOUT_RETRY_DELAYS.into_iter(); + loop { + match enqueue_telegram_outbox_batch(&self.pool, batch).await { + Err(StorageError::Postgres { + source: sqlx::Error::PoolTimedOut, + }) => { + let Some(delay) = retry_delays.next() else { + return Err(sqlx::Error::PoolTimedOut.into()); + }; + tracing::warn!( + batch_id = %batch.batch_id, + retry_in_ms = delay.as_millis(), + "no Postgres connection for Telegram outbox enqueue; retrying" + ); + tokio::time::sleep(delay).await; + } + result => return result, + } + } } /// Claim at most one currently-sendable operation per ordering key. @@ -1661,6 +1689,54 @@ mod tests { )); } + #[tokio::test] + async fn live_outbox_enqueue_retries_postgres_pool_timeouts() -> Result<(), Box> { + let Ok(dsn) = env::var("OPENPLOTVA_TEST_POSTGRES_DSN") else { + return Ok(()); + }; + let pool = PgPoolOptions::new() + .max_connections(1) + .acquire_timeout(std::time::Duration::from_millis(200)) + .connect(&dsn) + .await?; + crate::run_migrations_on(&pool).await?; + let suffix = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos(); + let batch_id = format!("outbox-pool-retry-test:{suffix}"); + let input = batch(&batch_id, 1); + let blob_hash = input.parts[0] + .blob + .as_ref() + .expect("test blob") + .sha256 + .clone(); + let store = PostgresTelegramOutboxStore::new(pool.clone()); + + let held = pool.acquire().await?; + let release = tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + drop(held); + }); + let queued = store.enqueue_batch(&input).await?; + release.await?; + + assert_eq!(queued.parts.len(), 1); + assert!(queued.parts.iter().all(|part| part.inserted)); + + // Other live tests claim any sendable row, so leave none behind. + sqlx::query("DELETE FROM telegram_outbox WHERE batch_id = $1") + .bind(&batch_id) + .execute(&pool) + .await?; + sqlx::query("DELETE FROM telegram_outbox_blobs WHERE sha256 = $1") + .bind(blob_hash) + .execute(&pool) + .await?; + Ok(()) + } + #[tokio::test] async fn outbox_roundtrip_is_idempotent_ordered_and_fenced() -> Result<(), Box> { let Ok(dsn) = env::var("OPENPLOTVA_TEST_POSTGRES_DSN") else { diff --git a/crates/openplotva-taskman/src/lib.rs b/crates/openplotva-taskman/src/lib.rs index ff4a0adf..3aafecbf 100644 --- a/crates/openplotva-taskman/src/lib.rs +++ b/crates/openplotva-taskman/src/lib.rs @@ -62,6 +62,9 @@ pub const MESSAGE_STATUS_PLACEHOLDER: &str = "placeholder"; pub const MESSAGE_STATUS_COMPLETED: &str = "completed"; pub const MESSAGE_STATUS_FAILED: &str = "failed"; pub const STUCK_JOB_ERROR_MESSAGE: &str = "Job stuck in processing state for too long"; +/// Upserted event that carries a running dialog turn's resumable agent state. +/// It is only needed while the job is active and is dropped when the job ends. +pub const AGENT_CHECKPOINT_EVENT_STAGE: &str = "agent_checkpoint"; #[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] #[serde(rename_all = "snake_case")] @@ -1474,16 +1477,19 @@ impl InMemoryTaskQueue { /// Recreate one job with a fresh ID, preserving queue, priority, title, timeout, and payload. pub fn restart(&self, job_id: i64, created_at: OffsetDateTime) -> Result { - let mut state = self.lock(); - let source = state - .records - .iter() - .find(|record| record.id == job_id) + let source = self + .record(job_id) .ok_or(TaskQueueError::JobNotFound(job_id))?; + Ok(self.restart_from(&source, created_at)) + } + + /// Recreate a job from a record that may no longer be held in memory. + pub fn restart_from(&self, source: &TaskQueueRecord, created_at: OffsetDateTime) -> i64 { let mut job = source.job.clone(); let queue_name = source.queue_name.clone(); job.created = created_at; + let mut state = self.lock(); let id = self.next_job_id(&mut state); let record = TaskQueueRecord { id, @@ -1510,7 +1516,7 @@ impl InMemoryTaskQueue { state.records.push(record.clone()); self.append_wal_record(task_queue_wal_upsert(record)); drop(state); - Ok(id) + id } /// Return one queue record by ID. @@ -1542,9 +1548,12 @@ impl InMemoryTaskQueue { if job_chat_id(&record.job) != chat_id { continue; } - record.status = JobStatus::Cancelled; - record.completed_at = Some(completed_at); - record.error = Some(reason.clone()); + mark_terminal( + record, + JobStatus::Cancelled, + Some(reason.clone()), + completed_at, + ); cancelled.push(record.id); records.push(record.clone()); } @@ -1613,19 +1622,40 @@ impl InMemoryTaskQueue { ids } - pub fn prune_terminal_before(&self, cutoff: OffsetDateTime) -> Vec { + /// Drop terminal jobs that ended before `cutoff` from memory. Postgres keeps + /// them until its own retention purge, so no WAL record is written. The newest + /// `keep_per_queue` terminal jobs of every queue stay for queue-time estimates, + /// and jobs with an undeleted placeholder stay for the placeholder cleanup. + pub fn evict_terminal_before(&self, cutoff: OffsetDateTime, keep_per_queue: usize) -> usize { let mut state = self.lock(); - let mut ids = Vec::new(); + let mut newest_terminal: BTreeMap<&str, Vec<(OffsetDateTime, i64)>> = BTreeMap::new(); + for record in state + .records + .iter() + .filter(|record| !record.status.is_active()) + { + newest_terminal + .entry(record.queue_name.as_str()) + .or_default() + .push((record.completed_at.unwrap_or(record.job.created), record.id)); + } + let kept: BTreeSet = newest_terminal + .into_values() + .flat_map(|mut ended| { + ended.sort_unstable_by(|left, right| right.cmp(left)); + ended.into_iter().take(keep_per_queue).map(|(_, id)| id) + }) + .collect(); + let before = state.records.len(); state.records.retain(|record| { - let prune = task_queue_record_terminal_before(record, cutoff); - if prune { - ids.push(record.id); - } - !prune + kept.contains(&record.id) + || !task_queue_record_terminal_before(record, cutoff) + || record + .messages + .iter() + .any(|message| message.status == MESSAGE_STATUS_PLACEHOLDER) }); - self.append_wal_records(ids.iter().copied().map(task_queue_wal_delete)); - drop(state); - ids + before - state.records.len() } /// Update one worker heartbeat timestamp. @@ -2295,9 +2325,7 @@ impl InMemoryTaskQueue { if record.status == JobStatus::Cancelled && status != JobStatus::Cancelled { return Err(TaskQueueError::JobCancelled(job_id)); } - record.status = status; - record.completed_at = Some(completed_at); - record.error = error; + mark_terminal(record, status, error, completed_at); let record = record.clone(); self.append_wal_record(task_queue_wal_upsert(record)); drop(state); @@ -3576,14 +3604,31 @@ fn fail_stuck_processing_records( if !is_stuck { continue; } - record.status = JobStatus::Failed; - record.completed_at = Some(now); - record.error = Some(STUCK_JOB_ERROR_MESSAGE.to_owned()); + mark_terminal( + record, + JobStatus::Failed, + Some(STUCK_JOB_ERROR_MESSAGE.to_owned()), + now, + ); failed.push(record.id); } failed } +fn mark_terminal( + record: &mut TaskQueueRecord, + status: JobStatus, + error: Option, + completed_at: OffsetDateTime, +) { + record.status = status; + record.completed_at = Some(completed_at); + record.error = error; + record + .events + .retain(|event| event.stage != AGENT_CHECKPOINT_EVENT_STAGE); +} + fn task_queue_record_terminal_before(record: &TaskQueueRecord, cutoff: OffsetDateTime) -> bool { if record.status.is_active() { return false; @@ -3704,7 +3749,7 @@ fn pending_record_matches_message(record: &TaskQueueRecord, chat_id: i64, messag #[cfg(test)] mod tests { use std::{ - collections::HashMap, + collections::{BTreeMap, BTreeSet, HashMap}, sync::{Arc, Mutex}, }; @@ -3712,14 +3757,14 @@ mod tests { use time::{Duration as TimeDuration, OffsetDateTime}; use super::{ - AGENT_QWEN_QUEUE_NAME, ASR_GPU1_QUEUE_NAME, ASR_PRIORITY, ASR_PROCESSING_TIMEOUT_SECONDS, - AgentJobParams, AgentRunStateBlob, AsrJobParams, CONTROL_QUEUE_NAME, ControlJobData, - ControlJobParams, ControlJobPayloadError, ControlKind, DEFAULT_PRIORITY, - DIALOG_AIFARM_QUEUE_NAME, DialogJobData, DialogJobParams, DialogJobPayloadError, - HIGH_PRIORITY, HIGHEST_PRIORITY, IMAGE_REGULAR_QUEUE_NAME, IMAGE_VIP_QUEUE_NAME, - ImageEditJobParams, ImageEditJobPayloadError, ImageGenJobParams, ImageGenJobPayloadError, - ImageJobData, InMemoryTaskQueue, JobPayload, JobStatus, JobType, LOWEST_PRIORITY, - MAX_JOB_EVENTS, MESSAGE_STATUS_COMPLETED, MESSAGE_STATUS_FAILED, + AGENT_CHECKPOINT_EVENT_STAGE, AGENT_QWEN_QUEUE_NAME, ASR_GPU1_QUEUE_NAME, ASR_PRIORITY, + ASR_PROCESSING_TIMEOUT_SECONDS, AgentJobParams, AgentRunStateBlob, AsrJobParams, + CONTROL_QUEUE_NAME, ControlJobData, ControlJobParams, ControlJobPayloadError, ControlKind, + DEFAULT_PRIORITY, DIALOG_AIFARM_QUEUE_NAME, DialogJobData, DialogJobParams, + DialogJobPayloadError, HIGH_PRIORITY, HIGHEST_PRIORITY, IMAGE_REGULAR_QUEUE_NAME, + IMAGE_VIP_QUEUE_NAME, ImageEditJobParams, ImageEditJobPayloadError, ImageGenJobParams, + ImageGenJobPayloadError, ImageJobData, InMemoryTaskQueue, JobPayload, JobStatus, JobType, + LOWEST_PRIORITY, MAX_JOB_EVENTS, MESSAGE_STATUS_COMPLETED, MESSAGE_STATUS_FAILED, MESSAGE_STATUS_PLACEHOLDER, MESSAGE_TYPE_RESULT, MUSIC_VIP_QUEUE_NAME, MusicGenJobParams, MusicGenJobPayloadError, QUEUE_ESTIMATE_DEFAULT_SAMPLE_SIZE, QUEUE_ESTIMATE_MIN_SAMPLE_SIZE, STUCK_JOB_ERROR_MESSAGE, TASK_QUEUE_SNAPSHOT_FORMAT, @@ -5665,39 +5710,150 @@ mod tests { } #[test] - fn in_memory_task_queue_prunes_old_terminal_records() -> Result<(), Box> - { - let queue = InMemoryTaskQueue::new(); + fn in_memory_task_queue_evicts_old_terminal_records_without_wal() + -> Result<(), Box> { + let journal = Arc::new(WalSinkStub::default()); + let queue = InMemoryTaskQueue::new_with_journal(journal.clone()); let now = OffsetDateTime::from_unix_timestamp(1_779_193_800)?; let old = now - TimeDuration::days(2); - let recent = now - TimeDuration::hours(1); - let cutoff = now - TimeDuration::days(1); + let recent = now - TimeDuration::minutes(10); + let cutoff = now - TimeDuration::hours(1); let old_done = queue.assign( TEXT_QUEUE_NAME, job_at("old", DEFAULT_PRIORITY, old, 1, 1, 1), ); - let recent_done = queue.assign( + let newest_old_done = queue.assign( TEXT_QUEUE_NAME, - job_at("recent", DEFAULT_PRIORITY, recent, 2, 1, 2), + job_at("newest old", DEFAULT_PRIORITY, old, 2, 1, 2), + ); + let recent_done = queue.assign( + CONTROL_QUEUE_NAME, + job_at("recent", DEFAULT_PRIORITY, recent, 3, 1, 3), ); let active = queue.assign( TEXT_QUEUE_NAME, - job_at("active", DEFAULT_PRIORITY, old, 3, 1, 3), + job_at("active", DEFAULT_PRIORITY, old, 4, 1, 4), + ); + let placeholder_done = queue.assign( + IMAGE_REGULAR_QUEUE_NAME, + job_at("placeholder", DEFAULT_PRIORITY, old, 5, 1, 5), + ); + let newest_image_done = queue.assign( + IMAGE_REGULAR_QUEUE_NAME, + job_at("newest image", DEFAULT_PRIORITY, old, 6, 1, 6), + ); + let old_image_done = queue.assign( + IMAGE_REGULAR_QUEUE_NAME, + job_at("old image", DEFAULT_PRIORITY, old, 7, 1, 7), ); queue.complete(old_done, old)?; + queue.complete(newest_old_done, old + TimeDuration::minutes(1))?; queue.fail(recent_done, "recent failure", recent)?; + queue.create_job_message(TaskQueueJobMessageParams { + job_id: placeholder_done, + message_type: "progress".to_owned(), + chat_id: 1, + message_id: 50, + created_at: old, + status: MESSAGE_STATUS_PLACEHOLDER.to_owned(), + })?; + queue.complete(placeholder_done, old)?; + queue.complete(newest_image_done, old + TimeDuration::minutes(5))?; + queue.complete(old_image_done, old - TimeDuration::minutes(1))?; + let wal_before = journal.records().len(); - assert_eq!(queue.prune_terminal_before(cutoff), vec![old_done]); - let statuses: HashMap = queue + assert_eq!(queue.evict_terminal_before(cutoff, 1), 2); + + let remaining: BTreeSet = queue .records() .into_iter() - .map(|record| (record.id, record.status)) + .map(|record| record.id) .collect(); - assert_eq!(statuses.get(&old_done), None); - assert_eq!(statuses.get(&recent_done), Some(&JobStatus::Failed)); - assert_eq!(statuses.get(&active), Some(&JobStatus::Pending)); + assert_eq!( + remaining, + BTreeSet::from([ + newest_old_done, + recent_done, + active, + placeholder_done, + newest_image_done + ]) + ); + assert_eq!(journal.records().len(), wal_before); + Ok(()) + } + + #[test] + fn terminal_transitions_drop_the_agent_checkpoint_event() + -> Result<(), Box> { + let now = OffsetDateTime::from_unix_timestamp(1_779_193_800)?; + let queue = InMemoryTaskQueue::new(); + let completed = queue.assign( + TEXT_QUEUE_NAME, + job_at("completed", DEFAULT_PRIORITY, now, 1, 1, 1), + ); + let cancelled = queue.assign( + TEXT_QUEUE_NAME, + job_at("cancelled", DEFAULT_PRIORITY, now, 2, 7, 2), + ); + for id in [completed, cancelled] { + queue.upsert_job_event( + id, + TaskQueueJobEvent { + stage: AGENT_CHECKPOINT_EVENT_STAGE.to_owned(), + data: BTreeMap::from([("checkpoint".to_owned(), "{}".to_owned())]), + ..TaskQueueJobEvent::default() + }, + now, + )?; + queue.append_job_event( + id, + TaskQueueJobEvent { + stage: "dialog_reply".to_owned(), + ..TaskQueueJobEvent::default() + }, + now, + )?; + } + + queue.complete(completed, now)?; + queue.cancel_user_jobs(2, 7, "reset", now); + + for id in [completed, cancelled] { + let stages: Vec = queue + .record(id) + .expect("terminal record") + .events + .into_iter() + .map(|event| event.stage) + .collect(); + assert_eq!(stages, vec!["dialog_reply".to_owned()]); + } + Ok(()) + } + + #[test] + fn restart_from_recreates_a_job_that_is_no_longer_in_memory() + -> Result<(), Box> { + let now = OffsetDateTime::from_unix_timestamp(1_779_193_800)?; + let source_queue = InMemoryTaskQueue::new(); + let id = source_queue.assign( + TEXT_QUEUE_NAME, + job_at("evicted", DEFAULT_PRIORITY, now, 1, 1, 1), + ); + source_queue.fail(id, "boom", now)?; + let source = source_queue.record(id).expect("failed record"); + + let queue = InMemoryTaskQueue::new(); + let restarted = queue.restart_from(&source, now + TimeDuration::minutes(5)); + + let record = queue.record(restarted).expect("restarted record"); + assert_eq!(record.status, JobStatus::Pending); + assert_eq!(record.queue_name, TEXT_QUEUE_NAME); + assert_eq!(record.job.title, "evicted"); + assert_eq!(record.job.created, now + TimeDuration::minutes(5)); Ok(()) } diff --git a/deploy/production/deploy-production.sh b/deploy/production/deploy-production.sh index ff39923d..66f13bcf 100755 --- a/deploy/production/deploy-production.sh +++ b/deploy/production/deploy-production.sh @@ -430,7 +430,9 @@ wait_for_http() { local url="$1" local name="$2" local output="${deploy_root}/${name}.json" - for _ in $(seq 1 90); do + # Startup restores the task queue and catches up the update backlog before + # health and readiness pass; allow up to 10 minutes for each. + for _ in $(seq 1 300); do if curl -fsS "$url" >"$output" 2>/dev/null; then log "${name} ok" return 0 diff --git a/docs/CODEBASE_MAP.md b/docs/CODEBASE_MAP.md index 5308864f..57f29b16 100644 --- a/docs/CODEBASE_MAP.md +++ b/docs/CODEBASE_MAP.md @@ -465,6 +465,11 @@ desc → created asc → id asc). Per-(chat,thread) lane lock on `dialog-aifarm` Two-tier timeouts (queue-start vs execution-start) so media queue-wait is excluded. Agent resumability: `release_orphaned_processing_for_startup` requeues non-agents but keeps agent rows in Processing for re-adoption. +Memory holds active jobs plus a short finished tail (1 h, the newest 32 per +queue, jobs with an undeleted placeholder); eviction writes no WAL. Startup loads +the same set, operator listings of finished jobs read `taskman_jobs`, and the +Postgres purge keeps finished rows for twice the completed-job retention. The +dialog `agent_checkpoint` event is dropped when a job finishes. #### `openplotva-updates` Redis update queue (list or stream backend) + the `openplotva-enqueue-update` diff --git a/migrations/201_drop_chat_interest_rollups.down.sql b/migrations/201_drop_chat_interest_rollups.down.sql new file mode 100644 index 00000000..53a2960a --- /dev/null +++ b/migrations/201_drop_chat_interest_rollups.down.sql @@ -0,0 +1,3 @@ +-- One-way: the chat_history/interest rollup rows are not restorable, and the +-- writer that produced them was removed with this change. +SELECT 1; diff --git a/migrations/201_drop_chat_interest_rollups.up.sql b/migrations/201_drop_chat_interest_rollups.up.sql new file mode 100644 index 00000000..17e0651f --- /dev/null +++ b/migrations/201_drop_chat_interest_rollups.up.sql @@ -0,0 +1,16 @@ +-- Retire the chat_history/interest telemetry rollup. Nothing reads it: dialog +-- memory context comes from memory_episodes directly. It re-aggregated every +-- memory episode on each analytics pass, so it held most of telemetry_rollups +-- and rewrote it daily. Keep the other rollups and rebuild the table without the +-- interest rows; TRUNCATE frees the space at once instead of leaving dead tuples +-- for vacuum. +CREATE TEMP TABLE kept_telemetry_rollups ON COMMIT DROP AS +SELECT * +FROM telemetry_rollups +WHERE NOT (source = 'chat_history' AND kind = 'interest'); + +TRUNCATE telemetry_rollups; + +INSERT INTO telemetry_rollups +SELECT * +FROM kept_telemetry_rollups;