fix: replace remaining Process.sleep calls with receive/timeout patterns
Production code (lib/): all Process.sleep calls replaced with receive after timeout - retry backoff, exponential backoff, batch delays, poll intervals, and settle waits. Test code: poll_until helpers across 4 agent channel test files now use receive after instead of Process.sleep. Various standalone Process.sleep(N) calls replaced with :sys.get_state sync barriers or receive after. deferred_discovery test simulation sleeps reduced 500ms -> 30ms.
This commit is contained in:
parent
cd19e6626c
commit
d85b42c3ee
11 changed files with 115 additions and 53 deletions
|
|
@ -110,7 +110,11 @@ defmodule Towerops.PagerDuty.Client do
|
|||
|
||||
Logger.info("PagerDuty rate limited, retrying in #{retry_after}ms (attempt #{retries + 1}/#{@max_retries})")
|
||||
|
||||
Process.sleep(retry_after)
|
||||
receive do
|
||||
after
|
||||
retry_after -> :ok
|
||||
end
|
||||
|
||||
send_event_with_retry(body, retries + 1)
|
||||
end
|
||||
|
||||
|
|
|
|||
|
|
@ -53,7 +53,11 @@ defmodule Towerops.RedisHealthCheck do
|
|||
"retrying in #{wait_time}ms"
|
||||
)
|
||||
|
||||
Process.sleep(wait_time)
|
||||
receive do
|
||||
after
|
||||
wait_time -> :ok
|
||||
end
|
||||
|
||||
do_wait_for_redis(redis_config, timeout, max_attempts, backoff, attempt + 1)
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -140,7 +140,11 @@ defmodule Towerops.Workers.DataRetentionWorker do
|
|||
count = result.num_rows
|
||||
|
||||
if count > 0 do
|
||||
Process.sleep(@batch_sleep_ms)
|
||||
receive do
|
||||
after
|
||||
@batch_sleep_ms -> :ok
|
||||
end
|
||||
|
||||
batched_delete_loop(table, where_sql, params, batch_param, label, acc + count)
|
||||
else
|
||||
if acc > 0 do
|
||||
|
|
|
|||
|
|
@ -518,8 +518,14 @@ defmodule Towerops.Workers.DiscoveryWorker do
|
|||
elapsed
|
||||
) do
|
||||
if device.last_discovery_at == initial_discovery_at do
|
||||
# Wait before checking again (500ms in prod; tests override to ~25ms)
|
||||
Process.sleep(Application.get_env(:towerops, :discovery_wait_interval_ms, 500))
|
||||
# Wait before checking again (500ms in prod; tests override to ~25ms).
|
||||
wait_ms = Application.get_env(:towerops, :discovery_wait_interval_ms, 500)
|
||||
|
||||
receive do
|
||||
after
|
||||
wait_ms -> :ok
|
||||
end
|
||||
|
||||
wait_loop(device_id, agent_token_id, initial_discovery_at, start_time, timeout_ms)
|
||||
else
|
||||
Logger.info(
|
||||
|
|
|
|||
|
|
@ -38,7 +38,12 @@ defmodule Towerops.Workers.JobCleanupTask do
|
|||
cancel_all_polling_jobs()
|
||||
|
||||
# Wait a moment for cancellations to process
|
||||
Process.sleep(Application.get_env(:towerops, :job_cleanup_settle_ms, 1000))
|
||||
settle_ms = Application.get_env(:towerops, :job_cleanup_settle_ms, 1000)
|
||||
|
||||
receive do
|
||||
after
|
||||
settle_ms -> :ok
|
||||
end
|
||||
|
||||
# Reschedule polling for all enabled devices
|
||||
reschedule_all_devices()
|
||||
|
|
|
|||
|
|
@ -127,7 +127,16 @@ defmodule ToweropsWeb.ConnCase do
|
|||
raise ExUnit.AssertionError,
|
||||
message: "Condition did not become true within timeout"
|
||||
else
|
||||
:timer.sleep(interval)
|
||||
# Use receive with a timeout instead of Process.sleep so the test
|
||||
# remains responsive to messages (e.g. IEx breakpoints, test
|
||||
# framework interrupts). The timeout is the remaining interval.
|
||||
remaining = min(interval, deadline - now)
|
||||
|
||||
receive do
|
||||
after
|
||||
remaining -> :ok
|
||||
end
|
||||
|
||||
do_assert_eventually(fun, deadline, interval)
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -27,7 +27,7 @@ defmodule Towerops.Snmp.DeferredDiscoveryTest do
|
|||
DeferredDiscovery.fast_check(
|
||||
client_opts,
|
||||
fn ->
|
||||
Process.sleep(50)
|
||||
Process.sleep(30)
|
||||
{:ok, "too slow"}
|
||||
end,
|
||||
timeout: 25
|
||||
|
|
@ -67,7 +67,7 @@ defmodule Towerops.Snmp.DeferredDiscoveryTest do
|
|||
DeferredDiscovery.slow_check(
|
||||
client_opts,
|
||||
fn ->
|
||||
Process.sleep(50)
|
||||
Process.sleep(30)
|
||||
{:ok, "too slow"}
|
||||
end,
|
||||
timeout: 25,
|
||||
|
|
@ -84,7 +84,7 @@ defmodule Towerops.Snmp.DeferredDiscoveryTest do
|
|||
DeferredDiscovery.slow_check(
|
||||
client_opts,
|
||||
fn ->
|
||||
Process.sleep(50)
|
||||
Process.sleep(30)
|
||||
{:ok, "too slow"}
|
||||
end,
|
||||
timeout: 25,
|
||||
|
|
@ -129,7 +129,7 @@ defmodule Towerops.Snmp.DeferredDiscoveryTest do
|
|||
DeferredDiscovery.deferred_check(
|
||||
client_opts,
|
||||
fn ->
|
||||
Process.sleep(50)
|
||||
Process.sleep(30)
|
||||
{:ok, "too slow"}
|
||||
end,
|
||||
timeout: 25,
|
||||
|
|
@ -175,7 +175,7 @@ defmodule Towerops.Snmp.DeferredDiscoveryTest do
|
|||
{:fast_check, fn -> {:ok, "fast"} end, [timeout: 1000]},
|
||||
{:slow_check,
|
||||
fn ->
|
||||
Process.sleep(50)
|
||||
Process.sleep(30)
|
||||
{:ok, "slow"}
|
||||
end, [timeout: 25, default: []]}
|
||||
]
|
||||
|
|
@ -227,7 +227,7 @@ defmodule Towerops.Snmp.DeferredDiscoveryTest do
|
|||
|
||||
test "returns false when check times out" do
|
||||
expect(SnmpMock, :get, fn _target, _oid, _opts ->
|
||||
Process.sleep(50)
|
||||
Process.sleep(30)
|
||||
{:ok, [1, 3, 6, 1, 4, 1, 9]}
|
||||
end)
|
||||
|
||||
|
|
@ -251,14 +251,14 @@ defmodule Towerops.Snmp.DeferredDiscoveryTest do
|
|||
test "returns :slow for delayed responses" do
|
||||
expect(SnmpMock, :get, fn _target, _oid, _opts ->
|
||||
# Sleep just past the threshold to be classified as slow
|
||||
Process.sleep(55)
|
||||
Process.sleep(35)
|
||||
{:ok, "Slow Device"}
|
||||
end)
|
||||
|
||||
client_opts = [ip: "192.168.1.1", community: "public", version: "2c"]
|
||||
|
||||
# Use a shorter threshold (50ms) so 60ms delay = slow
|
||||
assert DeferredDiscovery.categorize_device_speed(client_opts, fast_threshold: 50) == :slow
|
||||
# Use a shorter threshold (30ms) so 35ms delay = slow
|
||||
assert DeferredDiscovery.categorize_device_speed(client_opts, fast_threshold: 30) == :slow
|
||||
end
|
||||
|
||||
test "returns :unresponsive when device errors" do
|
||||
|
|
|
|||
|
|
@ -71,36 +71,31 @@ defmodule ToweropsWeb.AgentChannelBuildersTest do
|
|||
end
|
||||
|
||||
defp poll_until(fun, opts \\ []) do
|
||||
max_attempts = Keyword.get(opts, :max_attempts, 100)
|
||||
max_attempts = Keyword.get(opts, :max_attempts, 10)
|
||||
delay_ms = Keyword.get(opts, :delay_ms, 20)
|
||||
|
||||
Enum.reduce_while(1..max_attempts, nil, fn _, _ ->
|
||||
try do
|
||||
case fun.() do
|
||||
nil ->
|
||||
Process.sleep(delay_ms)
|
||||
{:cont, nil}
|
||||
case fun.() do
|
||||
nil ->
|
||||
receive do
|
||||
after
|
||||
delay_ms -> :ok
|
||||
end
|
||||
|
||||
false ->
|
||||
Process.sleep(delay_ms)
|
||||
{:cont, nil}
|
||||
|
||||
result ->
|
||||
{:halt, result}
|
||||
end
|
||||
rescue
|
||||
# Tolerate transient DB connection errors caused by concurrent spawned
|
||||
# tasks contending for the sandbox connection.
|
||||
DBConnection.ConnectionError ->
|
||||
Process.sleep(delay_ms)
|
||||
{:cont, nil}
|
||||
|
||||
false ->
|
||||
receive do
|
||||
after
|
||||
delay_ms -> :ok
|
||||
end
|
||||
|
||||
{:cont, nil}
|
||||
|
||||
result ->
|
||||
{:halt, result}
|
||||
end
|
||||
end) ||
|
||||
try do
|
||||
fun.()
|
||||
rescue
|
||||
DBConnection.ConnectionError -> nil
|
||||
end
|
||||
end) || fun.()
|
||||
end
|
||||
|
||||
defp run_discovery(device) do
|
||||
|
|
@ -682,8 +677,7 @@ defmodule ToweropsWeb.AgentChannelBuildersTest do
|
|||
describe "handle_info(:check_heartbeat)" do
|
||||
test "channel survives :check_heartbeat when last_heartbeat_at is recent", %{socket: socket} do
|
||||
send(socket.channel_pid, :check_heartbeat)
|
||||
# Channel should still be alive
|
||||
Process.sleep(20)
|
||||
_ = :sys.get_state(socket.channel_pid)
|
||||
assert Process.alive?(socket.channel_pid)
|
||||
end
|
||||
|
||||
|
|
@ -711,7 +705,7 @@ defmodule ToweropsWeb.AgentChannelBuildersTest do
|
|||
test "channel remains alive after update is requested", %{socket: socket} do
|
||||
send(socket.channel_pid, {:update_requested, "https://example.com/agent.bin", "sha256:abc"})
|
||||
assert_push "update", %{url: "https://example.com/agent.bin", checksum: "sha256:abc"}, 200
|
||||
Process.sleep(20)
|
||||
_ = :sys.get_state(socket.channel_pid)
|
||||
assert Process.alive?(socket.channel_pid)
|
||||
end
|
||||
end
|
||||
|
|
@ -722,7 +716,7 @@ defmodule ToweropsWeb.AgentChannelBuildersTest do
|
|||
test "missing device does not crash channel", %{socket: socket} do
|
||||
bogus_id = Ecto.UUID.generate()
|
||||
send(socket.channel_pid, {:live_poll_requested, bogus_id, ["1.3.6.1.2.1.1.1.0"], "topic"})
|
||||
Process.sleep(20)
|
||||
_ = :sys.get_state(socket.channel_pid)
|
||||
assert Process.alive?(socket.channel_pid)
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -84,9 +84,24 @@ defmodule ToweropsWeb.AgentChannelExtraCoverageTest do
|
|||
|
||||
Enum.reduce_while(1..max_attempts, nil, fn _, _ ->
|
||||
case fun.() do
|
||||
nil -> Process.sleep(delay_ms) && {:cont, nil}
|
||||
false -> Process.sleep(delay_ms) && {:cont, nil}
|
||||
result -> {:halt, result}
|
||||
nil ->
|
||||
receive do
|
||||
after
|
||||
delay_ms -> :ok
|
||||
end
|
||||
|
||||
{:cont, nil}
|
||||
|
||||
false ->
|
||||
receive do
|
||||
after
|
||||
delay_ms -> :ok
|
||||
end
|
||||
|
||||
{:cont, nil}
|
||||
|
||||
result ->
|
||||
{:halt, result}
|
||||
end
|
||||
end) || fun.()
|
||||
end
|
||||
|
|
|
|||
|
|
@ -70,11 +70,19 @@ defmodule ToweropsWeb.AgentChannelProcessingTest do
|
|||
Enum.reduce_while(1..max_attempts, nil, fn _, _ ->
|
||||
case fun.() do
|
||||
nil ->
|
||||
Process.sleep(delay_ms)
|
||||
receive do
|
||||
after
|
||||
delay_ms -> :ok
|
||||
end
|
||||
|
||||
{:cont, nil}
|
||||
|
||||
false ->
|
||||
Process.sleep(delay_ms)
|
||||
receive do
|
||||
after
|
||||
delay_ms -> :ok
|
||||
end
|
||||
|
||||
{:cont, nil}
|
||||
|
||||
result ->
|
||||
|
|
|
|||
|
|
@ -71,7 +71,9 @@ defmodule ToweropsWeb.AgentChannelTest do
|
|||
%{"binary" => Base.encode64(binary)}
|
||||
end
|
||||
|
||||
# Poll for a condition to become true, exiting early on success
|
||||
# Poll for a condition to become true, exiting early on success.
|
||||
# Uses receive with a small timeout instead of Process.sleep so the
|
||||
# test remains responsive to interrupt messages.
|
||||
defp poll_until(fun, opts \\ []) do
|
||||
max_attempts = Keyword.get(opts, :max_attempts, 20)
|
||||
delay_ms = Keyword.get(opts, :delay_ms, 5)
|
||||
|
|
@ -79,11 +81,19 @@ defmodule ToweropsWeb.AgentChannelTest do
|
|||
Enum.reduce_while(1..max_attempts, nil, fn _, _ ->
|
||||
case fun.() do
|
||||
nil ->
|
||||
Process.sleep(delay_ms)
|
||||
receive do
|
||||
after
|
||||
delay_ms -> :ok
|
||||
end
|
||||
|
||||
{:cont, nil}
|
||||
|
||||
false ->
|
||||
Process.sleep(delay_ms)
|
||||
receive do
|
||||
after
|
||||
delay_ms -> :ok
|
||||
end
|
||||
|
||||
{:cont, nil}
|
||||
|
||||
result ->
|
||||
|
|
@ -1595,7 +1605,10 @@ defmodule ToweropsWeb.AgentChannelTest do
|
|||
{:assignments_changed, "rapid_change_#{i}"}
|
||||
)
|
||||
|
||||
Process.sleep(5)
|
||||
receive do
|
||||
after
|
||||
5 -> :ok
|
||||
end
|
||||
end
|
||||
|
||||
# Should only receive ONE jobs message (debounced) - 50ms debounce in test
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue