fix(server): Rate limit outcomes emited only for events on "fast-path"

CHANGELOG.md

@@ -14,6 +14,7 @@
 - Accept big-endian minidumps. ([#789](https://github.com/getsentry/relay/pull/789))
 - Detect network outages and retry sending events instead of dropping them. ([#788](https://github.com/getsentry/relay/pull/788))
 - Rate limit outcomes emitted only for events. ([#806](https://github.com/getsentry/relay/pull/806))
+- Rate limit outcomes emited only for events on "fast-path". ([#809](https://github.com/getsentry/relay/pull/809))
 
 **Internal**:
 

relay-quotas/src/rate_limit.rs

@@ -286,6 +286,21 @@ impl RateLimits {
     pub fn longest(&self) -> Option<&RateLimit> {
         self.iter().max_by_key(|limit| limit.retry_after)
     }
+
+    /// Returns the most relevant rate limit.
+    ///
+    /// The most relevant rate limit is the longest rate limit for events and if there is no
+    /// rate limit for events then the longest rate limit for anything else
+    pub fn longest_error(&self) -> Option<&RateLimit> {
+        let is_event_related = |rate_limit: &&RateLimit| {
+            rate_limit.categories.is_empty()
+                || rate_limit.categories.iter().any(|cat| cat.is_error())
+        };
+
+        self.iter()
+            .filter(is_event_related)
+            .max_by_key(|limit| limit.retry_after)
+    }
 }
 
 /// Immutable rate limits iterator.

relay-server/src/actors/events.rs

@@ -22,7 +22,7 @@ use relay_general::protocol::{
 };
 use relay_general::store::ClockDriftProcessor;
 use relay_general::types::{Annotated, Array, Object, ProcessingAction, Value};
-use relay_quotas::{RateLimit, RateLimits};
+use relay_quotas::RateLimits;
 use relay_redis::RedisPool;
 
 use crate::actors::outcome::{DiscardReason, Outcome, OutcomeProducer, TrackOutcome};
@@ -142,9 +142,9 @@ impl ProcessingError {
             Self::InvalidTransaction => Some(Outcome::Invalid(DiscardReason::InvalidTransaction)),
             Self::DuplicateItem(_) => Some(Outcome::Invalid(DiscardReason::DuplicateItem)),
             Self::NoEventPayload => Some(Outcome::Invalid(DiscardReason::NoEventPayload)),
-            Self::RateLimited(ref rate_limits) => {
-                most_relevant(rate_limits).map(|r| Outcome::RateLimited(r.reason_code.clone()))
-            }
+            Self::RateLimited(ref rate_limits) => rate_limits
+                .longest_error()
+                .map(|r| Outcome::RateLimited(r.reason_code.clone())),
 
             // Processing-only outcomes (Sentry-internal Relays)
             #[cfg(feature = "processing")]
@@ -170,21 +170,6 @@ impl ProcessingError {
     }
 }
 
-/// Returns the most relevant rate limit.
-///
-/// The most relevant rate limit is the longest rate limit for events and if there is no
-/// rate limit for events then the longest rate limit for anything else
-fn most_relevant(rate_limits: &RateLimits) -> Option<&RateLimit> {
-    let is_event_related = |rate_limit: &&RateLimit| {
-        rate_limit.categories.is_empty() || rate_limit.categories.iter().any(|cat| cat.is_error())
-    };
-
-    rate_limits
-        .iter()
-        .filter(is_event_related)
-        .max_by_key(|limit| limit.retry_after)
-}
-
 type ExtractedEvent = (Annotated<Event>, usize);
 
 /// A state container for envelope processing.
@@ -1631,7 +1616,7 @@ mod tests {
     use chrono::{DateTime, TimeZone, Utc};
 
     use relay_common::DataCategory;
-    use relay_quotas::{RateLimitScope, RetryAfter};
+    use relay_quotas::{RateLimit, RateLimitScope, RetryAfter};
 
     fn create_breadcrumbs_item(breadcrumbs: &[(Option<DateTime<Utc>>, &str)]) -> Item {
         let mut data = Vec::new();
@@ -1799,7 +1784,7 @@ mod tests {
         });
 
         // only non event rate limits so nothing relevant
-        assert_eq!(most_relevant(&rate_limits), None)
+        assert_eq!(rate_limits.longest_error(), None)
     }
 
     #[test]
@@ -1826,7 +1811,7 @@ mod tests {
         };
         rate_limits.add(longest);
 
-        let limit = most_relevant(&rate_limits);
+        let limit = rate_limits.longest_error();
         //we do have an event rate limit
         assert!(limit.is_some());
         // the longest event rate limit is for org 42

relay-server/src/endpoints/common.rs

@@ -79,8 +79,8 @@ pub enum BadStoreRequest {
 }
 
 impl BadStoreRequest {
-    fn to_outcome(&self) -> Outcome {
-        match self {
+    fn to_outcome(&self) -> Option<Outcome> {
+        Some(match self {
             BadStoreRequest::UnsupportedProtocolVersion(_) => {
                 Outcome::Invalid(DiscardReason::AuthVersion)
             }
@@ -119,16 +119,14 @@ impl BadStoreRequest {
             }
 
             BadStoreRequest::RateLimited(rate_limits) => {
-                let reason_code = rate_limits
-                    .longest()
-                    .and_then(|limit| limit.reason_code.clone());
-
-                Outcome::RateLimited(reason_code)
+                return rate_limits
+                    .longest_error()
+                    .map(|r| Outcome::RateLimited(r.reason_code.clone()));
             }
 
             // should actually never create an outcome
             BadStoreRequest::InvalidEventId => Outcome::Invalid(DiscardReason::Internal),
-        }
+        })
     }
 }
 
@@ -457,15 +455,16 @@ where
             metric!(counter(RelayCounters::EnvelopeRejected) += 1);
 
             if is_event {
-                outcome_producer.do_send(TrackOutcome {
-                    timestamp: start_time,
-                    scoping: *scoping.borrow(),
-                    outcome: error.to_outcome(),
-                    event_id: *event_id.borrow(),
-                    remote_addr,
-                });
+                if let Some(outcome) = error.to_outcome() {
+                    outcome_producer.do_send(TrackOutcome {
+                        timestamp: start_time,
+                        scoping: *scoping.borrow(),
+                        outcome,
+                        event_id: *event_id.borrow(),
+                        remote_addr,
+                    });
+                }
             }
-
             if !emit_rate_limit && matches!(error, BadStoreRequest::RateLimited(_)) {
                 return Ok(create_response(*event_id.borrow()));
             }

tests/integration/test_store.py

@@ -550,7 +550,8 @@ def test_processing_quotas(
         retry_after2, rest = headers["x-sentry-rate-limits"].split(":", 1)
         assert int(retry_after2) == int(retry_after)
         assert rest == "%s:key" % category
-        outcomes_consumer.assert_rate_limited("get_lost", key_id=key_id)
+        if generates_outcomes:
+            outcomes_consumer.assert_rate_limited("get_lost", key_id=key_id)
 
     relay.dsn_public_key = second_key["publicKey"]