diff --git a/lib/oban_error_reporter.ex b/lib/oban_error_reporter.ex index 1ac0a27c5042..20c1ad6ae851 100644 --- a/lib/oban_error_reporter.ex +++ b/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 diff --git a/lib/workers/local_import_analytics_cleaner.ex b/lib/workers/local_import_analytics_cleaner.ex index f7aac2621bf7..354c121fcf74 100644 --- a/lib/workers/local_import_analytics_cleaner.ex +++ b/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 diff --git a/test/workers/local_import_analytics_cleaner_test.exs b/test/workers/local_import_analytics_cleaner_test.exs new file mode 100644 index 000000000000..5fefb31652fb --- /dev/null +++ b/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 diff --git a/test/workers/oban_error_reporter_test.exs b/test/workers/oban_error_reporter_test.exs index 397689d4f8c6..122d6ee461a9 100644 --- a/test/workers/oban_error_reporter_test.exs +++ b/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