graham/towerops
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: H22 — batch data retention deletes in 10k-row chunks

Browse source 
Replace 8 unbounded DELETE statements with a batched_delete loop to prevent
WAL amplification, autovacuum storms, and long row locks on large tables.
This commit is contained in:
Graham McIntire 2026-05-12 12:26:50 -05:00
parent 52cde2cbc5
commit d456283844
2 changed files with 80 additions and 123 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
bugs.md
View file

@ -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

181
lib/towerops/workers/data_retention_worker.ex
View file

@ -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