diff --git a/.gitignore b/.gitignore index 3351579b..7156d9d2 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,10 @@ # The directory Mix downloads your dependencies sources to. /deps/ +# Vendored package dependencies and build artifacts +vendor/*/deps/ +vendor/*/_build/ + # Where 3rd-party dependencies like ExDoc output generated docs. /doc/ diff --git a/lib/mix/tasks/oban.cancel_stuck_discovery.ex b/lib/mix/tasks/oban.cancel_stuck_discovery.ex new file mode 100644 index 00000000..01b8b808 --- /dev/null +++ b/lib/mix/tasks/oban.cancel_stuck_discovery.ex @@ -0,0 +1,57 @@ +defmodule Mix.Tasks.Oban.CancelStuckDiscovery do + @shortdoc "Cancels stuck DiscoveryWorker jobs" + + @moduledoc """ + Cancels all stuck DiscoveryWorker jobs that are in scheduled or retryable state. + + ## Usage + + # In production via kubectl: + kubectl exec -n towerops deployment/towerops -- /app/bin/towerops rpc "Mix.Tasks.Oban.CancelStuckDiscovery.run([])" + + # Locally: + mix oban.cancel_stuck_discovery + """ + use Mix.Task + + require Logger + + @impl Mix.Task + def run(_args) do + Mix.Task.run("app.start") + + cancelled_count = + Oban.Job + |> Oban.Repo.all(Towerops.Repo) + |> Enum.filter(fn job -> + job.worker == "Towerops.Workers.DiscoveryWorker" and + job.state in [:scheduled, :retryable, :executing] + end) + |> Enum.map(fn job -> + case Oban.cancel_job(job.id) do + {:ok, cancelled_job} -> + Logger.info( + "Cancelled stuck discovery job", + job_id: cancelled_job.id, + device_id: get_in(cancelled_job.args, ["device_id"]), + state: cancelled_job.state, + attempted_at: cancelled_job.attempted_at + ) + + 1 + + {:error, reason} -> + Logger.error( + "Failed to cancel job", + job_id: job.id, + error: inspect(reason) + ) + + 0 + end + end) + |> Enum.sum() + + Mix.shell().info("Cancelled #{cancelled_count} stuck discovery jobs") + end +end diff --git a/lib/towerops/workers/discovery_worker.ex b/lib/towerops/workers/discovery_worker.ex index fd0f0b48..8d82bd24 100644 --- a/lib/towerops/workers/discovery_worker.ex +++ b/lib/towerops/workers/discovery_worker.ex @@ -30,6 +30,9 @@ defmodule Towerops.Workers.DiscoveryWorker do # Timeout for waiting for agent-based discovery (60 seconds) @agent_discovery_timeout_ms 60_000 + # Overall job timeout (5 minutes) - job will be cancelled if it exceeds this + @job_timeout_ms 300_000 + # Consider agents online if they checked in within last 10 minutes @agent_online_threshold_minutes 10 @@ -41,8 +44,43 @@ defmodule Towerops.Workers.DiscoveryWorker do :discard device -> - perform_discovery(device) + # Discovery is a one-shot operation - always discard on completion (success or failure) + # This prevents stuck jobs from retrying indefinitely + # Use Task.async/await with timeout to prevent jobs from hanging indefinitely + task = Task.async(fn -> perform_discovery(device) end) + + case Task.yield(task, @job_timeout_ms) || Task.shutdown(task) do + {:ok, :ok} -> + :ok + + {:ok, {:error, reason}} -> + Logger.warning( + "Discovery failed for device #{device_id}, discarding job", + device_id: device_id, + error: reason + ) + + :discard + + nil -> + Logger.error( + "Discovery job timeout after #{@job_timeout_ms}ms, discarding to prevent retries", + device_id: device_id, + timeout_ms: @job_timeout_ms + ) + + :discard + end end + rescue + error -> + Logger.error( + "Discovery job crashed unexpectedly, discarding to prevent retries", + device_id: device_id, + error: Exception.format(:error, error, __STACKTRACE__) + ) + + :discard end defp perform_discovery(device) do