fix(projects): Treat fetch failures as pending

CHANGELOG.md

@@ -11,6 +11,7 @@
 - Report invalid spans with appropriate outcome reason. ([#4051](https://github.com/getsentry/relay/pull/4051))
 - Use the duration reported by the profiler instead of the transaction. ([#4058](https://github.com/getsentry/relay/pull/4058))
 - Incorrect pattern matches involving adjacent any and wildcard matchers. ([#4072](https://github.com/getsentry/relay/pull/4072))
+- Accept incoming requests even if there was an error fetching their project config. ([#4140](https://github.com/getsentry/relay/pull/4140))
 
 **Features:**
 

relay-server/src/services/projects/cache.rs

@@ -1,4 +1,5 @@
 use std::collections::{BTreeMap, BTreeSet};
+use std::convert::Infallible;
 use std::error::Error;
 use std::sync::Arc;
 use std::time::Duration;
@@ -493,12 +494,11 @@ impl ProjectSource {
         project_key: ProjectKey,
         no_cache: bool,
         cached_state: ProjectFetchState,
-    ) -> Result<ProjectFetchState, ()> {
+    ) -> Result<ProjectFetchState, ProjectSourceError> {
         let state_opt = self
             .local_source
             .send(FetchOptionalProjectState { project_key })
-            .await
-            .map_err(|_| ())?;
+            .await?;
 
         if let Some(state) = state_opt {
             return Ok(ProjectFetchState::new(state));
@@ -516,12 +516,11 @@ impl ProjectSource {
         if let Some(redis_source) = self.redis_source {
             let current_revision = current_revision.clone();
 
-            let redis_permit = self.redis_semaphore.acquire().await.map_err(|_| ())?;
+            let redis_permit = self.redis_semaphore.acquire().await?;
             let state_fetch_result = tokio::task::spawn_blocking(move || {
                 redis_source.get_config_if_changed(project_key, current_revision.as_deref())
             })
-            .await
-            .map_err(|_| ())?;
+            .await?;
             drop(redis_permit);
 
             match state_fetch_result {
@@ -553,8 +552,7 @@ impl ProjectSource {
                 current_revision,
                 no_cache,
             })
-            .await
-            .map_err(|_| ())?;
+            .await?;
 
         match state {
             UpstreamProjectState::New(state) => Ok(ProjectFetchState::new(state.sanitized())),
@@ -563,6 +561,22 @@ impl ProjectSource {
     }
 }
 
+#[derive(Debug, thiserror::Error)]
+enum ProjectSourceError {
+    #[error("redis permit error {0}")]
+    RedisPermit(#[from] tokio::sync::AcquireError),
+    #[error("redis join error {0}")]
+    RedisJoin(#[from] tokio::task::JoinError),
+    #[error("upstream error {0}")]
+    Upstream(#[from] relay_system::SendError),
+}
+
+impl From<Infallible> for ProjectSourceError {
+    fn from(value: Infallible) -> Self {
+        match value {}
+    }
+}
+
 /// Updates the cache with new project state information.
 struct UpdateProjectState {
     /// The public key to fetch the project by.
@@ -777,7 +791,13 @@ impl ProjectCacheBroker {
             let state = source
                 .fetch(project_key, no_cache, cached_state)
                 .await
-                .unwrap_or_else(|()| ProjectFetchState::disabled());
+                .unwrap_or_else(|e| {
+                    relay_log::error!(
+                        error = &e as &dyn Error,
+                        "Failed to fetch project from source"
+                    );
+                    ProjectFetchState::pending()
+                });
 
             let message = UpdateProjectState {
                 project_key,

tests/integration/test_query.py

@@ -154,17 +154,12 @@ def get_project_config():
         mini_sentry.clear_test_failures()
 
 
-def test_query_retry_maxed_out(mini_sentry, relay_with_processing, events_consumer):
+def test_query_retry_maxed_out(mini_sentry, relay):
     """
     Assert that a query is not retried an infinite amount of times.
-
-    This is not specific to processing or store, but here we have the outcomes
-    consumer which we can use to assert that an event has been dropped.
     """
     request_count = 0
 
-    events_consumer = events_consumer()
-
     original_get_project_config = mini_sentry.app.view_functions["get_project_config"]
 
     @mini_sentry.app.endpoint("get_project_config")
@@ -184,9 +179,7 @@ def get_project_config():
     for retry in range(RETRIES):  # 1 retry
         query_timeout += 1 * 1.5 ** (retry + 1)
 
-    relay = relay_with_processing(
-        {"limits": {"query_timeout": math.ceil(query_timeout)}}
-    )
+    relay = relay(mini_sentry, {"limits": {"query_timeout": math.ceil(query_timeout)}})
 
     # No error messages yet
     assert mini_sentry.test_failures.empty()
@@ -199,6 +192,12 @@ def get_project_config():
         assert {str(e) for _, e in mini_sentry.current_test_failures()} == {
             "Relay sent us event: error fetching project states: upstream request returned error 500 Internal Server Error: no error details",
         }
+
+        time.sleep(1)  # Wait for project to be cached
+
+        # Relay still accepts events for this project
+        next_response = relay.send_event(42)
+        assert "id" in next_response
     finally:
         mini_sentry.clear_test_failures()