cleanup worker tweaks

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]

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

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

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