Skip to content

Partition the Telegram outbox and LLM request events by day #256

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
iamwavecut merged 1 commit into from
Oct 13, 2026
Merged

Partition the Telegram outbox and LLM request events by day #256

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .env.example
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
30 changes: 28 additions & 2 deletions crates/openplotva-app/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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
Expand Down Expand Up @@ -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();
Expand Down
117 changes: 55 additions & 62 deletions crates/openplotva-app/src/runtime_llm.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -300,6 +290,7 @@ pub struct RuntimeLlmRequestEventCleanupReport {
pub enabled: bool,
pub ticks: u64,
pub deleted: u64,
pub dropped_partitions: u64,
pub errors: u64,
}

Expand Down Expand Up @@ -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<u64, sqlx::Error> {
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<Stop>(
pool: PgPool,
first_pass_delay: Duration,
interval: Duration,
retention_days: i32,
batch_size: i64,
stop: Stop,
) -> RuntimeLlmRequestEventCleanupReport
where
Stop: std::future::Future<Output = ()>,
{
let mut report = RuntimeLlmRequestEventCleanupReport {
enabled: retention_days > 0 && batch_size > 0,
enabled: retention_days > 0,
..RuntimeLlmRequestEventCleanupReport::default()
};
if !report.enabled {
Expand All @@ -957,42 +936,39 @@ where
() = &mut stop => return report,
() = tokio::time::sleep(first_pass_delay) => {}
}
'passes: loop {
loop {
let archive_result = tokio::select! {
biased;
() = &mut stop => break,
result = archive_old_analytics(&pool, retention_days) => result,
};
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;
Expand Down Expand Up @@ -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() {
Expand All @@ -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");
Expand All @@ -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) \
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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",
)
Expand All @@ -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) \
Expand All @@ -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'",
Expand All @@ -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;
}

Expand All @@ -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);
}

Expand Down
Loading
Loading