From d4562838447b42e56395922a59a76a60a13167a2 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 12 May 2026 12:26:50 -0500 Subject: [PATCH] =?UTF-8?q?fix:=20H22=20=E2=80=94=20batch=20data=20retenti?= =?UTF-8?q?on=20deletes=20in=2010k-row=20chunks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace 8 unbounded DELETE statements with a batched_delete loop to prevent WAL amplification, autovacuum storms, and long row locks on large tables. --- bugs.md | 22 --- lib/towerops/workers/data_retention_worker.ex | 181 ++++++++---------- 2 files changed, 80 insertions(+), 123 deletions(-) diff --git a/bugs.md b/bugs.md index 4a689d2b..8853366e 100644 --- a/bugs.md +++ b/bugs.md @@ -43,29 +43,7 @@ --- -### H13. PromEx Metrics Server Unauthenticated -**File:** `config/config.exs:106-117` - -**Severity:** HIGH — Internal metrics accessible without any auth - -**Description:** `auth_strategy: :none` on metrics server. Exposes DB query performance, memory info, business logic execution times. - -**Fix:** Add at minimum basic token auth, or document network isolation explicitly. - ---- - -### H22. DataRetentionWorker Unbounded DELETEs - -**File:** `lib/towerops/workers/data_retention_worker.ex:61-173` - -**Severity:** HIGH — Massive DELETE statements with no batching - -**Description:** Each table purged with a single DELETE. On millions of rows: WAL amplification, autovacuum storms, long row locks, blocked concurrent reads. - -**Fix:** Use batched deletes (`LIMIT 10000` in a loop) or partition-based truncation. - ---- ### H24. WeatherSyncWorker Never Starts diff --git a/lib/towerops/workers/data_retention_worker.ex b/lib/towerops/workers/data_retention_worker.ex index 9e4181be..7fbd5b2f 100644 --- a/lib/towerops/workers/data_retention_worker.ex +++ b/lib/towerops/workers/data_retention_worker.ex @@ -58,117 +58,96 @@ defmodule Towerops.Workers.DataRetentionWorker do end) end + @batch_size 10_000 + @batch_sleep_ms 100 + defp purge_for_orgs(org_ids, cutoff) do - # Tables with direct organization_id - direct_results = - Enum.map( - [{"check_results", "checked_at"}, {"wireless_client_readings", "checked_at"}], - fn {table, time_col} -> - result = - Repo.query!( - "DELETE FROM #{table} WHERE organization_id = ANY($1::uuid[]) AND #{time_col} < $2", - [org_ids, cutoff] - ) - - {table, result.num_rows} - end - ) - - # Resolved alerts only (keep unresolved regardless of age) - alert_result = - Repo.query!( - """ - DELETE FROM alerts - WHERE organization_id = ANY($1::uuid[]) - AND inserted_at < $2 - AND resolved_at IS NOT NULL - """, + [ + # Tables with direct organization_id + batched_delete( + "check_results", + "organization_id = ANY($1::uuid[]) AND checked_at < $2", + [org_ids, cutoff] + ), + batched_delete( + "wireless_client_readings", + "organization_id = ANY($1::uuid[]) AND checked_at < $2", + [org_ids, cutoff] + ), + # Resolved alerts only (keep unresolved regardless of age) + batched_delete( + "alerts", + "organization_id = ANY($1::uuid[]) AND inserted_at < $2 AND resolved_at IS NOT NULL", + [org_ids, cutoff], + label: "alerts_resolved" + ), + # monitoring_checks: device_id -> devices.organization_id + batched_delete( + "monitoring_checks", + "checked_at < $2 AND device_id IN (SELECT id FROM devices WHERE organization_id = ANY($1::uuid[]))", + [org_ids, cutoff] + ), + # snmp_sensor_readings + batched_delete( + "snmp_sensor_readings", + "checked_at < $2 AND sensor_id IN (SELECT s.id FROM snmp_sensors s JOIN snmp_devices sd ON sd.id = s.snmp_device_id JOIN devices d ON d.id = sd.device_id WHERE d.organization_id = ANY($1::uuid[]))", + [org_ids, cutoff] + ), + # snmp_interface_stats + batched_delete( + "snmp_interface_stats", + "checked_at < $2 AND interface_id IN (SELECT i.id FROM snmp_interfaces i JOIN snmp_devices sd ON sd.id = i.snmp_device_id JOIN devices d ON d.id = sd.device_id WHERE d.organization_id = ANY($1::uuid[]))", + [org_ids, cutoff] + ), + # snmp_processor_readings + batched_delete( + "snmp_processor_readings", + "checked_at < $2 AND processor_id IN (SELECT p.id FROM snmp_processors p JOIN snmp_devices sd ON sd.id = p.snmp_device_id JOIN devices d ON d.id = sd.device_id WHERE d.organization_id = ANY($1::uuid[]))", + [org_ids, cutoff] + ), + # snmp_storage_readings + batched_delete( + "snmp_storage_readings", + "checked_at < $2 AND storage_id IN (SELECT st.id FROM snmp_storage st JOIN snmp_devices sd ON sd.id = st.snmp_device_id JOIN devices d ON d.id = sd.device_id WHERE d.organization_id = ANY($1::uuid[]))", [org_ids, cutoff] ) + ] + end - # monitoring_checks: device_id -> devices.organization_id - mc_result = + # Loops DELETE in batches of @batch_size to avoid WAL amplification, + # autovacuum storms, and long row locks on large tables. + defp batched_delete(table, where_sql, params, opts \\ []) do + label = Keyword.get(opts, :label) || table + batch_param = "$#{length(params) + 1}" + total = batched_delete_loop(table, where_sql, params, batch_param, label, 0) + {label, total} + end + + defp batched_delete_loop(table, where_sql, params, batch_param, label, acc) do + result = Repo.query!( """ - DELETE FROM monitoring_checks - WHERE checked_at < $2 - AND device_id IN (SELECT id FROM devices WHERE organization_id = ANY($1::uuid[])) + DELETE FROM #{table} + WHERE id IN ( + SELECT id FROM #{table} + WHERE #{where_sql} + LIMIT #{batch_param} + ) """, - [org_ids, cutoff] + params ++ [@batch_size] ) - # snmp_sensor_readings: sensor_id -> snmp_sensors -> snmp_devices -> devices -> org - sr_result = - Repo.query!( - """ - DELETE FROM snmp_sensor_readings - WHERE checked_at < $2 - AND sensor_id IN ( - SELECT s.id FROM snmp_sensors s - JOIN snmp_devices sd ON sd.id = s.snmp_device_id - JOIN devices d ON d.id = sd.device_id - WHERE d.organization_id = ANY($1::uuid[]) - ) - """, - [org_ids, cutoff] - ) + count = result.num_rows - # snmp_interface_stats: interface_id -> snmp_interfaces -> snmp_devices -> devices -> org - is_result = - Repo.query!( - """ - DELETE FROM snmp_interface_stats - WHERE checked_at < $2 - AND interface_id IN ( - SELECT i.id FROM snmp_interfaces i - JOIN snmp_devices sd ON sd.id = i.snmp_device_id - JOIN devices d ON d.id = sd.device_id - WHERE d.organization_id = ANY($1::uuid[]) - ) - """, - [org_ids, cutoff] - ) + if count > 0 do + Process.sleep(@batch_sleep_ms) + batched_delete_loop(table, where_sql, params, batch_param, label, acc + count) + else + if acc > 0 do + Logger.info("Data retention: deleted #{acc} rows from #{label}") + end - # snmp_processor_readings: processor_id -> snmp_processors -> snmp_devices -> devices -> org - pr_result = - Repo.query!( - """ - DELETE FROM snmp_processor_readings - WHERE checked_at < $2 - AND processor_id IN ( - SELECT p.id FROM snmp_processors p - JOIN snmp_devices sd ON sd.id = p.snmp_device_id - JOIN devices d ON d.id = sd.device_id - WHERE d.organization_id = ANY($1::uuid[]) - ) - """, - [org_ids, cutoff] - ) - - # snmp_storage_readings: storage_id -> snmp_storage -> snmp_devices -> devices -> org - str_result = - Repo.query!( - """ - DELETE FROM snmp_storage_readings - WHERE checked_at < $2 - AND storage_id IN ( - SELECT st.id FROM snmp_storage st - JOIN snmp_devices sd ON sd.id = st.snmp_device_id - JOIN devices d ON d.id = sd.device_id - WHERE d.organization_id = ANY($1::uuid[]) - ) - """, - [org_ids, cutoff] - ) - - direct_results ++ - [ - {"alerts_resolved", alert_result.num_rows}, - {"monitoring_checks", mc_result.num_rows}, - {"snmp_sensor_readings", sr_result.num_rows}, - {"snmp_interface_stats", is_result.num_rows}, - {"snmp_processor_readings", pr_result.num_rows}, - {"snmp_storage_readings", str_result.num_rows} - ] + acc + end end end