fix: add recurring job dispatch timer for agent-polled devices

The server only sent polling/ping jobs to agents once on WebSocket join
and never re-dispatched them. After the first batch completed (~5-10s),
devices were never polled again until the agent reconnected.

:send_jobs now schedules a follow-up dispatch every 60s (configurable).
Unifies the timer management with :assignments_changed to prevent
accumulation.
This commit is contained in:
Graham McIntire 2026-03-12 12:07:34 -05:00
parent 793c38e784
commit 904ced77b9
No known key found for this signature in database
3 changed files with 66 additions and 6 deletions

View file

@ -93,6 +93,8 @@ config :towerops,
rate_limiting_enabled: false,
# Reduce debounce delay in tests for faster test execution (50ms vs 500ms in prod)
agent_channel_debounce_ms: 50,
# Reduce poll interval in tests for faster test execution (100ms vs 60s in prod)
agent_poll_interval_ms: 100,
# Reduce device deletion delay for faster tests (10ms vs 500ms in prod)
device_deletion_delay_ms: 10,
# Stripe test configuration

View file

@ -156,8 +156,15 @@ 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
cancel_poll_timer(socket)
updated_socket = build_and_push_jobs(socket)
{:noreply, updated_socket}
# Schedule next job dispatch cycle
interval_ms = poll_interval_ms()
timer = Process.send_after(self(), :send_jobs, interval_ms)
{:noreply, assign(updated_socket, :poll_timer, timer)}
end
# Poll a device immediately after discovery completes.
@ -247,15 +254,13 @@ defmodule ToweropsWeb.AgentChannel do
previous_device_ids: socket.assigns[:last_job_device_ids] || []
)
# Cancel any pending job refresh timer
if timer = socket.assigns[:job_refresh_timer] do
Process.cancel_timer(timer)
end
# Cancel any existing poll timer (regular cycle or pending debounce)
cancel_poll_timer(socket)
# Schedule new refresh after debounce delay (configurable, defaults to 500ms in prod, 50ms in test)
debounce_ms = Application.get_env(:towerops, :agent_channel_debounce_ms, 500)
timer = Process.send_after(self(), :send_jobs, debounce_ms)
{:noreply, assign(socket, :job_refresh_timer, timer)}
{:noreply, assign(socket, :poll_timer, timer)}
end
# Handle PubSub broadcast when discovery is requested for a device
@ -2157,6 +2162,16 @@ defmodule ToweropsWeb.AgentChannel do
defp routeros_config?(_), do: false
defp cancel_poll_timer(socket) do
if timer = socket.assigns[:poll_timer] do
Process.cancel_timer(timer)
end
end
defp poll_interval_ms do
Application.get_env(:towerops, :agent_poll_interval_ms, 60_000)
end
## Safe Encoding/Decoding Helpers
# Safely decode Base64-encoded protobuf binary.

View file

@ -1151,6 +1151,49 @@ defmodule ToweropsWeb.AgentChannelTest do
end
end
# ── Periodic job dispatch ────────────────────────────────────────────
describe "periodic job dispatch" do
test ":send_jobs schedules next dispatch after configured interval", %{socket: socket} do
# Trigger :send_jobs manually — it should push jobs AND schedule the next cycle
send(socket.channel_pid, :send_jobs)
assert_push "jobs", %{binary: _}, 200
# The periodic timer is configured to 100ms in test.exs
# Wait for the next automatic dispatch
assert_push "jobs", %{binary: _}, 500
end
test ":assignments_changed resets the poll timer and periodic cycle continues", %{socket: socket} do
# Trigger assignment change — debounced :send_jobs fires after 50ms
send(socket.channel_pid, {:assignments_changed, :assigned})
assert_push "jobs", %{binary: _}, 200
# After the debounced :send_jobs fires, it should schedule the next periodic cycle
# The next automatic dispatch should arrive after ~100ms (poll interval)
assert_push "jobs", %{binary: _}, 500
end
test "no duplicate timers accumulate from rapid :send_jobs calls", %{socket: socket} do
# Send multiple :send_jobs rapidly — each should cancel the previous timer
send(socket.channel_pid, :send_jobs)
assert_push "jobs", %{binary: _}, 200
send(socket.channel_pid, :send_jobs)
assert_push "jobs", %{binary: _}, 200
send(socket.channel_pid, :send_jobs)
assert_push "jobs", %{binary: _}, 200
# After the last :send_jobs, wait for the single periodic dispatch
# If timers accumulated, we'd see multiple rapid pushes here
assert_push "jobs", %{binary: _}, 300
# Only one timer should be running — no extra push within half the poll interval
refute_push "jobs", %{}, 50
end
end
# ── Existing tests (preserved) ──────────────────────────────────────
describe "poll after discovery" do