Skip to content

fix(retention): batch every delete — the sweep had never once succeeded #177

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
thejefflarson merged 1 commit into from
Aug 25, 2026
Merged

fix(retention): batch every delete — the sweep had never once succeeded #177

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
49 changes: 34 additions & 15 deletions server/src/online_ddl.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,23 +57,42 @@ pub struct OnlineIndex {
/// `include_columns_drifted`, which makes the lane detect and rebuild that
/// narrowing against the wide index migration 0017 built, rather than treating
/// the existing valid-and-same-name index as done forever.
pub const DESIRED_INDEXES: &[OnlineIndex] = &[OnlineIndex {
name: "metric_series_rollups_name_bucket_covering_idx",
table: "metric_series_rollups",
include_cols: &[
"service",
"kind",
"unit",
"is_monotonic",
"count",
"sum",
"avg",
"max",
],
create_sql: "CREATE INDEX CONCURRENTLY metric_series_rollups_name_bucket_covering_idx \
pub const DESIRED_INDEXES: &[OnlineIndex] = &[
OnlineIndex {
name: "metric_series_rollups_name_bucket_covering_idx",
table: "metric_series_rollups",
include_cols: &[
"service",
"kind",
"unit",
"is_monotonic",
"count",
"sum",
"avg",
"max",
],
create_sql: "CREATE INDEX CONCURRENTLY metric_series_rollups_name_bucket_covering_idx \
ON metric_series_rollups (name, bucket) \
INCLUDE (service, kind, unit, is_monotonic, count, sum, avg, max)",
}];
},
OnlineIndex {
// Freshness probe support. `selfmon` runs
// SELECT extract(epoch FROM now() - max(bucket)) FROM metric_series_rollups
// once a minute. Both other indexes on this table lead with `name`, so a bare
// `max(bucket)` with no `name` predicate can use neither and Postgres falls
// back to a full scan. Measured 2026-08-25 in production: a Parallel Seq Scan
// over 14.8M rows, 966k buffer reads, ~2.9s EVERY MINUTE. With this index the
// same query is an Index Only Scan reading one row — 0.142ms.
//
// `bucket DESC` so the backward scan `max()` wants is the index's natural
// order. No INCLUDE: the query selects nothing but `bucket`.
name: "metric_series_rollups_bucket_idx",
table: "metric_series_rollups",
include_cols: &[],
create_sql: "CREATE INDEX CONCURRENTLY metric_series_rollups_bucket_idx \
ON metric_series_rollups (bucket DESC)",
},
];

/// Session-level advisory-lock key that gates a whole lane run, so exactly one
/// replica builds during a rollout and the others skip cleanly. Arbitrary but
Expand Down
141 changes: 104 additions & 37 deletions server/src/retention.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,9 +9,14 @@
//! `WATCHER_RETENTION_METRICS_DAYS`. A table with no override falls back to the
//! existing global `WATCHER_RETENTION_DAYS` — an all-omitted config is exactly
//! today's single window, so this is a no-op for anyone who doesn't set the new
//! vars. This is deliberately per-*table*, not per-service: a per-service delete
//! over these tables would need `ctid`-batching like `prune_raw_metrics` below to
//! avoid the statement-timeout failure mode; that's a separate follow-up.
//! vars. This is deliberately per-*table*, not per-service.
//!
//! EVERY delete here is `ctid`-batched via [`prune_batched`]. That used to be
//! true only of `prune_raw_metrics`, on the assumption that a whole-table delete
//! was small enough to land inside the statement timeout. It is not: once a table
//! accumulates a backlog, the single DELETE times out, rolls back completely, and
//! the backlog it failed to clear makes the next attempt slower — a stall that
//! never recovers on its own.

use sqlx::PgPool;
use std::time::Duration;
Expand Down Expand Up @@ -56,6 +61,9 @@ pub async fn prune_once(
windows: Windows,
) -> anyhow::Result<u64> {
let mut total = 0;
// Tables that failed this sweep. Collected rather than propagated with `?`
// so ONE bad table cannot starve the ones after it — see below.
let mut failed: Vec<&str> = Vec::new();
// History tables age out on their own window (falling back to `days`).
for (table, col, override_days) in [
("spans", "start_time", windows.spans_days),
Expand All @@ -66,33 +74,111 @@ pub async fn prune_once(
if table_days <= 0 {
continue;
}
let sql = format!("DELETE FROM {table} WHERE {col} < now() - make_interval(days => $1)");
// AssertSqlSafe: sqlx 0.9 requires dynamic SQL be audited; table/col come
// only from the hardcoded list above, so there's no injection surface.
let r = sqlx::query(sqlx::AssertSqlSafe(sql))
.bind(table_days)
.execute(pool)
.await?;
if r.rows_affected() > 0 {
tracing::info!("retention: pruned {} rows from {table}", r.rows_affected());
total += r.rows_affected();
match prune_batched(pool, table, col, "days", table_days, HISTORY_PRUNE_BATCH).await {
Ok(n) => {
if n > 0 {
tracing::info!("retention: pruned {n} rows from {table}");
total += n;
}
}
// Keep going. Previously this was `?`, so the FIRST failing table
// aborted the sweep and every table after it in this list was never
// pruned at all. Observed 2026-08-25: the unbatched `logs` delete
// timed out every hour, and because `metric_series_rollups` comes
// after it, that table was never swept once and reached 20 GB.
Err(e) => {
tracing::warn!("retention: {table} sweep failed: {e}");
failed.push(table);
}
}
}
// Raw points: short hours-window cap (rollups hold the history).
if raw_hours > 0 {
let pruned = prune_raw_metrics(pool, raw_hours, RAW_PRUNE_BATCH).await?;
if pruned > 0 {
tracing::info!("retention: pruned {pruned} raw metric rows");
total += pruned;
match prune_raw_metrics(pool, raw_hours, RAW_PRUNE_BATCH).await {
Ok(pruned) => {
if pruned > 0 {
tracing::info!("retention: pruned {pruned} raw metric rows");
total += pruned;
}
}
Err(e) => {
tracing::warn!("retention: metrics sweep failed: {e}");
failed.push("metrics");
}
}
}
// Only a sweep where EVERY table succeeded counts. A partial sweep leaves
// some table growing, which is precisely the condition /healthz must keep
// reporting as stalled.
if !failed.is_empty() {
anyhow::bail!("retention incomplete; failed tables: {}", failed.join(", "));
}
// Record the successful sweep so self-telemetry can surface its recency and
// /healthz can flag a stall (a silent retention stall is exactly what let the
// metrics table grow to tens of GB un-paged).
crate::selfmon::record_retention_success(total);
Ok(total)
}

/// Rows deleted per batch for the history tables.
///
/// Deliberately smaller than [`RAW_PRUNE_BATCH`]: these tables carry far more
/// indexes than raw `metrics` — `logs` alone has five, including a GIN index on
/// `attributes` — and every deleted row must be removed from each one, so an
/// equal-sized batch costs several times as much.
const HISTORY_PRUNE_BATCH: i64 = 10_000;

/// Delete rows older than `amount` `unit`s from `table`, in batches of `batch`.
///
/// WHY BATCHING IS NOT OPTIONAL HERE. A single
/// `DELETE FROM <table> WHERE <col> < cutoff` over a large backlog exceeds the
/// connection's `statement_timeout`, and a cancelled DELETE **rolls back
/// entirely** — so the sweep deletes nothing, the backlog grows, and the next
/// sweep is slower still. It never recovers on its own.
///
/// That is not hypothetical: on 2026-08-25 `logs` held 11.5M rows past a 7-day
/// window, the hourly sweep hit its 60s timeout with `rows_affected=0` every
/// single time, and retention had NEVER completed a sweep in the life of the
/// process. Batching by `ctid` keeps each statement small enough to commit, so a
/// backlog drains across successive statements and each one is progress that
/// survives. In steady state the first batch is already short and the loop exits
/// after one pass.
/// `pub` so tests can drive it with a tiny `batch` and prove a backlog really
/// drains across statements — the property that was silently absent here.
pub async fn prune_batched(
pool: &PgPool,
table: &str,
col: &str,
unit: &str,
amount: i32,
batch: i64,
) -> anyhow::Result<u64> {
let mut pruned = 0u64;
loop {
let sql = format!(
"DELETE FROM {table} WHERE ctid IN (
SELECT ctid FROM {table}
WHERE {col} < now() - make_interval({unit} => $1)
LIMIT $2)"
);
// AssertSqlSafe: sqlx 0.9 requires dynamic SQL be audited. `table`, `col`
// and `unit` come only from hardcoded call sites in this module, never
// from config or request data, so there is no injection surface.
let r = sqlx::query(sqlx::AssertSqlSafe(sql))
.bind(amount)
.bind(batch)
.execute(pool)
.await?;
let n = r.rows_affected();
pruned += n;
// A short batch means nothing older than the cutoff remains.
if (n as i64) < batch {
break;
}
}
Ok(pruned)
}

/// Rows deleted per raw-metrics batch. Bounded so a large backlog drains across
/// many small statements instead of one huge DELETE.
const RAW_PRUNE_BATCH: i64 = 50_000;
Expand All @@ -107,24 +193,5 @@ const RAW_PRUNE_BATCH: i64 = 50_000;
/// successive iterations; in steady state the first batch is already partial and
/// the loop exits after one pass. Returns the total rows deleted.
pub async fn prune_raw_metrics(pool: &PgPool, raw_hours: i32, batch: i64) -> anyhow::Result<u64> {
let mut pruned = 0u64;
loop {
let r = sqlx::query(
"DELETE FROM metrics WHERE ctid IN (
SELECT ctid FROM metrics
WHERE time < now() - make_interval(hours => $1)
LIMIT $2)",
)
.bind(raw_hours)
.bind(batch)
.execute(pool)
.await?;
let n = r.rows_affected();
pruned += n;
// A short batch means no rows older than the cutoff remain.
if (n as i64) < batch {
break;
}
}
Ok(pruned)
prune_batched(pool, "metrics", "time", "hours", raw_hours, batch).await
}
91 changes: 91 additions & 0 deletions server/tests/smoke.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2291,6 +2291,97 @@ async fn retention_raw_metrics_drains_in_batches() {
assert_eq!(count(&pool, "metrics").await, 1);
}

/// The history tables (spans/logs/metric_series_rollups) must drain a backlog
/// across MULTIPLE statements, exactly like raw metrics.
///
/// This is the regression test for the bug that made retention permanently
/// stall: those deletes were issued as ONE unbatched `DELETE ... WHERE time <
/// cutoff`. Against a real backlog that exceeds the connection's
/// statement_timeout it is cancelled, rolls back completely, deletes nothing,
/// and the next sweep faces a larger backlog. On 2026-08-25 `logs` held 11.5M
/// expired rows and retention had never completed a single sweep.
///
/// The old shape passes the existing `retention_prunes_old_rows` test, because
/// two rows per table always fit in one statement. `batch = 1` is what actually
/// exercises the loop.
#[tokio::test]
#[serial]
async fn retention_history_tables_drain_in_batches() {
let Some(pool) = pool_or_skip().await else {
return;
};
let day = 86_400.0;
for _ in 0..3 {
insert_log_at(&pool, "svc", 10.0 * day).await;
}
insert_log_at(&pool, "svc", 1.0 * day).await;

let pruned = watcher_server::retention::prune_batched(&pool, "logs", "time", "days", 7, 1)
.await
.unwrap();
assert_eq!(
pruned, 3,
"backlog must drain across batches, not stop after one"
);
assert_eq!(count(&pool, "logs").await, 1, "in-window row must survive");
}

/// One failing table must not starve the tables after it in the sweep.
///
/// The loop used `?`, so the first failure aborted the whole sweep — and
/// `metric_series_rollups` is ordered AFTER `logs`, so once the `logs` delete
/// began timing out hourly, the rollups table was never swept again and grew to
/// 20 GB. Here a BEFORE DELETE trigger makes `spans` (the FIRST table) fail;
/// `logs` comes after it and must still be pruned.
#[tokio::test]
#[serial]
async fn retention_one_failing_table_does_not_starve_the_rest() {
let Some(pool) = pool_or_skip().await else {
return;
};
let day = 86_400.0;
insert_span_at(&pool, "svc", "old", "o", 10.0 * day).await;
insert_log_at(&pool, "svc", 10.0 * day).await;

sqlx::query(
"CREATE OR REPLACE FUNCTION retention_test_boom() RETURNS trigger AS $$
BEGIN RAISE EXCEPTION 'induced failure'; END; $$ LANGUAGE plpgsql",
)
.execute(&pool)
.await
.unwrap();
sqlx::query(
"CREATE TRIGGER retention_test_boom BEFORE DELETE ON spans
FOR EACH STATEMENT EXECUTE FUNCTION retention_test_boom()",
)
.execute(&pool)
.await
.unwrap();

let res = watcher_server::retention::prune_once(
&pool,
7,
0,
watcher_server::retention::Windows::default(),
)
.await;

sqlx::query("DROP TRIGGER IF EXISTS retention_test_boom ON spans")
.execute(&pool)
.await
.unwrap();

// The sweep as a whole must still report failure — a partial sweep leaves a
// table growing, so /healthz has to keep reporting the stall.
assert!(res.is_err(), "a failed table must fail the sweep");
// ...but the tables after the failing one must have been pruned anyway.
assert_eq!(
count(&pool, "logs").await,
0,
"logs is ordered after spans and must still be pruned when spans fails"
);
}

// --- Alerts ----------------------------------------------------------------

/// Apply declared rules through the real reconcile path. Rules are declarative
Expand Down