From a992c1f0dd7c601f37af0f5b2b95e4b2db3eff38 Mon Sep 17 00:00:00 2001 From: Claudio Ortolina Date: Mon, 2 Mar 2026 15:04:45 +0000 Subject: [PATCH] Introduce a Req.RateLimiter module to fix API rate limit issues --- docs/architecture.md | 9 +- lib/discogs.ex | 2 - lib/discogs/api.ex | 1 + lib/last_fm.ex | 2 - lib/last_fm/api.ex | 1 + lib/music_brainz.ex | 2 - lib/music_brainz/api.ex | 1 + lib/music_library/application.ex | 1 + .../worker/artist_refresh_discogs_data.ex | 6 +- .../artist_refresh_music_brainz_data.ex | 6 +- .../worker/fetch_artist_last_fm_data.ex | 13 +- .../record_refresh_music_brainz_data.ex | 11 +- lib/req/rate_limiter.ex | 71 ++++++++++ test/req/rate_limiter_test.exs | 122 ++++++++++++++++++ 14 files changed, 211 insertions(+), 37 deletions(-) create mode 100644 lib/req/rate_limiter.ex create mode 100644 test/req/rate_limiter_test.exs diff --git a/docs/architecture.md b/docs/architecture.md index 4818d3ef..315a55a0 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -114,6 +114,7 @@ Last.fm schemas (separate, not Ecto-persisted to main DB): | `Batch` | Generic batch runner: stream + transaction + error accumulation | | `Records.Batch` | Batch operations: refresh all MusicBrainz data, generate all embeddings (uses `Batch`) | | `Artists.Batch` | Batch refresh: MusicBrainz, Discogs, Wikipedia, Last.fm for all artists (uses `Batch`) | +| `Req.RateLimiter` | ETS-backed Req request step enforcing per-API minimum intervals between requests | | `Assets.Cache` | ETS-based asset cache with TTL | | `Assets.Image` / `Assets.Transform` | Image processing via Vix (libvips) | | `Colors.ColorFrequencyExtractor` | Color extraction via pixel sampling/histogram | @@ -149,10 +150,10 @@ stubbed via `Req.Test` (configured in `config/test.exs`). |-------|-------------|---------| | `default` | 10 | General async tasks | | `heavy_writes` | 1 | DB-intensive or serialized operations | -| `music_brainz` | 1 | Rate-limited MusicBrainz calls (500ms delay) | -| `discogs` | 1 | Rate-limited Discogs calls (1s delay) | -| `wikipedia` | 1 | Rate-limited Wikipedia calls | -| `last_fm` | 1 | Rate-limited Last.fm calls (500ms delay) | +| `music_brainz` | 1 | MusicBrainz calls (rate-limited at Req layer via `Req.RateLimiter`) | +| `discogs` | 1 | Discogs calls (rate-limited at Req layer via `Req.RateLimiter`) | +| `wikipedia` | 1 | Wikipedia calls | +| `last_fm` | 1 | Last.fm calls (rate-limited at Req layer via `Req.RateLimiter`) | ### On-Demand Workers diff --git a/lib/discogs.ex b/lib/discogs.ex index 38a10a25..4dc1c874 100644 --- a/lib/discogs.ex +++ b/lib/discogs.ex @@ -13,7 +13,5 @@ defmodule Discogs do API.get_artist_image(url, discogs_config) end - def api_cooldown, do: discogs_config().api_cooldown - defp discogs_config, do: Discogs.Config.resolve(:music_library) end diff --git a/lib/discogs/api.ex b/lib/discogs/api.ex index 7ee52299..93a89b88 100644 --- a/lib/discogs/api.ex +++ b/lib/discogs/api.ex @@ -34,6 +34,7 @@ defmodule Discogs.API do auth: "Discogs token=#{config.personal_access_token}" ) |> Req.Request.merge_options(config.req_options) + |> Req.RateLimiter.attach(name: :discogs, cooldown: config.api_cooldown) |> Req.Request.append_request_steps(log_attempt: &log_attempt/1) |> Req.Request.append_response_steps(log_error: &log_error/1) end diff --git a/lib/last_fm.ex b/lib/last_fm.ex index 2e32d224..a76780f2 100644 --- a/lib/last_fm.ex +++ b/lib/last_fm.ex @@ -93,7 +93,5 @@ defmodule LastFm do "https://www.last.fm/api/auth/?api_key=" <> last_fm_config.api_key end - def api_cooldown, do: last_fm_config().api_cooldown - defp last_fm_config, do: LastFm.Config.resolve(:music_library) end diff --git a/lib/last_fm/api.ex b/lib/last_fm/api.ex index 94419d8c..f5f1196e 100644 --- a/lib/last_fm/api.ex +++ b/lib/last_fm/api.ex @@ -146,6 +146,7 @@ defmodule LastFm.API do user_agent: config.user_agent ) |> Req.Request.merge_options(config.req_options) + |> Req.RateLimiter.attach(name: :last_fm, cooldown: config.api_cooldown) |> Req.Request.put_private(:api_key, config.api_key) |> Req.Request.append_request_steps(log_attempt: &log_attempt/1) |> Req.Request.append_response_steps(parse_error: &parse_error/1) diff --git a/lib/music_brainz.ex b/lib/music_brainz.ex index ae91a2b5..ee6c8eec 100644 --- a/lib/music_brainz.ex +++ b/lib/music_brainz.ex @@ -36,7 +36,5 @@ defmodule MusicBrainz do API.get_artist(musicbrainz_id, music_brainz_config()) end - def api_cooldown, do: music_brainz_config().api_cooldown - defp music_brainz_config, do: MusicBrainz.Config.resolve(:music_library) end diff --git a/lib/music_brainz/api.ex b/lib/music_brainz/api.ex index f454b078..70fdcf7e 100644 --- a/lib/music_brainz/api.ex +++ b/lib/music_brainz/api.ex @@ -492,6 +492,7 @@ defmodule MusicBrainz.API do user_agent: config.user_agent ) |> Req.Request.merge_options(config.req_options) + |> Req.RateLimiter.attach(name: :music_brainz, cooldown: config.api_cooldown) |> Req.Request.append_request_steps(log_attempt: &log_attempt/1) end diff --git a/lib/music_library/application.ex b/lib/music_library/application.ex index 4b63192d..d002b92f 100644 --- a/lib/music_library/application.ex +++ b/lib/music_library/application.ex @@ -10,6 +10,7 @@ defmodule MusicLibrary.Application do @impl true def start(_type, _args) do _ = Assets.Cache.new() + _ = Req.RateLimiter.new() children = [ MusicLibrary.Vault, diff --git a/lib/music_library/worker/artist_refresh_discogs_data.ex b/lib/music_library/worker/artist_refresh_discogs_data.ex index 85ee2f8b..e9d4de5d 100644 --- a/lib/music_library/worker/artist_refresh_discogs_data.ex +++ b/lib/music_library/worker/artist_refresh_discogs_data.ex @@ -3,10 +3,6 @@ defmodule MusicLibrary.Worker.ArtistRefreshDiscogsData do @impl Oban.Worker def perform(%Oban.Job{args: %{"id" => artist_info_id}}) do - result = MusicLibrary.Artists.refresh_discogs_data(artist_info_id) - - Process.sleep(Discogs.api_cooldown()) - - result + MusicLibrary.Artists.refresh_discogs_data(artist_info_id) end end diff --git a/lib/music_library/worker/artist_refresh_music_brainz_data.ex b/lib/music_library/worker/artist_refresh_music_brainz_data.ex index 35d61576..e14c5d1b 100644 --- a/lib/music_library/worker/artist_refresh_music_brainz_data.ex +++ b/lib/music_library/worker/artist_refresh_music_brainz_data.ex @@ -3,10 +3,6 @@ defmodule MusicLibrary.Worker.ArtistRefreshMusicBrainzData do @impl Oban.Worker def perform(%Oban.Job{args: %{"id" => artist_info_id}}) do - result = MusicLibrary.Artists.refresh_musicbrainz_data(artist_info_id) - - Process.sleep(MusicBrainz.api_cooldown()) - - result + MusicLibrary.Artists.refresh_musicbrainz_data(artist_info_id) end end diff --git a/lib/music_library/worker/fetch_artist_last_fm_data.ex b/lib/music_library/worker/fetch_artist_last_fm_data.ex index 53f0754f..45b2b5b3 100644 --- a/lib/music_library/worker/fetch_artist_last_fm_data.ex +++ b/lib/music_library/worker/fetch_artist_last_fm_data.ex @@ -3,14 +3,9 @@ defmodule MusicLibrary.Worker.FetchArtistLastFmData do @impl Oban.Worker def perform(%Oban.Job{args: %{"id" => artist_id}}) do - result = - case MusicLibrary.Artists.fetch_lastfm_data(artist_id) do - {:ok, _artist_info} -> :ok - error -> error - end - - Process.sleep(LastFm.api_cooldown()) - - result + case MusicLibrary.Artists.fetch_lastfm_data(artist_id) do + {:ok, _artist_info} -> :ok + error -> error + end end end diff --git a/lib/music_library/worker/record_refresh_music_brainz_data.ex b/lib/music_library/worker/record_refresh_music_brainz_data.ex index 4b38853a..0d08abbe 100644 --- a/lib/music_library/worker/record_refresh_music_brainz_data.ex +++ b/lib/music_library/worker/record_refresh_music_brainz_data.ex @@ -5,13 +5,8 @@ defmodule MusicLibrary.Worker.RecordRefreshMusicBrainzData do def perform(%Oban.Job{args: %{"id" => record_id}}) do record = MusicLibrary.Records.get_record!(record_id) - result = - with {:ok, updated_record} <- MusicLibrary.Records.refresh_musicbrainz_data(record) do - MusicLibrary.Records.notify_update(updated_record) - end - - Process.sleep(MusicBrainz.api_cooldown()) - - result + with {:ok, updated_record} <- MusicLibrary.Records.refresh_musicbrainz_data(record) do + MusicLibrary.Records.notify_update(updated_record) + end end end diff --git a/lib/req/rate_limiter.ex b/lib/req/rate_limiter.ex new file mode 100644 index 00000000..06a56948 --- /dev/null +++ b/lib/req/rate_limiter.ex @@ -0,0 +1,71 @@ +defmodule Req.RateLimiter do + @moduledoc """ + A Req request step that enforces minimum intervals between requests per API. + + Uses an ETS table to track the last request timestamp for each named API. + When a request is made before the cooldown has elapsed, the calling process + sleeps for the remaining time. + + ## Usage + + Req.new(...) + |> Req.RateLimiter.attach(name: :music_brainz, cooldown: 500) + + When `cooldown` is 0, the step is a no-op. + """ + + @table __MODULE__ + + @doc """ + Creates the ETS table used to track request timestamps. + Call once at application startup. + """ + def new do + :ets.new(@table, [:set, :public, :named_table]) + end + + @doc """ + Attaches the rate limiter as a request step on the given Req request. + + ## Options + + * `:name` - atom identifying the API (e.g. `:music_brainz`) + * `:cooldown` - minimum milliseconds between requests + + """ + def attach(request, opts) do + name = Keyword.fetch!(opts, :name) + cooldown = Keyword.fetch!(opts, :cooldown) + + request + |> Req.Request.put_private(:rate_limiter_name, name) + |> Req.Request.put_private(:rate_limiter_cooldown, cooldown) + |> Req.Request.prepend_request_steps(rate_limiter: &throttle/1) + end + + defp throttle(request) do + cooldown = Req.Request.get_private(request, :rate_limiter_cooldown) + + if cooldown > 0 do + name = Req.Request.get_private(request, :rate_limiter_name) + now = System.monotonic_time(:millisecond) + + case :ets.lookup(@table, name) do + [{^name, last_at}] -> + elapsed = now - last_at + remaining = cooldown - elapsed + + if remaining > 0 do + Process.sleep(remaining) + end + + [] -> + :ok + end + + :ets.insert(@table, {name, System.monotonic_time(:millisecond)}) + end + + request + end +end diff --git a/test/req/rate_limiter_test.exs b/test/req/rate_limiter_test.exs new file mode 100644 index 00000000..e8c11fc3 --- /dev/null +++ b/test/req/rate_limiter_test.exs @@ -0,0 +1,122 @@ +defmodule Req.RateLimiterTest do + use ExUnit.Case, async: true + + alias Req.RateLimiter + + describe "attach/2" do + test "attaches rate_limiter step and sets private fields" do + request = + Req.new(url: "https://example.com") + |> RateLimiter.attach(name: :test_attach, cooldown: 100) + + assert Req.Request.get_private(request, :rate_limiter_name) == :test_attach + assert Req.Request.get_private(request, :rate_limiter_cooldown) == 100 + + step_names = Enum.map(request.request_steps, &elem(&1, 0)) + assert :rate_limiter in step_names + end + + test "rate_limiter step runs before other request steps" do + request = + Req.new(url: "https://example.com") + |> Req.Request.append_request_steps(log_attempt: fn req -> req end) + |> RateLimiter.attach(name: :test_order, cooldown: 100) + + step_names = Enum.map(request.request_steps, &elem(&1, 0)) + rate_limiter_index = Enum.find_index(step_names, &(&1 == :rate_limiter)) + log_index = Enum.find_index(step_names, &(&1 == :log_attempt)) + + assert rate_limiter_index < log_index + end + end + + describe "throttle behavior" do + test "does not sleep when cooldown is 0" do + name = :"test_no_sleep_#{System.unique_integer([:positive])}" + + adapter = fn request -> + {request, Req.Response.new(status: 200, body: "ok")} + end + + request = + Req.new(url: "https://example.com", adapter: adapter) + |> RateLimiter.attach(name: name, cooldown: 0) + + start = System.monotonic_time(:millisecond) + {:ok, _} = Req.get(request) + {:ok, _} = Req.get(request) + elapsed = System.monotonic_time(:millisecond) - start + + assert elapsed < 50 + end + + test "enforces cooldown between rapid consecutive requests" do + name = :"test_cooldown_#{System.unique_integer([:positive])}" + cooldown = 100 + + adapter = fn request -> + {request, Req.Response.new(status: 200, body: "ok")} + end + + request = + Req.new(url: "https://example.com", adapter: adapter) + |> RateLimiter.attach(name: name, cooldown: cooldown) + + {:ok, _} = Req.get(request) + + start = System.monotonic_time(:millisecond) + {:ok, _} = Req.get(request) + elapsed = System.monotonic_time(:millisecond) - start + + assert elapsed >= cooldown - 10 + end + + test "does not sleep when enough time has elapsed" do + name = :"test_elapsed_#{System.unique_integer([:positive])}" + cooldown = 50 + + adapter = fn request -> + {request, Req.Response.new(status: 200, body: "ok")} + end + + request = + Req.new(url: "https://example.com", adapter: adapter) + |> RateLimiter.attach(name: name, cooldown: cooldown) + + {:ok, _} = Req.get(request) + Process.sleep(cooldown + 10) + + start = System.monotonic_time(:millisecond) + {:ok, _} = Req.get(request) + elapsed = System.monotonic_time(:millisecond) - start + + assert elapsed < 30 + end + + test "different API names are tracked independently" do + name_a = :"test_api_a_#{System.unique_integer([:positive])}" + name_b = :"test_api_b_#{System.unique_integer([:positive])}" + cooldown = 100 + + adapter = fn request -> + {request, Req.Response.new(status: 200, body: "ok")} + end + + request_a = + Req.new(url: "https://example.com", adapter: adapter) + |> RateLimiter.attach(name: name_a, cooldown: cooldown) + + request_b = + Req.new(url: "https://example.com", adapter: adapter) + |> RateLimiter.attach(name: name_b, cooldown: cooldown) + + {:ok, _} = Req.get(request_a) + + start = System.monotonic_time(:millisecond) + {:ok, _} = Req.get(request_b) + elapsed = System.monotonic_time(:millisecond) - start + + assert elapsed < 30 + end + end +end