diff --git a/CLAUDE.md b/CLAUDE.md index d1eb61ef..24fe49df 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -258,12 +258,14 @@ User submits QSO → enqueue_for_qso() → weather/hrrr/terrain/iemre workers 3. Safety net: `PropagationPruneWorker` every 15 min, deletes files older than 3h 4. `GridCachePruneWorker` every 15 min for in-memory/valkey grid cache -**Known cleanup gaps:** -- Building footprint tiles (`/data/buildings/`) — no expiration, no cleanup -- Canopy tiles (`/data/canopy/`) — no expiration, no cleanup -- Orphaned `.tmp` files from crashed writes — no periodic sweep -- HRDPS companion files (`.hrdps.prop`, `weather_scalars/{iso}.hrdps/`) — prune parity with HRRR needs verification -- Empty parent directories after prune — no cleanup of empty dirs +**Known cleanup gaps (confirmed):** +- Building footprint tiles (`/data/buildings/`) — no cleanup, no mtime-based expiry +- Canopy tiles (`/data/canopy/` + staging COGs) — no cleanup +- `.hrdps.prop` score files — `parse_valid_time` regex (`~r/^(.+)\.(?:prop|ntms)$/`) captures `iso.hrdps` which fails `DateTime.from_iso8601`; silently excluded from all prune + list operations +- `ScalarFile.prune_older_than` — defined but **never called** from `Propagation.prune_old_scores` (only Scores + Profiles called) +- HRDPS scalar dirs (`weather_scalars/{iso}.hrdps/`) — `list_valid_time_dirs` uses `DateTime.from_iso8601` directly which fails on `iso.hrdps` suffix; invisible to prune +- Orphaned `.tmp.*` files — atomic write (write→rename) leaves tmp files on crash; no scavenging in any Elixir or Rust writer (only `scores_file.rs` cleans its tmp on explicit rename error) +- Empty parent dirs after prune — no `rmdir` after individual file deletes in scores/profiles dirs ### Operational gotchas - **`kubectl exec deploy/prop` may route to the backfill pod**: the `prop` and `prop-backfill` deployments share label selectors, so `deploy/prop` can attach to either. When diagnosing cron / leader-only behavior, identify the hot pod by name first: `kubectl -n prop get pods -o name | grep -E '^pod/prop-[a-z0-9]+-[a-z0-9]+$' | head -1`. The backfill pod has `PROP_ROLE=backfill` and `peer: false` on its Oban; running RPC there is misleading because plugins gated on leader (Cron, etc.) appear inert by design. diff --git a/lib/microwaveprop/buildings/ms_footprints.ex b/lib/microwaveprop/buildings/ms_footprints.ex index 887022fd..57d86a18 100644 --- a/lib/microwaveprop/buildings/ms_footprints.ex +++ b/lib/microwaveprop/buildings/ms_footprints.ex @@ -152,4 +152,31 @@ defmodule Microwaveprop.Buildings.MsFootprints do end end end + + @doc """ + Delete building cache files whose mtime is older than `cutoff`. + Returns the number of files deleted. + """ + @spec prune_older_than(DateTime.t()) :: integer() + def prune_older_than(%DateTime{} = cutoff) do + cutoff_unix = DateTime.to_unix(cutoff) + + case File.ls(cache_dir()) do + {:ok, files} -> Enum.reduce(files, 0, &prune_file(&1, &2, cutoff_unix)) + _ -> 0 + end + end + + defp prune_file(file, acc, cutoff_unix) do + path = Path.join(cache_dir(), file) + stat = File.stat!(path) + mtime_unix = stat.mtime |> NaiveDateTime.from_erl!() |> DateTime.from_naive!("Etc/UTC") |> DateTime.to_unix() + + if mtime_unix < cutoff_unix do + _ = File.rm(path) + acc + 1 + else + acc + end + end end diff --git a/lib/microwaveprop/canopy.ex b/lib/microwaveprop/canopy.ex index f17a7415..f6a77963 100644 --- a/lib/microwaveprop/canopy.ex +++ b/lib/microwaveprop/canopy.ex @@ -94,4 +94,64 @@ defmodule Microwaveprop.Canopy do defp default_dir do Application.get_env(:microwaveprop, :canopy_tiles_dir, "/data/canopy") end + + @doc """ + Delete canopy tile cache files older than `cutoff` (by mtime). + Cleans up both tile cache and staging COG files. + Returns the number of files deleted. + """ + @spec prune_older_than(DateTime.t()) :: number() + def prune_older_than(%DateTime{} = cutoff) do + cutoff_unix = DateTime.to_unix(cutoff) + + canopy_dir = Application.get_env(:microwaveprop, :canopy_tiles_dir, "/data/canopy") + staging_dir = Application.get_env(:microwaveprop, :canopy_staging_dir, "/data/canopy/staging") + + tiles_deleted = prune_dir(canopy_dir, cutoff_unix) + staging_deleted = prune_dir(staging_dir, cutoff_unix) + + _ = maybe_rmdir_empty(staging_dir, staging_deleted) + + tiles_deleted + staging_deleted + end + + defp prune_dir(dir, cutoff_unix) do + case File.ls(dir) do + {:ok, files} -> Enum.reduce(files, 0, &prune_file(dir, &1, &2, cutoff_unix)) + _ -> 0 + end + end + + defp prune_file(dir, file, acc, cutoff_unix) do + path = Path.join(dir, file) + + case File.stat(path) do + {:ok, stat} -> + mtime_unix = + stat.mtime |> NaiveDateTime.from_erl!() |> DateTime.from_naive!("Etc/UTC") |> DateTime.to_unix() + + if mtime_unix < cutoff_unix do + _ = do_delete(path) + acc + 1 + else + acc + end + + _ -> + acc + end + end + + defp do_delete(path) do + _ = if File.dir?(path), do: File.rm_rf(path), else: File.rm(path) + end + + defp maybe_rmdir_empty(_dir, 0), do: :ok + + defp maybe_rmdir_empty(dir, _deleted) do + case File.ls(dir) do + {:ok, []} -> File.rmdir(dir) + _ -> :ok + end + end end diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index a338211c..81dc35b0 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -11,6 +11,7 @@ defmodule Microwaveprop.Propagation do alias Microwaveprop.Propagation.Scorer alias Microwaveprop.Propagation.ScoresFile alias Microwaveprop.Repo + alias Microwaveprop.Weather.ScalarFile alias Microwaveprop.Weather.SoundingParams require Logger @@ -234,17 +235,44 @@ defmodule Microwaveprop.Propagation do cutoff = DateTime.add(DateTime.utc_now(), -3, :hour) file_deleted = ScoresFile.prune_older_than(cutoff) profiles_deleted = ProfilesFile.prune_older_than(cutoff) + scalar_deleted = ScalarFile.prune_older_than(cutoff) - if file_deleted + profiles_deleted > 0 do + total = file_deleted + profiles_deleted + scalar_deleted + + if total > 0 do Logger.info( "PropagationScores: pruned #{file_deleted} old score files + " <> - "#{profiles_deleted} profile files (before #{cutoff})" + "#{profiles_deleted} profile files + #{scalar_deleted} scalar dirs " <> + "(before #{cutoff})" ) end + # Sweep orphaned .tmp.* files left by crashed atomic-write processes + tmp_deleted = + Enum.reduce([ScoresFile.base_dir(), ProfilesFile.base_dir(), ScalarFile.base_dir()], 0, &sweep_tmp_dir/2) + + if tmp_deleted > 0 do + Logger.info("PropagationScores: swept #{tmp_deleted} orphaned .tmp files") + end + :ok end + defp sweep_tmp_dir(dir, acc) do + case File.ls(dir) do + {:ok, entries} -> + entries + |> Enum.filter(&String.contains?(&1, ".tmp.")) + |> Enum.reduce(acc, fn entry, acc -> + _ = File.rm_rf(Path.join(dir, entry)) + acc + 1 + end) + + _ -> + acc + end + end + @doc """ Returns distinct valid_times for a band, ordered ascending. Always reads from the on-disk `ScoresFile` store — the `ScoreCache` only diff --git a/lib/microwaveprop/propagation/scores_file.ex b/lib/microwaveprop/propagation/scores_file.ex index 5b165102..874c151d 100644 --- a/lib/microwaveprop/propagation/scores_file.ex +++ b/lib/microwaveprop/propagation/scores_file.ex @@ -513,7 +513,7 @@ defmodule Microwaveprop.Propagation.ScoresFile do end defp parse_valid_time_dt(filename) do - with [_, iso] <- Regex.run(~r/^(.+)\.(?:prop|ntms)$/, filename), + with [_, iso] <- Regex.run(~r/^(.+)\.(?:hrdps\.prop|prop|ntms)$/, filename), {:ok, dt, _} <- DateTime.from_iso8601(iso) do dt else @@ -548,7 +548,7 @@ defmodule Microwaveprop.Propagation.ScoresFile do end defp parse_valid_time(filename) do - with [_, iso] <- Regex.run(~r/^(.+)\.(?:prop|ntms)$/, filename), + with [_, iso] <- Regex.run(~r/^(.+)\.(?:hrdps\.prop|prop|ntms)$/, filename), {:ok, dt, _} <- DateTime.from_iso8601(iso) do DateTime.to_unix(dt) else diff --git a/lib/microwaveprop/weather/scalar_file.ex b/lib/microwaveprop/weather/scalar_file.ex index 8711cef9..1290994a 100644 --- a/lib/microwaveprop/weather/scalar_file.ex +++ b/lib/microwaveprop/weather/scalar_file.ex @@ -509,7 +509,8 @@ defmodule Microwaveprop.Weather.ScalarFile do case File.ls(base) do {:ok, entries} -> for entry <- entries, - {:ok, dt, _} <- [DateTime.from_iso8601(entry)], + iso = String.replace_suffix(entry, ".hrdps", ""), + {:ok, dt, _} <- [DateTime.from_iso8601(iso)], do: {Path.join(base, entry), dt} _ -> diff --git a/lib/microwaveprop/workers/propagation_prune_worker.ex b/lib/microwaveprop/workers/propagation_prune_worker.ex index 48306c8b..130e757f 100644 --- a/lib/microwaveprop/workers/propagation_prune_worker.ex +++ b/lib/microwaveprop/workers/propagation_prune_worker.ex @@ -18,11 +18,28 @@ defmodule Microwaveprop.Workers.PropagationPruneWorker do max_attempts: 3, unique: [period: 300, states: :incomplete] + alias Microwaveprop.Buildings.MsFootprints + alias Microwaveprop.Canopy alias Microwaveprop.Propagation + require Logger + @impl Oban.Pro.Worker def process(%Oban.Job{}) do Propagation.prune_old_scores() + + cutoff_30d = DateTime.add(DateTime.utc_now(), -30, :day) + buildings_deleted = MsFootprints.prune_older_than(cutoff_30d) + canopy_deleted = Canopy.prune_older_than(cutoff_30d) + + if buildings_deleted > 0 do + Logger.info("MsFootprints: pruned #{buildings_deleted} cached tile files older than #{cutoff_30d}") + end + + if canopy_deleted > 0 do + Logger.info("Canopy: pruned #{canopy_deleted} cached tile/staging files older than #{cutoff_30d}") + end + :ok end end