From fa2ebdb973ddeb7c007ae8e41644c056f6a39869 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Thu, 12 Feb 2026 14:36:25 -0600 Subject: [PATCH] fix: add diagnostic logging for agent job refresh on device deletion Enhanced AgentChannel logging to track device IDs across job list updates. Added tests verifying server correctly excludes deleted devices from job lists. Root cause: Bug is in Go agent (appends jobs without clearing), not Phoenix server. --- lib/towerops_web/channels/agent_channel.ex | 19 +++- .../channels/agent_channel_test.exs | 98 +++++++++++++++++++ 2 files changed, 113 insertions(+), 4 deletions(-) diff --git a/lib/towerops_web/channels/agent_channel.ex b/lib/towerops_web/channels/agent_channel.ex index a4548259..b176a300 100644 --- a/lib/towerops_web/channels/agent_channel.ex +++ b/lib/towerops_web/channels/agent_channel.ex @@ -143,8 +143,8 @@ defmodule ToweropsWeb.AgentChannel do @impl true @spec handle_info(:send_jobs, Phoenix.Socket.t()) :: {:noreply, Phoenix.Socket.t()} def handle_info(:send_jobs, socket) do - build_and_push_jobs(socket) - {:noreply, socket} + updated_socket = build_and_push_jobs(socket) + {:noreply, updated_socket} end # Poll a device immediately after discovery completes. @@ -227,9 +227,11 @@ defmodule ToweropsWeb.AgentChannel do # Handle PubSub broadcast when device assignments change. # Debounces rapid changes — cancels any pending refresh and schedules a new one in 500ms. - def handle_info({:assignments_changed, _event}, socket) do + def handle_info({:assignments_changed, event}, socket) do Logger.info("Agent assignments changed, scheduling job refresh", - agent_token_id: socket.assigns.agent_token_id + agent_token_id: socket.assigns.agent_token_id, + event: event, + previous_device_ids: socket.assigns[:last_job_device_ids] || [] ) # Cancel any pending job refresh timer @@ -635,12 +637,21 @@ defmodule ToweropsWeb.AgentChannel do job_list = %AgentJobList{jobs: jobs} binary = AgentJobList.encode(job_list) + # Extract device IDs for better debugging + device_ids = jobs |> Enum.map(& &1.device_id) |> Enum.uniq() |> Enum.sort() + Logger.info("Sending #{length(jobs)} jobs to agent", agent_token_id: socket.assigns.agent_token_id, + device_count: length(device_ids), + device_ids: inspect(device_ids), job_ids: inspect(Enum.map(jobs, & &1.job_id)) ) + # Store last job list to detect changes on next refresh + socket = assign(socket, :last_job_device_ids, device_ids) + push(socket, "jobs", %{binary: Base.encode64(binary)}) + socket end @spec build_jobs_for_agent(Ecto.UUID.t()) :: [AgentJob.t()] diff --git a/test/towerops_web/channels/agent_channel_test.exs b/test/towerops_web/channels/agent_channel_test.exs index 60b1dcb9..292e8fc0 100644 --- a/test/towerops_web/channels/agent_channel_test.exs +++ b/test/towerops_web/channels/agent_channel_test.exs @@ -1199,4 +1199,102 @@ defmodule ToweropsWeb.AgentChannelTest do "Missing ssCpuUser OID (1.3.6.1.4.1.2021.11.9.0)" end end + + # ── Stage: Device deletion and job refresh ───────────────────────────────── + + describe "device deletion and job refresh" do + test "agent receives updated job list after device deletion", %{ + socket: socket, + device: device, + agent_token: agent_token + } do + # Create a second device to verify the list shrinks correctly + device2 = + DevicesFixtures.device_fixture(%{ + organization_id: device.organization_id, + snmp_version: "2c", + snmp_community: "public", + ip_address: "192.168.1.2" + }) + + {:ok, _assignment} = + AgentsFixtures.agent_assignment_fixture(agent_token.id, device2.id) + + # Trigger initial job refresh to get baseline + Phoenix.PubSub.broadcast( + Towerops.PubSub, + "agent:#{agent_token.id}:assignments", + {:assignments_changed, :test_setup} + ) + + # Should receive jobs message with 2 devices + assert_push "jobs", %{binary: initial_jobs_binary}, 1_000 + {:ok, decoded_initial} = Base.decode64(initial_jobs_binary) + initial_job_list = AgentJobList.decode(decoded_initial) + initial_device_ids = initial_job_list.jobs |> Enum.map(& &1.device_id) |> Enum.uniq() + assert length(initial_device_ids) == 2 + assert device.id in initial_device_ids + assert device2.id in initial_device_ids + + # Delete the first device + {:ok, _deleted} = Towerops.Devices.delete_device(device) + + # Should receive updated jobs message with only 1 device + assert_push "jobs", %{binary: updated_jobs_binary}, 1_000 + {:ok, decoded_updated} = Base.decode64(updated_jobs_binary) + updated_job_list = AgentJobList.decode(decoded_updated) + updated_device_ids = updated_job_list.jobs |> Enum.map(& &1.device_id) |> Enum.uniq() + + # Verify the deleted device is NOT in the new job list + assert length(updated_device_ids) == 1 + refute device.id in updated_device_ids + assert device2.id in updated_device_ids + end + + test "agent result for deleted device is rejected", %{socket: socket, device: device} do + # Delete the device + {:ok, _deleted} = Towerops.Devices.delete_device(device) + + # Wait for jobs refresh + assert_push "jobs", _, 1_000 + + # Try to send a result for the deleted device + result = %SnmpResult{ + device_id: device.id, + job_type: :POLL, + job_id: "poll:#{device.id}", + timestamp: DateTime.to_unix(DateTime.utc_now()), + oid_values: %{"1.3.6.1.2.1.1.1.0" => "Test"} + } + + payload = encode_payload(result) + push(socket, "result", payload) + + # Channel should remain alive (graceful error handling) + Process.sleep(100) + assert Process.alive?(socket.channel_pid) + + # The result should be logged as "Device not found" but not crash + # (This is verified by the channel staying alive) + end + + test "multiple rapid assignment changes are debounced", %{socket: socket, agent_token: agent_token} do + # Trigger multiple rapid assignment changes + for i <- 1..5 do + Phoenix.PubSub.broadcast( + Towerops.PubSub, + "agent:#{agent_token.id}:assignments", + {:assignments_changed, "rapid_change_#{i}"} + ) + + Process.sleep(10) + end + + # Should only receive ONE jobs message (debounced) + assert_push "jobs", %{binary: _jobs_binary}, 1_000 + + # Should NOT receive additional jobs messages + refute_push "jobs", _, 200 + end + end end