relay_server/services/projects/cache/
handle.rs

1use std::fmt;
2use std::sync::Arc;
3use std::time::{Duration, Instant};
4
5use relay_base_schema::project::ProjectKey;
6use relay_config::Config;
7use relay_system::Addr;
8use tokio::sync::broadcast;
9
10use super::state::Shared;
11use crate::services::projects::cache::service::ProjectChange;
12use crate::services::projects::cache::{Project, ProjectCache};
13use crate::services::projects::project::ProjectState;
14use crate::statsd::RelayTimers;
15
16/// A synchronous handle to the [`ProjectCache`].
17///
18/// The handle allows lock free access to cached projects. It also acts as an interface
19/// to the [`ProjectCacheService`](super::ProjectCacheService).
20#[derive(Clone)]
21pub struct ProjectCacheHandle {
22    pub(super) shared: Arc<Shared>,
23    pub(super) config: Arc<Config>,
24    pub(super) service: Addr<ProjectCache>,
25    pub(super) project_changes: broadcast::Sender<ProjectChange>,
26}
27
28impl ProjectCacheHandle {
29    /// Returns the current project state for the `project_key`.
30    pub fn get(&self, project_key: ProjectKey) -> Project<'_> {
31        let project = self.shared.get_or_create(project_key);
32        // Always trigger a fetch after retrieving the project to make sure the state is up to date.
33        self.fetch(project_key);
34
35        Project::new(project, self.config.current())
36    }
37
38    /// Awaits until the given project state becomes ready (enabled or disabled).
39    ///
40    /// Returns [`None`] if the project config cannot be resolved in the given time.
41    pub async fn ready(&self, project_key: ProjectKey, timeout: Duration) -> Option<Project<'_>> {
42        let project = self.get(project_key);
43        if !project.state().is_pending() {
44            return Some(project);
45        }
46
47        let t = Instant::now();
48        let result = tokio::time::timeout(timeout, self.ready_inner(project_key)).await;
49
50        relay_statsd::metric!(
51            timer(RelayTimers::ProjectStateReadyDuration) = t.elapsed(),
52            result = match &result {
53                Ok(project) => {
54                    match project.state() {
55                        ProjectState::Enabled(_) => "enabled",
56                        ProjectState::Dummy => "dummy",
57                        ProjectState::Disabled => "disabled",
58                        ProjectState::Pending => "pending",
59                    }
60                }
61                Err(_) => "timeout",
62            }
63        );
64
65        result.ok()
66    }
67
68    async fn ready_inner(&self, project_key: ProjectKey) -> Project<'_> {
69        loop {
70            let project = self.shared.get_or_create(project_key);
71            // Create the `Notified` before checking the project_state, to prevent missing
72            // an update between the check and the registration of the listener.
73            //
74            // From [`tokio::sync::futures::Notified::enabled`]:
75            // > notifications sent using notify_waiters [...] are received
76            // > as long as they happen after the creation of the Notified
77            let change_listener = project.outdated();
78            if !project.project_state().is_pending() {
79                drop(change_listener);
80                return Project::new(project, self.config.current());
81            }
82            change_listener.await;
83        }
84    }
85
86    /// Triggers a fetch/update check in the project cache for the supplied project.
87    pub fn fetch(&self, project_key: ProjectKey) {
88        self.service.send(ProjectCache::Fetch(project_key));
89    }
90
91    /// Returns a subscription to all [`ProjectChange`]'s.
92    ///
93    /// This stream notifies the subscriber about project state changes in the project cache.
94    /// Events may arrive in arbitrary order and be delivered multiple times.
95    pub fn changes(&self) -> broadcast::Receiver<ProjectChange> {
96        self.project_changes.subscribe()
97    }
98}
99
100impl fmt::Debug for ProjectCacheHandle {
101    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
102        f.debug_struct("ProjectCacheHandle")
103            .field("shared", &self.shared)
104            .finish()
105    }
106}
107
108#[cfg(test)]
109mod test {
110    use crate::services::projects::project::ProjectState;
111    use relay_config::Config;
112
113    use super::*;
114
115    impl ProjectCacheHandle {
116        /// Creates a new [`ProjectCacheHandle`] for testing only.
117        ///
118        /// A project cache handle created this way does not require a service to function.
119        pub fn for_test() -> Self {
120            Self {
121                shared: Default::default(),
122                config: Arc::new(Config::default()),
123                service: Addr::dummy(),
124                project_changes: broadcast::channel(999_999).0,
125            }
126        }
127
128        /// Sets the project state for a project.
129        ///
130        /// This can be used to emulate a project cache update in tests.
131        pub fn test_set_project_state(&self, project_key: ProjectKey, state: ProjectState) {
132            let is_pending = state.is_pending();
133            self.shared.test_set_project_state(project_key, state);
134            if is_pending {
135                let _ = self
136                    .project_changes
137                    .send(ProjectChange::Evicted(project_key));
138            } else {
139                let _ = self.project_changes.send(ProjectChange::Ready(project_key));
140            }
141        }
142
143        /// Returns `true` if there is a project created for this `project_key`.
144        ///
145        /// A project is automatically created on access via [`Self::get`].
146        pub fn test_has_project_created(&self, project_key: ProjectKey) -> bool {
147            self.shared.test_has_project_created(project_key)
148        }
149
150        /// The amount of fetches triggered for projects.
151        ///
152        /// A fetch is triggered for both [`Self::get`] and [`Self::fetch`].
153        pub fn test_num_fetches(&self) -> u64 {
154            self.service.len()
155        }
156    }
157}
relay_server/services/projects/cache/handle.rs

relay_server/services/projects/cache/
handle.rs
