Fix import purge on cleaner failure and reclaim leftover local CSV disk

lib/oban_error_reporter.ex

@@ -37,6 +37,7 @@ defmodule ObanErrorReporter do
 
   defp on_job_exception(%Oban.Job{
          queue: "analytics_imports",
+         worker: "Plausible.Workers.ImportAnalytics",
          args: %{"import_id" => import_id},
          state: "executing",
          attempt: attempt,
@@ -52,6 +53,7 @@ defmodule ObanErrorReporter do
 
   defp on_job_exception(%Oban.Job{
          queue: "analytics_imports",
+         worker: "Plausible.Workers.ImportAnalytics",
          args: %{"import_id" => import_id},
          state: "executing"
        }) do

lib/workers/local_import_analytics_cleaner.ex

@@ -5,17 +5,33 @@ defmodule Plausible.Workers.LocalImportAnalyticsCleaner do
 
   use Oban.Worker, queue: :analytics_imports, unique: [period: 3600]
 
+  require Logger
+
   @impl Oban.Worker
-  def perform(%Oban.Job{args: args}) do
+  def perform(%Oban.Job{args: args, attempt: attempt, max_attempts: max_attempts}) do
     %{"import_id" => import_id, "paths" => paths} = args
 
     if import_in_progress?(import_id) do
       {:snooze, _one_hour = 3600}
     else
-      Enum.each(paths, fn path ->
-        # credo:disable-for-next-line Credo.Check.Refactor.Nesting
-        if File.exists?(path), do: File.rm!(path)
-      end)
+      Enum.each(paths, &delete_path!(&1, import_id, attempt, max_attempts))
+    end
+  end
+
+  defp delete_path!(path, import_id, attempt, max_attempts) do
+    if File.exists?(path) do
+      try do
+        File.rm!(path)
+      rescue
+        e ->
+          if attempt >= max_attempts do
+            Logger.error(
+              "Failed to delete leftover local import file #{path} for import_id=#{import_id}: #{Exception.message(e)}"
+            )
+          end
+
+          reraise e, __STACKTRACE__
+      end
     end
   end
 

test/workers/local_import_analytics_cleaner_test.exs

@@ -0,0 +1,103 @@
+defmodule Plausible.Workers.LocalImportAnalyticsCleanerTest do
+  use Plausible.DataCase, async: true
+  import ExUnit.CaptureLog
+
+  alias Plausible.Workers.LocalImportAnalyticsCleaner
+  alias Plausible.Imported.SiteImport
+
+  require SiteImport
+
+  describe "perform/1" do
+    test "deletes local import files when import is no longer in progress" do
+      site_import = insert(:site_import, status: SiteImport.completed())
+      {dir, path} = writable_import_file!()
+
+      assert :ok =
+               LocalImportAnalyticsCleaner.perform(%Oban.Job{
+                 args: %{"import_id" => site_import.id, "paths" => [path]},
+                 attempt: 1,
+                 max_attempts: 20
+               })
+
+      refute File.exists?(path)
+      File.rm_rf!(dir)
+    end
+
+    test "snoozes when import is still in progress" do
+      site_import = insert(:site_import, status: SiteImport.importing())
+      {dir, path} = writable_import_file!()
+
+      assert {:snooze, 3600} =
+               LocalImportAnalyticsCleaner.perform(%Oban.Job{
+                 args: %{"import_id" => site_import.id, "paths" => [path]},
+                 attempt: 1,
+                 max_attempts: 20
+               })
+
+      assert File.exists?(path)
+      File.rm_rf!(dir)
+    end
+
+    test "logs when a leftover file cannot be deleted on the final attempt" do
+      site_import = insert(:site_import, status: SiteImport.completed())
+      {dir, path} = undeletable_import_file!()
+
+      log =
+        capture_log(fn ->
+          assert_raise File.Error, fn ->
+            LocalImportAnalyticsCleaner.perform(%Oban.Job{
+              args: %{"import_id" => site_import.id, "paths" => [path]},
+              attempt: 20,
+              max_attempts: 20
+            })
+          end
+        end)
+
+      assert log =~
+               "Failed to delete leftover local import file #{path} for import_id=#{site_import.id}"
+
+      assert File.exists?(path)
+      File.chmod!(dir, 0o755)
+      File.rm_rf!(dir)
+    end
+
+    test "does not log undeleted files before the final attempt" do
+      site_import = insert(:site_import, status: SiteImport.completed())
+      {dir, path} = undeletable_import_file!()
+
+      log =
+        capture_log(fn ->
+          assert_raise File.Error, fn ->
+            LocalImportAnalyticsCleaner.perform(%Oban.Job{
+              args: %{"import_id" => site_import.id, "paths" => [path]},
+              attempt: 1,
+              max_attempts: 20
+            })
+          end
+        end)
+
+      refute log =~ "Failed to delete leftover local import file"
+
+      assert File.exists?(path)
+      File.chmod!(dir, 0o755)
+      File.rm_rf!(dir)
+    end
+  end
+
+  defp writable_import_file! do
+    dir =
+      Path.join(System.tmp_dir!(), "plausible-import-cleaner-#{System.unique_integer([:positive])}")
+
+    File.mkdir_p!(dir)
+    path = Path.join(dir, "imported_visitors_20240101_20240131.csv")
+    File.write!(path, "date,visitors\n2024-01-01,1\n")
+    {dir, path}
+  end
+
+  defp undeletable_import_file! do
+    {dir, path} = writable_import_file!()
+    # Deleting a file requires write permission on the parent directory.
+    File.chmod!(dir, 0o555)
+    {dir, path}
+  end
+end

test/workers/oban_error_reporter_test.exs

@@ -1,5 +1,9 @@
 defmodule ObanErrorReporterTest do
-  use ExUnit.Case, async: true
+  use Plausible.DataCase
+  use Bamboo.Test
+  import ExUnit.CaptureLog
+
+  require Plausible.Imported.SiteImport
 
   describe "handle_event/4" do
     setup do
@@ -28,7 +32,7 @@ defmodule ObanErrorReporterTest do
 
     test "logs an error on failure" do
       log =
-        ExUnit.CaptureLog.capture_log(fn ->
+        capture_log(fn ->
           :ok =
             :telemetry.execute(
               [:oban, :job, :exception],
@@ -41,4 +45,149 @@ defmodule ObanErrorReporterTest do
       assert log =~ ":bad_job"
     end
   end
+
+  describe "on_job_exception/1 import failure handling" do
+    setup do
+      user = new_user()
+      site = new_site(owner: user)
+
+      site_import =
+        insert(:site_import,
+          site: site,
+          imported_by: user,
+          status: Plausible.Imported.SiteImport.completed()
+        )
+
+      populate_stats(site, site_import.id, [build(:imported_visitors)])
+
+      {:ok, site: site, site_import: site_import, user: user}
+    end
+
+    test "ImportAnalytics exception purges imported stats on transient failure", %{
+      site_import: site_import
+    } do
+      query = from(v in Plausible.Imported.Visitor, where: v.import_id == ^site_import.id)
+      assert await_clickhouse_count(query, 1)
+
+      capture_log(fn ->
+        ObanErrorReporter.handle_event(
+          [:oban, :job, :exception],
+          %{},
+          exception_meta(
+            worker: "Plausible.Workers.ImportAnalytics",
+            import_id: site_import.id,
+            attempt: 1,
+            max_attempts: 3
+          ),
+          nil
+        )
+      end)
+
+      assert await_clickhouse_count(query, 0)
+
+      site_import = Repo.reload!(site_import)
+      assert site_import.status == Plausible.Imported.SiteImport.completed()
+      assert_no_emails_delivered()
+    end
+
+    test "ImportAnalytics exception marks import failed on final attempt", %{
+      site_import: site_import,
+      user: user,
+      site: site
+    } do
+      capture_log(fn ->
+        ObanErrorReporter.handle_event(
+          [:oban, :job, :exception],
+          %{},
+          exception_meta(
+            worker: "Plausible.Workers.ImportAnalytics",
+            import_id: site_import.id,
+            attempt: 3,
+            max_attempts: 3
+          ),
+          nil
+        )
+      end)
+
+      site_import = Repo.reload!(site_import)
+      assert site_import.status == Plausible.Imported.SiteImport.failed()
+
+      assert_email_delivered_with(
+        to: [user],
+        subject: "Google Analytics import failed for #{site.domain}"
+      )
+    end
+
+    test "LocalImportAnalyticsCleaner exception does not purge stats or mark import failed", %{
+      site_import: site_import
+    } do
+      query = from(v in Plausible.Imported.Visitor, where: v.import_id == ^site_import.id)
+      assert await_clickhouse_count(query, 1)
+
+      capture_log(fn ->
+        ObanErrorReporter.handle_event(
+          [:oban, :job, :exception],
+          %{},
+          exception_meta(
+            worker: "Plausible.Workers.LocalImportAnalyticsCleaner",
+            import_id: site_import.id,
+            attempt: 1,
+            max_attempts: 20,
+            args: %{"import_id" => site_import.id, "paths" => ["/tmp/import.csv"]}
+          ),
+          nil
+        )
+      end)
+
+      assert await_clickhouse_count(query, 1)
+
+      site_import = Repo.reload!(site_import)
+      assert site_import.status == Plausible.Imported.SiteImport.completed()
+      assert_no_emails_delivered()
+    end
+
+    test "LocalImportAnalyticsCleaner final attempt does not mark import failed", %{
+      site_import: site_import
+    } do
+      capture_log(fn ->
+        ObanErrorReporter.handle_event(
+          [:oban, :job, :exception],
+          %{},
+          exception_meta(
+            worker: "Plausible.Workers.LocalImportAnalyticsCleaner",
+            import_id: site_import.id,
+            attempt: 20,
+            max_attempts: 20,
+            args: %{"import_id" => site_import.id, "paths" => ["/tmp/import.csv"]}
+          ),
+          nil
+        )
+      end)
+
+      site_import = Repo.reload!(site_import)
+      assert site_import.status == Plausible.Imported.SiteImport.completed()
+      assert_no_emails_delivered()
+    end
+  end
+
+  defp exception_meta(opts) do
+    import_id = Keyword.fetch!(opts, :import_id)
+    worker = Keyword.fetch!(opts, :worker)
+    attempt = Keyword.fetch!(opts, :attempt)
+    max_attempts = Keyword.fetch!(opts, :max_attempts)
+    args = Keyword.get(opts, :args, %{"import_id" => import_id})
+
+    %{
+      job: %Oban.Job{
+        queue: "analytics_imports",
+        worker: worker,
+        args: args,
+        state: "executing",
+        attempt: attempt,
+        max_attempts: max_attempts
+      },
+      reason: %RuntimeError{message: "boom"},
+      stacktrace: []
+    }
+  end
 end