ref(server): Always enforce cached rate limits

Author: Dav1dde

relay-server/src/services/processor.rs

@@ -1245,7 +1245,6 @@ impl EnvelopeProcessorService {
         })
     }
 
-    #[cfg(feature = "processing")]
     async fn enforce_quotas<Group>(
         &self,
         managed_envelope: &mut TypedEnvelope<Group>,
@@ -1273,28 +1272,30 @@ impl EnvelopeProcessorService {
             )
             .await?;
 
-        // Enforce all quotas consistently with Redis.
-        let consistent_result = RateLimiter::Consistent(rate_limiter)
-            .enforce(
-                managed_envelope,
-                cached_result.event,
-                extracted_metrics,
-                &global_config,
-                project_info,
-                rate_limits,
-            )
-            .await?;
+        if_processing!(self.inner.config, {
+            // Enforce all quotas consistently with Redis.
+            let consistent_result = RateLimiter::Consistent(rate_limiter)
+                .enforce(
+                    managed_envelope,
+                    cached_result.event,
+                    extracted_metrics,
+                    &global_config,
+                    project_info,
+                    rate_limits,
+                )
+                .await?;
 
-        // Update cached rate limits with the freshly computed ones.
-        if !consistent_result.rate_limits.is_empty() {
-            self.inner
-                .project_cache
-                .get(managed_envelope.scoping().project_key)
-                .rate_limits()
-                .merge(consistent_result.rate_limits);
-        }
+            // Update cached rate limits with the freshly computed ones.
+            if !consistent_result.rate_limits.is_empty() {
+                self.inner
+                    .project_cache
+                    .get(managed_envelope.scoping().project_key)
+                    .rate_limits()
+                    .merge(consistent_result.rate_limits);
+            }
 
-        Ok(consistent_result.event)
+            Ok(consistent_result.event)
+        } else { Ok(cached_result.event) })
     }
 
     /// Extract transaction metrics.
@@ -1629,17 +1630,15 @@ impl EnvelopeProcessorService {
             );
         }
 
-        if_processing!(self.inner.config, {
-            event = self
-                .enforce_quotas(
-                    managed_envelope,
-                    event,
-                    &mut extracted_metrics,
-                    project_info.clone(),
-                    rate_limits,
-                )
-                .await?;
-        });
+        event = self
+            .enforce_quotas(
+                managed_envelope,
+                event,
+                &mut extracted_metrics,
+                project_info.clone(),
+                rate_limits,
+            )
+            .await?;
 
         if event.value().is_some() {
             event::scrub(&mut event, project_info.clone())?;
@@ -1822,17 +1821,15 @@ impl EnvelopeProcessorService {
             //  - An empty envelope.
             //  - An envelope containing only processed profiles.
             // We need to make sure there are enough quotas for these profiles.
-            if_processing!(self.inner.config, {
-                event = self
-                    .enforce_quotas(
-                        managed_envelope,
-                        Annotated::empty(),
-                        &mut extracted_metrics,
-                        project_info.clone(),
-                        rate_limits,
-                    )
-                    .await?;
-            });
+            event = self
+                .enforce_quotas(
+                    managed_envelope,
+                    Annotated::empty(),
+                    &mut extracted_metrics,
+                    project_info.clone(),
+                    rate_limits,
+                )
+                .await?;
 
             return Ok(Some(extracted_metrics));
         }
@@ -1882,17 +1879,19 @@ impl EnvelopeProcessorService {
                     spans_extracted,
                 );
             }
+        });
 
-            event = self
-                .enforce_quotas(
-                    managed_envelope,
-                    event,
-                    &mut extracted_metrics,
-                    project_info.clone(),
-                    rate_limits,
-                )
-                .await?;
+        event = self
+            .enforce_quotas(
+                managed_envelope,
+                event,
+                &mut extracted_metrics,
+                project_info.clone(),
+                rate_limits,
+            )
+            .await?;
 
+        if_processing!(self.inner.config, {
             event = span::maybe_discard_transaction(managed_envelope, event, project_info);
         });
 
@@ -1925,24 +1924,25 @@ impl EnvelopeProcessorService {
         _rate_limits: Arc<RateLimits>,
     ) -> Result<Option<ProcessingExtractedMetrics>, ProcessingError> {
         profile_chunk::filter(managed_envelope, project_info.clone());
+
         if_processing!(self.inner.config, {
             profile_chunk::process(
                 managed_envelope,
                 project_info.clone(),
                 &self.inner.global_config.current(),
                 &self.inner.config,
             );
-
-            self.enforce_quotas(
-                managed_envelope,
-                Annotated::empty(),
-                &mut ProcessingExtractedMetrics::new(),
-                project_info,
-                _rate_limits,
-            )
-            .await?;
         });
 
+        self.enforce_quotas(
+            managed_envelope,
+            Annotated::empty(),
+            &mut ProcessingExtractedMetrics::new(),
+            project_info,
+            _rate_limits,
+        )
+        .await?;
+
         Ok(None)
     }
 
@@ -1968,16 +1968,14 @@ impl EnvelopeProcessorService {
             project_info.clone(),
         );
 
-        if_processing!(self.inner.config, {
-            self.enforce_quotas(
-                managed_envelope,
-                Annotated::empty(),
-                &mut extracted_metrics,
-                project_info.clone(),
-                _rate_limits,
-            )
-            .await?;
-        });
+        self.enforce_quotas(
+            managed_envelope,
+            Annotated::empty(),
+            &mut extracted_metrics,
+            project_info.clone(),
+            _rate_limits,
+        )
+        .await?;
 
         report::process_user_reports(managed_envelope);
         attachment::scrub(managed_envelope, project_info);
@@ -2000,16 +1998,15 @@ impl EnvelopeProcessorService {
             project_info.clone(),
             &self.inner.config,
         );
-        if_processing!(self.inner.config, {
-            self.enforce_quotas(
-                managed_envelope,
-                Annotated::empty(),
-                &mut extracted_metrics,
-                project_info,
-                rate_limits,
-            )
-            .await?;
-        });
+
+        self.enforce_quotas(
+            managed_envelope,
+            Annotated::empty(),
+            &mut extracted_metrics,
+            project_info,
+            rate_limits,
+        )
+        .await?;
 
         Ok(Some(extracted_metrics))
     }
@@ -2025,16 +2022,14 @@ impl EnvelopeProcessorService {
         #[allow(unused_mut)]
         let mut extracted_metrics = ProcessingExtractedMetrics::new();
 
-        if_processing!(self.inner.config, {
-            self.enforce_quotas(
-                managed_envelope,
-                Annotated::empty(),
-                &mut extracted_metrics,
-                project_info.clone(),
-                rate_limits,
-            )
-            .await?;
-        });
+        self.enforce_quotas(
+            managed_envelope,
+            Annotated::empty(),
+            &mut extracted_metrics,
+            project_info.clone(),
+            rate_limits,
+        )
+        .await?;
 
         report::process_client_reports(
             managed_envelope,
@@ -2065,44 +2060,40 @@ impl EnvelopeProcessorService {
             self.inner.geoip_lookup.as_ref(),
         )?;
 
-        if_processing!(self.inner.config, {
-            self.enforce_quotas(
-                managed_envelope,
-                Annotated::empty(),
-                &mut extracted_metrics,
-                project_info,
-                rate_limits,
-            )
-            .await?;
-        });
+        self.enforce_quotas(
+            managed_envelope,
+            Annotated::empty(),
+            &mut extracted_metrics,
+            project_info,
+            rate_limits,
+        )
+        .await?;
 
         Ok(Some(extracted_metrics))
     }
 
     /// Processes cron check-ins.
     async fn process_checkins(
         &self,
-        #[allow(unused_variables)] managed_envelope: &mut TypedEnvelope<CheckInGroup>,
-        #[allow(unused_variables)] project_id: ProjectId,
-        #[allow(unused_variables)] project_info: Arc<ProjectInfo>,
-        #[allow(unused_variables)] rate_limits: Arc<RateLimits>,
+        managed_envelope: &mut TypedEnvelope<CheckInGroup>,
+        project_id: ProjectId,
+        project_info: Arc<ProjectInfo>,
+        rate_limits: Arc<RateLimits>,
     ) -> Result<Option<ProcessingExtractedMetrics>, ProcessingError> {
-        #[allow(unused_mut)]
-        let mut extracted_metrics = ProcessingExtractedMetrics::new();
+        self.enforce_quotas(
+            managed_envelope,
+            Annotated::empty(),
+            &mut ProcessingExtractedMetrics::new(),
+            project_info,
+            rate_limits,
+        )
+        .await?;
 
         if_processing!(self.inner.config, {
-            self.enforce_quotas(
-                managed_envelope,
-                Annotated::empty(),
-                &mut extracted_metrics,
-                project_info,
-                rate_limits,
-            )
-            .await?;
             self.normalize_checkins(managed_envelope, project_id);
         });
 
-        Ok(Some(extracted_metrics))
+        Ok(None)
     }
 
     /// Process logs
@@ -2122,17 +2113,20 @@ impl EnvelopeProcessorService {
             project_info.clone(),
             &self.inner.global_config.current(),
         );
+
+        self.enforce_quotas(
+            managed_envelope,
+            Annotated::empty(),
+            &mut extracted_metrics,
+            project_info.clone(),
+            rate_limits,
+        )
+        .await?;
+
         if_processing!(self.inner.config, {
-            self.enforce_quotas(
-                managed_envelope,
-                Annotated::empty(),
-                &mut extracted_metrics,
-                project_info.clone(),
-                rate_limits,
-            )
-            .await?;
             ourlog::process(managed_envelope, project_info.clone());
         });
+
         Ok(Some(extracted_metrics))
     }
 
@@ -2176,17 +2170,17 @@ impl EnvelopeProcessorService {
                 &reservoir,
             )
             .await;
-
-            self.enforce_quotas(
-                managed_envelope,
-                Annotated::empty(),
-                &mut extracted_metrics,
-                project_info,
-                rate_limits,
-            )
-            .await?;
         });
 
+        self.enforce_quotas(
+            managed_envelope,
+            Annotated::empty(),
+            &mut extracted_metrics,
+            project_info,
+            rate_limits,
+        )
+        .await?;
+
         Ok(Some(extracted_metrics))
     }