feat(project-config): Query project configs with revision (#3947)

Follow-up to to #3892, now also implements the revision fetching in the
project config API.

Sentry does not need to be updated and this is not a breaking change and
works with any combinations of Relays.

Extends the project config API with another field `revisions` which
optionally contains the revisions for all requested public keys. If
there are are revisions, the response is extended with a field
`unchanged` which contains all project keys which were unchanged (have
the same revision), similar to how `pending` is handled.

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+**Features:**
+
+- Extend project config API to be revision aware. ([#3947](https://github.com/getsentry/relay/pull/3947)).
+
 **Bug Fixes**:
 
 - Keep frames from both ends of the stacktrace when trimming frames. ([#3905](https://github.com/getsentry/relay/pull/3905))

relay-server/src/endpoints/project_configs.rs

@@ -6,7 +6,8 @@ use axum::handler::Handler;
 use axum::http::Request;
 use axum::response::{IntoResponse, Result};
 use axum::{Json, RequestExt};
-use futures::future;
+use futures::stream::FuturesUnordered;
+use futures::StreamExt;
 use relay_base_schema::project::ProjectKey;
 use relay_dynamic_config::{ErrorBoundary, GlobalConfig};
 use serde::{Deserialize, Serialize};
@@ -79,6 +80,8 @@ struct GetProjectStatesResponseWrapper {
  configs: HashMap<ProjectKey, ProjectStateWrapper>,
  #[serde(skip_serializing_if = "Vec::is_empty")]
  pending: Vec<ProjectKey>,
+ #[serde(skip_serializing_if = "Vec::is_empty")]
+ unchanged: Vec<ProjectKey>,
  #[serde(skip_serializing_if = "Option::is_none")]
  global: Option<Arc<GlobalConfig>>,
  #[serde(skip_serializing_if = "Option::is_none")]
@@ -92,7 +95,13 @@ struct GetProjectStatesResponseWrapper {
 #[derive(Debug, Deserialize)]
 #[serde(rename_all = "camelCase")]
 struct GetProjectStatesRequest {
+ /// The list of all requested project configs.
  public_keys: Vec<ErrorBoundary<ProjectKey>>,
+ /// List of revisions for all project configs.
+ ///
+ /// This length of this list if specified must be the same length
+ /// as [`Self::public_keys`], the items are asssociated by index.
+ revisions: Option<ErrorBoundary<Vec<Option<String>>>>,
  #[serde(default)]
  full_config: bool,
  #[serde(default)]
@@ -101,6 +110,32 @@ struct GetProjectStatesRequest {
  global: bool,
 }
 
+fn into_valid_keys(
+ public_keys: Vec<ErrorBoundary<ProjectKey>>,
+ revisions: Option<ErrorBoundary<Vec<Option<String>>>>,
+) -> impl Iterator<Item = (ProjectKey, Option<String>)> {
+ let mut revisions = revisions.and_then(|e| e.ok()).unwrap_or_default();
+ if !revisions.is_empty() && revisions.len() != public_keys.len() {
+ // The downstream sent us a different amount of revisions than project keys,
+ // this indicates an error in the downstream code. Just to be safe, discard
+ // all revisions and carry on as if the downstream never sent any revisions.
+ relay_log::warn!(
+ "downstream sent {} project keys but {} revisions, discarding all revisions",
+ public_keys.len(),
+ revisions.len()
+ );
+ revisions.clear();
+ }
+ let revisions = revisions.into_iter().chain(std::iter::repeat(None));
+
+ std::iter::zip(public_keys, revisions).filter_map(|(public_key, revision)| {
+ // Skip unparsable public keys.
+ // The downstream Relay will consider them `ProjectState::missing`.
+ let public_key = public_key.ok()?;
+ Some((public_key, revision))
+ })
+}
+
 async fn inner(
  state: ServiceState,
  Query(version): Query<VersionQuery>,
@@ -112,21 +147,21 @@ async fn inner(
  let no_cache = inner.no_cache;
  let keys_len = inner.public_keys.len();
 
- // Skip unparsable public keys. The downstream Relay will consider them `ProjectState::missing`.
- let valid_keys = inner.public_keys.into_iter().filter_map(|e| e.ok());
- let futures = valid_keys.map(|project_key| async move {
- let state_result = if version.version >= ENDPOINT_V3 && !no_cache {
- project_cache
- .send(GetCachedProjectState::new(project_key))
- .await
- } else {
- project_cache
- .send(GetProjectState::new(project_key).no_cache(no_cache))
- .await
- };
+ let mut futures: FuturesUnordered<_> = into_valid_keys(inner.public_keys, inner.revisions)
+ .map(|(project_key, revision)| async move {
+ let state_result = if version.version >= ENDPOINT_V3 && !no_cache {
+ project_cache
+ .send(GetCachedProjectState::new(project_key))
+ .await
+ } else {
+ project_cache
+ .send(GetProjectState::new(project_key).no_cache(no_cache))
+ .await
+ };
 
- (project_key, state_result)
- });
+ (project_key, revision, state_result)
+ })
+ .collect();
 
  let (global, global_status) = if inner.global {
  match state.global_config().send(global_config::Get).await? {
@@ -143,8 +178,10 @@ async fn inner(
  };
 
  let mut pending = Vec::with_capacity(keys_len);
+ let mut unchanged = Vec::with_capacity(keys_len);
  let mut configs = HashMap::with_capacity(keys_len);
- for (project_key, state_result) in future::join_all(futures).await {
+
+ while let Some((project_key, revision, state_result)) = futures.next().await {
  let project_info = match state_result? {
  ProjectState::Enabled(info) => info,
  ProjectState::Disabled => {
@@ -157,6 +194,12 @@ async fn inner(
  }
  };
 
+ // Only ever omit responses when there was a valid revision in the first place.
+ if revision.is_some() && project_info.rev == revision {
+ unchanged.push(project_key);
+ continue;
+ }
+
  // If public key is known (even if rate-limited, which is Some(false)), it has
  // access to the project config
  let has_access = relay.internal
@@ -187,6 +230,7 @@ async fn inner(
  Ok(Json(GetProjectStatesResponseWrapper {
  configs,
  pending,
+ unchanged,
  global,
  global_status,
  }))

relay-server/src/services/project/state.rs

@@ -114,3 +114,13 @@ pub struct LimitedParsedProjectState {
  #[serde(flatten)]
  pub info: ProjectInfo,
 }
+
+/// Response indicating whether a project state needs to be updated
+/// or the upstream does not have a newer version.
+#[derive(Debug, Clone)]
+pub enum UpstreamProjectState {
+ /// The upstream sent a [`ProjectState`].
+ New(ProjectState),
+ /// The upstream indicated that there is no newer version of the state available.
+ NotModified,
+}

relay-server/src/services/project_cache.rs

@@ -8,6 +8,7 @@ use crate::services::buffer::{EnvelopeBufferError, EnvelopeBufferGuard, GuardedE
 use crate::services::processor::{
  EncodeMetrics, EnvelopeProcessor, MetricData, ProcessEnvelope, ProcessingGroup, ProjectMetrics,
 };
+use crate::services::project::state::UpstreamProjectState;
 use chrono::{DateTime, Utc};
 use hashbrown::HashSet;
 use relay_base_schema::project::ProjectKey;
@@ -459,7 +460,7 @@ impl ProjectSource {
  self,
  project_key: ProjectKey,
  no_cache: bool,
- _cached_state: ProjectFetchState,
+ cached_state: ProjectFetchState,
  ) -> Result<ProjectFetchState, ()> {
  let state_opt = self
  .local_source
@@ -478,29 +479,32 @@ impl ProjectSource {
  RelayMode::Managed => (), // Proceed with loading the config from redis or upstream
  }
 
+ let current_revision = cached_state.revision().map(String::from);
  #[cfg(feature = "processing")]
  if let Some(redis_source) = self.redis_source {
- let revision = _cached_state.revision().map(String::from);
+ let current_revision = current_revision.clone();
 
  let redis_permit = self.redis_semaphore.acquire().await.map_err(|_| ())?;
  let state_fetch_result = tokio::task::spawn_blocking(move || {
- redis_source.get_config_if_changed(project_key, revision.as_deref())
+ redis_source.get_config_if_changed(project_key, current_revision.as_deref())
  })
  .await
  .map_err(|_| ())?;
  drop(redis_permit);
 
  match state_fetch_result {
  // New state fetched from Redis, possibly pending.
- Ok(Some(state)) => {
+ Ok(UpstreamProjectState::New(state)) => {
  let state = state.sanitized();
  if !state.is_pending() {
  return Ok(ProjectFetchState::new(state));
  }
  }
  // Redis reported that we're holding an up-to-date version of the state already,
  // refresh the state and return the old cached state again.
- Ok(None) => return Ok(ProjectFetchState::refresh(_cached_state)),
+ Ok(UpstreamProjectState::NotModified) => {
+ return Ok(ProjectFetchState::refresh(cached_state))
+ }
  Err(error) => {
  relay_log::error!(
  error = &error as &dyn Error,
@@ -510,13 +514,20 @@ impl ProjectSource {
  };
  };
 
- self.upstream_source
+ let state = self
+ .upstream_source
  .send(FetchProjectState {
  project_key,
+ current_revision,
  no_cache,
  })
  .await
- .map_err(|_| ())
+ .map_err(|_| ())?;
+
+ match state {
+ UpstreamProjectState::New(state) => Ok(ProjectFetchState::new(state.sanitized())),
+ UpstreamProjectState::NotModified => Ok(ProjectFetchState::refresh(cached_state)),
+ }
  }
 }
 
@@ -1502,6 +1513,16 @@ pub struct FetchProjectState {
  /// The public key to fetch the project by.
  pub project_key: ProjectKey,
 
+ /// Currently cached revision if available.
+ ///
+ /// The upstream is allowed to omit full project configs
+ /// for requests for which the requester already has the most
+ /// recent revision.
+ ///
+ /// Settings this to `None` will essentially always re-fetch
+ /// the project config.
+ pub current_revision: Option<String>,
+
  /// If true, all caches should be skipped and a fresh state should be computed.
  pub no_cache: bool,
 }

relay-server/src/services/project_redis.rs

@@ -5,6 +5,7 @@ use relay_config::Config;
 use relay_redis::{RedisError, RedisPool};
 use relay_statsd::metric;
 
+use crate::services::project::state::UpstreamProjectState;
 use crate::services::project::{ParsedProjectState, ProjectState};
 use crate::statsd::{RelayCounters, RelayHistograms, RelayTimers};
 
@@ -54,16 +55,13 @@ impl RedisProjectSource {
 
  /// Fetches a project config from Redis.
  ///
- /// Returns `None` if the the project config stored in Redis has the same `revision`.
- /// Always returns a project state if the passed `revision` is `None`.
- ///
  /// The returned project state is [`ProjectState::Pending`] if the requested project config is not
  /// stored in Redis.
  pub fn get_config_if_changed(
  &self,
  key: ProjectKey,
  revision: Option<&str>,
- ) -> Result<Option<ProjectState>, RedisProjectError> {
+ ) -> Result<UpstreamProjectState, RedisProjectError> {
  let mut client = self.redis.client()?;
  let mut connection = client.connection()?;
 
@@ -82,7 +80,7 @@ impl RedisProjectSource {
  counter(RelayCounters::ProjectStateRedis) += 1,
  hit = "revision",
  );
- return Ok(None);
+ return Ok(UpstreamProjectState::NotModified);
  }
  }
 
@@ -96,7 +94,7 @@ impl RedisProjectSource {
  counter(RelayCounters::ProjectStateRedis) += 1,
  hit = "false"
  );
- return Ok(Some(ProjectState::Pending));
+ return Ok(UpstreamProjectState::New(ProjectState::Pending));
  };
 
  let response = ProjectState::from(parse_redis_response(response.as_slice())?);
@@ -113,13 +111,13 @@ impl RedisProjectSource {
  counter(RelayCounters::ProjectStateRedis) += 1,
  hit = "project_config_revision"
  );
- Ok(None)
+ Ok(UpstreamProjectState::NotModified)
  } else {
  metric!(
  counter(RelayCounters::ProjectStateRedis) += 1,
  hit = "project_config"
  );
- Ok(Some(response))
+ Ok(UpstreamProjectState::New(response))
  }
  }