wip

config/runtime.exs

@@ -229,6 +229,10 @@ if config_env() == :prod do
 
   config :towerops, ToweropsWeb.Endpoint,
     url: [host: host, port: 443, scheme: "https"],
+    check_origin: [
+      "//towerops.net",
+      "//*.towerops.net"
+    ],
     http: [
       # Enable IPv6 and bind on all interfaces.
       # Set it to  {0, 0, 0, 0, 0, 0, 0, 1} for local network only access.

lib/towerops/agents.ex

@@ -675,16 +675,63 @@ defmodule Towerops.Agents do
 
   """
   @spec should_phoenix_poll_device?(Device.t()) :: boolean()
-  def should_phoenix_poll_device?(%Device{} = device) do
-    case get_effective_agent_token(device) do
-      nil ->
-        # No agent assigned - Phoenix can poll
-        true
+  def should_phoenix_poll_device?(%Device{id: device_id}) do
+    # Use a single atomic query to check assignment and cloud poller status
+    # This prevents race conditions where assignment changes between reads
+    # Checks device assignment first, then site default, then org default
+    query =
+      from d in Device,
+        left_join: aa in AgentAssignment,
+        on: aa.device_id == d.id and aa.enabled == true,
+        left_join: device_token in AgentToken,
+        on: device_token.id == aa.agent_token_id,
+        left_join: s in assoc(d, :site),
+        left_join: site_token in AgentToken,
+        on: site_token.id == s.agent_token_id,
+        left_join: o in assoc(s, :organization),
+        left_join: org_token in AgentToken,
+        on: org_token.id == o.default_agent_token_id,
+        where: d.id == ^device_id,
+        select: %{
+          # Device-level assignment
+          has_device_assignment: not is_nil(aa.id),
+          device_is_cloud: device_token.is_cloud_poller,
+          # Site-level fallback
+          has_site_token: not is_nil(s.agent_token_id),
+          site_is_cloud: site_token.is_cloud_poller,
+          # Org-level fallback
+          has_org_token: not is_nil(o.default_agent_token_id),
+          org_is_cloud: org_token.is_cloud_poller
+        }
 
-      agent_token_id ->
-        # Check if the agent is a cloud poller
-        agent_token = get_agent_token!(agent_token_id)
-        agent_token.is_cloud_poller
+    case Repo.one(query) do
+      nil ->
+        # Device doesn't exist
+        false
+
+      result ->
+        # Determine effective agent using precedence: device > site > org
+        cond do
+          # 1. Device-level assignment
+          result.has_device_assignment ->
+            # Device assigned - check if cloud poller (nil = deleted token, treat as cloud)
+            result.device_is_cloud != false
+
+          # 2. Site-level default
+          result.has_site_token ->
+            # Site has default - check if cloud poller
+            result.site_is_cloud != false
+
+          # 3. Organization-level default
+          result.has_org_token ->
+            # Org has default - check if cloud poller
+            result.org_is_cloud != false
+
+          # 4. No agent anywhere
+          true ->
+            # No agent assigned at any level - Phoenix polls
+            true
+        end
     end
   end
 

lib/towerops/devices.ex

@@ -812,11 +812,16 @@ defmodule Towerops.Devices do
     _ = DeviceMonitorWorker.stop_monitoring(device.id)
     _ = DevicePollerWorker.stop_polling(device.id)
 
+    # Wait briefly for any in-flight jobs to complete or detect deletion
+    # This reduces (but doesn't eliminate) the window for orphaned writes
+    # In-flight jobs check device existence via verify_polling_assignment_unchanged
+    Process.sleep(500)
+
     # Get agent assignment before deleting (if any)
     assignment = Agents.get_device_assignment(device.id)
     agent_token_id = if assignment, do: assignment.agent_token_id
 
-    # Delete the device
+    # Delete the device (cascades to SNMP device, sensors, interfaces, etc.)
     result = Repo.delete(device)
 
     # Notify agent to stop processing jobs for this device

lib/towerops/snmp.ex

@@ -503,23 +503,18 @@ defmodule Towerops.Snmp do
 
   @doc """
   Creates or updates a neighbor record.
+
+  Uses atomic PostgreSQL ON CONFLICT to prevent race conditions when multiple
+  pollers try to upsert the same neighbor simultaneously.
   """
   def upsert_neighbor(attrs) do
-    case Repo.get_by(Neighbor,
-           interface_id: attrs.interface_id,
-           remote_chassis_id: attrs[:remote_chassis_id],
-           protocol: attrs.protocol
-         ) do
-      nil ->
-        %Neighbor{}
-        |> Neighbor.changeset(attrs)
-        |> Repo.insert()
-
-      neighbor ->
-        neighbor
-        |> Neighbor.changeset(attrs)
-        |> Repo.update()
-    end
+    %Neighbor{}
+    |> Neighbor.changeset(attrs)
+    |> Repo.insert(
+      on_conflict: {:replace_all_except, [:id, :inserted_at]},
+      conflict_target: [:interface_id, :remote_chassis_id, :protocol],
+      returning: true
+    )
   end
 
   @doc """
@@ -1501,23 +1496,18 @@ defmodule Towerops.Snmp do
 
   @doc """
   Creates or updates an ARP entry.
+
+  Uses atomic PostgreSQL ON CONFLICT to prevent race conditions when multiple
+  pollers try to upsert the same ARP entry simultaneously.
   """
   def upsert_arp_entry(attrs) do
-    case Repo.get_by(ArpEntry,
-           device_id: attrs.device_id,
-           ip_address: attrs.ip_address,
-           mac_address: attrs.mac_address
-         ) do
-      nil ->
-        %ArpEntry{}
-        |> ArpEntry.changeset(attrs)
-        |> Repo.insert()
-
-      arp_entry ->
-        arp_entry
-        |> ArpEntry.changeset(attrs)
-        |> Repo.update()
-    end
+    %ArpEntry{}
+    |> ArpEntry.changeset(attrs)
+    |> Repo.insert(
+      on_conflict: {:replace_all_except, [:id, :inserted_at]},
+      conflict_target: [:device_id, :ip_address, :mac_address],
+      returning: true
+    )
   end
 
   @doc """
@@ -1566,56 +1556,19 @@ defmodule Towerops.Snmp do
 
   @doc """
   Creates or updates a MAC address entry.
+
+  Uses atomic PostgreSQL ON CONFLICT to prevent race conditions when multiple
+  pollers try to upsert the same MAC address simultaneously. The unique constraint
+  handles duplicate detection automatically, eliminating the need for manual cleanup.
   """
   def upsert_mac_address(attrs) do
-    import Ecto.Query
-
-    require Logger
-
-    # Build query to find existing MAC address entry
-    query =
-      from m in MacAddress,
-        where: m.device_id == ^attrs.device_id,
-        where: m.mac_address == ^attrs.mac_address
-
-    # Handle vlan_id comparison (nil requires is_nil/1)
-    query =
-      case attrs[:vlan_id] do
-        nil -> where(query, [m], is_nil(m.vlan_id))
-        vlan_id -> where(query, [m], m.vlan_id == ^vlan_id)
-      end
-
-    # Use Repo.all to handle potential duplicates gracefully
-    # If duplicates exist, keep the first one and delete the rest
-    case Repo.all(query) do
-      [] ->
-        %MacAddress{}
-        |> MacAddress.changeset(attrs)
-        |> Repo.insert()
-
-      [mac_address] ->
-        mac_address
-        |> MacAddress.changeset(attrs)
-        |> Repo.update()
-
-      [first | duplicates] ->
-        # Handle duplicates: delete extras, update first
-        Logger.warning("Found #{length(duplicates)} duplicate MAC address entries, cleaning up",
-          device_id: attrs.device_id,
-          mac_address: attrs.mac_address,
-          vlan_id: attrs[:vlan_id],
-          duplicate_ids: Enum.map(duplicates, & &1.id)
-        )
-
-        # Delete duplicate entries
-        duplicate_ids = Enum.map(duplicates, & &1.id)
-        Repo.delete_all(from m in MacAddress, where: m.id in ^duplicate_ids)
-
-        # Update the remaining entry
-        first
-        |> MacAddress.changeset(attrs)
-        |> Repo.update()
-    end
+    %MacAddress{}
+    |> MacAddress.changeset(attrs)
+    |> Repo.insert(
+      on_conflict: {:replace_all_except, [:id, :inserted_at]},
+      conflict_target: [:device_id, :mac_address, :vlan_id],
+      returning: true
+    )
   end
 
   @doc """

lib/towerops/snmp/adapters/replay.ex

@@ -87,12 +87,18 @@ defmodule Towerops.Snmp.Adapters.Replay do
   def walk(opts, base_oid) when is_binary(base_oid) do
     oid_map = Keyword.fetch!(opts, :oid_map)
 
+    # Normalize base_oid (strip leading dot if present)
+    normalized_base = String.trim_leading(base_oid, ".")
+
     # Filter OIDs that start with base_oid (SNMP walk semantics)
     # Important: Must be children (start with base_oid + ".")
+    # Handle both absolute (.1.2.3) and relative (1.2.3) OID notation
     results =
       oid_map
       |> Enum.filter(fn {oid, _value} ->
-        String.starts_with?(oid, base_oid <> ".")
+        # Normalize OID (strip leading dot)
+        normalized_oid = String.trim_leading(oid, ".")
+        String.starts_with?(normalized_oid, normalized_base <> ".")
       end)
       |> Enum.map(fn {oid, value} ->
         {oid, parse_value(value)}
@@ -121,31 +127,35 @@ defmodule Towerops.Snmp.Adapters.Replay do
   @spec parse_value(String.t()) :: term()
   defp parse_value(value) when is_binary(value) do
     cond do
-      # MAC address (very specific pattern) - check FIRST
+      # Empty string (from ASN_NULL values) - check FIRST
+      # SNMP NULL values have no data, return nil
+      value == "" ->
+        Logger.debug("Parsing empty string as nil")
+        nil
+
+      # MAC address (very specific pattern) - check SECOND
       # Examples: aa:bb:cc:dd:ee:ff, AA-BB-CC-DD-EE-FF
       Regex.match?(~r/^([0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa-f]{2}$/, value) ->
         value
 
-      # IPv4 address (specific pattern) - check SECOND
+      # IPv4 address (specific pattern) - check THIRD
       # Example: 192.168.1.1
       Regex.match?(~r/^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$/, value) ->
         value
 
-      # OID list (contains dots but not IPv4) - check THIRD
-      # Example: "1.3.6.1.4.1.9" => [1, 3, 6, 1, 4, 1, 9]
-      # Must have at least 4 components to avoid IPv4 confusion
-      Regex.match?(~r/^[\d\.]+$/, value) and String.contains?(value, ".") and
-          length(String.split(value, ".")) > 4 ->
+      # OID-like string (contains dots but not IPv4) - check FOURTH
+      # Keep as string rather than parsing to avoid issues with SNMP shorthand notation
+      # Example: "1.3.6.1.4.1.9" => "1.3.6.1.4.1.9" (not parsed)
+      # Note: Leading dots (relative OIDs) and other SNMP notation are preserved
+      Regex.match?(~r/^\.?[\d\.]+$/, value) and String.contains?(value, ".") ->
         value
-        |> String.split(".")
-        |> Enum.map(&String.to_integer/1)
 
-      # Pure integer (no dots) - check FOURTH
+      # Pure integer (no dots) - check FIFTH
       # Example: "42" => 42
       Regex.match?(~r/^\d+$/, value) ->
         String.to_integer(value)
 
-      # Float - check FIFTH
+      # Float - check SIXTH
       # Example: "98.6" => 98.6
       Regex.match?(~r/^-?\d+\.\d+$/, value) ->
         String.to_float(value)
@@ -163,6 +173,8 @@ defmodule Towerops.Snmp.Adapters.Replay do
   defp oid_to_sortable(oid) do
     oid
     |> String.split(".")
+    # Filter out empty strings from leading dots
+    |> Enum.reject(&(&1 == ""))
     |> Enum.map(&String.to_integer/1)
   end
 end

lib/towerops/snmp/discovery.ex

@@ -217,12 +217,9 @@ defmodule Towerops.Snmp.Discovery do
          {:ok, processors} <- discover_processors_with_timeout(client_opts, timeouts),
          Logger.info("Discovering storage...", device_id: device.id),
          {:ok, storage} <- discover_storage_with_timeout(client_opts, timeouts),
-         Logger.info("Saving discovery results...", device_id: device.id),
-         {:ok, discovered_device} <- save_discovery_results(device, device_info, interfaces, sensors, vlans),
-         Logger.info("Syncing IP addresses...", device_id: device.id),
-         :ok <- sync_ip_addresses(discovered_device, ip_addresses),
-         Logger.info("Syncing processors...", device_id: device.id),
-         :ok <- sync_processors(discovered_device, processors),
+         Logger.info("Saving discovery results (including IP addresses and processors)...", device_id: device.id),
+         {:ok, discovered_device} <-
+           save_discovery_results(device, device_info, interfaces, sensors, vlans, ip_addresses, processors),
          Logger.info("Syncing storage...", device_id: device.id),
          :ok <- sync_storage(discovered_device, storage),
          Logger.info("Discovering neighbors...", device_id: device.id),
@@ -592,9 +589,11 @@ defmodule Towerops.Snmp.Discovery do
           device_info(),
           [interface_data()],
           [sensor_data()],
+          [map()],
+          [map()],
           [map()]
         ) :: {:ok, Device.t()} | {:error, term()}
-  defp save_discovery_results(device, device_info, interfaces, sensors, vlans) do
+  defp save_discovery_results(device, device_info, interfaces, sensors, vlans, ip_addresses, processors) do
     # Sanitize raw_discovery_data BEFORE starting transaction to avoid DB timeout
     # This can take several seconds for large discovery data (thousands of OIDs)
     sanitized_device_info = prepare_device_info_for_save(device_info, device.id)
@@ -608,9 +607,18 @@ defmodule Towerops.Snmp.Discovery do
       _ = sync_sensors(snmp_device, sensors)
       _ = sync_vlans(snmp_device, vlans)
 
+      # Sync IP addresses and processors inside transaction for atomicity
+      device_with_interfaces = %{
+        snmp_device
+        | interfaces: Enum.map(synced_interfaces, &Map.put(&1, :device_id, snmp_device.device_id))
+      }
+
+      _ = sync_ip_addresses(device_with_interfaces, ip_addresses)
+      _ = sync_processors(device_with_interfaces, processors)
+
       # Return device with interfaces loaded and device_id added
       # Use snmp_device.device_id which references the Equipment table
-      %{snmp_device | interfaces: Enum.map(synced_interfaces, &Map.put(&1, :device_id, snmp_device.device_id))}
+      device_with_interfaces
     end)
   end
 

lib/towerops/workers/device_monitor_worker.ex

@@ -72,13 +72,21 @@ defmodule Towerops.Workers.DeviceMonitorWorker do
   end
 
   defp schedule_next_check_with_error_handling(device_id) do
-    case schedule_next_check(device_id) do
-      {:ok, _job} ->
+    # Verify device still exists before scheduling next check
+    case Devices.get_device(device_id) do
+      nil ->
+        Logger.debug("Device deleted, not rescheduling monitoring", device_id: device_id)
         :ok
 
-      {:error, changeset} ->
-        Logger.error("Failed to schedule next monitoring check for device #{device_id}: #{inspect(changeset.errors)}")
-        :ok
+      _device ->
+        case schedule_next_check(device_id) do
+          {:ok, _job} ->
+            :ok
+
+          {:error, changeset} ->
+            Logger.error("Failed to schedule next monitoring check for device #{device_id}: #{inspect(changeset.errors)}")
+            :ok
+        end
     end
   end
 

lib/towerops/workers/device_poller_worker.ex

@@ -88,16 +88,25 @@ defmodule Towerops.Workers.DevicePollerWorker do
     end
   end
 
-  defp schedule_next_poll_with_error_handling(device_id, device) do
-    poll_interval = get_poll_interval(device)
-
-    case schedule_next_poll(device_id, poll_interval) do
-      {:ok, _job} ->
+  defp schedule_next_poll_with_error_handling(device_id, _stale_device) do
+    # Re-fetch device to get latest check_interval_seconds
+    # Configuration could have changed during the polling execution
+    case Devices.get_device(device_id) do
+      nil ->
+        Logger.debug("Device deleted, not rescheduling", device_id: device_id)
         :ok
 
-      {:error, changeset} ->
-        Logger.error("Failed to schedule next poll for device #{device_id}: #{inspect(changeset.errors)}")
-        :ok
+      fresh_device ->
+        poll_interval = get_poll_interval(fresh_device)
+
+        case schedule_next_poll(device_id, poll_interval) do
+          {:ok, _job} ->
+            :ok
+
+          {:error, changeset} ->
+            Logger.error("Failed to schedule next poll for device #{device_id}: #{inspect(changeset.errors)}")
+            :ok
+        end
     end
   end
 
@@ -175,15 +184,55 @@ defmodule Towerops.Workers.DevicePollerWorker do
     ]
 
     # Wait for all tasks with timeout, handle failures gracefully
-    tasks
-    |> Task.yield_many(30_000)
-    |> Enum.each(fn {task, result} ->
-      case result do
-        {:ok, _} -> :ok
-        {:exit, reason} -> Logger.error("Polling task failed: #{inspect(reason)}")
-        nil -> Task.shutdown(task, :brutal_kill)
-      end
-    end)
+    _results =
+      tasks
+      |> Task.yield_many(30_000)
+      |> Enum.map(fn {task, result} ->
+        case result do
+          {:ok, value} ->
+            {:ok, value}
+
+          {:exit, reason} ->
+            Logger.error("Polling task failed: #{inspect(reason)}")
+            {:error, reason}
+
+          nil ->
+            Task.shutdown(task, :brutal_kill)
+            {:error, :timeout}
+        end
+      end)
+
+    # Re-check assignment before processing results
+    # Device could have been reassigned to an agent during the 30-second polling window
+    case verify_polling_assignment_unchanged(device.id) do
+      :ok ->
+        # Assignment unchanged - process results normally
+        :ok
+
+      :reassigned ->
+        Logger.info("Device reassigned during polling, discarding results",
+          device_id: device.id,
+          device_name: device.name
+        )
+
+        :ok
+    end
+  end
+
+  # Verify device assignment hasn't changed since polling started
+  defp verify_polling_assignment_unchanged(device_id) do
+    case Devices.get_device(device_id) do
+      nil ->
+        # Device deleted during polling
+        :reassigned
+
+      device ->
+        if Agents.should_phoenix_poll_device?(device) do
+          :ok
+        else
+          :reassigned
+        end
+    end
   end
 
   defp poll_device_sensors(device, snmp_device, client_opts, now) do

lib/towerops_web/channels/agent_channel.ex

@@ -396,20 +396,20 @@ defmodule ToweropsWeb.AgentChannel do
   def handle_in("monitoring_check", %{"binary" => binary_b64}, socket) when is_binary(binary_b64) do
     with {:ok, binary} <- safe_base64_decode(binary_b64),
          {:ok, check} <- Validator.validate_monitoring_check_message(binary) do
-      maybe_debug_log(socket, "Received monitoring check from agent",
+      maybe_debug_log(socket, "Received ping result from agent",
         device_id: check.device_id,
         status: check.status,
         response_time_ms: check.response_time_ms,
         binary_size: byte_size(binary_b64)
       )
 
-      # Store monitoring check result in database
+      # Store ping result in database
       _ = store_monitoring_check(check, socket)
 
       {:noreply, socket}
     else
       {:error, {type, message}} ->
-        Logger.error("Invalid monitoring check from agent: #{type} - #{message}",
+        Logger.error("Invalid ping result from agent: #{type} - #{message}",
           agent_token_id: socket.assigns.agent_token_id,
           error_type: type,
           error_message: message,
@@ -419,7 +419,7 @@ defmodule ToweropsWeb.AgentChannel do
         {:noreply, socket}
 
       {:error, :base64_decode_failed} ->
-        Logger.error("Failed to decode monitoring check (invalid base64)",
+        Logger.error("Failed to decode ping result (invalid base64)",
           agent_token_id: socket.assigns.agent_token_id,
           binary_size: byte_size(binary_b64)
         )
@@ -894,7 +894,8 @@ defmodule ToweropsWeb.AgentChannel do
 
   defp process_snmp_result(organization_id, result, socket) do
     with {:ok, device} <- fetch_device(result.device_id),
-         :ok <- verify_device_organization(device, organization_id) do
+         :ok <- verify_device_organization(device, organization_id),
+         :ok <- verify_device_assignment(device, socket.assigns.agent_token_id) do
       process_job_result(device, result, socket)
     else
       {:error, :device_not_found} ->
@@ -902,6 +903,32 @@ defmodule ToweropsWeb.AgentChannel do
 
       {:error, :wrong_organization} ->
         Logger.error("Device #{result.device_id} not in agent's organization")
+
+      {:error, :device_reassigned} ->
+        Logger.warning("Ignoring stale result for reassigned device",
+          device_id: result.device_id,
+          agent_token_id: socket.assigns.agent_token_id
+        )
+    end
+  end
+
+  # Verify device is still assigned to this agent before processing results
+  defp verify_device_assignment(device, agent_token_id) do
+    # Get effective agent token (resolves inheritance from site/org)
+    effective_agent_token_id = Agents.get_effective_agent_token(device)
+
+    # Check if it matches the requesting agent
+    if effective_agent_token_id == agent_token_id do
+      # Also check if there's an explicit device assignment that's disabled
+      case Agents.get_device_assignment(device.id) do
+        %{enabled: false} ->
+          {:error, :device_reassigned}
+
+        _ ->
+          :ok
+      end
+    else
+      {:error, :device_reassigned}
     end
   end
 
@@ -911,8 +938,9 @@ defmodule ToweropsWeb.AgentChannel do
 
     with {:ok, device} <- fetch_device(check.device_id),
          :ok <- verify_device_organization(device, organization_id) do
-      # Convert protobuf timestamp to DateTime
-      checked_at = DateTime.from_unix!(check.timestamp, :second)
+      # Use server time to avoid clock skew issues
+      # Agent timestamps can be unreliable due to clock drift
+      checked_at = DateTime.truncate(DateTime.utc_now(), :second)
 
       # Convert status string to atom
       status = String.to_existing_atom(check.status)
@@ -930,7 +958,7 @@ defmodule ToweropsWeb.AgentChannel do
           :ok
 
         {:error, changeset} ->
-          Logger.error("Failed to store monitoring check",
+          Logger.error("Failed to store ping result",
             device_id: check.device_id,
             errors: inspect(changeset.errors)
           )
@@ -939,7 +967,7 @@ defmodule ToweropsWeb.AgentChannel do
       end
     else
       {:error, :device_not_found} ->
-        Logger.error("Device not found for monitoring check: #{check.device_id}")
+        Logger.error("Device not found for ping result: #{check.device_id}")
         {:error, :device_not_found}
 
       {:error, :wrong_organization} ->

lib/towerops_web/live/agent_live/show.html.heex

@@ -117,7 +117,7 @@
           <li>SNMP discovery results (full OID maps)</li>
           <li>Polling results (sensor readings, interface stats)</li>
           <li>Error messages with full context</li>
-          <li>Monitoring check submissions</li>
+          <li>Ping submissions</li>
         </ul>
       </div>
     </div>

lib/towerops_web/live/device_live/show.ex

@@ -52,16 +52,22 @@ defmodule ToweropsWeb.DeviceLive.Show do
       {:ok, _device} ->
         maybe_subscribe_and_schedule_refresh(socket, id)
 
-        tab = Map.get(params, "tab", "overview")
-        page = params |> Map.get("page", "1") |> String.to_integer()
+        # If no tab parameter, redirect to add it to the URL
+        if Map.has_key?(params, "tab") do
+          tab = params["tab"]
+          page = params |> Map.get("page", "1") |> String.to_integer()
 
-        socket =
-          socket
-          |> load_equipment_data(id)
-          |> assign(:active_tab, tab)
-          |> apply_backups_pagination(tab, page)
+          socket =
+            socket
+            |> load_equipment_data(id)
+            |> assign(:active_tab, tab)
+            |> apply_backups_pagination(tab, page)
 
-        {:noreply, socket}
+          {:noreply, socket}
+        else
+          # Redirect to add default tab to URL
+          {:noreply, push_patch(socket, to: ~p"/devices/#{id}?tab=overview")}
+        end
 
       {:error, :not_found} ->
         {:noreply,

lib/towerops_web/live/device_live/show.html.heex

@@ -244,7 +244,7 @@
             end
           ]}
         >
-          Logs
+          Events
         </.link>
 
         <%= if @snmp_device && @snmp_device.raw_discovery_data do %>

lib/towerops_web/live/graph_live/show.ex

@@ -222,7 +222,7 @@ defmodule ToweropsWeb.GraphLive.Show do
   # For unknown sensor types, use the type directly (supports count, pppoe_sessions, etc.)
   defp get_sensor_types_for_chart(sensor_type), do: [sensor_type]
 
-  defp get_chart_config("latency"), do: {"ICMP Latency", "ms", true}
+  defp get_chart_config("latency"), do: {"Ping Latency", "ms", true}
   defp get_chart_config("processors"), do: {"Processor Usage", "%", false}
   defp get_chart_config("memory"), do: {"Memory Usage", "%", false}
   defp get_chart_config("storage"), do: {"Storage Usage", "%", false}
@@ -404,7 +404,7 @@ defmodule ToweropsWeb.GraphLive.Show do
       nil
     else
       dataset = %{
-        label: "ICMP Latency",
+        label: "Ping Latency",
         data: Enum.map(checks, &latency_check_to_chart_point/1)
       }
 

priv/repo/migrations/20260209152641_fix_null_handling_in_unique_constraints.exs

@@ -0,0 +1,75 @@
+defmodule Towerops.Repo.Migrations.FixNullHandlingInUniqueConstraints do
+  use Ecto.Migration
+
+  @moduledoc """
+  Fix unique constraints to handle NULL values correctly.
+
+  PostgreSQL's default unique constraint treats NULL values as distinct,
+  allowing multiple rows with the same non-NULL values but different NULL values.
+
+  This migration replaces the existing unique constraints with ones that use
+  NULLS NOT DISTINCT, which treats NULL values as equal for uniqueness purposes.
+
+  This fixes race conditions where concurrent upserts with NULL values could
+  create duplicate records.
+  """
+
+  def up do
+    # Drop old unique indexes
+    drop_if_exists unique_index(:snmp_neighbors, [:interface_id, :remote_chassis_id, :protocol])
+    drop_if_exists unique_index(:snmp_mac_addresses, [:device_id, :mac_address, :vlan_id])
+
+    # Clean up existing duplicates before creating new constraints
+    # Keep the most recently updated record, delete older duplicates
+    execute("""
+    DELETE FROM snmp_neighbors
+    WHERE id IN (
+      SELECT id
+      FROM (
+        SELECT id,
+               ROW_NUMBER() OVER (
+                 PARTITION BY interface_id, COALESCE(remote_chassis_id, ''), protocol
+                 ORDER BY updated_at DESC
+               ) AS rn
+        FROM snmp_neighbors
+      ) t
+      WHERE rn > 1
+    )
+    """)
+
+    execute("""
+    DELETE FROM snmp_mac_addresses
+    WHERE id IN (
+      SELECT id
+      FROM (
+        SELECT id,
+               ROW_NUMBER() OVER (
+                 PARTITION BY device_id, mac_address, COALESCE(vlan_id::text, '')
+                 ORDER BY updated_at DESC
+               ) AS rn
+        FROM snmp_mac_addresses
+      ) t
+      WHERE rn > 1
+    )
+    """)
+
+    # Create new unique indexes with NULLS NOT DISTINCT
+    # This ensures NULL values are treated as equal for uniqueness
+    create unique_index(:snmp_neighbors, [:interface_id, :remote_chassis_id, :protocol],
+             nulls_distinct: false
+           )
+
+    create unique_index(:snmp_mac_addresses, [:device_id, :mac_address, :vlan_id],
+             nulls_distinct: false
+           )
+  end
+
+  def down do
+    # Revert to original unique indexes (which allow multiple NULLs)
+    drop_if_exists unique_index(:snmp_neighbors, [:interface_id, :remote_chassis_id, :protocol])
+    drop_if_exists unique_index(:snmp_mac_addresses, [:device_id, :mac_address, :vlan_id])
+
+    create unique_index(:snmp_neighbors, [:interface_id, :remote_chassis_id, :protocol])
+    create unique_index(:snmp_mac_addresses, [:device_id, :mac_address, :vlan_id])
+  end
+end

test/towerops/snmp_test.exs

@@ -8,6 +8,7 @@ defmodule Towerops.SnmpTest do
   alias Towerops.Snmp.Device
   alias Towerops.Snmp.Interface
   alias Towerops.Snmp.InterfaceStat
+  alias Towerops.Snmp.MacAddress
   alias Towerops.Snmp.Neighbor
   alias Towerops.Snmp.Sensor
   alias Towerops.Snmp.SensorReading
@@ -1050,6 +1051,7 @@ defmodule Towerops.SnmpTest do
           device_id: device.id,
           interface_id: interface.id,
           protocol: "lldp",
+          remote_chassis_id: "00:11:22:33:44:55",
           remote_system_name: "switch-b",
           last_discovered_at: DateTime.utc_now()
         })
@@ -1061,6 +1063,7 @@ defmodule Towerops.SnmpTest do
           device_id: device.id,
           interface_id: interface.id,
           protocol: "cdp",
+          remote_chassis_id: "aa:bb:cc:dd:ee:ff",
           remote_system_name: "router-a",
           last_discovered_at: DateTime.utc_now()
         })
@@ -1072,6 +1075,7 @@ defmodule Towerops.SnmpTest do
           device_id: device.id,
           interface_id: interface.id,
           protocol: "lldp",
+          remote_chassis_id: "11:22:33:44:55:66",
           remote_system_name: "switch-a",
           last_discovered_at: DateTime.utc_now()
         })
@@ -2298,4 +2302,209 @@ defmodule Towerops.SnmpTest do
       assert Enum.at(discovered, 2).hostname == "old-device"
     end
   end
+
+  describe "concurrent upsert operations (race condition tests)" do
+    test "concurrent neighbor upserts don't create duplicates", %{device: device, snmp_device: snmp_device} do
+      interface =
+        %Interface{}
+        |> Interface.changeset(%{
+          snmp_device_id: snmp_device.id,
+          if_index: 1,
+          if_name: "eth0"
+        })
+        |> Repo.insert!()
+
+      # Spawn 10 concurrent workers all upserting the same neighbor
+      tasks =
+        for i <- 1..10 do
+          Task.async(fn ->
+            Snmp.upsert_neighbor(%{
+              device_id: device.id,
+              interface_id: interface.id,
+              remote_chassis_id: "00:11:22:33:44:55",
+              protocol: "lldp",
+              remote_system_name: "neighbor-#{i}",
+              last_discovered_at: DateTime.utc_now()
+            })
+          end)
+        end
+
+      # Wait for all tasks to complete
+      results = Task.await_many(tasks, 5_000)
+
+      # All should succeed
+      assert Enum.all?(results, fn
+               {:ok, _} -> true
+               _ -> false
+             end)
+
+      # Should have exactly 1 neighbor, not 10
+      neighbors =
+        Repo.all(
+          from n in Neighbor,
+            where: n.interface_id == ^interface.id and n.remote_chassis_id == "00:11:22:33:44:55"
+        )
+
+      assert length(neighbors) == 1
+
+      # The final name should be one of the concurrent writes (non-deterministic but valid)
+      neighbor = hd(neighbors)
+      assert String.starts_with?(neighbor.remote_system_name, "neighbor-")
+    end
+
+    test "concurrent ARP entry upserts don't create duplicates", %{device: device} do
+      # Spawn 10 concurrent workers all upserting the same ARP entry
+      tasks =
+        for _i <- 1..10 do
+          Task.async(fn ->
+            Snmp.upsert_arp_entry(%{
+              device_id: device.id,
+              ip_address: "192.168.1.100",
+              mac_address: "aa:bb:cc:dd:ee:ff",
+              entry_type: "dynamic",
+              last_seen_at: DateTime.utc_now()
+            })
+          end)
+        end
+
+      # Wait for all tasks to complete
+      results = Task.await_many(tasks, 5_000)
+
+      # All should succeed
+      assert Enum.all?(results, fn
+               {:ok, _} -> true
+               _ -> false
+             end)
+
+      # Should have exactly 1 ARP entry, not 10
+      arp_entries =
+        Repo.all(
+          from a in ArpEntry,
+            where:
+              a.device_id == ^device.id and a.ip_address == "192.168.1.100" and
+                a.mac_address == "aa:bb:cc:dd:ee:ff"
+        )
+
+      assert length(arp_entries) == 1
+    end
+
+    test "concurrent MAC address upserts don't create duplicates", %{device: device} do
+      # Spawn 10 concurrent workers all upserting the same MAC address
+      tasks =
+        for i <- 1..10 do
+          Task.async(fn ->
+            Snmp.upsert_mac_address(%{
+              device_id: device.id,
+              mac_address: "11:22:33:44:55:66",
+              vlan_id: 100,
+              port_index: i,
+              last_seen_at: DateTime.utc_now()
+            })
+          end)
+        end
+
+      # Wait for all tasks to complete
+      results = Task.await_many(tasks, 5_000)
+
+      # All should succeed
+      assert Enum.all?(results, fn
+               {:ok, _} -> true
+               _ -> false
+             end)
+
+      # Should have exactly 1 MAC address, not 10
+      mac_addresses =
+        Repo.all(
+          from m in MacAddress,
+            where: m.device_id == ^device.id and m.mac_address == "11:22:33:44:55:66" and m.vlan_id == 100
+        )
+
+      assert length(mac_addresses) == 1
+
+      # The final port_index should be one of the concurrent writes
+      mac = hd(mac_addresses)
+      assert mac.port_index in 1..10
+    end
+
+    test "concurrent MAC address upserts with nil vlan_id don't create duplicates", %{device: device} do
+      # Spawn 10 concurrent workers all upserting the same MAC address with nil vlan_id
+      tasks =
+        for i <- 1..10 do
+          Task.async(fn ->
+            Snmp.upsert_mac_address(%{
+              device_id: device.id,
+              mac_address: "aa:bb:cc:dd:ee:ff",
+              vlan_id: nil,
+              port_index: i,
+              last_seen_at: DateTime.utc_now()
+            })
+          end)
+        end
+
+      # Wait for all tasks to complete
+      results = Task.await_many(tasks, 5_000)
+
+      # All should succeed
+      assert Enum.all?(results, fn
+               {:ok, _} -> true
+               _ -> false
+             end)
+
+      # Should have exactly 1 MAC address, not 10
+      mac_addresses =
+        Repo.all(
+          from m in MacAddress,
+            where: m.device_id == ^device.id and m.mac_address == "aa:bb:cc:dd:ee:ff" and is_nil(m.vlan_id)
+        )
+
+      assert length(mac_addresses) == 1
+    end
+
+    test "concurrent neighbor upserts with nil chassis_id handled correctly", %{
+      device: device,
+      snmp_device: snmp_device
+    } do
+      interface =
+        %Interface{}
+        |> Interface.changeset(%{
+          snmp_device_id: snmp_device.id,
+          if_index: 1,
+          if_name: "eth0"
+        })
+        |> Repo.insert!()
+
+      # Spawn 10 concurrent workers all upserting neighbors with nil chassis_id
+      tasks =
+        for i <- 1..10 do
+          Task.async(fn ->
+            Snmp.upsert_neighbor(%{
+              device_id: device.id,
+              interface_id: interface.id,
+              remote_chassis_id: nil,
+              protocol: "cdp",
+              remote_system_name: "neighbor-#{i}",
+              last_discovered_at: DateTime.utc_now()
+            })
+          end)
+        end
+
+      # Wait for all tasks to complete
+      results = Task.await_many(tasks, 5_000)
+
+      # All should succeed
+      assert Enum.all?(results, fn
+               {:ok, _} -> true
+               _ -> false
+             end)
+
+      # Should have exactly 1 neighbor with nil chassis_id
+      neighbors =
+        Repo.all(
+          from n in Neighbor,
+            where: n.interface_id == ^interface.id and is_nil(n.remote_chassis_id) and n.protocol == "cdp"
+        )
+
+      assert length(neighbors) == 1
+    end
+  end
 end

test/towerops/workers/device_poller_worker_test.exs

@@ -343,4 +343,142 @@ defmodule Towerops.Workers.DevicePollerWorkerTest do
       refute cancelled_jobs == []
     end
   end
+
+  describe "race condition handling" do
+    test "handles device deletion during poll gracefully", %{organization: org, site: site} do
+      {:ok, device} =
+        Devices.create_device(%{
+          name: "Test Device",
+          ip_address: "192.168.1.100",
+          snmp_enabled: true,
+          snmp_version: "2c",
+          snmp_community: "public",
+          site_id: site.id,
+          organization_id: org.id
+        })
+
+      # Perform polling on device without SNMP setup (no sensors/interfaces)
+      # Should complete without crashing
+      assert :ok = DevicePollerWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
+
+      # Delete device
+      Devices.delete_device(device)
+
+      # Verify polling with non-existent device doesn't crash
+      assert :ok = DevicePollerWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
+    end
+
+    test "discards poll results when device is reassigned to agent", %{
+      organization: org,
+      site: site
+    } do
+      # Create a local agent token (is_cloud_poller defaults to false)
+      {:ok, agent_token, _token} = Towerops.Agents.create_agent_token(org.id, "Local Agent")
+
+      {:ok, device} =
+        Devices.create_device(%{
+          name: "Test Device",
+          ip_address: "192.168.1.100",
+          snmp_enabled: true,
+          snmp_version: "2c",
+          snmp_community: "public",
+          site_id: site.id,
+          organization_id: org.id
+        })
+
+      # Create SNMP device
+      snmp_device =
+        %Device{}
+        |> Device.changeset(%{
+          device_id: device.id,
+          sys_name: "test-device",
+          sys_descr: "Test Device"
+        })
+        |> Repo.insert!()
+
+      # Create a sensor
+      %Sensor{}
+      |> Sensor.changeset(%{
+        snmp_device_id: snmp_device.id,
+        sensor_type: "temperature",
+        sensor_index: "1",
+        sensor_oid: "1.3.6.1.2.1.99.1.1.1.4.1",
+        sensor_descr: "Test Sensor",
+        sensor_unit: "C",
+        sensor_divisor: 1,
+        current_reading: 25.0
+      })
+      |> Repo.insert!()
+
+      # Initial state: Phoenix should poll (no agent assigned)
+      assert Towerops.Agents.should_phoenix_poll_device?(device)
+
+      # Mock SNMP to simulate slow polling
+      poll_started = :erlang.monotonic_time(:millisecond)
+
+      expect(SnmpMock, :get, fn _, _, _ ->
+        # Simulate device reassignment during poll
+        if :erlang.monotonic_time(:millisecond) - poll_started < 100 do
+          # First call - assign to agent
+          {:ok, _} = Towerops.Agents.assign_device_to_agent(agent_token.id, device.id)
+          Process.sleep(50)
+        end
+
+        {:ok, 30}
+      end)
+
+      expect(SnmpMock, :walk, fn _, _, _ -> {:ok, []} end)
+
+      # Perform polling - should detect reassignment and discard results
+      assert :ok = DevicePollerWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
+
+      # Verify device is now assigned to local agent
+      device = Repo.reload!(device)
+      refute Towerops.Agents.should_phoenix_poll_device?(device)
+    end
+
+    test "uses fresh poll interval when rescheduling after config change", %{
+      organization: org,
+      site: site
+    } do
+      {:ok, device} =
+        Devices.create_device(%{
+          name: "Test Device",
+          ip_address: "192.168.1.100",
+          snmp_enabled: true,
+          check_interval_seconds: 60,
+          site_id: site.id,
+          organization_id: org.id
+        })
+
+      # Change interval during "polling" (before scheduling next poll)
+      {:ok, _device} = Devices.update_device(device, %{check_interval_seconds: 300})
+
+      # Perform polling - should use NEW interval (300s) when scheduling next poll
+      # Device has no SNMP setup, so it won't actually poll but will reschedule
+      assert :ok = DevicePollerWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
+
+      # The code path re-fetches the device for fresh config before rescheduling
+    end
+
+    test "stops rescheduling when device is deleted", %{organization: org, site: site} do
+      {:ok, device} =
+        Devices.create_device(%{
+          name: "Test Device",
+          ip_address: "192.168.1.100",
+          snmp_enabled: true,
+          site_id: site.id,
+          organization_id: org.id
+        })
+
+      # Perform polling
+      assert :ok = DevicePollerWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
+
+      # Delete device
+      Devices.delete_device(device)
+
+      # Perform polling again - should not crash and not reschedule
+      assert :ok = DevicePollerWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
+    end
+  end
 end