fix(rate-limits): align the SDK with Sentry logging specs

lib/sentry/config.ex

@@ -369,6 +369,13 @@ defmodule Sentry.Config do
       doc: """
       The level to use when Sentry fails to
       send an event due to an API failure or other reasons.
+
+      Rate limits are the exception: Sentry logs them once per rate-limit window,
+      and logs the individual events dropped while a limit is active at the
+      `:debug` level.
+
+      All messages logged by the SDK itself carry `domain: [:sentry]`, so you can
+      filter them out through `Logger`'s metadata filters.
       """
     ],
     filter: [

lib/sentry/opentelemetry/span_processor.ex

@@ -10,7 +10,7 @@ if Sentry.OpenTelemetry.VersionChecker.tracing_compatible?() do
     alias OpenTelemetry.SemConv.Incubating.URLAttributes, as: URLAttributes
     require OpenTelemetry.SemConv.Incubating.MessagingAttributes, as: MessagingAttributes
 
-    alias Sentry.LoggerUtils
+    alias Sentry.{ClientError, LoggerUtils}
 
     alias Sentry.{Transaction, OpenTelemetry.SpanStorage, OpenTelemetry.SpanRecord}
     alias Sentry.Interfaces.Span
@@ -180,6 +180,13 @@ if Sentry.OpenTelemetry.VersionChecker.tracing_compatible?() do
           :excluded ->
             true
 
+          {:error, %ClientError{reason: :rate_limited} = error} ->
+            LoggerUtils.debug(fn ->
+              "Failed to send transaction to Sentry: #{inspect(error)}"
+            end)
+
+            {:error, :invalid_span}
+
           {:error, error} ->
             LoggerUtils.log(fn -> "Failed to send transaction to Sentry: #{inspect(error)}" end)
             {:error, :invalid_span}

lib/sentry/telemetry/scheduler.ex

@@ -36,6 +36,7 @@ defmodule Sentry.Telemetry.Scheduler do
 
   alias Sentry.{
     CheckIn,
+    ClientError,
     ClientReport,
     Config,
     Envelope,
@@ -442,6 +443,9 @@ defmodule Sentry.Telemetry.Scheduler do
         {:ok, _id} ->
           :ok
 
+        {:error, %ClientError{reason: :rate_limited} = error} ->
+          {:error, error}
+
         {:error, error} ->
           LoggerUtils.log(fn ->
             "Sentry: failed to send envelope: #{Exception.message(error)}"

lib/sentry/transport.ex

@@ -61,8 +61,9 @@ defmodule Sentry.Transport do
       {:ok, id} ->
         {:ok, id}
 
+      # A 429 is counted upstream, so recording a client report here would
+      # double-count the discarded items.
       {:error, :rate_limited} ->
-        ClientReport.Sender.record_discarded_events(:ratelimit_backoff, items)
         {:error, ClientError.new(:rate_limited)}
 
       {:error, {:envelope_too_large, {status, headers, body}}} ->
@@ -219,16 +220,19 @@ defmodule Sentry.Transport do
     if Enum.any?(events, &(Map.has_key?(&1, :source) && &1.source == :logger)) do
       :ok
     else
-      message =
-        case send_result do
-          {:error, %ClientError{} = error} ->
-            "Failed to send Sentry event. #{Exception.message(error)}"
+      log_send_result(send_result)
+    end
+  end
 
-          {:ok, _} ->
-            nil
-        end
+  defp log_send_result({:error, %ClientError{reason: :rate_limited} = error}) do
+    LoggerUtils.debug(fn -> ["Failed to send Sentry event. ", Exception.message(error)] end)
+  end
 
-      if message, do: LoggerUtils.log(fn -> [message] end)
-    end
+  defp log_send_result({:error, %ClientError{} = error}) do
+    LoggerUtils.log(fn -> ["Failed to send Sentry event. ", Exception.message(error)] end)
+  end
+
+  defp log_send_result({:ok, _envelope_id}) do
+    :ok
   end
 end

lib/sentry/transport/rate_limiter.ex

@@ -24,6 +24,8 @@ defmodule Sentry.Transport.RateLimiter do
 
   use GenServer
 
+  alias Sentry.LoggerUtils
+
   @default_sweep_interval_ms 60_000
 
   defstruct [:table_name]
@@ -137,7 +139,10 @@ defmodule Sentry.Transport.RateLimiter do
   """
   @spec update_global_rate_limit(pos_integer()) :: :ok
   def update_global_rate_limit(retry_after_seconds) when is_integer(retry_after_seconds) do
-    store_max_expiry(:global, System.system_time(:millisecond) + retry_after_seconds * 1000)
+    now = System.system_time(:millisecond)
+    expiry = now + retry_after_seconds * 1000
+
+    store_limits([{:global, expiry}], now)
   end
 
   @doc """
@@ -158,44 +163,96 @@ defmodule Sentry.Transport.RateLimiter do
 
     rate_limits_header
     |> parse_rate_limits_header()
-    |> Enum.each(fn {category, retry_after_ms} ->
-      store_max_expiry(category, now + retry_after_ms)
+    |> Enum.map(fn {category, retry_after_ms} -> {category, now + retry_after_ms} end)
+    |> store_limits(now)
+  end
+
+  defp store_limits(limits, now) do
+    limits
+    |> Enum.reduce(%{}, fn {category, expiry}, acc ->
+      Map.update(acc, category, expiry, &max(&1, expiry))
+    end)
+    |> Enum.filter(fn {category, expiry} ->
+      store_max_expiry(category, expiry, now) == :started and expiry > now
     end)
+    |> log_new_limits(now)
+  end
+
+  defp log_new_limits([], _now), do: :ok
+
+  defp log_new_limits(limits, now) do
+    LoggerUtils.log(fn ->
+      [
+        "Sentry is rate-limiting ",
+        Enum.map_join(limits, ", ", &format_limit(&1, now)),
+        ". Data is dropped locally until the limit expires."
+      ]
+    end)
+  end
 
-    :ok
+  defp format_limit({category, expiry}, now) do
+    "#{format_category(category)} for #{format_duration(expiry - now)} (until #{format_expiry(expiry)})"
   end
 
-  ## Private Helpers
+  defp format_category(:global), do: "all data categories"
+  defp format_category(category), do: ~s(the "#{category}" data category)
+
+  defp format_duration(duration_ms) when duration_ms >= 1000, do: "#{div(duration_ms, 1000)}s"
+  defp format_duration(duration_ms), do: "#{duration_ms}ms"
+
+  defp format_expiry(expiry) do
+    expiry
+    |> DateTime.from_unix!(:millisecond)
+    |> DateTime.truncate(:second)
+    |> DateTime.to_iso8601()
+  end
 
   # Rate limits may only ever be extended: a response carrying a shorter delay
   # than the one already in flight must not let the SDK resume sending early.
   #
   # Senders handle responses concurrently, so this compares and swaps rather than
   # reading and then writing: two responses racing on the same category would
   # otherwise both read the old expiry and let the shorter one land last.
-  defp store_max_expiry(category, expiry) do
+  # The write also reports whether it opened a new window (`:started`) or merely
+  # pushed an active one further out (`:extended`), so that callers announcing a
+  # limit do so exactly once even when responses are handled concurrently.
+  defp store_max_expiry(category, expiry, now) do
     table = name()
 
     if :ets.insert_new(table, {category, expiry}) do
-      :ok
+      :started
     else
-      replace_shorter_expiry(table, category, expiry)
+      replace_lapsed_expiry(table, category, expiry, now)
+    end
+  end
+
+  defp replace_lapsed_expiry(table, category, expiry, now) do
+    match_spec = [
+      {{category, :"$1"}, [{:<, :"$1", now}, {:<, :"$1", expiry}],
+       [{{{:const, category}, {:const, expiry}}}]}
+    ]
+
+    case :ets.select_replace(table, match_spec) do
+      1 -> :started
+      0 -> replace_shorter_expiry(table, category, expiry, now)
     end
   end
 
-  defp replace_shorter_expiry(table, category, expiry) do
+  defp replace_shorter_expiry(table, category, expiry, now) do
     match_spec = [
       {{category, :"$1"}, [{:<, :"$1", expiry}], [{{{:const, category}, {:const, expiry}}}]}
     ]
 
     case :ets.select_replace(table, match_spec) do
       1 ->
-        :ok
+        :extended
 
       0 ->
         # Either the stored expiry is already the longer one, or the sweeper
         # pruned the entry between the two calls and it has to be re-inserted.
-        if :ets.member(table, category), do: :ok, else: store_max_expiry(category, expiry)
+        if :ets.member(table, category),
+          do: :extended,
+          else: store_max_expiry(category, expiry, now)
     end
   end
 

test/sentry/opentelemetry/span_processor_test.exs

@@ -134,6 +134,16 @@ defmodule Sentry.Opentelemetry.SpanProcessorTest do
     assert log =~ ~r/domain=(\w+\.)*sentry \[info\]\s+Failed to send transaction to Sentry/
   end
 
+  @tag span_storage: true
+  test "stays quiet when the transaction is dropped by an active rate limit" do
+    put_test_config(environment_name: "test", traces_sample_rate: 1.0)
+    set_rate_limit("transaction")
+
+    log = capture_log(fn -> TestEndpoint.child_instrumented_function("one") end)
+
+    refute log =~ ~r/\[warning\]\s+Failed to send transaction/
+  end
+
   defp assert_valid_iso8601(timestamp) do
     case DateTime.from_iso8601(timestamp) do
       {:ok, datetime, _offset} ->

test/sentry/telemetry/scheduler_test.exs

@@ -470,6 +470,45 @@ defmodule Sentry.Telemetry.SchedulerTest do
       stop_buffers(buffers)
       Bypass.down(bypass)
     end
+
+    test "stays quiet when a direct transport send is rate limited during flush" do
+      %{bypass: bypass} = setup_bypass()
+      prev_retries = Application.get_env(:sentry, :request_retries)
+      Application.put_env(:sentry, :request_retries, [])
+
+      on_exit(fn ->
+        if prev_retries do
+          Application.put_env(:sentry, :request_retries, prev_retries)
+        else
+          Application.delete_env(:sentry, :request_retries)
+        end
+      end)
+
+      Bypass.expect(bypass, "POST", "/api/1/envelope/", fn conn ->
+        conn
+        |> Plug.Conn.put_resp_header("X-Sentry-Rate-Limits", "0:error:key")
+        |> Plug.Conn.resp(429, ~s<{}>)
+      end)
+
+      buffers = start_test_buffers(batch_size: 1)
+      uid = System.unique_integer([:positive])
+
+      {:ok, pid} =
+        Scheduler.start_link(
+          buffers: buffers,
+          name: :"test_scheduler_rate_limited_#{uid}"
+        )
+
+      Buffer.add(buffers.log, make_log_event("rate-limited-send"))
+
+      log = capture_log(fn -> Scheduler.flush(pid) end)
+
+      refute log =~ "failed to send envelope"
+
+      GenServer.stop(pid)
+      stop_buffers(buffers)
+      Bypass.down(bypass)
+    end
   end
 
   # Helper functions

test/sentry/transport/rate_limiter_test.exs

@@ -2,6 +2,7 @@ defmodule Sentry.Transport.RateLimiterTest do
   use Sentry.Case, async: true
 
   import Sentry.TestHelpers
+  import ExUnit.CaptureLog
 
   alias Sentry.Transport.RateLimiter
 
@@ -236,6 +237,76 @@ defmodule Sentry.Transport.RateLimiterTest do
     end
   end
 
+  describe "logging of new rate limits" do
+    test "announces every newly limited category in a single message" do
+      log =
+        capture_log(fn ->
+          RateLimiter.update_rate_limits("60:error;transaction:key, 120:attachment:org")
+        end)
+
+      assert length(String.split(log, "Sentry is rate-limiting")) == 2
+      assert log =~ ~s("error")
+      assert log =~ ~s("transaction")
+      assert log =~ ~s("attachment")
+      assert log =~ "60s"
+      assert log =~ "120s"
+    end
+
+    test "stays quiet when a shorter limit arrives for an active category" do
+      RateLimiter.update_rate_limits("60:error:key")
+
+      log = capture_log(fn -> RateLimiter.update_rate_limits("30:error:key") end)
+
+      refute log =~ "Sentry is rate-limiting"
+    end
+
+    test "announces a new limit landing on an expired entry" do
+      set_rate_limit("error", duration: -10)
+
+      log = capture_log(fn -> RateLimiter.update_rate_limits("60:error:key") end)
+
+      assert log =~ "Sentry is rate-limiting"
+      assert log =~ ~s("error")
+    end
+
+    test "announces a limit once when responses race" do
+      rounds = 20
+
+      announcements =
+        for round <- 1..rounds, reduce: 0 do
+          acc ->
+            log = capture_log(fn -> race_update_rate_limits("error-#{round}") end)
+            acc + length(String.split(log, "Sentry is rate-limiting")) - 1
+        end
+
+      assert announcements == rounds
+    end
+  end
+
+  defp race_update_rate_limits(category, senders \\ 40) do
+    table = table_name()
+    release = :atomics.new(1, [])
+    parent = self()
+
+    tasks =
+      for _ <- 1..senders do
+        Task.async(fn ->
+          Process.put(:rate_limiter_table_name, table)
+          send(parent, :ready)
+          await_release(release)
+          RateLimiter.update_rate_limits("60:#{category}:key")
+        end)
+      end
+
+    for _ <- 1..senders, do: assert_receive(:ready)
+    :atomics.put(release, 1, 1)
+    Task.await_many(tasks, 5000)
+  end
+
+  defp await_release(release) do
+    if :atomics.get(release, 1) == 1, do: :ok, else: await_release(release)
+  end
+
   defp table_name, do: Process.get(:rate_limiter_table_name)
 
   defp stored_expiry(category) do