feat(cache): jitter ScoreCacheReconciler sweep interval to avoid thundering herd
Every pod was rescheduling its sweep on a fixed 60 s `Process.send_after` tick, so N replicas hit the shared NFS `/data/scores` mount and Postgres in lock-step every minute. Add a uniform jitter of up to 20 s on top of the 60 s base (effective range 60-80 s) so per-replica schedules drift apart. Expose `next_sweep_interval/0,2` as a public helper so the randomized window can be tested deterministically with `:rand` draws rather than mocking timers. Existing reschedule test passes `jitter_max_ms: 0` to keep its 2 s wait window honest.
This commit is contained in:
parent
6c6e413105
commit
1591ac740d
2 changed files with 50 additions and 3 deletions
|
|
@ -31,6 +31,7 @@ defmodule Microwaveprop.Propagation.ScoreCacheReconciler do
|
|||
require Logger
|
||||
|
||||
@default_interval_ms 60_000
|
||||
@default_jitter_max_ms 20_000
|
||||
|
||||
@spec start_link(keyword()) :: GenServer.on_start() | :ignore
|
||||
def start_link(opts) do
|
||||
|
|
@ -52,16 +53,34 @@ defmodule Microwaveprop.Propagation.ScoreCacheReconciler do
|
|||
end)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Returns the next sweep delay in milliseconds: the configured base
|
||||
interval plus a uniformly random jitter in `[1, jitter_max_ms]`.
|
||||
|
||||
Scheduling each pod's next tick with a randomized offset prevents the
|
||||
cluster-wide "every replica sweeps at :00" thundering herd against the
|
||||
shared NFS scores mount and Postgres.
|
||||
"""
|
||||
@spec next_sweep_interval() :: non_neg_integer()
|
||||
def next_sweep_interval(base_ms \\ @default_interval_ms, jitter_max_ms \\ @default_jitter_max_ms)
|
||||
|
||||
def next_sweep_interval(base_ms, jitter_max_ms) when jitter_max_ms > 0 do
|
||||
base_ms + :rand.uniform(jitter_max_ms)
|
||||
end
|
||||
|
||||
def next_sweep_interval(base_ms, _jitter_max_ms), do: base_ms
|
||||
|
||||
@impl true
|
||||
def init(opts) do
|
||||
interval = Keyword.get(opts, :interval_ms, @default_interval_ms)
|
||||
jitter_max = Keyword.get(opts, :jitter_max_ms, @default_jitter_max_ms)
|
||||
|
||||
_ =
|
||||
if Keyword.get(opts, :run_on_start, true) do
|
||||
_ = Process.send_after(self(), :sweep, 500)
|
||||
end
|
||||
|
||||
{:ok, %{interval_ms: interval}}
|
||||
{:ok, %{interval_ms: interval, jitter_max_ms: jitter_max}}
|
||||
end
|
||||
|
||||
@impl true
|
||||
|
|
@ -72,7 +91,8 @@ defmodule Microwaveprop.Propagation.ScoreCacheReconciler do
|
|||
Logger.info("ScoreCacheReconciler: warmed #{warmed} {band, valid_time} pairs from disk")
|
||||
end
|
||||
|
||||
Process.send_after(self(), :sweep, state.interval_ms)
|
||||
delay = next_sweep_interval(state.interval_ms, state.jitter_max_ms)
|
||||
Process.send_after(self(), :sweep, delay)
|
||||
{:noreply, state}
|
||||
end
|
||||
|
||||
|
|
|
|||
|
|
@ -47,6 +47,31 @@ defmodule Microwaveprop.Propagation.ScoreCacheReconcilerTest do
|
|||
]
|
||||
end
|
||||
|
||||
describe "next_sweep_interval/0" do
|
||||
test "returns a value in [base + 1, base + jitter_max] with defaults" do
|
||||
# Exercise the randomized default many times; each draw must land
|
||||
# inside the advertised (60_000, 80_000] window. `:rand.uniform/1`
|
||||
# returns >= 1 so the minimum is base + 1, not base.
|
||||
for _ <- 1..500 do
|
||||
delay = ScoreCacheReconciler.next_sweep_interval()
|
||||
assert delay >= 60_001
|
||||
assert delay <= 80_000
|
||||
end
|
||||
end
|
||||
|
||||
test "honours explicit base and jitter arguments" do
|
||||
for _ <- 1..200 do
|
||||
delay = ScoreCacheReconciler.next_sweep_interval(1_000, 250)
|
||||
assert delay >= 1_001
|
||||
assert delay <= 1_250
|
||||
end
|
||||
end
|
||||
|
||||
test "returns the base interval unchanged when jitter_max is 0" do
|
||||
assert ScoreCacheReconciler.next_sweep_interval(5_000, 0) == 5_000
|
||||
end
|
||||
end
|
||||
|
||||
describe "sweep_once/0" do
|
||||
test "warms ScoreCache for every {band, valid_time} on disk and missing from cache",
|
||||
%{band_mhz: band_mhz} do
|
||||
|
|
@ -146,7 +171,9 @@ defmodule Microwaveprop.Propagation.ScoreCacheReconcilerTest do
|
|||
|
||||
ScoresFile.write!(band_mhz, first, sample_scores())
|
||||
|
||||
start_supervised!({ScoreCacheReconciler, run_on_start: true, interval_ms: 150})
|
||||
# jitter_max_ms: 0 keeps the reschedule deterministic for the
|
||||
# 2s wait_for_cache window; the jitter default is 20s.
|
||||
start_supervised!({ScoreCacheReconciler, run_on_start: true, interval_ms: 150, jitter_max_ms: 0})
|
||||
|
||||
# First sweep warms `first` but knows nothing about `second`.
|
||||
assert {:ok, _} = wait_for_cache(band_mhz, first, 2_000)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue