Reliability improvements: Task.Supervisor, discovery retries, polling visibility

- Add Towerops.TaskSupervisor to supervision tree, replace bare spawn/1
  in agent channel with supervised tasks for polling data processing
- Change DiscoveryWorker from max_attempts: 1 to 3 with retryable error
  classification (transient errors retry, permanent errors discard)
- Name all 9 parallel polling tasks and log timeouts/crashes instead of
  silently dropping results via Stream.run()
- Add tests for TaskSupervisor presence and error classification

lib/towerops/application.ex

@@ -101,7 +101,9 @@ defmodule Towerops.Application do
         Towerops.Profiles.YamlProfiles,
         # SnmpKit services for SNMP operations (used for SNMP protocol, not MIB resolution)
         SnmpKit.SnmpMgr.Config,
-        SnmpKit.SnmpMgr.MIB
+        SnmpKit.SnmpMgr.MIB,
+        # Task supervisor for fire-and-forget background tasks (e.g. polling data processing)
+        {Task.Supervisor, name: Towerops.TaskSupervisor}
       ] ++
         dev_only_workers() ++
         background_workers() ++

lib/towerops/workers/device_poller_worker.ex

@@ -171,6 +171,18 @@ defmodule Towerops.Workers.DevicePollerWorker do
     now = DateTime.truncate(DateTime.utc_now(), :second)
 
     # Run all polling operations in parallel using async tasks
+    task_names = [
+      "sensors",
+      "state_sensors",
+      "interfaces",
+      "interface_changes",
+      "neighbors",
+      "arp",
+      "mac",
+      "processors",
+      "storage"
+    ]
+
     tasks = [
       Task.async(fn -> poll_device_sensors(device, snmp_device, client_opts, now) end),
       Task.async(fn -> poll_device_state_sensors(device, snmp_device, client_opts, now) end),
@@ -183,24 +195,30 @@ defmodule Towerops.Workers.DevicePollerWorker do
       Task.async(fn -> poll_device_storage(device, snmp_device, client_opts, now) end)
     ]
 
-    # Wait for all tasks with timeout, handle failures gracefully
-    _results =
-      tasks
-      |> Task.yield_many(30_000)
-      |> Enum.map(fn {task, result} ->
-        case result do
-          {:ok, value} ->
-            {:ok, value}
+    # Wait for all tasks with timeout, log failures
+    tasks
+    |> Task.yield_many(30_000)
+    |> Enum.zip(task_names)
+    |> Enum.each(fn {{task, result}, name} ->
+      case result do
+        {:ok, _value} ->
+          :ok
 
-          {:exit, reason} ->
-            Logger.error("Polling task failed: #{inspect(reason)}")
-            {:error, reason}
+        {:exit, reason} ->
+          Logger.warning("Polling task '#{name}' crashed for #{device.name}: #{inspect(reason)}",
+            device_id: device.id,
+            task: name
+          )
 
-          nil ->
-            Task.shutdown(task, :brutal_kill)
-            {:error, :timeout}
-        end
-      end)
+        nil ->
+          Task.shutdown(task, :brutal_kill)
+
+          Logger.warning("Polling task '#{name}' timed out for #{device.name}",
+            device_id: device.id,
+            task: name
+          )
+      end
+    end)
 
     # Re-check assignment before processing results
     # Device could have been reassigned to an agent during the 30-second polling window
@@ -549,12 +567,16 @@ defmodule Towerops.Workers.DevicePollerWorker do
       fn sensor ->
         result = poll_sensor_value(sensor, client_opts)
         handle_sensor_poll_result(sensor, result, timestamp)
+        sensor.sensor_descr
       end,
       max_concurrency: 2,
       timeout: 40_000,
       on_timeout: :kill_task
     )
-    |> Stream.run()
+    |> Enum.each(fn
+      {:ok, _} -> :ok
+      {:exit, reason} -> Logger.warning("Sensor poll task failed: #{inspect(reason)}")
+    end)
   end
 
   defp poll_state_sensors(state_sensors, client_opts, timestamp, device_id) do
@@ -563,12 +585,16 @@ defmodule Towerops.Workers.DevicePollerWorker do
       fn state_sensor ->
         result = poll_state_sensor_value(state_sensor, client_opts)
         handle_state_sensor_poll_result(state_sensor, result, timestamp, device_id)
+        state_sensor.sensor_descr
       end,
       max_concurrency: 2,
       timeout: 40_000,
       on_timeout: :kill_task
     )
-    |> Stream.run()
+    |> Enum.each(fn
+      {:ok, _} -> :ok
+      {:exit, reason} -> Logger.warning("State sensor poll task failed: #{inspect(reason)}")
+    end)
   end
 
   defp poll_state_sensor_value(state_sensor, client_opts) do

lib/towerops/workers/discovery_worker.ex

@@ -19,7 +19,7 @@ defmodule Towerops.Workers.DiscoveryWorker do
   use Oban.Worker,
     queue: :discovery,
     unique: [period: 60, states: [:available, :scheduled, :executing]],
-    max_attempts: 1
+    max_attempts: 3
 
   alias Towerops.Agents
   alias Towerops.Devices
@@ -39,7 +39,7 @@ defmodule Towerops.Workers.DiscoveryWorker do
   @agent_online_threshold_minutes 10
 
   @impl Oban.Worker
-  @spec perform(Oban.Job.t()) :: :ok | :discard
+  @spec perform(Oban.Job.t()) :: :ok | :discard | {:error, term()}
   def perform(%Oban.Job{args: %{"device_id" => device_id}} = job) do
     Events.broadcast_job_event(job, :started)
     start_time = System.monotonic_time(:second)
@@ -63,31 +63,37 @@ defmodule Towerops.Workers.DiscoveryWorker do
 
         :discard ->
           Events.broadcast_job_event(job, :completed, %{duration: duration})
+
+        {:error, reason} ->
+          Events.broadcast_job_event(job, :failed, %{
+            error: inspect(reason),
+            duration: duration
+          })
       end
 
       result
     rescue
       error ->
         duration = System.monotonic_time(:second) - start_time
+        error_string = Exception.format(:error, error, __STACKTRACE__)
 
         Logger.error(
-          "Discovery job crashed unexpectedly, discarding to prevent retries",
+          "Discovery job crashed unexpectedly",
           device_id: device_id,
-          error: Exception.format(:error, error, __STACKTRACE__)
+          error: error_string
         )
 
         Events.broadcast_job_event(job, :failed, %{
-          error: Exception.format(:error, error, __STACKTRACE__),
+          error: error_string,
           duration: duration
         })
 
+        # Crashes are not retryable - they indicate a code bug, not a transient issue
         :discard
     end
   end
 
   defp perform_device_discovery(device, device_id) do
-    # 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)
 
@@ -96,22 +102,32 @@ defmodule Towerops.Workers.DiscoveryWorker do
         :ok
 
       {:ok, {:error, reason}} ->
-        Logger.warning(
-          "Discovery failed for device #{device_id}, discarding job",
-          device_id: device_id,
-          error: reason
-        )
+        if retryable_error?(reason) do
+          Logger.warning(
+            "Discovery failed for device #{device_id} with transient error, will retry",
+            device_id: device_id,
+            error: inspect(reason)
+          )
 
-        :discard
+          {:error, reason}
+        else
+          Logger.warning(
+            "Discovery failed for device #{device_id} with permanent error, discarding",
+            device_id: device_id,
+            error: inspect(reason)
+          )
+
+          :discard
+        end
 
       nil ->
         Logger.error(
-          "Discovery job timeout after #{@job_timeout_ms}ms, discarding to prevent retries",
+          "Discovery job timeout after #{@job_timeout_ms}ms, will retry",
           device_id: device_id,
           timeout_ms: @job_timeout_ms
         )
 
-        :discard
+        {:error, :job_timeout}
     end
   end
 
@@ -254,6 +270,20 @@ defmodule Towerops.Workers.DiscoveryWorker do
     end
   end
 
+  @doc """
+  Classifies whether a discovery error is transient (retryable) or permanent.
+
+  Transient errors (network issues, timeouts) will be retried by Oban.
+  Permanent errors (device not found, invalid config) are discarded immediately.
+  Unknown errors default to retryable to avoid silent data loss.
+  """
+  @spec retryable_error?(atom() | term()) :: boolean()
+  def retryable_error?(:device_not_found), do: false
+  def retryable_error?(:invalid_config), do: false
+  def retryable_error?(:no_snmp_credentials), do: false
+  def retryable_error?(:device_deleted), do: false
+  def retryable_error?(_reason), do: true
+
   # Private helpers
 
   defp get_assigned_agent(device) do

lib/towerops_web/channels/agent_channel.ex

@@ -1138,7 +1138,7 @@ defmodule ToweropsWeb.AgentChannel do
     device_name = device.name
     snmp_device_id = device.snmp_device.id
 
-    spawn(fn ->
+    Task.Supervisor.start_child(Towerops.TaskSupervisor, fn ->
       try do
         alias Towerops.Snmp.ArpDiscovery
         alias Towerops.Snmp.MacDiscovery

test/towerops/application_test.exs

@@ -0,0 +1,9 @@
+defmodule Towerops.ApplicationTest do
+  use ExUnit.Case, async: true
+
+  describe "supervision tree" do
+    test "Towerops.TaskSupervisor is running" do
+      assert Process.whereis(Towerops.TaskSupervisor)
+    end
+  end
+end

test/towerops/workers/discovery_worker_test.exs

@@ -78,14 +78,14 @@ defmodule Towerops.Workers.DiscoveryWorkerTest do
                DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => non_existent_id}})
     end
 
-    test "returns discard when discovery fails", %{device: device} do
-      # Mock SNMP failure (test_connection fails immediately)
-      expect(SnmpMock, :get, fn _target, _oid, _opts ->
+    test "returns error for transient failures to allow retry", %{device: device} do
+      # Mock SNMP failure with a transient error (timeout)
+      stub(SnmpMock, :get, fn _target, _oid, _opts ->
         {:error, :timeout}
       end)
 
-      # Discovery worker always returns :discard to prevent infinite retries
-      assert :discard = DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
+      # Transient failures should return {:error, reason} so Oban can retry
+      assert {:error, _reason} = DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
     end
 
     test "successfully completes discovery with proper mocks", %{device: device} do
@@ -119,7 +119,7 @@ defmodule Towerops.Workers.DiscoveryWorkerTest do
     test "logs error when discovery fails", %{device: device} do
       import ExUnit.CaptureLog
 
-      expect(SnmpMock, :get, fn _target, _oid, _opts ->
+      stub(SnmpMock, :get, fn _target, _oid, _opts ->
         {:error, :network_unreachable}
       end)
 
@@ -131,4 +131,34 @@ defmodule Towerops.Workers.DiscoveryWorkerTest do
       assert log =~ "Direct SNMP discovery failed:"
     end
   end
+
+  describe "retryable_error?/1" do
+    test "timeout is retryable" do
+      assert DiscoveryWorker.retryable_error?(:timeout)
+    end
+
+    test "network_unreachable is retryable" do
+      assert DiscoveryWorker.retryable_error?(:network_unreachable)
+    end
+
+    test "connection_refused is retryable" do
+      assert DiscoveryWorker.retryable_error?(:connection_refused)
+    end
+
+    test "econnrefused is retryable" do
+      assert DiscoveryWorker.retryable_error?(:econnrefused)
+    end
+
+    test "device_not_found is not retryable" do
+      refute DiscoveryWorker.retryable_error?(:device_not_found)
+    end
+
+    test "invalid_config is not retryable" do
+      refute DiscoveryWorker.retryable_error?(:invalid_config)
+    end
+
+    test "unknown errors are retryable by default" do
+      assert DiscoveryWorker.retryable_error?(:some_unknown_error)
+    end
+  end
 end