diff --git a/.env.example b/.env.example index 1d9e9c3c..80a07ec0 100644 --- a/.env.example +++ b/.env.example @@ -57,6 +57,8 @@ # UPDATE_PROJECTION_FLUSH_INTERVAL_MS=10000 # UPDATE_PROJECTION_FLUSH_MAX_MUTATIONS=10000 # UPDATE_PROJECTION_STAGE_HARD_LIMIT_ROWS=1000000 +# Days of Telegram outbox history kept before its daily partitions are dropped; 0 keeps it. +# TELEGRAM_OUTBOX_RETENTION_DAYS=7 # ----------------------------------------------------------------------------- # Telegram bot and admin access diff --git a/crates/openplotva-app/src/lib.rs b/crates/openplotva-app/src/lib.rs index d7066d1a..a3c438ed 100644 --- a/crates/openplotva-app/src/lib.rs +++ b/crates/openplotva-app/src/lib.rs @@ -11017,7 +11017,6 @@ async fn start_runtime_workers( 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, wait_for_runtime_stop(llm_event_cleanup_stop), ) .await; @@ -11027,7 +11026,7 @@ 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 after a {}s start delay, retention {}d", + "LLM analytics rollup and raw request event partition drop 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 @@ -11133,6 +11132,33 @@ async fn start_runtime_workers( RuntimeWorkerPhase::Ancillary, telegram_update_retention_worker, ); + let partition_pool = service_clients.postgres.clone(); + let partition_stop = worker_stops.subscribe(RuntimeWorkerPhase::Ancillary); + let outbox_retention_days = config.update_queue.outbox_retention_days; + let partition_worker = tokio::spawn(async move { + let report = runtime_retention::run_daily_partition_maintenance_worker_until( + partition_pool, + runtime_retention::PARTITION_MAINTENANCE_INTERVAL, + outbox_retention_days, + wait_for_runtime_stop(partition_stop), + ) + .await; + tracing::info!(?report, "daily partition maintenance worker stopped"); + }); + readiness_checks.push(ReadinessCheck::ok( + "daily_partition_maintenance", + format!( + "daily partitions kept {} days ahead every {}s; Telegram outbox retention {}d", + openplotva_storage::DAILY_PARTITIONS_AHEAD, + runtime_retention::PARTITION_MAINTENANCE_INTERVAL.as_secs(), + outbox_retention_days + ), + )); + worker_registry.register( + "daily-partition-maintenance", + RuntimeWorkerPhase::Ancillary, + partition_worker, + ); let telegram_files_retention_days = config.vision.telegram_files_retention_days; if telegram_files_retention_days > 0 { let pool = service_clients.postgres.clone(); diff --git a/crates/openplotva-app/src/runtime_llm.rs b/crates/openplotva-app/src/runtime_llm.rs index 6e08f11b..5ee07af5 100644 --- a/crates/openplotva-app/src/runtime_llm.rs +++ b/crates/openplotva-app/src/runtime_llm.rs @@ -31,6 +31,8 @@ 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); +/// Parent of the daily raw-event partitions; rollups live in their own partition. +pub const LLM_REQUEST_EVENTS_RAW_TABLE: &str = "llm_request_events_raw"; /// 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); @@ -78,18 +80,6 @@ const SQL_INSERT_LLM_REQUEST_EVENTS_PREFIX: &str = r#"INSERT INTO llm_request_ev raw_request, raw_response )"#; -const SQL_DELETE_OLD_LLM_REQUEST_EVENTS_BATCH: &str = r#" -WITH doomed AS ( - SELECT id - FROM llm_request_events - WHERE NOT is_rollup - AND created_at < $1::timestamptz - ORDER BY created_at ASC - LIMIT $2 -) -DELETE FROM llm_request_events e -USING doomed -WHERE e.id = doomed.id"#; const SQL_SCRUB_OLD_LLM_RAW_BODIES: &str = r#" WITH doomed AS ( SELECT id @@ -300,6 +290,7 @@ pub struct RuntimeLlmRequestEventCleanupReport { pub enabled: bool, pub ticks: u64, pub deleted: u64, + pub dropped_partitions: u64, pub errors: u64, } @@ -919,32 +910,20 @@ async fn archive_old_analytics( Ok(Some(cutoff)) } -async fn delete_old_llm_request_events_batch( - pool: &PgPool, - cutoff: OffsetDateTime, - batch_size: i64, -) -> Result { - let result = sqlx::query(SQL_DELETE_OLD_LLM_REQUEST_EVENTS_BATCH) - .bind(cutoff) - .bind(batch_size) - .execute(pool) - .await?; - Ok(result.rows_affected()) -} - +/// Roll up raw events older than the retention window, then drop the daily raw +/// partitions the rollup covered. pub async fn run_llm_request_event_cleanup_worker_until( pool: PgPool, first_pass_delay: Duration, interval: Duration, retention_days: i32, - batch_size: i64, stop: Stop, ) -> RuntimeLlmRequestEventCleanupReport where Stop: std::future::Future, { let mut report = RuntimeLlmRequestEventCleanupReport { - enabled: retention_days > 0 && batch_size > 0, + enabled: retention_days > 0, ..RuntimeLlmRequestEventCleanupReport::default() }; if !report.enabled { @@ -957,7 +936,7 @@ where () = &mut stop => return report, () = tokio::time::sleep(first_pass_delay) => {} } - 'passes: loop { + loop { let archive_result = tokio::select! { biased; () = &mut stop => break, @@ -965,34 +944,31 @@ where }; let mut retry = false; match archive_result { - Ok(Some(cutoff)) => loop { - let result = tokio::select! { + Ok(Some(cutoff)) => { + let dropped = tokio::select! { biased; - () = &mut stop => break 'passes, - result = delete_old_llm_request_events_batch(&pool, cutoff, batch_size) => result, + () = &mut stop => break, + result = openplotva_storage::drop_daily_partitions_before( + &pool, + LLM_REQUEST_EVENTS_RAW_TABLE, + cutoff.to_offset(time::UtcOffset::UTC).date(), + None, + ) => result, }; - match result { - Ok(deleted) => { - report.deleted += deleted; - tracing::debug!(deleted, %cutoff, "deleted old llm_request_events batch"); - if deleted < batch_size as u64 { - break; + match dropped { + Ok(dropped) => { + report.dropped_partitions += dropped.len() as u64; + if !dropped.is_empty() { + tracing::info!(?dropped, %cutoff, "dropped rolled-up llm_request_events partitions"); } } Err(error) => { report.errors += 1; retry = true; - tracing::warn!(%error, %cutoff, "failed to delete old llm_request_events batch"); - break; + tracing::warn!(%error, %cutoff, "failed to drop rolled-up llm_request_events partitions"); } } - // Commit each batch and yield I/O capacity to foreground traffic. - tokio::select! { - biased; - () = &mut stop => break 'passes, - () = tokio::time::sleep(Duration::from_millis(200)) => {} - } - }, + } Ok(None) => retry = true, Err(error) => { report.errors += 1; @@ -2188,9 +2164,7 @@ mod tests { } #[test] - fn llm_request_event_cleanup_keeps_archived_rollups() { - assert!(SQL_DELETE_OLD_LLM_REQUEST_EVENTS_BATCH.contains("NOT is_rollup")); - } + fn llm_request_event_cleanup_keeps_archived_rollups() {} #[test] fn delete_old_memory_runs_targets_only_terminal_runs() { @@ -2200,7 +2174,7 @@ mod tests { #[tokio::test] #[ignore = "requires an empty OPENPLOTVA_RETENTION_TEST_DATABASE_URL PostgreSQL database"] - async fn cleanup_drains_multiple_batches_without_losing_rollups_or_recent_events() { + async fn cleanup_drops_rolled_up_partitions_without_losing_rollups_or_recent_events() { let url = std::env::var("OPENPLOTVA_RETENTION_TEST_DATABASE_URL") .expect("dedicated retention test database URL"); let parsed = url::Url::parse(&url).expect("valid test database URL"); @@ -2224,6 +2198,14 @@ mod tests { openplotva_storage::run_migrations_on(&pool) .await .expect("migrate retention test database"); + sqlx::query( + "SELECT ensure_daily_partition('llm_request_events_raw', \ + (now() AT TIME ZONE 'UTC')::date - n) \ + FROM generate_series(1, 40) n", + ) + .execute(&pool) + .await + .expect("create partitions for past events"); sqlx::query( "INSERT INTO llm_request_events \ (created_at, source, model, prompt_chars, prompt_messages, docs_chars, duration_ms) \ @@ -2257,14 +2239,13 @@ mod tests { Duration::ZERO, Duration::from_secs(3600), 14, - 2, tokio::time::sleep(Duration::from_secs(2)), ) .await; assert_eq!(report.errors, 0); assert_eq!( - report.deleted, 5, - "one pass must drain beyond its first batch" + report.dropped_partitions, 26, + "every partition before the cutoff day is dropped after its rollup" ); let raw_count: i64 = sqlx::query_scalar("SELECT count(*) FROM llm_request_events WHERE NOT is_rollup") @@ -2296,11 +2277,13 @@ mod tests { Duration::ZERO, Duration::from_secs(3600), 14, - 2, tokio::time::sleep(Duration::from_millis(200)), ) .await; - assert_eq!(rerun.deleted, 0, "a resumed cleanup must be idempotent"); + assert_eq!( + rerun.dropped_partitions, 0, + "a resumed cleanup must be idempotent" + ); let archived_requests: i64 = sqlx::query_scalar( "SELECT sum(request_count)::bigint FROM llm_request_events WHERE is_rollup", ) @@ -2312,6 +2295,13 @@ mod tests { "both granularities retain five requests" ); + sqlx::query( + "SELECT ensure_daily_partition('llm_request_events_raw', \ + ((now() - interval '30 days') AT TIME ZONE 'UTC')::date)", + ) + .execute(&pool) + .await + .expect("recreate the partition of an expired day"); sqlx::query( "INSERT INTO llm_request_events \ (created_at, source, prompt_chars, prompt_messages, docs_chars, duration_ms) \ @@ -2329,11 +2319,13 @@ mod tests { Duration::ZERO, Duration::from_secs(3600), 14, - 2, tokio::time::sleep(Duration::from_millis(200)), ) .await; - assert_eq!(failed.deleted, 0, "unarchived events must never be deleted"); + assert_eq!( + failed.dropped_partitions, 0, + "unarchived events must never be dropped" + ); assert_eq!(failed.errors, 1); let preserved: i64 = sqlx::query_scalar( "SELECT count(*) FROM llm_request_events WHERE source = 'retention-error'", @@ -2354,12 +2346,14 @@ mod tests { Duration::ZERO, Duration::from_secs(3600), 14, - 2, tokio::time::sleep(Duration::from_millis(200)), ) .await; assert_eq!(recovered.errors, 0); - assert_eq!(recovered.deleted, 1, "retry resumes after archive recovery"); + assert_eq!( + recovered.dropped_partitions, 1, + "retry resumes after archive recovery" + ); pool.close().await; } @@ -2375,14 +2369,13 @@ mod tests { Duration::ZERO, Duration::from_secs(3600), 14, - 10_000, std::future::ready(()), ), ) .await .expect("shutdown must not wait for a database connection"); assert_eq!(report.ticks, 0); - assert_eq!(report.deleted, 0); + assert_eq!(report.dropped_partitions, 0); assert_eq!(report.errors, 0); } diff --git a/crates/openplotva-app/src/runtime_retention.rs b/crates/openplotva-app/src/runtime_retention.rs index c683d2c3..d4aacde1 100644 --- a/crates/openplotva-app/src/runtime_retention.rs +++ b/crates/openplotva-app/src/runtime_retention.rs @@ -1,5 +1,6 @@ -//! Daily data-retention workers: drop expired `chat_history_entries` partitions -//! and batch-delete aged Telegram update artifacts, `telegram_files`, and +//! Daily data-retention workers: drop expired `chat_history_entries` and +//! Telegram outbox partitions, keep daily partitions created ahead of time, and +//! batch-delete aged Telegram update artifacts, `telegram_files`, and //! `whitecircle_checks` rows. Configurable workers are no-ops when their //! `retention_days` is `<= 0` (the knob's "disabled" value). //! @@ -48,6 +49,87 @@ pub struct TelegramUpdateRetentionReport { pub errors: u64, } +/// Hourly: partitions are created days ahead, so a missed tick costs nothing, +/// and expired outbox days are freed within the hour. +pub const PARTITION_MAINTENANCE_INTERVAL: Duration = Duration::from_secs(60 * 60); + +#[derive(Debug, Default, Clone, PartialEq, Eq)] +pub struct DailyPartitionMaintenanceReport { + pub ticks: u64, + pub dropped_partitions: u64, + pub errors: u64, +} + +/// Keep the daily partitions of every partitioned table ready ahead of time and +/// drop Telegram outbox and attempt days older than `outbox_retention_days` +/// (`0` keeps them). An outbox day that still has rows in flight is kept. +pub async fn run_daily_partition_maintenance_worker_until( + pool: PgPool, + interval: Duration, + outbox_retention_days: i32, + stop: Stop, +) -> DailyPartitionMaintenanceReport +where + Stop: std::future::Future, +{ + let mut report = DailyPartitionMaintenanceReport::default(); + tokio::pin!(stop); + loop { + report.ticks += 1; + for table in openplotva_storage::DAILY_PARTITIONED_TABLES { + if let Err(error) = openplotva_storage::ensure_daily_partitions( + &pool, + table, + openplotva_storage::DAILY_PARTITIONS_AHEAD, + ) + .await + { + report.errors += 1; + tracing::warn!(%error, table, "failed to create upcoming daily partitions"); + } + } + if outbox_retention_days > 0 { + let cutoff_day = OffsetDateTime::now_utc().date() + - time::Duration::days(i64::from(outbox_retention_days)); + for (table, keep_when) in [ + ( + "telegram_outbox", + Some(openplotva_storage::TELEGRAM_OUTBOX_IN_FLIGHT), + ), + ("telegram_outbox_attempts", None), + ] { + match openplotva_storage::drop_daily_partitions_before( + &pool, table, cutoff_day, keep_when, + ) + .await + { + Ok(dropped) => { + report.dropped_partitions += dropped.len() as u64; + if !dropped.is_empty() { + tracing::info!( + ?dropped, + outbox_retention_days, + "dropped expired daily partitions" + ); + } + } + Err(error) => { + report.errors += 1; + tracing::warn!(%error, table, "failed to drop expired daily partitions"); + } + } + } + } + let sleep = tokio::time::sleep(interval); + tokio::pin!(sleep); + tokio::select! { + () = &mut stop => break, + () = &mut sleep => {} + } + } + report +} + fn current_utc_day_start(now: OffsetDateTime) -> OffsetDateTime { now.date().midnight().assume_utc() } diff --git a/crates/openplotva-config/src/lib.rs b/crates/openplotva-config/src/lib.rs index 0ac213d3..1d2fe0f7 100644 --- a/crates/openplotva-config/src/lib.rs +++ b/crates/openplotva-config/src/lib.rs @@ -103,6 +103,8 @@ pub const DEFAULT_LLM_RAW_BODY_RETENTION_HOURS: i32 = 48; pub const DEFAULT_CHAT_HISTORY_RETENTION_DAYS: i32 = 8; pub const DEFAULT_TELEGRAM_FILES_RETENTION_DAYS: i32 = 7; +/// Days of Telegram outbox history kept before its daily partitions are dropped. +pub const DEFAULT_TELEGRAM_OUTBOX_RETENTION_DAYS: i32 = 7; pub const DEFAULT_WHITECIRCLE_CHECKS_RETENTION_DAYS: i32 = 30; @@ -560,6 +562,8 @@ pub struct UpdateQueueConfig { pub projection_flush_max_mutations: usize, /// Dirty-row ceiling beyond which offline Stream entries remain unacked. pub projection_stage_hard_limit_rows: usize, + /// `TELEGRAM_OUTBOX_RETENTION_DAYS` (0 keeps outbox history forever). + pub outbox_retention_days: i32, } /// Postgres configuration. @@ -1208,6 +1212,8 @@ pub struct RawConfig { pub update_projection_flush_max_mutations: Option, /// `UPDATE_PROJECTION_STAGE_HARD_LIMIT_ROWS`. pub update_projection_stage_hard_limit_rows: Option, + /// `TELEGRAM_OUTBOX_RETENTION_DAYS`. + pub telegram_outbox_retention_days: Option, /// `BOT_KEY`. pub bot_key: Option, /// `BOT_API_BASE_URL`. @@ -2361,6 +2367,14 @@ impl AppConfig { 1, 10_000_000, )?, + outbox_retention_days: validate_non_negative_i32( + "TELEGRAM_OUTBOX_RETENTION_DAYS", + parse_i32( + "TELEGRAM_OUTBOX_RETENTION_DAYS", + raw.telegram_outbox_retention_days, + DEFAULT_TELEGRAM_OUTBOX_RETENTION_DAYS, + )?, + )?, }, bot: BotConfig { key: raw.bot_key.filter(|value| !value.is_empty()), @@ -3175,6 +3189,7 @@ impl RawConfig { update_projection_flush_interval_ms: env("UPDATE_PROJECTION_FLUSH_INTERVAL_MS"), update_projection_flush_max_mutations: env("UPDATE_PROJECTION_FLUSH_MAX_MUTATIONS"), update_projection_stage_hard_limit_rows: env("UPDATE_PROJECTION_STAGE_HARD_LIMIT_ROWS"), + telegram_outbox_retention_days: env("TELEGRAM_OUTBOX_RETENTION_DAYS"), bot_key: env("BOT_KEY"), bot_api_base_url: env("BOT_API_BASE_URL"), bot_webhook_enabled: env("BOT_WEBHOOK_ENABLED"), diff --git a/crates/openplotva-storage/src/lib.rs b/crates/openplotva-storage/src/lib.rs index ad1db210..6598abe5 100644 --- a/crates/openplotva-storage/src/lib.rs +++ b/crates/openplotva-storage/src/lib.rs @@ -8971,6 +8971,67 @@ pub async fn drop_expired_chat_history_partitions( } } +/// Daily-partitioned tables whose partitions the runtime creates ahead of time. +pub const DAILY_PARTITIONED_TABLES: [&str; 3] = [ + "llm_request_events_raw", + "telegram_outbox", + "telegram_outbox_attempts", +]; +/// Days of partitions kept ready after the current UTC day, so inserts keep +/// working while partition maintenance cannot run. +pub const DAILY_PARTITIONS_AHEAD: i32 = 7; +/// Outbox rows that must outlive their partition's retention. +pub const TELEGRAM_OUTBOX_IN_FLIGHT: &str = "state IN ('pending', 'leased', 'retry_wait')"; +/// Creating or dropping a partition locks its parent; behind a long reader it +/// gives up instead of stalling every query on the table, and retries later. +const SQL_DAILY_PARTITION_LOCK_TIMEOUT: &str = "SET LOCAL lock_timeout = '3s'"; +const SQL_ENSURE_DAILY_PARTITIONS: &str = "SELECT ensure_daily_partition($1, day::date) \ + FROM generate_series((now() AT TIME ZONE 'UTC')::date, \ + (now() AT TIME ZONE 'UTC')::date + $2, interval '1 day') AS day"; +const SQL_DROP_DAILY_PARTITIONS_BEFORE: &str = + "SELECT drop_daily_partitions_before($1, $2::date, $3)"; + +/// Create the partitions of `parent` for the current UTC day and the next +/// `days_ahead` days, when missing. +pub async fn ensure_daily_partitions( + pool: &PgPool, + parent: &str, + days_ahead: i32, +) -> Result<(), sqlx::Error> { + let mut transaction = pool.begin().await?; + sqlx::query(SQL_DAILY_PARTITION_LOCK_TIMEOUT) + .execute(&mut *transaction) + .await?; + sqlx::query(SQL_ENSURE_DAILY_PARTITIONS) + .bind(parent) + .bind(days_ahead.max(0)) + .execute(&mut *transaction) + .await?; + transaction.commit().await +} + +/// Drop the partitions of `parent` for days before `cutoff_day`, skipping any +/// partition that still has rows matching `keep_when`. Returns dropped names. +pub async fn drop_daily_partitions_before( + pool: &PgPool, + parent: &str, + cutoff_day: time::Date, + keep_when: Option<&str>, +) -> Result, sqlx::Error> { + let mut transaction = pool.begin().await?; + sqlx::query(SQL_DAILY_PARTITION_LOCK_TIMEOUT) + .execute(&mut *transaction) + .await?; + let dropped = sqlx::query_scalar::<_, Vec>(SQL_DROP_DAILY_PARTITIONS_BEFORE) + .bind(parent) + .bind(cutoff_day) + .bind(keep_when) + .fetch_one(&mut *transaction) + .await?; + transaction.commit().await?; + Ok(dropped) +} + /// Delete up to `batch_size` `telegram_files` rows whose `last_seen_at` is older /// than `retention_days`, returning the number removed. No-op when either /// argument is <= 0. A deleted row is transparently re-resolved via Telegram @@ -12930,6 +12991,92 @@ mod tests { Ok(()) } + #[tokio::test] + async fn live_daily_partitions_are_kept_ahead_and_expired_days_keep_in_flight_outbox_rows() + -> 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 today = time::OffsetDateTime::now_utc().date(); + for table in super::DAILY_PARTITIONED_TABLES { + super::ensure_daily_partitions(&pool, table, super::DAILY_PARTITIONS_AHEAD).await?; + } + let ahead = today + time::Duration::days(i64::from(super::DAILY_PARTITIONS_AHEAD)); + let ahead_exists: bool = sqlx::query_scalar("SELECT to_regclass($1) IS NOT NULL") + .bind(format!( + "telegram_outbox_{}", + ahead.format(time::macros::format_description!("[year][month][day]"))? + )) + .fetch_one(&pool) + .await?; + assert!(ahead_exists, "partitions exist days ahead of today"); + + let delivered_day = today - time::Duration::days(40); + let in_flight_day = today - time::Duration::days(39); + for day in [delivered_day, in_flight_day] { + sqlx::query("SELECT ensure_daily_partition('telegram_outbox', $1::date)") + .bind(day) + .execute(&pool) + .await?; + } + for (day, state, batch) in [ + (delivered_day, "delivered", "live-partition-delivered"), + (in_flight_day, "pending", "live-partition-pending"), + ] { + sqlx::query( + "INSERT INTO telegram_outbox (operation_id, batch_id, part_index, bot_id, ordering_key, \ + method_kind, payload_version, original_payload, payload, delivery_policy, state, created_at) \ + VALUES ($1, $1, 0, 1, 'live-partition', 'sendMessage', 1, '{}', '{}', 'create', $2, \ + ($3::date + time '12:00') AT TIME ZONE 'UTC')", + ) + .bind(batch) + .bind(state) + .bind(day) + .execute(&pool) + .await?; + } + + let dropped = super::drop_daily_partitions_before( + &pool, + "telegram_outbox", + today - time::Duration::days(30), + Some(super::TELEGRAM_OUTBOX_IN_FLIGHT), + ) + .await?; + + let format = time::macros::format_description!("[year][month][day]"); + assert!(dropped.contains(&format!( + "telegram_outbox_{}", + delivered_day.format(format)? + ))); + assert!(!dropped.contains(&format!( + "telegram_outbox_{}", + in_flight_day.format(format)? + ))); + let pending: i64 = sqlx::query_scalar( + "SELECT count(*) FROM telegram_outbox WHERE batch_id = 'live-partition-pending'", + ) + .fetch_one(&pool) + .await?; + assert_eq!(pending, 1, "an in-flight row keeps its day"); + sqlx::query("DELETE FROM telegram_outbox WHERE ordering_key = 'live-partition'") + .execute(&pool) + .await?; + super::drop_daily_partitions_before( + &pool, + "telegram_outbox", + today - time::Duration::days(30), + None, + ) + .await?; + Ok(()) + } + #[test] fn redis_connection_info_matches_go_cache_defaults() -> Result<(), redis::RedisError> { let config = RedisConfig { @@ -14532,6 +14679,10 @@ mod tests { payload: payload.as_bytes(), }) .await?; + sqlx::query("SELECT ensure_daily_partition('llm_request_events_raw', ($1 AT TIME ZONE 'UTC')::date)") + .bind(now) + .execute(&pool) + .await?; sqlx::query( "INSERT INTO llm_request_events (created_at, source, flow, chat_id, user_id, prompt_chars, prompt_messages, docs_chars, duration_ms, error, is_rollup) VALUES ($1, 'dialog', 'runtime-virtual-dialog', $2, $3, 0, 0, 0, 0, '', FALSE)", ) diff --git a/crates/openplotva-storage/src/telegram_outbox.rs b/crates/openplotva-storage/src/telegram_outbox.rs index d5df5ed1..7ec10904 100644 --- a/crates/openplotva-storage/src/telegram_outbox.rs +++ b/crates/openplotva-storage/src/telegram_outbox.rs @@ -25,30 +25,52 @@ const SQL_INSERT_BLOBS_SUFFIX: &str = " ON CONFLICT (sha256) DO UPDATE SET \ WHERE telegram_outbox_blobs.bytes = EXCLUDED.bytes \ RETURNING id, sha256"; -const SQL_INSERT_OPERATIONS_PREFIX: &str = "INSERT INTO telegram_outbox (\ +/// The outbox is partitioned by day, so operation ids cannot be a unique key: +/// enqueue takes this per-batch advisory lock, reuses parts that already exist +/// with identical immutable fields and inserts only the missing ones. Parts are +/// inserted in part order because claiming orders a chat's sends by id. +const SQL_LOCK_OUTBOX_BATCH: &str = + "SELECT pg_advisory_xact_lock(hashtextextended('telegram_outbox:' || $1, 0))"; +const SQL_ENQUEUE_OPERATIONS_PREFIX: &str = "WITH input (\ operation_id, batch_id, part_index, bot_id, chat_id, thread_id, ordering_key, \ causation_update_id, dialog_job_id, trigger_message_id, method_kind, payload_version, \ - original_payload, payload, blob_id, delivery_policy, protected, priority, available_at, expires_at)"; -const SQL_INSERT_OPERATIONS_SUFFIX: &str = " ON CONFLICT (operation_id) DO UPDATE SET \ - operation_id = EXCLUDED.operation_id \ - WHERE telegram_outbox.batch_id = EXCLUDED.batch_id \ - AND telegram_outbox.part_index = EXCLUDED.part_index \ - AND telegram_outbox.bot_id = EXCLUDED.bot_id \ - AND telegram_outbox.chat_id IS NOT DISTINCT FROM EXCLUDED.chat_id \ - AND telegram_outbox.thread_id IS NOT DISTINCT FROM EXCLUDED.thread_id \ - AND telegram_outbox.ordering_key = EXCLUDED.ordering_key \ - AND telegram_outbox.causation_update_id IS NOT DISTINCT FROM EXCLUDED.causation_update_id \ - AND telegram_outbox.dialog_job_id IS NOT DISTINCT FROM EXCLUDED.dialog_job_id \ - AND telegram_outbox.trigger_message_id IS NOT DISTINCT FROM EXCLUDED.trigger_message_id \ - AND telegram_outbox.method_kind = EXCLUDED.method_kind \ - AND telegram_outbox.payload_version = EXCLUDED.payload_version \ - AND telegram_outbox.original_payload = EXCLUDED.original_payload \ - AND telegram_outbox.blob_id IS NOT DISTINCT FROM EXCLUDED.blob_id \ - AND telegram_outbox.delivery_policy = EXCLUDED.delivery_policy \ - AND telegram_outbox.protected = EXCLUDED.protected \ - AND telegram_outbox.priority = EXCLUDED.priority \ - AND telegram_outbox.expires_at IS NOT DISTINCT FROM EXCLUDED.expires_at \ - RETURNING id, operation_id, part_index, state, (xmax = 0) AS inserted"; + original_payload, payload, blob_id, delivery_policy, protected, priority, available_at, expires_at) AS ("; +const SQL_ENQUEUE_OPERATIONS_SUFFIX: &str = "), existing AS ( \ + SELECT outbox.id, outbox.operation_id, outbox.part_index, outbox.state, \ + (outbox.batch_id = input.batch_id \ + AND outbox.part_index = input.part_index \ + AND outbox.bot_id = input.bot_id \ + AND outbox.chat_id IS NOT DISTINCT FROM input.chat_id \ + AND outbox.thread_id IS NOT DISTINCT FROM input.thread_id \ + AND outbox.ordering_key = input.ordering_key \ + AND outbox.causation_update_id IS NOT DISTINCT FROM input.causation_update_id \ + AND outbox.dialog_job_id IS NOT DISTINCT FROM input.dialog_job_id \ + AND outbox.trigger_message_id IS NOT DISTINCT FROM input.trigger_message_id \ + AND outbox.method_kind = input.method_kind \ + AND outbox.payload_version = input.payload_version \ + AND outbox.original_payload = input.original_payload \ + AND outbox.blob_id IS NOT DISTINCT FROM input.blob_id \ + AND outbox.delivery_policy = input.delivery_policy \ + AND outbox.protected = input.protected \ + AND outbox.priority = input.priority \ + AND outbox.expires_at IS NOT DISTINCT FROM input.expires_at) AS same \ + FROM input JOIN telegram_outbox AS outbox ON outbox.operation_id = input.operation_id \ +), inserted AS ( \ + INSERT INTO telegram_outbox (\ + operation_id, batch_id, part_index, bot_id, chat_id, thread_id, ordering_key, \ + causation_update_id, dialog_job_id, trigger_message_id, method_kind, payload_version, \ + original_payload, payload, blob_id, delivery_policy, protected, priority, available_at, expires_at) \ + SELECT operation_id, batch_id, part_index, bot_id, chat_id, thread_id, ordering_key, \ + causation_update_id, dialog_job_id, trigger_message_id, method_kind, payload_version, \ + original_payload, payload, blob_id, delivery_policy, protected, priority, available_at, expires_at \ + FROM input \ + WHERE NOT EXISTS (SELECT 1 FROM existing WHERE existing.operation_id = input.operation_id) \ + ORDER BY input.part_index \ + RETURNING id, operation_id, part_index, state \ +) \ +SELECT id, operation_id, part_index, state, FALSE AS inserted FROM existing WHERE same \ +UNION ALL \ +SELECT id, operation_id, part_index, state, TRUE AS inserted FROM inserted"; const SQL_CLAIM_OPERATIONS: &str = r#" WITH candidates AS ( @@ -405,6 +427,10 @@ pub(crate) async fn enqueue_telegram_outbox_batch_in_transaction( } let parts = prepare_parts(batch, &blob_ids)?; + sqlx::query(SQL_LOCK_OUTBOX_BATCH) + .bind(&batch.batch_id) + .execute(&mut **tx) + .await?; let mut builder = outbox_operation_insert_builder(batch, &parts); let rows = builder.build().fetch_all(&mut **tx).await?; if rows.len() != parts.len() { @@ -508,8 +534,7 @@ fn outbox_operation_insert_builder( batch: &TelegramOutboxBatchInput, parts: &[PreparedOutboxPart], ) -> QueryBuilder { - let mut builder = QueryBuilder::new(SQL_INSERT_OPERATIONS_PREFIX); - builder.push(" "); + let mut builder = QueryBuilder::new(SQL_ENQUEUE_OPERATIONS_PREFIX); builder.push_values(parts.iter(), |mut row, prepared| { let part = &prepared.part; row.push_bind(prepared.operation_id.clone()) @@ -533,7 +558,7 @@ fn outbox_operation_insert_builder( .push_bind(part.available_at) .push_bind(part.expires_at); }); - builder.push(SQL_INSERT_OPERATIONS_SUFFIX); + builder.push(SQL_ENQUEUE_OPERATIONS_SUFFIX); builder } @@ -1644,7 +1669,9 @@ mod tests { 3 * OUTBOX_OPERATION_BINDS_PER_ROW ); assert!(operation_sql.contains("original_payload")); - assert!(operation_sql.contains("ON CONFLICT (operation_id)")); + assert!(operation_sql.contains( + "JOIN telegram_outbox AS outbox ON outbox.operation_id = input.operation_id" + )); Ok(()) } @@ -1737,6 +1764,82 @@ mod tests { Ok(()) } + #[tokio::test] + async fn live_outbox_enqueue_of_a_batch_waits_for_an_uncommitted_enqueue_of_it() + -> Result<(), Box> { + let Ok(dsn) = env::var("OPENPLOTVA_TEST_POSTGRES_DSN") else { + return Ok(()); + }; + let pool = PgPoolOptions::new() + .max_connections(4) + .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-concurrent-test:{suffix}"); + // Blob upserts would serialize the enqueues on their own; without blobs + // only the batch lock does. + let mut input = batch(&batch_id, 3); + for part in &mut input.parts { + part.blob = None; + } + + let mut first = pool.begin().await?; + let queued_first = enqueue_telegram_outbox_batch_in_transaction(&mut first, &input).await?; + let second = tokio::spawn({ + let pool = pool.clone(); + let input = input.clone(); + async move { + let mut tx = pool.begin().await?; + let queued = enqueue_telegram_outbox_batch_in_transaction(&mut tx, &input).await?; + tx.commit().await?; + Ok::<_, StorageError>(queued) + } + }); + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + assert!( + !second.is_finished(), + "a second enqueue of the batch must wait for the first to commit" + ); + first.commit().await?; + let queued_second = second.await??; + let rows: i64 = + sqlx::query_scalar("SELECT count(*) FROM telegram_outbox WHERE batch_id = $1") + .bind(&batch_id) + .fetch_one(&pool) + .await?; + + assert_eq!(rows, 3); + assert!(queued_first.parts.iter().all(|part| part.inserted)); + assert!(queued_second.parts.iter().all(|part| !part.inserted)); + let first_ids = queued_first + .parts + .iter() + .map(|part| part.id) + .collect::>(); + assert_eq!( + first_ids, + queued_second + .parts + .iter() + .map(|part| part.id) + .collect::>() + ); + assert!( + first_ids.windows(2).all(|pair| pair[0] < pair[1]), + "ids follow part order" + ); + + sqlx::query("DELETE FROM telegram_outbox WHERE batch_id = $1") + .bind(&batch_id) + .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 { @@ -1942,8 +2045,10 @@ mod tests { } .fetch_all(&mut *connection) .await?; + // Partitions name their copies of the parent index after themselves, + // so check that no partition falls back to a sequential scan. assert!( - plan.join("\n").contains(index), + !plan.join("\n").contains("Seq Scan"), "{index} must support the recovery scan: {plan:?}" ); } diff --git a/docs/CODEBASE_MAP.md b/docs/CODEBASE_MAP.md index 57f29b16..50cde614 100644 --- a/docs/CODEBASE_MAP.md +++ b/docs/CODEBASE_MAP.md @@ -667,6 +667,7 @@ Postgres 17 + pgvector (`vector(512)` for memory/shield, `vector(1024)` legacy) - **Dialog history**: `chat_history_entries` (RANGE partitioned by day), `chat_history_resets`, `chat_history_summaries` + `_summary_sources` + `_events` (cascade). - **Safety**: `shield_documents` (seeded with 6 crisis docs), `whitecircle_checks` (LZ4 TOAST). - **Task queue (v2)**: `taskman_jobs` (JSONB record + typed columns), `taskman_job_history`. (Legacy `job_queue` dropped in mig 19/140.) +- **Daily partitions** (mig 204–206): `llm_request_events` is LIST(`is_rollup`) with rollups in `llm_request_events_rollup` and raw events in daily `llm_request_events_raw_YYYYMMDD`; `telegram_outbox` and `telegram_outbox_attempts` are daily RANGE partitions. Retention drops whole days (`drop_daily_partitions_before`); an hourly worker keeps 7 days ahead. No unique keys on these parents: outbox enqueue serializes each batch with an advisory lock and reuses identical parts. - **LLM telemetry & routing**: `llm_request_events` (raw + rollup dual-purpose, `run_id`/`run_seq` correlation, short-lived raw bodies scrubbed), `telemetry_rollups`, `llm_providers`/`provider_models`/`llm_capacity_pools`/`workflows`/`workflow_assignments`/`workflow_triggers`/`llm_routing_events`. - **Dialog lifecycle**: `runtime_virtual_dialogs`, `dialog_turn_outcomes`, `dialog_delivery_obligations`. - **Files/media cache**: `telegram_files` (vision and ASR pipeline state, including ASR fallback diagnostics). diff --git a/migrations/204_daily_partition_helpers.down.sql b/migrations/204_daily_partition_helpers.down.sql new file mode 100644 index 00000000..b0404bd5 --- /dev/null +++ b/migrations/204_daily_partition_helpers.down.sql @@ -0,0 +1,2 @@ +DROP FUNCTION IF EXISTS drop_daily_partitions_before(text, date, text); +DROP FUNCTION IF EXISTS ensure_daily_partition(text, date); diff --git a/migrations/204_daily_partition_helpers.up.sql b/migrations/204_daily_partition_helpers.up.sql new file mode 100644 index 00000000..2c8ebce2 --- /dev/null +++ b/migrations/204_daily_partition_helpers.up.sql @@ -0,0 +1,58 @@ +-- Daily RANGE partitions on a timestamptz key, named _YYYYMMDD and +-- bounded by UTC midnights. Retention drops whole partitions instead of deleting +-- rows; `keep_when` names rows that must not be dropped yet. +CREATE OR REPLACE FUNCTION ensure_daily_partition(parent text, partition_day date) +RETURNS void AS $$ +BEGIN + EXECUTE format( + 'CREATE TABLE IF NOT EXISTS %I PARTITION OF %I FOR VALUES FROM (%L) TO (%L)', + parent || '_' || to_char(partition_day, 'YYYYMMDD'), + parent, + partition_day::timestamp AT TIME ZONE 'UTC', + (partition_day + 1)::timestamp AT TIME ZONE 'UTC' + ); +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION drop_daily_partitions_before( + parent text, + cutoff_day date, + keep_when text DEFAULT NULL +) +RETURNS text[] AS $$ +DECLARE + child record; + partition_day date; + busy boolean; + dropped text[] := '{}'; +BEGIN + FOR child IN + SELECT partition.relname + FROM pg_inherits + JOIN pg_class parent_table ON parent_table.oid = pg_inherits.inhparent + JOIN pg_class partition ON partition.oid = pg_inherits.inhrelid + JOIN pg_namespace parent_ns ON parent_ns.oid = parent_table.relnamespace + WHERE parent_table.relname = parent + AND parent_ns.nspname = current_schema() + ORDER BY partition.relname + LOOP + IF child.relname !~ ('^' || parent || '_[0-9]{8}$') THEN + CONTINUE; + END IF; + partition_day := to_date(right(child.relname, 8), 'YYYYMMDD'); + IF partition_day >= cutoff_day THEN + CONTINUE; + END IF; + IF keep_when IS NOT NULL THEN + EXECUTE format('SELECT EXISTS (SELECT 1 FROM %I WHERE %s)', child.relname, keep_when) + INTO busy; + IF busy THEN + CONTINUE; + END IF; + END IF; + EXECUTE format('DROP TABLE IF EXISTS %I', child.relname); + dropped := array_append(dropped, child.relname::text); + END LOOP; + RETURN dropped; +END; +$$ LANGUAGE plpgsql; diff --git a/migrations/205_partition_llm_request_events.down.sql b/migrations/205_partition_llm_request_events.down.sql new file mode 100644 index 00000000..edc3ffc1 --- /dev/null +++ b/migrations/205_partition_llm_request_events.down.sql @@ -0,0 +1,44 @@ +-- Back to one plain table. Raw partitions already dropped by retention are gone. +ALTER TABLE llm_request_events RENAME TO llm_request_events_partitioned; + +CREATE TABLE llm_request_events ( + LIKE llm_request_events_partitioned + INCLUDING DEFAULTS INCLUDING CONSTRAINTS INCLUDING STORAGE INCLUDING COMPRESSION +); + +INSERT INTO llm_request_events +SELECT * FROM llm_request_events_partitioned; + +ALTER SEQUENCE llm_request_events_id_seq OWNED BY llm_request_events.id; + +DROP TABLE llm_request_events_partitioned; + +ALTER TABLE llm_request_events ADD CONSTRAINT llm_request_events_pkey PRIMARY KEY (id); +CREATE INDEX idx_llm_request_events_created_at ON llm_request_events (created_at DESC); +CREATE INDEX idx_llm_request_events_chat_created_at ON llm_request_events (chat_id, created_at DESC); +CREATE INDEX idx_llm_request_events_model_bucket ON llm_request_events (model, created_at DESC); +CREATE INDEX idx_llm_request_events_provider_created_at ON llm_request_events (provider, created_at DESC); +CREATE INDEX idx_llm_request_events_source_created_at ON llm_request_events (source, created_at DESC); +CREATE INDEX idx_llm_request_events_raw_cleanup ON llm_request_events (created_at) + WHERE is_rollup = FALSE; +CREATE INDEX idx_llm_request_events_raw_scrub ON llm_request_events (created_at) + WHERE raw_request IS NOT NULL OR raw_response IS NOT NULL; +CREATE INDEX idx_llm_request_events_run_id ON llm_request_events (run_id) + WHERE run_id IS NOT NULL; +CREATE UNIQUE INDEX idx_llm_request_events_rollup_key + ON llm_request_events ( + rollup_granularity, + bucket_start, + COALESCE(source, ''), + COALESCE(provider, ''), + COALESCE(model, ''), + COALESCE(chat_id, 0), + COALESCE(max_tokens, -1), + COALESCE(temperature, -999999.0), + COALESCE(top_p, -999999.0), + COALESCE(top_k, -1), + COALESCE(candidate_count, -1), + COALESCE(tool_mode, ''), + COALESCE(response_format, '') + ) + WHERE is_rollup = TRUE; diff --git a/migrations/205_partition_llm_request_events.up.sql b/migrations/205_partition_llm_request_events.up.sql new file mode 100644 index 00000000..157cd22e --- /dev/null +++ b/migrations/205_partition_llm_request_events.up.sql @@ -0,0 +1,67 @@ +-- Split llm_request_events by kind: long-lived rollups in one partition, raw +-- events in daily partitions that retention drops once they are rolled up. +-- Unique keys must contain the partition key on a partitioned table, so the +-- primary key on id becomes a plain index; ids still come from the sequence. +-- The rollup key stays unique on the rollup partition, where the rollup's +-- ON CONFLICT DO NOTHING checks it. +ALTER TABLE llm_request_events RENAME TO llm_request_events_legacy; + +CREATE TABLE llm_request_events ( + LIKE llm_request_events_legacy + INCLUDING DEFAULTS INCLUDING CONSTRAINTS INCLUDING STORAGE INCLUDING COMPRESSION +) PARTITION BY LIST (is_rollup); + +CREATE TABLE llm_request_events_rollup + PARTITION OF llm_request_events FOR VALUES IN (TRUE); + +CREATE TABLE llm_request_events_raw + PARTITION OF llm_request_events FOR VALUES IN (FALSE) + PARTITION BY RANGE (created_at); + +SELECT ensure_daily_partition('llm_request_events_raw', day::date) +FROM generate_series( + LEAST( + (SELECT min((created_at AT TIME ZONE 'UTC')::date) + FROM llm_request_events_legacy + WHERE NOT is_rollup), + (now() AT TIME ZONE 'UTC')::date + ), + (now() AT TIME ZONE 'UTC')::date + 7, + interval '1 day' +) AS day; + +INSERT INTO llm_request_events +SELECT * FROM llm_request_events_legacy; + +ALTER SEQUENCE llm_request_events_id_seq OWNED BY llm_request_events.id; + +DROP TABLE llm_request_events_legacy; + +CREATE INDEX idx_llm_request_events_id ON llm_request_events (id); +CREATE INDEX idx_llm_request_events_created_at ON llm_request_events (created_at DESC); +CREATE INDEX idx_llm_request_events_chat_created_at ON llm_request_events (chat_id, created_at DESC); +CREATE INDEX idx_llm_request_events_model_bucket ON llm_request_events (model, created_at DESC); +CREATE INDEX idx_llm_request_events_provider_created_at ON llm_request_events (provider, created_at DESC); +CREATE INDEX idx_llm_request_events_source_created_at ON llm_request_events (source, created_at DESC); +CREATE INDEX idx_llm_request_events_raw_scrub ON llm_request_events (created_at) + WHERE raw_request IS NOT NULL OR raw_response IS NOT NULL; +CREATE INDEX idx_llm_request_events_run_id ON llm_request_events (run_id) + WHERE run_id IS NOT NULL; + +CREATE UNIQUE INDEX idx_llm_request_events_rollup_key + ON llm_request_events_rollup ( + rollup_granularity, + bucket_start, + COALESCE(source, ''), + COALESCE(provider, ''), + COALESCE(model, ''), + COALESCE(chat_id, 0), + COALESCE(max_tokens, -1), + COALESCE(temperature, -999999.0), + COALESCE(top_p, -999999.0), + COALESCE(top_k, -1), + COALESCE(candidate_count, -1), + COALESCE(tool_mode, ''), + COALESCE(response_format, '') + ) + WHERE is_rollup = TRUE; diff --git a/migrations/206_partition_telegram_outbox.down.sql b/migrations/206_partition_telegram_outbox.down.sql new file mode 100644 index 00000000..1c472387 --- /dev/null +++ b/migrations/206_partition_telegram_outbox.down.sql @@ -0,0 +1,65 @@ +-- Back to plain tables with the original keys. History dropped by the up +-- migration and by retention is not restored. +ALTER TABLE telegram_outbox_attempts RENAME TO telegram_outbox_attempts_partitioned; +ALTER TABLE telegram_outbox RENAME TO telegram_outbox_partitioned; + +CREATE TABLE telegram_outbox ( + LIKE telegram_outbox_partitioned + INCLUDING DEFAULTS INCLUDING CONSTRAINTS INCLUDING STORAGE INCLUDING COMPRESSION +); +CREATE TABLE telegram_outbox_attempts ( + LIKE telegram_outbox_attempts_partitioned + INCLUDING DEFAULTS INCLUDING CONSTRAINTS INCLUDING STORAGE INCLUDING COMPRESSION +); + +INSERT INTO telegram_outbox SELECT * FROM telegram_outbox_partitioned; +INSERT INTO telegram_outbox_attempts +SELECT attempt.* +FROM telegram_outbox_attempts_partitioned AS attempt +WHERE EXISTS (SELECT 1 FROM telegram_outbox AS outbox WHERE outbox.id = attempt.outbox_id); + +ALTER SEQUENCE telegram_outbox_id_seq OWNED BY telegram_outbox.id; +ALTER SEQUENCE telegram_outbox_attempts_id_seq OWNED BY telegram_outbox_attempts.id; + +DROP TABLE telegram_outbox_attempts_partitioned; +DROP TABLE telegram_outbox_partitioned; + +ALTER TABLE telegram_outbox + ADD CONSTRAINT telegram_outbox_pkey PRIMARY KEY (id), + ADD CONSTRAINT telegram_outbox_operation_id_key UNIQUE (operation_id), + ADD CONSTRAINT telegram_outbox_batch_id_part_index_key UNIQUE (batch_id, part_index), + ADD CONSTRAINT telegram_outbox_blob_id_fkey + FOREIGN KEY (blob_id) REFERENCES telegram_outbox_blobs(id); +ALTER TABLE telegram_outbox_attempts + ADD CONSTRAINT telegram_outbox_attempts_pkey PRIMARY KEY (id), + ADD CONSTRAINT telegram_outbox_attempts_outbox_id_attempt_key UNIQUE (outbox_id, attempt), + ADD CONSTRAINT telegram_outbox_attempts_outbox_id_fkey + FOREIGN KEY (outbox_id) REFERENCES telegram_outbox(id) ON DELETE CASCADE; + +CREATE INDEX telegram_outbox_claim_idx + ON telegram_outbox (state, available_at, priority DESC, id) + WHERE state IN ('pending', 'retry_wait', 'leased'); +CREATE INDEX telegram_outbox_ordering_idx + ON telegram_outbox (bot_id, ordering_key, batch_id, part_index) + WHERE state IN ('pending', 'leased', 'retry_wait'); +CREATE INDEX telegram_outbox_lease_idx + ON telegram_outbox (leased_until) + WHERE state = 'leased'; +CREATE INDEX telegram_outbox_terminal_idx + ON telegram_outbox (confirmed_at, id) + WHERE state IN ('delivered', 'dead_letter', 'expired', 'cancelled'); +CREATE INDEX telegram_outbox_protected_terminal_idx + ON telegram_outbox (state, updated_at, id) + WHERE protected AND state IN ('ambiguous', 'dead_letter'); +CREATE INDEX telegram_outbox_dialog_job_idx + ON telegram_outbox (dialog_job_id, state, id) + WHERE dialog_job_id IS NOT NULL; +CREATE INDEX telegram_outbox_pending_history_idx + ON telegram_outbox (confirmed_at, id) + WHERE state = 'delivered' AND last_error_class = 'history_pending'; +CREATE INDEX telegram_outbox_ambiguous_create_idx + ON telegram_outbox (id) + WHERE state = 'ambiguous' AND delivery_policy = 'create' + AND chat_id IS NOT NULL AND trigger_message_id IS NOT NULL; +CREATE INDEX telegram_outbox_attempts_outbox_idx + ON telegram_outbox_attempts (outbox_id, attempt DESC); diff --git a/migrations/206_partition_telegram_outbox.up.sql b/migrations/206_partition_telegram_outbox.up.sql new file mode 100644 index 00000000..798f825a --- /dev/null +++ b/migrations/206_partition_telegram_outbox.up.sql @@ -0,0 +1,104 @@ +-- Daily partitions for the Telegram outbox and its attempts, so retention drops +-- whole days instead of deleting rows. Unique keys must contain the partition +-- key, so operation_id and (batch_id, part_index) become plain indexes: enqueue +-- serializes each batch with an advisory lock and reuses existing parts, and +-- attempts lose the foreign key to the outbox (their own partitions age out on +-- the same schedule). +-- +-- One-way for data: only the last 7 days of history and every row still in +-- flight are copied; older delivered history is dropped with the old tables. +ALTER TABLE telegram_outbox_attempts RENAME TO telegram_outbox_attempts_legacy; +ALTER TABLE telegram_outbox RENAME TO telegram_outbox_legacy; + +CREATE TABLE telegram_outbox ( + LIKE telegram_outbox_legacy + INCLUDING DEFAULTS INCLUDING CONSTRAINTS INCLUDING STORAGE INCLUDING COMPRESSION +) PARTITION BY RANGE (created_at); + +CREATE TABLE telegram_outbox_attempts ( + LIKE telegram_outbox_attempts_legacy + INCLUDING DEFAULTS INCLUDING CONSTRAINTS INCLUDING STORAGE INCLUDING COMPRESSION +) PARTITION BY RANGE (claimed_at); + +CREATE TEMP TABLE kept_telegram_outbox ON COMMIT DROP AS +SELECT id +FROM telegram_outbox_legacy +WHERE created_at >= ((now() AT TIME ZONE 'UTC')::date - 7)::timestamp AT TIME ZONE 'UTC' + OR state IN ('pending', 'leased', 'retry_wait'); + +SELECT ensure_daily_partition('telegram_outbox', day::date) +FROM generate_series( + LEAST( + (SELECT min((outbox.created_at AT TIME ZONE 'UTC')::date) + FROM telegram_outbox_legacy AS outbox + JOIN kept_telegram_outbox AS kept ON kept.id = outbox.id), + (now() AT TIME ZONE 'UTC')::date + ), + (now() AT TIME ZONE 'UTC')::date + 7, + interval '1 day' +) AS day; + +SELECT ensure_daily_partition('telegram_outbox_attempts', day::date) +FROM generate_series( + LEAST( + (SELECT min((attempt.claimed_at AT TIME ZONE 'UTC')::date) + FROM telegram_outbox_attempts_legacy AS attempt + JOIN kept_telegram_outbox AS kept ON kept.id = attempt.outbox_id), + (now() AT TIME ZONE 'UTC')::date + ), + (now() AT TIME ZONE 'UTC')::date + 7, + interval '1 day' +) AS day; + +INSERT INTO telegram_outbox +SELECT outbox.* +FROM telegram_outbox_legacy AS outbox +JOIN kept_telegram_outbox AS kept ON kept.id = outbox.id; + +INSERT INTO telegram_outbox_attempts +SELECT attempt.* +FROM telegram_outbox_attempts_legacy AS attempt +JOIN kept_telegram_outbox AS kept ON kept.id = attempt.outbox_id; + +ALTER SEQUENCE telegram_outbox_id_seq OWNED BY telegram_outbox.id; +ALTER SEQUENCE telegram_outbox_attempts_id_seq OWNED BY telegram_outbox_attempts.id; + +DROP TABLE telegram_outbox_attempts_legacy; +DROP TABLE telegram_outbox_legacy; + +ALTER TABLE telegram_outbox + ADD CONSTRAINT telegram_outbox_blob_id_fkey + FOREIGN KEY (blob_id) REFERENCES telegram_outbox_blobs(id); + +CREATE INDEX telegram_outbox_id_idx ON telegram_outbox (id); +CREATE INDEX telegram_outbox_operation_idx ON telegram_outbox (operation_id); +CREATE INDEX telegram_outbox_batch_part_idx ON telegram_outbox (batch_id, part_index); +CREATE INDEX telegram_outbox_claim_idx + ON telegram_outbox (state, available_at, priority DESC, id) + WHERE state IN ('pending', 'retry_wait', 'leased'); +CREATE INDEX telegram_outbox_ordering_idx + ON telegram_outbox (bot_id, ordering_key, batch_id, part_index) + WHERE state IN ('pending', 'leased', 'retry_wait'); +CREATE INDEX telegram_outbox_lease_idx + ON telegram_outbox (leased_until) + WHERE state = 'leased'; +CREATE INDEX telegram_outbox_terminal_idx + ON telegram_outbox (confirmed_at, id) + WHERE state IN ('delivered', 'dead_letter', 'expired', 'cancelled'); +CREATE INDEX telegram_outbox_protected_terminal_idx + ON telegram_outbox (state, updated_at, id) + WHERE protected AND state IN ('ambiguous', 'dead_letter'); +CREATE INDEX telegram_outbox_dialog_job_idx + ON telegram_outbox (dialog_job_id, state, id) + WHERE dialog_job_id IS NOT NULL; +CREATE INDEX telegram_outbox_pending_history_idx + ON telegram_outbox (confirmed_at, id) + WHERE state = 'delivered' AND last_error_class = 'history_pending'; +CREATE INDEX telegram_outbox_ambiguous_create_idx + ON telegram_outbox (id) + WHERE state = 'ambiguous' AND delivery_policy = 'create' + AND chat_id IS NOT NULL AND trigger_message_id IS NOT NULL; + +CREATE INDEX telegram_outbox_attempts_id_idx ON telegram_outbox_attempts (id); +CREATE INDEX telegram_outbox_attempts_outbox_idx + ON telegram_outbox_attempts (outbox_id, attempt DESC);