fix: mark project file as failed when export worker errors

CHANGELOG.md

@@ -21,6 +21,12 @@ and this project adheres to
 
 ### Fixed
 
+- ExportWorker now marks the ProjectFile as `:failed` when the export process
+  errors, preventing records from being stuck permanently as `:in_progress` with
+  a nil path. The data retention cron also handles orphaned files with nil paths
+  gracefully instead of crashing.
+  [#4454](https://github.com/OpenFn/lightning/issues/4454)
+
 ## [2.16.1-pre1] - 2026-04-04
 
 ### Added

lib/lightning/projects.ex

@@ -1021,13 +1021,22 @@ defmodule Lightning.Projects do
           f.project_id == ^project_id and f.inserted_at < ago(^period, "day")
       )
       |> Repo.all()
-      |> Enum.each(fn %{path: object_path} = project_file ->
-        result = Lightning.Storage.delete(object_path)
+      |> Enum.each(fn
+        %{path: nil} = project_file ->
+          Logger.warning(
+            "Deleting orphaned project file #{project_file.id} " <>
+              "with nil path (likely a failed export)"
+          )
 
-        if match?({:ok, _res}, result) or
-             match?({:error, %{status: 404}}, result) do
           Repo.delete(project_file)
-        end
+
+        %{path: object_path} = project_file ->
+          result = Lightning.Storage.delete(object_path)
+
+          if match?({:ok, _res}, result) or
+               match?({:error, %{status: 404}}, result) do
+            Repo.delete(project_file)
+          end
       end)
     end
 

lib/lightning/workorders/export_worker.ex

@@ -53,32 +53,32 @@ defmodule Lightning.WorkOrders.ExportWorker do
       }) do
     search_params = SearchParams.from_map(params)
 
-    result =
-      with {:ok, project_file} <- get_project_file(project_file_id),
-           {:ok, project_file} <-
-             update_project_file(project_file, %{status: :in_progress}),
-           {:ok, project} <- get_project(project_id),
-           {:ok, zip_file} <-
-             process_export(project, search_params, project_file),
-           {:ok, storage_path} <- store_project_file(zip_file, project_file) do
-        update_project_file(project_file, %{
-          status: :completed,
-          path: storage_path
-        })
-      end
-
-    case result do
-      {:ok, project_file} ->
-        UserNotifier.notify_history_export_completion(
-          project_file.created_by,
-          project_file
-        )
-
-        Logger.info("Export completed successfully.")
-        :ok
+    with {:ok, project_file} <- get_project_file(project_file_id),
+         {:ok, project_file} <-
+           update_project_file(project_file, %{status: :in_progress}),
+         {:ok, project} <- get_project(project_id),
+         {:ok, zip_file} <-
+           process_export(project, search_params, project_file),
+         {:ok, storage_path} <-
+           store_project_file(zip_file, project_file),
+         {:ok, project_file} <-
+           update_project_file(project_file, %{
+             status: :completed,
+             path: storage_path
+           }) do
+      UserNotifier.notify_history_export_completion(
+        project_file.created_by,
+        project_file
+      )
 
+      Logger.info("Export completed successfully.")
+      :ok
+    else
       {:error, reason} ->
+        mark_project_file_failed(project_file_id)
+
         Logger.error("Export failed with reason: #{inspect(reason)}")
+
         {:error, reason}
     end
   end
@@ -499,4 +499,27 @@ defmodule Lightning.WorkOrders.ExportWorker do
       %{id: log_line.id, message: log_line.message, run_id: log_line.run_id}
     end)
   end
+
+  defp mark_project_file_failed(project_file_id) do
+    case Repo.get(Projects.File, project_file_id) do
+      nil ->
+        :ok
+
+      project_file ->
+        project_file
+        |> Projects.File.mark_failed()
+        |> Repo.update()
+        |> case do
+          {:ok, _project_file} ->
+            :ok
+
+          {:error, changeset} ->
+            Logger.error(
+              "Failed to mark project file #{project_file_id} as failed: #{inspect(changeset.errors)}"
+            )
+
+            {:error, changeset}
+        end
+    end
+  end
 end

test/lightning/export_worker_test.exs

@@ -183,6 +183,27 @@ defmodule Lightning.ExportWorkerTest do
 
       :zip.zip_close(zip_handle)
     end
+
+    test "marks project file as failed when export fails",
+         %{
+           project_file: project_file,
+           search_params: search_params
+         } do
+      non_existent_project_id = Ecto.UUID.generate()
+
+      assert {:error, :project_not_found} ==
+               ExportWorker.perform(%Oban.Job{
+                 args: %{
+                   "project_id" => non_existent_project_id,
+                   "project_file" => project_file.id,
+                   "search_params" => to_string_key_map(search_params)
+                 }
+               })
+
+      project_file = Repo.reload(project_file)
+      assert project_file.status == :failed
+      assert is_nil(project_file.path)
+    end
   end
 
   def extract_and_read(zip_file_path, target_file_name) do

test/lightning/projects_test.exs

@@ -1500,6 +1500,28 @@ defmodule Lightning.ProjectsTest do
       assert Repo.get(Projects.File, project_file2.id)
     end
 
+    test "deletes orphaned project files with nil path" do
+      project =
+        insert(:project, history_retention_period: 7)
+
+      more_days_ago = Date.utc_today() |> Date.add(-8)
+
+      orphaned_file =
+        insert(:project_file,
+          project: project,
+          path: nil,
+          status: :in_progress,
+          inserted_at: DateTime.new!(more_days_ago, ~T[00:00:00])
+        )
+
+      :ok =
+        Projects.perform(%Oban.Job{
+          args: %{"type" => "data_retention"}
+        })
+
+      refute Repo.get(Projects.File, orphaned_file.id)
+    end
+
     test "deletes channel request history based on started_at" do
       project = insert(:project, history_retention_period: 7)
       channel = insert(:channel, project: project)