feat(server): Delay serving global config in managed mode before upstream response

relay-server/src/actors/global_config.rs

@@ -141,7 +141,7 @@ pub enum Status {
     ///
     /// This variant should never be sent after the first `Ready` has occured.
     #[default]
-    Waiting,
+    Pending,
 }
 
 impl Status {
@@ -153,7 +153,7 @@ impl Status {
     pub fn get_ready_or_default(self) -> Arc<GlobalConfig> {
         match self {
             Status::Ready(global_config) => global_config,
-            Status::Waiting => Arc::default(),
+            Status::Pending => Arc::default(),
         }
     }
 }
@@ -188,7 +188,7 @@ impl GlobalConfigService {
     /// Creates a new [`GlobalConfigService`].
     pub fn new(config: Arc<Config>, upstream: Addr<UpstreamRelay>) -> Self {
         let (internal_tx, internal_rx) = mpsc::channel(1);
-        let (global_config_watch, _) = watch::channel(Status::Waiting);
+        let (global_config_watch, _) = watch::channel(Status::Pending);
 
         Self {
             config,
@@ -262,8 +262,6 @@ impl GlobalConfigService {
                         if !self.global_config_watch.borrow().is_ready() {
                             relay_log::info!("received global config from upstream");
                         }
-                        // Notifying subscribers only fails when there are no
-                        // subscribers.
                         self.global_config_watch
                             .send_replace(Status::Ready(Arc::new(global_config)));
                         success = true;

relay-server/src/actors/project_cache.rs

@@ -479,6 +479,12 @@ enum GlobalConfigStatus {
     Pending(BTreeSet<ProjectKey>),
 }
 
+impl GlobalConfigStatus {
+    fn is_ready(&self) -> bool {
+        matches!(self, GlobalConfigStatus::Ready(_))
+    }
+}
+
 impl ProjectCacheBroker {
     fn set_global_config(&mut self, global_config: Arc<GlobalConfig>) {
         if let GlobalConfigStatus::Pending(project_keys) = std::mem::replace(
@@ -502,17 +508,13 @@ impl ProjectCacheBroker {
         }
     }
 
-    fn get_global_config(&self) -> Option<Arc<GlobalConfig>> {
+    fn global_config_value(&self) -> Option<Arc<GlobalConfig>> {
         match &self.global_config {
             GlobalConfigStatus::Ready(gc) => Some(gc.clone()),
             GlobalConfigStatus::Pending(_) => None,
         }
     }
 
-    fn global_config_ready(&self) -> bool {
-        matches!(self.global_config, GlobalConfigStatus::Ready(_))
-    }
-
     /// Adds the value to the queue for the provided key.
     pub fn enqueue(&mut self, key: QueueKey, value: ManagedEnvelope) {
         self.index.entry(key.own_key).or_default().insert(key);
@@ -733,7 +735,7 @@ impl ProjectCacheBroker {
     fn handle_processing(&mut self, managed_envelope: ManagedEnvelope) {
         let project_key = managed_envelope.envelope().meta().public_key();
 
-        let Some(global_config) = self.get_global_config() else {
+        let Some(global_config) = self.global_config_value() else {
             // This indicates a logical error in our approach, this function should only
             // be called when we have a global config.
             relay_log::error!("attempted to process envelope without global config");
@@ -820,7 +822,7 @@ impl ProjectCacheBroker {
         if project_state.is_some()
             && (sampling_state.is_some() || sampling_key.is_none())
             && !self.buffer_guard.is_over_high_watermark()
-            && self.global_config_ready()
+            && self.global_config.is_ready()
         {
             return self.handle_processing(context);
         }
@@ -965,7 +967,7 @@ impl Service for ProjectCacheService {
                     relay_log::info!("global config received");
                     GlobalConfigStatus::Ready(global_config)
                 }
-                global_config::Status::Waiting => {
+                global_config::Status::Pending => {
                     relay_log::info!("waiting for global config");
                     GlobalConfigStatus::Pending(BTreeSet::new())
                 }
@@ -989,22 +991,23 @@ impl Service for ProjectCacheService {
 
             loop {
                 tokio::select! {
-                biased;
-                Ok(()) = subscription.changed() => {
-                    match subscription.borrow().clone() {
-                        global_config::Status::Ready(global_config) => broker.set_global_config(global_config),
-                        // The watch should only be updated if it gets a new value.
-                        // This would imply a logical bug.
-                        global_config::Status::Waiting => relay_log::error!("still waiting for the global config"),
+                    biased;
+
+                    Ok(()) = subscription.changed() => {
+                        match subscription.borrow().clone() {
+                            global_config::Status::Ready(global_config) => broker.set_global_config(global_config),
+                            // The watch should only be updated if it gets a new value.
+                            // This would imply a logical bug.
+                            global_config::Status::Pending => relay_log::error!("still waiting for the global config"),
                         }
                     },
-                Some(message) = state_rx.recv() => broker.merge_state(message),
-                // Buffer will not dequeue the envelopes from the spool if there is not enough
-                // permits in `BufferGuard` available. Currently this is 50%.
-                Some(managed_envelope) = buffer_rx.recv() => broker.handle_processing(managed_envelope),
-                _ = ticker.tick() => broker.evict_stale_project_caches(),
-                Some(message) = rx.recv() => broker.handle_message(message),
-                else => break,
+                    Some(message) = state_rx.recv() => broker.merge_state(message),
+                    // Buffer will not dequeue the envelopes from the spool if there is not enough
+                    // permits in `BufferGuard` available. Currently this is 50%.
+                    Some(managed_envelope) = buffer_rx.recv() => broker.handle_processing(managed_envelope),
+                    _ = ticker.tick() => broker.evict_stale_project_caches(),
+                    Some(message) = rx.recv() => broker.handle_message(message),
+                    else => break,
                 }
             }
 

tests/integration/test_envelope.py

@@ -579,3 +579,51 @@ def test_sample_rates_metrics(mini_sentry, relay_with_processing, events_consume
 
     event, _ = events_consumer.get_event()
     assert event["_metrics"]["sample_rates"] == sample_rates
+
+
+# Checks that we buffer envelopes until we have a global config in processing mode.
+def test_buffer_envelopes_without_global_config(
+    mini_sentry, relay_with_processing, events_consumer
+):
+    include_global = False
+    original_endpoint = mini_sentry.app.view_functions["get_project_config"]
+
+    @mini_sentry.app.endpoint("get_project_config")
+    def get_project_config():
+        nonlocal include_global
+
+        res = original_endpoint().get_json()
+        if not include_global:
+            res.pop("global", None)
+        return res
+
+    mini_sentry.add_basic_project_config(42)
+    events_consumer = events_consumer()
+    options = {"cache": {"global_config_fetch_interval": 1}}
+    relay = relay_with_processing(options=options)
+
+    envelope = Envelope()
+    envelope.add_event({"message": "hello, world!"})
+    relay.send_envelope(42, envelope)
+
+    error_raised = False
+    try:
+        # Event is still in buffer because we have not provided a global config
+        events_consumer.get_event(timeout=1)
+    except AssertionError:
+        error_raised = True
+    assert error_raised
+
+    include_global = True
+    # Clear errors because we log error when we request global config yet we dont receive it.
+    assert len(mini_sentry.test_failures) > 0
+    for err in mini_sentry.test_failures:
+        assert (
+            str(err[1])
+            == "Relay sent us event: global config missing in upstream response"
+        )
+    mini_sentry.test_failures.clear()
+
+    # Global configs are fetched in 10 second intervals, so the event should have come
+    # through after a 10 sec timeout.
+    events_consumer.get_event(timeout=10)

tests/integration/test_query.py

@@ -140,9 +140,9 @@ def get_project_config():
     try:
         relay.send_event(42)
 
-        event = mini_sentry.captured_events.get(timeout=8).get_event()
+        event = mini_sentry.captured_events.get(timeout=12).get_event()
         assert event["logentry"] == {"formatted": "Hello, World!"}
-        assert retry_count == 2
+        assert retry_count == 4
 
         if mini_sentry.test_failures:
             for _, error in mini_sentry.test_failures: