relay_server/services/projects/source/
mod.rs

1use relay_base_schema::project::ProjectKey;
2use relay_config::{Config, RelayMode};
3#[cfg(feature = "processing")]
4use relay_redis::RedisClients;
5use relay_system::{Addr, ServiceSpawn, ServiceSpawnExt as _};
6use std::convert::Infallible;
7use std::sync::Arc;
8
9#[cfg(feature = "processing")]
10pub mod redis;
11pub mod upstream;
12
13use crate::services::projects::project::{ProjectState, Revision};
14use crate::services::upstream::UpstreamRelay;
15
16#[cfg(feature = "processing")]
17use self::redis::RedisProjectSource;
18use self::upstream::{UpstreamProjectSource, UpstreamProjectSourceService};
19
20/// Helper type that contains all configured sources for project cache fetching.
21#[derive(Clone, Debug)]
22pub struct ProjectSource {
23    config: Arc<Config>,
24    upstream_source: Addr<UpstreamProjectSource>,
25    #[cfg(feature = "processing")]
26    redis_source: Option<RedisProjectSource>,
27}
28
29impl ProjectSource {
30    /// Starts all project source services in the given [`ServiceSpawn`].
31    pub async fn start_in(
32        services: &dyn ServiceSpawn,
33        config: Arc<Config>,
34        upstream_relay: Addr<UpstreamRelay>,
35        #[cfg(feature = "processing")] _redis: Option<RedisClients>,
36    ) -> Self {
37        let upstream_source = services.start(UpstreamProjectSourceService::new(
38            config.clone(),
39            upstream_relay,
40        ));
41
42        #[cfg(feature = "processing")]
43        let redis_source =
44            _redis.map(|pool| RedisProjectSource::new(config.clone(), pool.project_configs));
45
46        Self {
47            config,
48            upstream_source,
49            #[cfg(feature = "processing")]
50            redis_source,
51        }
52    }
53
54    /// Fetches a project with `project_key` from the configured sources.
55    ///
56    /// Returns a fully sanitized project.
57    pub async fn fetch(
58        self,
59        project_key: ProjectKey,
60        no_cache: bool,
61        current_revision: Revision,
62    ) -> Result<SourceProjectState, ProjectSourceError> {
63        let config = self.config.current();
64
65        match config.relay_mode() {
66            RelayMode::Proxy => return Ok(ProjectState::Dummy.into()),
67            RelayMode::Managed => (), // Proceed with loading the config from redis or upstream
68        }
69
70        #[cfg(feature = "processing")]
71        if let Some(redis_source) = self.redis_source {
72            let current_revision = current_revision.clone();
73
74            let state_fetch_result = redis_source
75                .get_config_if_changed(project_key, current_revision)
76                .await;
77
78            match state_fetch_result {
79                // New state fetched from Redis, possibly pending.
80                //
81                // If it is pending, we must fallback to fetching from the upstream.
82                Ok(SourceProjectState::New(state)) => {
83                    let state = state.sanitized(config.processing_enabled());
84                    if !state.is_pending() {
85                        return Ok(state.into());
86                    }
87                }
88                // Redis reported that we're holding an up-to-date version of the state already,
89                // refresh the state and return the old cached state again.
90                Ok(SourceProjectState::NotModified) => return Ok(SourceProjectState::NotModified),
91                Err(error) => {
92                    relay_log::error!(
93                        error = &error as &dyn std::error::Error,
94                        "failed to fetch project from Redis",
95                    );
96                }
97            };
98        };
99
100        let state = self
101            .upstream_source
102            .send(FetchProjectState {
103                project_key,
104                current_revision,
105                no_cache,
106            })
107            .await
108            .map_err(|_| ProjectSourceError::FatalUpstream)??;
109
110        Ok(match state {
111            SourceProjectState::New(state) => {
112                SourceProjectState::New(state.sanitized(config.processing_enabled()))
113            }
114            SourceProjectState::NotModified => SourceProjectState::NotModified,
115        })
116    }
117}
118
119#[derive(Debug, thiserror::Error)]
120pub enum ProjectSourceError {
121    /// Error returned from the upstream.
122    #[error("upstream error: {0}")]
123    Upstream(#[from] upstream::Error),
124    /// Upstream did not return a result.
125    ///
126    /// This happens when the upstream does not reply to the request.
127    /// This should never happen.
128    #[error("fatal upstream error")]
129    FatalUpstream,
130}
131
132impl From<Infallible> for ProjectSourceError {
133    fn from(value: Infallible) -> Self {
134        match value {}
135    }
136}
137
138#[derive(Clone, Debug)]
139pub struct FetchProjectState {
140    /// The public key to fetch the project by.
141    pub project_key: ProjectKey,
142
143    /// Currently cached revision if available.
144    ///
145    /// The upstream is allowed to omit full project configs
146    /// for requests for which the requester already has the most
147    /// recent revision.
148    pub current_revision: Revision,
149
150    /// If true, all caches should be skipped and a fresh state should be computed.
151    pub no_cache: bool,
152}
153
154/// Response indicating whether a project state needs to be updated
155/// or the upstream does not have a newer version.
156#[derive(Debug, Clone)]
157pub enum SourceProjectState {
158    /// The upstream sent a [`ProjectState`].
159    New(ProjectState),
160    /// The upstream indicated that there is no newer version of the state available.
161    NotModified,
162}
163
164impl From<ProjectState> for SourceProjectState {
165    fn from(value: ProjectState) -> Self {
166        Self::New(value)
167    }
168}
relay_server/services/projects/source/mod.rs

relay_server/services/projects/source/
mod.rs
