From 389ce8edebec26efe70a97021c3539aa22013f16 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 14 Jul 2025 12:31:47 -0500 Subject: [PATCH] cleanup worker tweaks --- config/config.exs | 2 +- lib/aprsme/workers/packet_cleanup_worker.ex | 142 ++++++++++++++++-- ...13160000_ensure_packet_cleanup_indexes.exs | 24 +++ .../workers/packet_cleanup_worker_test.exs | 79 +++++++++- 4 files changed, 224 insertions(+), 23 deletions(-) create mode 100644 priv/repo/migrations/20250713160000_ensure_packet_cleanup_indexes.exs diff --git a/config/config.exs b/config/config.exs index cf41a10..4ad2210 100644 --- a/config/config.exs +++ b/config/config.exs @@ -38,7 +38,7 @@ config :aprsme, Oban, {Oban.Plugins.Pruner, max_age: 60 * 60 * 24 * 7}, {Oban.Plugins.Cron, crontab: [ - {"0 0 * * *", Aprsme.Workers.PacketCleanupWorker} + {"0 */6 * * *", Aprsme.Workers.PacketCleanupWorker} ]} ], queues: [default: 10, maintenance: 2] diff --git a/lib/aprsme/workers/packet_cleanup_worker.ex b/lib/aprsme/workers/packet_cleanup_worker.ex index 7a90896..935218f 100644 --- a/lib/aprsme/workers/packet_cleanup_worker.ex +++ b/lib/aprsme/workers/packet_cleanup_worker.ex @@ -6,23 +6,32 @@ defmodule Aprsme.Workers.PacketCleanupWorker do 1. Removing packets older than the retention period (1 year by default) 2. Providing granular cleanup with custom age parameters 3. Logging statistics about cleanup operations + 4. Supporting batch processing for large datasets - This worker is scheduled to run daily at midnight UTC via Oban's cron feature. + This worker is scheduled to run every 6 hours via Oban's cron feature for more frequent, + smaller cleanup operations to prevent large deletion spikes. ## Functions - `perform/1` - Standard cleanup using configured retention period - `cleanup_packets_older_than/1` - Cleanup packets older than specified days + - `cleanup_packets_in_batches/1` - Batch cleanup for large datasets """ use Oban.Worker, queue: :maintenance, max_attempts: 3 # Import modules needed for database operations + import Ecto.Query alias Aprsme.Packet alias Aprsme.Repo require Logger + # Batch size for cleanup operations to prevent long-running transactions + @batch_size 10_000 + # Maximum time to spend on cleanup in milliseconds (5 minutes) + @max_cleanup_time 5 * 60 * 1000 + @impl Oban.Worker @spec perform(Oban.Job.t()) :: :ok | {:error, String.t()} def perform(%Oban.Job{args: %{"cleanup_days" => days}}) when is_integer(days) do @@ -31,8 +40,8 @@ defmodule Aprsme.Workers.PacketCleanupWorker do # Count packets before cleanup for statistics _total_before = count_total_packets() - # Perform the cleanup of old packets with custom age - deleted_count = packets_module().clean_packets_older_than(days) + # Perform the cleanup of old packets with custom age using batch processing + {deleted_count, _} = cleanup_packets_older_than_batched(days) # Log results Logger.info("APRS packet cleanup complete: removed #{deleted_count} packets older than #{days} days") @@ -48,8 +57,8 @@ defmodule Aprsme.Workers.PacketCleanupWorker do # Count packets before cleanup for statistics _total_before = count_total_packets() - # Perform the cleanup of old packets (older than 1 year by default) - deleted_count = cleanup_old_packets() + # Perform the cleanup of old packets (older than 1 year by default) using batch processing + {deleted_count, _} = cleanup_old_packets_batched() # Log results retention_days = Application.get_env(:aprsme, :packet_retention_days, 365) @@ -64,33 +73,134 @@ defmodule Aprsme.Workers.PacketCleanupWorker do Repo.aggregate(Packet, :count, :id) end - defp cleanup_old_packets do - # Use the existing function from Packets context - packets_module().clean_old_packets() + defp cleanup_old_packets_batched do + retention_days = Application.get_env(:aprsme, :packet_retention_days, 365) + cleanup_packets_older_than_batched(retention_days) end @doc """ - Perform cleanup of packets older than a specific number of days. + Perform cleanup of packets older than a specific number of days using batch processing. - This function provides more granular control over packet cleanup operations. + This function provides more granular control over packet cleanup operations and + processes deletions in smaller batches to prevent long-running transactions. ## Parameters - `days` - Number of days to retain packets (packets older than this will be deleted) + ## Returns + - {deleted_count, batches_processed} + """ + @spec cleanup_packets_older_than_batched(pos_integer()) :: {non_neg_integer(), non_neg_integer()} + def cleanup_packets_older_than_batched(days) when is_integer(days) and days > 0 do + Logger.info("Starting batched APRS packet cleanup for packets older than #{days} days") + + cutoff_time = DateTime.add(DateTime.utc_now(), -days * 86_400, :second) + start_time = System.monotonic_time(:millisecond) + + {deleted_count, batches_processed} = cleanup_packets_in_batches(cutoff_time, start_time) + + duration = System.monotonic_time(:millisecond) - start_time + + Logger.info( + "APRS packet cleanup complete: removed #{deleted_count} packets in #{batches_processed} batches over #{duration}ms" + ) + + {deleted_count, batches_processed} + end + + @doc """ + Clean packets older than a specific number of days. + + This function allows for more granular cleanup operations by specifying + the exact age threshold for packet deletion. + + ## Parameters + - `days` - Number of days to keep (packets older than this will be deleted) + ## Returns - Number of packets deleted """ @spec cleanup_packets_older_than(pos_integer()) :: non_neg_integer() def cleanup_packets_older_than(days) when is_integer(days) and days > 0 do - Logger.info("Starting APRS packet cleanup for packets older than #{days} days") - - deleted_count = packets_module().clean_packets_older_than(days) - - Logger.info("APRS packet cleanup complete: removed #{deleted_count} packets older than #{days} days") - + {deleted_count, _} = cleanup_packets_older_than_batched(days) deleted_count end + @doc """ + Clean packets in batches to prevent long-running transactions. + + This function processes deletions in smaller batches and respects time limits + to ensure the cleanup doesn't impact system performance. + + ## Parameters + - `cutoff_time` - DateTime before which packets should be deleted + - `start_time` - Monotonic time when cleanup started (for duration tracking) + + ## Returns + - {total_deleted_count, batches_processed} + """ + @spec cleanup_packets_in_batches(DateTime.t(), integer()) :: {non_neg_integer(), non_neg_integer()} + def cleanup_packets_in_batches(cutoff_time, start_time) do + cleanup_packets_in_batches(cutoff_time, start_time, 0, 0) + end + + defp cleanup_packets_in_batches(cutoff_time, start_time, total_deleted, batches_processed) do + # Check if we've exceeded the maximum cleanup time + current_time = System.monotonic_time(:millisecond) + + if current_time - start_time > @max_cleanup_time do + Logger.info("Cleanup time limit reached (#{@max_cleanup_time}ms), stopping batch processing") + {total_deleted, batches_processed} + else + # Get batch of packet IDs to delete + packet_ids = get_packet_ids_for_deletion(cutoff_time, @batch_size) + + case packet_ids do + [] -> + # No more packets to delete + {total_deleted, batches_processed} + + ids -> + # Delete the batch + {deleted_count, _} = Repo.delete_all(from(p in Packet, where: p.id in ^ids)) + + new_total_deleted = total_deleted + deleted_count + new_batches_processed = batches_processed + 1 + + Logger.debug( + "Cleanup batch #{new_batches_processed}: deleted #{deleted_count} packets (total: #{new_total_deleted})" + ) + + # Continue with next batch + cleanup_packets_in_batches(cutoff_time, start_time, new_total_deleted, new_batches_processed) + end + end + end + + @doc """ + Get packet IDs for deletion in batches. + + This function queries for packet IDs that are older than the cutoff time, + limiting the result to the specified batch size. + + ## Parameters + - `cutoff_time` - DateTime before which packets should be deleted + - `batch_size` - Maximum number of IDs to return + + ## Returns + - List of packet IDs to delete + """ + @spec get_packet_ids_for_deletion(DateTime.t(), pos_integer()) :: [integer()] + def get_packet_ids_for_deletion(cutoff_time, batch_size) do + Repo.all( + from(p in Packet, + where: p.received_at < ^cutoff_time, + select: p.id, + limit: ^batch_size + ) + ) + end + defp packets_module do Application.get_env(:aprsme, :packets_module, Aprsme.Packets) end diff --git a/priv/repo/migrations/20250713160000_ensure_packet_cleanup_indexes.exs b/priv/repo/migrations/20250713160000_ensure_packet_cleanup_indexes.exs new file mode 100644 index 0000000..e14544d --- /dev/null +++ b/priv/repo/migrations/20250713160000_ensure_packet_cleanup_indexes.exs @@ -0,0 +1,24 @@ +defmodule Aprsme.Repo.Migrations.EnsurePacketCleanupIndexes do + use Ecto.Migration + @disable_ddl_transaction true + @disable_migration_lock true + + def up do + # Ensure index on received_at for efficient time-based deletes + execute """ + CREATE INDEX CONCURRENTLY IF NOT EXISTS packets_received_at_idx + ON packets (received_at) + """ + + # Ensure index on id for efficient batch deletes + execute """ + CREATE INDEX CONCURRENTLY IF NOT EXISTS packets_id_idx + ON packets (id) + """ + end + + def down do + execute "DROP INDEX IF EXISTS packets_received_at_idx" + execute "DROP INDEX IF EXISTS packets_id_idx" + end +end diff --git a/test/aprsme/workers/packet_cleanup_worker_test.exs b/test/aprsme/workers/packet_cleanup_worker_test.exs index be14d1e..0edb9ad 100644 --- a/test/aprsme/workers/packet_cleanup_worker_test.exs +++ b/test/aprsme/workers/packet_cleanup_worker_test.exs @@ -14,8 +14,16 @@ defmodule Aprsme.Workers.PacketCleanupWorkerTest do days = 30 job = %Oban.Job{args: %{"cleanup_days" => days}} - expect(PacketsMock, :clean_packets_older_than, fn ^days -> - 5 + # Mock the Repo.all call for get_packet_ids_for_deletion + expect(Aprsme.Repo, :all, fn _query -> + # Return some packet IDs + [1, 2, 3, 4, 5] + end) + + # Mock the Repo.delete_all call + expect(Aprsme.Repo, :delete_all, fn _query -> + # Return deleted count + {5, nil} end) assert :ok = PacketCleanupWorker.perform(job) @@ -26,8 +34,16 @@ defmodule Aprsme.Workers.PacketCleanupWorkerTest do test "cleans up packets using the default retention period" do job = %Oban.Job{args: %{}} - expect(PacketsMock, :clean_old_packets, fn -> - 10 + # Mock the Repo.all call for get_packet_ids_for_deletion + expect(Aprsme.Repo, :all, fn _query -> + # Return some packet IDs + [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] + end) + + # Mock the Repo.delete_all call + expect(Aprsme.Repo, :delete_all, fn _query -> + # Return deleted count + {10, nil} end) assert :ok = PacketCleanupWorker.perform(job) @@ -39,11 +55,62 @@ defmodule Aprsme.Workers.PacketCleanupWorkerTest do days = 60 deleted_count = 15 - expect(PacketsMock, :clean_packets_older_than, fn ^days -> - deleted_count + # Mock the Repo.all call for get_packet_ids_for_deletion + expect(Aprsme.Repo, :all, fn _query -> + # Return packet IDs + Enum.to_list(1..deleted_count) + end) + + # Mock the Repo.delete_all call + expect(Aprsme.Repo, :delete_all, fn _query -> + # Return deleted count + {deleted_count, nil} end) assert ^deleted_count = PacketCleanupWorker.cleanup_packets_older_than(days) end end + + describe "cleanup_packets_older_than_batched/1" do + test "returns correct tuple format with deleted count and batches" do + days = 30 + + # Mock the Repo.all call for get_packet_ids_for_deletion + expect(Aprsme.Repo, :all, fn _query -> + # Return some packet IDs + [1, 2, 3, 4, 5] + end) + + # Mock the Repo.delete_all call + expect(Aprsme.Repo, :delete_all, fn _query -> + # Return deleted count + {5, nil} + end) + + {deleted_count, batches} = PacketCleanupWorker.cleanup_packets_older_than_batched(days) + + assert is_integer(deleted_count) + assert is_integer(batches) + assert deleted_count >= 0 + assert batches >= 0 + end + end + + describe "get_packet_ids_for_deletion/2" do + test "returns list of packet IDs within batch size limit" do + cutoff_time = DateTime.add(DateTime.utc_now(), -30, :day) + batch_size = 100 + + # Mock the Repo.all call + expect(Aprsme.Repo, :all, fn _query -> + # Return some mock packet IDs + [1, 2, 3, 4, 5] + end) + + result = PacketCleanupWorker.get_packet_ids_for_deletion(cutoff_time, batch_size) + + assert is_list(result) + assert length(result) <= batch_size + end + end end