review changes

relay-server/src/endpoints/project_configs.rs

@@ -114,7 +114,18 @@ fn into_valid_keys(
  public_keys: Vec<ErrorBoundary<ProjectKey>>,
  revisions: Option<ErrorBoundary<Vec<Option<String>>>>,
 ) -> impl Iterator<Item = (ProjectKey, Option<String>)> {
- let revisions = revisions.and_then(|e| e.ok()).unwrap_or_default();
+ 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)| {

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

@@ -122,5 +122,5 @@ pub enum UpstreamProjectState {
  /// The upstream sent a [`ProjectState`].
  New(ProjectState),
  /// The upstream indicated that there is no newer version of the state available.
- Unchanged,
+ NotModified,
 }

relay-server/src/services/project_cache.rs

@@ -502,7 +502,7 @@ impl ProjectSource {
  }
  // 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(UpstreamProjectState::Unchanged) => {
+ Ok(UpstreamProjectState::NotModified) => {
  return Ok(ProjectFetchState::refresh(cached_state))
  }
  Err(error) => {
@@ -526,7 +526,7 @@ impl ProjectSource {
 
  match state {
  UpstreamProjectState::New(state) => Ok(ProjectFetchState::new(state.sanitized())),
- UpstreamProjectState::Unchanged => Ok(ProjectFetchState::refresh(cached_state)),
+ UpstreamProjectState::NotModified => Ok(ProjectFetchState::refresh(cached_state)),
  }
  }
 }

relay-server/src/services/project_upstream.rs

@@ -132,18 +132,19 @@ impl ProjectStateChannel {
  self.no_cache = true;
  }
 
- pub fn attach(&mut self, sender: BroadcastSender<UpstreamProjectState>) {
- self.channel.attach(sender)
- }
-
- /// Updates the revision in the current channel.
+ /// Attaches a new sender to the same channel.
  ///
- /// If the new revision is equal to the contained revision this is a no-op.
+ /// Also makes sure the new sender's revision matches the already requested revision.
  /// If the new revision is different from the contained revision this clears the revision.
  /// To not have multiple fetches per revision per batch, we need to find a common denominator
  /// for requests with different revisions, which is always to fetch the full project config.
- pub fn sync_revision(&mut self, new_revision: Option<String>) {
- if self.revision != new_revision {
+ pub fn attach(
+ &mut self,
+ sender: BroadcastSender<UpstreamProjectState>,
+ revision: Option<String>,
+ ) {
+ self.channel.attach(sender);
+ if self.revision != revision {
  self.revision = None;
  }
  }
@@ -571,8 +572,7 @@ impl UpstreamProjectSourceService {
  }
  Entry::Occupied(mut entry) => {
  let channel = entry.get_mut();
- channel.attach(sender);
- channel.sync_revision(current_revision);
+ channel.attach(sender, current_revision);
  // Ensure upstream skips caches if one of the recipients requests an uncached response. This
  // operation is additive across requests.
  if no_cache {